Work stealing, scheduling and speculation

Who gets the next piece of work, when is that decided, and when does running a duplicate pay off?

Work stealing

The problem. Task durations are unknown and unequal. So any a priori assignment of n tasks to p workers is imbalanced by an amount you cannot predict.

Two obvious designs, both flawed

• A central queue that all workers pull from solves imbalance perfectly. But it is a shared resource, and it queues badly: waiting grows as 1/(1 − ρ) (the "queues explode near full load" rule). This is fine when each task takes a millisecond. It is fatal when each unit takes a hundred microseconds, as with morsels (the small chunks of rows in morsel-driven execution).
• Per-worker queues remove the contention but bring back the imbalance.

Work stealing takes the good part of both. Each worker owns a deque (a double-ended queue). It pushes and pops at its own end, with no locking in the common case. When its deque is empty, it picks a random victim and steals from the other end of the victim's deque.

worker loop:
  while true:
    t = my_deque.pop_bottom()          # own end, no lock in common case
    if t is none:
      v = random_worker()              # random victim
      t = v.deque.steal_top()          # opposite end: oldest, largest task
    if t is none: continue/backoff
    run(t)                             # may push new subtasks on my own bottom

Two properties make it strong

1. It is provably good. Randomized work stealing achieves time about T₁/p + O(T∞): the work-span bound within a constant factor, with no central coordinator and no global information. In many places across the platform randomization replaces coordination, and the evidence is empirical. Here there is a theorem. This is the principle randomization instead of coordination at its strongest.
2. The direction of the steal is derived, not arbitrary. You steal from the far end because it holds the oldest and largest task, which is also the one least likely to be in the victim's cache. A large task means more work per steal, so fewer steals are needed.

The costs are locality and determinism.

• A stolen task runs on a core with a cold cache, and possibly on another NUMA node (a CPU socket whose memory is farther away). The real price of a steal is the cache refill. This is why morsel-driven schedulers steal from workers on the same NUMA node first, and cross sockets only as a last resort.
• Execution order becomes run-dependent. Performance is then not reproducible. Where the reduction is not associative, and floating-point addition is not exactly associative, even the results can differ between runs (for example (a + b) + c can differ from a + (b + c) in the last digits). Parallelism costs order even when nobody asked for an ordering.

The limit that matters most: a steal cannot split a task. Work stealing corrects the assignment of units. It cannot correct a unit that is itself too large. So it fixes imbalance and does nothing for skew.

Static versus dynamic task scheduling

This is the "when is it decided?" question. Derive it from the value of information rather than describing it as a style preference.

Static versus dynamic scheduling is a choice between simplicity and adaptivity. Static scheduling fixes the assignment of tasks to workers before execution, using the planner's estimates. What it buys is co-optimization: all decisions are made together, with global knowledge of the plan. You can place tasks near their data, gang-schedule stages that must run at the same time, reserve memory coherently, and pay zero runtime overhead.

Dynamic scheduling delays each assignment until a worker is free, and uses observations instead of estimates. What it buys is correction. What it costs is a round trip per decision and the loss of global co-optimization, because each decision is locally greedy (the later decisions have not happened yet).

The value of dynamism equals the estimate's error times the cost of being wrong. Because cardinality error (the planner's error in estimating how many rows a step outputs) compounds with join depth, the error term grows with plan depth. So dynamic scheduling pays more as plans get deeper and queries get longer, and is worth almost nothing for a shallow, short query, where static is both cheaper and better.

The cost side is governed by granularity. A decision that costs 1 ms is negligible against a 1-second task and absurd against a 1-microsecond one. The rule is: the granularity of adaptivity must match the cost of the decision. That is why cluster schedulers adapt per task, morsel schedulers adapt per morsel with an almost free decision, and nothing adapts per tuple.

The execution model chooses for you. This is the non-obvious link to Across machines:

• Streamed stage edges need every stage resident at once, so the degree of parallelism must be fixed before launch, so the scheduler must be static.
• Materialized edges let you run one stage at a time, so the next stage's DOP (degree of parallelism) can be chosen from the previous stage's measured output, so the scheduler can be dynamic.

The execution model determines which scheduling model is even available. Pipelined MPP engines are static because they are pipelined, not because their authors liked static scheduling.

Control flow = flow control, once more. Dynamic scheduling is pull: the worker asks for work when it is free, so workers pace themselves and the coordinator needs no model of their capacity. Static assignment is push: the coordinator decides, so it must model capacity it cannot observe, and it is wrong in exactly the way the principle predicts. (Recall: the party that decides when the next unit moves is the only party that can refuse it.)

Speculative execution

The problem. Task duration has a tail, and elapsed time lives in the tail. From Three idlenesses, E[max]/E[mean] ≈ 1 + c·√(2 ln n) for light tails, and worse for real ones. Speculation attacks the tail directly: when a task looks late, start a duplicate of it and accept whichever copy finishes first.

The economics. Speculating the slowest fraction q of tasks costs q extra capacity and buys a latency reduction that is bounded by the tail's excess.

• At low utilization, the extra capacity was idle anyway, so speculation is nearly pure gain.
• At high utilization, the capacity is not free. It is taken from other queries. Response time near saturation goes as 1/(1 − ρ), so adding duplicate work when ρ is close to 1 makes the cluster slower in total than the latency it saves.

Speculation is a trade of throughput for latency, and it turns negative-sum precisely when the cluster is busy. That is also when people tend to turn it on. A mechanism that looks purely beneficial in isolation is constrained by the queue formula.

Detection is where naive implementations fail. A rule such as "slower than 1.5× the median elapsed time" cannot tell a task that is slow from a task that is big. On skewed data the naive rule speculates the largest task. The duplicate takes just as long, because it has just as much work, and you have paid double capacity for nothing. The correct target is estimated remaining time: speculate the tasks that will finish last. That needs progress measurement (for example, fraction of input bytes consumed), not elapsed time. Straggler detection is an attempt to infer a cause from a symptom that two different causes share.

Two preconditions on everything else

1. Tasks must be side-effect-free, because two copies will run.
2. Output must be commit-arbitrated: exactly one attempt's output is published. This is the principle commit arbitration. When two attempts write the same output, one designated atomic act decides which one counts. Atomicity over the task's many written bytes is implemented as atomicity over one designated commit (for example, an atomic rename or a registry entry), and the commit protocol exists only to supply it.

A correctness corollary that is often missed. If a task is non-deterministic and a downstream consumer has already read part of the losing attempt's output, the two attempts' outputs are not interchangeable, and the result is silently wrong. Speculation requires determinism, and that is a correctness requirement, not a performance one.

Related idea: hedged requests (sending a second copy of a slow request to another server) are the same mechanism at the level of a service. We only mention them here.

Skew: the problem that speculation and stealing cannot solve

Speculation fixes slow. Work stealing fixes misassigned. Neither can fix big. Skew needs its own mechanisms because its cause is different in kind: the work unit itself is too large, and no reassignment or duplication divides it. Skew and elasticity develops those mechanisms. Here we only note the boundary, because it is why Mateo's doubled cluster did not help.