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 Effect.makeSemaphore 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 { import Effect
Effect } from "effect"
2

3
// Create a semaphore with 3 permits
4
const const mutex: Effect.Effect<Effect.Semaphore, never, never>
mutex = import Effect
Effect.const makeSemaphore: (permits: number) => Effect.Effect<Effect.Semaphore>Creates a new Semaphore
makeSemaphore(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 { import Effect
Effect } from "effect"
2

3
const const task: Effect.Effect<void, never, never>
task = import Effect
Effect.const gen: <YieldWrap<Effect.Effect<void, never, never>>, void>(f: (resume: Effect.Adapter) => Generator<YieldWrap<Effect.Effect<void, never, never>>, void, never>) => Effect.Effect<...> (+1 overload)
gen(function* () {
4
  yield* import Effect
Effect.const log: (...message: ReadonlyArray<any>) => Effect.Effect<void, never, never>Logs one or more messages or error causes at the current log level, which is INFO by default.
This function allows logging multiple items at once and can include detailed error information using `Cause` instances.

To adjust the log level, use the `Logger.withMinimumLogLevel` function.
log("start")
5
  yield* import Effect
Effect.const sleep: (duration: DurationInput) => Effect.Effect<void>Returns an effect that suspends for the specified duration. This method is
asynchronous, and does not actually block the fiber executing the effect.
sleep("2 seconds")
6
  yield* import Effect
Effect.const log: (...message: ReadonlyArray<any>) => Effect.Effect<void, never, never>Logs one or more messages or error causes at the current log level, which is INFO by default.
This function allows logging multiple items at once and can include detailed error information using `Cause` instances.

To adjust the log level, use the `Logger.withMinimumLogLevel` function.
log("end")
7
})
8

9
const const program: Effect.Effect<void, never, never>
program = import Effect
Effect.const gen: <YieldWrap<Effect.Effect<Effect.Semaphore, never, never>> | YieldWrap<Effect.Effect<[void, void, void], never, never>>, void>(f: (resume: Effect.Adapter) => Generator<...>) => Effect.Effect<...> (+1 overload)
gen(function* () {
10
  const const mutex: Effect.Semaphore
mutex = yield* import Effect
Effect.const makeSemaphore: (permits: number) => Effect.Effect<Effect.Semaphore>Creates a new Semaphore
makeSemaphore(1)
11

12
  // Wrap the task to require one permit, forcing sequential execution
13
  const const semTask: Effect.Effect<void, never, never>
semTask = const mutex: Effect.Semaphore
mutex
14
    .(method) Semaphore.withPermits(permits: number): <A, E, R>(self: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R>when the given amount of permits are available, run the effect and release the permits when finished
withPermits(1)(const task: Effect.Effect<void, never, never>
task)
15
    .(method) Pipeable.pipe<Effect.Effect<void, never, never>, Effect.Effect<void, never, never>>(this: Effect.Effect<...>, ab: (_: Effect.Effect<void, never, never>) => Effect.Effect<void, never, never>): Effect.Effect<...> (+21 overloads)
pipe(import Effect
Effect.const withLogSpan: (label: string) => <A, E, R>(effect: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R> (+1 overload)Adds a log span to your effects, which tracks and logs the duration of
operations or tasks. This is useful for performance monitoring and debugging
time-sensitive processes.
withLogSpan("elapsed"))
16

17
  // Run 3 tasks concurrently, but they execute sequentially
18
  // due to the one-permit semaphore
19
  yield* import Effect
Effect.const all: <readonly [Effect.Effect<void, never, never>, Effect.Effect<void, never, never>, Effect.Effect<void, never, never>], {
    concurrency: "unbounded";
}>(arg: readonly [Effect.Effect<void, never, never>, Effect.Effect<...>, Effect.Effect<...>], options?: {
    ...;
} | undefined) => Effect.Effect<...>Runs all the provided effects in sequence respecting the structure provided in input.

Supports multiple arguments, a single argument tuple / array or record / struct.
all([const semTask: Effect.Effect<void, never, never>
semTask, const semTask: Effect.Effect<void, never, never>
semTask, const semTask: Effect.Effect<void, never, never>
semTask], {
20
    (property) concurrency: "unbounded"
concurrency: "unbounded"
21
  })
22
})
23

24
import Effect
Effect.const runFork: <void, never>(effect: Effect.Effect<void, never, never>, options?: RunForkOptions) => RuntimeFiber<void, never>Executes an effect and returns a `RuntimeFiber` that represents the running computation.

Use `runFork` when you want to start an effect without blocking the current execution flow.
It returns a fiber that you can observe, interrupt, or join as needed.
runFork(const program: Effect.Effect<void, never, never>
program)
25
/*
26
Output:
27
timestamp=... level=INFO fiber=#1 message=start elapsed=3ms
28
timestamp=... level=INFO fiber=#1 message=end elapsed=2010ms
29
timestamp=... level=INFO fiber=#2 message=start elapsed=2012ms
30
timestamp=... level=INFO fiber=#2 message=end elapsed=4017ms
31
timestamp=... level=INFO fiber=#3 message=start elapsed=4018ms
32
timestamp=... level=INFO fiber=#3 message=end elapsed=6026ms
33
*/

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 { import Effect
Effect } from "effect"
2

3
const const program: Effect.Effect<void, never, never>
program = import Effect
Effect.const gen: <YieldWrap<Effect.Effect<Effect.Semaphore, never, never>> | YieldWrap<Effect.Effect<void[], never, never>>, void>(f: (resume: Effect.Adapter) => Generator<...>) => Effect.Effect<...> (+1 overload)
gen(function* () {
4
  const const mutex: Effect.Semaphore
mutex = yield* import Effect
Effect.const makeSemaphore: (permits: number) => Effect.Effect<Effect.Semaphore>Creates a new Semaphore
makeSemaphore(5)
5

6
  const const tasks: Effect.Effect<void, never, never>[]
tasks = [1, 2, 3, 4, 5].(method) Array<number>.map<Effect.Effect<void, never, never>>(callbackfn: (value: number, index: number, array: number[]) => Effect.Effect<void, never, never>, thisArg?: any): Effect.Effect<void, never, never>[]Calls a defined callback function on each element of an array, and returns an array that contains the results.
map(((parameter) n: number
n) =>
7
    const mutex: Effect.Semaphore
mutex
8
      .(method) Semaphore.withPermits(permits: number): <A, E, R>(self: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R>when the given amount of permits are available, run the effect and release the permits when finished
withPermits((parameter) n: number
n)(
9
        import Effect
Effect.const delay: <void, never, never>(self: Effect.Effect<void, never, never>, duration: DurationInput) => Effect.Effect<void, never, never> (+1 overload)Returns an effect that is delayed from this effect by the specified
`Duration`.
delay(import Effect
Effect.const log: (...message: ReadonlyArray<any>) => Effect.Effect<void, never, never>Logs one or more messages or error causes at the current log level, which is INFO by default.
This function allows logging multiple items at once and can include detailed error information using `Cause` instances.

To adjust the log level, use the `Logger.withMinimumLogLevel` function.
log(`process: ${(parameter) n: number
n}`), "2 seconds")
10
      )
11
      .(method) Pipeable.pipe<Effect.Effect<void, never, never>, Effect.Effect<void, never, never>>(this: Effect.Effect<...>, ab: (_: Effect.Effect<void, never, never>) => Effect.Effect<void, never, never>): Effect.Effect<...> (+21 overloads)
pipe(import Effect
Effect.const withLogSpan: (label: string) => <A, E, R>(effect: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R> (+1 overload)Adds a log span to your effects, which tracks and logs the duration of
operations or tasks. This is useful for performance monitoring and debugging
time-sensitive processes.
withLogSpan("elapsed"))
12
  )
13

14
  yield* import Effect
Effect.const all: <Effect.Effect<void, never, never>[], {
    concurrency: "unbounded";
}>(arg: Effect.Effect<void, never, never>[], options?: {
    concurrency: "unbounded";
} | undefined) => Effect.Effect<...>Runs all the provided effects in sequence respecting the structure provided in input.

Supports multiple arguments, a single argument tuple / array or record / struct.
all(const tasks: Effect.Effect<void, never, never>[]
tasks, { (property) concurrency: "unbounded"
concurrency: "unbounded" })
15
})
16

17
import Effect
Effect.const runFork: <void, never>(effect: Effect.Effect<void, never, never>, options?: RunForkOptions) => RuntimeFiber<void, never>Executes an effect and returns a `RuntimeFiber` that represents the running computation.

Use `runFork` when you want to start an effect without blocking the current execution flow.
It returns a fiber that you can observe, interrupt, or join as needed.
runFork(const program: Effect.Effect<void, never, never>
program)
18
/*
19
Output:
20
timestamp=... level=INFO fiber=#1 message="process: 1" elapsed=2011ms
21
timestamp=... level=INFO fiber=#2 message="process: 2" elapsed=2017ms
22
timestamp=... level=INFO fiber=#3 message="process: 3" elapsed=4020ms
23
timestamp=... level=INFO fiber=#4 message="process: 4" elapsed=6025ms
24
timestamp=... level=INFO fiber=#5 message="process: 5" elapsed=8034ms
25
*/