38 - Semaphore: Bounded Concurrency

Previous · Next

A semaphore is a generalized mutex: it issues a fixed number of permits. An effect wrapped with withPermits(k) cannot run until k permits are free, and the permits are returned when the effect completes, on success, failure, or interruption.

Use it to bound the concurrency of a resource: connection pools, rate-limited APIs, mutexes.

What you’ll work through

index.test.ts walks three beats: constructing a semaphore with Semaphore.make, gating a single effect with withPermits(1), and combining the semaphore with Effect.all + concurrency: "unbounded" to cap a pool of concurrent workers.

By the end you should be able to bound any resource — a connection pool, a rate-limited API, a mutex — without writing release logic, because permits return automatically on success, failure, and interruption.

Why this matters. Permits return on interruption thanks to the finalizer machinery in step 37. A semaphore cannot deadlock from a fiber that died holding a permit.

Concept

1
makeSemaphore(N)          N permits
2

3
  withPermits(1)(effect)  borrow 1 permit, run effect, return permit
4
  withPermits(k)(effect)  borrow k permits, run effect, return permits
5

6
permits = 3:
7
  [✓][✓][✓]            three active tasks
8
   ◀──── 1 finishes  one permit returned; the next waiter resumes

A new permit is returned automatically when the effect ends, including when it is interrupted. You cannot leak a permit by forgetting to release it.

`Semaphore.make` — the mutex case

Semaphore.make(n) builds a semaphore holding n permits and returns it as an Effect<Semaphore>. The most common shape is n = 1 — a mutex — wrapped around a critical section with sem.withPermits(1)(...) so only one fiber at a time runs the guarded effect.

Permits are returned automatically on success, failure, and interruption.

1
import { Effect } from "effect"
2

3
const program = Effect.gen(function* () {
4
  //      ┌─── Effect.Semaphore
5
  //      ▼
6
  const sem = yield* Semaphore.make(1)   // 1 permit = mutex
7

8
  // Two fibers race for the critical section; only one runs at a time.
9
  yield* Effect.all(
10
    [criticalSection, criticalSection].map((eff) => sem.withPermits(1)(eff)),
11
    { concurrency: "unbounded" },
12
  )
13
})

Rule of thumb. Build the semaphore once — at service initialization or at the top of a scoped program — and share it. A semaphore constructed inside the guarded function is a new semaphore on every call and enforces nothing.

Your turn build a 1-permit semaphore, wrap two concurrent fibers with `sem.withPermits(1)(...)`, and assert their critical sections never overlap.

`semaphore.withPermits`

Wraps an effect so that it can only run when k permits are available.

1
//      ┌─── Effect<string>
2
//      ▼
3
const program = Effect.gen(function* () {
4
  const sem = yield* Semaphore.make(1)
5
  const critical = sem.withPermits(1)(Effect.succeed("inside"))
6
  return yield* critical
7
})

Details

withPermits(k) accepts an Effect<A, E, R> and returns an Effect<A, E, R>. Permits are acquired in FIFO order; pending callers form a queue.

When to Use

withPermits(1) on a 1-permit semaphore, a mutex.
withPermits(1) on an N-permit semaphore, connection pool, rate limit.
withPermits(k) for tasks that occupy multiple slots (e.g. an expensive operation that reserves two connections).

Composite Example: Connection Pool Cap

1
//      ┌─── Effect<ReadonlyArray<number>>
2
//      ▼
3
const program = Effect.gen(function* () {
4
  const pool = yield* Semaphore.make(5)
5
  const query = (id: number) =>
6
    pool.withPermits(1)(
7
      Effect.sleep("5 millis").pipe(Effect.as(id * 2)),
8
    )
9

10
  return yield* Effect.all(
11
    Array.from({ length: 20 }, (_, i) => query(i)),
12
    { concurrency: "unbounded" },
13
  )
14
})
15
// 20 results, but at most 5 queries were in flight at any moment.

The Effect.all concurrency is unbounded, but the semaphore is the real cap.

