Threads & channels

The dispatch loop from last lesson 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: Send and Sync are checked by the compiler, so a data race is a build error.

Prerequisites

• Enum vs trait dispatch

→

Leads to

• Node buffers & spill

You’ll be able to: describe the one-thread-per-source ingest model, explain how a bounded channel produces backpressure, and say what the Send bound on RecordSource guarantees.

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 @47d2e12

// 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.

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 @47d2e12

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

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 @47d2e12

/// 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 &T may be shared across threads.” 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 Phase 2 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.

Feel the backpressure

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:

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.

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

// 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

Inspect the concurrency

✓ 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 — the threading is race-free by construction, not by careful convention.

// 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

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 explain what the Send bound on RecordSource guarantees and why it is not Sync

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

You’ve seen records flow concurrently from sources into the dispatcher. But a blocking operator — a sort, 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.

Mark this lesson complete