The worker pool — threads & channels

The dispatch loop doesn’t sit idle waiting for a source to hand it the next record. Clinker runs each source on its own OS thread, reading and decoding in parallel, pushing records through a bounded channel into the dispatcher. That overlaps slow IO with compute, and because the channel is bounded, it throttles a fast reader when the consumer falls behind, keeping memory in check. The thing that makes this concurrency safe rather than terrifying is Rust’s ownership model: the Send and Sync marker traits are checked by the compiler, so a data race is a build error. We use just enough of threads and channels to read the executor’s ingest path; The Rust Book, ch. 16 is the canonical treatment of fearless concurrency if you want it from first principles.

Prerequisites

• Dispatching a node

→

Leads to

• Node buffers & spill

What you’ll be able to do

• Describe the one-thread-per-source ingest model and explain why running producers and consumer concurrently overlaps slow IO with compute.
• Explain how a bounded channel produces backpressure, and name what StreamEvent carries through it.
• Distinguish the Send bound from the Sync bound, and say which one RecordSource requires and why it deliberately does not require the other.
• Write a producer/consumer program with a bounded channel that visibly stalls the producer when the consumer falls behind.

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

• thread: an OS-scheduled line of execution; the engine gives each source its own.
• channel: a typed queue that hands values from a producer thread to a consumer thread.
• sender / receiver: the two endpoints of a channel (tx puts in, rx takes out).
• Send: ownership of the value may move to another thread.
• Sync: a shared reference to the value may be used from several threads at once.
• backpressure: a full buffer pacing a fast producer to a slow consumer.

Predict first

✋ Predict first

One producer thread sends rows 0..6 into a bounded channel of capacity 2. The consumer is slow: each row it receives takes real CPU work. Both threads print as they go. Before the consumer takes its FIRST row, how many rows can the producer have already sent: all six, or some smaller number? Write your number before revealing.

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

Reveal the answer

At most three, not all six. The channel holds two unconsumed rows; the producer also holds one row in hand mid-send. So the producer can get rows 0 and 1 buffered, then block on sending row 2 because the buffer is full. It cannot run ahead to row 5 while the consumer is still on row 0.

The wrong mental model is “send just drops the row into an infinite mailbox and returns.” It does not. On a bounded channel, send is a rendezvous point: once the buffer is full, send parks the producer thread until the consumer frees a slot. That blocking is the backpressure, the same mechanism that stops a fast file reader from piling a million records into the engine’s memory while a slow pipeline chews through them. Hold that “at most three” prediction; the first runnable example below lets you watch it happen.

One thread per source

Every declared source gets a dedicated, named thread that drives its reader and pushes records:

clinker-exec ·ingest.rs ·ingest_source fn @19acdcb4

// crates/clinker-exec/src/executor/mod.rs — one OS thread per Source
let handle = std::thread::Builder::new()
    .name(format!("clinker-ingest-{}", src_cfg.name))
    .spawn(move || ingest_source(src_cfg_owned, source_input, config_clone, stream, shutdown))?;

The producer thread loops next_record() and pushes each one into the channel; the dispatch loop (the consumer) drains it. Producers and consumer run concurrently: while source A’s thread is blocked decoding a slow file, the dispatcher is busy running operators on records already delivered. That overlap is the whole point: slow IO on one thread happens during compute on another, instead of one waiting on the other.

The bounded channel is the backpressure

The channel connecting a reader thread to the dispatcher is a bounded crossbeam_channel. Its element is a StreamEvent, either a body record or a document-boundary marker:

clinker-exec ·stream_event.rs ·StreamEvent type @19acdcb4

pub enum StreamEvent {
    Record(Record, u64),       // a row and its source row number
    Punctuation(Punctuation),  // a document-boundary marker (open / close)
}

crates/clinker-exec/src/executor/source_stream.rs

let (tx, rx) = crossbeam_channel::bounded(capacity);  // DEFAULT_CAPACITY = 1024
// ...
self.tx.send(StreamEvent::record(record, row_num))?;  // BLOCKS when the channel is full

Here tx is the sender the reader thread owns, and rx is the receiver the dispatcher drains.