Bounding a batch with `Semaphore.make`

The standard pattern for bounding a batch of work — a pool of HTTP requests, parallel DB writes, concurrent calls to a rate-limited API — is Semaphore.make(N) paired with withPermits(1) around each unit of work and Effect.all with concurrency: "unbounded".

1
import { Effect } from "effect"
2

3
const program = Effect.gen(function* () {
4
  const sem = yield* Semaphore.make(3) // pool size 3
5

6
  return yield* Effect.all(
7
    requests.map((req) => sem.withPermits(1)(fetchOne(req))),
8
    { concurrency: "unbounded" },
9
  )
10
})

Effect.all would normally fan out all requests at once. The semaphore is the real cap — at most 3 of them are actually in flight; the rest wait for a permit.

Permits are returned automatically on success, failure, and interruption.

Your turn wrap each item in `sem.withPermits(1)(...)` and confirm the observed in-flight count never exceeds the semaphore size, even with unbounded outer concurrency.

Rules of thumb

One semaphore per resource pool. Don’t share permits across logically distinct workloads — they will contend for the wrong reason.
withPermits(1) is the common case. Higher counts weight a task more heavily within the same pool.
Pair with Effect.all + unbounded concurrency. The semaphore does the bounding; let the scheduler hand out work eagerly.

Common Pitfalls

1, Constructing the semaphore inside the guarded function. A new semaphore on every call enforces nothing. Build it once at service initialization and share it.

2, Using a semaphore where a queue would be clearer. A semaphore bounds active work. A queue buffers pending work.

If producers can outpace consumers, you want a queue (possibly fed into a semaphore-bounded worker pool).

3, Asking for more permits than the semaphore holds. withPermits(k) with k greater than the total permits will suspend forever. Validate at construction time.

4, Forgetting that permits are returned on interruption. This is a feature, not a risk: a fiber interrupted while holding a permit returns it before it dies. No deadlocks from forgotten releases.

access — the test asserts ordering, not just correctness. - [ ] Scale up to an N-permit semaphore and observe that at most N effects run concurrently even when the surrounding `Effect.all` uses `concurrency: "unbounded"`. - [ ] Confirm permits are returned on interruption — the semaphore is the cap, the scheduler is the firehose.

The full test suite

The index.test.ts for this lesson, verbatim from steps/38-semaphore/.

1
import { describe, it, expect } from "vite-plus/test"
2
import { Effect, Pool, Ref, Semaphore } from "effect"
3

4
// =============================================================================
5
// 41 - Semaphore: Bounded Concurrency
6
// =============================================================================
7
//
8
// Semaphore.make(N)       Effect<Semaphore>
9
// semaphore.withPermits(k)(eff) borrow k permits for the duration of eff
10

