Basic Concurrency

Concurrency Options

Effect provides options to manage how effects are executed, particularly focusing on controlling how many effects run concurrently.

type Options = {
  readonly concurrency?: Concurrency
}

The concurrency option is used to determine the level of concurrency, with the following values:

type Concurrency = number | "unbounded"

Let’s explore each configuration in detail.

Sequential Execution (Default)

By default, if you don’t specify any concurrency option, effects will run sequentially, one after the other. This means each effect starts only after the previous one completes.

Example (Sequential Execution)

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

3
// Helper function to simulate a task with a delay
4
const makeTask = (n: number, delay: Duration.Input) =>
5
  Effect.promise(
6
    () =>
7
      new Promise<void>((resolve) => {
8
        console.log(`start task${n}`) // Logs when the task starts
9
        setTimeout(() => {
10
          console.log(`task${n} done`) // Logs when the task finishes
11
          resolve()
12
        }, Duration.toMillis(delay))
13
      }),
14
  )
15

16
const task1 = makeTask(1, "200 millis")
17
const task2 = makeTask(2, "100 millis")
18

19
const sequential = Effect.all([task1, task2])
20

21
await Effect.runPromise(sequential) // => [undefined, undefined]
22
/*
23
Output:
24
start task1
25
task1 done
26
start task2 <-- task2 starts only after task1 completes
27
task2 done
28
*/

Numbered Concurrency

You can control how many effects run concurrently by setting a number for concurrency. For example, concurrency: 2 allows up to two effects to run at the same time.

Example (Limiting to 2 Concurrent Tasks)

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

3
// Helper function to simulate a task with a delay
4
const makeTask = (n: number, delay: Duration.Input) =>
5
  Effect.promise(
6
    () =>
7
      new Promise<void>((resolve) => {
8
        console.log(`start task${n}`) // Logs when the task starts
9
        setTimeout(() => {
10
          console.log(`task${n} done`) // Logs when the task finishes
11
          resolve()
12
        }, Duration.toMillis(delay))
13
      }),
14
  )
15

16
const task1 = makeTask(1, "200 millis")
17
const task2 = makeTask(2, "100 millis")
18
const task3 = makeTask(3, "210 millis")
19
const task4 = makeTask(4, "110 millis")
20
const task5 = makeTask(5, "150 millis")
21

22
const numbered = Effect.all([task1, task2, task3, task4, task5], {
23
  concurrency: 2,
24
})
25

26
await Effect.runPromise(numbered) // => [undefined, undefined, undefined, undefined, undefined]
27
/*
28
Output:
29
start task1
30
start task2 <-- active tasks: task1, task2
31
task2 done
32
start task3 <-- active tasks: task1, task3
33
task1 done
34
start task4 <-- active tasks: task3, task4
35
task4 done
36
start task5 <-- active tasks: task3, task5
37
task3 done
38
task5 done
39
*/

Unbounded Concurrency

When concurrency: "unbounded" is used, there’s no limit to the number of effects running concurrently.

Example (Unbounded Concurrency)

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

3
// Helper function to simulate a task with a delay
4
const makeTask = (n: number, delay: Duration.Input) =>
5
  Effect.promise(
6
    () =>
7
      new Promise<void>((resolve) => {
8
        console.log(`start task${n}`) // Logs when the task starts
9
        setTimeout(() => {
10
          console.log(`task${n} done`) // Logs when the task finishes
11
          resolve()
12
        }, Duration.toMillis(delay))
13
      }),
14
  )
15

16
const task1 = makeTask(1, "200 millis")
17
const task2 = makeTask(2, "100 millis")
18
const task3 = makeTask(3, "210 millis")
19
const task4 = makeTask(4, "110 millis")
20
const task5 = makeTask(5, "150 millis")
21

22
const unbounded = Effect.all([task1, task2, task3, task4, task5], {
23
  concurrency: "unbounded",
24
})
25

26
await Effect.runPromise(unbounded) // => [undefined, undefined, undefined, undefined, undefined]
27
/*
28
Output:
29
start task1
30
start task2
31
start task3
32
start task4
33
start task5
34
task2 done
35
task4 done
36
task5 done
37
task1 done
38
task3 done
39
*/

Interruptions