Read what “bounded” buys you. The channel holds at most capacity events. When it’s full, the producer’s send blocks the reader thread until the consumer drains one. So a fast reader cannot race ahead of a slow pipeline and pile a million records into memory; it’s paced to the consumer. That is backpressure, and it falls out of the channel’s bound for free, with no explicit throttling logic:

 source A ──[reader thread]──▶ bounded channel (cap 1024) ──┐
 source B ──[reader thread]──▶ bounded channel (cap 1024) ──┼──▶ dispatch loop ──▶ operators
 source C ──[reader thread]──▶ bounded channel (cap 1024) ──┘
                    ▲                                            a full channel blocks
                    └──────────── backpressure ─────────────────  its reader thread

Send and Sync: the compiler proves it safe

Moving a reader onto another thread is only sound if the reader can cross threads. Rust spells that requirement out in the trait bound: a RecordSource must be Send:

clinker-exec ·mod.rs ·RecordSource trait @19acdcb4

/// Must be `Send` for the per-Source `std::thread` to own it; not `Sync`
/// — each source is single-threaded streaming.
pub trait RecordSource: Send { /* schema(), next_record(), ... */ }

Two auto-traits carry the whole thread-safety story:

• Send = “ownership of this value may move to another thread.” The reader is Send, so the spawn can take it by move. A non-Send type (say, one holding an Rc) would make that spawn a compile error, caught before it ever runs.
• Sync = “a shared reference may be used from several threads at once.” The reader is deliberately not Sync: one thread owns it, no sharing. But data that is shared (the schema, the document context) is held behind Arc and is Sync, so many threads can read it at once. (That’s the Data & Representation Arc story, now load-bearing for concurrency.)

You don’t sprinkle locks and hope. The bounds are checked at compile time: if the types line up, the threading is race-free by construction. Heavy CPU operators (sorts, joins) take a second path, a shared Rayon thread pool the executor installs work onto, but the same Send/ Sync discipline governs what may cross into it.

Worked → completion → faded

Now you build the producer/consumer pattern yourself, scaffolded down to a from-scratch task. Each rung adds exactly one new idea: first watch backpressure, then create the channel that causes it, then spawn the producer thread that a Send bound makes safe.

Worked: watch the producer stall

std has a bounded channel too, sync_channel. Here one producer thread feeds a deliberately slow consumer through a tiny buffer; watch the producer stall when the buffer fills. This is the “at most three” prediction from the top, made runnable. Read the annotations, then run it.

rust // editable

reset ▶ run

use std::sync::mpsc::sync_channel;
use std::thread;

fn main() {
  // Bounded buffer of 2 — like the executor's per-source channel, shrunk.
  let (tx, rx) = sync_channel::<u32>(2);

  // Producer on its OWN thread (one thread per source, in the engine).
  let producer = thread::spawn(move || {
      for row in 0..6 {
          // send() BLOCKS once 2 unconsumed rows are buffered — backpressure.
          tx.send(row).unwrap();
          println!("produced {row}");
      }
      println!("producer done");
  });

  // Slow consumer = the dispatch loop doing per-record work.
  for got in rx {
      let mut acc = 0u64;
      for i in 0..3_000_000 { acc = acc.wrapping_add(i); } // simulate work
      std::hint::black_box(acc);
      println!("    consumed {got}");
  }
  producer.join().unwrap();
}

