Skip to content

Chunk: merge and advance as one step #1083

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
frankmcsherry wants to merge 1 commit into
base: master-next
Choose a base branch
from
Draft

Chunk: merge and advance as one step #1083

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 34 additions & 9 deletions differential-dataflow/src/trace/chunk/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -88,6 +88,29 @@ pub trait Chunk: Sized + Clone {
/// point it appends the queue to the output without the method's assistance.
fn merge(in1: &mut VecDeque<Self>, in2: &mut VecDeque<Self>, out: &mut VecDeque<Self>);

/// Merge the fronts of two input deques and advance the result by `frontier`, as one step.
///
/// This is [`merge`](Chunk::merge) followed by [`advance`](Chunk::advance), and that is its
/// default. An implementor for whom writing a chunk is the expensive part can override it to
/// write each merged update once, where the two steps write it twice.
///
/// `carry` holds merged updates that are not yet advanced, as `advance` leaves them at the
/// front of its input: a trailing `(key, val)` group that may continue. The method may leave
/// such a group there in turn. Advanced, consolidated chunks go to `out`.
/// The obligations of `merge` hold: an invocation consumes at least one of its inputs.
fn merge_advance(
in1: &mut VecDeque<Self>,
in2: &mut VecDeque<Self>,
frontier: AntichainRef<Self::Time>,
carry: &mut VecDeque<Self>,
out: &mut VecDeque<Self>,
) {
let mut produced = VecDeque::new();
Self::merge(in1, in2, &mut produced);
carry.extend(produced);
Self::advance(carry, frontier, false, out);
}

/// Partition `input` updates into `keep` (greater or equal `frontier`) or not (`ship`).
///
/// An implementation should yield with some frequency to allow the output to "settle".
Expand Down Expand Up @@ -378,15 +401,20 @@ where
self.idx2 += 1;
}

// Merge's per-tick output (a burst's worth, or one tail chunk), measured
// for fuel before it joins the carry already in `merged`.
let mut produced = VecDeque::new();
// A step's work, for fuel, is the updates it takes from the inputs: a burst's
// worth merged, or one tail chunk.
let work: usize;
if !self.in1.is_empty() && !self.in2.is_empty() {
// Both sides have data: drain the loaded burst.
C::merge(&mut self.in1, &mut self.in2, &mut produced);
// Both sides have data: merge and advance the fronts of the loaded burst.
let loaded = |in1: &VecDeque<C>, in2: &VecDeque<C>| in1.iter().chain(in2.iter()).map(C::len).sum::<usize>();
let before = loaded(&self.in1, &self.in2);
C::merge_advance(&mut self.in1, &mut self.in2, self.frontier.borrow(), &mut self.merged, &mut self.advanced);
work = before - loaded(&self.in1, &self.in2);
} else if let Some(chunk) = self.in1.pop_front().or_else(|| self.in2.pop_front()) {
// Exactly one side has data: flush its verbatim tail, one chunk a step.
produced.push_back(chunk);
work = chunk.len();
self.merged.push_back(chunk);
C::advance(&mut self.merged, self.frontier.borrow(), false, &mut self.advanced);
} else {
// Both sources drained: final flush of advance's and settle's carries.
C::advance(&mut self.merged, self.frontier.borrow(), true, &mut self.advanced);
Expand All @@ -395,9 +423,6 @@ where
break;
}

let work: usize = produced.iter().map(C::len).sum();
self.merged.extend(produced);
C::advance(&mut self.merged, self.frontier.borrow(), false, &mut self.advanced);
// Maintain grading at the yield boundary: this step may exhaust `fuel` and
// suspend with `advanced` held, and held chunk state must be graded.
C::settle(&mut self.advanced, false, &mut self.settled);
Expand Down
227 changes: 185 additions & 42 deletions interactive/src/runtime/chunk.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -136,56 +136,59 @@ where
}
}

/// Concatenate a run of (globally-sorted) chunks into one combined `(kv, times, diffs)`, the
/// key/val columns a range at a time ([`concat_blocks`]).
fn concat(chunks: &[Self]) -> (CValue, ColTimes<T>, Vec<R>) {
let kvs: Vec<CValue> = chunks.iter().map(Self::kv).collect();
let total: usize = chunks.iter().map(Self::len_).sum();
let (mut times, mut diffs) = (ColTimes::new(), Vec::with_capacity(total));
for ch in chunks {
times.push_range(ch.times(), 0, ch.len_());
diffs.extend_from_slice(ch.diffs());
}
let kv = if total == 0 { CValue::Unit(0) } else { concat_blocks(&kvs) };
(kv, times, diffs)
}

