When the machine holding a piece's bytes is busy, should the scheduler wait for it or start elsewhere?

## The choice, stated precisely A **piece of the input** is one slice that a single **worker thread** — one lane inside a worker process on one machine — reads and processes from start to finish. Suppose the bytes for that slice sit on the local disks of two machines, and both of those machines currently have every lane occupied. A lane is free on a third machine that does not hold the bytes. The scheduler has exactly two moves: - **Wait.** Keep the piece unassigned for a moment and hope a lane frees on a machine that holds the bytes. Cost: idle capacity now, on a lane that could have been working. - **Start now, further away.** Assign the piece to the free lane and let it fetch the bytes over the network. Cost: a remote read that lasts as long as the piece reads, not just at the start. Neither is free, and they are denominated differently — one in idle seconds across the cluster, one in network bytes and read latency for one piece. ## What actually decides it 1. **How long the piece will run.** A piece that finishes in two seconds cannot repay a five-second wait; a piece that runs ten minutes repays it easily. This is the dominant term. 2. **How large the local-versus-remote gap is here.** On a cluster where the network delivers close to local disk bandwidth, there is nothing to wait for. 3. **How likely the holding machine frees up soon.** If it is saturated for the next hour, waiting is a pure loss dressed as an optimisation. 4. **What the idle lane is worth.** On a busy shared cluster, an idle lane is someone else's queued work. On an idle cluster it costs almost nothing, so waiting is cheaper than it looks. ## Giving up in ordered steps A scheduler that models nearness does not make a binary choice; it walks an ordered set of rungs, from nearest to least near, dropping a rung each time a bounded wait expires: | rung | what it means | read cost | |---|---|---| | nearest | a lane on a machine whose own disk holds the bytes | local disk | | middle | a lane on a machine sharing a switch with one that holds the bytes | one hop, contended only within that group | | last | any lane anywhere in the cluster | across the shared fabric | The important property is that the rungs are **ordered and time-bounded**, not that any particular number of them exists. Some clusters have a meaningful middle rung because their network is hierarchical; some have a flat fabric where the middle rung is indistinguishable from the last one, and modelling it buys nothing. ## Where engines genuinely differ - **Some runtimes express an ordered preference and wait briefly at each rung. Others express none** and place every piece on the first free lane, because the deployment they target has no machine that is nearer. - In a **finite job** — a run over an input that ends, so the pieces are derived from measurable stored bytes — this decision is taken per piece, hundreds or thousands of times per run. In a **continuous job** — a run over an input with no end, where the author declares how many copies of each operator run — placement is settled once when the job starts and is not re-decided as records arrive, so there is no per-piece waiting to tune. - The older execution model that materialises every intermediate result to disk between the two halves of a job took this decision at a coarser grain, one round of pieces at a time, rather than continuously. ## The two failure modes - **Waiting too long.** Symptom: the cluster looks busy in the scheduler and idle in the machine metrics. Lanes sit unassigned while pieces queue for a small set of machines, and the run's wall-clock time is dominated by scheduling gaps rather than by work. This is especially sharp when the input's bytes are concentrated on few machines, so many pieces want the same few. - **Never waiting.** Symptom: every read is remote, the shared network saturates, and read bandwidth per lane collapses as parallelism rises — adding lanes makes the run slower, not faster. Both failures look like "the job is slow", and they are distinguished by one observation: whether lanes were idle or busy while the run dragged. ## What an interviewer is listening for That you treat it as an economic trade with a break-even, not as a setting. The strong answer names the quantity that sets the break-even — the piece's expected runtime against the expected wait, scaled by the read-speed gap — and says out loud that on a deployment where no machine holds the bytes, the whole decision is vacuous and waiting is strictly a bug.