From 4bee9f1e2da65a307f9ba9a70256a2382bc2d03c Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Sat, 15 Aug 2026 07:58:10 -0400 Subject: [PATCH 1/5] Reuse returned containers across scheduling boundaries --- communication/src/allocator/thread.rs | 77 +++++++++++++++++-- container/src/lib.rs | 33 +++++++- .../src/dataflow/operators/generic/handles.rs | 23 +++++- 3 files changed, 119 insertions(+), 14 deletions(-) diff --git a/communication/src/allocator/thread.rs b/communication/src/allocator/thread.rs index 9857ed5ba..72f933e83 100644 --- a/communication/src/allocator/thread.rs +++ b/communication/src/allocator/thread.rs @@ -94,13 +94,78 @@ impl Pull for Puller { #[inline] fn pull(&mut self) -> &mut Option { let mut borrow = self.source.borrow_mut(); - // if let Some(element) = self.current.take() { - // // TODO : Arbitrary constant. - // if borrow.1.len() < 16 { - // borrow.1.push_back(element); - // } - // } + if let Some(element) = self.current.take() { + // Retain a bounded number of values for the producer to reclaim. + // Consumers that took the value with `recv()` leave `current` as + // `None` and therefore correctly return nothing. + if borrow.1.len() < 16 { + borrow.1.push_back(element); + } + } self.current = borrow.0.pop_front(); &mut self.current } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn pull_returns_unconsumed_resources_to_the_producer() { + let events = Rc::new(RefCell::new(Vec::new())); + let (mut pusher, mut puller) = Thread::new_from(0, events); + + let mut first = Vec::with_capacity(1024); + first.extend(0..16); + let first_capacity = first.capacity(); + let mut first = Some(first); + pusher.push(&mut first); + assert!(first.is_none()); + + puller.pull().as_mut().unwrap().clear(); + assert!(puller.pull().is_none()); + + let mut second = Some(vec![99]); + pusher.push(&mut second); + let returned = second.expect("producer should reclaim the prior value"); + assert!(returned.is_empty()); + assert_eq!(returned.capacity(), first_capacity); + } + + #[test] + fn recv_transfers_ownership_without_returning_a_resource() { + let events = Rc::new(RefCell::new(Vec::new())); + let (mut pusher, mut puller) = Thread::new_from(0, events); + + pusher.send(vec![1, 2, 3]); + assert_eq!(puller.recv(), Some(vec![1, 2, 3])); + assert!(puller.pull().is_none()); + + let mut next = Some(vec![4]); + pusher.push(&mut next); + assert!(next.is_none(), "a taken value must not appear in the return queue"); + } + + #[test] + fn resource_return_queue_is_bounded() { + let events = Rc::new(RefCell::new(Vec::new())); + let (mut pusher, mut puller) = Thread::new_from(0, events); + + for value in 0..32 { + pusher.send(vec![value]); + } + for _ in 0..32 { + puller.pull().as_mut().unwrap().clear(); + } + assert!(puller.pull().is_none()); + + let mut returned = 0; + for _ in 0..32 { + let mut item = Some(Vec::new()); + pusher.push(&mut item); + returned += usize::from(item.is_some()); + } + assert_eq!(returned, 16); + } +} diff --git a/container/src/lib.rs b/container/src/lib.rs index 400512475..41acd0f48 100644 --- a/container/src/lib.rs +++ b/container/src/lib.rs @@ -136,8 +136,9 @@ mod noop { /// A default container builder that uses length and preferred capacity to chunk data. /// -/// Maintains a single empty allocation between [`Self::push_into`] and [`Self::extract`], but not -/// across [`Self::finish`] to maintain a low memory footprint. +/// Maintains a single empty allocation for reuse across calls, including across +/// [`Self::finish`]. This bounds retained memory to one container while avoiding +/// an allocation boundary at each logical timestamp. /// /// Maintains FIFO order. #[derive(Default, Debug)] @@ -184,13 +185,37 @@ impl ContainerBuilder for CapacityContainerBuilder if !self.current.is_empty() { self.pending.push_back(std::mem::take(&mut self.current)); } - self.empty = self.pending.pop_front(); - self.empty.as_mut() + if let Some(container) = self.pending.pop_front() { + self.empty = Some(container); + self.empty.as_mut() + } else { + None + } } } impl LengthPreservingContainerBuilder for CapacityContainerBuilder { } +#[cfg(test)] +mod tests { + use super::{CapacityContainerBuilder, ContainerBuilder, PushInto}; + + #[test] + fn capacity_builder_retains_returned_container_across_finish() { + let mut builder = CapacityContainerBuilder::>::default(); + builder.push_into(1); + + let container = builder.finish().expect("one finished container"); + let allocation = container.as_ptr(); + container.clear(); + assert!(builder.finish().is_none()); + + builder.push_into(2); + let reused = builder.finish().expect("one finished container"); + assert_eq!(reused.as_ptr(), allocation); + } +} + impl Accountable for Vec { #[inline] fn record_count(&self) -> i64 { i64::try_from(Vec::len(self)).unwrap() } #[inline] fn is_empty(&self) -> bool { Vec::is_empty(self) } diff --git a/timely/src/dataflow/operators/generic/handles.rs b/timely/src/dataflow/operators/generic/handles.rs index 9205e627e..090318d89 100644 --- a/timely/src/dataflow/operators/generic/handles.rs +++ b/timely/src/dataflow/operators/generic/handles.rs @@ -31,6 +31,8 @@ pub struct InputHandleCore>> { /// Staged capabilities and containers. staging: VecDeque<(InputCapability, C)>, staged: Vec, + /// Consumed containers available to swap back into subsequently pulled messages. + spares: Vec, } impl>> InputHandleCore { @@ -54,9 +56,16 @@ impl>> InputHandleCore(&mut self, mut logic: F) where F: FnMut(InputCapability, std::slice::IterMut::), C: Default { - while let Some((cap, data)) = self.next() { - let data = std::mem::take(data); - self.staging.push_back((cap, data)); + let pull_counter = &mut self.pull_counter; + let internal = &self.internal; + let summaries = &self.summaries; + let spares = &mut self.spares; + while let Some((guard, bundle)) = pull_counter.next_guarded() { + let cap = InputCapability::new(Rc::clone(internal), Rc::clone(summaries), guard); + let data = &mut bundle.data; + let mut received = spares.pop().unwrap_or_default(); + std::mem::swap(data, &mut received); + self.staging.push_back((cap, received)); } self.staging.make_contiguous().sort_unstable_by(|x,y| x.0.time().cmp(&y.0.time())); @@ -65,7 +74,12 @@ impl>> InputHandleCore>>( summaries, staging: Default::default(), staged: Default::default(), + spares: Default::default(), } } From be7c8536794b181f3db871d477443bdf03c5589b Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Sat, 15 Aug 2026 07:58:10 -0400 Subject: [PATCH 2/5] Promote allocation-conscious columnar transport --- timely/examples/columnar.rs | 219 ++++------------ timely/examples/transport_alloc.rs | 403 +++++++++++++++++++++++++++++ timely/src/container/columnar.rs | 323 +++++++++++++++++++++++ timely/src/lib.rs | 3 + 4 files changed, 785 insertions(+), 163 deletions(-) create mode 100644 timely/examples/transport_alloc.rs create mode 100644 timely/src/container/columnar.rs diff --git a/timely/examples/columnar.rs b/timely/examples/columnar.rs index 1a87ab164..a3315ef08 100644 --- a/timely/examples/columnar.rs +++ b/timely/examples/columnar.rs @@ -3,12 +3,13 @@ use std::collections::HashMap; use columnar::Index; -use timely::Accountable; +use timely::container::columnar::{ColumnarBuilder, ColumnarContainer}; use timely::container::CapacityContainerBuilder; use timely::dataflow::channels::pact::{ExchangeCore, Pipeline}; -use timely::dataflow::InputHandle; use timely::dataflow::operators::{InspectCore, Operator, Probe}; +use timely::dataflow::InputHandle; use timely::dataflow::ProbeHandle; +use timely::Accountable; // Creates `WordCountContainer` and `WordCountReference` structs, // as well as various implementations relating them to `WordCount`. @@ -19,9 +20,8 @@ struct WordCount { } fn main() { - type InnerContainer = ::Container; - type Container = Column; + type Container = ColumnarContainer; use columnar::Len; @@ -39,27 +39,34 @@ fn main() { worker.dataflow::(|scope| { input .to_stream(scope) - .unary( - Pipeline, - "Split", - |_cap, _info| { - move |input, output| { - input.for_each_time(|time, data| { - let mut session = output.session(&time); - for data in data { - for wordcount in data.borrow().into_index_iter().flat_map(|wordcount| { - wordcount.text.split(|b| b.is_ascii_whitespace()).filter(|s| !s.is_empty()).map(move |text| WordCountReference { text, diff: wordcount.diff }) - }) { - session.give(wordcount); - } + .unary(Pipeline, "Split", |_cap, _info| { + move |input, output| { + input.for_each_time(|time, data| { + let mut session = output.session(&time); + for data in data { + for wordcount in + data.borrow().into_index_iter().flat_map(|wordcount| { + wordcount + .text + .split(|b| b.is_ascii_whitespace()) + .filter(|s| !s.is_empty()) + .map(move |text| WordCountReference { + text, + diff: wordcount.diff, + }) + }) + { + session.give(wordcount); } - }); - } - }, - ) + } + }); + } + }) .container::() .unary_frontier( - ExchangeCore::,_>::new_core(|x: &WordCountReference<&[u8],&i64>| x.text.len() as u64), + ExchangeCore::, _>::new_core( + |x: &WordCountReference<&[u8], &i64>| x.text.len() as u64, + ), "WordCount", |_capability, _info| { let mut queues = HashMap::new(); @@ -71,7 +78,6 @@ fn main() { .entry(time.retain(output.output_index())) .or_insert(Vec::new()) .extend(data.map(std::mem::take)); - }); for (key, val) in queues.iter_mut() { @@ -79,16 +85,22 @@ fn main() { let mut session = output.session(key); for batch in val.drain(..) { for wordcount in batch.borrow().into_index_iter() { - let total = - if let Some(count) = counts.get_mut(wordcount.text) { + let total = if let Some(count) = + counts.get_mut(wordcount.text) + { *count += wordcount.diff; *count - } - else { - counts.insert(wordcount.text.to_vec(), *wordcount.diff); + } else { + counts.insert( + wordcount.text.to_vec(), + *wordcount.diff, + ); *wordcount.diff }; - session.give(WordCountReference { text: wordcount.text, diff: total }); + session.give(WordCountReference { + text: wordcount.text, + diff: total, + }); } } } @@ -99,23 +111,28 @@ fn main() { }, ) .container::() - .inspect_container(|x| { - match x { - Ok((time, data)) => { - println!("seen at: {:?}\t{:?} records", time, data.record_count()); - for wc in data.borrow().into_index_iter() { - println!(" {}: {}", std::str::from_utf8(wc.text).unwrap_or(""), wc.diff); - } - }, - Err(frontier) => println!("frontier advanced to {:?}", frontier), + .inspect_container(|x| match x { + Ok((time, data)) => { + println!("seen at: {:?}\t{:?} records", time, data.record_count()); + for wc in data.borrow().into_index_iter() { + println!( + " {}: {}", + std::str::from_utf8(wc.text).unwrap_or(""), + wc.diff + ); + } } + Err(frontier) => println!("frontier advanced to {:?}", frontier), }) .probe_with(&probe); }); // introduce data and watch! for round in 0..10 { - input.send(WordCountReference { text: "flat container", diff: 1 }); + input.send(WordCountReference { + text: "flat container", + diff: 1, + }); input.advance_to(round + 1); while probe.less_than(input.time()) { worker.step(); @@ -124,127 +141,3 @@ fn main() { }) .unwrap(); } - - -pub use container::Column; -mod container { - - use columnar::bytes::stash::Stash; - - #[derive(Clone, Default)] - pub struct Column { pub stash: Stash } - - use columnar::{Len, Index}; - use columnar::bytes::indexed; - use columnar::common::IterOwn; - - impl Column { - /// Borrows the contents no matter their representation. - #[inline(always)] pub fn borrow(&self) -> C::Borrowed<'_> { self.stash.borrow() } - } - - impl timely::Accountable for Column { - #[inline] fn record_count(&self) -> i64 { i64::try_from(self.borrow().len()).unwrap() } - #[inline] fn is_empty(&self) -> bool { self.borrow().is_empty() } - } - impl timely::container::DrainContainer for Column { - type Item<'a> = C::Ref<'a>; - type DrainIter<'a> = IterOwn>; - fn drain<'a>(&'a mut self) -> Self::DrainIter<'a> { self.borrow().into_index_iter() } - } - - impl timely::container::SizableContainer for Column { - fn at_capacity(&self) -> bool { - match &self.stash { - Stash::Typed(t) => { - let length_in_bytes = 8 * indexed::length_in_words(&t.borrow()); - length_in_bytes >= (1 << 20) - }, - Stash::Bytes(_) => true, - Stash::Align(_) => true, - } - } - fn ensure_capacity(&mut self, _stash: &mut Option) { } - } - - impl timely::container::PushInto for Column where C: columnar::Push { - #[inline] fn push_into(&mut self, item: T) { use columnar::Push; self.stash.push(item) } - } - - impl timely::dataflow::channels::ContainerBytes for Column { - fn from_bytes(bytes: timely::bytes::arc::Bytes) -> Self { Self { stash: Stash::try_from_bytes(bytes).expect("valid columnar data") } } - fn length_in_bytes(&self) -> usize { self.stash.length_in_bytes() } - fn into_bytes(&self, writer: &mut W) { self.stash.write_bytes(writer).expect("write failed") } - } -} - - -use builder::ColumnBuilder; -mod builder { - - use std::collections::VecDeque; - use columnar::bytes::{indexed, stash::Stash}; - use super::Column; - - /// A container builder for `Column`. - #[derive(Default)] - pub struct ColumnBuilder { - /// Container that we're writing to. - current: C, - /// Empty allocation. - empty: Option>, - /// Completed containers pending to be sent. - pending: VecDeque>, - } - - impl timely::container::PushInto for ColumnBuilder where C: columnar::Push { - #[inline] - fn push_into(&mut self, item: T) { - self.current.push(item); - // If there is less than 10% slop with 2MB backing allocations, mint a container. - let words = indexed::length_in_words(&self.current.borrow()); - let round = (words + ((1 << 18) - 1)) & !((1 << 18) - 1); - if round - words < round / 10 { - let mut alloc = Vec::with_capacity(round); - indexed::encode(&mut alloc, &self.current.borrow()); - self.pending.push_back(Column { stash: Stash::Align(alloc.into_boxed_slice().into()) }); - self.current.clear(); - } - } - } - - use timely::container::{ContainerBuilder, LengthPreservingContainerBuilder}; - impl ContainerBuilder for ColumnBuilder { - type Container = Column; - - #[inline] - fn extract(&mut self) -> Option<&mut Self::Container> { - if let Some(container) = self.pending.pop_front() { - self.empty = Some(container); - self.empty.as_mut() - } else { - None - } - } - - #[inline] - fn finish(&mut self) -> Option<&mut Self::Container> { - if !self.current.is_empty() { - self.pending.push_back(Column { stash: Stash::Typed(std::mem::take(&mut self.current)) }); - } - self.empty = self.pending.pop_front(); - self.empty.as_mut() - } - - #[inline] - fn relax(&mut self) { - // The caller is responsible for draining all contents; assert that we are empty. - // The assertion is not strictly necessary, but it helps catch bugs. - assert!(self.current.is_empty()); - assert!(self.pending.is_empty()); - *self = Self::default(); - } - } - - impl LengthPreservingContainerBuilder for ColumnBuilder { } -} diff --git a/timely/examples/transport_alloc.rs b/timely/examples/transport_alloc.rs new file mode 100644 index 000000000..68aa2cbd7 --- /dev/null +++ b/timely/examples/transport_alloc.rs @@ -0,0 +1,403 @@ +//! Measures steady-state allocations in typed and binary exchange paths. +//! +//! Run with, for example: +//! `cargo run --release --example transport_alloc -- binary columnar 4 100 10000`. + +use std::alloc::{GlobalAlloc, Layout, System}; +use std::hint::black_box; +use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; +use std::sync::{Arc, Barrier}; +use std::time::Instant; + +use serde::{Deserialize, Serialize}; +use timely::container::columnar::{ColumnarBuilder, ColumnarContainer}; +use timely::container::{CapacityContainerBuilder, ContainerBuilder, PushInto}; +use timely::dataflow::operators::{Exchange, InspectCore, Probe}; +use timely::dataflow::{InputHandle, ProbeHandle}; +use timely::Accountable; + +struct CountingAllocator; + +static ALLOCATIONS: AtomicUsize = AtomicUsize::new(0); +static ALLOCATED_BYTES: AtomicUsize = AtomicUsize::new(0); +static MEASURING: AtomicBool = AtomicBool::new(false); +static ALLOCATION_COUNTS: [AtomicUsize; 6] = [const { AtomicUsize::new(0) }; 6]; +static ALLOCATION_BYTES: [AtomicUsize; 6] = [const { AtomicUsize::new(0) }; 6]; +static RECEIVED_CONTAINERS: AtomicUsize = AtomicUsize::new(0); +static BYTE_BACKED_CONTAINERS: AtomicUsize = AtomicUsize::new(0); + +unsafe impl GlobalAlloc for CountingAllocator { + unsafe fn alloc(&self, layout: Layout) -> *mut u8 { + record_allocation(layout.size()); + // SAFETY: Delegates the allocation with the unchanged layout. + unsafe { System.alloc(layout) } + } + + unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) { + // SAFETY: Delegates the deallocation with the original pointer and layout. + unsafe { System.dealloc(ptr, layout) } + } + + unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, new_size: usize) -> *mut u8 { + record_allocation(new_size); + // SAFETY: Delegates the reallocation with the original allocation metadata. + unsafe { System.realloc(ptr, layout, new_size) } + } +} + +#[inline] +fn record_allocation(size: usize) { + if !MEASURING.load(Ordering::Relaxed) { + return; + } + let bucket = match size { + 0..=256 => 0, + 257..=4_096 => 1, + 4_097..=65_536 => 2, + 65_537..=262_144 => 3, + 262_145..=1_048_576 => 4, + _ => 5, + }; + ALLOCATIONS.fetch_add(1, Ordering::Relaxed); + ALLOCATED_BYTES.fetch_add(size, Ordering::Relaxed); + ALLOCATION_COUNTS[bucket].fetch_add(1, Ordering::Relaxed); + ALLOCATION_BYTES[bucket].fetch_add(size, Ordering::Relaxed); +} + +fn reset_allocations() { + ALLOCATIONS.store(0, Ordering::SeqCst); + ALLOCATED_BYTES.store(0, Ordering::SeqCst); + for counter in ALLOCATION_COUNTS.iter().chain(ALLOCATION_BYTES.iter()) { + counter.store(0, Ordering::SeqCst); + } +} + +fn allocation_histogram() -> String { + let labels = ["0-256", "257-4K", "4K-64K", "64K-256K", "256K-1M", ">1M"]; + labels + .iter() + .enumerate() + .map(|(index, label)| { + format!( + "{label}:{}:{}", + ALLOCATION_COUNTS[index].load(Ordering::SeqCst), + ALLOCATION_BYTES[index].load(Ordering::SeqCst), + ) + }) + .collect::>() + .join(",") +} + +#[global_allocator] +static GLOBAL: CountingAllocator = CountingAllocator; + +#[derive(Clone, Serialize, Deserialize, columnar::Columnar)] +struct Record { + key: u64, + payload: [u64; 7], +} + +#[derive(Clone, Copy)] +enum Transport { + Typed, + Binary, +} + +impl Transport { + fn config(self, workers: usize) -> timely::CommunicationConfig { + match self { + Transport::Typed => timely::CommunicationConfig::Process(workers), + Transport::Binary => timely::CommunicationConfig::ProcessBinary(workers), + } + } + + fn name(self) -> &'static str { + match self { + Transport::Typed => "typed", + Transport::Binary => "binary", + } + } +} + +fn main() { + let args = std::env::args().skip(1).collect::>(); + if args.len() != 5 { + eprintln!("usage: transport_alloc "); + std::process::exit(2); + } + + let transport = match args[0].as_str() { + "typed" => Transport::Typed, + "binary" => Transport::Binary, + other => panic!("unknown transport: {other}"), + }; + let workers = args[2].parse().expect("workers must be an integer"); + let rounds = args[3].parse().expect("rounds must be an integer"); + let records = args[4] + .parse() + .expect("records-per-round must be an integer"); + + match args[1].as_str() { + "vec" => run_vec(transport, workers, rounds, records), + "columnar" => run_columnar(transport, workers, rounds, records), + "columnar-builder" => run_columnar_builder(rounds, records), + other => panic!("unknown container: {other}"), + } +} + +fn run_columnar_builder(rounds: usize, records: usize) { + type Columns = ::Container; + let mut builder = ColumnarBuilder::::default(); + for record in 0..records { + let key = record as u64; + let payload = [key; 7]; + builder.push_into(RecordReference { + key: &key, + payload: &payload, + }); + } + while let Some(container) = builder.finish() { + black_box(container.borrow()); + } + builder.relax(); + reset_allocations(); + MEASURING.store(true, Ordering::SeqCst); + let start = Instant::now(); + for round in 0..rounds { + for record in 0..records { + let key = (round * records + record) as u64; + let payload = [key; 7]; + builder.push_into(RecordReference { + key: &key, + payload: &payload, + }); + } + while let Some(container) = builder.finish() { + black_box(container.borrow()); + } + builder.relax(); + } + let count = rounds * records; + let elapsed = start.elapsed().as_secs_f64(); + MEASURING.store(false, Ordering::SeqCst); + let allocations = ALLOCATIONS.load(Ordering::SeqCst); + let bytes = ALLOCATED_BYTES.load(Ordering::SeqCst); + println!( + "transport=none container=columnar-builder workers=1 records={} seconds={elapsed:.6} records_per_second={:.0} allocations={} allocated_bytes={} bytes_per_record={:.3} allocation_histogram={}", + count, + count as f64 / elapsed, + allocations, + bytes, + bytes as f64 / count as f64, + allocation_histogram(), + ); +} + +fn run_vec(transport: Transport, workers: usize, rounds: usize, records: usize) { + run( + transport, + "vec", + workers, + rounds, + records, + move |worker, shared| { + let mut input = InputHandle::>>::new(); + let mut probe = ProbeHandle::new(); + let seen = Arc::clone(&shared.seen); + + worker.dataflow::(|scope| { + input + .to_stream(scope) + .exchange(|record| record.key) + .inspect_container(move |event| { + if let Ok((_time, data)) = event { + seen.fetch_add(data.len(), Ordering::Relaxed); + black_box(data); + } + }) + .probe_with(&mut probe); + }); + + for record in 0..records { + let key = (record + worker.index()) as u64; + input.send(Record { + key, + payload: [key; 7], + }); + } + input.advance_to(1); + while probe.less_than(input.time()) { + worker.step(); + } + shared.start_measurement(worker.index()); + let start = Instant::now(); + for round in 0..rounds { + for record in 0..records { + let key = (round * records + record + worker.index()) as u64; + input.send(Record { + key, + payload: [key; 7], + }); + } + input.advance_to(round + 2); + while probe.less_than(input.time()) { + worker.step(); + } + } + shared.finish_measurement(start.elapsed(), worker.index()); + }, + ); +} + +fn run_columnar(transport: Transport, workers: usize, rounds: usize, records: usize) { + type Columns = ::Container; + type Container = ColumnarContainer; + + run( + transport, + "columnar", + workers, + rounds, + records, + move |worker, shared| { + let mut input = InputHandle::>::new_with_builder(); + let mut probe = ProbeHandle::new(); + let seen = Arc::clone(&shared.seen); + + worker.dataflow::(|scope| { + input + .to_stream(scope) + .exchange(|record| *record.key) + .inspect_container(move |event| { + if let Ok((_time, data)) = event { + seen.fetch_add(data.record_count() as usize, Ordering::Relaxed); + RECEIVED_CONTAINERS.fetch_add(1, Ordering::Relaxed); + BYTE_BACKED_CONTAINERS + .fetch_add(usize::from(data.is_bytes()), Ordering::Relaxed); + black_box(data.borrow()); + } + }) + .probe_with(&mut probe); + }); + + for record in 0..records { + let key = (record + worker.index()) as u64; + let payload = [key; 7]; + input.send(RecordReference { + key: &key, + payload: &payload, + }); + } + input.advance_to(1); + while probe.less_than(input.time()) { + worker.step(); + } + shared.start_measurement(worker.index()); + let start = Instant::now(); + for round in 0..rounds { + for record in 0..records { + let key = (round * records + record + worker.index()) as u64; + let payload = [key; 7]; + input.send(RecordReference { + key: &key, + payload: &payload, + }); + } + input.advance_to(round + 2); + while probe.less_than(input.time()) { + worker.step(); + } + } + shared.finish_measurement(start.elapsed(), worker.index()); + }, + ); + + // Keep the alias checked as part of the example; it also documents the + // concrete container users select for columnar streams. + let _: Option = None; +} + +struct Shared { + barrier: Barrier, + elapsed_ns: AtomicU64, + seen: Arc, + transport: &'static str, + container: &'static str, + workers: usize, + expected: usize, +} + +impl Shared { + fn start_measurement(&self, worker: usize) { + self.barrier.wait(); + if worker == 0 { + reset_allocations(); + self.seen.store(0, Ordering::SeqCst); + MEASURING.store(true, Ordering::SeqCst); + RECEIVED_CONTAINERS.store(0, Ordering::SeqCst); + BYTE_BACKED_CONTAINERS.store(0, Ordering::SeqCst); + } + self.barrier.wait(); + } + + fn finish_measurement(&self, elapsed: std::time::Duration, worker: usize) { + self.elapsed_ns + .fetch_max(elapsed.as_nanos() as u64, Ordering::Relaxed); + self.barrier.wait(); + if worker == 0 { + MEASURING.store(false, Ordering::SeqCst); + let seen = self.seen.load(Ordering::Relaxed); + assert_eq!(seen, self.expected); + let allocations = ALLOCATIONS.load(Ordering::SeqCst); + let bytes = ALLOCATED_BYTES.load(Ordering::SeqCst); + let received_containers = RECEIVED_CONTAINERS.load(Ordering::SeqCst); + let byte_backed_containers = BYTE_BACKED_CONTAINERS.load(Ordering::SeqCst); + let seconds = self.elapsed_ns.load(Ordering::Relaxed) as f64 / 1_000_000_000.0; + println!( + "transport={} container={} workers={} records={} seconds={seconds:.6} records_per_second={:.0} allocations={} allocated_bytes={} bytes_per_record={:.3} received_containers={} byte_backed_containers={} allocation_histogram={}", + self.transport, + self.container, + self.workers, + seen, + seen as f64 / seconds, + allocations, + bytes, + bytes as f64 / seen as f64, + received_containers, + byte_backed_containers, + allocation_histogram(), + ); + } + self.barrier.wait(); + } +} + +fn run( + transport: Transport, + container: &'static str, + workers: usize, + rounds: usize, + records: usize, + logic: F, +) where + F: Fn(&mut timely::worker::Worker, &Arc) + Send + Sync + 'static, +{ + let expected = workers * rounds * records; + let shared = Arc::new(Shared { + barrier: Barrier::new(workers), + elapsed_ns: AtomicU64::new(0), + seen: Arc::new(AtomicUsize::new(0)), + transport: transport.name(), + container, + workers, + expected, + }); + let worker_shared = Arc::clone(&shared); + let config = timely::Config { + communication: transport.config(workers), + worker: timely::WorkerConfig::default(), + }; + + timely::execute(config, move |worker| logic(worker, &worker_shared)) + .expect("timely execution should initialize"); +} diff --git a/timely/src/container/columnar.rs b/timely/src/container/columnar.rs new file mode 100644 index 000000000..37ce0156f --- /dev/null +++ b/timely/src/container/columnar.rs @@ -0,0 +1,323 @@ +//! Columnar containers for allocation-conscious data transport. +//! +//! [`ColumnarContainer`] stores either mutable typed columns or an immutable +//! view over serialized communication bytes. With binary communication, a +//! receiver can therefore inspect records without reconstructing owned rows. +//! [`ColumnarBuilder`] assembles typed columns and reclaims column allocations +//! returned by synchronous serializers. + +use std::collections::VecDeque; + +use ::columnar::bytes::stash::Stash; +use ::columnar::{Index, Len}; + +use crate::bytes::arc::Bytes; +use crate::container::{ + Accountable, ContainerBuilder, DrainContainer, LengthPreservingContainerBuilder, PushInto, + SizableContainer, +}; +use crate::dataflow::channels::ContainerBytes; + +/// Preferred serialized size of a columnar transport container. +pub const DEFAULT_BUFFER_BYTES: usize = 1 << 20; + +/// A columnar container that is either typed or backed by communication bytes. +#[derive(Clone, Default)] +pub struct ColumnarContainer { + stash: Stash, +} + +impl ColumnarContainer { + /// Borrows the columnar contents independent of their current representation. + #[inline(always)] + pub fn borrow(&self) -> C::Borrowed<'_> { + self.stash.borrow() + } + + /// Returns true when the container directly retains serialized bytes. + pub fn is_bytes(&self) -> bool { + matches!(self.stash, Stash::Bytes(_)) + } + + fn typed(container: C) -> Self { + Self { + stash: Stash::Typed(container), + } + } + + fn take_typed(&mut self) -> Option { + match std::mem::take(&mut self.stash) { + Stash::Typed(mut container) => { + ::columnar::Clear::clear(&mut container); + Some(container) + } + Stash::Bytes(_) | Stash::Align(_) => None, + } + } +} + +impl Accountable for ColumnarContainer { + #[inline] + fn record_count(&self) -> i64 { + i64::try_from(self.borrow().len()).expect("columnar record count must fit in i64") + } + + #[inline] + fn is_empty(&self) -> bool { + self.borrow().is_empty() + } +} + +impl DrainContainer for ColumnarContainer { + type Item<'a> + = C::Ref<'a> + where + C: 'a; + type DrainIter<'a> + = ::columnar::common::IterOwn> + where + C: 'a; + + #[inline] + fn drain(&mut self) -> Self::DrainIter<'_> { + self.borrow().into_index_iter() + } +} + +impl SizableContainer for ColumnarContainer { + fn at_capacity(&self) -> bool { + self.stash.length_in_bytes() >= DEFAULT_BUFFER_BYTES + } + + fn ensure_capacity(&mut self, spare: &mut Option) { + if matches!(self.stash, Stash::Typed(_)) { + // `CapacityContainerBuilder` leaves a default typed container in + // `current` after sending and places the container returned by the + // pusher in `spare`. At the start of the next batch, prefer that + // returned allocation. Once a record has been pushed this branch + // no longer swaps, so the working container remains stable. + if self.is_empty() + && spare + .as_ref() + .is_some_and(|candidate| matches!(candidate.stash, Stash::Typed(_))) + { + std::mem::swap(self, spare.as_mut().expect("checked above")); + if let Stash::Typed(container) = &mut self.stash { + ::columnar::Clear::clear(container); + } + } + return; + } + if let Some(mut spare) = spare.take() { + if let Some(container) = spare.take_typed() { + self.stash = Stash::Typed(container); + return; + } + } + self.stash = Stash::Typed(C::default()); + } +} + +impl PushInto for ColumnarContainer +where + C: ::columnar::Container + ::columnar::ContainerBytes + ::columnar::Push, +{ + #[inline] + fn push_into(&mut self, item: T) { + ::columnar::Push::push(&mut self.stash, item); + } +} + +impl ContainerBytes for ColumnarContainer { + fn from_bytes(bytes: Bytes) -> Self { + Self { + stash: Stash::try_from_bytes(bytes).expect("valid columnar container bytes"), + } + } + + fn length_in_bytes(&self) -> usize { + self.stash.length_in_bytes() + } + + fn into_bytes(&self, writer: &mut W) { + self.stash + .write_bytes(writer) + .expect("columnar container write failed") + } +} + +/// Builds bounded-size [`ColumnarContainer`] batches from individual records. +/// +/// Typed column allocations returned by a pusher are retained in a small pool. +/// This is particularly effective with binary communication, whose serializer +/// returns the typed container immediately after writing it into a byte slab. +pub struct ColumnarBuilder { + current: C, + needs_current: bool, + returned: Option>, + spares: Vec, + pending: VecDeque>, +} + +impl Default for ColumnarBuilder { + fn default() -> Self { + Self { + current: C::default(), + needs_current: false, + returned: None, + spares: Vec::new(), + pending: VecDeque::new(), + } + } +} + +impl + ColumnarBuilder +{ + const MAX_SPARES: usize = 2; + + fn reclaim_returned(&mut self) { + if let Some(mut returned) = self.returned.take() { + if let Some(container) = returned.take_typed() { + // Prefer the most recently returned containers. Early sends + // commonly leave allocation-free defaults behind; retaining + // those forever would crowd out useful allocations that make + // the round trip through a channel later. + if self.spares.len() == Self::MAX_SPARES { + self.spares.remove(0); + } + self.spares.push(container); + } + } + } + + fn ensure_current(&mut self) { + if self.needs_current { + self.reclaim_returned(); + self.current = self.spares.pop().unwrap_or_default(); + self.needs_current = false; + } + } + + fn emit_current(&mut self) { + if !self.current.is_empty() { + self.pending + .push_back(ColumnarContainer::typed(std::mem::take(&mut self.current))); + self.needs_current = true; + } + } +} + +impl PushInto for ColumnarBuilder +where + C: ::columnar::ContainerBytes + ::columnar::Push, +{ + #[inline] + fn push_into(&mut self, item: T) { + assert!( + PREFERRED_BYTES > 0, + "preferred columnar batch size must be non-zero" + ); + self.ensure_current(); + ::columnar::Push::push(&mut self.current, item); + if ::columnar::bytes::indexed::length_in_words(&self.current.borrow()) * 8 + >= PREFERRED_BYTES + { + self.emit_current(); + } + } +} + +impl ContainerBuilder + for ColumnarBuilder +{ + type Container = ColumnarContainer; + + fn extract(&mut self) -> Option<&mut Self::Container> { + self.reclaim_returned(); + self.returned = self.pending.pop_front(); + self.returned.as_mut() + } + + fn finish(&mut self) -> Option<&mut Self::Container> { + if !self.needs_current { + self.emit_current(); + } + self.extract() + } + + fn relax(&mut self) { + assert!( + self.pending.is_empty(), + "finish must drain pending columnar containers" + ); + assert!(self.needs_current || self.current.is_empty()); + // `relax` occurs at the end of each pushed sequence, often once per + // scheduling activation. Releasing the returned columns here would + // turn normal progress boundaries into allocation boundaries. Keep a + // bounded working set; dropping the builder still releases it. + self.reclaim_returned(); + self.ensure_current(); + } +} + +impl LengthPreservingContainerBuilder + for ColumnarBuilder +{ +} + +#[cfg(test)] +mod tests { + use super::*; + + #[derive(::columnar::Columnar)] + struct TestRecord { + key: u64, + value: String, + } + + type Columns = ::Container; + + #[test] + fn serialized_container_retains_received_bytes() { + let mut original = ColumnarContainer::::default(); + original.push_into(TestRecordReference { + key: &7, + value: "seven", + }); + + let mut encoded = Vec::new(); + ContainerBytes::into_bytes(&original, &mut encoded); + let received = ColumnarContainer::::from_bytes( + crate::bytes::arc::BytesMut::from(encoded).freeze(), + ); + + assert!(received.is_bytes()); + assert_eq!(received.record_count(), 1); + let record = received.borrow().get(0); + assert_eq!(*record.key, 7); + assert_eq!(record.value, b"seven"); + } + + #[test] + fn builder_reclaims_a_returned_typed_container() { + let mut builder = ColumnarBuilder::::default(); + for key in 0..16 { + builder.push_into(TestRecordReference { + key: &key, + value: "value", + }); + } + + assert!(builder.extract().is_some()); + while builder.extract().is_some() {} + let reclaimed = builder.spares.len(); + assert!(reclaimed > 0); + + builder.push_into(TestRecordReference { + key: &99, + value: "again", + }); + assert_eq!(builder.spares.len(), reclaimed - 1); + } +} diff --git a/timely/src/lib.rs b/timely/src/lib.rs index 39625f7e3..63deb47b0 100644 --- a/timely/src/lib.rs +++ b/timely/src/lib.rs @@ -69,6 +69,9 @@ pub use timely_container::Accountable; /// Re-export of the `timely_container` crate. pub mod container { pub use timely_container::*; + + /// Columnar containers that can retain serialized communication bytes. + pub mod columnar; } /// Re-export of the `timely_communication` crate. From 8f4b38c431b9ba3ed6c7cb24a066d2e4566741e4 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Sat, 15 Aug 2026 07:58:10 -0400 Subject: [PATCH 3/5] Document transport allocation results and design options --- TRANSPORT_ENGINEERING.md | 189 ++++++++++++++++++++++++++++ mdbook/Cargo.toml | 1 + mdbook/src/chapter_5/chapter_5_3.md | 48 +++++++ 3 files changed, 238 insertions(+) create mode 100644 TRANSPORT_ENGINEERING.md diff --git a/TRANSPORT_ENGINEERING.md b/TRANSPORT_ENGINEERING.md new file mode 100644 index 000000000..fe399cd6c --- /dev/null +++ b/TRANSPORT_ENGINEERING.md @@ -0,0 +1,189 @@ +# Transport and progress engineering assessment + +This assessment covers the current six-crate workspace, the columnar transport +experiment, and the concurrent progress structure proposed in pull request +807. Measurements below were made on an Apple Silicon host with four workers, +release builds, 5,000 fixed-width 64-byte records per worker per logical round, +and all-to-all routing. Allocation counts include `alloc` and `realloc`; bytes +are requested bytes, not live-set size. + +## Implemented transport tranche + +- `timely::container::columnar::{ColumnarContainer, ColumnarBuilder}` promotes + the former example-only implementation into supported infrastructure. + Received binary containers retain their serialized `Bytes`; typed column + allocations are recycled through a bounded two-container builder pool. +- `CapacityContainerBuilder` now retains one returned spare across `finish()`. + Previously, the drain-completing call overwrote that spare with `None`, making + every logical timestamp an allocation boundary. +- `ColumnarContainer::ensure_capacity` selects the typed allocation returned by + the pusher when starting the next empty batch. The default typed container in + `current` previously masked the useful spare. +- The thread-local channel returns at most 16 consumed values to its producer. + Ownership-taking `recv()` calls do not return a value, and tests cover both + cases and the bound. Generic input handles swap a bounded set of consumed + containers back into later pull slots, completing that ownership round trip + without reclaiming containers explicitly taken by user code. +- `timely/examples/transport_alloc.rs` supplies a repeatable typed/binary × + vector/columnar matrix, allocation-size histogram, warmup, record-count + validation, and a direct builder measurement. + +### Results + +The diagnostic, pre-fix binary-columnar run allocated 257.6 bytes/record +(28,367 allocation calls for one million records). The same exact unwarmed run +after retaining and selecting the returned spare allocated 26.4 bytes/record +(2,921 calls), a 9.8x reduction in requested bytes and 9.7x fewer allocation +calls. + +The reduction came from two necessary fixes. `CapacityContainerBuilder::finish` +discarded the container returned by the pusher on its final drain call. Fixing +that alone did not improve the measurement: `ColumnarContainer::ensure_capacity` +still preferred an allocation-free typed `current` over the useful typed spare. +Once it selected the returned spare, repeated geometric growth fell from +126.8 MB to 3.9 MB in the 4–64 KiB allocation bucket and from 93.1 MB to +3.2 MB in the 64–256 KiB bucket. The 18.6 MB of communication-slab growth was +unchanged. The promoted builder, thread-local resource return, and input-handle +swap complete other ownership paths but did not cause this headline +`ProcessBinary` reduction. + +After a one-round warmup, a two-million-record matrix produced: + +| Transport | Container | allocations | bytes/record | records/s | +|---|---:|---:|---:|---:| +| typed | `Vec` | 19,318 | 66.6 | 353M | +| binary | `Vec` | 16,863 | 72.5 | 97M | +| typed | columnar | 53,617 | 236.4 | 152M | +| binary | columnar | 4,050 | 10.1 | 176M | +| none | direct columnar builder | 0 | 0.0 | 538M | + +These are diagnostic microbenchmarks, not promises about application +throughput. The allocation result is robust: direct construction is +allocation-free after warmup, and binary columnar eliminates the repeated +4 KiB–256 KiB geometric growth seen in typed columnar. Its remaining measured +bytes are almost entirely nineteen 1 MiB communication-slab acquisitions and +therefore amortize with a longer run. + +After porting the change from the original 0.29 snapshot to upstream 0.31 and +its `columnar` 0.13 dependency, the warmed binary-columnar case reproduced at +9.58 allocated bytes/record and 95.9 million records/second. The original exact +pre/post timing moved from 64.2 to 84.4 million records/second, but those runs +were only 12–22 ms; treat the apparent 31% speedup as directional until an +interleaved multi-second A/B benchmark is available. + +Typed columnar remains a regression because `CommunicationConfig::Process` +uses one-way `std::sync::mpsc` ownership transfer. There is no route by which a +target can return a consumed container to the correct source. The columnar +layout multiplies geometric growth across its component columns, so losing the +whole batch allocation each round is more expensive than losing one vector. +The supported recommendation is consequently columnar plus `ProcessBinary` +(or cluster zero-copy communication), not columnar plus typed process channels. + +## Pull request 807: progress exchange + +The PR's central diagnosis is right: broadcast MPSC queues make each sender +clone progress batches for every reader, preserve obsolete intermediate state, +and charge a laggard for the full history rather than the consolidated net. +Its shared compacting chain demonstrably protects laggards and bounds backlog, +but the single shared head moves the cost to the healthy case. The PR's own +measurements show a 2–7x synthetic send-path loss and a 6.6x loss at eight +workers in the progress-heavy `event_driven` workload, while data-heavy +PageRank is approximately unchanged. That agrees with the reported experience +that no overall improvement was measurable. + +My disposition would also be “keep as an experiment, do not make it the sole +default.” It optimizes an important failure mode, but forces every healthy +worker through a globally written cache line and nested lock protocol. It is a +resource-governance improvement, not yet a throughput improvement. + +### A more promising shape + +Use a bounded, hierarchical combining tree rather than either W broadcast +queues or one global chain: + +1. Each worker publishes into a single-producer local delta slot/log, with a + monotonically increasing generation. It never clones per reader. +2. One combiner per small socket-local group (for example 4–8 workers) drains + changed generations into a consolidated group accumulator. Writers use a + `try_lock`; on contention they retain and consolidate into their local slot + rather than waiting on a global head. +3. Group accumulators feed a second-level accumulator only when their net + changes. Readers track a generation per group and fold the latest + consolidated snapshots. +4. Put an explicit byte/entry budget on every local slot. A lagging publisher + consolidates more aggressively; a lagging reader does not prevent writers + or other readers from reclaiming historical nodes. + +This gives cross-writer cancellation within groups, reduces shared-cacheline +fan-in from W to roughly the group size, and makes laggard work proportional to +current consolidated state rather than elapsed sends. It does change the +proof obligation: publication must expose an atomic snapshot/generation pair, +and reclamation must wait until all readers have acknowledged that generation. +An epoch or two-buffer seqlock cell is simpler to audit than a mutable linked +chain. + +Two useful variants should be benchmarked before implementation: + +- **Striped ledger by topology, not by key.** Each send remains atomic and goes + to the writer's group stripe; a reader consolidates the small set of stripes. + This preserves send atomicity while allowing cross-writer cancellation inside + each stripe. +- **RCU snapshot plus delta inbox.** Writers append small deltas to bounded + per-writer SPSC rings. A combiner periodically publishes an immutable + consolidated `Arc` snapshot. Readers normally clone one snapshot and process + only deltas newer than its generation. A laggard jumps to a newer snapshot + instead of replaying history. + +The benchmark acceptance criteria should be stated as a Pareto frontier: +healthy progress-heavy throughput, p99 send/receive latency, retained bytes +with one unread worker, and catch-up work after 1/16/1024 scheduling rounds. +A single throughput number hides the protection that motivated the structure. + +## Broader engineering assessment + +The codebase has unusually clean conceptual seams: bytes, containers, +communication, progress, scheduling, and operator construction are separate +crates/modules; the `Push<&mut Option>` ownership slot is a strong and +underused abstraction; and progress correctness is largely isolated from data +representation. Tests are small and generally exercise semantic contracts. + +The highest complexity is concentrated in `progress/reachability.rs`, +`progress/frontier.rs`, `progress/subgraph.rs`, `worker.rs`, and the generic +operator builders. Their complexity is mostly inherent, but several incidental +costs can be removed: + +- Consolidate the three generic builder implementations (`builder_raw`, + `builder_rc`, and `builder_ref`) around one internal wiring/state machine. + Keep the public APIs as adapters; today duplicated frontier, capability, and + shutdown bookkeeping makes changes harder to audit. +- Split `progress/subgraph.rs` into topology construction, runtime progress + exchange, and scheduling/activation state. This would make it possible to + replace the progress medium without editing the progress calculus. +- Treat `ContainerBuilder::relax` as “trim excess, retain a bounded working + set,” not “drop all storage.” It is called at scheduling boundaries and can + silently become a steady-state allocation boundary. +- Make resource-return behavior an explicit allocator capability. `Process` + cannot return typed resources to a source; binary allocators return the + typed input immediately; thread-local channels can return consumed values. + Encoding this distinction in types or diagnostics would prevent container + choices whose recycling assumptions cannot be met. +- Separate benchmark-only concurrent structures from exported communication + primitives. `communication/src/chain.rs` is currently not exported or wired + into `Progcaster`; its presence in the source tree otherwise suggests a + supported facility that does not exist. + +## Deferred work + +- Do not skip `columnar::Stash::try_from_bytes` validation by default. The + receive path is already byte-backed; unchecked construction would weaken a + network trust boundary for little demonstrated gain. +- A bidirectional typed process channel could recycle containers, but routing a + returned generic `T` to its original sender requires per-source receive lanes + or protocol metadata. That is a larger allocator redesign and should be + measured against simply using `ProcessBinary`. +- The next columnar experiment should use variable-width strings and nested + records. Fixed-width rows establish recycling behavior, but do not quantify + the layout's main cache-locality and allocation-count advantage. +- Run PR 807 and the hierarchical variants on a many-core, multi-socket Linux + host. The current single-socket Apple Silicon result is enough to reject a + universal default, not enough to reject the laggard-protection design goal. diff --git a/mdbook/Cargo.toml b/mdbook/Cargo.toml index ce2643b4d..50c8b6e85 100644 --- a/mdbook/Cargo.toml +++ b/mdbook/Cargo.toml @@ -9,6 +9,7 @@ publish = false workspace = true [dependencies] +columnar = { workspace = true } timely = { path = "../timely" } timely_bytes = { path = "../bytes" } timely_communication = { path = "../communication" } diff --git a/mdbook/src/chapter_5/chapter_5_3.md b/mdbook/src/chapter_5/chapter_5_3.md index 0766d49e5..7266fea97 100644 --- a/mdbook/src/chapter_5/chapter_5_3.md +++ b/mdbook/src/chapter_5/chapter_5_3.md @@ -23,6 +23,54 @@ What we want to achieve is: In Timely, we provide a set of `core` operators that are generic on the container type they can handle. In most cases, the `core` operators are an immediate generalization of their non-core variant, providing the semantically equivalent functionality. ++## Columnar transport + +The `timely::container::columnar` module provides `ColumnarContainer` and +`ColumnarBuilder` for records deriving `columnar::Columnar`. The container can +hold mutable typed columns while it is being assembled and retain a borrowed +view over communication bytes after binary transport. This avoids rebuilding +owned rows at the receiver. + +```rust +use columnar::Index; +use timely::container::columnar::ColumnarBuilder; +use timely::dataflow::operators::{Exchange, InspectCore}; +use timely::dataflow::InputHandle; + +#[derive(columnar::Columnar)] +struct Record { + key: u64, + value: String, +} + +type Columns = ::Container; + +timely::example(|scope| { + let mut input = InputHandle::>::new_with_builder(); + input + .to_stream(scope) + .exchange(|record| *record.key) + .inspect_container(|event| { + if let Ok((_time, records)) = event { + for record in records.borrow().into_index_iter() { + println!("{}: {:?}", record.key, record.value); + } + } + }); + + input.send(RecordReference { key: &0, value: "zero" }); +}); +``` + +`CommunicationConfig::ProcessBinary` and cluster communication serialize a +typed columnar container into shared byte slabs and return its column +allocations immediately. This is the path on which the builder's bounded +recycling is most effective. `CommunicationConfig::Process` moves typed +containers through one-way inter-thread channels; it currently has no matching +resource-return path, so columnar exchange builders may need to regrow their +columns at each logical time. + + ## Limitations From 6bded495222d6bce789ae893ef90e3a88282f6d6 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Sat, 15 Aug 2026 08:15:34 -0400 Subject: [PATCH 4/5] Bound quiescent transport memory --- TRANSPORT_ENGINEERING.md | 144 +++++++++--------- communication/src/allocator/thread.rs | 77 +--------- container/src/lib.rs | 33 +--- mdbook/src/chapter_5/chapter_5_3.md | 11 +- timely/examples/columnar.rs | 8 +- timely/examples/transport_alloc.rs | 6 +- timely/src/container/columnar.rs | 74 ++++++--- .../src/dataflow/operators/generic/handles.rs | 23 +-- 8 files changed, 156 insertions(+), 220 deletions(-) diff --git a/TRANSPORT_ENGINEERING.md b/TRANSPORT_ENGINEERING.md index fe399cd6c..695ab17b5 100644 --- a/TRANSPORT_ENGINEERING.md +++ b/TRANSPORT_ENGINEERING.md @@ -7,77 +7,81 @@ release builds, 5,000 fixed-width 64-byte records per worker per logical round, and all-to-all routing. Allocation counts include `alloc` and `realloc`; bytes are requested bytes, not live-set size. -## Implemented transport tranche +## Memory-footprint constraint + +Transport scratch space must obey this quiescent-state invariant: + +> Quiescent retained transport capacity is bounded per worker or allocator, +> independent of the number of logical channels and destination workers. + +A fixed per-channel pool does not satisfy this property, even when its local +bound looks small. An exchange distributor owns one builder per destination. +Retaining one preferred-size container in every builder can therefore scale as +`channels × source workers × destination workers × container capacity`. At +10,000 exchange channels, 100 workers, and 1 MiB containers, the theoretical +process-wide bound is 100 TB. Sixteen-container input or thread-channel pools +scale as `channels × workers × 16 × capacity` and are also unacceptable. + +The zero-copy allocator is the appropriate place to retain a process-wide, +budgeted set of compact byte slabs. Typed column scratch space should disappear +when its logical channel becomes quiescent. + +## Revised transport tranche - `timely::container::columnar::{ColumnarContainer, ColumnarBuilder}` promotes the former example-only implementation into supported infrastructure. - Received binary containers retain their serialized `Bytes`; typed column - allocations are recycled through a bounded two-container builder pool. -- `CapacityContainerBuilder` now retains one returned spare across `finish()`. - Previously, the drain-completing call overwrote that spare with `None`, making - every logical timestamp an allocation boundary. -- `ColumnarContainer::ensure_capacity` selects the typed allocation returned by - the pusher when starting the next empty batch. The default typed container in - `current` previously masked the useful spare. -- The thread-local channel returns at most 16 consumed values to its producer. - Ownership-taking `recv()` calls do not return a value, and tests cover both - cases and the bound. Generic input handles swap a bounded set of consumed - containers back into later pull slots, completing that ownership round trip - without reclaiming containers explicitly taken by user code. + Binary receivers retain a view into compact communication `Bytes` rather + than reconstructing owned rows. +- A columnar builder may recycle returned typed columns only while its current + sequence is active. The final `finish()` call and `relax()` both release + `current`, returned containers, and the bounded transient spare list. +- The proposed generic changes to `CapacityContainerBuilder`, + `InputHandleCore`, and the thread-local allocator were removed. Their + seemingly small per-instance bounds multiplied by channel and worker counts. +- Tests include 10,000 independently activated and quiesced builders and assert + that none retains a current allocation or pooled container. - `timely/examples/transport_alloc.rs` supplies a repeatable typed/binary × - vector/columnar matrix, allocation-size histogram, warmup, record-count - validation, and a direct builder measurement. - -### Results - -The diagnostic, pre-fix binary-columnar run allocated 257.6 bytes/record -(28,367 allocation calls for one million records). The same exact unwarmed run -after retaining and selecting the returned spare allocated 26.4 bytes/record -(2,921 calls), a 9.8x reduction in requested bytes and 9.7x fewer allocation -calls. - -The reduction came from two necessary fixes. `CapacityContainerBuilder::finish` -discarded the container returned by the pusher on its final drain call. Fixing -that alone did not improve the measurement: `ColumnarContainer::ensure_capacity` -still preferred an allocation-free typed `current` over the useful typed spare. -Once it selected the returned spare, repeated geometric growth fell from -126.8 MB to 3.9 MB in the 4–64 KiB allocation bucket and from 93.1 MB to -3.2 MB in the 64–256 KiB bucket. The 18.6 MB of communication-slab growth was -unchanged. The promoted builder, thread-local resource return, and input-handle -swap complete other ownership paths but did not cause this headline -`ProcessBinary` reduction. - -After a one-round warmup, a two-million-record matrix produced: - -| Transport | Container | allocations | bytes/record | records/s | -|---|---:|---:|---:|---:| -| typed | `Vec` | 19,318 | 66.6 | 353M | -| binary | `Vec` | 16,863 | 72.5 | 97M | -| typed | columnar | 53,617 | 236.4 | 152M | -| binary | columnar | 4,050 | 10.1 | 176M | -| none | direct columnar builder | 0 | 0.0 | 538M | - -These are diagnostic microbenchmarks, not promises about application -throughput. The allocation result is robust: direct construction is -allocation-free after warmup, and binary columnar eliminates the repeated -4 KiB–256 KiB geometric growth seen in typed columnar. Its remaining measured -bytes are almost entirely nineteen 1 MiB communication-slab acquisitions and -therefore amortize with a longer run. - -After porting the change from the original 0.29 snapshot to upstream 0.31 and -its `columnar` 0.13 dependency, the warmed binary-columnar case reproduced at -9.58 allocated bytes/record and 95.9 million records/second. The original exact -pre/post timing moved from 64.2 to 84.4 million records/second, but those runs -were only 12–22 ms; treat the apparent 31% speedup as directional until an -interleaved multi-second A/B benchmark is available. - -Typed columnar remains a regression because `CommunicationConfig::Process` -uses one-way `std::sync::mpsc` ownership transfer. There is no route by which a -target can return a consumed container to the correct source. The columnar -layout multiplies geometric growth across its component columns, so losing the -whole batch allocation each round is more expensive than losing one vector. -The supported recommendation is consequently columnar plus `ProcessBinary` -(or cluster zero-copy communication), not columnar plus typed process channels. + vector/columnar matrix, allocation-size histogram, warmup, and record-count + validation. + +### Allocation versus retention + +The initial recycling experiment found a real allocation mechanism. For one +million records, binary columnar fell from 257.6 to 26.4 requested bytes/record +and allocation calls fell 9.7x. Repeated geometric growth fell from 126.8 MB to +3.9 MB in the 4–64 KiB bucket and from 93.1 MB to 3.2 MB in the 64–256 KiB +bucket. The exact short-run throughput moved from 64.2 to 84.4 million +records/second. + +That result was not free: it converted allocation churn into long-lived +per-destination typed column capacity. On upstream 0.31 the retained variant +reached 9.58 allocated bytes/record and 95.9 million records/second after +warmup, but violated the quiescent-state invariant above. Those numbers are +recorded as a rejected point in the tradeoff space, not as the behavior of the +revised PR. + +With all per-channel retention removed, the same two-million-record matrix +produced: + +| Transport | Container | allocated bytes/record | records/s | +|---|---:|---:|---:| +| typed | `Vec` | 68.4 | 309M | +| binary | `Vec` | 80.7 | 98M | +| typed | columnar | 472.3 | 85M | +| binary | columnar | 480.3 | 64M | +| none | direct columnar builder | 236.0 | 134M | + +The fixed-width columnar microbenchmark is now deliberately allocation-heavy: +source and exchange scratch columns are regrown after each quiescence boundary. +It demonstrates that the 9.8x allocation reduction was purchased with retained +state. Columnar transport can still be useful for variable-width records, +receiver-side borrowed access, and avoiding reconstruction of owned rows, but +the fixed-width result is not a throughput recommendation. + +A future recycling design should use a worker-wide byte budget shared among +active channels, rather than a count embedded in each builder. Doing that well +requires a capacity-reporting/reinitialization contract for generic containers; +it is deferred rather than hidden behind an unsafe aggregate memory bound. ## Pull request 807: progress exchange @@ -159,9 +163,9 @@ costs can be removed: - Split `progress/subgraph.rs` into topology construction, runtime progress exchange, and scheduling/activation state. This would make it possible to replace the progress medium without editing the progress calculus. -- Treat `ContainerBuilder::relax` as “trim excess, retain a bounded working - set,” not “drop all storage.” It is called at scheduling boundaries and can - silently become a steady-state allocation boundary. +- Specify `ContainerBuilder::relax` as a quiescence and memory-reclamation + boundary. Any future retention should be charged to an explicit worker-wide + byte budget, not an implicit per-builder count. - Make resource-return behavior an explicit allocator capability. `Process` cannot return typed resources to a source; binary allocators return the typed input immediately; thread-local channels can return consumed values. diff --git a/communication/src/allocator/thread.rs b/communication/src/allocator/thread.rs index 72f933e83..9857ed5ba 100644 --- a/communication/src/allocator/thread.rs +++ b/communication/src/allocator/thread.rs @@ -94,78 +94,13 @@ impl Pull for Puller { #[inline] fn pull(&mut self) -> &mut Option { let mut borrow = self.source.borrow_mut(); - if let Some(element) = self.current.take() { - // Retain a bounded number of values for the producer to reclaim. - // Consumers that took the value with `recv()` leave `current` as - // `None` and therefore correctly return nothing. - if borrow.1.len() < 16 { - borrow.1.push_back(element); - } - } + // if let Some(element) = self.current.take() { + // // TODO : Arbitrary constant. + // if borrow.1.len() < 16 { + // borrow.1.push_back(element); + // } + // } self.current = borrow.0.pop_front(); &mut self.current } } - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn pull_returns_unconsumed_resources_to_the_producer() { - let events = Rc::new(RefCell::new(Vec::new())); - let (mut pusher, mut puller) = Thread::new_from(0, events); - - let mut first = Vec::with_capacity(1024); - first.extend(0..16); - let first_capacity = first.capacity(); - let mut first = Some(first); - pusher.push(&mut first); - assert!(first.is_none()); - - puller.pull().as_mut().unwrap().clear(); - assert!(puller.pull().is_none()); - - let mut second = Some(vec![99]); - pusher.push(&mut second); - let returned = second.expect("producer should reclaim the prior value"); - assert!(returned.is_empty()); - assert_eq!(returned.capacity(), first_capacity); - } - - #[test] - fn recv_transfers_ownership_without_returning_a_resource() { - let events = Rc::new(RefCell::new(Vec::new())); - let (mut pusher, mut puller) = Thread::new_from(0, events); - - pusher.send(vec![1, 2, 3]); - assert_eq!(puller.recv(), Some(vec![1, 2, 3])); - assert!(puller.pull().is_none()); - - let mut next = Some(vec![4]); - pusher.push(&mut next); - assert!(next.is_none(), "a taken value must not appear in the return queue"); - } - - #[test] - fn resource_return_queue_is_bounded() { - let events = Rc::new(RefCell::new(Vec::new())); - let (mut pusher, mut puller) = Thread::new_from(0, events); - - for value in 0..32 { - pusher.send(vec![value]); - } - for _ in 0..32 { - puller.pull().as_mut().unwrap().clear(); - } - assert!(puller.pull().is_none()); - - let mut returned = 0; - for _ in 0..32 { - let mut item = Some(Vec::new()); - pusher.push(&mut item); - returned += usize::from(item.is_some()); - } - assert_eq!(returned, 16); - } -} diff --git a/container/src/lib.rs b/container/src/lib.rs index 41acd0f48..400512475 100644 --- a/container/src/lib.rs +++ b/container/src/lib.rs @@ -136,9 +136,8 @@ mod noop { /// A default container builder that uses length and preferred capacity to chunk data. /// -/// Maintains a single empty allocation for reuse across calls, including across -/// [`Self::finish`]. This bounds retained memory to one container while avoiding -/// an allocation boundary at each logical timestamp. +/// Maintains a single empty allocation between [`Self::push_into`] and [`Self::extract`], but not +/// across [`Self::finish`] to maintain a low memory footprint. /// /// Maintains FIFO order. #[derive(Default, Debug)] @@ -185,37 +184,13 @@ impl ContainerBuilder for CapacityContainerBuilder if !self.current.is_empty() { self.pending.push_back(std::mem::take(&mut self.current)); } - if let Some(container) = self.pending.pop_front() { - self.empty = Some(container); - self.empty.as_mut() - } else { - None - } + self.empty = self.pending.pop_front(); + self.empty.as_mut() } } impl LengthPreservingContainerBuilder for CapacityContainerBuilder { } -#[cfg(test)] -mod tests { - use super::{CapacityContainerBuilder, ContainerBuilder, PushInto}; - - #[test] - fn capacity_builder_retains_returned_container_across_finish() { - let mut builder = CapacityContainerBuilder::>::default(); - builder.push_into(1); - - let container = builder.finish().expect("one finished container"); - let allocation = container.as_ptr(); - container.clear(); - assert!(builder.finish().is_none()); - - builder.push_into(2); - let reused = builder.finish().expect("one finished container"); - assert_eq!(reused.as_ptr(), allocation); - } -} - impl Accountable for Vec { #[inline] fn record_count(&self) -> i64 { i64::try_from(Vec::len(self)).unwrap() } #[inline] fn is_empty(&self) -> bool { Vec::is_empty(self) } diff --git a/mdbook/src/chapter_5/chapter_5_3.md b/mdbook/src/chapter_5/chapter_5_3.md index 7266fea97..f57ef5d46 100644 --- a/mdbook/src/chapter_5/chapter_5_3.md +++ b/mdbook/src/chapter_5/chapter_5_3.md @@ -64,11 +64,12 @@ timely::example(|scope| { `CommunicationConfig::ProcessBinary` and cluster communication serialize a typed columnar container into shared byte slabs and return its column -allocations immediately. This is the path on which the builder's bounded -recycling is most effective. `CommunicationConfig::Process` moves typed -containers through one-way inter-thread channels; it currently has no matching -resource-return path, so columnar exchange builders may need to regrow their -columns at each logical time. +allocations immediately. A builder can reuse those columns while its current +sequence is active. Draining `finish` or calling `relax` releases the typed +working set: quiescent memory must not grow in proportion to the number of +logical channels or destination workers. `CommunicationConfig::Process` moves +typed containers through one-way inter-thread channels and has no matching +resource-return path. ## Limitations diff --git a/timely/examples/columnar.rs b/timely/examples/columnar.rs index a3315ef08..7c0f88a5f 100644 --- a/timely/examples/columnar.rs +++ b/timely/examples/columnar.rs @@ -3,13 +3,13 @@ use std::collections::HashMap; use columnar::Index; -use timely::container::columnar::{ColumnarBuilder, ColumnarContainer}; +use timely::Accountable; use timely::container::CapacityContainerBuilder; -use timely::dataflow::channels::pact::{ExchangeCore, Pipeline}; -use timely::dataflow::operators::{InspectCore, Operator, Probe}; +use timely::container::columnar::{ColumnarBuilder, ColumnarContainer}; use timely::dataflow::InputHandle; use timely::dataflow::ProbeHandle; -use timely::Accountable; +use timely::dataflow::channels::pact::{ExchangeCore, Pipeline}; +use timely::dataflow::operators::{InspectCore, Operator, Probe}; // Creates `WordCountContainer` and `WordCountReference` structs, // as well as various implementations relating them to `WordCount`. diff --git a/timely/examples/transport_alloc.rs b/timely/examples/transport_alloc.rs index 68aa2cbd7..dd3a23f50 100644 --- a/timely/examples/transport_alloc.rs +++ b/timely/examples/transport_alloc.rs @@ -10,11 +10,11 @@ use std::sync::{Arc, Barrier}; use std::time::Instant; use serde::{Deserialize, Serialize}; +use timely::Accountable; use timely::container::columnar::{ColumnarBuilder, ColumnarContainer}; use timely::container::{CapacityContainerBuilder, ContainerBuilder, PushInto}; use timely::dataflow::operators::{Exchange, InspectCore, Probe}; use timely::dataflow::{InputHandle, ProbeHandle}; -use timely::Accountable; struct CountingAllocator; @@ -122,7 +122,9 @@ impl Transport { fn main() { let args = std::env::args().skip(1).collect::>(); if args.len() != 5 { - eprintln!("usage: transport_alloc "); + eprintln!( + "usage: transport_alloc " + ); std::process::exit(2); } diff --git a/timely/src/container/columnar.rs b/timely/src/container/columnar.rs index 37ce0156f..10c483c02 100644 --- a/timely/src/container/columnar.rs +++ b/timely/src/container/columnar.rs @@ -148,9 +148,10 @@ impl ContainerBytes for ColumnarContainer { /// Builds bounded-size [`ColumnarContainer`] batches from individual records. /// -/// Typed column allocations returned by a pusher are retained in a small pool. -/// This is particularly effective with binary communication, whose serializer -/// returns the typed container immediately after writing it into a byte slab. +/// Typed column allocations returned by a pusher are reused while a sequence is +/// active. Draining [`ContainerBuilder::finish`] or calling +/// [`ContainerBuilder::relax`] releases all typed allocations, so quiescent +/// memory does not scale with the number of builders or logical channels. pub struct ColumnarBuilder { current: C, needs_current: bool, @@ -243,7 +244,20 @@ impl ContainerBuild if !self.needs_current { self.emit_current(); } - self.extract() + self.reclaim_returned(); + if let Some(container) = self.pending.pop_front() { + self.returned = Some(container); + self.returned.as_mut() + } else { + // `finish` is called until it returns `None`. Treat that final + // call as a quiescence boundary: retaining even one preferred-size + // allocation per builder becomes prohibitive in wide dataflows. + self.current = C::default(); + self.needs_current = false; + self.returned = None; + self.spares.clear(); + None + } } fn relax(&mut self) { @@ -252,12 +266,11 @@ impl ContainerBuild "finish must drain pending columnar containers" ); assert!(self.needs_current || self.current.is_empty()); - // `relax` occurs at the end of each pushed sequence, often once per - // scheduling activation. Releasing the returned columns here would - // turn normal progress boundaries into allocation boundaries. Keep a - // bounded working set; dropping the builder still releases it. - self.reclaim_returned(); - self.ensure_current(); + // A fixed per-builder pool has an unacceptable aggregate bound for + // dataflows with many channels and workers. The zero-copy transport + // owns the process-wide byte slabs worth retaining; typed column + // scratch space is transient. + *self = Self::default(); } } @@ -300,7 +313,7 @@ mod tests { } #[test] - fn builder_reclaims_a_returned_typed_container() { + fn builder_releases_typed_containers_after_finish() { let mut builder = ColumnarBuilder::::default(); for key in 0..16 { builder.push_into(TestRecordReference { @@ -309,15 +322,36 @@ mod tests { }); } - assert!(builder.extract().is_some()); - while builder.extract().is_some() {} - let reclaimed = builder.spares.len(); - assert!(reclaimed > 0); + while builder.finish().is_some() {} - builder.push_into(TestRecordReference { - key: &99, - value: "again", - }); - assert_eq!(builder.spares.len(), reclaimed - 1); + assert!(builder.current.is_empty()); + assert!(!builder.needs_current); + assert!(builder.returned.is_none()); + assert!(builder.spares.is_empty()); + assert!(builder.pending.is_empty()); + } + + #[test] + fn wide_quiescent_builder_set_has_no_pooled_containers() { + let mut builders = std::iter::repeat_with(ColumnarBuilder::::default) + .take(10_000) + .collect::>(); + + for (key, builder) in builders.iter_mut().enumerate() { + let key = key as u64; + builder.push_into(TestRecordReference { + key: &key, + value: "value", + }); + while builder.finish().is_some() {} + } + + for builder in builders { + assert!(builder.current.is_empty()); + assert!(!builder.needs_current); + assert!(builder.returned.is_none()); + assert!(builder.spares.is_empty()); + assert!(builder.pending.is_empty()); + } } } diff --git a/timely/src/dataflow/operators/generic/handles.rs b/timely/src/dataflow/operators/generic/handles.rs index 090318d89..9205e627e 100644 --- a/timely/src/dataflow/operators/generic/handles.rs +++ b/timely/src/dataflow/operators/generic/handles.rs @@ -31,8 +31,6 @@ pub struct InputHandleCore>> { /// Staged capabilities and containers. staging: VecDeque<(InputCapability, C)>, staged: Vec, - /// Consumed containers available to swap back into subsequently pulled messages. - spares: Vec, } impl>> InputHandleCore { @@ -56,16 +54,9 @@ impl>> InputHandleCore(&mut self, mut logic: F) where F: FnMut(InputCapability, std::slice::IterMut::), C: Default { - let pull_counter = &mut self.pull_counter; - let internal = &self.internal; - let summaries = &self.summaries; - let spares = &mut self.spares; - while let Some((guard, bundle)) = pull_counter.next_guarded() { - let cap = InputCapability::new(Rc::clone(internal), Rc::clone(summaries), guard); - let data = &mut bundle.data; - let mut received = spares.pop().unwrap_or_default(); - std::mem::swap(data, &mut received); - self.staging.push_back((cap, received)); + while let Some((cap, data)) = self.next() { + let data = std::mem::take(data); + self.staging.push_back((cap, data)); } self.staging.make_contiguous().sort_unstable_by(|x,y| x.0.time().cmp(&y.0.time())); @@ -74,12 +65,7 @@ impl>> InputHandleCore>>( summaries, staging: Default::default(), staged: Default::default(), - spares: Default::default(), } } From 25abca1439295aee05ec510c019595a7a770b4db Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Sat, 15 Aug 2026 15:06:14 -0400 Subject: [PATCH 5/5] Move columnar containers into container crate --- TRANSPORT_ENGINEERING.md | 3 ++- container/Cargo.toml | 8 ++++++ .../container => container/src}/columnar.rs | 25 ++++++++++--------- container/src/lib.rs | 4 +++ mdbook/src/chapter_5/chapter_5_3.md | 5 ++-- timely/src/dataflow/channels/mod.rs | 16 ++++++++++++ timely/src/lib.rs | 3 --- 7 files changed, 46 insertions(+), 18 deletions(-) rename {timely/src/container => container/src}/columnar.rs (94%) diff --git a/TRANSPORT_ENGINEERING.md b/TRANSPORT_ENGINEERING.md index 695ab17b5..18fa5f69f 100644 --- a/TRANSPORT_ENGINEERING.md +++ b/TRANSPORT_ENGINEERING.md @@ -28,7 +28,8 @@ when its logical channel becomes quiescent. ## Revised transport tranche -- `timely::container::columnar::{ColumnarContainer, ColumnarBuilder}` promotes +- `timely_container::columnar::{ColumnarContainer, ColumnarBuilder}` (also + re-exported from `timely::container`) promotes the former example-only implementation into supported infrastructure. Binary receivers retain a view into compact communication `Bytes` rather than reconstructing owned rows. diff --git a/container/Cargo.toml b/container/Cargo.toml index feb8330f0..c93dd5811 100644 --- a/container/Cargo.toml +++ b/container/Cargo.toml @@ -8,3 +8,11 @@ rust-version.workspace = true [lints] workspace = true + +[features] +default = ["columnar"] +columnar = ["dep:columnar", "dep:timely_bytes"] + +[dependencies] +columnar = { workspace = true, optional = true } +timely_bytes = { path = "../bytes", version = "0.31", optional = true } diff --git a/timely/src/container/columnar.rs b/container/src/columnar.rs similarity index 94% rename from timely/src/container/columnar.rs rename to container/src/columnar.rs index 10c483c02..0c1cc46b3 100644 --- a/timely/src/container/columnar.rs +++ b/container/src/columnar.rs @@ -11,12 +11,12 @@ use std::collections::VecDeque; use ::columnar::bytes::stash::Stash; use ::columnar::{Index, Len}; -use crate::bytes::arc::Bytes; -use crate::container::{ +use timely_bytes::arc::Bytes; + +use crate::{ Accountable, ContainerBuilder, DrainContainer, LengthPreservingContainerBuilder, PushInto, SizableContainer, }; -use crate::dataflow::channels::ContainerBytes; /// Preferred serialized size of a columnar transport container. pub const DEFAULT_BUFFER_BYTES: usize = 1 << 20; @@ -128,21 +128,22 @@ where } } -impl ContainerBytes for ColumnarContainer { - fn from_bytes(bytes: Bytes) -> Self { +impl ColumnarContainer { + /// Wraps and validates bytes containing a columnar encoding. + pub fn from_bytes(bytes: Bytes) -> Self { Self { stash: Stash::try_from_bytes(bytes).expect("valid columnar container bytes"), } } - fn length_in_bytes(&self) -> usize { + /// Reports the number of bytes in this container's wire encoding. + pub fn length_in_bytes(&self) -> usize { self.stash.length_in_bytes() } - fn into_bytes(&self, writer: &mut W) { - self.stash - .write_bytes(writer) - .expect("columnar container write failed") + /// Writes this container's columnar encoding. + pub fn write_bytes(&self, writer: &mut W) -> std::io::Result<()> { + self.stash.write_bytes(writer) } } @@ -300,9 +301,9 @@ mod tests { }); let mut encoded = Vec::new(); - ContainerBytes::into_bytes(&original, &mut encoded); + original.write_bytes(&mut encoded).unwrap(); let received = ColumnarContainer::::from_bytes( - crate::bytes::arc::BytesMut::from(encoded).freeze(), + timely_bytes::arc::BytesMut::from(encoded).freeze(), ); assert!(received.is_bytes()); diff --git a/container/src/lib.rs b/container/src/lib.rs index 400512475..5d2761e85 100644 --- a/container/src/lib.rs +++ b/container/src/lib.rs @@ -4,6 +4,10 @@ use std::collections::VecDeque; +/// Allocation-conscious columnar containers and builders. +#[cfg(feature = "columnar")] +pub mod columnar; + /// A type containing a number of records accounted for by progress tracking. /// /// The object stores a number of updates and thus is able to describe it count diff --git a/mdbook/src/chapter_5/chapter_5_3.md b/mdbook/src/chapter_5/chapter_5_3.md index f57ef5d46..6f4afb1ad 100644 --- a/mdbook/src/chapter_5/chapter_5_3.md +++ b/mdbook/src/chapter_5/chapter_5_3.md @@ -25,8 +25,9 @@ In most cases, the `core` operators are an immediate generalization of their non +## Columnar transport -The `timely::container::columnar` module provides `ColumnarContainer` and -`ColumnarBuilder` for records deriving `columnar::Columnar`. The container can +The default-enabled `columnar` feature of `timely_container` provides +`ColumnarContainer` and `ColumnarBuilder`, re-exported through +`timely::container::columnar`, for records deriving `columnar::Columnar`. The container can hold mutable typed columns while it is being assembled and retain a borrowed view over communication bytes after binary transport. This avoids rebuilding owned rows at the receiver. diff --git a/timely/src/dataflow/channels/mod.rs b/timely/src/dataflow/channels/mod.rs index ce109363c..4ed189dd8 100644 --- a/timely/src/dataflow/channels/mod.rs +++ b/timely/src/dataflow/channels/mod.rs @@ -117,6 +117,22 @@ mod implementations { use serde::{Serialize, Deserialize}; use crate::dataflow::channels::ContainerBytes; + impl ContainerBytes + for crate::container::columnar::ColumnarContainer + { + fn from_bytes(bytes: crate::bytes::arc::Bytes) -> Self { + Self::from_bytes(bytes) + } + + fn length_in_bytes(&self) -> usize { + self.length_in_bytes() + } + + fn into_bytes(&self, writer: &mut W) { + self.write_bytes(writer).expect("columnar container write failed") + } + } + impl Deserialize<'a>> ContainerBytes for Vec { fn from_bytes(bytes: crate::bytes::arc::Bytes) -> Self { ::bincode::deserialize(&bytes[..]).expect("bincode::deserialize() failed") diff --git a/timely/src/lib.rs b/timely/src/lib.rs index 63deb47b0..39625f7e3 100644 --- a/timely/src/lib.rs +++ b/timely/src/lib.rs @@ -69,9 +69,6 @@ pub use timely_container::Accountable; /// Re-export of the `timely_container` crate. pub mod container { pub use timely_container::*; - - /// Columnar containers that can retain serialized communication bytes. - pub mod columnar; } /// Re-export of the `timely_communication` crate.