All effects in Effect are executed by fibers. If you didn’t create the fiber yourself, it was created by an operation you’re using (if it’s concurrent) or by the Effect runtime system.

A fiber is created any time an effect is run. When running effects concurrently, a fiber is created for each concurrent effect.

To summarize:

An Effect is a higher-level concept that describes an effectful computation. It is lazy and immutable, meaning it represents a computation that may produce a value or fail but does not immediately execute.
A fiber, on the other hand, represents the running execution of an Effect. It can be interrupted or awaited to retrieve its result. Think of it as a way to control and interact with the ongoing computation.

Fibers can be interrupted in various ways. Let’s explore some of these scenarios and see examples of how to interrupt fibers in Effect.

interrupt

A fiber can be interrupted using the Effect.interrupt effect on that particular fiber.

This effect models the explicit interruption of the fiber in which it runs. When executed, it causes the fiber to stop its operation immediately, capturing the interruption details such as the fiber’s ID and its start time. The resulting interruption can be observed in the Exit type if the effect is run with functions like runPromiseExit.

Example (Without Interruption)

In this case, the program runs without any interruption, logging the start and completion of the task.

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

3
const program = Effect.gen(function* () {
4
  console.log("start")
5
  yield* Effect.sleep("2 seconds")
6
  console.log("done")
7
  return "some result"
8
})
9

10
await Effect.runPromiseExit(program) // => Exit.succeed("some result")
11
/*
12
Output:
13
start
14
done
15
*/

Example (With Interruption)

Here, the fiber is interrupted after the log "start" but before the "done" log. The Effect.interrupt stops the fiber, and it never reaches the final log.

1
import { Effect } from "effect"
2

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

9
const exit = await Effect.runPromiseExit(program)
10
exit._tag // => "Failure"
11
/*
12
Output:
13
start
14
{
15
  _id: 'Exit',
16
  _tag: 'Failure',
17
  cause: {
18
    _id: 'Cause',
19
    _tag: 'Interrupt',
20
    fiberId: {
21
      _id: 'FiberId',
22
      _tag: 'Runtime',
23
      id: 0,
24
      startTimeMillis: ...
25
    }
26
  }
27
}
28
*/

onInterrupt

Registers a cleanup effect to run when an effect is interrupted.

This function allows you to specify an effect to run when the fiber is interrupted. This effect will be executed when the fiber is interrupted, allowing you to perform cleanup or other actions.

Example (Running a Cleanup Action on Interruption)

In this example, we set up a handler that logs “Cleanup completed” whenever the fiber is interrupted. We then show three cases: a successful effect, a failing effect, and an interrupted effect, demonstrating how the handler is triggered depending on how the effect ends.

1
import { Console, Effect, Exit } from "effect"
2

3
// This handler is executed when the fiber is interrupted
4
const handler = Effect.onInterrupt((_fibers) =>
5
  Console.log("Cleanup completed"),
6
)
7

8
const success = Console.log("Task completed").pipe(
9
  Effect.as("some result"),
10
  handler,
11
)
12

13
await Effect.runPromise(success) // => "some result"
14
/*
15
Output:
16
Task completed
17
*/
18

19
const failure = Console.log("Task failed").pipe(
20
  Effect.andThen(Effect.fail("some error")),
21
  handler,
22
)
23

24
await Effect.runPromiseExit(failure) // => Exit.fail("some error")
25
/*
26
Output:
27
Task failed
28
*/
29

30
const interruption = Console.log("Task interrupted").pipe(
31
  Effect.andThen(Effect.interrupt),
32
  handler,
33
)
34

35
const interruptionExit = await Effect.runPromiseExit(interruption)
36
interruptionExit._tag // => "Failure"
37
/*
38
Output:
39
Task interrupted
40
Cleanup completed
41
*/

Interruption of Concurrent Effects

When running multiple effects concurrently, such as with Effect.forEach, if one of the effects is interrupted, it causes all concurrent effects to be interrupted as well.

The resulting cause includes information about which fibers were interrupted.

Example (Interrupting Concurrent Effects)

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

