Skip to content

Commit 02e18e1

Browse files
frankmcsherryfrankmcsherry
committed
Pivot container builders above capability checking
1 parent 04aefdd commit 02e18e1

File tree

14 files changed

+253
-182
lines changed

14 files changed

+253
-182
lines changed

‎timely/src/dataflow/channels/pushers/mod.rs‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,3 +6,9 @@ pub mod tee;
66
pub mod exchange;
77
pub mod counter;
88
pub mod buffer;
9+
pub mod progress;
10+
11+
/// An output pusher which validates capabilities, records progress, and tees output.
12+
pub type Output<T, C> = progress::Progress<T, counter::Counter<T, tee::Tee<T, C>>>;
13+
/// An output session that will flush the output when dropped.
14+
pub type OutputSession<'a, T, C> = progress::ProgressSession<'a, T, C, counter::Counter<T, tee::Tee<T, C>>>;

‎timely/src/dataflow/channels/pushers/progress.rs‎

Lines changed: 67 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,67 @@
1+
//! A wrapper that allows containers to be sent by validating capabilities.
2+
3+
use std::rc::Rc;
4+
use std::cell::RefCell;
5+
6+
use crate::progress::{ChangeBatch, Timestamp};
7+
use crate::dataflow::channels::Message;
8+
use crate::dataflow::operators::CapabilityTrait;
9+
use crate::communication::Push;
10+
use crate::Container;
11+
12+
/// A wrapper that allows containers to be sent by validating capabilities.
13+
#[derive(Debug)]
14+
pub struct Progress<T, P> {
15+
pushee: P,
16+
internal: Rc<RefCell<ChangeBatch<T>>>,
17+
port: usize,
18+
}
19+
20+
impl<T: Timestamp, P> Progress<T, P> {
21+
/// Ships a container using a provided capability.
22+
///
23+
/// On return, the container may hold undefined contents and should be cleared before it is reused.
24+
#[inline] pub fn give<C: Container, CT: CapabilityTrait<T>>(&mut self, capability: &CT, container: &mut C) where P: Push<Message<T, C>> {
25+
debug_assert!(self.valid(capability), "Attempted to open output session with invalid capability");
26+
if !container.is_empty() { Message::push_at(container, capability.time().clone(), &mut self.pushee); }
27+
}
28+
/// Activates a `Progress` into a `ProgressSession` which will flush when dropped.
29+
pub fn activate<'a, C>(&'a mut self) -> ProgressSession<'a, T, C, P> where P: Push<Message<T, C>> {
30+
ProgressSession {
31+
borrow: self,
32+
marker: std::marker::PhantomData,
33+
}
34+
}
35+
/// Determines if the capability is valid for this output.
36+
pub fn valid<CT: CapabilityTrait<T>>(&self, capability: &CT) -> bool {
37+
capability.valid_for_output(&self.internal, self.port)
38+
}
39+
}
40+
41+
impl<T, P> Progress<T, P> where T : Ord+Clone+'static {
42+
/// Allocates a new `Progress` from a pushee and capability validation information.
43+
pub fn new(pushee: P, internal: Rc<RefCell<ChangeBatch<T>>>, port: usize) -> Self {
44+
Self { pushee, internal, port }
45+
}
46+
}
47+
48+
/// A session that provides access to a `Progress` but will flush when dropped.
49+
///
50+
/// The type of the container `C` must be known, as long as the flushing action requires a specific `Push` implementation.
51+
pub struct ProgressSession<'a, T: Timestamp, C, P: Push<Message<T, C>>> {
52+
borrow: &'a mut Progress<T, P>,
53+
marker: std::marker::PhantomData<C>,
54+
}
55+
56+
impl<'a, T: Timestamp, C, P: Push<Message<T, C>>> std::ops::Deref for ProgressSession<'a, T, C, P> {
57+
type Target = Progress<T, P>;
58+
fn deref(&self) -> &Self::Target { self.borrow }
59+
}
60+
61+
impl<'a, T: Timestamp, C, P: Push<Message<T, C>>> std::ops::DerefMut for ProgressSession<'a, T, C, P> {
62+
fn deref_mut(&mut self) -> &mut Self::Target { self.borrow }
63+
}
64+
65+
impl<'a, T: Timestamp, C, P: Push<Message<T, C>>> Drop for ProgressSession<'a, T, C, P> {
66+
fn drop(&mut self) { self.borrow.pushee.push(&mut None); }
67+
}

