Sharing the cluster

The problem

A resource manager serves several goals that pull against one another:

Utilization

Keep expensive hardware busy. Idle machines cost almost as much as busy ones.

Latency

Start and finish work quickly, especially work a user is waiting on.

Fairness

Give each user or team its share, and prevent starvation.

Isolation

Keep one job's spike from slowing its neighbors.

Availability

Spread replicas across failure domains so one rack or zone failing doesn't take a service down.

Scheduling overhead

Decide fast. A perfect placement that takes a minute to compute is often worse than a good one made instantly.

Compared with scheduling on one machine, three things make this harder. The scheduler's view of node load is always slightly stale. Moving work between machines costs network transfer and lost cache state. And any decision-maker can fail.

Allocating resources

Most systems describe work by its requests, the resources reserved for it, and its limits, the most it may use. Placing tasks with several resource dimensions onto machines is a multi-dimensional bin-packing problem. Computing an optimal packing is NP-hard, so schedulers use heuristics such as best fit, worst fit, or scoring functions.

Packing on one dimension at a time wastes the others. Resources left over on a machine that can't be used, because another resource there is exhausted, are called stranded.

Because jobs usually use less than they request, many clusters overcommit: they reserve more than physically exists and rely on priority and preemption to evict low-priority batch work when production services need the room. Production clusters report large gains from this, but it only works with good isolation and usage prediction.

PapersGrandl et al., "Multi-Resource Packing for Cluster Schedulers," SIGCOMM 2014 · Rzadca et al., "Autopilot: Workload Autoscaling at Google," EuroSys 2020
DocsKubernetes requests and limits · Pod priority and preemption

Scheduler architectures

Who makes placement decisions, and with what view of the cluster, is the central design choice.

Centralized scheduling scales further than intuition suggests. Kubernetes and Borg run one logical scheduler for thousands of nodes by caching cluster state and scoring only a sample of the feasible nodes. Decentralization pays off mainly for very short tasks, where scheduling delay dominates.

PapersVerma et al., "Large-Scale Cluster Management at Google with Borg," EuroSys 2015 · Hindman et al., "Mesos," NSDI 2011 · Vavilapalli et al., "Apache Hadoop YARN," SoCC 2013 · Schwarzkopf et al., "Omega," EuroSys 2013 · Ousterhout et al., "Sparrow: Distributed, Low Latency Scheduling," SOSP 2013 · Burns et al., "Borg, Omega, and Kubernetes," CACM 59(5), 2016

Load balancing

Load balancing spreads incoming work, such as requests, connections, or tasks, across servers so none is overloaded while others sit idle.

The last row is one of the most useful results in the field. Sending each task to one random server leaves the busiest server with a maximum load that grows like log n / log log n. Sampling just two servers and picking the less loaded drops that to log log n / log 2, an exponential improvement for almost no extra cost. Because it only needs fresh load for two servers, it works well when global load information is stale.

PapersAzar, Broder, Karlin & Upfal, "Balanced Allocations," SIAM J. Computing 29(1), 1999 · Mitzenmacher, "The Power of Two Choices in Randomized Load Balancing," IEEE TPDS 12(10), 2001 · Karger et al., "Consistent Hashing and Random Trees," STOC 1997 · Mirrokni, Thorup & Zadimoghaddam, "Consistent Hashing with Bounded Loads," SODA 2018
DocsNGINX load balancing · Envoy load balancers · Google SRE book, "Load Balancing in the Datacenter"

Dynamic balancing and migration

Static policies decide placement once, using known information such as task sizes. Dynamic policies react to current load, which matters when work is unpredictable. Two classic dynamic strategies differ in who starts the transfer:

• Sender-initiated: an overloaded node looks for a lightly loaded one to take work. This works well at low to moderate load, but at high load every node is searching and the probes add to the overload.
• Receiver-initiated: an idle node asks others for work. This works well at high load, since idle nodes are rare and their requests are likely to find work.

Work stealing is the receiver-initiated idea refined for parallel runtimes. Each worker keeps a double-ended queue of tasks and works from one end. An idle worker steals from the other end of a random victim's queue, taking the oldest task, which is usually the largest piece of remaining work.

