Semaphore

A semaphore is a synchronization mechanism used to manage access to a shared resource. In Effect, semaphores help control resource access or coordinate tasks within asynchronous, concurrent operations.

A semaphore acts as a generalized mutex, allowing a set number of permits to be held and released concurrently. Permits act like tickets, giving tasks or fibers controlled access to a shared resource. When no permits are available, tasks trying to acquire one will wait until a permit is released.

Creating a Semaphore

The Semaphore.make function initializes a semaphore with a specified number of permits. Each permit allows one task to access a resource or perform an operation concurrently, and multiple permits enable a configurable level of concurrency.

Example (Creating a Semaphore with 3 Permits)

1
import { Effect, Semaphore } from "effect"
2

3
// Create a semaphore with 3 permits
4
const mutex = Semaphore.make(3)
5

6
// The semaphore was created with exactly 3 permits available
7
const acquired = await Effect.runPromise(
8
  mutex.pipe(Effect.flatMap((sem) => Semaphore.take(sem, 3))),
9
)
10
acquired // => 3

withPermits

The withPermits method lets you specify the number of permits required to run an effect. Once the specified permits are available, it runs the effect, automatically releasing the permits when the task completes.

Example (Forcing Sequential Task Execution with a One-Permit Semaphore)

In this example, three tasks are started concurrently, but they run sequentially because the one-permit semaphore only allows one task to proceed at a time.

1
import { Effect, Semaphore } from "effect"
2

3
const task = Effect.gen(function* () {
4
  yield* Effect.log("start")
5
  yield* Effect.sleep("2 seconds")
6
  yield* Effect.log("end")
7
})
8

9
const program = Effect.gen(function* () {
10
  const mutex = yield* Semaphore.make(1)
11

12
  // Wrap the task to require one permit, forcing sequential execution
13
  const semTask = mutex.withPermits(1)(task).pipe(Effect.withLogSpan("elapsed"))
14

15
  // Run 3 tasks concurrently, but they execute sequentially
16
  // due to the one-permit semaphore
17
  yield* Effect.all([semTask, semTask, semTask], {
18
    concurrency: "unbounded",
19
  })
20
})
21

22
await Effect.runPromise(program) // => undefined
23
/*
24
Output:
25
timestamp=... level=INFO fiber=#1 message=start elapsed=3ms
26
timestamp=... level=INFO fiber=#1 message=end elapsed=2010ms
27
timestamp=... level=INFO fiber=#2 message=start elapsed=2012ms
28
timestamp=... level=INFO fiber=#2 message=end elapsed=4017ms
29
timestamp=... level=INFO fiber=#3 message=start elapsed=4018ms
30
timestamp=... level=INFO fiber=#3 message=end elapsed=6026ms
31
*/

Example (Using Multiple Permits to Control Concurrent Task Execution)

In this example, we create a semaphore with five permits and use withPermits(n) to allocate a different number of permits for each task:

1
import { Effect, Semaphore } from "effect"
2

3
const program = Effect.gen(function* () {
4
  const mutex = yield* Semaphore.make(5)
5

6
  const tasks = [1, 2, 3, 4, 5].map((n) =>
7
    mutex
8
      .withPermits(n)(Effect.delay(Effect.log(`process: ${n}`), "2 seconds"))
9
      .pipe(Effect.withLogSpan("elapsed")),
10
  )
11

12
  yield* Effect.all(tasks, { concurrency: "unbounded" })
13
})
14

15
await Effect.runPromise(program) // => undefined
16
/*
17
Output:
18
timestamp=... level=INFO fiber=#1 message="process: 1" elapsed=2011ms
19
timestamp=... level=INFO fiber=#2 message="process: 2" elapsed=2017ms
20
timestamp=... level=INFO fiber=#3 message="process: 3" elapsed=4020ms
21
timestamp=... level=INFO fiber=#4 message="process: 4" elapsed=6025ms
22
timestamp=... level=INFO fiber=#5 message="process: 5" elapsed=8034ms
23
*/