11
describe("41 - Semaphore", () => {
12
  // @beat
13
  //   id: mutex
14
  //   shape: one-todo
15
  //   title: One-permit semaphore is a mutex
16
  //   teaches: Semaphore.make(1) returns a Semaphore; semaphore.withPermits(1)(eff) serializes the critical section.
17
  //   hint: Semaphore.make(1), then mutex.withPermits(1)(criticalSection)
18
  describe("withPermits(1) — mutex", () => {
19
    it("serializes the critical section", async () => {
20
      const program = /* @stub */ Effect.gen(function* () {
21
        const log = yield* Ref.make<ReadonlyArray<number>>([])
22
        const mutex = yield* Semaphore.make(1)
23

24
        const task = (id: number) =>
25
          mutex.withPermits(1)(
26
            Effect.gen(function* () {
27
              yield* Ref.update(log, (arr) => [...arr, id])
28
              yield* Effect.sleep("5 millis")
29
              yield* Ref.update(log, (arr) => [...arr, -id])
30
            }),
31
          )
32

33
        yield* Effect.all([task(1), task(2), task(3)], {
34
          concurrency: "unbounded",
35
        })
36
        return yield* Ref.get(log)
37
      }) /* @end */
38

39
      const events = await Effect.runPromise(program)
40
      // Every positive id is immediately followed by its negative — no interleaving.
41
      expect(events.length).toBe(6)
42
      const active = new Set<number>()
43
      for (const entry of events) {
44
        if (entry > 0) {
45
          expect(active.size).toBe(0)
46
          active.add(entry)
47
        } else {
48
          active.delete(-entry)
49
        }
50
      }
51
    })
52
  })
53

54
  // @end-beat
55

56
  // @beat
57
  //   id: pool-cap
58
  //   shape: one-todo
59
  //   title: N-permit semaphore caps active concurrency
60
  //   teaches: Semaphore.make(N) issues N permits; withPermits(1) on it bounds simultaneous workers to N.
61
  //   hint: Semaphore.make(3) then sem.withPermits(1)(task) — at most 3 tasks run at once
62
  describe("withPermits(1) — N permits", () => {
63
    it("caps active tasks to the permit count", async () => {
64
      const program = /* @stub */ Effect.gen(function* () {
65
        const active = yield* Ref.make(0)
66
        const peak = yield* Ref.make(0)
67
        const sem = yield* Semaphore.make(3)
68

69
        const task = sem.withPermits(1)(
70
          Effect.gen(function* () {
71
            yield* Ref.update(active, (n) => n + 1)
72
            const current = yield* Ref.get(active)
73
            yield* Ref.update(peak, (m) => Math.max(m, current))
74
            yield* Effect.sleep("10 millis")
75
            yield* Ref.update(active, (n) => n - 1)
76
          }),
77
        )
78

79
        yield* Effect.all(Array.from({ length: 6 }, () => task), {
80
          concurrency: "unbounded",
81
        })
82
        return yield* Ref.get(peak)
83
      }) /* @end */
84

85
      const peak = await Effect.runPromise(program)
86
      expect(peak).toBeLessThanOrEqual(3)
87
      expect(peak).toBeGreaterThan(0)
88
    })
89
  })
90
  // @end-beat
91

92
  // @beat
93
  //   id: multi-permit
94
  //   shape: multi-todo
95
  //   title: withPermits(k) reserves multiple slots
96
  //   teaches: A task that occupies multiple slots calls withPermits(k); permits are returned even on interruption.
97
  //   hint: predict the two log entries pushed by `heavy` and `light` — both run and both record
98
  describe("withPermits(k) — multi-permit tasks", () => {
99
    it("allows tasks to reserve multiple permits", async () => {
100
      const program = Effect.gen(function* () {
101
        const sem = yield* Semaphore.make(2)
102
        const log: Array<string> = []
103

104
        const heavy = sem.withPermits(2)(Effect.sync(() => log.push("heavy")))
105
        const light = sem.withPermits(1)(Effect.sync(() => log.push("light")))
106

107
        yield* Effect.all([heavy, light], { concurrency: "unbounded" })
108
        return log
109
      })
110

111
      const log = await Effect.runPromise(program)
112
      // Predict the two strings the tasks pushed into log
113
      expect(log).toContain(/* @stub */ "heavy" /* @end */)
114
      expect(log).toContain(/* @stub */ "light" /* @end */)
115
    })
116
  })
117
  // @end-beat
118

119
  describe("Connection pool cap", () => {
120
    it("never exceeds the pool size under high concurrency", async () => {
121
      const program = Effect.gen(function* () {
122
        const POOL_SIZE = 5
123
        const pool = yield* Semaphore.make(POOL_SIZE)
124
        const active = yield* Ref.make(0)
125
        const peak = yield* Ref.make(0)
126

127
        const query = (id: number) =>
128
          pool.withPermits(1)(
129
            Effect.gen(function* () {
130
              yield* Ref.update(active, (n) => n + 1)
131
              const current = yield* Ref.get(active)
132
              yield* Ref.update(peak, (m) => Math.max(m, current))
133
              yield* Effect.sleep("5 millis")
134
              yield* Ref.update(active, (n) => n - 1)
135
              return id * 2
136
            }),
137
          )
138

139
        yield* Effect.all(
140
          Array.from({ length: 20 }, (_, i) => query(i)),
141
          { concurrency: "unbounded" },
142
        )
143
        return yield* Ref.get(peak)
144
      })
145

146
      expect(await Effect.runPromise(program)).toBeLessThanOrEqual(5)
147
    })
148
  })
149

150
  // @beat
151
  //   id: limit-batch
152
  //   shape: one-todo
153
  //   title: Cap a parallel batch to N concurrent workers
154
  //   teaches: Semaphore.make(N) + withPermits(1) is the standard pattern for bounding a pool — use it whenever you have more work than your downstream (a DB, an API) can handle concurrently.
155
  //   hint: yield* Semaphore.make(2), then sem.withPermits(1)(task) wraps the work; assert peak never exceeds 2
156

157
  describe("bounded batch with semaphore", () => {
158
    it("processes 8 tasks with at most 2 running at once", async () => {
159
      let inFlight = 0
160
      let peak = 0
161

162
      const program = /* @stub */ Effect.gen(function* () {
163
        const sem = yield* Semaphore.make(2)
164

165
        const task = (id: number) =>
166
          sem.withPermits(1)(
167
            Effect.gen(function* () {
168
              inFlight++
169
              peak = Math.max(peak, inFlight)
170
              yield* Effect.sleep("5 millis")
171
              inFlight--
172
              return id
173
            }),
174
          )
175

176
        return yield* Effect.all(
177
          Array.from({ length: 8 }, (_, i) => task(i)),
178
          { concurrency: "unbounded" },
179
        )
180
      }) /* @end */
181

182
      const results = await Effect.runPromise(program)
183
      expect(results).toHaveLength(8)
184
      expect(peak).toBeLessThanOrEqual(2)
185
      expect(peak).toBeGreaterThan(0)
186
    })
187
  })
188
  // @end-beat
189
})
190

