Fan-out and map-reduce

Parallel workers under a rate limit: watch calls queue for token capacity and the speed-up saturate, against Amdahl's law and the bound the token bucket sets.

Loading the animation…

Concept

Map-reduce is the pattern that buys speed: split the work, run the pieces in parallel, combine. In LangGraph the split is a conditional edge that returns one Send per piece. Each Send becomes its own task in the next super-step, whose input is the payload it carries rather than the graph state; all of them run in that one super-step, their writes are combined by the key's reducer at the barrier, and the reduce node runs once, in the super-step after. The engine's runtime models Send exactly as LangGraph 1.2.14 does: 39 sessions are checked checkpoint for checkpoint against the real library, including a map-reduce, a crash in one worker (with a checkpointer, its siblings' writes are kept and the resume re-runs only the failed worker), an interrupt in one worker, an empty fan-out and sends from two routers beside ordinary edges (the Send tasks' writes are applied after the edge-triggered tasks', in send order).

Parallel on paper is not parallel in practice. Here a document of 32 chunks of 2,000 tokens is summarised chunk by chunk and the summaries reduced, against an endpoint with a limit of 50 requests and 8,000 output tokens a minute and a choice of input-token limits. The width is how many calls run at once: LangGraph's max_concurrency run setting caps the tasks of a super-step that run together.

Without a binding limit (13.3 s at width 32 and 200,000 input tokens a minute) the speed-up follows Amdahl's law exactly: the split and the reduce cannot be parallelised, the summaries are a fraction 0.935 of the one-at-a-time time, and width 32 gives 10.58× rather than 32×.
At 80,000 input tokens a minute the curve bends: the workers need 102,840 uncached input tokens, the bucket holds 80,000 and refills at 80,000 a minute, so the later calls wait for capacity (hatched). Width 16 reaches 5.50×, not Amdahl's 8.08×, and widening further buys nothing.
At 40,000 the bucket is the whole story: every width from 2 up finishes in 1 min 42 s (1.38×), under the bucket's bound of 1.49×. More parallelism only makes more calls wait.

The lesson generalises: a fan-out is limited by the slower of two ceilings, the serial part of the work and the rate at which the endpoint will accept tokens. Which one binds is a matter of arithmetic you can do before you build it.

Maths

Let $T_1$ be the time one call at a time and $f$ the fraction of it that can run in parallel (everything but the split and the reduce). With $N$ calls at once and nothing else in the way, $T_N = T_1\big((1-f) + f/N\big)$ : Amdahl's law (G. M. Amdahl, Validity of the single processor approach to achieving large scale computing capabilities, AFIPS 1967, doi:10.1145/1465482.1465560), which the engine's tests check the simulation reproduces exactly when the limits are out of reach.

A token bucket with capacity $B$ and refill rate $r$ admits at most $B + r\,t$ tokens by time $t$ . To admit all $\sum_i n_i$ uncached input tokens, the last call cannot start before $T_{\text{bucket}} = (\sum_i n_i - B)/r$ , so $T_N \ge T_{\text{bucket}}$ whatever $N$ is. At 40,000 tokens a minute the calls need 102,840 uncached input tokens, so the bound is (102,840 − 40,000) at 40,000 a minute, 1 min 34 s, against 2 min 21 s one call at a time. The same bound holds for requests and output tokens; the binding one is the largest. The engine's tests also check that no window of the simulated run admits more input tokens than $B + r\,\Delta t$ .

Code

The admission rule of the discrete-event simulation, cut from the vendored engine (des.ts):

if (buckets.rq + EPS >= 1 && buckets.it + EPS >= ni && buckets.ot + EPS >= 0) {

And how the graph runtime turns a router's send leaf into Send packets (graph.ts):

for (const item of (state[g.over] ?? []) as unknown[]) {
  const arg: Obj = { [g.as]: clone(item) };
  for (const k of (g.with ?? []) as string[]) arg[k] = clone(state[k] ?? null);
  out.push({ send: g.send, arg });
}