3
const program = Effect.forEach(
4
  [1, 2, 3],
5
  (n) =>
6
    Effect.gen(function* () {
7
      console.log(`start #${n}`)
8
      yield* Effect.sleep(`${n} seconds`)
9
      if (n > 1) {
10
        return yield* Effect.interrupt
11
      }
12
      console.log(`done #${n}`)
13
    }).pipe(Effect.onInterrupt(() => Console.log(`interrupted #${n}`))),
14
  { concurrency: "unbounded" },
15
)
16

17
const exit = await Effect.runPromiseExit(program)
18
console.log(JSON.stringify(exit, null, 2))
19
exit._tag // => "Failure"
20
/*
21
Output:
22
start #1
23
start #2
24
start #3
25
done #1
26
interrupted #2
27
interrupted #3
28
{
29
  "_id": "Exit",
30
  "_tag": "Failure",
31
  "cause": {
32
    "_id": "Cause",
33
    "_tag": "Parallel",
34
    "left": {
35
      "_id": "Cause",
36
      "_tag": "Interrupt",
37
      "fiberId": {
38
        "_id": "FiberId",
39
        "_tag": "Runtime",
40
        "id": 3,
41
        "startTimeMillis": ...
42
      }
43
    },
44
    "right": {
45
      "_id": "Cause",
46
      "_tag": "Sequential",
47
      "left": {
48
        "_id": "Cause",
49
        "_tag": "Empty"
50
      },
51
      "right": {
52
        "_id": "Cause",
53
        "_tag": "Interrupt",
54
        "fiberId": {
55
          "_id": "FiberId",
56
          "_tag": "Runtime",
57
          "id": 0,
58
          "startTimeMillis": ...
59
        }
60
      }
61
    }
62
  }
63
}
64
*/

Racing

race

This function takes two effects and runs them concurrently. The first effect that successfully completes will determine the result of the race, and the other effect will be interrupted.

If neither effect succeeds, the function will fail with a cause containing all the errors.

This is useful when you want to run two effects concurrently, but only care about the first one to succeed. It is commonly used in cases like timeouts, retries, or when you want to optimize for the faster response without worrying about the other effect.

Example (Both Tasks Succeed)

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

3
const task1 = Effect.succeed("task1").pipe(
4
  Effect.delay("200 millis"),
5
  Effect.tap(Console.log("task1 done")),
6
  Effect.onInterrupt(() => Console.log("task1 interrupted")),
7
)
8
const task2 = Effect.succeed("task2").pipe(
9
  Effect.delay("100 millis"),
10
  Effect.tap(Console.log("task2 done")),
11
  Effect.onInterrupt(() => Console.log("task2 interrupted")),
12
)
13

14
const program = Effect.race(task1, task2)
15

16
await Effect.runPromise(program) // => "task2"
17
/*
18
Output:
19
task2 done
20
task1 interrupted
21
*/

Example (One Task Fails, One Succeeds)

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

3
const task1 = Effect.fail("task1").pipe(
4
  Effect.delay("100 millis"),
5
  Effect.tap(Console.log("task1 done")),
6
  Effect.onInterrupt(() => Console.log("task1 interrupted")),
7
)
8
const task2 = Effect.succeed("task2").pipe(
9
  Effect.delay("200 millis"),
10
  Effect.tap(Console.log("task2 done")),
11
  Effect.onInterrupt(() => Console.log("task2 interrupted")),
12
)
13

14
const program = Effect.race(task1, task2)
15

16
await Effect.runPromise(program) // => "task2"
17
/*
18
Output:
19
task2 done
20
*/

Example (Both Tasks Fail)

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

3
const task1 = Effect.fail("task1").pipe(
4
  Effect.delay("100 millis"),
5
  Effect.tap(Console.log("task1 done")),
6
  Effect.onInterrupt(() => Console.log("task1 interrupted")),
7
)
8
const task2 = Effect.fail("task2").pipe(
9
  Effect.delay("200 millis"),
10
  Effect.tap(Console.log("task2 done")),
11
  Effect.onInterrupt(() => Console.log("task2 interrupted")),
12
)
13

14
const program = Effect.race(task1, task2)
15

