# Introduction
A lot of systems fan out work to many sub-tasks and then wait for all of them to finish. A build that compiles a few thousand `.cc` files in parallel and then links them. A scatter-gather query that hits 64 shards and joins their responses. A distributed training step that waits for every worker's gradient before applying the update.
That looks something like:
```csharp
await Task.WhenAll(files.Select(DoSomethingReallyCoolAsync));
```
The completion time of the whole thing is the completion time of the slowest sub-task. This post works out how badly that hurts as $k$ grows, and what we can do about it.
# Assumptions
We'll assume for most of this post that there are $k$ sub-tasks with i.i.d. completion times $N_1, \dots, N_k$. This isn't always true in practice, but it's a good starting point for understanding the problem. In real systems, correlated sub-tasks are one reason why things will turn out to be much worse than in this idealized world; mutexes are perhaps one of the easiest ways to see how this can bite you.
We'll also assume $N_i \sim \text{LogNormal}(\mu, \sigma^2)$ (i.e., the completion time of the sub-tasks is [log-normal](https://en.wikipedia.org/wiki/Log-normal_distribution)). Two things make this a reasonable starting point:
1. Measured task and request latencies are almost always positively skewed: most finish quickly, a few run arbitrarily long ("ugh, this random thing hung again!"). A symmetric distribution like the normal can't reproduce that long right tail.
2. There's a mechanism behind the skew. A sub-task's latency is the product of many independent multiplicative effects: queueing, cache misses, network jitter, lock contention. Taking logs turns that product into a sum, so $\ln N_i = \sum_j \ln f_j$ is a sum of many independent terms and is approximately normal by the [central limit theorem](https://en.wikipedia.org/wiki/Central_limit_theorem). That is exactly the statement that $N_i$ is log-normal, a multiplicative CLT sometimes called [Gibrat's law](https://en.wikipedia.org/wiki/Gibrat%27s_law).
Log-normal sits between the light-tailed exponential and genuinely heavy-tailed power laws: heavy enough to throw the occasional brutal straggler, light enough that the maximum keeps all its moments and obeys the clean $\sqrt{2 \ln k}$ scaling derived below. The scaling does depend on that tail. If real sub-task latencies are power-law in the tail (they sometimes are), the maximum grows much faster than this model predicts, so read the log-normal numbers as an optimistic case; bounded or lighter-tailed sub-tasks make it grow slower. The qualitative picture (barriers hurt as $k$ grows, $\sigma$ dominates at scale, partial barriers help a lot) carries over to any reasonable choice.
The difference is easiest to see with both densities drawn together. Below, the normal is matched to the log-normal's mean and variance, so only the shape differs: the normal is symmetric and puts real probability on negative completion times, while the log-normal stays positive and trails off into a long right tail.
The two share a mean and a standard deviation, yet the log-normal's mean, median, and mode all disagree, the signature of a right-skewed distribution:
That skew carries into the CDF: the log-normal crawls toward 1 through its long right tail, while the normal has already spent real probability on negative completion times before the clock starts.
# The effect of maximums
The maximum of $k$ i.i.d. random variables $N_i$ is also a random variable $M$, and its distribution is given by:
$$
\begin{aligned}
F_{M}(x) &= P(\max(N_1, \dots, N_k) \leq x) \\
&= P(N_1 \leq x, \dots, N_k \leq x) \\
&= P(N_1 \leq x) \times \dots \times P(N_k \leq x) \\
&= F(x)^k
\end{aligned}
$$
Where $F$ is the cumulative distribution function (CDF) of $N_i$, so $F_M$ is the CDF of $M$. Already from this we can see why the maximum is painful: at every fixed $x$ with $F(x) < 1$, $F(x)^k \to 0$ as $k$ grows, so $M$ concentrates on larger and larger values. The more sub-tasks we have, the more likely at least one of them takes too long and drags the maximum upwards.
This equality means we can get a closed-form expression for the CDF of the maximum of $k$ log-normal random variables:
$$
F_{M}(x) = \left[ \Phi\left( \frac{\ln(x) - \mu}{\sigma} \right) \right]^k
$$
Where $\Phi$ is the CDF of the standard normal distribution. Here's what that looks like:
We can use this to get the PDF of the maximum of $k$ log-normal random variables by differentiating, but it's a bit of a mess and doesn't help for things we'd like to do like computing expected values. Looking into this led me down the [extreme value theory](https://en.wikipedia.org/wiki/Extreme_value_theory) rabbit hole, as you'll see below.
# Expected value of the maximum
As it turns out, the PDF is messy and a closed-form expected value isn't pretty either, but we can simulate it. Below is a Monte Carlo estimate of $\mathbb{E}[M]$ along with the 68% and 95% central ranges of $M$'s distribution (the 16th/84th and 2.5th/97.5th percentiles of $M$ for each $k$), for $k$ ranging from 1 to $k_{\max}$:
Two things to notice:
1. The expected value grows monotonically with $k$, at a decreasing rate.
2. The spread also grows with $k$.
[Extreme value theory](https://en.wikipedia.org/wiki/Extreme_value_theory) tells us the log-normal distribution sits in the [Gumbel domain of attraction](https://en.wikipedia.org/wiki/Generalized_extreme_value_distribution), and after suitable centering and scaling, the maximum of $k$ i.i.d. samples converges to a Gumbel distribution as $k \to \infty$ (yeah, I know, just wtf?).
The underlying normal $X_i = \ln N_i \sim \mathcal{N}(\mu, \sigma^2)$ has a well-known leading-order asymptotic[^leading-order] for its maximum:
[^leading-order]: "Leading order" pins down _how fast_ the maximum grows with $k$, not its value at the $k$ you actually run: the relative error vanishes as $k \to \infty$, but the absolute gap need not be small. So basically: it's a ballpark estimate rather than an approximation.
$$
\mathbb{E}\left[\max_i X_i\right] \approx \mu + \sigma \sqrt{2 \ln k}
$$
Exponentiating, the typical value of $M$ scales like
$$
\exp\left( \mu + \sigma \sqrt{2 \ln k} \right)
$$
to leading order.[^sqrt-2lnk] There are $O(1/\sqrt{\ln k})$ corrections involving $\ln \ln k$ (visible in the Monte Carlo plot above as a meaningful gap at finite $k$), but the scaling is what tells the story.
[^sqrt-2lnk]: $\sqrt{2 \ln k}$ is the level a single one of $k$ standard normals is expected to exceed: set $k\,(1 - \Phi(b)) = 1$ and use the Gaussian tail $1 - \Phi(b) \approx \varphi(b)/b$, which gives $b = \sqrt{2 \ln k}$ to leading order. It is the centering sequence in the [Fisher-Tippett-Gnedenko theorem](https://en.wikipedia.org/wiki/Fisher%E2%80%93Tippett%E2%80%93Gnedenko_theorem) for the normal; the full expansion subtracts the $\ln \ln k$ terms mentioned above.
The exponent grows like $\sqrt{\ln k}$, so the maximum grows _unboundedly_ in $k$, but quite slowly. There are a few practical takeaways from this:
1. The variance is way more important than the mean: $\sigma$ dominates completion time as soon as $k$ is large enough that $\sigma\sqrt{2\ln k} > \mu$. In that regime, the same absolute reduction in $\sigma$ shaves $\sqrt{2\ln k}$ times as much off the exponent as the same reduction in $\mu$, and the gap grows with $k$.
2. The maximum only ever moves up, and adding more parallel sub-tasks has diminishing returns on latency.
3. A single rare slow sub-task ruins everything. The completion time is extremely sensitive to the tail of the per-task distribution.
# What if we don't need all of them?
The full barrier (wait for all $k$ sub-tasks) is the worst case. In practice, we often don't need all sub-tasks to complete: we can tolerate some number $c$ of stragglers, and use only the $k - c$ fastest results. This is the [order statistic](https://en.wikipedia.org/wiki/Order_statistic) generalization of the maximum.
If we sort the sub-task completion times $N_{(1)} \leq N_{(2)} \leq \dots \leq N_{(k)}$, the time to get the first $j$ of them done is $N_{(j)}$, whose CDF is:
$$
F_{(j)}(x) = \sum_{i=j}^{k} \binom{k}{i} F(x)^i (1 - F(x))^{k-i}
$$
When $j = k$ this collapses back to $F(x)^k$ (the maximum). The case $j < k$ is where it gets interesting:
The plot above shows the expected wait time as a function of $k$ for three "barrier fractions": waiting for 100% of sub-tasks (the full barrier), waiting for the fastest 95%, and waiting for the fastest 50% (the median):
- The full barrier grows roughly like $\exp(\sigma \sqrt{2 \ln k})$, as discussed above.
- Tolerating a 5% straggler tail caps the wait at a much lower value: each additional sub-task gives a few more "outs", which together absorb the tail.
- Tolerating a 50% straggler tail makes the wait time roughly constant in $k$. As $k \to \infty$, the sample median converges to the population median $e^\mu$, which doesn't depend on $\sigma$ at all.
This idea shows up all over the place, with one famous case being in distributed training. Synchronous SGD waits for all $N$ workers' gradients before stepping, so the step time is the max of $N$ and one slow worker stalls everyone. [Chen et al. (2016)](https://arxiv.org/abs/1604.00981) add $b$ backup workers and step as soon as the fastest $N$ of the $N+b$ report, dropping the slowest $b$. The step time becomes the $N$-th order statistic of $N+b$ instead of the max of $N$, and because they still aggregate exactly $N$ gradients the effective batch (and its variance) is unchanged (at the cost of the $b$ gradients computed and discarded each step).
# What can we do about it?
If the system has a hard barrier with no $c$ to spare, the previous trick is unavailable. There are still a few things that can be done.
## Reduce σ
Since the maximum grows in $\exp(\sigma \sqrt{2 \ln k})$, shaving $\sigma$ has outsized impact. At $k = 10{,}000$ we have $\sqrt{2 \ln k} \approx 4.3$, so every $0.1$ trimmed from $\sigma$ divides the typical maximum by $e^{0.43} \approx 1.5$, while the same $0.1$ off $\mu$ only divides it by $e^{0.1} \approx 1.1$.
Common causes of variance in sub-task completion times include GC pauses, cold caches, head-of-line blocking, mutex contention, network jitter, and noisy neighbors. Hunting them down is usually the highest-leverage thing you can do[^spc].
[^spc]: And a great tool for this is [Statistical Process Control](https://en.wikipedia.org/wiki/Statistical_process_control). I really enjoy anything written by [Donald J. Wheeler](https://spcpress.com/djw_columns.php) to learn about this.
## Reduce k
Either through batching (process many items per sub-task) or by hierarchical fan-out (splitting a $k$-way fan-out into a $\sqrt{k}$-way fan-out of $\sqrt{k}$-way fan-outs). Batching is the one that actually moves the max, because it cuts $k$ directly:
```csharp
// One task per file becomes one task per batch of 16: k drops 16x,
// at the cost of each task now doing 16x the work.
await Task.WhenAll(files.Chunk(16).Select(CompileBatchAsync));
```
Hierarchy mainly buys you bookkeeping (fewer children per node, easier failure handling); for pure latency it only helps when each intermediate does its own non-trivial work, since the global max in a routing-only tree is still the max of all $k$ leaves regardless of tree shape.
## Hedge requests
If a sub-task can be served by multiple replicas, send the request to $r$ of them and take the first response:
```csharp
// Race r replicas, take the first answer, cancel the losers.
async Task HedgedAsync(IReadOnlyList replicas)
{
using var cts = new CancellationTokenSource();
var inFlight = replicas.Select(r => r.QueryAsync(cts.Token)).ToList();
var winner = await Task.WhenAny(inFlight);
cts.Cancel();
return await winner;
}
```
Assuming the per-replica latencies are independent, this replaces the per-replica distribution $F$ with $1 - (1 - F)^r$, which _squares_ (for $r = 2$) the tail probability. The downside is sending $r \times$ the load to the underlying system, which is rarely free. A common compromise is to wait until some quantile (e.g., p95) before issuing the hedged request, which dramatically shortens the tail while sending only ~5% extra load:
```csharp
// Cheaper: only hedge the requests that have already gone slow.
async Task HedgeAfterAsync(Replica primary, Replica backup, TimeSpan p95)
{
var first = primary.QueryAsync();
if (await Task.WhenAny(first, Task.Delay(p95)) == first)
return await first; // primary beat the p95 deadline
using var cts = new CancellationTokenSource();
var second = backup.QueryAsync(cts.Token);
var winner = await Task.WhenAny(first, second);
cts.Cancel();
return await winner;
}
```
Dean and Barroso's [The Tail at Scale](https://research.google/pubs/the-tail-at-scale/) is the canonical reference for this. Simulating all three strategies shows how much the barrier moves: full hedging ($r = 2$) flattens it almost completely, and the p95 compromise recovers most of that benefit for a fraction of the extra load.
## Speculatively re-issue stragglers
A variant of hedging for when the sub-tasks aren't naturally replicated. If a sub-task is taking longer than expected (e.g., past some quantile of the distribution), re-issue it to a different worker and take whichever response comes first. It's the same race as `HedgeAfterAsync` above, except the backup goes to a fresh worker rather than a replica, so the hedge plot describes it too. This is the trick [MapReduce](https://research.google/pubs/mapreduce-simplified-data-processing-on-large-clusters/) uses to deal with straggler tasks.
# Conclusion
Barriers turn the per-task tail latency into the dominant cost of fan-out, and the cost grows with the number of sub-tasks. The math says the growth is slow ($\sqrt{\ln k}$ in the exponent for log-normal), but in practice that's still painful: doubling $k$ pushes the maximum up, and any additional variance shows up amplified at the barrier.
The two best mitigations I've found in practice are (a) hunting down sources of variance in the per-task distribution, and (b) avoiding the full barrier when at all possible. "$k - c$ of $k$" is often sufficient and dramatically cheaper; everything else (hedging, speculative re-execution, hierarchical fan-out) is downstream of those two.