Work Distribution, Join Barriers, and Idle Time at the End of a Stage

Parallelism is the standard response to a stage that takes too long, and it works on the mean in the way the arithmetic suggests: divide the work into $n$ parts and the stage takes about $1/n$ of the time. On the tail it does the opposite. The stage now takes as long as its slowest part, and the maximum of $n$ samples from any distribution is larger than a single sample, so the tail of the parallel stage is worse than the tail of the serial one even as its mean improves.

The two effects are not in tension; they are both consequences of the same structure, and the structure is what has to be measured before the parallelism is added.

The maximum of n samples

If each of $n$ workers independently draws a duration from a distribution with cumulative function $F$, the probability that all of them finish by time $t$ is $F(t)^n$, so the distribution of the stage’s completion time is $F^n$. Its median is the value $t$ where $F(t)^n = 0.5$, which is the $(0.5^{1/n})$-th quantile of the single-worker distribution.

For $n = 10$, $0.5^{1/10} \approx 0.933$, so the stage’s median is the single worker’s 93rd percentile. For $n = 100$, $0.5^{0.01} \approx 0.993$, so the stage’s median is the single worker’s 99.3rd percentile. This is the fan-out arithmetic that makes a backend p99 into a frontend median, and it is why adding parallelism to a stage with a heavy-tailed service distribution can make the user-visible latency worse even when the stage’s mean time falls.

The mean, by contrast, does what intuition expects when durations are independent and the maximum is not dominated by a rare extreme: the expected maximum of $n$ exponential variables with mean $\mu$ is $\mu H_n$, where $H_n$ is the $n$-th harmonic number, so the stage takes $\mu H_n$ rather than $\mu$. For $n=10$ that is about 2.9 times a single worker’s mean, compared to 10 times for the serial version, so parallelism still wins on the mean by a factor of roughly 3.4. It wins by much less than the naive factor of 10, and the gap is the price of the barrier.

Where the variance comes from

Independent, identically distributed workers are the assumption that makes the arithmetic tractable and the assumption that real systems violate. The variance in completion time has several sources, and they behave differently under splitting.

Work-size variance is intrinsic to the input. Splitting a fixed total into $n$ equal parts removes it; splitting into $n$ parts by a key or a hash preserves whatever imbalance the key distribution has, so a partition that happens to contain twice the data takes twice as long. The distribution of partition sizes, not the mean partition size, determines the stage’s completion time.

Resource variance is environmental. Workers share memory bandwidth, cache, network, and storage, so a worker that happens to be co-scheduled with others runs slower. This source grows with the number of workers, which means adding parallelism increases the variance it is trying to average out. The effect is the contention side of the throughput problem, and it explains why a stage split across 64 threads on 8 cores is often slower than the same stage split across 8.

Scheduling variance is the queue. A worker that is dispatched to a busy host waits for a slot, and that wait is not part of the work but is part of the completion time. With a shared queue the wait is small and uniform; with per-worker queues the wait depends on which worker the task landed on, and the landing is often the largest single contributor to a straggler.

Straggler events are the fourth source and the one that dominates tails in practice. A worker that is descheduled, that hits a page fault, that waits on a garbage collection, or that loses a network packet can take an order of magnitude longer than its peers without doing any extra work. These events are rare per worker and common per stage, because the stage waits for all workers.

The barrier converts one slow worker into a slow request

A join barrier waits for every participant. That makes the request’s latency the maximum of the participants’ latencies, with no partial credit. If one participant is delayed by a 200-millisecond garbage collection pause, the request takes 200 milliseconds longer regardless of how quickly the other 99 finished.

The consequence is that the request’s latency distribution inherits the union of all the failure modes of all its participants. With 100 participants, a per-participant probability of a rare delay multiplies by 100 at the request level. A one-in-a-thousand per-worker straggler becomes a one-in-ten request-level straggler.

The mitigation strategies all trade against something. Redundant execution, issuing the same task to two workers and taking the first result, reduces the tail at the cost of doubled work; the reduction is substantial because the probability that both copies are slow is the square of the per-copy probability. Speculative execution, issuing a second copy only after a delay, does the same with less average cost but a fixed latency floor equal to the speculation delay. Both are responses to variance rather than to mean, and neither helps if the variance is systematic, as when one host is genuinely slower for all of its work.

