Memory pressure signalling and operator cooperation

Why can the manager only ask operators to spill, and how do cooperative protocols fail?

You cannot preempt memory, so the only lever left is to ask.

Why it must be cooperative rather than authoritative

The memory manager knows how many bytes are outstanding and who holds them. It does not know which bytes are cheap to give back. Is this region a fully merged sort run that can simply be dropped? Or is it a half-built hash table whose loss costs four minutes of work? The operator knows. A manager that chose victim bytes itself would be swap, and every argument against swap from Memory is a space resource applies. So the manager publishes the shortage and the holders decide.

This is this mechanism's theme for the third time. First, kernel to engine: the engine builds a buffer pool over the OS page cache. Second, engine to operator: ring buffers over a replacement policy. Third, now: manager to consumer, cooperative spill over forced eviction. Each upper level holds information the level below cannot get. Each level above responds by turning a decision into a request. (This is the take-back rule: the application takes the policy back when it holds information the layer below lacks.)

The protocol shape

Every system that has this protocol follows the same pattern:

1. Consumers register with the manager and report current usage.
2. On an allocation failure, the manager calls spill(N) on consumers, typically largest first.
3. Each consumer returns the number of bytes it actually freed, which may be fewer than N.
4. If cooperation is not enough, something gets killed.

# Sketch of the protocol (not a real engine)
def allocate(requester, nbytes):
    while manager.free < nbytes:
        victim = max(manager.consumers, key=lambda c: c.used)  # largest first
        freed = victim.spill(nbytes - manager.free)            # may free less than asked
        if freed == 0:
            raise OutOfMemory(requester)                       # nobody can help: fail or kill
    manager.reserve(requester, nbytes)

Spark's TaskMemoryManager and MemoryConsumer.spill() are exactly this. In Spark 2.x the cooperation scope is within one task: when an acquisition fails, other consumers of the same task are asked to spill. Between tasks, arbitration is the fixed 1/2N to 1/N fair share from Memory grants and reservations per operator. So one fat task cannot be made to give way to a starving neighbour. It simply keeps what it acquired. A fixed share is arbitration that never negotiates, and the price is that a workload with uneven task footprints (which is most real workloads) leaves memory stranded.

Four failure modes of cooperative protocols

These generalize well beyond memory.

1. Non-participants. Any holder that is not registered with the manager is invisible and cannot be asked to spill: a UDF's own data structures, a third-party library's buffers, the runtime's own overhead. A cooperative accounting system is only as accurate as its least cooperative participant, and in a system that allows user code, an uncooperative participant always exists. This, not an arithmetic error, is where most real OOMs come from: the accounting was correct about everything it could see. (This is Dmitri's third puzzle, the UDF.)
2. The response is slow; the shortage is immediate. Writing gigabytes to disk takes time, but the allocation that triggered the request needs memory now. So either the requester blocks, which creates a latency cliff and the possibility of two consumers each waiting for the other (a deadlock), or you signal pressure before it is critical, using a reserve. This gives two watermarks: a soft one (begin asking consumers to spill) and a hard one (stop admitting new work, then kill). The reserve buys time for the answer. Its size is set by how long a spill takes, not by how much memory you have. Worked example. If memory is consumed at 2 GB/s and a spill takes 3 s to free enough space, you need about 2 GB/s × 3 s = 6 GB of reserve between the soft watermark and the point of failure. A bigger machine with the same allocation rate and the same spill time needs the same 6 GB.
3. Oscillation. A control loop with a single threshold spills, frees memory, refills and spills again. It needs hysteresis: a gap between the "start" threshold and the "stop" threshold (for example, start spilling at 90% used and stop at 75%). Autoscaling that flaps up and down and congestion control without damping fail for the same reason.
4. Fairness inversion in victim selection. Largest first recovers the most memory per request and destroys the most accumulated work. Youngest first destroys the least work and may not free enough to help. Presto's node-level behaviour of killing the largest query is the honest extreme: maximum recovery, maximum waste. There is no correct answer. The cost is relocated onto whichever query you choose, not reduced, because costs are conserved.

One-line compression: you cannot preempt memory so you ask, and a cooperative protocol is only as accurate as its least cooperative participant. The reserve you hold back is what pays for the latency of the answer.