Concurrency #

import { Effect, Fiber } from "effect"

const worker = (name: string) =>
  Effect.gen(function* () {
    for (let i = 1; i <= 3; i++) {
      console.log(name, "step", i)
      yield* Effect.yieldNow   // give the other fibers a turn
    }
    return name + " done"
  })

const program = Effect.gen(function* () {
  // forkChild starts the work in a new fiber and returns immediately
  const a = yield* Effect.forkChild(worker("A"))
  const b = yield* Effect.forkChild(worker("B"))
  console.log("parent: both forked, neither has run yet")

  // join waits for the fiber and gives back its success value
  const resultA = yield* Fiber.join(a)
  const resultB = yield* Fiber.join(b)
  console.log(resultA, "/", resultB)
})

Effect.runPromise(program)

import { Cause, Effect, Exit, Fiber } from "effect"

const slowJob = Effect.gen(function* () {
  yield* Effect.sleep("50 millis")
  return "report ready"
}).pipe(Effect.onInterrupt(() => Effect.sync(() => console.log("slowJob: cleaning up"))))

const program = Effect.gen(function* () {
  const ok = yield* Effect.forkChild(Effect.succeed(42))
  const bad = yield* Effect.forkChild(Effect.fail("disk full"))
  const slow = yield* Effect.forkChild(slowJob)

  // await never throws: it hands you the Exit to inspect
  const exit1 = yield* Fiber.await(ok)
  const exit2 = yield* Fiber.await(bad)
  console.log(exit1._tag, Exit.isSuccess(exit1) ? exit1.value : "")
  console.log(exit2._tag, Exit.isFailure(exit2) ? Cause.squash(exit2.cause) : "")

  // interrupt asks the fiber to stop and waits until it has stopped
  yield* Effect.sleep("5 millis")
  yield* Fiber.interrupt(slow)
  const exit3 = yield* Fiber.await(slow)
  const reasons = Exit.isFailure(exit3) ? exit3.cause.reasons.map((r) => r._tag).join(",") : ""
  console.log(exit3._tag, "reason:", reasons)
})

Effect.runPromise(program)

import { Effect, Fiber } from "effect"

const ticker = (name: string) =>
  Effect.gen(function* () {
    console.log(name, "started")
    yield* Effect.sleep("30 millis")
    console.log(name, "finished")
  }).pipe(Effect.onInterrupt(() => Effect.sync(() => console.log(name, "interrupted"))))

// 1. A child dies with its parent. The parent returns after 5ms, long before 30ms.
const parent = Effect.gen(function* () {
  yield* Effect.forkChild(ticker("child"))
  yield* Effect.sleep("5 millis")
  console.log("parent finished first")
})

const program = Effect.gen(function* () {
  const parentFiber = yield* Effect.forkChild(parent)
  yield* Fiber.join(parentFiber)

  // 2. A scoped fiber dies when the scope closes
  yield* Effect.scoped(Effect.gen(function* () {
    yield* Effect.forkScoped(ticker("scoped"))
    yield* Effect.sleep("5 millis")
    console.log("leaving scope")
  }))

  // 3. A detached fiber has no owner: we must join it ourselves or it leaks
  const detached = yield* Effect.forkDetach(ticker("detached"))
  yield* Effect.sleep("5 millis")
  console.log("parent moving on, detached still runs")
  yield* Fiber.join(detached)
})

Effect.runPromise(program)

import { Effect, Ref } from "effect"

