The memory arbitrator — interior mutability

A buffer spills when memory is tight, but “memory” is a single budget shared by every operator running at once, each on its own thread, each charging and releasing bytes. They need one shared object that tracks the total and decides who spills. Here Rust’s ownership model seems to fight you: you share a value across threads with Arc, and Arc gives only shared (&) access, never &mut. So how does anyone update the counters? The answer is interior mutability: a type built to mutate through &self, using atomics, a copy-on-write snapshot, and a condition variable. The MemoryArbitrator is the engine’s masterclass in it. We use just enough of the Rust mechanism to read the real type, and The Rust Book, ch. 15 is the canonical treatment of interior mutability and RefCell if you want it from first principles.

Prerequisites

• Node buffers & spill

→

Leads to

• Bounded memory in action

What you’ll be able to do

• Explain why an Arc-shared type must mutate through &self, and name interior mutability as the mechanism that allows it.
• Distinguish the interior-mutable primitives by the access pattern each fits: Cell/RefCell (single-threaded), Atomic*, Mutex, ArcSwap, Condvar.
• Read the arbitrator’s atomic counters and copy-on-write registry in real engine signatures and say what makes each write safe without a global lock.
• Write a shared counter that four threads mutate through &self, and predict the compile error when you swap the atomic for a plain field.

New terms in this lesson (each is also expanded inline at the point you first need it):

• interior mutability: mutating through a & instead of a &mut.
• Cell / RefCell: the single-threaded form of the same idea.
• atomic: a lock-free, hardware-safe shared counter or flag.
• Mutex: a lock that serializes access to whatever it guards.
• Condvar: parks a thread until another signals, with no CPU spin.
• arbitrator: the engine’s single shared memory-budget authority.

Predict first

✋ Predict first

The MemoryArbitrator is Arc-shared across every operator thread, and Arc<T> only ever hands out &T, never &mut T. Yet add_bytes(&self, n) clearly RAISES a counter, and observe(&self) RAISES a high-water mark. How can a method that takes &self mutate the struct's fields at all, and what makes that not a data race? Write your one-sentence theory before revealing.

Decide on your answer before you open the reveal — committing a guess is what makes it stick.

Reveal the answer

The fields are interior-mutable types: each one turns a shared & into a safe write on its own, so the &self method never needs a &mut.

• A plain u64 field can only be written through &mut self, which Arc will never give you.
• An AtomicU64 field can be written through &self: fetch_add is a single lock-free CPU instruction the hardware makes indivisible, so many threads racing on it is safe by construction, not a data race.
• The borrow checker’s “one writer at a time” rule hasn’t been broken; it’s been moved. Instead of the compiler proving exclusivity, the hardware (for atomics) or a run-time lock (for Mutex) enforces it.

The wrong mental model to drop: “&self means read-only, so the method can’t really be changing anything.” &self means shared, not immutable, and interior-mutable types are exactly the ones that exploit that gap. Hold that distinction; every type below is one way of filling it.

The &self problem

Share a value across threads and you reach for Arc<T>. But Arc<T> only ever hands out &T, never &mut T, because two threads holding &mut to the same value is precisely the data race the borrow checker forbids. So a shared budget tracker can only offer &self methods. To mutate through them, its fields must be interior-mutable types, ones whose own API turns a & into a safe write. That’s the entire trick behind Atomic*, Mutex, RwLock, and ArcSwap.

The family splits on one axis: does it cross threads?

• Cell<T> / RefCell<T> are the single-threaded form. Cell swaps whole values through &self; RefCell hands out a borrow checked at run time and panics if you break the one-writer rule. Neither is thread-safe, so the arbitrator can’t use them, but they’re the simplest way to see the idea.
• Atomic*, Mutex, RwLock, and ArcSwap are the thread-safe forms, the ones a value crossing threads behind an Arc must use.

The arbitrator is Arc-shared across every operator thread, and its fields are exactly these thread-safe ones:

clinker-exec ·memory.rs ·MemoryArbitrator type @19acdcb4

