concurrency（concurrency 搜索）

Concurrency in Effect

Overview

Effect provides lightweight fiber-based concurrency:

• Fibers - Lightweight threads managed by Effect runtime
• Structured concurrency - Parent fibers supervise children
• Safe interruption - Clean cancellation with resource cleanup
• Concurrent primitives - Queue, Deferred, Semaphore, PubSub

Basic Parallel Execution

Effect.all with Concurrency

import { Effect } from "effect";

const results = yield * Effect.all([fetchUser(1), fetchUser(2), fetchUser(3)], { concurrency: "unbounded" });

const results = yield * Effect.all(tasks, { concurrency: 5 });

const results = yield * Effect.all(tasks);

Effect.forEach with Concurrency

const users = yield * Effect.forEach(userIds, (id) => fetchUser(id), { concurrency: 10 });

Fibers

Creating Fibers with fork

const program = Effect.gen(function* () {
  const fiber = yield* Effect.fork(longRunningTask);

  yield* doOtherWork();

  const result = yield* Fiber.join(fiber);
});

Fork Variants

const fiber = yield * Effect.fork(task);

const fiber = yield * Effect.forkDaemon(task);

const fiber = yield * Effect.forkIn(scope)(task);

const fiber = yield * Effect.forkWithErrorHandler(task, onError);

Fiber Operations

import { Fiber } from "effect";

const result = yield * Fiber.join(fiber);

const exit = yield * Fiber.await(fiber);

yield * Fiber.interrupt(fiber);

const maybeResult = yield * Fiber.poll(fiber);

Racing

Effect.race - First to Complete

const fastest = yield * Effect.race(fetchFromServer1(), fetchFromServer2());

Effect.raceAll - Race Many

const fastest = yield * Effect.raceAll([fetchFromCDN1(), fetchFromCDN2(), fetchFromCDN3()]);

Effect.raceFirst - Include Failures

const first = yield * Effect.raceFirst(task1, task2);

Deferred - One-Time Promise

import { Deferred } from "effect";

const program = Effect.gen(function* () {
  const deferred = yield* Deferred.make<string, never>();

  const fiber = yield* Effect.fork(
    Effect.gen(function* () {
      const value = yield* Deferred.await(deferred);
      yield* Effect.log(`Got: ${value}`);
    }),
  );

  yield* Deferred.succeed(deferred, "Hello!");

  yield* Fiber.join(fiber);
});

Queue - Concurrent Queue

import { Queue } from "effect";

const program = Effect.gen(function* () {
  const queue = yield* Queue.bounded<number>(100);

  yield* Effect.fork(Effect.forEach([1, 2, 3, 4, 5], (n) => Queue.offer(queue, n)));

  const items = yield* Effect.forEach(Array.from({ length: 5 }), () => Queue.take(queue));
});

Queue Variants

const bounded = yield * Queue.bounded<number>(100);

const unbounded = yield * Queue.unbounded<number>();

const dropping = yield * Queue.dropping<number>(100);

const sliding = yield * Queue.sliding<number>(100);

PubSub - Publish/Subscribe

import { PubSub } from "effect";

const program = Effect.gen(function* () {
  const pubsub = yield* PubSub.bounded<string>(100);

  const sub1 = yield* PubSub.subscribe(pubsub);
  const sub2 = yield* PubSub.subscribe(pubsub);

  yield* PubSub.publish(pubsub, "Hello!");

  const msg1 = yield* Queue.take(sub1);
  const msg2 = yield* Queue.take(sub2);
});

Semaphore - Limit Concurrency

import { Effect } from "effect";

const program = Effect.gen(function* () {
  const semaphore = yield* Effect.makeSemaphore(3);

  yield* Effect.forEach(tasks, (task) => semaphore.withPermits(1)(task), { concurrency: "unbounded" });
});

Latch - Coordination Point

import { Latch } from "effect";

const program = Effect.gen(function* () {
  const latch = yield* Latch.make(false);

  yield* Effect.fork(
    Effect.forEach(
      workers,
      (worker) =>
        Effect.gen(function* () {
          yield* Latch.await(latch);
          yield* worker.start();
        }),
      { concurrency: "unbounded" },
    ),
  );

  yield* Latch.open(latch);
});

Interruption

Interrupting Fibers

const fiber = yield * Effect.fork(longTask);

yield * Fiber.interrupt(fiber);

Uninterruptible Regions

const critical = Effect.uninterruptible(
  Effect.gen(function* () {
    yield* beginTransaction();
    yield* performOperations();
    yield* commitTransaction();
  }),
);

Interruptible Within Uninterruptible

const program = Effect.uninterruptible(
  Effect.gen(function* () {
    yield* criticalSetup();

    // This part can be interrupted
    yield* Effect.interruptible(longOperation);

    yield* criticalTeardown();
  }),
);

Supervision

Structured concurrency ensures child fibers are managed:

const parent = Effect.gen(function* () {
  const child1 = yield* Effect.fork(task1);
  const child2 = yield* Effect.fork(task2);

  // If parent fails/interrupts, children are interrupted
  yield* failingOperation();
});
// child1 and child2 automatically interrupted

Daemon Fibers

Escape supervision with daemon:

const daemon = yield * Effect.forkDaemon(backgroundTask);

Common Patterns

Timeout with Fallback

const withTimeout = task.pipe(Effect.timeout("5 seconds"), Effect.map(Option.getOrElse(() => defaultValue)));

Worker Pool

const workerPool = Effect.gen(function* () {
  const semaphore = yield* Effect.makeSemaphore(numWorkers);

  return (task: Effect.Effect<A>) => semaphore.withPermits(1)(task);
});

Parallel with Error Collection

const results =
  yield *
  Effect.all(tasks, {
    concurrency: "unbounded",
    mode: "either", // Collect all results
  });

Best Practices

1. Use Effect.all concurrency for simple parallelism
2. Use Semaphore to limit concurrent operations
3. Prefer structured concurrency over daemon fibers
4. Handle interruption in long-running effects
5. Use Queue for producer/consumer patterns
6. Use Deferred for one-time coordination

Additional Resources

For comprehensive concurrency documentation, consult ${CLAUDE_PLUGIN_ROOT}/references/llms-full.txt.

Search for these sections:

• "Fibers" for fiber management
• "Basic Concurrency" for parallel execution
• "Deferred" for synchronization primitives
• "Queue" for concurrent queues
• "PubSub" for publish/subscribe
• "Semaphore" for concurrency limiting