16
const exit = await Effect.runPromiseExit(program)
17
console.log(exit)
18
exit._tag // => "Failure"
19
/*
20
Output:
21
{
22
  _id: 'Exit',
23
  _tag: 'Failure',
24
  cause: {
25
    _id: 'Cause',
26
    _tag: 'Parallel',
27
    left: { _id: 'Cause', _tag: 'Fail', failure: 'task1' },
28
    right: { _id: 'Cause', _tag: 'Fail', failure: 'task2' }
29
  }
30
}
31
*/

If you want to handle the result of whichever task completes first, whether it succeeds or fails, you can use the Effect.result function. This function wraps the result in a Result type, allowing you to see if the result was a success (Success) or a failure (Failure):

Example (Handling Success or Failure with Result)

1
import { Effect, Console, Result } from "effect"
2

3
const task1 = Effect.fail("task1").pipe(
4
  Effect.delay("100 millis"),
5
  Effect.tap(Console.log("task1 done")),
6
  Effect.onInterrupt(() => Console.log("task1 interrupted")),
7
)
8
const task2 = Effect.succeed("task2").pipe(
9
  Effect.delay("200 millis"),
10
  Effect.tap(Console.log("task2 done")),
11
  Effect.onInterrupt(() => Console.log("task2 interrupted")),
12
)
13

14
// Run both tasks concurrently, wrapping the result
15
// in Result to capture success or failure
16
const program = Effect.race(Effect.result(task1), Effect.result(task2))
17

18
await Effect.runPromise(program) // => Result.fail("task1")
19
/*
20
Output:
21
task2 interrupted
22
{ _id: 'Result', _tag: 'Failure', failure: 'task1' }
23
*/

raceAll

This function runs multiple effects concurrently and returns the result of the first one to succeed. If one effect succeeds, the others will be interrupted.

If none of the effects succeed, the function will fail with the last error encountered.

This is useful when you want to race multiple effects, but only care about the first one to succeed. It is commonly used in cases like timeouts, retries, or when you want to optimize for the faster response without worrying about the other effects.

Example (All Tasks Succeed)

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

3
const task1 = Effect.succeed("task1").pipe(
4
  Effect.delay("100 millis"),
5
  Effect.tap(Console.log("task1 done")),
6
  Effect.onInterrupt(() => Console.log("task1 interrupted")),
7
)
8
const task2 = Effect.succeed("task2").pipe(
9
  Effect.delay("200 millis"),
10
  Effect.tap(Console.log("task2 done")),
11
  Effect.onInterrupt(() => Console.log("task2 interrupted")),
12
)
13

14
const task3 = Effect.succeed("task3").pipe(
15
  Effect.delay("150 millis"),
16
  Effect.tap(Console.log("task3 done")),
17
  Effect.onInterrupt(() => Console.log("task3 interrupted")),
18
)
19

20
const program = Effect.raceAll([task1, task2, task3])
21

22
await Effect.runPromise(program) // => "task1"
23
/*
24
Output:
25
task1 done
26
task2 interrupted
27
task3 interrupted
28
*/

Example (One Task Fails, Two Tasks Succeed)

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

3
const task1 = Effect.fail("task1").pipe(
4
  Effect.delay("100 millis"),
5
  Effect.tap(Console.log("task1 done")),
6
  Effect.onInterrupt(() => Console.log("task1 interrupted")),
7
)
8
const task2 = Effect.succeed("task2").pipe(
9
  Effect.delay("200 millis"),
10
  Effect.tap(Console.log("task2 done")),
11
  Effect.onInterrupt(() => Console.log("task2 interrupted")),
12
)
13

14
const task3 = Effect.succeed("task3").pipe(
15
  Effect.delay("150 millis"),
16
  Effect.tap(Console.log("task3 done")),
17
  Effect.onInterrupt(() => Console.log("task3 interrupted")),
18
)
19

20
const program = Effect.raceAll([task1, task2, task3])
21

22
await Effect.runPromise(program) // => "task3"
23
/*
24
Output:
25
task3 done
26
task2 interrupted
27
*/

Example (All Tasks Fail)

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

3
const task1 = Effect.fail("task1").pipe(
4
  Effect.delay("100 millis"),
5
  Effect.tap(Console.log("task1 done")),
6
  Effect.onInterrupt(() => Console.log("task1 interrupted")),
7
)
8
const task2 = Effect.fail("task2").pipe(
9
  Effect.delay("200 millis"),
10
  Effect.tap(Console.log("task2 done")),
11
  Effect.onInterrupt(() => Console.log("task2 interrupted")),
12
)
13