pub struct MemoryArbitrator {
    limit: AtomicU64,                  // the hard limit, in bytes
    peak_rss: AtomicU64,               // highest process RSS observed
    cumulative_spill_bytes: AtomicU64, // total bytes spilled to disk
    // copy-on-write registry of operators to poll / pause / spill:
    consumers: ArcSwap<Vec<(ConsumerId, Arc<dyn MemoryConsumer>)>>,
    policy: Box<dyn ArbitrationPolicy>,
    // ... more atomic counters
}

Every counter is an AtomicU64; the operator registry is an ArcSwap (a lock-free, copy-on-write cell). None of these needs &mut to change, and that’s what lets one Arc<MemoryArbitrator> be read and updated by all the threads at once.

 thread (sort op)   ─┐
 thread (agg op)    ─┼──▶  Arc<MemoryArbitrator>
 thread (join op)   ─┤        limit, peak_rss, cumulative_spill : AtomicU64
 thread (source)    ─┘        consumers : ArcSwap<Vec<…>>   (lock-free snapshot)
        every thread holds &self; updates go through the atomics — no &mut, no global lock

Mutating through &self

Watch the signatures: these methods change the arbitrator’s state, yet every one takes &self. The mutation rides an atomic operation:

crates/clinker-exec/src/pipeline/memory.rs

pub fn observe(&self) {
    if let Some(rss) = rss_bytes() {
        self.peak_rss.fetch_max(rss, Ordering::Relaxed);  // raise the high-water mark
    }
}

The per-operator byte counter lives on a ConsumerHandle (an Arc-shared handle each operator holds) and charges/releases bytes the same lock-free way:

clinker-exec ·memory.rs ·ConsumerHandle type @19acdcb4

pub fn add_bytes(&self, n: u64) {   // &self — charge memory from any thread
    let _ = self.bytes.fetch_update(Ordering::Relaxed, Ordering::Relaxed,
        |cur| Some(cur.saturating_add(n)));
}
pub fn sub_bytes(&self, n: u64) {   // &self — release it again
    let _ = self.bytes.fetch_update(Ordering::Relaxed, Ordering::Relaxed,
        |cur| Some(cur.saturating_sub(n)));
}

Why atomics rather than wrapping the whole thing in a Mutex? Because charging bytes happens on the hot path, per batch, from many threads. A fetch_add is a single lock-free CPU instruction; a Mutex would serialize every operator through one lock. The operator registry uses ArcSwap for the same reason: the frequent readers (which operator is using what?) load an immutable snapshot with no lock, while the rare register/unregister clones the Vec and atomically swaps it in. Hot path lock-free; rare path copy-on-write.

The pause signal: a Condvar in miniature

Not all coordination is a counter. When the arbitrator decides to pause a source for backpressure, the source’s thread has to actually block until resumed, and busy-spinning would burn a core. That’s what a condition variable is for. The PauseSignal pairs an AtomicBool flag with a Mutex + Condvar:

clinker-exec ·memory.rs ·PauseSignal type @19acdcb4

pub struct PauseSignal { paused: AtomicBool, mu: Mutex<()>, cv: Condvar }

impl PauseSignal {
    pub fn resume(&self) {
        self.paused.store(false, Ordering::Release);
        self.cv.notify_all();              // wake every parked waiter
    }
    pub fn wait_while_paused(&self) {
        if !self.is_paused() { return; }   // lock-free fast path when not paused
        let mut g = self.mu.lock().unwrap();
        while self.is_paused() {
            g = self.cv.wait(g).unwrap();  // park the thread — no CPU spin
        }
    }
}

A paused source calls wait_while_paused() and the OS puts its thread to sleep; resume() wakes it with notify_all. The AtomicBool fast path means the common “not paused” case costs nothing, no lock at all. Three interior-mutable primitives, each chosen for its access pattern: atomics for hot counters, ArcSwap for a mostly-read registry, Condvar for genuine blocking.

This pairs with the bounded channel from the worker pool: a full channel throttles a producer implicitly (its send blocks); the PauseSignal is the explicit backpressure the arbitrator reaches for when the budget, not the channel, is the constraint. Same goal, different lever.

Worked → completion → faded

You’ve read the arbitrator’s three primitives. Now build the core idea yourself, one shared, interior-mutable counter, in three rungs: a worked program, a one-gap completion, then the whole thing from a skeleton. One new idea per rung; the type never changes, only how much you write.

