7. Concurrency

Ordinary calls run one after another. Running things at the same time is always written with an explicit combinator, and the concurrent parts are programs, so they start only when the combinator starts them.

Running several things at once

const dashboard = Effect.fn("dashboard")(function* (id: string) {
  const [profile, orders] = yield* Effect.all([loadProfile(id), recentOrders(id)], {
    concurrency: 2
  })
  return { profile, orders }
})

function dashboard(id: string) {
  const [profile, orders] = all([loadProfile(id), recentOrders(id)], {
    concurrency: 2
  })
  return { profile, orders }
}

const thumbnails = forEach(images, (image) => resize(image), { concurrency: 8 })
const fastest = race(fromCache(id), fromNetwork(id))
const first = await.race [fromCache(id), fromNetwork(id)]
const quick = timeout(slowQuery(id), "500 millis")

Each combinator has its own rule for a failing branch, and Effeck adds nothing to Effect's semantics here:

A catch around any of these sees the failure as usual. When Effect.race fails because both branches failed, catch receives the first failure (Chapter 3).

Background work with a lifetime

fork starts a program on a new fiber owned by the enclosing block. When the block exits, by any path, the fiber is interrupted, and the block waits for its cleanup before releasing anything acquired earlier in the block. A function body is a block, so fork needs no using nearby.

const serve = Effect.fn("serve")(
  function* (port: number) {
    const http = yield* Http
    const server = yield* http.listen(port)
    yield* Effect.forkScoped(heartbeat())
    return yield* server.awaitShutdown()
  },
  Effect.scoped
)

function serve(port: number) {
  using server = Http.listen(port)
  fork heartbeat()          // interrupted when serve exits, before the server closes
  return server.awaitShutdown()
}

fork is a keyword (D166): fork buildReport(id) forks one call, and fork { ... } forks a block. It returns a Fiber, and join waits for it (D225):

function withProgress(id: string) {
  const pending = fork buildReport(id)
  sendProgressEmail(id)
  return join pending           // waits for it; fails if it failed
}
// Report!ReportError | MailError!Reports | Mailer

fork adds the child's requirements to R and nothing else: the fiber cannot outlive withProgress, so no Scope appears in the contract, and its failures reach you only through join. The call form fork(buildReport(id)) means the same. For a longer lifetime, say so: forkIn(scope, program) ties the fiber to a scope you hold, detach(program) ties it to nothing (D170), Effect.forkScoped ties it to the caller's scope and adds Scope to R, and Effect.forkChild ties it to the current fiber.

join waits for a Fiber, or an array or object of fibers, and keeps the shape (Fiber.join). When one fiber fails, the others still running are interrupted and the failure propagates, as Effect.all fails fast. A Deferred waits with its method, d.await() (D227):

const summary = Effect.fn("summary")(function* (id: string) {
  const profile = yield* Effect.forkChild(loadProfile(id))
  const orders = yield* Effect.forkChild(recentOrders(id))
  const [p, o] = yield* Fiber.joinAll([profile, orders])
  return { profile: p, orders: o }
})

function summary(id: string) {
  const profile = fork loadProfile(id)
  const orders = fork recentOrders(id)
  return { profile: join profile, orders: join orders }
}

To start several programs and wait for them in one step, use an await operation instead of forking each: await.all [loadProfile(id), recentOrders(id)] (below).

await waits for promises in JavaScript, so in Effeck code a plain await is an error that names its replacement (D226): await fiber is EK4609, whose quick fix writes join fiber; await d on a Deferred is EK4610, whose quick fix writes d.await(); await loadProfile(id) is EK4601, because a call already runs and waits where it stands; a promise is EK4602, and waits in a hatch, try(async () => …), where await keeps its JavaScript meaning; anything else is EK4603. An await in a hatch's function that is not async is EK4604. join on anything but fibers is EK4611, and join inside a hatch's function, which is plain TypeScript, is EK4612.

Await operations

await.all, await.allSettled, await.race, and await.any (D196) come from the TC39 await operations proposal (stage 1) and, for an object operand, the Await Dictionary proposal (stage 3). Each behaves as the Promise method of the same name. Each element is a program passed by name (D147), started in its own fiber; an element that is a fiber or a Deferred already running is waited for, not restarted.

const next = (jobs: Queue.Queue<Job>, shutdown: Deferred.Deferred<void>) =>
  Effect.raceAllFirst([
    Effect.map(Queue.take(jobs), (job) => run(job)),
    Effect.as(Deferred.await(shutdown), "stopped"),
    Effect.flatMap(Effect.sleep("30 seconds"), () => heartbeat())
  ])

function next(jobs: Queue.Queue<Job>, shutdown: Deferred.Deferred<void>) {
  return await.race [
    run(jobs.take()),
    do {
      shutdown.await()
      "stopped"
    },
    do {
      sleep("30 seconds")
      heartbeat()
    }
  ]
}

• await.all [a(), b()] and await.all { x: a(), y: b() } are Effect.all with unbounded concurrency: every result, the shape kept, an object's keys in the order written, and a failure interrupts the rest.
• await.allSettled is Effect.all with mode: "result": each outcome is a Result, and it never fails.
• await.race is Effect.raceFirst over any number of elements: the first to finish wins, success or failure, and the others are interrupted. Of elements ready at once, the first written wins. An empty race never finishes, as Promise.race([]) does not.
• await.any is Effect.raceAll: the first success wins; it fails only when every element fails, with every failure.
• A do { ... } expression (D187) gives an element statements of its own. A race over an object is EK4608.
• Controversial (D196): await.x behaves like Promise.x, while a named Effect function keeps Effect's meaning, so await.race [a(), b()] takes the first to finish and race(a(), b()) the first success.
• select (D168) was replaced by await.race (D197) and is EK4607. To poll without waiting, call Queue.poll and test the result with an if.

Concurrent loops

for concurrent runs its body for each item in its own fiber and joins them all before the loop ends (D169). for concurrent(n) runs at most n at once:

const saveAll = (ids: ReadonlyArray<string>) =>
  Effect.forEach(ids, (id) => Effect.flatMap(fetchPage(id), save), { concurrency: 8, discard: true })

function saveAll(ids: ReadonlyArray<string>) {
  for concurrent(8) (const id of ids) save(fetchPage(id))
}

A failure interrupts the items still running and fails the loop. continue ends one item. A break that would leave the loop is EK4605 and a return is EK4606, because each item runs in its own fiber; throw to stop every item. Use forEach when you need the results.

An expression-bodied arrow has no block to own its child: () => fork(p) uses Effect.forkScoped and keeps Scope in R, unlike () => { return fork(p) } (D96). Returning a block-owned child currently returns a fiber already interrupted when that block exited; the compiler does not warn.

What a new fiber sees

A forked fiber starts with the services of the code that forked it, as in Effect. Later changes in either fiber do not flow to the other. Handlers, introduced in Chapter 9, are the one thing that does not cross to a new fiber; Chapter 9 shows the compiler's check.