The third strategy is to shorten the barrier: to return partial results as they arrive, so the request’s latency is the time to the first or the best $k$ results rather than the time to all. That changes the semantics of the operation, and it is only available when the caller can use an incomplete answer. Search is the classic case; a transactional write is not.

Unequal work and the idle tail

A stage with $n$ workers and a fixed total amount of work has a natural inefficiency at its end: when the last worker finishes, the others have been idle. The fraction of the stage’s duration during which the average worker is busy is the stage’s efficiency, and for a fixed number of equal-cost tasks it is the ratio of the number of tasks to the number of workers times the number of rounds.

With 100 tasks on 8 workers, the schedule runs 13 rounds; the last round has 4 tasks on 8 workers, so half the capacity is idle for one round out of 13, and the efficiency is $100/(8 \times 13) = 96$ percent. With 17 tasks on 8 workers, the schedule runs 3 rounds with the last round using 1 worker, giving $17/24 = 71$ percent. The efficiency depends on the remainder, not on the scale, so a fixed partition into a number of tasks that is not a multiple of the worker count wastes up to almost a full round.

Dynamic scheduling, where each worker takes the next task when it finishes rather than being assigned a fixed set, removes this specific waste and reduces the impact of work-size variance, because a worker that gets a large task does not leave a fixed gap. It does not remove the tail, since the last task still determines the completion time, and it does not remove stragglers. It does convert the waste from a deterministic remainder into a random one, which is better on average and harder to predict.

Measuring the stage rather than the worker

The measurement that matters for a parallel stage is the distribution of the stage’s completion time, not the distribution of the workers’ task durations. The two have different shapes by construction, and a dashboard of worker durations systematically understates the stage’s tail.

Three quantities make the gap visible. The stage’s elapsed time against the maximum worker time shows whether the barrier is the constraint. The sum of worker times against the stage’s elapsed time times the worker count gives the utilization of the stage, which reveals the idle tail. The count of stage completions against the count of worker tasks gives the fan-out, which is needed to interpret the other two.

import time
from concurrent.futures import ThreadPoolExecutor

def stage(tasks, workers):
    """Run a barrier stage and report the split between work and waiting."""
    t0 = time.monotonic()
    with ThreadPoolExecutor(max_workers=workers) as pool:
        futures = [pool.submit(t) for t in tasks]
        results = [f.result() for f in futures]   # barrier: waits for all
    elapsed = time.monotonic() - t0
    busy = sum(f.duration for f in futures)       # instrumented per task
    return {
        'elapsed': elapsed,
        'max_task': max(f.duration for f in futures),
        'busy_fraction': busy / (elapsed * workers),  # 1.0 means no idle tail
        'fanout': len(tasks) / workers,
    }

The busy_fraction term is the direct measure of the idle tail. A value near 1 means the stage is work-bound and adding workers helps; a value well below 1 with a max_task close to elapsed means the stage is straggler-bound, and adding workers will not help because the slowest task sets the duration regardless of how many are running.

The interaction with the queue

Parallelism changes the queueing behaviour of the stage in a way that is easy to miss. A stage that uses $n$ workers on a host with $c$ cores, where $n > c$, has more runnable threads than cores, so each worker’s execution is interleaved with the others’ and the time to complete any one task rises. The stage’s throughput may improve while each task’s latency worsens, and the barrier converts that per-task worsening directly into stage latency.

The operating point that matters is the ratio of workers to cores, not the ratio of work to workers. Running at more workers than cores is a bet that the tasks are I/O-bound enough to block, which is true for network calls and false for computation. When the bet is wrong, the stage’s latency curve acquires the steep shape of a saturated resource, which is the same nonlinearity described in Utilization and the Nonlinear Growth of Waiting Time, and the added parallelism buys throughput at the cost of the very tail it was meant to reduce.

The other interaction is with the request-level queue. If each request spawns $n$ workers, then $r$ concurrent requests produce $rn$ runnable tasks, and the host’s scheduler sees a load that is $n$ times what the request rate suggests. The queue depth per core is the quantity that governs scheduler delay, and it is invisible in a per-request view. The relationship between outstanding requests and time in system is fixed by Little’s Law, treated in Little’s Law: Concurrency, Arrival Rate, and Time in System, but the identity says nothing about how that time is distributed across the workers a request occupies.