Task migration moves running work from one node to another, from a whole virtual machine to a single process or container. It is worth it only if the expected gain beats the cost:

migrate when  time saved on the new node  >  state size / bandwidth + pause + warm-up

Moving a 4 GB VM over a 10 Gb/s link takes at least 3.2 seconds of transfer, plus time to rebuild caches on the new host. A perfectly balanced cluster is therefore not the goal. Rebalancing too aggressively spends more on monitoring and migration than the imbalance costs, so practical systems tolerate some imbalance and act only past a threshold.

PapersEager, Lazowska & Zahorjan, "Adaptive Load Sharing in Homogeneous Distributed Systems," IEEE TSE 12(5), 1986 · Eager, Lazowska & Zahorjan, "A Comparison of Receiver-Initiated and Sender-Initiated Adaptive Load Sharing," Performance Evaluation 6(1), 1986 · Blumofe & Leiserson, "Scheduling Multithreaded Computations by Work Stealing," JACM 46(5), 1999 · Clark et al., "Live Migration of Virtual Machines," NSDI 2005

Scheduling policies

A scheduler decides the order and location of work. Different metrics favor different policies:

Two more constraints are specific to distributed work. Parallel jobs whose tasks communicate constantly, such as distributed ML training, need gang scheduling: every task starts together or none do. A partially placed job just holds resources while it waits. Long-running services also need spreading constraints that keep replicas on different machines, racks, or zones.

Fair sharing

With one resource, fairness is simple: give each user an equal share, and redistribute what someone doesn't need (max-min fairness). With several resources, users need different mixes. Dominant Resource Fairness (DRF) equalizes each user's dominant share: the largest fraction of any single resource they hold.

DRF has properties that simpler schemes lack. No user prefers another's allocation. No user can gain by lying about their demands. And each user does at least as well as with a static 1/n partition of the cluster. Hierarchical versions apply it per organization, then per team, then per user.

PaperGhodsi et al., "Dominant Resource Fairness: Fair Allocation of Multiple Resource Types," NSDI 2011
DocsSpark job scheduling and fair pools

Locality and stragglers

Data locality means running a task where its input data already lives, to avoid copying it across the network. Strict fairness can fight locality: the job whose turn it is may have no data on the free machine. Delay scheduling resolves this by letting that job skip its turn for a few seconds, waiting for a slot on a machine that holds its data. A short wait buys high locality at almost no cost to fairness.

Stragglers are tasks that run far slower than their peers because of a bad disk, a noisy neighbor, or uneven data. In a job of a thousand parallel tasks, the slowest one sets the finish time. MapReduce introduced speculative execution: near the end of a job, launch backup copies of slow tasks and take whichever finishes first. Interactive services use the same idea as hedged requests, sending a second copy of a request if the first hasn't answered within the 95th-percentile latency.

PapersZaharia et al., "Delay Scheduling," EuroSys 2010 · Dean & Ghemawat, "MapReduce: Simplified Data Processing on Large Clusters," CACM 51(1), 2008 · Dean & Barroso, "The Tail at Scale," CACM 56(2), 2013

Elasticity

Instead of only placing work on a fixed pool, modern systems resize the pool. Horizontal autoscaling adds or removes replicas; vertical autoscaling changes the resources each replica gets. The Kubernetes horizontal autoscaler uses a simple proportional rule:

desired replicas = ceil( current replicas × current metric / target metric )
example: 4 replicas at 90% CPU, target 60%  →  ceil(4 × 90 / 60) = 6 replicas

Autoscalers need damping. Without stabilization windows and cooldowns, they oscillate, adding capacity just as a spike ends and then removing it just before the next. Scaling also lags demand by however long a new replica takes to start, so systems keep headroom or scale on predicted load. Google's Autopilot sets requests automatically from past usage, which reduced slack (reserved but unused resources) in production.

DocsKubernetes horizontal pod autoscaling
PaperRzadca et al., "Autopilot," EuroSys 2020

Trade-offs

In practice

Tools and reading