14
const task3 = Effect.fail("task3").pipe(
15
  Effect.delay("150 millis"),
16
  Effect.tap(Console.log("task3 done")),
17
  Effect.onInterrupt(() => Console.log("task3 interrupted")),
18
)
19

20
const program = Effect.raceAll([task1, task2, task3])
21

22
const exit = await Effect.runPromiseExit(program)
23
console.log(exit)
24
exit._tag // => "Failure"
25
/*
26
Output:
27
{
28
  _id: 'Exit',
29
  _tag: 'Failure',
30
  cause: { _id: 'Cause', _tag: 'Fail', failure: 'task2' }
31
}
32
*/

raceFirst

This function takes two effects and runs them concurrently, returning the result of the first one that completes, regardless of whether it succeeds or fails.

This function is useful when you want to race two operations, and you want to proceed with whichever one finishes first, regardless of whether it succeeds or fails.

Example (Both Tasks Succeed)

1
import { Effect, Console, Exit } from "effect"
2

3
const task1 = Effect.succeed("task1").pipe(
4
  Effect.delay("100 millis"),
5
  Effect.tap(Console.log("task1 done")),
6
  Effect.onInterrupt(() =>
7
    Console.log("task1 interrupted").pipe(Effect.delay("100 millis")),
8
  ),
9
)
10
const task2 = Effect.succeed("task2").pipe(
11
  Effect.delay("200 millis"),
12
  Effect.tap(Console.log("task2 done")),
13
  Effect.onInterrupt(() =>
14
    Console.log("task2 interrupted").pipe(Effect.delay("100 millis")),
15
  ),
16
)
17

18
const program = Effect.raceFirst(task1, task2).pipe(
19
  Effect.tap(Console.log("more work...")),
20
)
21

22
await Effect.runPromiseExit(program) // => Exit.succeed("task1")
23
/*
24
Output:
25
task1 done
26
task2 interrupted
27
more work...
28
*/

Example (One Task Fails, One Succeeds)

1
import { Effect, Console, Exit } from "effect"
2

3
const task1 = Effect.fail("task1").pipe(
4
  Effect.delay("100 millis"),
5
  Effect.tap(Console.log("task1 done")),
6
  Effect.onInterrupt(() =>
7
    Console.log("task1 interrupted").pipe(Effect.delay("100 millis")),
8
  ),
9
)
10
const task2 = Effect.succeed("task2").pipe(
11
  Effect.delay("200 millis"),
12
  Effect.tap(Console.log("task2 done")),
13
  Effect.onInterrupt(() =>
14
    Console.log("task2 interrupted").pipe(Effect.delay("100 millis")),
15
  ),
16
)
17

18
const program = Effect.raceFirst(task1, task2).pipe(
19
  Effect.tap(Console.log("more work...")),
20
)
21

22
await Effect.runPromiseExit(program) // => Exit.fail("task1")
23
/*
24
Output:
25
task2 interrupted
26
*/

Observing the Winner

The optional onWinner callback receives the winning fiber and its index (0 for the first effect, 1 for the second). The callback is observational: the result of raceFirst is still determined by the first effect to complete.

Example (Observing Which Task Completes First)

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

3
const task1 = Effect.succeed("task1").pipe(
4
  Effect.delay("100 millis"),
5
  Effect.tap(Console.log("task1 done")),
6
  Effect.onInterrupt(() =>
7
    Console.log("task1 interrupted").pipe(Effect.delay("100 millis")),
8
  ),
9
)
10
const task2 = Effect.succeed("task2").pipe(
11
  Effect.delay("200 millis"),
12
  Effect.tap(Console.log("task2 done")),
13
  Effect.onInterrupt(() =>
14
    Console.log("task2 interrupted").pipe(Effect.delay("100 millis")),
15
  ),
16
)
17

18
const program = Effect.raceFirst(task1, task2, {
19
  onWinner: ({ index }) => console.log(`task${index + 1} won`),
20
})
21

22
Effect.runFork(program)
23
/*
24
Output:
25
task1 done
26
task1 won
27
task2 interrupted
28
*/