From 719f84ead8ac37f99dbca21ac38b3282f2e05a35 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 11 Sep 2026 16:19:43 -0400 Subject: [PATCH] Remove operators that read a time off the capability, and the flow-control and Result helpers Removed: delay/delay_batch/delay_total, count/accumulate, aggregate, state_machine, branch, branch_when, reclock, iterator_source (flow_controlled), and ResultStream. None has a user in differential-dataflow or Materialize. The first seven key their logic on the capability's time, which records do not carry; since #813 a message's stamp is a set of times, and since #817 these operators required a total order. branch_when in particular routes a whole container by its capability, which is wrong for any stream whose records carry their own times. The pingpong example carries a round counter in its data; the loop examples in the docs and the book terminate by data; the flow-control chapter assigns timestamps with a small unary operator and delayed capabilities. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_012k2GSwxmvD2LvckkoXi6GK --- mdbook/src/chapter_4/chapter_4_2.md | 5 +- mdbook/src/chapter_4/chapter_4_3.md | 14 +- timely/examples/flow_controlled.rs | 33 ---- timely/examples/pingpong.rs | 9 +- .../src/dataflow/operators/core/feedback.rs | 14 +- timely/src/dataflow/operators/core/mod.rs | 2 - timely/src/dataflow/operators/core/reclock.rs | 89 --------- timely/src/dataflow/operators/mod.rs | 2 - .../operators/vec/aggregation/aggregate.rs | 109 ---------- .../dataflow/operators/vec/aggregation/mod.rs | 18 -- .../vec/aggregation/state_machine.rs | 113 ----------- timely/src/dataflow/operators/vec/branch.rs | 138 ------------- timely/src/dataflow/operators/vec/count.rs | 74 ------- timely/src/dataflow/operators/vec/delay.rs | 155 --------------- .../dataflow/operators/vec/flow_controlled.rs | 124 ------------ timely/src/dataflow/operators/vec/mod.rs | 9 - timely/src/dataflow/operators/vec/result.rs | 187 ------------------ 17 files changed, 27 insertions(+), 1068 deletions(-) delete mode 100644 timely/examples/flow_controlled.rs delete mode 100644 timely/src/dataflow/operators/core/reclock.rs delete mode 100644 timely/src/dataflow/operators/vec/aggregation/aggregate.rs delete mode 100644 timely/src/dataflow/operators/vec/aggregation/mod.rs delete mode 100644 timely/src/dataflow/operators/vec/aggregation/state_machine.rs delete mode 100644 timely/src/dataflow/operators/vec/branch.rs delete mode 100644 timely/src/dataflow/operators/vec/count.rs delete mode 100644 timely/src/dataflow/operators/vec/delay.rs delete mode 100644 timely/src/dataflow/operators/vec/flow_controlled.rs delete mode 100644 timely/src/dataflow/operators/vec/result.rs diff --git a/mdbook/src/chapter_4/chapter_4_2.md b/mdbook/src/chapter_4/chapter_4_2.md index 36244e976..7a809ef51 100644 --- a/mdbook/src/chapter_4/chapter_4_2.md +++ b/mdbook/src/chapter_4/chapter_4_2.md @@ -14,7 +14,7 @@ We are going to check the [Collatz conjecture](https://en.wikipedia.org/wiki/Col extern crate timely; use timely::dataflow::operators::{Feedback, ToStream, Concat, Inspect, ConnectLoop}; -use timely::dataflow::operators::vec::{Map, Filter, BranchWhen}; +use timely::dataflow::operators::vec::{Map, Filter}; fn main() { timely::example(|scope| { @@ -29,13 +29,12 @@ fn main() { .map(|x| if x % 2 == 0 { x / 2 } else { 3 * x + 1 } ) .inspect(|x| println!("{:?}", x)) .filter(|x| *x != 1) - .branch_when(|t| t < &100).1 .connect_loop(handle); }); } ``` -This program first creates a loop variable, using the `feedback` method on scopes. This method comes from the `Feedback` extension trait in `dataflow::operators`, in case you can't find it. When we create a new loop variable, we have to tell timely dataflow by how much we should increment the timestamp each time around the loop. To be more specific, we have to give a path summary which often is just a number that tells us by how much to increment the timestamp. When we later connect the output of an operation back to this loop variable we can specify an upper bound on the number of iterations by using the `branch_when` method. +This program first creates a loop variable, using the `feedback` method on scopes. This method comes from the `Feedback` extension trait in `dataflow::operators`, in case you can't find it. When we create a new loop variable, we have to tell timely dataflow by how much we should increment the timestamp each time around the loop. To be more specific, we have to give a path summary which often is just a number that tells us by how much to increment the timestamp. The loop here ends on its own, once every number has reached one and been filtered out; a loop that might not end needs its own bound, for example a round counter carried in the data and a `filter` on it. We start with a stream of the numbers from one through nine, because we have to start somewhere. Our plan is to repeatedly apply the Collatz step, and then discard any numbers equal to one, but we want to apply this not only to our input but also to whatever comes back around our loop variable. So, the very first step is to `concat` our input stream with the feedback stream. Then we can apply the Collatz step, filter out the ones, and then connect the resulting stream as the definition of the feedback stream. diff --git a/mdbook/src/chapter_4/chapter_4_3.md b/mdbook/src/chapter_4/chapter_4_3.md index 438614cee..a4c87cee4 100644 --- a/mdbook/src/chapter_4/chapter_4_3.md +++ b/mdbook/src/chapter_4/chapter_4_3.md @@ -54,13 +54,13 @@ but without actually producing the 4,999,950,000 intermediate records all at onc One way to do this is to build a self-regulating dataflow, into which we can immediately dump all the records, but which will buffer records until it is certain that the work for prior records has drained. We will write this out in all the gory details, but these operators are certainly things that could be packaged up and reused. -The idea here is to take our stream of work, and to use the `delay` operator to assign new timestamps to the records. We will spread the work out so that each timestamp has at most (in this case) 100 numbers. We can write a `binary` operator that will buffer received records until their timestamp is "next", meaning all strictly prior work has drained from the dataflow fragment. How do we do this? We turn our previously unary operator into a binary operator that has a feedback edge connected to its second input. We use the frontier of that feedback input to control when we emit data. +The idea here is to take our stream of work, and to use a small operator to assign new timestamps to the records, by sending each under a capability delayed to its new time. We will spread the work out so that each timestamp has at most (in this case) 100 numbers. We can write a `binary` operator that will buffer received records until their timestamp is "next", meaning all strictly prior work has drained from the dataflow fragment. How do we do this? We turn our previously unary operator into a binary operator that has a feedback edge connected to its second input. We use the frontier of that feedback input to control when we emit data. ```rust,no_run extern crate timely; use timely::dataflow::operators::{Feedback, ToStream, Operator, ConnectLoop}; -use timely::dataflow::operators::vec::{Delay, Map, Filter}; +use timely::dataflow::operators::vec::{Map, Filter}; use timely::dataflow::channels::pact::Pipeline; fn main() { @@ -74,8 +74,16 @@ fn main() { // Produce all numbers less than each input number. (1 .. 100_000u64) .to_stream(scope) + .container::>() // Assign timestamps to records so that not much work is in each time. - .delay(|number, time| *number / 100 ) + .unary(Pipeline, "Delay", |_capability, _info| move |input, output| { + input.for_each_stamp(|cap, data| { + for number in data.flat_map(|d| d.drain(..)) { + output.session(&cap.delayed(&(number / 100), 0)).give(number); + } + }); + }) + .container::>() // Buffer records until all prior timestamps have completed. .binary_frontier(cycle, Pipeline, Pipeline, "Buffer", move |capability, info| { diff --git a/timely/examples/flow_controlled.rs b/timely/examples/flow_controlled.rs deleted file mode 100644 index dcfe69034..000000000 --- a/timely/examples/flow_controlled.rs +++ /dev/null @@ -1,33 +0,0 @@ -use timely::dataflow::operators::vec::flow_controlled::{iterator_source, IteratorSourceInput}; -use timely::dataflow::operators::{probe, Probe, Inspect}; - -fn main() { - timely::execute_from_args(std::env::args(), |worker| { - let mut input = (0u64..100000).peekable(); - worker.dataflow(|scope| { - let probe_handle = probe::Handle::new(); - let probe_handle_2 = probe_handle.clone(); - - iterator_source( - scope, - "Source", - move |prev_t| { - if let Some(first_x) = input.peek().cloned() { - let next_t = first_x / 100 * 100; - Some(IteratorSourceInput { - lower_bound: Default::default(), - data: vec![ - (next_t, - input.by_ref().take(10).map(|x| (/* "timestamp" */ x, x)).collect::>())], - target: *prev_t, - }) - } else { - None - } - }, - probe_handle_2) - .inspect(|d| eprintln!("{:?}", d)) - .probe_with(&probe_handle); - }); - }).unwrap(); -} diff --git a/timely/examples/pingpong.rs b/timely/examples/pingpong.rs index e674879a0..bffeccc7d 100644 --- a/timely/examples/pingpong.rs +++ b/timely/examples/pingpong.rs @@ -1,4 +1,4 @@ -use timely::dataflow::operators::{ToStream, Exchange, Feedback, Concat, ConnectLoop, vec::{Map, BranchWhen}}; +use timely::dataflow::operators::{ToStream, Exchange, Feedback, Concat, ConnectLoop, vec::{Map, Filter}}; fn main() { @@ -11,13 +11,16 @@ fn main() { let peers = worker.peers(); worker.dataflow::(move |scope| { let (helper, cycle) = scope.feedback(1); + // Each record counts the rounds it has made; all start at zero, and each + // makes exactly `iterations` trips around the loop. (0 .. elements) - .filter(move |&x| (x as usize) % peers == index) + .filter(move |&i| (i as usize) % peers == index) + .map(|_| 0u64) .to_stream(scope) .concat(cycle) .exchange(|&x| x) .map_in_place(|x| *x += 1) - .branch_when(move |t| t < &iterations).1 + .filter(move |&x| x <= iterations) .connect_loop(helper); }); }).unwrap(); diff --git a/timely/src/dataflow/operators/core/feedback.rs b/timely/src/dataflow/operators/core/feedback.rs index 6ce566ad2..d436e4418 100644 --- a/timely/src/dataflow/operators/core/feedback.rs +++ b/timely/src/dataflow/operators/core/feedback.rs @@ -22,16 +22,17 @@ pub trait Feedback<'scope, T: Timestamp> { /// ``` /// use timely::dataflow::Scope; /// use timely::dataflow::operators::{Feedback, ConnectLoop, ToStream, Concat, Inspect}; - /// use timely::dataflow::operators::vec::BranchWhen; + /// use timely::dataflow::operators::vec::{Map, Filter}; /// /// timely::example(|scope| { - /// // circulate 0..10 for 100 iterations. + /// // circulate 0..10, incrementing each, until each reaches 100. /// let (handle, cycle) = scope.feedback(1); /// (0..10).to_stream(scope) /// .container::>() /// .concat(cycle) /// .inspect(|x| println!("seen: {:?}", x)) - /// .branch_when(|t| t < &100).1 + /// .map(|x| x + 1) + /// .filter(|x| *x < 100) /// .connect_loop(handle); /// }); /// ``` @@ -94,16 +95,17 @@ pub trait ConnectLoop<'scope, T: Timestamp, C: Container> { /// ``` /// use timely::dataflow::Scope; /// use timely::dataflow::operators::{Feedback, ConnectLoop, ToStream, Concat, Inspect}; - /// use timely::dataflow::operators::vec::BranchWhen; + /// use timely::dataflow::operators::vec::{Map, Filter}; /// /// timely::example(|scope| { - /// // circulate 0..10 for 100 iterations. + /// // circulate 0..10, incrementing each, until each reaches 100. /// let (handle, cycle) = scope.feedback(1); /// (0..10).to_stream(scope) /// .container::>() /// .concat(cycle) /// .inspect(|x| println!("seen: {:?}", x)) - /// .branch_when(|t| t < &100).1 + /// .map(|x| x + 1) + /// .filter(|x| *x < 100) /// .connect_loop(handle); /// }); /// ``` diff --git a/timely/src/dataflow/operators/core/mod.rs b/timely/src/dataflow/operators/core/mod.rs index 521d4f05d..7ac86bc7b 100644 --- a/timely/src/dataflow/operators/core/mod.rs +++ b/timely/src/dataflow/operators/core/mod.rs @@ -14,7 +14,6 @@ pub mod ok_err; pub mod partition; pub mod probe; pub mod rc; -pub mod reclock; pub mod to_stream; pub mod unordered_input; @@ -31,5 +30,4 @@ pub use ok_err::OkErr; pub use partition::Partition; pub use probe::Probe; pub use to_stream::{ToStream, ToStreamBuilder}; -pub use reclock::Reclock; pub use unordered_input::{UnorderedInput, UnorderedHandle}; diff --git a/timely/src/dataflow/operators/core/reclock.rs b/timely/src/dataflow/operators/core/reclock.rs deleted file mode 100644 index 76a939d06..000000000 --- a/timely/src/dataflow/operators/core/reclock.rs +++ /dev/null @@ -1,89 +0,0 @@ -//! Extension methods for `Stream` based on record-by-record transformation. - -use crate::Container; -use crate::progress::Timestamp; -use crate::dataflow::Stream; -use crate::dataflow::channels::pact::Pipeline; -use crate::dataflow::operators::generic::operator::Operator; - -/// Extension trait for reclocking a stream. -pub trait Reclock<'scope, T: Timestamp> { - /// Delays records until an input is observed on the `clock` input. - /// - /// The source stream is buffered until a record is seen on the clock input, - /// at which point a notification is requested and all data with time less - /// or equal to the clock time are sent. This method does not ensure that all - /// workers receive the same clock records, which can be accomplished with - /// `broadcast`. - /// - /// # Examples - /// - /// ``` - /// use timely::dataflow::operators::{ToStream, Reclock, Capture}; - /// use timely::dataflow::operators::vec::{Delay, Map}; - /// use timely::dataflow::operators::capture::Extract; - /// - /// let captured = timely::example(|scope| { - /// - /// // produce data 0..10 at times 0..10. - /// let data = (0..10).to_stream(scope) - /// .delay(|x,t| *x); - /// - /// // product clock ticks at three times. - /// let clock = vec![3, 5, 8].into_iter() - /// .to_stream(scope) - /// .delay(|x,t| *x) - /// .map(|_| ()); - /// - /// // reclock the data. - /// data.reclock(clock) - /// .capture() - /// }); - /// - /// let extracted = captured.extract(); - /// assert_eq!(extracted.len(), 3); - /// assert_eq!(extracted[0], (3, vec![0,1,2,3])); - /// assert_eq!(extracted[1], (5, vec![4,5])); - /// assert_eq!(extracted[2], (8, vec![6,7,8])); - /// ``` - fn reclock(self, clock: Stream<'scope, T, TC>) -> Self; -} - -impl<'scope, T: Timestamp, C: Container> Reclock<'scope, T> for Stream<'scope, T, C> { - fn reclock(self, clock: Stream<'scope, T, TC>) -> Self { - - let mut stash = vec![]; - - self.binary_notify(clock, Pipeline, Pipeline, "Reclock", vec![], move |input1, input2, output, notificator| { - - // stash each data input with its stamp; a message with no capabilities - // could never be released, and is discarded. - input1.for_each_stamp(|cap, data| { - if !cap.stamp().is_empty() { - for data in data { - stash.push((cap.stamp().clone(), std::mem::take(data))); - } - } - }); - - // request notification at each clock time, to flush stash. - input2.for_each_stamp(|cap, _data| { - for cap in cap.retain_stamp(output.output_index()).iter() { - notificator.notify_at(cap.clone()); - } - }); - - // each time with complete stash can be flushed: data whose stamp has an - // element less or equal to the clock time may be sent at the clock time. - notificator.for_each(|cap,_,_| { - let mut session = output.session(&cap); - for &mut (ref stamp, ref mut data) in &mut stash { - if stamp.less_equal(cap.time()) { - session.give_container(data); - } - } - stash.retain(|x| !x.0.less_equal(cap.time())); - }); - }) - } -} diff --git a/timely/src/dataflow/operators/mod.rs b/timely/src/dataflow/operators/mod.rs index eefd779a3..99c1fad9c 100644 --- a/timely/src/dataflow/operators/mod.rs +++ b/timely/src/dataflow/operators/mod.rs @@ -17,7 +17,6 @@ pub use self::exchange::Exchange; pub use self::generic::Operator; pub use self::generic::{Notificator, FrontierNotificator}; -pub use self::reclock::Reclock; pub mod core; pub mod vec; @@ -36,7 +35,6 @@ pub use self::core::input::Input; pub mod generic; -pub use self::core::reclock; // keep "mint" module-private mod capability; diff --git a/timely/src/dataflow/operators/vec/aggregation/aggregate.rs b/timely/src/dataflow/operators/vec/aggregation/aggregate.rs deleted file mode 100644 index 3aea54e31..000000000 --- a/timely/src/dataflow/operators/vec/aggregation/aggregate.rs +++ /dev/null @@ -1,109 +0,0 @@ -//! General purpose intra-timestamp aggregation -use std::hash::Hash; -use std::collections::HashMap; - -use crate::ExchangeData; -use crate::order::TotalOrder; -use crate::progress::Timestamp; -use crate::dataflow::StreamVec; -use crate::dataflow::operators::generic::operator::Operator; -use crate::dataflow::channels::pact::Exchange; - -/// Generic intra-timestamp aggregation -/// -/// Extension method supporting aggregation of keyed data within timestamp. -/// For inter-timestamp aggregation, consider `StateMachine`. -pub trait Aggregate<'scope, T: Timestamp, K: ExchangeData+Hash, V: ExchangeData> { - /// Aggregates data of the form `(key, val)`, using user-supplied logic. - /// - /// The `aggregate` method is implemented for streams of `(K, V)` data, - /// and takes functions `fold`, `emit`, and `hash`; used to combine new `V` - /// data with existing `D` state, to produce `R` output from `D` state, and - /// to route `K` keys, respectively. - /// - /// Aggregation happens within each time, and results are produced once the - /// time is complete. - /// - /// # Examples - /// ``` - /// use timely::dataflow::operators::{ToStream, Inspect}; - /// use timely::dataflow::operators::vec::{Map, aggregation::Aggregate}; - /// - /// timely::example(|scope| { - /// - /// (0..10).to_stream(scope) - /// .map(|x| (x % 2, x)) - /// .aggregate( - /// |_key, val, agg| { *agg += val; }, - /// |key, agg: i32| (key, agg), - /// |key| *key as u64 - /// ) - /// .inspect(|x| assert!(*x == (0, 20) || *x == (1, 25))); - /// }); - /// ``` - /// - /// By changing the type of the aggregate value, one can accumulate into different types. - /// Here we accumulate the data into a `Vec` and report its length (which we could - /// obviously do more efficiently; imagine we were doing a hash instead). - /// - /// ``` - /// use timely::dataflow::operators::{ToStream, Inspect}; - /// use timely::dataflow::operators::vec::{Map, aggregation::Aggregate}; - /// - /// timely::example(|scope| { - /// - /// (0..10) - /// .to_stream(scope) - /// .map(|x| (x % 2, x)) - /// .aggregate::<_,Vec,_,_,_>( - /// |_key, val, agg| { agg.push(val); }, - /// |key, agg| (key, agg.len()), - /// |key| *key as u64 - /// ) - /// .inspect(|x| assert!(*x == (0, 5) || *x == (1, 5))); - /// }); - /// ``` - fn aggregateR+'static, H: Fn(&K)->u64+'static>( - self, - fold: F, - emit: E, - hash: H) -> StreamVec<'scope, T, R> where T: Eq; -} - -impl<'scope, T: Timestamp + TotalOrder + Hash, K: ExchangeData+Clone+Hash+Eq, V: ExchangeData> Aggregate<'scope, T, K, V> for StreamVec<'scope, T, (K, V)> { - - fn aggregateR+'static, H: Fn(&K)->u64+'static>( - self, - fold: F, - emit: E, - hash: H) -> StreamVec<'scope, T, R> where T: Eq { - - let mut aggregates = HashMap::new(); - self.unary_notify(Exchange::new(move |(k, _)| hash(k)), "Aggregate", vec![], move |input, output, notificator| { - - // read each input, fold into aggregates - input.for_each_stamp(|cap, data| { - // A message with no time makes no progress claims, and has no time to aggregate at. - if let Some(cap) = cap.retain_least(output.output_index()) { - let agg_time = aggregates.entry(cap.time().clone()).or_insert_with(HashMap::new); - for (key, val) in data.flat_map(|d| d.drain(..)) { - let agg = agg_time.entry(key.clone()).or_insert_with(Default::default); - fold(&key, val, agg); - } - notificator.notify_at(cap); - } - }); - - // pop completed aggregates, send along whatever - notificator.for_each(|cap,_,_| { - if let Some(aggs) = aggregates.remove(cap.time()) { - let mut session = output.session(&cap); - for (key, agg) in aggs { - session.give(emit(key, agg)); - } - } - }); - }) - - } -} diff --git a/timely/src/dataflow/operators/vec/aggregation/mod.rs b/timely/src/dataflow/operators/vec/aggregation/mod.rs deleted file mode 100644 index 407cf1eb3..000000000 --- a/timely/src/dataflow/operators/vec/aggregation/mod.rs +++ /dev/null @@ -1,18 +0,0 @@ -//! Aggregation operators of various flavors -//! -//! Two traits, `Aggregate` and `StateMachine`, which support the accumulation of streamed information. -//! -//! `Aggregate` accumulates records within times, and releases the accumulations once the time is complete. -//! -//! `StateMachine` responds to a sequence of keyed events, maintaining and updating a state for each key. -//! The user logic may produce output records for each transition, and optionally de-register the state to -//! clean up when appropriate. -//! -//! The two methods are often combined, using first `Aggregate` to reduce the volume of information, and then -//! `StateMachine` to track an accumulation across timestamps. - -pub use self::aggregate::Aggregate; -pub use self::state_machine::StateMachine; - -pub mod state_machine; -pub mod aggregate; diff --git a/timely/src/dataflow/operators/vec/aggregation/state_machine.rs b/timely/src/dataflow/operators/vec/aggregation/state_machine.rs deleted file mode 100644 index 288725894..000000000 --- a/timely/src/dataflow/operators/vec/aggregation/state_machine.rs +++ /dev/null @@ -1,113 +0,0 @@ -//! General purpose state transition operator. -use std::hash::Hash; -use std::collections::HashMap; - -use crate::ExchangeData; -use crate::order::TotalOrder; -use crate::progress::Timestamp; -use crate::dataflow::StreamVec; -use crate::dataflow::operators::generic::operator::Operator; -use crate::dataflow::channels::pact::Exchange; - -/// Provides the `state_machine` method. -/// -/// Generic state-transition machinery: each key has a state, and receives a sequence of events. -/// Events are applied in time-order, but no other promises are made. Each state transition can -/// produce output, which is sent. -/// -/// `state_machine` will buffer inputs if earlier inputs may still arrive. it will directly apply -/// updates for the current time reflected in the notificator, though. In the case of partially -/// ordered times, the only guarantee is that updates are not applied out of order, not that there -/// is some total order on times respecting the total order (updates may be interleaved). -pub trait StateMachine<'scope, T: Timestamp, K: ExchangeData+Hash+Eq, V: ExchangeData> { - /// Tracks a state for each presented key, using user-supplied state transition logic. - /// - /// The transition logic `fold` may mutate the state, and produce both output records and - /// a `bool` indicating that it is appropriate to deregister the state, cleaning up once - /// the state is no longer helpful. - /// - /// # Examples - /// ``` - /// use timely::dataflow::operators::{ToStream, Inspect}; - /// use timely::dataflow::operators::vec::{Map, aggregation::StateMachine}; - /// - /// timely::example(|scope| { - /// - /// // these results happen to be right, but aren't guaranteed. - /// // the system is at liberty to re-order within a timestamp. - /// let result = vec![(0,0), (0,2), (0,6), (0,12), (0,20), - /// (1,1), (1,4), (1,9), (1,16), (1,25)]; - /// - /// (0..10).to_stream(scope) - /// .map(|x| (x % 2, x)) - /// .state_machine( - /// |_key, val, agg| { *agg += val; (false, Some((*_key, *agg))) }, - /// |key| *key as u64 - /// ) - /// .inspect(move |x| assert!(result.contains(x))); - /// }); - /// ``` - fn state_machine< - R: 'static, // output type - D: Default+'static, // per-key state (data) - I: IntoIterator, // type of output iterator - F: Fn(&K, V, &mut D)->(bool, I)+'static, // state update logic - H: Fn(&K)->u64+'static, // "hash" function for keys - >(self, fold: F, hash: H) -> StreamVec<'scope, T, R> where T : Hash+Eq ; -} - -impl<'scope, T: Timestamp + TotalOrder, K: ExchangeData+Hash+Eq+Clone, V: ExchangeData> StateMachine<'scope, T, K, V> for StreamVec<'scope, T, (K, V)> { - fn state_machine< - R: 'static, // output type - D: Default+'static, // per-key state (data) - I: IntoIterator, // type of output iterator - F: Fn(&K, V, &mut D)->(bool, I)+'static, // state update logic - H: Fn(&K)->u64+'static, // "hash" function for keys - >(self, fold: F, hash: H) -> StreamVec<'scope, T, R> where T : Hash+Eq { - - let mut pending: HashMap<_, Vec<(K, V)>> = HashMap::new(); // times -> (keys -> state) - let mut states = HashMap::new(); // keys -> state - - self.unary_notify(Exchange::new(move |(k, _)| hash(k)), "StateMachine", vec![], move |input, output, notificator| { - - // go through each time with data, process each (key, val) pair. - notificator.for_each(|cap,_,_| { - if let Some(pend) = pending.remove(cap.time()) { - let mut session = output.session(&cap); - for (key, val) in pend { - let (remove, output) = { - let state = states.entry(key.clone()).or_insert_with(Default::default); - fold(&key, val, state) - }; - if remove { states.remove(&key); } - session.give_iterator(output.into_iter()); - } - } - }); - - // stash each input and request a notification when ready - input.for_each_stamp(|cap, data| { - // A message with no time makes no progress claims, and has no place in time order. - if let Some(cap) = cap.retain_least(output.output_index()) { - // stash if not time yet - if notificator.frontier(0).less_than(cap.time()) { - for data in data { pending.entry(cap.time().clone()).or_insert_with(Vec::new).append(data); } - notificator.notify_at(cap); - } - else { - // else we can process immediately - let mut session = output.session(&cap); - for (key, val) in data.flat_map(|d| d.drain(..)) { - let (remove, output) = { - let state = states.entry(key.clone()).or_insert_with(Default::default); - fold(&key, val, state) - }; - if remove { states.remove(&key); } - session.give_iterator(output.into_iter()); - } - } - } - }); - }) - } -} diff --git a/timely/src/dataflow/operators/vec/branch.rs b/timely/src/dataflow/operators/vec/branch.rs deleted file mode 100644 index f6c5220dd..000000000 --- a/timely/src/dataflow/operators/vec/branch.rs +++ /dev/null @@ -1,138 +0,0 @@ -//! Operators that separate one stream into two streams based on some condition - -use crate::dataflow::channels::pact::Pipeline; -use crate::order::TotalOrder; -use crate::progress::Timestamp; -use crate::dataflow::operators::generic::OutputBuilder; -use crate::dataflow::operators::generic::builder_rc::OperatorBuilder; -use crate::dataflow::{StreamVec, Stream}; -use crate::Container; - -/// Extension trait for `StreamVec`. -pub trait Branch : Sized { - /// Takes one input stream and splits it into two output streams. - /// For each record, the supplied closure is called with a reference to - /// the data and its time. If it returns `true`, the record will be sent - /// to the second returned stream, otherwise it will be sent to the first. - /// - /// If the result of the closure only depends on the time, not the data, - /// `branch_when` should be used instead. - /// - /// # Examples - /// ``` - /// use timely::dataflow::operators::{ToStream, Inspect, vec::Branch}; - /// - /// timely::example(|scope| { - /// let (odd, even) = (0..10) - /// .to_stream(scope) - /// .branch(|_time, x| *x % 2 == 0); - /// - /// even.inspect(|x| println!("even numbers: {:?}", x)); - /// odd.inspect(|x| println!("odd numbers: {:?}", x)); - /// }); - /// ``` - fn branch(self, condition: impl Fn(&T, &D) -> bool + 'static) -> (Self, Self); -} - -impl<'scope, T: Timestamp + TotalOrder, D: 'static> Branch for StreamVec<'scope, T, D> { - fn branch(self, condition: impl Fn(&T, &D) -> bool + 'static) -> (Self, Self) { - let mut builder = OperatorBuilder::new("Branch".to_owned(), self.scope()); - - let mut input = builder.new_input(self, Pipeline); - builder.set_notify_for(0, crate::progress::operate::FrontierInterest::Never); - let (output1, stream1) = builder.new_output(); - let (output2, stream2) = builder.new_output(); - - let mut output1 = OutputBuilder::from(output1); - let mut output2 = OutputBuilder::from(output2); - - builder.build(move |_| { - move |_frontiers| { - let mut output1_handle = output1.activate(); - let mut output2_handle = output2.activate(); - - input.for_each_stamp(|cap, data| { - // A message with no time makes no progress claims, and cannot be routed by time. - if let Some(t) = cap.least() { - let mut out1 = output1_handle.session(&cap); - let mut out2 = output2_handle.session(&cap); - for datum in data.flat_map(|d| d.drain(..)) { - if condition(t, &datum) { - out2.give(datum); - } else { - out1.give(datum); - } - } - } - }); - } - }); - - (stream1, stream2) - } -} - -/// Extension trait for `Stream`. -pub trait BranchWhen: Sized { - /// Takes one input stream and splits it into two output streams. - /// For each time, the supplied closure is called. If it returns `true`, - /// the records for that will be sent to the second returned stream, otherwise - /// they will be sent to the first. - /// - /// The closure sees the least time of each message's stamp, and so this operator - /// exists only for totally ordered timestamps; deciding routing by the value of a - /// timestamp has no meaning for a message stamped by several incomparable times. - /// - /// # Examples - /// ``` - /// use timely::dataflow::operators::{ToStream, Inspect}; - /// use timely::dataflow::operators::vec::{BranchWhen, Delay}; - /// - /// timely::example(|scope| { - /// let (before_five, after_five) = (0..10) - /// .to_stream(scope) - /// .container::>() - /// .delay(|x,t| *x) // data 0..10 at time 0..10 - /// .branch_when(|time| time >= &5); - /// - /// before_five.inspect(|x| println!("Times 0-4: {:?}", x)); - /// after_five.inspect(|x| println!("Times 5 and later: {:?}", x)); - /// }); - /// ``` - fn branch_when(self, condition: impl Fn(&T) -> bool + 'static) -> (Self, Self); -} - -impl<'scope, T: Timestamp + TotalOrder, C: Container> BranchWhen for Stream<'scope, T, C> { - fn branch_when(self, condition: impl Fn(&T) -> bool + 'static) -> (Self, Self) { - let mut builder = OperatorBuilder::new("Branch".to_owned(), self.scope()); - - let mut input = builder.new_input(self, Pipeline); - builder.set_notify_for(0, crate::progress::operate::FrontierInterest::Never); - let (output1, stream1) = builder.new_output(); - let (output2, stream2) = builder.new_output(); - - let mut output1 = OutputBuilder::from(output1); - let mut output2 = OutputBuilder::from(output2); - - builder.build(move |_| { - - move |_frontiers| { - let mut output1_handle = output1.activate(); - let mut output2_handle = output2.activate(); - - input.for_each_stamp(|cap, data| { - if let Some(t) = cap.least() { - let mut out = if condition(t) { - output2_handle.session(&cap) - } else { - output1_handle.session(&cap) - }; - out.give_containers(data); - } - }); - } - }); - - (stream1, stream2) - } -} diff --git a/timely/src/dataflow/operators/vec/count.rs b/timely/src/dataflow/operators/vec/count.rs deleted file mode 100644 index 3fb8a086f..000000000 --- a/timely/src/dataflow/operators/vec/count.rs +++ /dev/null @@ -1,74 +0,0 @@ -//! Counts the number of records at each time. -use std::collections::HashMap; - -use crate::dataflow::channels::pact::Pipeline; -use crate::order::TotalOrder; -use crate::progress::Timestamp; -use crate::dataflow::StreamVec; -use crate::dataflow::operators::generic::operator::Operator; - -/// Accumulates records within a timestamp. -pub trait Accumulate<'scope, T: Timestamp, D: 'static> : Sized { - /// Accumulates records within a timestamp. - /// - /// # Examples - /// - /// ``` - /// use timely::dataflow::operators::{ToStream, Capture}; - /// use timely::dataflow::operators::vec::count::Accumulate; - /// use timely::dataflow::operators::capture::Extract; - /// - /// let captured = timely::example(|scope| { - /// (0..10).to_stream(scope) - /// .accumulate(0, |sum, data| { for &x in data.iter() { *sum += x; } }) - /// .capture() - /// }); - /// - /// let extracted = captured.extract(); - /// assert_eq!(extracted, vec![(0, vec![45])]); - /// ``` - fn accumulate(self, default: A, logic: impl Fn(&mut A, &mut Vec)+'static) -> StreamVec<'scope, T, A>; - /// Counts the number of records observed at each time. - /// - /// # Examples - /// - /// ``` - /// use timely::dataflow::operators::{ToStream, Capture}; - /// use timely::dataflow::operators::vec::count::Accumulate; - /// use timely::dataflow::operators::capture::Extract; - /// - /// let captured = timely::example(|scope| { - /// (0..10).to_stream(scope) - /// .count() - /// .capture() - /// }); - /// - /// let extracted = captured.extract(); - /// assert_eq!(extracted, vec![(0, vec![10])]); - /// ``` - fn count(self) -> StreamVec<'scope, T, usize> { self.accumulate(0, |sum, data| *sum += data.len()) } -} - -impl<'scope, T: Timestamp + TotalOrder + ::std::hash::Hash, D: 'static> Accumulate<'scope, T, D> for StreamVec<'scope, T, D> { - fn accumulate(self, default: A, logic: impl Fn(&mut A, &mut Vec)+'static) -> StreamVec<'scope, T, A> { - - let mut accums = HashMap::new(); - self.unary_notify(Pipeline, "Accumulate", vec![], move |input, output, notificator| { - input.for_each_stamp(|cap, data| { - // A message with no time makes no progress claims, and has no time to accumulate at. - if let Some(cap) = cap.retain_least(output.output_index()) { - for data in data { - logic(accums.entry(cap.time().clone()).or_insert_with(|| default.clone()), data); - } - notificator.notify_at(cap); - } - }); - - notificator.for_each(|cap,_,_| { - if let Some(accum) = accums.remove(cap.time()) { - output.session(&cap).give(accum); - } - }); - }) - } -} diff --git a/timely/src/dataflow/operators/vec/delay.rs b/timely/src/dataflow/operators/vec/delay.rs deleted file mode 100644 index 7500779de..000000000 --- a/timely/src/dataflow/operators/vec/delay.rs +++ /dev/null @@ -1,155 +0,0 @@ -//! Operators acting on timestamps to logically delay records - -use std::collections::HashMap; - -use crate::order::TotalOrder; -use crate::progress::Timestamp; -use crate::dataflow::channels::pact::Pipeline; -use crate::dataflow::StreamVec; -use crate::dataflow::operators::generic::operator::Operator; - -/// Methods to advance the timestamps of records or batches of records. -pub trait Delay { - - /// Advances the timestamp of records using a supplied function. - /// - /// The function *must* advance the timestamp; the operator will test that the - /// new timestamp is greater or equal to the old timestamp, and will assert if - /// it is not. - /// - /// # Examples - /// - /// The following example takes the sequence `0..10` at time `0` - /// and delays each element `i` to time `i`. - /// - /// ``` - /// use timely::dataflow::operators::{ToStream, Operator}; - /// use timely::dataflow::operators::vec::Delay; - /// use timely::dataflow::channels::pact::Pipeline; - /// - /// timely::example(|scope| { - /// (0..10).to_stream(scope) - /// .delay(|data, time| *data) - /// .sink(Pipeline, "example", |(input, frontier)| { - /// input.for_each_stamp(|cap, data| { - /// println!("data at time: {:?}", cap); - /// }); - /// }); - /// }); - /// ``` - fn delayT+'static>(self, func: L) -> Self; - - /// Advances the timestamp of records using a supplied function. - /// - /// This method is a specialization of `delay` for when the timestamp is totally - /// ordered. In this case, we can use a priority queue rather than an unsorted - /// list to manage the potentially available timestamps. - /// - /// # Examples - /// - /// The following example takes the sequence `0..10` at time `0` - /// and delays each element `i` to time `i`. - /// - /// ``` - /// use timely::dataflow::operators::{ToStream, Operator}; - /// use timely::dataflow::operators::vec::Delay; - /// use timely::dataflow::channels::pact::Pipeline; - /// - /// timely::example(|scope| { - /// (0..10).to_stream(scope) - /// .delay(|data, time| *data) - /// .sink(Pipeline, "example", |(input, frontier)| { - /// input.for_each_stamp(|cap, data| { - /// println!("data at time: {:?}", cap); - /// }); - /// }); - /// }); - /// ``` - fn delay_totalT+'static>(self, func: L) -> Self - where T: TotalOrder; - - /// Advances the timestamp of batches of records using a supplied function. - /// - /// The operator will test that the new timestamp is greater or equal to the - /// old timestamp, and will assert if it is not. The batch version does not - /// consult the data, and may only view the timestamp itself. - /// - /// # Examples - /// - /// The following example takes the sequence `0..10` at time `0` - /// and delays each batch (there is just one) to time `1`. - /// - /// ``` - /// use timely::dataflow::operators::{ToStream, Operator}; - /// use timely::dataflow::operators::vec::Delay; - /// use timely::dataflow::channels::pact::Pipeline; - /// - /// timely::example(|scope| { - /// (0..10).to_stream(scope) - /// .delay_batch(|time| time + 1) - /// .sink(Pipeline, "example", |(input, frontier)| { - /// input.for_each_stamp(|cap, data| { - /// println!("data at time: {:?}", cap); - /// }); - /// }); - /// }); - /// ``` - fn delay_batchT+'static>(self, func: L) -> Self; -} - -impl Delay for StreamVec<'_, T, D> { - fn delayT+'static>(self, mut func: L) -> Self { - let mut elements = HashMap::new(); - self.unary_notify(Pipeline, "Delay", vec![], move |input, output, notificator| { - input.for_each_stamp(|cap, data| { - // A message with no time makes no progress claims, and has no time to delay from. - if let Some(old_time) = cap.least() { - for datum in data.flat_map(|d| d.drain(..)) { - let new_time = func(&datum, old_time); - assert!(old_time.less_equal(&new_time)); - elements.entry(new_time.clone()) - .or_insert_with(|| { notificator.notify_at(cap.delayed(&new_time, output.output_index())); Vec::new() }) - .push(datum); - } - } - }); - - // for each available notification, send corresponding set - notificator.for_each(|cap,_,_| { - if let Some(mut data) = elements.remove(cap.time()) { - output.session(&cap).give_iterator(data.drain(..)); - } - }); - }) - } - - fn delay_totalT+'static>(self, func: L) -> Self - where T: TotalOrder - { - self.delay(func) - } - - fn delay_batchT+'static>(self, mut func: L) -> Self { - let mut elements = HashMap::new(); - self.unary_notify(Pipeline, "Delay", vec![], move |input, output, notificator| { - input.for_each_stamp(|cap, data| { - if let Some(old_time) = cap.least() { - let new_time = func(old_time); - assert!(old_time.less_equal(&new_time)); - elements.entry(new_time.clone()) - .or_insert_with(|| { notificator.notify_at(cap.delayed(&new_time, output.output_index())); Vec::new() }) - .extend(data.map(std::mem::take)); - } - }); - - // for each available notification, send corresponding set - notificator.for_each(|cap,_,_| { - if let Some(mut datas) = elements.remove(cap.time()) { - for mut data in datas.drain(..) { - output.session(&cap).give_container(&mut data); - } - } - }); - }) - } -} diff --git a/timely/src/dataflow/operators/vec/flow_controlled.rs b/timely/src/dataflow/operators/vec/flow_controlled.rs deleted file mode 100644 index 7897ec358..000000000 --- a/timely/src/dataflow/operators/vec/flow_controlled.rs +++ /dev/null @@ -1,124 +0,0 @@ -//! Methods to construct flow-controlled sources. - -use crate::order::TotalOrder; -use crate::progress::timestamp::Timestamp; -use crate::dataflow::operators::generic::operator::source; -use crate::dataflow::operators::probe::Handle; -use crate::dataflow::{StreamVec, Scope}; - -/// Output of the input reading function for iterator_source. -pub struct IteratorSourceInput, I: IntoIterator> { - /// Lower bound on timestamps that can be emitted by this input in the future. - pub lower_bound: T, - /// Any `T: IntoIterator` of new input data in the form (time, data): time must be - /// monotonically increasing. - pub data: I, - /// A timestamp that represents the frontier that the probe should have - /// reached before the function is invoked again to ingest additional input. - pub target: T, -} - -/// Construct a source that repeatedly calls the provided function to ingest input. -/// -/// The function can return `None` to signal the end of the input. -/// Otherwise, it should return a [`IteratorSourceInput`], where: -/// * `lower_bound` is a lower bound on timestamps that can be emitted by this input in the future, -/// `Default::default()` can be used if this isn't needed (the source will assume that -/// the timestamps in `data` are monotonically increasing and will release capabilities -/// accordingly); -/// * `data` is any `T: IntoIterator` of new input data in the form (time, data): time must be -/// monotonically increasing; -/// * `target` is a timestamp that represents the frontier that the probe should have -/// reached before the function is invoked again to ingest additional input. -/// The function will receive the current lower bound of timestamps that can be inserted, -/// `lower_bound`. -/// -/// # Example -/// ```rust -/// use timely::dataflow::operators::vec::flow_controlled::{iterator_source, IteratorSourceInput}; -/// use timely::dataflow::operators::{probe, Probe, Inspect}; -/// -/// timely::execute_from_args(std::env::args(), |worker| { -/// let mut input = (0u64..100000).peekable(); -/// worker.dataflow(|scope| { -/// let mut probe_handle = probe::Handle::new(); -/// let probe_handle_2 = probe_handle.clone(); -/// -/// let mut next_t: u64 = 0; -/// iterator_source( -/// scope, -/// "Source", -/// move |prev_t| { -/// if let Some(first_x) = input.peek().cloned() { -/// next_t = first_x / 100 * 100; -/// Some(IteratorSourceInput { -/// lower_bound: Default::default(), -/// data: vec![ -/// (next_t, -/// input.by_ref().take(10).map(|x| (/* "timestamp" */ x, x)).collect::>())], -/// target: *prev_t, -/// }) -/// } else { -/// None -/// } -/// }, -/// probe_handle_2) -/// .inspect(|d| eprintln!("{:?}", d)) -/// .probe_with(&mut probe_handle); -/// }); -/// }).unwrap(); -/// ``` -pub fn iterator_source< - 'scope, - T: Timestamp, - D: 'static, - DI: IntoIterator, - I: IntoIterator, - F: FnMut(&T)->Option>+'static>( - scope: Scope<'scope, T>, - name: &str, - mut input_f: F, - probe: Handle, - ) -> StreamVec<'scope, T, D> where T: TotalOrder { - - let mut target = T::minimum(); - source(scope, name, |cap, info| { - let mut cap = Some(cap); - let activator = scope.activator_for(info.address); - move |output| { - cap = cap.take().and_then(|mut cap| { - loop { - if !probe.less_than(&target) { - if let Some(IteratorSourceInput { - lower_bound, - data, - target: new_target, - }) = input_f(cap.time()) { - target = new_target; - let mut has_data = false; - for (t, ds) in data.into_iter() { - cap = if cap.time() != &t { cap.delayed(&t) } else { cap }; - let mut session = output.session(&cap); - session.give_iterator(ds.into_iter()); - has_data = true; - } - - cap = if cap.time().less_than(&lower_bound) { cap.delayed(&lower_bound) } else { cap }; - if !has_data { - break Some(cap); - } - } else { - break None; - } - } else { - break Some(cap); - } - } - }); - - if cap.is_some() { - activator.activate(); - } - } - }) -} diff --git a/timely/src/dataflow/operators/vec/mod.rs b/timely/src/dataflow/operators/vec/mod.rs index d3f1920a4..196f393dd 100644 --- a/timely/src/dataflow/operators/vec/mod.rs +++ b/timely/src/dataflow/operators/vec/mod.rs @@ -9,26 +9,17 @@ //! around resource management (allocation and deallocation, across threads). pub mod input; -pub mod flow_controlled; pub mod unordered_input; pub mod partition; pub mod map; pub mod filter; -pub mod delay; pub mod broadcast; pub mod to_stream; -pub mod branch; -pub mod result; -pub mod aggregation; -pub mod count; pub use self::input::Input; pub use self::unordered_input::UnorderedInput; pub use self::partition::Partition; pub use self::map::Map; pub use self::filter::Filter; -pub use self::delay::Delay; pub use self::broadcast::Broadcast; -pub use self::branch::{Branch, BranchWhen}; -pub use self::result::ResultStream; pub use self::to_stream::ToStream; diff --git a/timely/src/dataflow/operators/vec/result.rs b/timely/src/dataflow/operators/vec/result.rs deleted file mode 100644 index 429e80c10..000000000 --- a/timely/src/dataflow/operators/vec/result.rs +++ /dev/null @@ -1,187 +0,0 @@ -//! Extension methods for `StreamVec` containing `Result`s. - -use crate::dataflow::operators::vec::Map; -use crate::progress::Timestamp; -use crate::dataflow::StreamVec; - -/// Extension trait for `StreamVec`. -pub trait ResultStream<'scope, T: Timestamp, D: 'static, E: 'static> { - /// Returns a new instance of `self` containing only `ok` records. - /// - /// # Examples - /// ``` - /// use timely::dataflow::operators::{ToStream, Inspect, vec::ResultStream}; - /// - /// timely::example(|scope| { - /// vec![Ok(0), Err(())].to_stream(scope) - /// .ok() - /// .inspect(|x| println!("seen: {:?}", x)); - /// }); - /// ``` - fn ok(self) -> StreamVec<'scope, T, D>; - - /// Returns a new instance of `self` containing only `err` records. - /// - /// # Examples - /// ``` - /// use timely::dataflow::operators::{ToStream, Inspect, vec::ResultStream}; - /// - /// timely::example(|scope| { - /// vec![Ok(0), Err(())].to_stream(scope) - /// .err() - /// .inspect(|x| println!("seen: {:?}", x)); - /// }); - /// ``` - fn err(self) -> StreamVec<'scope, T, E>; - - /// Returns a new instance of `self` applying `logic` on all `Ok` records. - /// - /// # Examples - /// ``` - /// use timely::dataflow::operators::{ToStream, Inspect, vec::ResultStream}; - /// - /// timely::example(|scope| { - /// vec![Ok(0), Err(())].to_stream(scope) - /// .map_ok(|x| x + 1) - /// .inspect(|x| println!("seen: {:?}", x)); - /// }); - /// ``` - fn map_ok D2 + 'static>(self, logic: L) -> StreamVec<'scope, T, Result>; - - /// Returns a new instance of `self` applying `logic` on all `Err` records. - /// - /// # Examples - /// ``` - /// use timely::dataflow::operators::{ToStream, Inspect, vec::ResultStream}; - /// - /// timely::example(|scope| { - /// vec![Ok(0), Err(())].to_stream(scope) - /// .map_err(|_| 1) - /// .inspect(|x| println!("seen: {:?}", x)); - /// }); - /// ``` - fn map_err E2 + 'static>(self, logic: L) -> StreamVec<'scope, T, Result>; - - /// Returns a new instance of `self` applying `logic` on all `Ok` records, passes through `Err` - /// records. - /// - /// # Examples - /// ``` - /// use timely::dataflow::operators::{ToStream, Inspect, vec::ResultStream}; - /// - /// timely::example(|scope| { - /// vec![Ok(0), Err(())].to_stream(scope) - /// .and_then(|x| Ok(1 + 1)) - /// .inspect(|x| println!("seen: {:?}", x)); - /// }); - /// ``` - fn and_then Result + 'static>( - self, - logic: L, - ) -> StreamVec<'scope, T, Result>; - - /// Returns a new instance of `self` applying `logic` on all `Ok` records. - /// - /// # Examples - /// ``` - /// use timely::dataflow::operators::{ToStream, Inspect, vec::ResultStream}; - /// - /// timely::example(|scope| { - /// vec![Ok(1), Err(())].to_stream(scope) - /// .unwrap_or_else(|_| 0) - /// .inspect(|x| println!("seen: {:?}", x)); - /// }); - /// ``` - fn unwrap_or_else D + 'static>(self, logic: L) -> StreamVec<'scope, T, D>; -} - -impl<'scope, T: Timestamp, D: 'static, E: 'static> ResultStream<'scope, T, D, E> for StreamVec<'scope, T, Result> { - fn ok(self) -> StreamVec<'scope, T, D> { - self.flat_map(Result::ok) - } - - fn err(self) -> StreamVec<'scope, T, E> { - self.flat_map(Result::err) - } - - fn map_ok D2 + 'static>(self, mut logic: L) -> StreamVec<'scope, T, Result> { - self.map(move |r| r.map(&mut logic)) - } - - fn map_err E2 + 'static>(self, mut logic: L) -> StreamVec<'scope, T, Result> { - self.map(move |r| r.map_err(&mut logic)) - } - - fn and_then Result + 'static>(self, mut logic: L) -> StreamVec<'scope, T, Result> { - self.map(move |r| r.and_then(&mut logic)) - } - - fn unwrap_or_else D + 'static>(self, mut logic: L) -> StreamVec<'scope, T, D> { - self.map(move |r| r.unwrap_or_else(&mut logic)) - } -} - -#[cfg(test)] -mod tests { - use crate::dataflow::operators::{vec::{ToStream, ResultStream}, Capture, capture::Extract}; - - #[test] - fn test_ok() { - let output = crate::example(|scope| { - vec![Ok(0), Err(())].to_stream(scope) - .ok() - .capture() - }); - assert_eq!(output.extract()[0].1, vec![0]); - } - - #[test] - fn test_err() { - let output = crate::example(|scope| { - vec![Ok(0), Err(())].to_stream(scope) - .err() - .capture() - }); - assert_eq!(output.extract()[0].1, vec![()]); - } - - #[test] - fn test_map_ok() { - let output = crate::example(|scope| { - vec![Ok(0), Err(())].to_stream(scope) - .map_ok(|_| 10) - .capture() - }); - assert_eq!(output.extract()[0].1, vec![Ok(10), Err(())]); - } - - #[test] - fn test_map_err() { - let output = crate::example(|scope| { - vec![Ok(0), Err(())].to_stream(scope) - .map_err(|_| 10) - .capture() - }); - assert_eq!(output.extract()[0].1, vec![Ok(0), Err(10)]); - } - - #[test] - fn test_and_then() { - let output = crate::example(|scope| { - vec![Ok(0), Err(())].to_stream(scope) - .and_then(|_| Ok(1)) - .capture() - }); - assert_eq!(output.extract()[0].1, vec![Ok(1), Err(())]); - } - - #[test] - fn test_unwrap_or_else() { - let output = crate::example(|scope| { - vec![Ok(0), Err(())].to_stream(scope) - .unwrap_or_else(|_| 10) - .capture() - }); - assert_eq!(output.extract()[0].1, vec![0, 10]); - } -}