use std::sync::mpsc::sync_channel; use std::thread; fn main() { // Bounded buffer of 2 — like the executor's per-source channel, shrunk. let (tx, rx) = sync_channel::<u32>(2); // Producer on its OWN thread (one thread per source, in the engine). let producer = thread::spawn(move || { for row in 0..6 { // send() BLOCKS once 2 unconsumed rows are buffered — backpressure. tx.send(row).unwrap(); println!("produced {row}"); } println!("producer done"); }); // Slow consumer = the dispatch loop doing per-record work. for got in rx { let mut acc = 0u64; for i in 0..3_000_000 { acc = acc.wrapping_add(i); } // simulate work std::hint::black_box(acc); println!(" consumed {got}"); } producer.join().unwrap(); }

The producer can get at most two rows ahead before send blocks; it then advances only as the consumer drains. Shrink the slow loop to nothing and the producer races to “done” immediately: the bound is the only thing pacing it. Grow the channel and you trade memory for slack. That dial is exactly what the executor’s DEFAULT_CAPACITY sets.

Completion: size the channel

Below, the producer and consumer are written for you; the channel’s capacity is the gap. The producer sends rows 0..5 and prints each send; the consumer is slow. You want the producer to be able to get at most one row ahead of the consumer, so it should print produced 0, then stall on sending row 1 until the first row is consumed. Fill in the capacity that does that.

rust // editable

reset ▶ run

use std::sync::mpsc::sync_channel;
use std::thread;

fn main() {
  // TODO: pick the capacity that lets the producer get exactly ONE row ahead.
  let (tx, rx) = sync_channel::<u32>(/* capacity? */ 0);

  let producer = thread::spawn(move || {
      for row in 0..5 {
          tx.send(row).unwrap();
          println!("produced {row}");
      }
  });

  for got in rx {
      let mut acc = 0u64;
      for i in 0..3_000_000 { acc = acc.wrapping_add(i); }
      std::hint::black_box(acc);
      println!("    consumed {got}");
  }
  producer.join().unwrap();
}

use std::sync::mpsc::sync_channel; use std::thread; fn main() { // TODO: pick the capacity that lets the producer get exactly ONE row ahead. let (tx, rx) = sync_channel::<u32>(/* capacity? */ 0); let producer = thread::spawn(move || { for row in 0..5 { tx.send(row).unwrap(); println!("produced {row}"); } }); for got in rx { let mut acc = 0u64; for i in 0..3_000_000 { acc = acc.wrapping_add(i); } std::hint::black_box(acc); println!(" consumed {got}"); } producer.join().unwrap(); }

💡 Hint 1

“One row ahead” means one row sitting in the buffer while the producer holds the next one mid-send. A capacity of 0 is a special case: it’s a pure rendezvous where send waits for a matching receive, so the producer never gets any rows ahead. You want one slot of slack.

Show solution

let (tx, rx) = sync_channel::<u32>(1);

A capacity of 1 lets exactly one unconsumed row sit in the buffer. The producer sends row 0 (buffered), sends row 1 (now blocks, buffer full), and only advances when the consumer takes row 0. Capacity is the backpressure dial: 0 is strict rendezvous, larger values trade memory for more producer slack. The engine’s DEFAULT_CAPACITY = 1024 is the same dial turned up.

Faded: spawn the producer thread

Your turn with no producer written. Complete main so that a producer thread sends the three records [10, 20, 30] through the bounded channel, and the main thread (the consumer) sums what it receives and prints the total (60). The reason this compiles at all is the bound you just learned: i32 is Send, so moving the sender and the data onto the spawned thread is allowed.

rust // editable

reset ▶ run

use std::sync::mpsc::sync_channel;
use std::thread;

fn main() {
  let (tx, rx) = sync_channel::<i32>(1);

  // TODO: spawn a producer thread that sends 10, 20, 30 through `tx`,
  // then drops `tx` so the consumer's loop can end.
  // (Hint: `move` the sender onto the thread.)

  let mut total = 0;
  for got in rx {
      total += got;
  }
  println!("total = {total}"); // expect: total = 60
}

use std::sync::mpsc::sync_channel; use std::thread; fn main() { let (tx, rx) = sync_channel::<i32>(1); // TODO: spawn a producer thread that sends 10, 20, 30 through `tx`, // then drops `tx` so the consumer's loop can end. // (Hint: `move` the sender onto the thread.) let mut total = 0; for got in rx { total += got; } println!("total = {total}"); // expect: total = 60 }

💡 Hint 1

The receiver’s for got in rx loop ends only when every sender is dropped. The spawned closure owns tx after move, so tx is dropped when the closure returns, and that’s what lets the loop finish. You do not need to join the handle for the total to be correct here, but capturing it is tidier.

Show solution

let producer = thread::spawn(move || {
    for row in [10, 20, 30] {
        tx.send(row).unwrap();
    }
    // tx drops here when the closure returns — the consumer loop can now end.
});

let mut total = 0;
for got in rx {
    total += got;
}
producer.join().unwrap();
println!("total = {total}"); // total = 60

thread::spawn(move || …) moves tx (a Send value) onto the new thread, which then owns it. Because the engine’s RecordSource: Send, this is exactly the move the executor makes with a real reader, only here the “record” is an i32. Swap the i32 for a non-Send type like Rc<i32> and this stops compiling: the compiler refuses the move before the program ever runs.

Comfortable with threads, bounded channels, and Send/Sync? Skip ahead →

Why-bridge: where this rides on Arc

The split is deliberate. The reader is Send-not-Sync because exactly one thread owns and drives it; there is no &reader shared anywhere. But the things many reader threads (and the dispatcher, and Rayon workers) all need to see at once (the schema, the document context) are held behind Arc, which is Sync. So the rule of thumb the engine follows is: move what one thread owns (Send); share what many threads read (Arc, which is Sync).

That is why Arc, not Rc, is the smart pointer in the hot path. Rc’s reference count is a plain integer with no synchronization, so Rc is neither Send nor Sync, and handing one to thread::spawn is a compile error. Arc’s count is atomic, which is precisely what earns it Send + Sync and lets one Schema be shared across every ingest thread safely.

Want Arc and atomic reference counts from first principles, with no engine framing? Read the docs →

Retrieval checkpoint

✓ Checkpoint — threads & channels

💡 Hint 1

Read the : Send bound on RecordSource and its comment. Then find the bounded(...) call: the capacity argument is the backpressure dial.

What the tour establishes

Each source runs ingest_source on its own named thread, pushing StreamEvents through a crossbeam_channel::bounded channel; a full channel blocks the reader (backpressure). The reader is RecordSource: Send so it can be moved onto its thread, but not Sync (one owner, no sharing); shared data rides Arc and is Sync. The compiler checks all of it, so the threading is race-free by construction, not by careful convention.

// quick check

How does a bounded channel create backpressure between a source reader and the dispatch loop?

• It drops records when full to keep up
• When the channel is full, the producer's send blocks the reader thread until the consumer drains an event, so a fast reader is paced to the consumer and can't pile unbounded records into memory
• It speeds up the consumer automatically

// quick check

Why is RecordSource required to be Send but deliberately not Sync?

• Send makes it faster; Sync makes it slower
• It must move onto its own ingest thread (Send), but it is owned and driven by exactly one thread with no shared &-access (so Sync is unneeded); shared data instead rides Arc, which is Sync
• Sync types cannot be put in a channel

// quick check

Spaced recall from the sharing-and-boxing lesson: the engine shares one Schema across millions of records and across every ingest thread using Arc, never Rc. Why must it be Arc here?

• Arc copies the whole Schema on each clone, so each thread gets its own copy and never races
• Arc's reference count is atomic, which is what makes it Send + Sync so it can cross threads safely; Rc's count is a plain integer, so Rc is neither Send nor Sync and a spawn holding one is a compile error
• Rc and Arc are interchangeable across threads; Arc is just the more common name

// quick check

Which statement about send on a bounded channel is the MISCONCEPTION to correct?

• When the channel is full, send blocks the calling thread until a slot frees up
• send always returns immediately and the channel grows to hold whatever you put in; the bound only affects performance, not whether send blocks
• Backpressure is a consequence of that blocking, not a separate throttling mechanism the engine codes by hand

You should be able to:

• You can describe the one-thread-per-source ingest model and why it overlaps IO with compute
• You can explain how a bounded channel produces backpressure and what StreamEvent carries
• You can distinguish Send from Sync, and say which RecordSource requires and why it is deliberately not the other
• You wrote a producer/consumer program with a bounded channel that stalls the producer when the consumer falls behind
• You can say why Arc (not Rc) is the pointer that crosses ingest threads

Verify in the checkout:

grep -n 'pub enum StreamEvent' crates/clinker-exec/src/executor/stream_event.rs
grep -n 'fn ingest_source' crates/clinker-exec/src/executor/ingest.rs
grep -n 'crossbeam_channel::bounded' crates/clinker-exec/src/executor/source_stream.rs
grep -n 'trait RecordSource' crates/clinker-exec/src/source/mod.rs

Connections

You’ve seen records flow concurrently from sources into the dispatcher, watched a bounded channel stall a fast producer, and built the producer/consumer pattern yourself, resting on the Send bound that makes the move safe and the Arc/Sync split that lets shared data be read from many threads. But a blocking operator, a sort or a big aggregation, has to hold records while it works, and there may be more than fit in memory. Next: the buffer that can live in RAM, on disk, or both.

Prerequisites

• Dispatching a node

→

Leads to

• Node buffers & spill

Go deeper on the Rust (optional, one-directional, the same Arc/shared-ownership concept taught from first principles in The Rust Book):

• Want Arc and atomic reference counts on a standalone example? Read the docs →

Glossary terms used: thread, channel, sender / receiver, Send, Sync, backpressure.

Mark this lesson complete