const program = Effect.gen(function* () {
  const inFlight = yield* Ref.make(0)
  const maxInFlight = yield* Ref.make(0)

  // A fake request that records how many copies of itself run at once
  const fetchUser = (id: number) =>
    Effect.gen(function* () {
      const now = yield* Ref.updateAndGet(inFlight, (n) => n + 1)
      yield* Ref.update(maxInFlight, (m) => Math.max(m, now))
      yield* Effect.sleep("5 millis")
      yield* Ref.update(inFlight, (n) => n - 1)
      return "user-" + id
    })

  const ids = [1, 2, 3, 4, 5, 6]

  const sequential = yield* Effect.forEach(ids, fetchUser)                     // one at a time
  console.log("sequential", sequential.join(","), "max:", yield* Ref.get(maxInFlight))

  yield* Ref.set(maxInFlight, 0)
  const limited = yield* Effect.forEach(ids, fetchUser, { concurrency: 2 })      // at most two
  console.log("limited   ", limited.join(","), "max:", yield* Ref.get(maxInFlight))

  yield* Ref.set(maxInFlight, 0)
  const all = yield* Effect.all(ids.map(fetchUser), { concurrency: "unbounded" }) // everything
  console.log("unbounded ", all.join(","), "max:", yield* Ref.get(maxInFlight))
})

Effect.runPromise(program)

import { Effect, Option, Ref } from "effect"

const program = Effect.gen(function* () {
  const cancelled = yield* Ref.make<Array<string>>([])

  // A fake request that records when it is cancelled
  const request = (name: string, ms: number) =>
    Effect.sleep(ms).pipe(                   // a plain number means milliseconds
      Effect.as(name),
      Effect.onInterrupt(() => Ref.update(cancelled, (list) => [...list, name]))
    )

  // race: first success wins, the other one is interrupted
  const winner = yield* Effect.race(request("mirror-eu", 50), request("mirror-us", 5))
  console.log("winner:", winner, "| cancelled:", yield* Ref.get(cancelled))

  // raceAll: same for a list
  yield* Ref.set(cancelled, [])
  const fastest = yield* Effect.raceAll([request("a", 40), request("b", 5), request("c", 60)])
  console.log("fastest:", fastest, "| cancelled:", (yield* Ref.get(cancelled)).sort())

  // timeoutOption: None on timeout, and the slow work is interrupted
  yield* Ref.set(cancelled, [])
  const maybe = yield* Effect.timeoutOption(request("report", 50), "5 millis")
  console.log("timed out:", Option.isNone(maybe), "| cancelled:", yield* Ref.get(cancelled))

  // timeout: a typed TimeoutError that you can catch by its tag
  const value = yield* Effect.timeout(request("report", 50), "5 millis").pipe(
    Effect.catchTag("TimeoutError", () => Effect.succeed("fallback"))
  )
  console.log("value:", value)
})

Effect.runPromise(program)

import { Cause, Effect, Exit, Fiber } from "effect"

const saveFile = Effect.gen(function* () {
  // This block runs fully or not at all: interruption waits for it
  yield* Effect.uninterruptible(Effect.gen(function* () {
    console.log("write header")
    yield* Effect.sleep("10 millis")
    console.log("write body")
  }))
  yield* Effect.sleep("50 millis")     // interruptible again: this is where we get stopped
  console.log("never printed")
}).pipe(Effect.onInterrupt(() => Effect.sync(() => console.log("saveFile: closing handle"))))

const program = Effect.gen(function* () {
  const fiber = yield* Effect.forkChild(saveFile)
  yield* Effect.sleep("2 millis")
  yield* Fiber.interrupt(fiber)         // arrives during the uninterruptible block
  const exit = yield* Fiber.await(fiber)
  console.log("interrupted:", Exit.hasInterrupts(exit))

  // Interrupting yourself: the Cause says Interrupt, not Fail
  const self = yield* Effect.exit(Effect.gen(function* () {
    yield* Effect.interrupt
    return "unreachable"
  }))
  if (Exit.isFailure(self)) {
    const tags = self.cause.reasons.map((r) => r._tag).join(",")
    console.log("reasons:", tags, "| hasFails:", Cause.hasFails(self.cause))
  }

  // In Effect.all one failure interrupts the siblings
  const exit2 = yield* Effect.exit(Effect.all([
    Effect.sleep("50 millis").pipe(Effect.onInterrupt(() => Effect.sync(() => console.log("sibling cancelled")))),
    Effect.sleep("5 millis").pipe(Effect.andThen(Effect.fail("bad input")))
  ], { concurrency: "unbounded" }))
  console.log(exit2._tag, Exit.isFailure(exit2) ? Cause.squash(exit2.cause) : "")
})