fn len_(&self) -> usize { self.0.times.len() }
}

impl<T, R> Chunk for CorgiChunk<T, R>
where
T: ColTime,
R: Semigroup + Clone + 'static,
{
type Time = T;
const TARGET: usize = TARGET;

fn len(&self) -> usize { self.0.times.len() }

/// Merge of the two front chunks through their shared horizon, FULLY consolidating equal
/// `(key, val, time)` triples and pushing back the survivor's suffix (the fueled-merger
/// contract). Batched: corgi's `survey_groups` reports the interleaving of the two `(key, val)`
/// columns as maximal ranges exclusive to one side and equal classes as their ranges on BOTH
/// sides, so rows only one side holds are copied by range, and only the classes both sides
/// hold — where consolidation can happen — are merged row by row, on their times, which corgi
/// does not own. The shape is walked once per level per chunk, never once per row.
fn merge(in1: &mut VecDeque<Self>, in2: &mut VecDeque<Self>, out: &mut VecDeque<Self>) {
/// The merge of the two front chunks through their shared horizon, as [`Chunk::merge`]
/// describes; and with `compact`, a frontier and a carry, the advancing and consolidating of
/// the merged rows as well, as [`Chunk::advance`] would do next.
///
/// With `compact` the merged order stays as indices into the sources. Complete `(key, val)`
/// groups have their times advanced, each group of several rows is sorted and summed on its
/// times, and only the surviving rows are gathered. The key and value columns are so written
/// once, where merging and then advancing writes them twice.
/// The last group may continue in a later chunk. It is held back in the carry, unadvanced,
/// and a group the step before held back joins this step as a third source.
fn merge_fronts(
in1: &mut VecDeque<Self>,
in2: &mut VecDeque<Self>,
mut compact: Option<(AntichainRef<T>, &mut VecDeque<Self>)>,
out: &mut VecDeque<Self>,
) {
let c1 = in1.pop_front().unwrap();
let c2 = in2.pop_front().unwrap();
let (kv1, kv2) = (c1.kv(), c2.kv());
let (n1, n2) = (c1.len_(), c2.len_());
let (t1, d1) = (c1.times(), c1.diffs());
let (t2, d2) = (c2.times(), c2.diffs());
// A group the step before held back: one `(key, val)`, whose rows come first.
let held = compact.as_mut().and_then(|(_, carry)| carry.pop_front());
let kv0 = held.as_ref().map(Self::kv);
let n0 = held.as_ref().map_or(0, Self::len_);

let runs = survey_groups(&kv1, &kv2);
let (mut tags, mut offs) = (Vec::with_capacity(n1 + n2), Vec::with_capacity(n1 + n2));
let (mut tags, mut offs) = (Vec::with_capacity(n0 + n1 + n2), Vec::with_capacity(n0 + n1 + n2));
// Emitted chunks own this allocation. Size it by surviving rows so cancellation
// does not leave a small result holding storage for both full inputs. Times are
// reserved for both inputs, to copy ranges without regrowing, and shrunk below if
// fewer than half the rows survive.
let (mut times, mut diffs): (ColTimes<T>, Vec<R>) = (ColTimes::new(), Vec::new());
times.reserve(t1.width().max(t2.width()), n1 + n2);
times.reserve(t1.width().max(t2.width()), n0 + n1 + n2);
if let Some(held) = &held {
tags.resize(n0, 2);
offs.extend(0..n0);
times.push_range(held.times(), 0, n0);
diffs.extend_from_slice(held.diffs());
}
// When compacting: where each group starts in the merged order. A run exclusive to one
// side starts a group, as does each group within it, read off that side's own group ends.
// A class both sides hold is one group.
let ends = compact.as_ref().map(|_| (group_bounds(&kv1), group_bounds(&kv2)));
let (mut e1, mut e2) = (0, 0);
let mut starts: Vec<usize> = Vec::new();
if n0 > 0 { starts.push(0); }
let mark = |starts: &mut Vec<usize>, ends: &[usize], e: &mut usize, base: usize, lo: usize, hi: usize| {
starts.push(base);
while *e < ends.len() && ends[*e] <= lo { *e += 1; }
while *e < ends.len() && ends[*e] < hi { starts.push(base + ends[*e] - lo); *e += 1; }
};
// Where the survivor's pushed-back suffix starts: the last run, if it is exclusive.
let (mut p1, mut p2) = (n1, n2);
let copy = |tags: &mut Vec<usize>, offs: &mut Vec<usize>, times: &mut ColTimes<T>, diffs: &mut Vec<R>, side: usize, lo: usize, hi: usize| {
Expand All @@ -199,14 +202,21 @@ where
let last = r + 1 == runs.len();
match *run {
GroupRun::A(lo, hi) => {
if last { p1 = lo; } else { copy(&mut tags, &mut offs, &mut times, &mut diffs, 0, lo, hi); }
if last { p1 = lo; } else {
if let Some((ends1, _)) = &ends { mark(&mut starts, ends1, &mut e1, tags.len(), lo, hi); }
copy(&mut tags, &mut offs, &mut times, &mut diffs, 0, lo, hi);
}
}
GroupRun::B(lo, hi) => {
if last { p2 = lo; } else { copy(&mut tags, &mut offs, &mut times, &mut diffs, 1, lo, hi); }
if last { p2 = lo; } else {
if let Some((_, ends2)) = &ends { mark(&mut starts, ends2, &mut e2, tags.len(), lo, hi); }
copy(&mut tags, &mut offs, &mut times, &mut diffs, 1, lo, hi);
}
}
GroupRun::Both(a_lo, a_hi, b_lo, b_hi) => {
// Both classes hold one `(key, val)`, sorted by time: merge on time, summing the
// diffs of equal times, which is the consolidation.
if ends.is_some() { starts.push(tags.len()); }
let (mut i, mut j) = (a_lo, b_lo);
while i < a_hi && j < b_hi {
match t1.cmp_cross(i, t2, j) {
Expand Down Expand Up @@ -238,9 +248,62 @@ where
// The survey can be large. Its last use precedes materializing the
// output columns, so do not retain it across that allocation peak.
drop(runs);
if times.len() * 2 < n1 + n2 { times.shrink_to_fit(); }
let srcs = [Some(&kv1), Some(&kv2)];
Self::emit(&srcs, &tags, &offs, times, diffs, out);
let mut srcs = vec![Some(&kv1), Some(&kv2)];
if let Some(kv0) = &kv0 { srcs.push(Some(kv0)); }
match compact {
None => {
if times.len() * 2 < n1 + n2 { times.shrink_to_fit(); }
Self::emit(&srcs, &tags, &offs, times, diffs, out);
}
Some((frontier, carry)) => {
let m = tags.len();
// A class whose rows all cancelled leaves a start with no rows.
starts.dedup();
if starts.last() == Some(&m) { starts.pop(); }
// The held group continues if the first merged row has its `(key, val)`.
if n0 > 0 && starts.get(1) == Some(&n0) {
let first = if tags[n0] == 0 { &kv1 } else { &kv2 };
if equal_idx(kv0.as_ref().unwrap(), first, &[0], &[offs[n0]])[0] == 0 { starts.remove(1); }
}
if let Some(&g) = starts.last() {
// The last group may continue in a later chunk: hold it back, as it is.
let mut last = ColTimes::new();
last.push_range(&times, g, m);
carry.push_front(Self::from_kv(gather_lanes(&srcs, &tags[g..], &offs[g..]), last, diffs[g..].to_vec()));
// Advance the complete groups' times in one lane-wise pass. Advancing can reorder
// a group's times and make some equal, so each group of several rows is sorted
// on its times and equal times sum. All rows of a group share `(key, val)`, so
// any of them stands for the group in the gather.
times.advance_by(frontier, g);
let (mut otags, mut ooffs, mut rows) = (Vec::with_capacity(g), Vec::with_capacity(g), Vec::with_capacity(g));
let mut odiffs: Vec<R> = Vec::with_capacity(g);
let mut order: Vec<usize> = Vec::new();
for group in starts.windows(2) {
let (lo, hi) = (group[0], group[1]);
if hi - lo == 1 {
if !diffs[lo].is_zero() { otags.push(tags[lo]); ooffs.push(offs[lo]); rows.push(lo); odiffs.push(diffs[lo].clone()); }
continue;
}
order.clear();
order.extend(lo..hi);
order.sort_by(|&a, &b| times.cmp(a, b));
let mut k = 0;
while k < order.len() {
let rep = order[k];
let mut d = diffs[rep].clone();
k += 1;
while k < order.len() && times.cmp(order[k], rep) == Ordering::Equal {
d.plus_equals(&diffs[order[k]]);
k += 1;
}
if !d.is_zero() { otags.push(tags[rep]); ooffs.push(offs[rep]); rows.push(rep); odiffs.push(d); }
}
}
// The one gather of the key and value columns.
if !rows.is_empty() { Self::emit(&srcs, &otags, &ooffs, times.gather(&rows), odiffs, out); }
}
}
}
drop(tags);
drop(offs);

Expand All @@ -259,6 +322,63 @@ where
}
}

/// Concatenate a run of (globally-sorted) chunks into one combined `(kv, times, diffs)`, the
/// key/val columns a range at a time ([`concat_blocks`]).
fn concat(chunks: &[Self]) -> (CValue, ColTimes<T>, Vec<R>) {
let kvs: Vec<CValue> = chunks.iter().map(Self::kv).collect();
let total: usize = chunks.iter().map(Self::len_).sum();
let (mut times, mut diffs) = (ColTimes::new(), Vec::with_capacity(total));
for ch in chunks {
times.push_range(ch.times(), 0, ch.len_());
diffs.extend_from_slice(ch.diffs());
}
let kv = if total == 0 { CValue::Unit(0) } else { concat_blocks(&kvs) };
(kv, times, diffs)
}

fn len_(&self) -> usize { self.0.times.len() }
}

impl<T, R> Chunk for CorgiChunk<T, R>
where
T: ColTime,
R: Semigroup + Clone + 'static,
{
type Time = T;
const TARGET: usize = TARGET;

fn len(&self) -> usize { self.0.times.len() }

/// Merge of the two front chunks through their shared horizon, FULLY consolidating equal
/// `(key, val, time)` triples and pushing back the survivor's suffix (the fueled-merger
/// contract). Batched: corgi's `survey_groups` reports the interleaving of the two `(key, val)`
/// columns as maximal ranges exclusive to one side and equal classes as their ranges on BOTH
/// sides, so rows only one side holds are copied by range, and only the classes both sides
/// hold — where consolidation can happen — are merged row by row, on their times, which corgi
/// does not own. The shape is walked once per level per chunk, never once per row.
fn merge(in1: &mut VecDeque<Self>, in2: &mut VecDeque<Self>, out: &mut VecDeque<Self>) {
Self::merge_fronts(in1, in2, None, out);
}

/// Merge and advance as one step ([`merge_fronts`](Self::merge_fronts)), which gathers the
/// key and value columns once. With more than a single group held over, the two steps.
fn merge_advance(
in1: &mut VecDeque<Self>,
in2: &mut VecDeque<Self>,
frontier: AntichainRef<T>,
carry: &mut VecDeque<Self>,
out: &mut VecDeque<Self>,
) {
if carry.len() > 1 {
let mut produced = VecDeque::new();
Self::merge(in1, in2, &mut produced);
carry.extend(produced);
Self::advance(carry, frontier, false, out);
} else {
Self::merge_fronts(in1, in2, Some((frontier, carry)), out);
}
}

fn extract(
input: &mut VecDeque<Self>,
frontier: AntichainRef<T>,
Expand Down Expand Up @@ -1181,4 +1301,27 @@ mod test {
assert_eq!(result.as_ref().map_or_else(BTreeMap::new, read_batch), want, "u1={u1:?}\nu2={u2:?}\nf={f}");
}
}

// A `(key, val)` group that spans every chunk of both inputs is held back from step to step,
// and advanced once, whole. Compacted to a frontier past all of its times it is one update.
#[test]
fn merge_advance_carries_a_group_across_chunks() {
let mut u1: Vec<((u64, u64), u64, i64)> = (0..9).map(|t| ((7, 7), 2 * t, 1)).collect();
let mut u2: Vec<((u64, u64), u64, i64)> = (0..9).map(|t| ((7, 7), 2 * t + 1, 1)).collect();
u1.push(((1, 1), 0, 1));
u2.push(((9, 9), 0, 1));
for sz in 1..5 {
let (s1, s2) = (batch(&u1, sz), batch(&u2, sz));
let frontier = Antichain::from_elem(100);
let mut merger = ChunkBatchMerger::new(&s1, &s2, frontier.borrow());
loop {
let mut fuel = 1isize;
merger.work(&s1, &s2, &mut fuel);
if fuel > 0 { break; }
}
let result = merger.done().expect("three updates remain");
assert_eq!(read_batch(&result), reference(&u1, &u2, 100), "chunks of {sz}");
assert_eq!(result.chunks.iter().map(Chunk::len).sum::<usize>(), 3, "chunks of {sz}");
}
}
}
Loading