timely/dataflow/operators/generic/

operator.rs

//! Methods to construct generic streaming and blocking unary operators.

use crate::progress::frontier::MutableAntichain;
use crate::progress::Timestamp;
use crate::dataflow::channels::pact::ParallelizationContract;

use crate::dataflow::operators::generic::handles::{InputHandleCore, OutputBuilderSession, OutputBuilder};
use crate::dataflow::operators::capability::Capability;

use crate::dataflow::{Scope, Stream};

use super::builder_rc::OperatorBuilder;
use crate::dataflow::operators::generic::OperatorInfo;
use crate::{Container, ContainerBuilder};
use crate::container::CapacityContainerBuilder;

/// Methods to construct generic streaming and blocking operators.
pub trait Operator<'scope, T: Timestamp, C1> {
    /// Creates a new dataflow operator that partitions its input stream by a parallelization
    /// strategy `pact`, and repeatedly invokes `logic`, the function returned by the function passed as `constructor`.
    /// `logic` can read from the input stream, write to the output stream, and inspect the frontier at the input.
    ///
    /// # Examples
    /// ```
    /// use std::collections::HashMap;
    /// use timely::dataflow::operators::ToStream;
    /// use timely::dataflow::operators::generic::Operator;
    /// use timely::dataflow::channels::pact::Pipeline;
    ///
    /// timely::example(|scope| {
    ///     (0u64..10)
    ///         .to_stream(scope)
    ///         .container::<Vec<_>>()
    ///         .unary_frontier(Pipeline, "example", |default_cap, _info| {
    ///             let mut cap = Some(default_cap.delayed(&12));
    ///             let mut stash = HashMap::new();
    ///             move |(input, frontier), output| {
    ///                 if let Some(ref c) = cap.take() {
    ///                     output.session(&c).give(12);
    ///                 }
    ///                 // Stash data under a capability for its least time.
    ///                 input.for_each_stamp(|cap, data| {
    ///                     if let Some(cap) = cap.retain_least(output.output_index()) {
    ///                         stash.entry(cap)
    ///                              .or_insert(Vec::new())
    ///                              .extend(data.flat_map(|d| d.drain(..)));
    ///                     }
    ///                 });
    ///                 // Send stashed data whose time the input frontier has passed.
    ///                 stash.retain(|cap, vec| {
    ///                     if frontier.frontier().less_equal(cap.time()) { true } else {
    ///                         output.session(cap).give_iterator(vec.drain(..));
    ///                         false
    ///                     }
    ///                 });
    ///             }
    ///         })
    ///         .container::<Vec<_>>();
    /// });
    /// ```
    fn unary_frontier<CB, B, L, P>(self, pact: P, name: &str, constructor: B) -> Stream<'scope, T, CB::Container>
    where
        CB: ContainerBuilder,
        B: FnOnce(Capability<T>, OperatorInfo) -> L,
        L: FnMut((&mut InputHandleCore<T, C1, P::Puller>, &MutableAntichain<T>),
                 &mut OutputBuilderSession<'_, T, CB>)+'static,
        P: ParallelizationContract<T, C1>;


    /// Creates a new dataflow operator that partitions its input stream by a parallelization
    /// strategy `pact`, and repeatedly invokes `logic`, the function returned by the function passed as `constructor`.
    /// `logic` can read from the input stream, and write to the output stream.
    ///
    /// # Examples
    /// ```
    /// use timely::dataflow::operators::ToStream;
    /// use timely::dataflow::operators::generic::operator::Operator;
    /// use timely::dataflow::channels::pact::Pipeline;
    /// use timely::dataflow::Scope;
    ///
    /// timely::example(|scope| {
    ///     (0u64..10)
    ///         .to_stream(scope)
    ///         .container::<Vec<_>>()
    ///         .unary(Pipeline, "example", |default_cap, _info| {
    ///             let mut cap = Some(default_cap.delayed(&12));
    ///             move |input, output| {
    ///                 if let Some(ref c) = cap.take() {
    ///                     output.session(&c).give(100);
    ///                 }
    ///                 input.for_each_stamp(|cap, data| {
    ///                     output.session(&cap).give_containers(data);
    ///                 });
    ///             }
    ///         });
    /// });
    /// ```
    fn unary<CB, B, L, P>(self, pact: P, name: &str, constructor: B) -> Stream<'scope, T, CB::Container>
    where
        CB: ContainerBuilder,
        B: FnOnce(Capability<T>, OperatorInfo) -> L,
        L: FnMut(&mut InputHandleCore<T, C1, P::Puller>,
                 &mut OutputBuilderSession<T, CB>)+'static,
        P: ParallelizationContract<T, C1>;

    /// Creates a new dataflow operator that partitions its input streams by a parallelization
    /// strategy `pact`, and repeatedly invokes `logic`, the function returned by the function passed as `constructor`.
    /// `logic` can read from the input streams, write to the output stream, and inspect the frontier at the inputs.
    ///
    /// # Examples
    /// ```
    /// use std::collections::HashMap;
    /// use timely::dataflow::operators::{Input, Inspect};
    /// use timely::dataflow::operators::generic::operator::Operator;
    /// use timely::dataflow::channels::pact::Pipeline;
    ///
    /// timely::execute(timely::Config::thread(), |worker| {
    ///    let (mut in1, mut in2) = worker.dataflow::<usize,_,_>(|scope| {
    ///        let (in1_handle, in1) = scope.new_input::<Vec<_>>();
    ///        let (in2_handle, in2) = scope.new_input::<Vec<_>>();
    ///        in1.binary_frontier(in2, Pipeline, Pipeline, "example", |mut _default_cap, _info| {
    ///            let mut stash = HashMap::new();
    ///            move |(input1, frontier1), (input2, frontier2), output| {
    ///                // Stash data from either input under a capability for its least time.
    ///                input1.for_each_stamp(|cap, data| {
    ///                    if let Some(cap) = cap.retain_least(output.output_index()) {
    ///                        stash.entry(cap).or_insert(Vec::new()).extend(data.flat_map(|d| d.drain(..)));
    ///                    }
    ///                });
    ///                input2.for_each_stamp(|cap, data| {
    ///                    if let Some(cap) = cap.retain_least(output.output_index()) {
    ///                        stash.entry(cap).or_insert(Vec::new()).extend(data.flat_map(|d| d.drain(..)));
    ///                    }
    ///                });
    ///                // Send stashed data whose time both input frontiers have passed.
    ///                stash.retain(|cap, vec| {
    ///                    if frontier1.frontier().less_equal(cap.time()) || frontier2.frontier().less_equal(cap.time()) { true } else {
    ///                        output.session(cap).give_iterator(vec.drain(..));
    ///                        false
    ///                    }
    ///                });
    ///            }
    ///        })
    ///        .container::<Vec<_>>()
    ///        .inspect_core(|e| if let Ok((s, x)) = e { println!("{:?} -> {:?}", s, x) });
    ///
    ///        (in1_handle, in2_handle)
    ///    });
    ///
    ///    for i in 1..10 {
    ///        in1.send(i - 1);
    ///        in1.advance_to(i);
    ///        in2.send(i - 1);
    ///        in2.advance_to(i);
    ///    }
    /// }).unwrap();
    /// ```
    fn binary_frontier<C2, CB, B, L, P1, P2>(self, other: Stream<'scope, T, C2>, pact1: P1, pact2: P2, name: &str, constructor: B) -> Stream<'scope, T, CB::Container>
    where
        C2: Container,
        CB: ContainerBuilder,
        B: FnOnce(Capability<T>, OperatorInfo) -> L,
        L: FnMut((&mut InputHandleCore<T, C1, P1::Puller>, &MutableAntichain<T>),
                 (&mut InputHandleCore<T, C2, P2::Puller>, &MutableAntichain<T>),
                 &mut OutputBuilderSession<'_, T, CB>)+'static,
        P1: ParallelizationContract<T, C1>,
        P2: ParallelizationContract<T, C2>;


    /// Creates a new dataflow operator that partitions its input streams by a parallelization
    /// strategy `pact`, and repeatedly invokes `logic`, the function returned by the function passed as `constructor`.
    /// `logic` can read from the input streams, write to the output stream, and inspect the frontier at the inputs.
    ///
    /// # Examples
    /// ```
    /// use timely::dataflow::operators::{ToStream, Inspect};
    /// use timely::dataflow::operators::generic::operator::Operator;
    /// use timely::dataflow::channels::pact::Pipeline;
    /// use timely::dataflow::Scope;
    ///
    /// timely::example(|scope| {
    ///     let stream1 = (0u64..10).to_stream(scope).container::<Vec<_>>();
    ///     let stream2 = (0u64..10).to_stream(scope).container::<Vec<_>>();
    ///     stream1
    ///         .binary(stream2, Pipeline, Pipeline, "example", |default_cap, _info| {
    ///             let mut cap = Some(default_cap.delayed(&12));
    ///             move |input1, input2, output| {
    ///                 if let Some(ref c) = cap.take() {
    ///                     output.session(&c).give(100);
    ///                 }
    ///                 input1.for_each_stamp(|cap, data| output.session(&cap).give_containers(data));
    ///                 input2.for_each_stamp(|cap, data| output.session(&cap).give_containers(data));
    ///             }
    ///         }).inspect(|x| println!("{:?}", x));
    /// });
    /// ```
    fn binary<C2, CB, B, L, P1, P2>(self, other: Stream<'scope, T, C2>, pact1: P1, pact2: P2, name: &str, constructor: B) -> Stream<'scope, T, CB::Container>
    where
        C2: Container,
        CB: ContainerBuilder,
        B: FnOnce(Capability<T>, OperatorInfo) -> L,
        L: FnMut(&mut InputHandleCore<T, C1, P1::Puller>,
                 &mut InputHandleCore<T, C2, P2::Puller>,
                 &mut OutputBuilderSession<'_, T, CB>)+'static,
        P1: ParallelizationContract<T, C1>,
        P2: ParallelizationContract<T, C2>;

    /// Creates a new dataflow operator that partitions its input stream by a parallelization
    /// strategy `pact`, and repeatedly invokes the function `logic` which can read from the input stream
    /// and inspect the frontier at the input.
    ///
    /// # Examples
    /// ```
    /// use timely::dataflow::operators::ToStream;
    /// use timely::dataflow::operators::generic::operator::Operator;
    /// use timely::dataflow::channels::pact::Pipeline;
    /// use timely::dataflow::Scope;
    ///
    /// timely::example(|scope| {
    ///     (0u64..10)
    ///         .to_stream(scope)
    ///         .container::<Vec<_>>()
    ///         .sink(Pipeline, "example", |(input, frontier)| {
    ///             input.for_each_stamp(|cap, data| {
    ///                 for datum in data.flatten() {
    ///                     println!("{:?}:\t{:?}", cap, datum);
    ///                 }
    ///             });
    ///         });
    /// });
    /// ```
    fn sink<L, P>(self, pact: P, name: &str, logic: L)
    where
        L: FnMut((&mut InputHandleCore<T, C1, P::Puller>, &MutableAntichain<T>))+'static,
        P: ParallelizationContract<T, C1>;
}

impl<'scope, T: Timestamp, C1: Container> Operator<'scope, T, C1> for Stream<'scope, T, C1> {

    fn unary_frontier<CB, B, L, P>(self, pact: P, name: &str, constructor: B) -> Stream<'scope, T, CB::Container>
    where
        CB: ContainerBuilder,
        B: FnOnce(Capability<T>, OperatorInfo) -> L,
        L: FnMut((&mut InputHandleCore<T, C1, P::Puller>, &MutableAntichain<T>),
                 &mut OutputBuilderSession<'_, T, CB>)+'static,
        P: ParallelizationContract<T, C1> {

        let mut builder = OperatorBuilder::new(name.to_owned(), self.scope());
        let operator_info = builder.operator_info();

        let mut input = builder.new_input(self, pact);
        let (output, stream) = builder.new_output();
        let mut output = OutputBuilder::from(output);

        builder.build(move |mut capabilities| {
            // `capabilities` should be a single-element vector.
            let capability = capabilities.pop().unwrap();
            let mut logic = constructor(capability, operator_info);
            move |frontiers| {
                let mut output_handle = output.activate();
                logic((&mut input, &frontiers[0]), &mut output_handle);
            }
        });

        stream
    }


    fn unary<CB, B, L, P>(self, pact: P, name: &str, constructor: B) -> Stream<'scope, T, CB::Container>
    where
        CB: ContainerBuilder,
        B: FnOnce(Capability<T>, OperatorInfo) -> L,
        L: FnMut(&mut InputHandleCore<T, C1, P::Puller>,
                 &mut OutputBuilderSession<'_, T, CB>)+'static,
        P: ParallelizationContract<T, C1> {

        let mut builder = OperatorBuilder::new(name.to_owned(), self.scope());
        let operator_info = builder.operator_info();

        let mut input = builder.new_input(self, pact);
        builder.set_notify_for(0, crate::progress::operate::FrontierInterest::Never);
        let (output, stream) = builder.new_output();
        let mut output = OutputBuilder::from(output);

        builder.build(move |mut capabilities| {
            // `capabilities` should be a single-element vector.
            let capability = capabilities.pop().unwrap();
            let mut logic = constructor(capability, operator_info);
            move |_frontiers| logic(&mut input, &mut output.activate())
        });

        stream
    }

    fn binary_frontier<C2, CB, B, L, P1, P2>(self, other: Stream<'scope, T, C2>, pact1: P1, pact2: P2, name: &str, constructor: B) -> Stream<'scope, T, CB::Container>
    where
        C2: Container,
        CB: ContainerBuilder,
        B: FnOnce(Capability<T>, OperatorInfo) -> L,
        L: FnMut((&mut InputHandleCore<T, C1, P1::Puller>, &MutableAntichain<T>),
                 (&mut InputHandleCore<T, C2, P2::Puller>, &MutableAntichain<T>),
                 &mut OutputBuilderSession<'_, T, CB>)+'static,
        P1: ParallelizationContract<T, C1>,
        P2: ParallelizationContract<T, C2> {

        let mut builder = OperatorBuilder::new(name.to_owned(), self.scope());
        let operator_info = builder.operator_info();

        let mut input1 = builder.new_input(self, pact1);
        let mut input2 = builder.new_input(other, pact2);
        let (output, stream) = builder.new_output();
        let mut output = OutputBuilder::from(output);

        builder.build(move |mut capabilities| {
            // `capabilities` should be a single-element vector.
            let capability = capabilities.pop().unwrap();
            let mut logic = constructor(capability, operator_info);
            move |frontiers| {
                let mut output_handle = output.activate();
                logic((&mut input1, &frontiers[0]), (&mut input2, &frontiers[1]), &mut output_handle);
            }
        });

        stream
    }



    fn binary<C2, CB, B, L, P1, P2>(self, other: Stream<'scope, T, C2>, pact1: P1, pact2: P2, name: &str, constructor: B) -> Stream<'scope, T, CB::Container>
    where
        C2: Container,
        CB: ContainerBuilder,
        B: FnOnce(Capability<T>, OperatorInfo) -> L,
        L: FnMut(&mut InputHandleCore<T, C1, P1::Puller>,
                 &mut InputHandleCore<T, C2, P2::Puller>,
                 &mut OutputBuilderSession<'_, T, CB>)+'static,
        P1: ParallelizationContract<T, C1>,
        P2: ParallelizationContract<T, C2> {

        let mut builder = OperatorBuilder::new(name.to_owned(), self.scope());
        let operator_info = builder.operator_info();

        let mut input1 = builder.new_input(self, pact1);
        let mut input2 = builder.new_input(other, pact2);
        builder.set_notify_for(0, crate::progress::operate::FrontierInterest::Never);
        builder.set_notify_for(1, crate::progress::operate::FrontierInterest::Never);
        let (output, stream) = builder.new_output();
        let mut output = OutputBuilder::from(output);

        builder.build(move |mut capabilities| {
            // `capabilities` should be a single-element vector.
            let capability = capabilities.pop().unwrap();
            let mut logic = constructor(capability, operator_info);
            move |_frontiers| {
                let mut output_handle = output.activate();
                logic(&mut input1, &mut input2, &mut output_handle);
            }
        });

        stream
    }

    fn sink<L, P>(self, pact: P, name: &str, mut logic: L)
    where
        L: FnMut((&mut InputHandleCore<T, C1, P::Puller>, &MutableAntichain<T>))+'static,
        P: ParallelizationContract<T, C1> {

        let mut builder = OperatorBuilder::new(name.to_owned(), self.scope());
        let mut input = builder.new_input(self, pact);

        builder.build(|_capabilities| {
            move |frontiers| {
                logic((&mut input, &frontiers[0]));
            }
        });
    }
}

/// Creates a new data stream source for a scope.
///
/// The source is defined by a name, and a constructor which takes a default capability to
/// a method that can be repeatedly called on a output handle. The method is then repeatedly
/// invoked, and is expected to eventually send data and downgrade and release capabilities.
///
/// # Examples
/// ```
/// use timely::dataflow::operators::Inspect;
/// use timely::dataflow::operators::generic::operator::source;
/// use timely::dataflow::Scope;
///
/// timely::example(|scope| {
///
///     source(scope, "Source", |capability, info| {
///
///         let activator = scope.activator_for(info.address);
///
///         let mut cap = Some(capability);
///         move |output| {
///
///             let mut done = false;
///             if let Some(cap) = cap.as_mut() {
///                 // get some data and send it.
///                 let time = cap.time().clone();
///                 output.session(&cap)
///                       .give(*cap.time());
///
///                 // downgrade capability.
///                 cap.downgrade(&(time + 1));
///                 done = time > 20;
///             }
///
///             if done { cap = None; }
///             else    { activator.activate(); }
///         }
///     })
///     .container::<Vec<_>>()
///     .inspect(|x| println!("number: {:?}", x));
/// });
/// ```
pub fn source<'scope, T: Timestamp, CB, B, L>(scope: Scope<'scope, T>, name: &str, constructor: B) -> Stream<'scope, T, CB::Container>
where
    CB: ContainerBuilder,
    B: FnOnce(Capability<T>, OperatorInfo) -> L,
    L: FnMut(&mut OutputBuilderSession<'_, T, CB>)+'static {

    let mut builder = OperatorBuilder::new(name.to_owned(), scope);
    let operator_info = builder.operator_info();

    let (output, stream) = builder.new_output();
    let mut output = OutputBuilder::from(output);

    builder.build(move |mut capabilities| {
        // `capabilities` should be a single-element vector.
        let capability = capabilities.pop().unwrap();
        let mut logic = constructor(capability, operator_info);
        move |_frontier| {
            logic(&mut output.activate());
        }
    });

    stream
}

/// Constructs an empty stream.
///
/// This method is useful in patterns where an input is required, but there is no
/// meaningful data to provide. The replaces patterns like `stream.filter(|_| false)`
/// which are just silly.
///
/// # Examples
/// ```
/// use timely::dataflow::operators::Inspect;
/// use timely::dataflow::operators::generic::operator::empty;
/// use timely::dataflow::Scope;
///
/// timely::example(|scope| {
///
///
///     empty::<_, Vec<_>>(scope)     // type required in this example
///         .inspect(|()| panic!("never called"));
///
/// });
/// ```
pub fn empty<'scope, T: Timestamp, C: Container>(scope: Scope<'scope, T>) -> Stream<'scope, T, C> {
    source::<_, CapacityContainerBuilder<C>, _, _>(scope, "Empty", |_capability, _info| |_output| {
        // drop capability, do nothing
    })
}