diff --git a/differential-dataflow/src/operators/arrange/agent.rs b/differential-dataflow/src/operators/arrange/agent.rs index a310be8bb..f383b04e3 100644 --- a/differential-dataflow/src/operators/arrange/agent.rs +++ b/differential-dataflow/src/operators/arrange/agent.rs @@ -568,6 +568,15 @@ pub mod trace_box { /// /// This is used to inspect batches for purposes of resource accounting in external systems. pub fn trace(&self) -> &Tr { &self.trace } + /// The accumulated logical-compaction holds of all referees of the trace. + /// + /// This is the `MutableAntichain` whose frontier the trace compacts to. Exposed so an + /// external referee (for example a cross-runtime publisher) can compute the frontier the + /// trace would compact to without its own hold, by cloning this and subtracting that hold. + pub fn logical_compaction(&self) -> &MutableAntichain { &self.logical_compaction } + /// The accumulated physical-compaction holds of all referees of the trace. See + /// [`TraceBox::logical_compaction`]. + pub fn physical_compaction(&self) -> &MutableAntichain { &self.physical_compaction } /// Replaces elements of `lower` with those of `upper`. #[inline] pub fn adjust_logical_compaction(&mut self, lower: AntichainRef, upper: AntichainRef) { diff --git a/differential-dataflow/src/operators/arrange/mod.rs b/differential-dataflow/src/operators/arrange/mod.rs index fedbfdc23..2dbda1960 100644 --- a/differential-dataflow/src/operators/arrange/mod.rs +++ b/differential-dataflow/src/operators/arrange/mod.rs @@ -63,6 +63,7 @@ type TraceAgentQueueWriter = Weak<(Activator, RefCell>)>; pub mod writer; pub mod agent; pub mod arrangement; +pub mod sharing; pub mod upsert; diff --git a/differential-dataflow/src/operators/arrange/sharing.rs b/differential-dataflow/src/operators/arrange/sharing.rs new file mode 100644 index 000000000..727e06135 --- /dev/null +++ b/differential-dataflow/src/operators/arrange/sharing.rs @@ -0,0 +1,825 @@ +//! Sharing arrangements across timely runtimes. +//! +//! An arrangement is normally readable only from the worker that maintains it: its batches are +//! reference counted with `Rc` and its trace handle is `Rc>`, both pinned to one +//! thread. This module lets a worker publish an arrangement whose batches are `Arc`'d (and whose +//! contents are `Send + Sync`) through a *publication point*, from which readers on other threads +//! take consistent snapshots or import the arrangement into a second timely runtime. +//! +//! The unit that crosses the thread boundary is not the [`Spine`](super::super::super::trace::implementations::spine_fueled::Spine), +//! which holds thread-local state and has a single writer, but the spine's *contents*: a chain of +//! immutable `Arc`'d batches together with the trace's `since` and `upper` frontiers. Because +//! batches are immutable, a chain plus frontiers is a self-describing, consistent view. When the +//! publishing worker later merges batches, a reader holding an older chain is unaffected: its `Arc`s +//! keep the pre-merge batches alive until it drops them. +//! +//! ## Pieces +//! +//! * [`Arranged::publish`] attaches a publisher to an arrangement on the owning worker and returns a +//! [`Published`] whose [`Published::handle`] hands out `Clone + Send` [`SharedTraceHandle`]s. +//! * [`SharedTraceHandle`] implements [`TraceReader`], so it drives compaction and cursors like any +//! trace handle. [`SharedTraceHandle::snapshot_at`] serves a point or full-scan read from any +//! thread. [`SharedTraceHandle::import`] replays the shared arrangement into another scope. +//! +//! ## Compaction +//! +//! Every reader registers logical and physical holds. The publisher forwards their *meet* (the +//! greatest lower bound, so the trace never compacts past the least reader's hold) to its own +//! `TraceAgent`, which is the sole writer of the trace's compaction frontiers. Publishing itself +//! carries no compaction floor: with no readers the publisher advances its hold to the +//! writer-driven frontier (the meet of the other agents' holds), so the trace compacts and merges +//! as the writer advances. The publisher never forwards the empty frontier, which would +//! irreversibly release the trace. + +use std::collections::{BTreeMap, VecDeque}; +use std::sync::{Arc, Condvar, Mutex}; + +use timely::dataflow::operators::generic::{source, Operator}; +use timely::dataflow::operators::CapabilitySet; +use timely::dataflow::Scope; +use timely::progress::frontier::{AntichainRef, MutableAntichain}; +use timely::progress::Antichain; +use timely::scheduling::activate::SyncActivator; + +use crate::lattice::{antichain_meet, Lattice}; +use crate::trace::cursor::{cursor_list, CursorList, Navigable}; +use crate::trace::wrappers::frontier::{BatchFrontier, TraceFrontier}; +use crate::trace::{BatchReader, TraceReader}; + +use super::{Arranged, TraceAgent, TraceReplayInstruction}; + +/// The queue and wakeup for one importer registered against a publication point. +struct ImportQueue { + /// Replay instructions the publisher appends and the importer drains, mirroring the local + /// arrange replay queue. Batches carry a hint time that lower-bounds their updates. + instructions: VecDeque>, + /// Wakes the importer's source operator on the reader's worker when new instructions arrive. + activator: SyncActivator, +} + +/// State shared between a publisher (on the owning worker) and its readers (on any thread). +/// +/// The `chain`, `since`, and `upper` are always updated together under the lock, so every reader +/// observes a frontier-consistent view. +struct SharedTraceState { + /// The published chain, sourced from `map_batches`: contiguous descriptions including the + /// seal-only empty batches that never travel on the arrangement stream. Coverage is at least + /// `upper` (within a worker step it may briefly run ahead, never behind). + chain: Vec, + /// Logical compaction frontier of the published view. Reads at times not beyond `since` are not + /// accurate. A snapshot must pick a time at or beyond it. + since: Antichain, + /// Seal frontier: the join of the chain's batch uppers. Batches strictly below `upper` are + /// complete and readable. + upper: Antichain, + /// Per-registration logical holds. The publisher forwards their meet, falling back to the + /// writer-driven frontier when empty (never the destructive empty meet of zero holds). + logical_holds: BTreeMap>, + /// Per-registration physical holds, forwarded independently of the logical holds. + physical_holds: BTreeMap>, + /// Importer queues, keyed by registration id. A handle may back several registrations, so this + /// is keyed separately from any handle. + queues: BTreeMap>, + /// Monotonic source of registration ids for holds and queues. + next_id: usize, + /// Set when the publisher drops. A terminal empty frontier is enqueued to each importer, so + /// readers close only after draining what was already published. + closed: bool, +} + +impl SharedTraceState +where + Tr::Time: Lattice + Clone, +{ + /// The meet of `holds`, or `fallback` when there are none. + /// + /// Never returns the empty meet of zero holds, which is the empty frontier and would tell the + /// trace to compact everything, permanently releasing the publisher's capability. + fn compaction_target( + holds: &BTreeMap>, + fallback: &Antichain, + ) -> Antichain { + let mut iter = holds.values(); + match iter.next() { + None => fallback.clone(), + Some(first) => { + let mut acc = first.clone(); + for hold in iter { + acc = antichain_meet(&acc.borrow()[..], &hold.borrow()[..]); + } + acc + } + } + } +} + +/// A publication point paired with its live [`Condvar`], so peek waiters can block for `upper` to +/// advance without a lost-wakeup race: the publisher signals under the same lock it uses to move +/// `upper`. +struct SharedTrace { + state: Mutex>, + upper_changed: Condvar, + /// Total peer count (workers-per-process times processes) of the scope that published this + /// arrangement. Set once at publish time and never mutated afterward, so `import` reads it + /// without taking `state`'s lock. Pairwise import (importer worker `i` reads publisher worker + /// `i`) is sound only when an importing scope shards keys the same way, which requires this to + /// match the importing scope's own `peers()`. + peers: usize, +} + +type SharedTraceRef = Arc>; + +/// The result of publishing an arrangement. Holding it keeps the publication point registered; +/// dropping it does not stop the publisher (the publisher lives with its dataflow), but no further +/// handles can be minted from it. +pub struct Published { + shared: SharedTraceRef, +} + +impl Published +where + Tr::Time: Lattice + Clone, +{ + /// Hands out a `Clone + Send` handle to the published arrangement. + /// + /// The handle registers a logical hold at the current published `since`, so the arrangement + /// will not compact past it until the handle (and all its clones) drop. + pub fn handle(&self) -> SharedTraceHandle { + SharedTraceHandle::register(Arc::clone(&self.shared)) + } +} + +/// A `Clone + Send` reader of a published arrangement. +/// +/// Implements [`TraceReader`], so downstream operators drive its compaction and acquire cursors as +/// with any trace handle. Each clone carries an independent registration: cloning mints a fresh id +/// and copies the source's holds, so two consumers of one import cannot release each other's holds. +pub struct SharedTraceHandle { + shared: SharedTraceRef, + /// This handle's hold registration id. + id: usize, + /// This handle's own logical frontier, mirrored into `logical_holds[id]`. Kept locally so + /// `get_logical_compaction` can return a borrow. + logical: Antichain, + /// This handle's own physical frontier, mirrored into `physical_holds[id]`. + physical: Antichain, +} + +impl SharedTraceHandle +where + Tr::Time: Lattice + Clone, +{ + /// Registers a fresh hold at the current published `since` and returns a handle for it. + fn register(shared: SharedTraceRef) -> Self { + let (id, since) = { + let mut state = shared.state.lock().expect("shared trace poisoned"); + let id = state.next_id; + state.next_id += 1; + let since = state.since.clone(); + state.logical_holds.insert(id, since.clone()); + state.physical_holds.insert(id, since.clone()); + (id, since) + }; + Self { + shared, + id, + logical: since.clone(), + physical: since, + } + } + + /// Takes a consistent snapshot of the published arrangement as of `time`, blocking until `upper` + /// passes `time`. + /// + /// Works from any thread. Returns `None` when the snapshot cannot serve `time`, which is either + /// the publisher closed before `upper` passed `time`, or compaction has advanced `since` beyond + /// `time` so the accumulation at `time` is no longer accurate. The gate on `since` mirrors the + /// single-runtime peek path, which errors when the compaction frontier is beyond the read time + /// rather than returning coalesced results. A caller that needs to tell the two `None` cases + /// apart should inspect `since` before the read. + pub fn snapshot_at(&self, time: &Tr::Time) -> Option> { + let mut state = self.shared.state.lock().expect("shared trace poisoned"); + // `upper` not less-equal `time` means all updates at `time` are sealed. + while state.upper.less_equal(time) { + if state.closed { + return None; + } + state = self + .shared + .upper_changed + .wait(state) + .expect("shared trace poisoned"); + } + // `since` beyond `time` means times at `time` have been coalesced and a read there would be + // inaccurate. Fail to `None` rather than serve stale data. + if !state.since.less_equal(time) { + return None; + } + Some(TraceSnapshot { + chain: state.chain.clone(), + since: state.since.clone(), + upper: state.upper.clone(), + }) + } + + /// Writes this handle's logical or physical hold into the shared registry. + fn update_hold(&self, logical: bool) { + let mut state = self.shared.state.lock().expect("shared trace poisoned"); + if logical { + state.logical_holds.insert(self.id, self.logical.clone()); + } else { + state.physical_holds.insert(self.id, self.physical.clone()); + } + } +} + +impl Clone for SharedTraceHandle +where + Tr::Time: Lattice + Clone, +{ + fn clone(&self) -> Self { + // A clone must be an independent hold: `import` returns `Arranged { trace: handle.clone() }` + // and `Arranged` is itself `Clone`, so distinct downstream operators drive compaction on + // distinct clones. Sharing one hold slot would let the faster operator release the slower + // one's hold. This mirrors `TraceAgent::clone`, which registers an independent counted hold. + let id = { + let mut state = self.shared.state.lock().expect("shared trace poisoned"); + let id = state.next_id; + state.next_id += 1; + state.logical_holds.insert(id, self.logical.clone()); + state.physical_holds.insert(id, self.physical.clone()); + id + }; + Self { + shared: Arc::clone(&self.shared), + id, + logical: self.logical.clone(), + physical: self.physical.clone(), + } + } +} + +impl Drop for SharedTraceHandle { + fn drop(&mut self) { + if let Ok(mut state) = self.shared.state.lock() { + state.logical_holds.remove(&self.id); + state.physical_holds.remove(&self.id); + } + } +} + +impl TraceReader for SharedTraceHandle +where + Tr::Time: Lattice + Clone, +{ + type Time = Tr::Time; + type Batch = Tr::Batch; + + fn batches_through(&mut self, upper: AntichainRef) -> Option> { + let state = self.shared.state.lock().expect("shared trace poisoned"); + // A clean cut of the published chain: all non-empty batches whose upper is not beyond + // `upper`, and none whose lower is beyond `upper`. Empty batches are dropped, as + // `Spine::batches_through` does. + let mut out = Vec::new(); + for batch in state.chain.iter() { + // A batch whose lower is beyond the cut, and everything after it in the totally + // ordered chain, lies past `upper`. Empty batches never carry updates to read. + if timely::PartialOrder::less_equal(&upper, &batch.lower().borrow()) { + break; + } + if !batch.is_empty() { + // Fail-stop on a batch that straddles the cut (`lower < upper < batch.upper()`), + // matching `Spine::batches_through`. Returning it would hand back updates at times + // not before `upper`, corrupting a downstream `cursor_through` consumer such as + // `join`. The published chain is totally ordered by description, so this cut is + // clean unless a caller requested a frontier that is not batch-aligned. + assert!( + timely::PartialOrder::less_equal(&batch.upper().borrow(), &upper), + "batches_through: upper straddles batch" + ); + out.push(batch.clone()); + } + } + Some(out) + } + + fn set_logical_compaction(&mut self, frontier: AntichainRef) { + self.logical = frontier.to_owned(); + self.update_hold(true); + } + + fn get_logical_compaction(&mut self) -> AntichainRef<'_, Tr::Time> { + self.logical.borrow() + } + + fn set_physical_compaction(&mut self, frontier: AntichainRef<'_, Tr::Time>) { + self.physical = frontier.to_owned(); + self.update_hold(false); + } + + fn get_physical_compaction(&mut self) -> AntichainRef<'_, Tr::Time> { + self.physical.borrow() + } + + fn map_batches(&self, mut f: F) { + let state = self.shared.state.lock().expect("shared trace poisoned"); + for batch in state.chain.iter() { + f(batch); + } + } +} + +/// Smallest time, used only to satisfy the borrow in the cut check for empty lower frontiers. +fn batch_min() -> Tr::Time { + ::minimum() +} + +/// An owned, consistent snapshot of a published arrangement: an immutable chain plus its frontiers. +/// +/// Holding it pins the chain's batches, keeping their memory alive even as the publishing worker +/// merges. Dropping it releases them, from whatever thread drops last. +pub struct TraceSnapshot { + chain: Vec, + since: Antichain, + upper: Antichain, +} + +impl TraceSnapshot { + /// The logical compaction frontier of the snapshot. Times not beyond it are not accurate. + pub fn since(&self) -> AntichainRef<'_, Tr::Time> { + self.since.borrow() + } + /// The seal frontier of the snapshot. Times strictly below it are complete. + pub fn upper(&self) -> AntichainRef<'_, Tr::Time> { + self.upper.borrow() + } + /// A cursor merging the snapshot's batch cursors, with the batches as its storage. + pub fn cursor(&self) -> (CursorList<::Cursor>, Vec) + where + Tr::Batch: Navigable, + { + cursor_list(self.chain.clone()) + } +} + +impl<'scope, Tr> Arranged<'scope, TraceAgent> +where + Tr: crate::trace::Trace + 'static, + Tr::Batch: Send + Sync, + Tr::Time: Lattice + Clone + Send + Sync, +{ + /// Publishes this arrangement through a publication point on the owning worker. + /// + /// Attaches a publisher operator to the arrangement stream. On each activation the publisher + /// refreshes the published chain, `since`, and `upper` from the trace, appends newly arrived + /// batches to importer queues, and forwards the meet of reader holds to the trace's compaction. + /// The returned [`Published`] mints [`SharedTraceHandle`]s that read the arrangement from any + /// thread. + pub fn publish(&self) -> Published { + self.publish_named("PublishShared") + } + + /// [`Arranged::publish`], with a name for the publisher operator. + pub fn publish_named(&self, name: &str) -> Published { + // The publisher owns a `TraceAgent` clone: its read capability is the aggregate lease for + // all readers, so the trace cannot compact or drop out from under them. + let mut agent = self.trace.clone(); + + let initial_since = agent.get_logical_compaction().to_owned(); + let shared: SharedTraceRef = Arc::new(SharedTrace { + state: Mutex::new(SharedTraceState { + chain: Vec::new(), + since: initial_since, + // NOTE: `Antichain::from_elem(minimum)`, never `Antichain::new()`. The empty + // frontier reads as "complete through the end of time", making every snapshot wait + // vacuously true and returning empty results instead of blocking. + upper: Antichain::from_elem(batch_min::()), + logical_holds: BTreeMap::new(), + physical_holds: BTreeMap::new(), + queues: BTreeMap::new(), + next_id: 0, + closed: false, + }), + upper_changed: Condvar::new(), + peers: self.stream.scope().peers(), + }); + + let publisher = Publisher { + shared: Arc::clone(&shared), + }; + + let sink_shared = Arc::clone(&shared); + self.stream.clone().sink( + timely::dataflow::channels::pact::Pipeline, + name, + move |(input, _frontier)| { + // Keep `publisher` alive with the operator, so operator (dataflow) drop closes the + // publication point. + let _publisher = &publisher; + + // Batches arriving on the stream, each with a capability time that lower-bounds the + // batch's updates. Empty seal batches do not travel the stream; they are picked up + // from the trace below. + let mut arrived: Vec<(Tr::Batch, Tr::Time)> = Vec::new(); + input.for_each(|cap, data| { + let hint = cap.time().clone(); + for batch in data.drain(..) { + arrived.push((batch, hint.clone())); + } + }); + + // Refresh the chain and frontiers from the trace, the authoritative source. + let mut chain = Vec::new(); + let mut upper = Antichain::new(); + agent.map_batches(|batch| { + chain.push(batch.clone()); + upper = batch.upper().to_owned(); + }); + if chain.is_empty() { + upper = Antichain::from_elem(batch_min::()); + } + // Contract: publishing carries no independent compaction floor. The trace's writer + // (and, in Materialize, the controller) drive `since` through their own trace + // handles. Only a live importer's registered hold may hold the trace back, and it + // releases on drop. The publisher keeps a holding agent solely so importer holds + // have somewhere to forward to, so that hold must FOLLOW the writer rather than pin + // the trace. + // + // The writer-driven frontier is the meet of all other agents' holds, i.e. the + // frontier the trace would compact to if the publisher were not holding it. The + // `TraceBox` accumulates every agent's hold in a `MutableAntichain`; cloning it and + // subtracting the publisher's own contribution yields that meet. Advancing the + // publisher's hold to it lets the trace compact and merge as the writer advances. + // + // NOTE: the per-batch `since` from `map_batches` cannot serve as this source. Those + // frontiers advance only when the Spine actually compacts, which the publisher's own + // pinning hold prevents, so they stay frozen at the publish-time `since`. Reading + // the trace-box meet breaks that circularity. + let publisher_logical = agent.get_logical_compaction().to_owned(); + let publisher_physical = agent.get_physical_compaction().to_owned(); + // The writer-driven frontier for one dimension: the meet of the accumulated holds + // with the publisher's own contribution removed. An empty result means the + // publisher is the sole holder. Never forward the empty frontier: it would compact + // everything and release the publisher's capability, so fall back to the publisher's + // current hold and keep the current floor. + let others_meet = + |accumulated: &MutableAntichain, own: &Antichain| { + let mut others = accumulated.clone(); + others.update_iter(own.iter().map(|t| (t.clone(), -1))); + if others.frontier().is_empty() { + own.clone() + } else { + others.frontier().to_owned() + } + }; + let (writer_logical, writer_physical) = { + let trace_box = agent.trace_box_unstable(); + let trace_box = trace_box.borrow(); + ( + others_meet(&trace_box.logical_compaction, &publisher_logical), + others_meet(&trace_box.physical_compaction, &publisher_physical), + ) + }; + + let (logical_target, physical_target) = { + let mut state = sink_shared.state.lock().expect("shared trace poisoned"); + + for (batch, hint) in arrived.drain(..) { + for queue in state.queues.values_mut() { + queue.instructions.push_back(TraceReplayInstruction::Batch( + batch.clone(), + Some(hint.clone()), + )); + } + } + + let upper_advanced = state.upper != upper; + if upper_advanced { + for queue in state.queues.values_mut() { + queue + .instructions + .push_back(TraceReplayInstruction::Frontier(upper.clone())); + } + } + + // Fall back to the writer-driven frontier (not the publisher's frozen hold) when + // there are no reader holds, so with zero readers the target follows the writer. + let logical = SharedTraceState::::compaction_target( + &state.logical_holds, + &writer_logical, + ); + let physical = SharedTraceState::::compaction_target( + &state.physical_holds, + &writer_physical, + ); + + state.chain = chain; + // Publish the trace's real logical compaction after we forward `logical` below. + // Agent compaction only advances (joins), so the publisher's hold becomes + // `join(publisher_logical, logical)`. The trace's real compaction is the meet of + // every agent's hold: the publisher's post-forward hold and the writer-driven + // meet of the others. Publishing exactly that keeps the gate in step with the + // trace, so a reader hold keeps its own frontier readable rather than being + // raced past by the writer. It is never below the real compaction (it equals + // it), so a handle registering in this window cannot latch an anti-conservative + // `since` that claims accuracy at already-merged times. + let publisher_after = publisher_logical.join(&logical); + state.since = + antichain_meet(&publisher_after.borrow()[..], &writer_logical.borrow()[..]); + state.upper = upper; + + // Wake importers and any peek waiters. + for queue in state.queues.values() { + let _ = queue.activator.activate(); + } + if upper_advanced { + sink_shared.upper_changed.notify_all(); + } + + (logical, physical) + }; + + // Apply compaction to the agent OUTSIDE the lock: `set_physical_compaction` can run + // an unbounded merge synchronously, which must not block concurrent readers. + agent.set_logical_compaction(logical_target.borrow()); + agent.set_physical_compaction(physical_target.borrow()); + }, + ); + + Published { shared } + } +} + +/// Guard that marks the publication point closed when the publisher operator drops, waking readers +/// so they drain and shut down. +struct Publisher { + shared: SharedTraceRef, +} + +impl Drop for Publisher { + fn drop(&mut self) { + if let Ok(mut state) = self.shared.state.lock() { + state.closed = true; + let empty = Antichain::new(); + for queue in state.queues.values_mut() { + queue + .instructions + .push_back(TraceReplayInstruction::Frontier(empty.clone())); + let _ = queue.activator.activate(); + } + } + self.shared.upper_changed.notify_all(); + } +} + +impl SharedTraceHandle +where + Tr: 'static, + Tr::Time: Lattice + Clone, + Tr::Batch: Navigable, +{ + /// Imports the published arrangement into `scope`, returning an [`Arranged`] backed by this + /// shared trace. + /// + /// Requires `scope`'s total peer count (workers-per-process times processes) to equal the + /// publisher's, panicking otherwise. Pairwise import (importer worker `i` reads publisher + /// worker `i`) is sound only when both sides shard by the same `key.hashed() % peers`; a + /// mismatched peer count would silently read the wrong shard instead of failing loudly. + /// + /// The importer registers a replay queue seeded with the current chain and drains it as the + /// publisher appends. The registration is owned by the source operator, so dropping the import + /// dataflow deregisters it and releases its holds even while other handle clones and the reader + /// worker live on. + pub fn import<'scope>( + &self, + scope: Scope<'scope, Tr::Time>, + name: &str, + ) -> Arranged<'scope, SharedTraceHandle> { + assert_eq!( + scope.peers(), + self.shared.peers, + "shared-trace import requires equal total peers (workers_per_process * num_processes)" + ); + + let shared = Arc::clone(&self.shared); + let trace = self.clone(); + + let stream = source(scope, name, move |capability, info| { + let activator = scope.worker().sync_activator_for(info.address.to_vec()); + + // Register under one lock acquisition: mint an id, seed the queue with the current + // chain (hint `minimum`, as the local replay does for historical batches) followed by + // the current upper, and install the queue. Later batches append; earlier ones are + // seeded. Nothing is missed or duplicated. + let reg_id = { + let mut state = shared.state.lock().expect("shared trace poisoned"); + let reg_id = state.next_id; + state.next_id += 1; + let mut instructions = VecDeque::new(); + for batch in state.chain.iter() { + instructions.push_back(TraceReplayInstruction::Batch( + batch.clone(), + Some(batch_min::()), + )); + } + instructions.push_back(TraceReplayInstruction::Frontier(state.upper.clone())); + // If the publisher already closed, its one-shot terminal frontier has been and gone, + // so seed our own. Otherwise a late importer would drain the chain and then wait + // forever for a frontier that never arrives, leaking its capability. Mirrors the + // `state.closed` guard in `snapshot_at`. + if state.closed { + instructions.push_back(TraceReplayInstruction::Frontier(Antichain::new())); + } + state.queues.insert( + reg_id, + ImportQueue { + instructions, + activator, + }, + ); + reg_id + }; + + let mut capabilities = Some(CapabilitySet::new()); + capabilities.as_mut().unwrap().insert(capability); + + // Deregisters the queue when the source operator (and thus this closure) drops. + let _guard = QueueGuard { + shared: Arc::clone(&shared), + reg_id, + }; + + move |output| { + let _guard = &_guard; + let mut drained = Vec::new(); + { + let mut state = shared.state.lock().expect("shared trace poisoned"); + if let Some(queue) = state.queues.get_mut(®_id) { + drained.extend(queue.instructions.drain(..)); + } + } + + if let Some(caps) = capabilities.as_mut() { + for instruction in drained { + match instruction { + TraceReplayInstruction::Frontier(frontier) => { + if frontier.is_empty() { + // Terminal frontier: publisher closed, no more capabilities. + capabilities = None; + break; + } + caps.downgrade(&frontier.borrow()[..]); + } + TraceReplayInstruction::Batch(batch, hint) => { + if let Some(time) = hint { + if !batch.is_empty() { + // Emit under a capability delayed to the batch's hint, a + // lower bound on its times. Never the batch upper: that + // would let the frontier pass updates still in the batch. + let cap = caps.delayed(&time); + output.session(&cap).give(batch); + } + } + } + } + } + } + } + }); + + Arranged { stream, trace } + } + + /// Imports a static snapshot of the published arrangement at `as_of`, for single-time reads. + /// + /// Unlike [`Self::import`], which replays the live change stream, this emits the published chain + /// exactly once (when the arrangement has sealed through `as_of`) and then releases its + /// capability, so the imported collection is complete for a single logical time. Times are + /// presented through a [`TraceFrontier`]/[`BatchFrontier`] wrapper advanced to `as_of` and + /// bounded by `until`, so a downstream reader sees the accumulation at `as_of` with no + /// pre-`as_of` retractions. This is what a one-shot peek or its query dataflow needs, and it + /// sidesteps the live import's forward-batch capability handling entirely: there is no forward + /// stream. + /// + /// The wrapper advances times on read, so the shared `Arc` batches are reused as-is, never + /// re-arranged. + /// + /// The snapshot is emitted only once `upper` is strictly beyond `as_of` (the same gate as + /// [`Self::snapshot_at`]), so the accumulation at `as_of` is final. Until then the operator holds + /// its capability at the minimum time and re-checks on each publisher activation, so the output + /// frontier does not advance and a downstream result read waits rather than observing a partial + /// (often empty) snapshot. The caller's read hold must keep `since` at or below `as_of`, matching + /// [`Self::snapshot_at`]; a `since` past `as_of` would make the read inaccurate. + /// + /// Requires `scope`'s total peer count to equal the publisher's, as [`Self::import`] does. + pub fn import_snapshot_at<'scope>( + &self, + scope: Scope<'scope, Tr::Time>, + name: &str, + as_of: Antichain, + until: Antichain, + ) -> Arranged<'scope, TraceFrontier>> { + assert_eq!( + scope.peers(), + self.shared.peers, + "shared-trace import requires equal total peers (workers_per_process * num_processes)" + ); + + let trace = TraceFrontier::make_from(self.clone(), as_of.borrow(), until.borrow()); + let shared = Arc::clone(&self.shared); + + let stream = source(scope, name, move |capability, info| { + let activator = scope.worker().sync_activator_for(info.address.to_vec()); + + // Register an (instruction-less) queue purely to receive the publisher's activations on + // each `upper` advance and on close, so we re-check the seal condition without polling. + let reg_id = { + let mut state = shared.state.lock().expect("shared trace poisoned"); + let reg_id = state.next_id; + state.next_id += 1; + state.queues.insert( + reg_id, + ImportQueue { + instructions: VecDeque::new(), + activator, + }, + ); + reg_id + }; + + // Deregisters the queue when the source operator (and thus this closure) drops. + let _guard = QueueGuard { + shared: Arc::clone(&shared), + reg_id, + }; + + let mut capabilities = Some(CapabilitySet::new()); + capabilities.as_mut().unwrap().insert(capability); + + move |output| { + let _guard = &_guard; + let Some(caps) = capabilities.as_mut() else { + // The snapshot was already emitted and the capability released. + return; + }; + + let (chain, upper, closed) = { + let mut state = shared.state.lock().expect("shared trace poisoned"); + // Drain any queued instructions to reset the coalescing activator flag; this + // path reads `chain`/`upper` directly rather than replaying instructions. + if let Some(queue) = state.queues.get_mut(®_id) { + queue.instructions.clear(); + } + (state.chain.clone(), state.upper.clone(), state.closed) + }; + + // Emit once `upper` is strictly beyond every `as_of` time, mirroring the wait in + // `snapshot_at`: only then is the accumulation at `as_of` final (all updates at times + // not beyond `as_of` are sealed, so advancing them to `as_of` yields a fixed answer). + // Keying on `as_of` rather than `until` is deliberate: `until` may be the empty + // (unbounded) frontier for a long-lived index, which a live `upper` never reaches. A + // premature emit at `upper == as_of` would ship an incomplete (often empty) snapshot + // and then close, so the read must wait for `upper` to pass `as_of`. If the publisher + // has closed we emit whatever has been sealed. + let sealed = closed || as_of.iter().all(|t| !upper.less_equal(t)); + if !sealed { + // Not yet sealed. Hold the capability and wait for the next publisher activation. + return; + } + + for batch in chain { + if !batch.is_empty() { + // Emit whole `Arc` batches under a capability at the minimum time, exactly as + // the live `import` seeds historical batches. The `BatchFrontier` wrapper + // advances the batches' times to `as_of` on read, so the emitted capability + // (minimum) is always at or below the presented times. + let cap = caps.delayed(&batch_min::()); + output + .session(&cap) + .give(BatchFrontier::make_from(batch, as_of.borrow(), until.borrow())); + } + } + + // Release the capability: the single-time snapshot is complete, so the output + // frontier advances to the empty antichain. + capabilities = None; + } + }); + + Arranged { stream, trace } + } +} + +/// Deregisters an importer's replay queue when its source operator drops. +struct QueueGuard { + shared: SharedTraceRef, + reg_id: usize, +} + +impl Drop for QueueGuard { + fn drop(&mut self) { + if let Ok(mut state) = self.shared.state.lock() { + state.queues.remove(&self.reg_id); + } + } +} diff --git a/differential-dataflow/src/trace/implementations/mod.rs b/differential-dataflow/src/trace/implementations/mod.rs index 9b69cb12d..70041d3d4 100644 --- a/differential-dataflow/src/trace/implementations/mod.rs +++ b/differential-dataflow/src/trace/implementations/mod.rs @@ -53,6 +53,13 @@ pub use self::ord_neu::OrdKeySpine as KeySpine; pub use self::ord_neu::OrdKeyBatcher as KeyBatcher; pub use self::ord_neu::RcOrdKeyBuilder as KeyBuilder; +// `Arc`-backed variants of the `Val`/`Key` aliases, whose batches can be shared across threads. +// The batcher is shared with the `Rc` variants, since batchers do not reference-count batches. +pub use self::ord_neu::ArcOrdValSpine as ArcValSpine; +pub use self::ord_neu::ArcOrdValBuilder as ArcValBuilder; +pub use self::ord_neu::ArcOrdKeySpine as ArcKeySpine; +pub use self::ord_neu::ArcOrdKeyBuilder as ArcKeyBuilder; + use std::convert::TryInto; use serde::{Deserialize, Serialize}; diff --git a/differential-dataflow/src/trace/implementations/ord_neu.rs b/differential-dataflow/src/trace/implementations/ord_neu.rs index cf0c22a5f..1c5f58232 100644 --- a/differential-dataflow/src/trace/implementations/ord_neu.rs +++ b/differential-dataflow/src/trace/implementations/ord_neu.rs @@ -9,11 +9,13 @@ //! and should consume fewer resources (computation and memory) when it applies. use std::rc::Rc; +use std::sync::Arc; use crate::trace::implementations::spine_fueled::Spine; use crate::trace::implementations::merge_batcher::MergeBatcher; use crate::trace::implementations::merge_batcher::vec::VecMerger; use crate::trace::rc_blanket_impls::RcBuilder; +use crate::trace::arc_blanket_impls::ArcBuilder; use super::{Layout, Vector}; @@ -22,17 +24,29 @@ pub use self::key_batch::{OrdKeyBatch, OrdKeyBuilder}; /// A trace implementation using a spine of ordered lists. pub type OrdValSpine = Spine>>>; +/// An `Arc`-backed variant of [`OrdValSpine`]. +/// +/// Its batches are `Arc`'d rather than `Rc`'d, so a batch whose contents are `Send + Sync` can be +/// read from a thread other than the one maintaining the trace. Prefer [`OrdValSpine`] unless that +/// cross-thread sharing is needed, since atomic reference counting is marginally more expensive. +pub type ArcOrdValSpine = Spine>>>; /// A batcher using ordered lists. pub type OrdValBatcher = MergeBatcher>; /// A builder using ordered lists. pub type RcOrdValBuilder = RcBuilder, Vec<((K,V),T,R)>>>; +/// An `Arc`-backed variant of [`RcOrdValBuilder`], pairing with [`ArcOrdValSpine`]. +pub type ArcOrdValBuilder = ArcBuilder, Vec<((K,V),T,R)>>>; /// A trace implementation using a spine of ordered lists. pub type OrdKeySpine = Spine>>>; +/// An `Arc`-backed variant of [`OrdKeySpine`], readable from other threads. See [`ArcOrdValSpine`]. +pub type ArcOrdKeySpine = Spine>>>; /// A batcher for ordered lists. pub type OrdKeyBatcher = MergeBatcher>; /// A builder for ordered lists. pub type RcOrdKeyBuilder = RcBuilder, Vec<((K,()),T,R)>>>; +/// An `Arc`-backed variant of [`RcOrdKeyBuilder`], pairing with [`ArcOrdKeySpine`]. +pub type ArcOrdKeyBuilder = ArcBuilder, Vec<((K,()),T,R)>>>; pub use layers::{Vals, Upds}; /// Layers are containers of lists of some type. diff --git a/differential-dataflow/src/trace/mod.rs b/differential-dataflow/src/trace/mod.rs index c6f864f53..579b6c377 100644 --- a/differential-dataflow/src/trace/mod.rs +++ b/differential-dataflow/src/trace/mod.rs @@ -437,3 +437,122 @@ pub mod rc_blanket_impls { fn done(self) -> Rc { Rc::new(self.merger.done()) } } } + +/// Blanket implementations for atomically reference counted batches. +/// +/// These mirror [`rc_blanket_impls`], but use `Arc` so that batches whose contents +/// are `Send + Sync` can be shared across threads. This enables reading a trace's +/// batches from outside the worker that maintains the trace, for example to import +/// a snapshot of an arrangement into a dataflow running elsewhere. +pub mod arc_blanket_impls { + + use std::sync::Arc; + + use timely::progress::{Antichain, frontier::AntichainRef}; + use super::{Batch, BatchReader, Builder, Merger, Navigable, Cursor, Description}; + + impl Navigable for Arc { + /// The type used to enumerate the batch's contents. + type Cursor = ArcBatchCursor; + /// Acquires a cursor to the batch's contents. + fn cursor(&self) -> Self::Cursor { + ArcBatchCursor::new((**self).cursor()) + } + } + + impl BatchReader for Arc { + + type Time = B::Time; + /// The number of updates in the batch. + fn len(&self) -> usize { (**self).len() } + /// Describes the times of the updates in the batch. + fn description(&self) -> &Description { (**self).description() } + } + + /// Wrapper to provide cursor to nested scope. + pub struct ArcBatchCursor { + cursor: C, + } + + impl ArcBatchCursor { + fn new(cursor: C) -> Self { + ArcBatchCursor { + cursor, + } + } + } + + impl Cursor for ArcBatchCursor { + + type Storage = Arc; + + type Key<'a> = C::Key<'a>; + type ValOwn = C::ValOwn; + type Val<'a> = C::Val<'a>; + type Time = C::Time; + type TimeGat<'a> = C::TimeGat<'a>; + type Diff = C::Diff; + type DiffGat<'a> = C::DiffGat<'a>; + type KeyContainer = C::KeyContainer; + type ValContainer = C::ValContainer; + type TimeContainer = C::TimeContainer; + type DiffContainer = C::DiffContainer; + + #[inline] fn key_valid(&self, storage: &Self::Storage) -> bool { self.cursor.key_valid(storage) } + #[inline] fn val_valid(&self, storage: &Self::Storage) -> bool { self.cursor.val_valid(storage) } + + #[inline] fn key<'a>(&self, storage: &'a Self::Storage) -> Self::Key<'a> { self.cursor.key(storage) } + #[inline] fn val<'a>(&self, storage: &'a Self::Storage) -> Self::Val<'a> { self.cursor.val(storage) } + + #[inline] fn get_key<'a>(&self, storage: &'a Self::Storage) -> Option> { self.cursor.get_key(storage) } + #[inline] fn get_val<'a>(&self, storage: &'a Self::Storage) -> Option> { self.cursor.get_val(storage) } + + #[inline] + fn map_times, Self::DiffGat<'_>)>(&mut self, storage: &Self::Storage, logic: L) { + self.cursor.map_times(storage, logic) + } + + #[inline] fn step_key(&mut self, storage: &Self::Storage) { self.cursor.step_key(storage) } + #[inline] fn seek_key(&mut self, storage: &Self::Storage, key: Self::Key<'_>) { self.cursor.seek_key(storage, key) } + + #[inline] fn step_val(&mut self, storage: &Self::Storage) { self.cursor.step_val(storage) } + #[inline] fn seek_val(&mut self, storage: &Self::Storage, val: Self::Val<'_>) { self.cursor.seek_val(storage, val) } + + #[inline] fn rewind_keys(&mut self, storage: &Self::Storage) { self.cursor.rewind_keys(storage) } + #[inline] fn rewind_vals(&mut self, storage: &Self::Storage) { self.cursor.rewind_vals(storage) } + } + + /// An immutable collection of updates. + impl Batch for Arc { + type Merger = ArcMerger; + fn empty(lower: Antichain, upper: Antichain) -> Self { + Arc::new(B::empty(lower, upper)) + } + } + + /// Wrapper type for building atomically reference counted batches. + pub struct ArcBuilder { builder: B } + + /// Functionality for building batches from ordered update sequences. + impl Builder for ArcBuilder { + type Input = B::Input; + type Time = B::Time; + type Output = Arc; + fn with_capacity(keys: usize, vals: usize, upds: usize) -> Self { ArcBuilder { builder: B::with_capacity(keys, vals, upds) } } + fn push(&mut self, input: &mut Self::Input) { self.builder.push(input) } + fn done(self, description: Description) -> Arc { Arc::new(self.builder.done(description)) } + fn seal(chain: &mut Vec, description: Description) -> Self::Output { + Arc::new(B::seal(chain, description)) + } + } + + /// Wrapper type for merging atomically reference counted batches. + pub struct ArcMerger { merger: B::Merger } + + /// Represents a merge in progress. + impl Merger> for ArcMerger { + fn new(source1: &Arc, source2: &Arc, compaction_frontier: AntichainRef) -> Self { ArcMerger { merger: B::begin_merge(source1, source2, compaction_frontier) } } + fn work(&mut self, source1: &Arc, source2: &Arc, fuel: &mut isize) { self.merger.work(source1, source2, fuel) } + fn done(self) -> Arc { Arc::new(self.merger.done()) } + } +} diff --git a/differential-dataflow/tests/sharing.rs b/differential-dataflow/tests/sharing.rs new file mode 100644 index 000000000..4e5841fae --- /dev/null +++ b/differential-dataflow/tests/sharing.rs @@ -0,0 +1,314 @@ +//! Sharing arrangements across threads and runtimes. + +use std::sync::{mpsc, Mutex}; + +use timely::dataflow::operators::capture::Extract; +use timely::dataflow::operators::{Capture, Probe}; +use timely::dataflow::ProbeHandle; + +use timely::progress::Antichain; + +use differential_dataflow::input::InputSession; +use differential_dataflow::operators::arrange::sharing::SharedTraceHandle; +use differential_dataflow::operators::arrange::Arrange; +use differential_dataflow::trace::cursor::Cursor; +use differential_dataflow::trace::implementations::{ + ArcValBuilder, ArcValSpine, ValBatcher, ValBuilder, ValSpine, +}; +use differential_dataflow::trace::TraceReader; + +// The shared spine is `Arc`-backed, since `publish` requires `Send + Sync` batches. The reduce +// output below stays on the default `Rc`-backed `ValSpine`/`ValBuilder`, as it is not shared. +type Spine = ArcValSpine; +type Handle = SharedTraceHandle; + +/// A worker on one thread publishes an arrangement. A separate thread holds a `Send` handle, waits +/// for the publication frontier to pass a time, snapshots, and reads the collection at that time. +#[test] +fn snapshot_from_another_thread() { + let (handle_tx, handle_rx) = mpsc::channel::(); + // The reader raises this once it has its snapshot, so the publishing worker knows it can stop + // stepping. A retained trace handle keeps the dataflow from quiescing, so the worker never + // finishes on its own. + let done = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + let reader_done = std::sync::Arc::clone(&done); + + // Reader thread: receive the handle, snapshot as of time 2, accumulate. + let reader = std::thread::spawn(move || { + let handle = handle_rx.recv().unwrap(); + let snapshot = handle.snapshot_at(&2).expect("publisher closed early"); + reader_done.store(true, std::sync::atomic::Ordering::SeqCst); + let (mut cursor, storage) = snapshot.cursor(); + // Accumulate `(key, val) -> diff` for all times <= 2. + let mut acc: Vec<((u64, u64), isize)> = Vec::new(); + while cursor.key_valid(&storage) { + let key = *cursor.key(&storage); + while cursor.val_valid(&storage) { + let val = *cursor.val(&storage); + let mut sum = 0isize; + cursor.map_times(&storage, |t, d| { + if *t <= 2 { + sum += *d; + } + }); + if sum != 0 { + acc.push(((key, val), sum)); + } + cursor.step_val(&storage); + } + cursor.step_key(&storage); + } + acc.sort(); + acc + }); + + timely::execute_directly(move |worker| { + let mut input = InputSession::::new(); + let published = worker.dataflow(|scope| { + let arranged = input + .to_collection(scope) + .arrange::, ArcValBuilder<_, _, _, _>, Spine>(); + arranged.publish() + }); + handle_tx.send(published.handle()).unwrap(); + + // Time 0: (1,10)+1, (2,20)+1. Time 1: retract (2,20). Time 2: (3,30)+1. + input.advance_to(0); + input.insert((1, 10)); + input.insert((2, 20)); + input.advance_to(1); + input.remove((2, 20)); + input.advance_to(2); + input.insert((3, 30)); + input.advance_to(3); + input.flush(); + + // Step until the reader has taken its snapshot. The publisher advances `upper` as it steps, + // which unblocks the reader's `snapshot_at`. + while !done.load(std::sync::atomic::Ordering::SeqCst) { + worker.step(); + } + }); + + let got = reader.join().unwrap(); + // As of time 2: (1,10) present, (2,20) inserted then retracted, (3,30) inserted at 2. + assert_eq!(got, vec![((1, 10), 1), ((3, 30), 1)]); +} + +/// A publisher in one dataflow, an importer in another dataflow of the same worker, connected only +/// through a `SharedTraceHandle`. The importer reduces the shared arrangement and captures the +/// result, which must match a direct reduction of the input. +#[test] +fn import_through_shared_handle() { + let (capture_tx, capture_rx) = mpsc::channel(); + + timely::execute_directly(move |worker| { + let mut input = InputSession::::new(); + + // Publisher dataflow. + let handle = worker.dataflow(|scope| { + let arranged = input + .to_collection(scope) + .arrange::, ArcValBuilder<_, _, _, _>, Spine>(); + arranged.publish().handle() + }); + + // Importer dataflow: import the shared arrangement, count values per key, capture. + let mut probe = ProbeHandle::new(); + worker.dataflow(|scope| { + let imported = handle.import(scope, "Import"); + let counted = imported + .reduce_abelian::<_, ValBuilder<_, _, _, _>, ValSpine, _, _>( + "Count", + |_key, input, output| { + let count = input.iter().map(|&(_, d)| d).sum::() as u64; + output.push((count, 1)); + }, + |vec, key, upds| { + vec.clear(); + vec.extend(upds.drain(..).map(|(v, t, r)| ((*key, v), t, r))); + }, + ) + .as_collection(|k, v| (*k, *v)); + counted + .inner + .probe_with(&mut probe) + .capture_into(capture_tx.clone()); + }); + + // Key 1 has two values (10, 20), key 2 has one value (30), all at time 0. + input.advance_to(0); + input.insert((1, 10)); + input.insert((1, 20)); + input.insert((2, 30)); + input.advance_to(1); + input.flush(); + + // Step until the imported-and-reduced output has sealed time 0. The worker closure then + // returns, dropping the dataflows and flushing the capture. + while probe.less_than(&1) { + worker.step(); + } + }); + + // Reduce output at time 0: key 1 -> count 2, key 2 -> count 1. + let mut results: Vec<((u64, u64), u64, isize)> = capture_rx + .extract() + .into_iter() + .flat_map(|(_, data)| data) + .collect(); + results.sort(); + assert_eq!(results, vec![((1, 2), 0, 1), ((2, 1), 0, 1)]); +} + +/// Feeds `input` a fresh update at `time` and steps the worker, so the publisher operator +/// reactivates and republishes its `since` from the trace. The publisher only runs on stream +/// input, so a bare compaction advance on another handle is invisible until the next batch. +fn tick( + worker: &mut timely::worker::Worker, + input: &mut InputSession, + time: u64, +) { + input.advance_to(time); + input.insert((time, time)); + input.advance_to(time + 1); + input.flush(); + for _ in 0..20 { + worker.step(); + } +} + +/// Publishing must not pin compaction. With no registered reader holds, as the trace's writer +/// advances logical (and physical) compaction the publisher's own hold must follow, so the trace +/// actually compacts. We keep a writer handle on the original trace, advance its compaction to +/// `10`, tick the publisher, and assert the published `since` advanced: a fresh snapshot rejects a +/// read at `5` (a compacted time) while still serving `10`. Before the fix the publisher pins the +/// trace at the publish-time frontier, so `since` stays at `0` and the read at `5` is served. +#[test] +fn publish_without_readers_does_not_pin_compaction() { + timely::execute_directly(move |worker| { + let mut input = InputSession::::new(); + // Keep a writer handle (a plain `TraceAgent` clone) alongside the publication. This is the + // only hold besides the publisher's own agent. Crucially, no `SharedTraceHandle` is minted, + // so `state.logical_holds` stays empty: zero registered reader holds. + let (mut writer, published) = worker.dataflow(|scope| { + let arranged = input + .to_collection(scope) + .arrange::, ArcValBuilder<_, _, _, _>, Spine>(); + (arranged.trace.clone(), arranged.publish()) + }); + + // Seed some updates and let the publisher settle. + for t in 0..5 { + tick(worker, &mut input, t); + } + + // The writer advances its logical and physical compaction to 10, as a controller would, + // then a fresh batch reactivates the publisher so it republishes `since`. + writer.set_logical_compaction(Antichain::from_elem(10).borrow()); + writer.set_physical_compaction(Antichain::from_elem(10).borrow()); + tick(worker, &mut input, 10); + + // The published `since` followed the writer to 10: a fresh handle cannot snapshot at the + // compacted time 5, but can at 10. + let handle = published.handle(); + assert!( + handle.snapshot_at(&5).is_none(), + "snapshot at a compacted time must be rejected (since did not advance)" + ); + assert!( + handle.snapshot_at(&10).is_some(), + "snapshot at the compaction frontier must succeed" + ); + }); +} + +/// A live reader hold holds the trace back at its own frontier, and releasing it (drop) lets the +/// trace follow the writer. A reader registered at `since` 0 keeps time 0 readable even as the +/// writer advances to 10: the publisher forwards the reader's hold, so the trace stays at 0 and the +/// published `since` stays at 0. After the reader drops, a further tick lets the publisher follow +/// the writer, and a read at the now-compacted time 5 is rejected. This guards the fix on the "with +/// holds" side. Before the fix the publisher pins the trace regardless, so the post-drop compaction +/// never happens. +#[test] +fn import_hold_pins_then_releases() { + timely::execute_directly(move |worker| { + let mut input = InputSession::::new(); + let (mut writer, published) = worker.dataflow(|scope| { + let arranged = input + .to_collection(scope) + .arrange::, ArcValBuilder<_, _, _, _>, Spine>(); + (arranged.trace.clone(), arranged.publish()) + }); + + // A live reader hold registered at the publish-time `since` (0). + let reader = published.handle(); + + for t in 0..5 { + tick(worker, &mut input, t); + } + + // The writer advances to 10, but the live reader hold (0) holds the trace back: the + // publisher forwards the reader's hold, so the published `since` stays at 0 and a read at 0 + // still succeeds. + writer.set_logical_compaction(Antichain::from_elem(10).borrow()); + writer.set_physical_compaction(Antichain::from_elem(10).borrow()); + tick(worker, &mut input, 10); + assert!( + reader.snapshot_at(&0).is_some(), + "a live reader hold must keep its frontier readable" + ); + + // Drop the reader. Now only the publisher holds, and it follows the writer, so the trace + // compacts to 10. A handle minted afterward (so it does not itself hold the trace at 0) + // observes the advanced `since`: a read at the compacted time 5 is rejected. Before the fix + // the publisher pins the trace even with no readers, so `since` stays at 0 and this read + // succeeds. + drop(reader); + tick(worker, &mut input, 11); + let observer = published.handle(); + assert!( + observer.snapshot_at(&5).is_none(), + "after the reader drops, the trace must compact to the writer frontier" + ); + }); +} + +/// A publisher on a two-worker runtime (`peers() == 2`) hands its handle to an importer on a +/// single-threaded runtime (`peers() == 1`). Pairwise import assumes both sides shard keys the +/// same way, which requires equal total peers, so `import` must assert and panic rather than +/// silently reading the wrong shard. +#[test] +#[should_panic(expected = "peers")] +fn import_asserts_equal_peers() { + let (handle_tx, handle_rx) = mpsc::channel::(); + let handle_tx = Mutex::new(handle_tx); + + // Publisher runtime: two worker threads, so the publishing scope's `peers()` is 2. + timely::execute(timely::Config::process(2), move |worker| { + let mut input = InputSession::::new(); + let published = worker.dataflow(|scope| { + let arranged = input + .to_collection(scope) + .arrange::, ArcValBuilder<_, _, _, _>, Spine>(); + arranged.publish() + }); + // Only one worker needs to hand out a handle; the others publish redundantly (mirroring + // real SPMD dataflows) but nobody reads their handles. + if worker.index() == 0 { + handle_tx.lock().unwrap().send(published.handle()).unwrap(); + } + }) + .expect("publisher runtime failed to start"); + + let handle = handle_rx.recv().expect("publisher did not send a handle"); + + // Importer runtime: single-threaded (`execute_directly` never spawns worker threads), so + // `peers()` is 1, mismatching the publisher's 2. `import` runs on this same thread, so its + // panic unwinds directly into the test rather than being swallowed at a thread boundary. + timely::execute_directly(move |worker| { + worker.dataflow::(|scope| { + let _imported = handle.import(scope, "Import"); + }); + }); +} diff --git a/differential-dataflow/tests/trace.rs b/differential-dataflow/tests/trace.rs index d30e8ef2d..1863e7bb5 100644 --- a/differential-dataflow/tests/trace.rs +++ b/differential-dataflow/tests/trace.rs @@ -57,3 +57,34 @@ fn test_trace() { let vec_4 = cursor4.to_vec(&storage4, |k| k.clone(), |v| v.clone()); assert_eq!(vec_4, vec_3); } + +/// Batches of the `Arc`-backed spines can be handed to another thread, which can hold and read +/// them independently. This is the property that enables sharing a trace's contents outside the +/// worker that maintains it. The default `Rc`-backed spines do not have it, by design. +#[test] +fn test_batches_read_from_other_thread() { + use differential_dataflow::trace::Navigable; + use differential_dataflow::trace::implementations::ord_neu::{ArcOrdValBuilder, OrdValBatcher}; + + fn assert_send_sync(_: &T) {} + + let mut batcher = OrdValBatcher::::new(None, 0); + batcher.push_into(vec![ + ((1, 2), 0, 1), + ((2, 3), 1, 1), + ]); + let (mut chain, description) = batcher.seal(Antichain::from_elem(2)); + let batch = ArcOrdValBuilder::::seal(&mut chain, description); + + assert_send_sync(&batch); + + let read = std::thread::spawn(move || { + let mut cursor = batch.cursor(); + cursor.to_vec(&batch, |k| k.clone(), |v| v.clone()) + }).join().unwrap(); + + assert_eq!(read, vec![ + ((1, 2), vec![(0, 1)]), + ((2, 3), vec![(1, 1)]), + ]); +}