Effect.runPromise(program)

import { Cause, Deferred, Effect, Fiber, Queue } from "effect"

const program = Effect.gen(function* () {
  // Deferred: a one-shot signal. The server waits, the parent completes it.
  const configReady = yield* Deferred.make<string>()
  const server = yield* Effect.forkChild(Effect.gen(function* () {
    console.log("server: waiting for config")
    const config = yield* Deferred.await(configReady)   // suspends here
    console.log("server: started with", config)
  }))
  yield* Effect.sleep("5 millis")
  yield* Deferred.succeed(configReady, "port=8080")
  yield* Fiber.join(server)

  // Queue: capacity 2, so the producer is slowed down to the consumer's pace
  const jobs = yield* Queue.bounded<number, Cause.Done>(2)

  const producer = yield* Effect.forkChild(Effect.gen(function* () {
    for (const n of [1, 2, 3, 4, 5]) {
      if (yield* Queue.isFull(jobs)) console.log("producer: queue full, waiting")
      yield* Queue.offer(jobs, n)                        // suspends while full
      console.log("producer: offered", n)
    }
    yield* Queue.end(jobs)                               // no more jobs
  }))

  const consumer = yield* Effect.forkChild(Effect.gen(function* () {
    const seen: Array<number> = []
    yield* Effect.gen(function* () {
      seen.push(yield* Queue.take(jobs))
      yield* Effect.sleep("5 millis")                    // a slow consumer
    }).pipe(Effect.forever, Effect.catchTag("Done", () => Effect.void))
    return seen
  }))

  yield* Fiber.join(producer)
  console.log("consumer got", (yield* Fiber.join(consumer)).join(","))
})

Effect.runPromise(program)

import { Cause, Effect, Fiber, Queue, Ref } from "effect"

const program = Effect.gen(function* () {
  // 1. bounded: offer waits while the queue is full. The log proves the order.
  const log = yield* Ref.make<Array<string>>([])
  const note = (line: string) => Ref.update(log, (xs) => [...xs, line])
  const bounded = yield* Queue.bounded<number>(1)
  const producer = yield* Effect.forkChild(Effect.gen(function* () {
    yield* Queue.offer(bounded, 1)
    yield* note("offered 1")
    yield* Queue.offer(bounded, 2)          // suspends: the only slot is taken
    yield* note("offered 2")
  }))
  yield* Effect.sleep("5 millis")
  yield* note("consumer takes")
  yield* Queue.take(bounded)                // frees the slot, the producer continues
  yield* Fiber.join(producer)
  console.log("bounded :", (yield* Ref.get(log)).join(", "))

  // 2. dropping: a full queue refuses the new value, and offer returns false
  const dropping = yield* Queue.dropping<number>(2)
  const accepted = yield* Effect.forEach([1, 2, 3], (n) => Queue.offer(dropping, n))
  console.log("dropping: accepted", accepted.join(","), "| kept", (yield* Queue.takeAll(dropping)).join(","))

  // 3. sliding: a full queue removes the oldest value, and offer returns true
  const sliding = yield* Queue.sliding<number>(2)
  const accepted2 = yield* Effect.forEach([1, 2, 3], (n) => Queue.offer(sliding, n))
  console.log("sliding : accepted", accepted2.join(","), "| kept", (yield* Queue.takeAll(sliding)).join(","))

  // 4. end: the queue keeps its values, consumers drain them, then take fails with Done
  const jobs = yield* Queue.unbounded<number, Cause.Done>()
  yield* Queue.offerAll(jobs, [1, 2, 3, 4, 5])
  yield* Queue.end(jobs)
  console.log("size after end:", yield* Queue.size(jobs))
  console.log("batch of 1..3 :", (yield* Queue.takeBetween(jobs, 1, 3)).join(","))
  console.log("the rest      :", (yield* Queue.takeAll(jobs)).join(","))
  const last = yield* Queue.take(jobs).pipe(
    Effect.catchTag("Done", () => Effect.succeed("Done: no more jobs"))
  )
  console.log("then          :", last)
})

