diff --git a/TRANSPORT_ENGINEERING.md b/TRANSPORT_ENGINEERING.md new file mode 100644 index 000000000..18fa5f69f --- /dev/null +++ b/TRANSPORT_ENGINEERING.md @@ -0,0 +1,194 @@ +# 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. + +## 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}` (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. +- 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, 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 + +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. +- 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. + 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/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/container/src/columnar.rs b/container/src/columnar.rs new file mode 100644 index 000000000..0c1cc46b3 --- /dev/null +++ b/container/src/columnar.rs @@ -0,0 +1,358 @@ +//! 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 timely_bytes::arc::Bytes; + +use crate::{ + Accountable, ContainerBuilder, DrainContainer, LengthPreservingContainerBuilder, PushInto, + SizableContainer, +}; + +/// 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 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"), + } + } + + /// Reports the number of bytes in this container's wire encoding. + pub fn length_in_bytes(&self) -> usize { + self.stash.length_in_bytes() + } + + /// Writes this container's columnar encoding. + pub fn write_bytes(&self, writer: &mut W) -> std::io::Result<()> { + self.stash.write_bytes(writer) + } +} + +/// Builds bounded-size [`ColumnarContainer`] batches from individual records. +/// +/// 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, + 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.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) { + assert!( + self.pending.is_empty(), + "finish must drain pending columnar containers" + ); + assert!(self.needs_current || self.current.is_empty()); + // 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(); + } +} + +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(); + original.write_bytes(&mut encoded).unwrap(); + let received = ColumnarContainer::::from_bytes( + timely_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_releases_typed_containers_after_finish() { + let mut builder = ColumnarBuilder::::default(); + for key in 0..16 { + builder.push_into(TestRecordReference { + key: &key, + value: "value", + }); + } + + while builder.finish().is_some() {} + + 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/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/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..6f4afb1ad 100644 --- a/mdbook/src/chapter_5/chapter_5_3.md +++ b/mdbook/src/chapter_5/chapter_5_3.md @@ -23,6 +23,56 @@ 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 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. + +```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. 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 1a87ab164..7c0f88a5f 100644 --- a/timely/examples/columnar.rs +++ b/timely/examples/columnar.rs @@ -5,10 +5,11 @@ use std::collections::HashMap; use columnar::Index; use timely::Accountable; use timely::container::CapacityContainerBuilder; -use timely::dataflow::channels::pact::{ExchangeCore, Pipeline}; +use timely::container::columnar::{ColumnarBuilder, ColumnarContainer}; use timely::dataflow::InputHandle; -use timely::dataflow::operators::{InspectCore, Operator, Probe}; use timely::dataflow::ProbeHandle; +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`. @@ -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..dd3a23f50 --- /dev/null +++ b/timely/examples/transport_alloc.rs @@ -0,0 +1,405 @@ +//! 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::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}; + +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/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")