‎timely/src/dataflow/operators/branch.rs‎

Lines changed: 11 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
//! Operators that separate one stream into two streams based on some condition
22
33
use crate::dataflow::channels::pact::Pipeline;
4+
use crate::dataflow::operators::generic::OutputBuilder;
45
use crate::dataflow::operators::generic::builder_rc::OperatorBuilder;
56
use crate::dataflow::{Scope, Stream, StreamCore};
67
use crate::{Container, Data};
@@ -43,8 +44,11 @@ impl<S: Scope, D: Data> Branch<S, D> for Stream<S, D> {
4344
builder.set_notify(false);
4445

4546
let mut input = builder.new_input(self, Pipeline);
46-
let (mut output1, stream1) = builder.new_output();
47-
let (mut output2, stream2) = builder.new_output();
47+
let (output1, stream1) = builder.new_output();
48+
let (output2, stream2) = builder.new_output();
49+
50+
let mut output1 = OutputBuilder::from(output1);
51+
let mut output2 = OutputBuilder::from(output2);
4852

4953
builder.build(move |_| {
5054
move |_frontiers| {
@@ -99,8 +103,11 @@ impl<S: Scope, C: Container> BranchWhen<S::Timestamp> for StreamCore<S, C> {
99103
builder.set_notify(false);
100104

101105
let mut input = builder.new_input(self, Pipeline);
102-
let (mut output1, stream1) = builder.new_output();
103-
let (mut output2, stream2) = builder.new_output();
106+
let (output1, stream1) = builder.new_output();
107+
let (output2, stream2) = builder.new_output();
108+
109+
let mut output1 = OutputBuilder::from(output1);
110+
let mut output2 = OutputBuilder::from(output2);
104111

105112
builder.build(move |_| {
106113

‎timely/src/dataflow/operators/capability.rs‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@ use crate::dataflow::channels::pullers::counter::ConsumedGuard;
3737
pub trait CapabilityTrait<T: Timestamp> {
3838
/// The timestamp associated with the capability.
3939
fn time(&self) -> &T;
40+
/// Validates that the capability is valid for a specific internal buffer and output port.
4041
fn valid_for_output(&self, query_buffer: &Rc<RefCell<ChangeBatch<T>>>, port: usize) -> bool;
4142
}
4243

@@ -229,7 +230,7 @@ type CapabilityUpdates<T> = Rc<RefCell<Vec<Rc<RefCell<ChangeBatch<T>>>>>>;
229230

230231
/// An capability of an input port.
231232
///
232-
/// Holding onto this capability will implicitly holds onto a capability for all the outputs
233+
/// Holding onto this capability will implicitly hold on to a capability for all the outputs
233234
/// ports this input is connected to, after the connection summaries have been applied.
234235
///
235236
/// This input capability supplies a `retain_for_output(self)` method which consumes the input

‎timely/src/dataflow/operators/core/concat.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -85,7 +85,7 @@ impl<G: Scope, C: Container> Concatenate<G, C> for G {
8585
let mut output = output.activate();
8686
for handle in handles.iter_mut() {
8787
handle.for_each(|time, data| {
88-
output.session(&time).give_container(data);
88+
output.give(&time, data);
8989
})
9090
}
9191
}

‎timely/src/dataflow/operators/core/feedback.rs‎

Lines changed: 2 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,7 @@
11
//! Create cycles in a timely dataflow graph.
22
33
use crate::Container;
4-
use crate::container::CapacityContainerBuilder;
54
use crate::dataflow::channels::pact::Pipeline;
6-
use crate::dataflow::channels::pushers::Tee;
7-
use crate::dataflow::operators::generic::OutputWrapper;
85
use crate::dataflow::operators::generic::builder_rc::OperatorBuilder;
96
use crate::dataflow::scopes::child::Iterative;
107
use crate::dataflow::{StreamCore, Scope};
@@ -121,9 +118,7 @@ impl<G: Scope, C: Container> ConnectLoop<G, C> for StreamCore<G, C> {
121118
input.for_each(|cap, data| {
122119
if let Some(new_time) = summary.results_in(cap.time()) {
123120
let new_cap = cap.delayed(&new_time);
124-
output
125-
.session(&new_cap)
126-
.give_container(data);
121+
output.give(&new_cap, data);
127122
}
128123
});
129124
});
@@ -135,5 +130,5 @@ impl<G: Scope, C: Container> ConnectLoop<G, C> for StreamCore<G, C> {
135130
pub struct Handle<G: Scope, C: Container> {
136131
builder: OperatorBuilder<G>,
137132
summary: <G::Timestamp as Timestamp>::Summary,
138-
output: OutputWrapper<G::Timestamp, CapacityContainerBuilder<C>, Tee<G::Timestamp, C>>,
133+
output: crate::dataflow::channels::pushers::Output<G::Timestamp, C>,
139134
}

‎timely/src/dataflow/operators/core/ok_err.rs‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ use crate::Container;
44
use crate::container::{DrainContainer, SizableContainer, PushInto};
55
use crate::dataflow::channels::pact::Pipeline;
66
use crate::dataflow::operators::generic::builder_rc::OperatorBuilder;
7+
use crate::dataflow::operators::generic::OutputBuilder;
78
use crate::dataflow::{Scope, StreamCore};
89

910
/// Extension trait for `Stream`.
@@ -53,8 +54,11 @@ impl<S: Scope, C: Container + DrainContainer> OkErr<S, C> for StreamCore<S, C> {
5354
builder.set_notify(false);
5455

5556
let mut input = builder.new_input(self, Pipeline);
56-
let (mut output1, stream1) = builder.new_output();
57-
let (mut output2, stream2) = builder.new_output();
57+
let (output1, stream1) = builder.new_output();
58+
let (output2, stream2) = builder.new_output();
59+
60+
let mut output1 = OutputBuilder::from(output1);
61+
let mut output2 = OutputBuilder::from(output2);
5862

5963
builder.build(move |_| {
6064
move |_frontiers| {

‎timely/src/dataflow/operators/core/partition.rs‎

Lines changed: 24 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,9 @@
11
//! Partition a stream of records into multiple streams.
2+
use std::collections::BTreeMap;
23

34
use crate::container::{DrainContainer, ContainerBuilder, PushInto};
45
use crate::dataflow::channels::pact::Pipeline;
56
use crate::dataflow::operators::generic::builder_rc::OperatorBuilder;
6-
use crate::dataflow::operators::InputCapability;
77
use crate::dataflow::{Scope, StreamCore};
88
use crate::Container;
99

@@ -46,43 +46,41 @@ impl<G: Scope, C: Container + DrainContainer> Partition<G, C> for StreamCore<G,
4646
let mut outputs = Vec::with_capacity(parts as usize);
4747
let mut streams = Vec::with_capacity(parts as usize);
4848

49+
let mut c_build = CB::default();
50+
4951
for _ in 0..parts {
50-
let (output, stream) = builder.new_output::<CB>();
52+
let (output, stream) = builder.new_output::<CB::Container>();
5153
outputs.push(output);
5254
streams.push(stream);
5355
}
5456

57+
let mut caps = BTreeMap::default();
58+
5559
builder.build(move |_| {
56-
let mut todo = vec![];
60+
let mut todo: BTreeMap<_,Vec<_>> = Default::default();
5761
move |_frontiers| {
5862
let mut handles = outputs.iter_mut().map(|o| o.activate()).collect::<Vec<_>>();
59-
60-
// The capability associated with each session in `sessions`.
61-
let mut sessions_cap: Option<InputCapability<G::Timestamp>> = None;
62-
let mut sessions = vec![];
63-
63+
let mut parts = BTreeMap::<u64,Vec<_>>::default();
6464
while let Some((cap, data)) = input.next() {
65-
todo.push((cap, std::mem::take(data)));
65+
todo.entry(cap.time().clone()).or_default().push(std::mem::take(data));
66+
caps.insert(cap.time().clone(), cap);
6667
}
67-
todo.sort_unstable_by(|a, b| a.0.cmp(&b.0));
6868

69-
for (cap, mut data) in todo.drain(..) {
70-
if sessions_cap.as_ref().map_or(true, |s_cap| s_cap.time() != cap.time()) {
71-
sessions = handles.iter_mut().map(|h| (None, Some(h))).collect();
72-
sessions_cap = Some(cap);
69+
while let Some((time, dataz)) = todo.pop_first() {
70+
let cap = caps.remove(&time).unwrap();
71+
for mut data in dataz.into_iter() {
72+
for datum in data.drain() {
73+
let (part, datum) = route(datum);
74+
parts.entry(part).or_default().push(datum);
75+
}
7376
}
74-
for datum in data.drain() {
75-
let (part, datum2) = route(datum);
76-
77-
let session = match sessions[part as usize] {
78-
(Some(ref mut s), _) => s,
79-
(ref mut session_slot, ref mut handle) => {
80-
let handle = handle.take().unwrap();
81-
let session = handle.session_with_builder(sessions_cap.as_ref().unwrap());
82-
session_slot.insert(session)
83-
}
84-
};
85-
session.give(datum2);
77+
while let Some((part, data)) = parts.pop_first() {
78+
for datum in data.into_iter() {
79+
c_build.push_into(datum);
80+
}
81+
while let Some(container) = c_build.finish() {
82+
handles[part as usize].give(&cap, container);
83+
}
8684
}
8785
}
8886
}

‎timely/src/dataflow/operators/generic/builder_rc.rs‎

Lines changed: 21 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -9,15 +9,13 @@ use crate::progress::operate::SharedProgress;
99
use crate::progress::frontier::{Antichain, MutableAntichain};
1010

1111
use crate::Container;
12-
use crate::container::ContainerBuilder;
1312
use crate::dataflow::{Scope, StreamCore};
14-
use crate::dataflow::channels::pushers::Tee;
1513
use crate::dataflow::channels::pushers::Counter as PushCounter;
16-
use crate::dataflow::channels::pushers::buffer::Buffer as PushBuffer;
14+
use crate::dataflow::channels::pushers;
1715
use crate::dataflow::channels::pact::ParallelizationContract;
1816
use crate::dataflow::channels::pullers::Counter as PullCounter;
1917
use crate::dataflow::operators::capability::Capability;
20-
use crate::dataflow::operators::generic::handles::{InputHandleCore, new_input_handle, OutputWrapper};
18+
use crate::dataflow::operators::generic::handles::{InputHandleCore, new_input_handle};
2119
use crate::dataflow::operators::generic::operator_info::OperatorInfo;
2220
use crate::dataflow::operators::generic::builder_raw::OperatorShape;
2321
use crate::progress::operate::PortConnectivity;
@@ -90,7 +88,7 @@ impl<G: Scope> OperatorBuilder<G> {
9088
}
9189

9290
/// Adds a new output to a generic operator builder, returning the `Push` implementor to use.
93-
pub fn new_output<CB: ContainerBuilder>(&mut self) -> (OutputWrapper<G::Timestamp, CB, Tee<G::Timestamp, CB::Container>>, StreamCore<G, CB::Container>) {
91+
pub fn new_output<C: Container>(&mut self) -> (pushers::Output<G::Timestamp, C>, StreamCore<G, C>) {
9492
let connection = (0..self.builder.shape().inputs()).map(|i| (i, Antichain::from_elem(Default::default())));
9593
self.new_output_connection(connection)
9694
}
@@ -103,9 +101,10 @@ impl<G: Scope> OperatorBuilder<G> {
103101
///
104102
/// Commonly the connections are either the unit summary, indicating the same timestamp might be produced as output, or an empty
105103
/// antichain indicating that there is no connection from the input to the output.
106-
pub fn new_output_connection<CB: ContainerBuilder, I>(&mut self, connection: I) -> (
107-
OutputWrapper<G::Timestamp, CB, Tee<G::Timestamp, CB::Container>>,
108-
StreamCore<G, CB::Container>
104+
pub fn new_output_connection<C: Container, I>(&mut self, connection: I) -> (
105+
// OutputWrapper<G::Timestamp, CB, Tee<G::Timestamp, CB::Container>>,
106+
pushers::Output<G::Timestamp, C>,
107+
StreamCore<G, C>,
109108
)
110109
where
111110
I: IntoIterator<Item = (usize, Antichain<<G::Timestamp as Timestamp>::Summary>)> + Clone,
@@ -116,14 +115,16 @@ impl<G: Scope> OperatorBuilder<G> {
116115
let internal = Rc::new(RefCell::new(ChangeBatch::new()));
117116
self.internal.borrow_mut().push(Rc::clone(&internal));
118117

119-
let mut buffer = PushBuffer::new(PushCounter::new(tee));
120-
self.produced.push(Rc::clone(buffer.inner().produced()));
118+
let counter = PushCounter::new(tee);
119+
self.produced.push(Rc::clone(counter.produced()));
121120

122121
for (input, entry) in connection {
123122
self.summaries[input].borrow_mut().add_port(new_output, entry);
124123
}
125124

126-
(OutputWrapper::new(buffer, internal, new_output), stream)
125+
let output = pushers::Output::new(counter, internal, new_output);
126+
(output, stream)
127+
// (OutputWrapper::new(buffer, internal, new_output), stream)
127128
}
128129

129130
/// Creates an operator implementation from supplied logic constructor.
@@ -222,7 +223,7 @@ impl<G: Scope> OperatorBuilder<G> {
222223

223224
#[cfg(test)]
224225
mod tests {
225-
use crate::container::CapacityContainerBuilder;
226+
use crate::dataflow::operators::generic::OutputBuilder;
226227

227228
#[test]
228229
#[should_panic]
@@ -238,8 +239,10 @@ mod tests {
238239
let mut builder = OperatorBuilder::new("Failure".to_owned(), scope.clone());
239240

240241
// let mut input = builder.new_input(stream, Pipeline);
241-
let (mut output1, _stream1) = builder.new_output::<CapacityContainerBuilder<Vec<()>>>();
242-
let (mut output2, _stream2) = builder.new_output::<CapacityContainerBuilder<Vec<()>>>();
242+
let (output1, _stream1) = builder.new_output::<Vec<()>>();
243+
let (output2, _stream2) = builder.new_output::<Vec<()>>();
244+
let mut output1 = OutputBuilder::from(output1);
245+
let mut output2 = OutputBuilder::from(output2);
243246

244247
builder.build(move |capabilities| {
245248
move |_frontiers| {
@@ -268,8 +271,10 @@ mod tests {
268271
let mut builder = OperatorBuilder::new("Failure".to_owned(), scope.clone());
269272

270273
// let mut input = builder.new_input(stream, Pipeline);
271-
let (mut output1, _stream1) = builder.new_output::<CapacityContainerBuilder<Vec<()>>>();
272-
let (mut output2, _stream2) = builder.new_output::<CapacityContainerBuilder<Vec<()>>>();
274+
let (output1, _stream1) = builder.new_output::<Vec<()>>();
275+
let (output2, _stream2) = builder.new_output::<Vec<()>>();
276+
let mut output1 = OutputBuilder::from(output1);
277+
let mut output2 = OutputBuilder::from(output2);
273278

274279
builder.build(move |mut capabilities| {
275280
move |_frontiers| {

0 commit comments

Comments
 (0)