Effect.runPromise(program)

import { Deferred, Effect, Fiber, Latch, Option, Ref } from "effect"

const program = Effect.gen(function* () {
  const gate = yield* Latch.make()             // starts closed
  const log = yield* Ref.make<Array<string>>([])
  const note = (line: string) => Ref.update(log, (xs) => [...xs, line])

  const worker = (name: string) =>
    Effect.gen(function* () {
      yield* note(name + " waits")
      yield* Latch.await(gate)                 // suspends while the gate is closed
      yield* note(name + " runs")
    })

  const fibers = yield* Effect.forEach(["a", "b", "c"], (n) => Effect.forkChild(worker(n)))
  yield* Effect.sleep("5 millis")
  console.log("open:", Latch.isOpen(gate), "|", (yield* Ref.get(log)).join(", "))

  yield* Latch.open(gate)                      // wakes every waiter at once
  yield* Fiber.joinAll(fibers)
  console.log("open:", Latch.isOpen(gate), "|", (yield* Ref.get(log)).join(", "))

  // whenOpen on an open latch runs the effect at once
  console.log(yield* Latch.whenOpen(gate, Effect.succeed("late worker runs at once")))

  // close makes the gate reusable: a new waiter suspends again
  yield* Latch.close(gate)
  const late = yield* Latch.await(gate).pipe(Effect.as("passed"), Effect.timeoutOption("5 millis"))
  console.log("after close:", Option.isNone(late) ? "a new waiter suspends" : "passed")

  // Deferred: 1 value, 1 time. The second completion is ignored.
  const cell = yield* Deferred.make<number>()
  const first = yield* Deferred.succeed(cell, 1)
  const second = yield* Deferred.succeed(cell, 2)
  console.log("deferred:", first, second, "value", yield* Deferred.await(cell))
})

Effect.runPromise(program)

import { Effect, PubSub, Ref, Semaphore } from "effect"

const program = Effect.gen(function* () {
  // PubSub: every subscriber sees every message
  yield* Effect.scoped(Effect.gen(function* () {
    const events = yield* PubSub.unbounded<string>()
    const audit = yield* PubSub.subscribe(events)      // released when the scope closes
    const mailer = yield* PubSub.subscribe(events)
    yield* PubSub.publish(events, "order:created")
    yield* PubSub.publish(events, "order:paid")
    console.log("audit  saw", yield* PubSub.take(audit), yield* PubSub.take(audit))
    console.log("mailer saw", yield* PubSub.take(mailer), yield* PubSub.take(mailer))
  }))

  // Semaphore: at most 2 database queries at once, whoever calls
  const dbPool = yield* Semaphore.make(2)
  const inFlight = yield* Ref.make(0)
  const maxInFlight = yield* Ref.make(0)

  const query = (sql: string) =>
    Semaphore.withPermits(dbPool, 1)(Effect.gen(function* () {
      const now = yield* Ref.updateAndGet(inFlight, (n) => n + 1)
      yield* Ref.update(maxInFlight, (m) => Math.max(m, now))
      yield* Effect.sleep("5 millis")
      yield* Ref.update(inFlight, (n) => n - 1)
      return sql + " ok"
    }))

  // Launch everything at once; the semaphore, not the option, enforces the limit
  const results = yield* Effect.all(
    ["select 1", "select 2", "select 3", "select 4", "select 5"].map(query),
    { concurrency: "unbounded" }
  )
  console.log(results.join(" | "))
  console.log("max concurrent queries:", yield* Ref.get(maxInFlight))
})

Effect.runPromise(program)

import { Effect, Option, Ref, Semaphore } from "effect"