Worked: share a counter across threads

This is the whole idea in std: an Arc<AtomicU64> mutated by four threads through &self, no &mut and no lock anywhere. Run it and watch the total land on 10000 every time, regardless of how the threads interleave.

rust // editable

reset ▶ run

use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::thread;

fn main() {
  // One shared counter — like the arbitrator's charged-bytes total.
  let charged = Arc::new(AtomicU64::new(0));

  let mut handles = Vec::new();
  for op in 1..=4 {
      let charged = Arc::clone(&charged);   // every thread shares the SAME counter
      handles.push(thread::spawn(move || {
          // &self access only — yet this mutates, via the atomic. No &mut needed.
          charged.fetch_add(1000 * op, Ordering::Relaxed);
      }));
  }
  for h in handles { h.join().unwrap(); }

  // 1000 + 2000 + 3000 + 4000 — correct regardless of thread interleaving
  println!("total charged: {}", charged.load(Ordering::Relaxed));
}

use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; use std::thread; fn main() { // One shared counter — like the arbitrator's charged-bytes total. let charged = Arc::new(AtomicU64::new(0)); let mut handles = Vec::new(); for op in 1..=4 { let charged = Arc::clone(&charged); // every thread shares the SAME counter handles.push(thread::spawn(move || { // &self access only — yet this mutates, via the atomic. No &mut needed. charged.fetch_add(1000 * op, Ordering::Relaxed); })); } for h in handles { h.join().unwrap(); } // 1000 + 2000 + 3000 + 4000 — correct regardless of thread interleaving println!("total charged: {}", charged.load(Ordering::Relaxed)); }

Four threads, one counter, each charging through a shared &. The total is always 10000: atomic fetch_add makes the increments race-free without any lock. That is interior mutability in one line: a & that writes.

Completion: release bytes again

The arbitrator doesn’t only charge; operators release memory when a batch is freed. Here the charge step is written for you; complete the matching release so the net total is correct. Each thread charges 1000 then releases 400, so four threads should leave 4 * 600 = 2400.

rust // editable

reset ▶ run

use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::thread;

fn main() {
  let charged = Arc::new(AtomicU64::new(0));

  let mut handles = Vec::new();
  for _ in 0..4 {
      let charged = Arc::clone(&charged);
      handles.push(thread::spawn(move || {
          charged.fetch_add(1000, Ordering::Relaxed);     // charge a batch
          // TODO: release 400 of it again, through &self, lock-free.
          // The released bytes are no longer in use, so SUBTRACT them.
      }));
  }
  for h in handles { h.join().unwrap(); }

  // expect 4 * (1000 - 400) = 2400
  println!("net charged: {}", charged.load(Ordering::Relaxed));
}

use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; use std::thread; fn main() { let charged = Arc::new(AtomicU64::new(0)); let mut handles = Vec::new(); for _ in 0..4 { let charged = Arc::clone(&charged); handles.push(thread::spawn(move || { charged.fetch_add(1000, Ordering::Relaxed); // charge a batch // TODO: release 400 of it again, through &self, lock-free. // The released bytes are no longer in use, so SUBTRACT them. })); } for h in handles { h.join().unwrap(); } // expect 4 * (1000 - 400) = 2400 println!("net charged: {}", charged.load(Ordering::Relaxed)); }

💡 Hint 1

AtomicU64 has a fetch_sub that mirrors fetch_add: same &self, same lock-free atomic, opposite direction. It takes the amount to subtract and the same Ordering::Relaxed.

Show solution

charged.fetch_sub(400, Ordering::Relaxed);          // release, through &self

fetch_sub is the release half of the same lock-free pattern add_bytes/sub_bytes use in the real ConsumerHandle. No &mut, no Mutex: charge and release are both single atomic instructions, which is exactly why they can sit on the per-batch hot path without serializing the operators against each other.

Faded: a tiny budget gate

Now write the whole thing from a skeleton. Build a one-method type, Budget, that holds an AtomicU64 of used bytes and exposes try_charge(&self, n) -> bool: it adds n only if the total would stay within a fixed LIMIT, returning whether the charge was admitted. Notice the signature is &self: this is a shared object, like the arbitrator. Two threads will race on it.

rust // editable

reset ▶ run

use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::thread;

const LIMIT: u64 = 1000;

struct Budget {
  used: AtomicU64,
}

impl Budget {
  fn new() -> Self {
      Budget { used: AtomicU64::new(0) }
  }

  // TODO: charge `n` bytes through &self ONLY if used + n <= LIMIT.
  // Return true if admitted, false if it would overflow the budget.
  // Hint: read the current value, then commit with a single atomic op.
  fn try_charge(&self, n: u64) -> bool {
      todo!()
  }
}

fn main() {
  let budget = Arc::new(Budget::new());

  let mut handles = Vec::new();
  for _ in 0..2 {
      let budget = Arc::clone(&budget);
      handles.push(thread::spawn(move || {
          // each thread tries to charge 800; only one can fit under LIMIT=1000
          println!("admitted: {}", budget.try_charge(800));
      }));
  }
  for h in handles { h.join().unwrap(); }

  println!("final used: {}", budget.used.load(Ordering::Relaxed));
}

use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; use std::thread; const LIMIT: u64 = 1000; struct Budget { used: AtomicU64, } impl Budget { fn new() -> Self { Budget { used: AtomicU64::new(0) } } // TODO: charge `n` bytes through &self ONLY if used + n <= LIMIT. // Return true if admitted, false if it would overflow the budget. // Hint: read the current value, then commit with a single atomic op. fn try_charge(&self, n: u64) -> bool { todo!() } } fn main() { let budget = Arc::new(Budget::new()); let mut handles = Vec::new(); for _ in 0..2 { let budget = Arc::clone(&budget); handles.push(thread::spawn(move || { // each thread tries to charge 800; only one can fit under LIMIT=1000 println!("admitted: {}", budget.try_charge(800)); })); } for h in handles { h.join().unwrap(); } println!("final used: {}", budget.used.load(Ordering::Relaxed)); }

💡 Hint 1

fetch_update is the read-modify-write that decides atomically. Give it a closure returning Some(new) to commit or None to reject; it returns Ok if your closure committed, Err if it declined. Map that Result to a bool.

Show solution

fn try_charge(&self, n: u64) -> bool {
    self.used
        .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |cur| {
            let next = cur + n;
            if next <= LIMIT { Some(next) } else { None } // commit, or decline
        })
        .is_ok()                                          // Ok = admitted, Err = rejected
}

fetch_update loops internally until its compare-and-swap wins, so the check-then-set is one atomic decision: two threads can’t both see used == 0, both decide 800 fits, and both commit 1600. Exactly one of the racing try_charge(800) calls returns true; the final used is 800, never 1600. This is the arbitrator’s admission logic in miniature: a &self gate on a shared budget, correct under any interleaving, with no lock on the hot path.

Real-source grounding

The skeleton you just wrote is the shape of the real thing. In clinker-exec, the same fetch_update read-modify-write drives add_bytes/sub_bytes on the ConsumerHandle, and fetch_max drives observe on the arbitrator, every one a &self method on an Arc-shared type.

What the compiler enforces: because the shared fields are AtomicU64 (which is Sync), MemoryArbitrator is Sync, so Arc<MemoryArbitrator> can be moved into every operator thread. Swap one counter for a plain u64 and the &self methods stop compiling. Incrementing a plain u64 needs &mut self, which Arc never hands out, so the shared-and-mutate pattern becomes a build error, not a latent data race.

What a junior might misread: seeing &self on add_bytes and concluding “this method only reads.” &self means shared, not read-only; the atomic field is what makes the write sound. The other misread is reaching for a Mutex to guard a single counter: correct, but it serializes the hot path the atomic keeps lock-free.

Why-bridge: one primitive per access pattern

Why three different interior-mutable types instead of wrapping the whole arbitrator in one big Mutex? Because each field has a different access pattern, and the cheapest correct primitive differs for each:

• Counters (limit, peak_rss, charged bytes) are written per-batch from many threads on the hot path → AtomicU64, a single lock-free instruction.
• The consumer registry is read constantly (who is using what?) and written rarely (an operator registers once) → ArcSwap, lock-free reads with copy-on-write writes.
• The pause flag needs a thread to genuinely block until resumed → AtomicBool for the cheap not-paused check, plus Mutex + Condvar for the rare real wait.

One global Mutex would be correct but would funnel every per-batch charge through a single lock, erasing the parallelism the one-thread-per-source design exists to get.

Want Arc and shared-across-threads ownership from first principles, with no engine framing? Read the docs →

Retrieval checkpoint

✓ Checkpoint — interior mutability

💡 Hint 1

Look at how many methods on the arbitrator and ConsumerHandle take &self yet clearly change state. Then ask: what type makes each of those writes safe without a lock?

What the tour establishes

MemoryArbitrator is Arc-shared, so all its mutators take &self. Counters are AtomicU64 (lock-free fetch_add/fetch_max on the hot path); the consumer registry is ArcSwap (lock-free reads, copy-on-write writes); the PauseSignal is AtomicBool + Mutex + Condvar (cheap not-paused fast path, real blocking when paused). Each interior-mutable primitive is matched to its access pattern, and that’s the design lesson.

// quick check

Why are the MemoryArbitrator's counters AtomicU64 with &self methods, rather than plain u64 fields mutated through &mut self?

• Atomics use less memory than u64
• The arbitrator is Arc-shared across many threads, which only grants &self; interior-mutable atomics let those threads update the counters lock-free, where &mut self would be impossible to obtain and a Mutex would serialize the hot path
• &mut self is not allowed on any struct in Rust

// quick check

You write a counter type for a single-threaded cache and want to mutate it through &self. Which interior-mutable primitive is the lightest correct choice, and why NOT the one the arbitrator uses?

• AtomicU64, because it is always the fastest interior-mutable type
• Cell<u64> (or RefCell), because the value never crosses threads; an atomic pays for cross-thread synchronization you don't need single-threaded
• Mutex<u64>, because every interior mutation requires a lock

// quick check

Why does PauseSignal check an AtomicBool before taking the Condvar's lock in wait_while_paused?

• To make pausing slower and safer
• So the common 'not paused' case is a single lock-free atomic load with no mutex acquisition: only an actually-paused thread pays for the lock and parks
• Because Condvar requires an AtomicBool

// quick check

Spaced recall from the worker pool lesson: the ingest channel between a source thread and the dispatcher is a BOUNDED crossbeam channel. How does that bound relate to the arbitrator's PauseSignal?

• They are the same mechanism: the PauseSignal IS the channel
• Both are backpressure: a full bounded channel throttles a producer implicitly (its send blocks), while the PauseSignal is explicit backpressure the arbitrator triggers when the memory budget, not the channel, is the constraint
• The bounded channel makes the PauseSignal unnecessary, so the arbitrator never pauses anyone

// quick check

An engineer says: 'add_bytes takes &self, so it can't actually be changing the arbitrator; &self is read-only.' Which correction names the misconception?

• Correct: &self methods never mutate; add_bytes must return a new arbitrator
• Misconception: &self means SHARED, not immutable. Interior-mutable fields (here an atomic) are exactly the types that write through a shared reference, which is why add_bytes can mutate while taking &self
• It's fine because add_bytes secretly takes &mut self at run time

You should be able to:

• You can explain why an Arc-shared type must mutate through &self and name interior mutability as the mechanism
• You can match each primitive (Cell/RefCell, Atomic*, Mutex, ArcSwap, Condvar) to the access pattern it fits
• You can explain why the byte counters are atomics and the consumer registry is ArcSwap
• You wrote a &self try_charge gate on a shared AtomicU64 that is correct under any thread interleaving
• You can explain what the PauseSignal's Condvar does and why the AtomicBool fast path matters

Verify in the checkout:

grep -n 'struct MemoryArbitrator' crates/clinker-exec/src/pipeline/memory.rs
grep -n 'fn add_bytes' crates/clinker-exec/src/pipeline/memory.rs
grep -n 'struct PauseSignal' crates/clinker-exec/src/pipeline/memory.rs
grep -n 'fetch_add\|fetch_max\|fetch_update' crates/clinker-exec/src/pipeline/memory.rs | head

Connections

You’ve seen how the arbitrator stores shared state safely. Next: what it decides with it, spill, pause, or abort, and how it refuses an impossible budget before a single record moves.

Mark this lesson complete