191
// @beat
192
//   id: scoped-resource-pool
193
//   shape: read-run
194
//   title: Reuse a bounded set of scoped resources
195
//   teaches: Pool bounds expensive resources and releases them with Scope; use Semaphore for permits and Pool when the permits represent reusable resources.
196
//   hint: Pool.make takes an acquireRelease effect and a maximum size; Pool.get borrows within the surrounding scope.
197
describe("Pool", () => {
198
  it("reuses a bounded set of scoped resources", async () => {
199
    const program = Effect.scoped(Effect.gen(function*() {
200
      const allocations = yield* Ref.make(0)
201
      const pool = yield* Pool.make({
202
        acquire: Effect.acquireRelease(
203
          Ref.updateAndGet(allocations, (n) => n + 1),
204
          () => Effect.void,
205
        ),
206
        size: 2,
207
      })
208
      const resource = yield* Pool.get(pool)
209
      return { resource, allocations: yield* Ref.get(allocations) }
210
    }))
211
    const result = await Effect.runPromise(program)
212
    expect(result.resource).toBeGreaterThan(0)
213
    expect(result.allocations).toBeLessThanOrEqual(2)
214
  })
215
})
216
// @end-beat

07·04Semaphore

38 - Semaphore: Bounded Concurrency

What you’ll work through

Concept

`Semaphore.make` — the mutex case

`semaphore.withPermits`

Composite Example: Connection Pool Cap

Bounding a batch with `Semaphore.make`

Rules of thumb

Common Pitfalls

See Also

The full test suite

07·04Semaphore

38 - Semaphore: Bounded Concurrency

What you’ll work through

Concept

Semaphore.make — the mutex case

semaphore.withPermits

Composite Example: Connection Pool Cap

Bounding a batch with Semaphore.make

Rules of thumb

Common Pitfalls

See Also

The full test suite

`Semaphore.make` — the mutex case

`semaphore.withPermits`

Bounding a batch with `Semaphore.make`