const program = Effect.gen(function* () {
  // 1. Mutex: 1 permit protects a read-wait-write on a plain variable
  let balance = 100
  const lock = yield* Semaphore.make(1)
  const withdraw = (amount: number) =>
    Semaphore.withPermits(lock, 1)(Effect.gen(function* () {
      const current = balance
      yield* Effect.sleep("1 millis")       // without the lock, other fibers run here
      balance = current - amount
    }))
  yield* Effect.forEach([10, 20, 30], withdraw, { concurrency: "unbounded", discard: true })
  console.log("balance:", balance)          // 40. Without the lock: 90, 80, or 70

  // 2. Weighted permits: "big" needs 3 of 4 slots, so it waits for "small-a" to end
  const slots = yield* Semaphore.make(4)
  const order = yield* Ref.make<Array<string>>([])
  const job = (name: string, cost: number, ms: number) =>
    Semaphore.withPermits(slots, cost)(
      Effect.sleep(ms).pipe(Effect.andThen(Ref.update(order, (xs) => [...xs, name])))
    )
  yield* Effect.all(
    [job("small-a", 1, 5), job("small-b", 1, 25), job("big", 3, 1)],
    { concurrency: "unbounded" }
  )
  console.log("finish order:", (yield* Ref.get(order)).join(", "))

  // 3. Skip when busy: the second flush finds no permit and returns none at once
  const flushLock = yield* Semaphore.make(1)
  const flush = Semaphore.withPermitsIfAvailable(flushLock, 1)(
    Effect.sleep("5 millis").pipe(Effect.as("flushed"))
  )
  const [first, second] = yield* Effect.all([flush, flush], { concurrency: "unbounded" })
  console.log("first:", Option.getOrElse(first, () => "skipped"), "| second:", Option.getOrElse(second, () => "skipped"))
})

Effect.runPromise(program)

import { Effect, Fiber } from "effect"

const compute = Effect.gen(function* () {
  yield* Effect.sleep("2 millis")
  return 20
})

const program = Effect.gen(function* () {
  const result = yield* Effect.forkChild(compute)
  console.log("total:", result + 1)
})

Effect.runPromise(program)

import { Effect, Ref } from "effect"

const program = Effect.gen(function* () {
  const inFlight = yield* Ref.make(0)
  const maxInFlight = yield* Ref.make(0)

  const lookup = (id: number) =>
    Effect.gen(function* () {
      const now = yield* Ref.updateAndGet(inFlight, (n) => n + 1)
      yield* Ref.update(maxInFlight, (m) => Math.max(m, now))
      yield* Effect.sleep("5 millis")
      yield* Ref.update(inFlight, (n) => n - 1)
      return id
    })

  yield* Effect.forEach([1, 2, 3, 4, 5, 6], lookup, { parallel: 3 })
  console.log("max in flight:", yield* Ref.get(maxInFlight))
})

Effect.runPromise(program)

import { Effect, Fiber } from "effect"

const sendEmail = Effect.gen(function* () {
  yield* Effect.sleep("10 millis")
  console.log("email sent")
})

const handleRequest = Effect.gen(function* () {
  yield* Effect.forkChild(sendEmail)
  console.log("request handled")
})

Effect.runPromise(handleRequest)

import { Effect } from "effect"

const slowLookup = Effect.sleep("50 millis").pipe(Effect.as("fresh price"))

const priceWithTimeout: Effect.Effect<string> = Effect.timeout(slowLookup, "5 millis")

const program = Effect.gen(function* () {
  const price = yield* priceWithTimeout
  console.log("price:", price)
})

Effect.runPromise(program)

import { Effect, Fiber } from "effect"

const writeRecord = Effect.gen(function* () {
  console.log("write header")
  yield* Effect.sleep("10 millis")
  console.log("write body")
  yield* Effect.sleep("50 millis")
  console.log("archived")
})

const program = Effect.gen(function* () {
  const writer = yield* Effect.forkChild(writeRecord)
  yield* Effect.sleep("2 millis")
  yield* Fiber.interrupt(writer)
  console.log("writer stopped")
})

Effect.runPromise(program)

import { Effect, Ref, Semaphore } from "effect"

const program = Effect.gen(function* () {
  const inFlight = yield* Ref.make(0)
  const maxInFlight = yield* Ref.make(0)

  const query = (id: number) =>
    Effect.gen(function* () {
      const pool = yield* Semaphore.make(2)
      return yield* Semaphore.withPermits(pool, 1)(Effect.gen(function* () {
        const now = yield* Ref.updateAndGet(inFlight, (n) => n + 1)
        yield* Ref.update(maxInFlight, (m) => Math.max(m, now))
        yield* Effect.sleep("5 millis")
        yield* Ref.update(inFlight, (n) => n - 1)
        return id
      }))
    })

  yield* Effect.all([1, 2, 3, 4, 5].map(query), { concurrency: "unbounded" })
  console.log("max concurrent:", yield* Ref.get(maxInFlight))
})

Effect.runPromise(program)

import { Effect } from "effect"

const primary = Effect.sleep("2 millis").pipe(Effect.andThen(Effect.fail("primary down")))
const backup = Effect.sleep("20 millis").pipe(Effect.as("backup answered"))

const program = Effect.gen(function* () {
  const answer = yield* Effect.raceFirst(primary, backup)
  console.log(answer)
})

Effect.runPromise(program)

import { Effect } from "effect"

const heartbeat = Effect.gen(function* () {
  console.log("heartbeat started")
  yield* Effect.sleep("1 second")
}).pipe(Effect.onInterrupt(() => Effect.sync(() => console.log("heartbeat stopped"))))

const program = Effect.gen(function* () {
  yield* Effect.forkScoped(heartbeat)
  yield* Effect.sleep("5 millis")
  console.log("request done")
})

Effect.runPromise(program)

import { Effect, Option, Result, Semaphore } from "effect"

const program = Effect.gen(function* () {
  const lock = yield* Semaphore.make(1)

  const risky = (fail: boolean) =>
    Effect.gen(function* () {
      yield* Semaphore.take(lock, 1)
      if (fail) yield* Effect.fail("boom")
      yield* Semaphore.release(lock, 1)
      return "done"
    })

  const first = yield* Effect.result(risky(true))
  console.log("first:", Result.isFailure(first) ? "failed" : "ok")

  const second = yield* risky(false).pipe(Effect.timeoutOption("50 millis"))
  console.log("second:", Option.isSome(second) ? second.value : "hung")
})

Effect.runPromise(program)

import { Effect, Fiber, Queue } from "effect"

const program = Effect.gen(function* () {
  const jobs = yield* Queue.bounded<number>(2)

  const producer = yield* Effect.forkChild(Effect.gen(function* () {
    yield* Queue.offerAll(jobs, [10, 20, 30])
    yield* Queue.end(jobs)
  }))

  const consumer = yield* Effect.forkChild(Effect.gen(function* () {
    let total = 0
    yield* Effect.gen(function* () {
      total += yield* Queue.take(jobs)
    }).pipe(Effect.forever, Effect.catchTag("Done", () => Effect.void))
    return total
  }))

  yield* Fiber.join(producer)
  console.log("total:", yield* Fiber.join(consumer))
})

Effect.runPromise(program)

import { Effect, Fiber, Latch, Option } from "effect"

const program = Effect.gen(function* () {
  const ready = yield* Latch.make()

  // 2 workers wait before the cache is warm
  const worker = (name: string) => Latch.whenOpen(ready, Effect.succeed(name + " served"))
  const early = yield* Effect.forEach(["a", "b"], (n) => Effect.forkChild(worker(n)))

  yield* Effect.sleep("5 millis")            // cache warm-up
  yield* Latch.release(ready)
  console.log((yield* Fiber.joinAll(early)).join(", "))

  // a worker that arrives after the warm-up must not wait
  const late = yield* worker("c").pipe(Effect.timeoutOption("5 millis"))
  console.log(Option.isSome(late) ? late.value : "c timed out")
})

Effect.runPromise(program)