Struct Stream 

pub struct Stream<'scope, T: Timestamp, C> { /* private fields */ }

Abstraction of a stream of C: Container records timestamped with T.

Internally Stream maintains a list of data recipients who should be presented with data produced by the source of the stream.

Implementations

impl<'scope, T: Timestamp, C> Stream<'scope, T, C>

pub fn connect_to<P: Push<Message<T, C>> + 'static>( self, target: Target, pusher: P, identifier: usize, )

where C: 'static,

Connects the stream to a destination.

The destination is described both by a Target, for progress tracking information, and a P: Push where the records should actually be sent. The identifier is unique to the edge and is used only for logging purposes.

pub fn new( source: Source, output: TeeHelper<T, C>, scope: Scope<'scope, T>, ) -> Self

Allocates a Stream from a supplied Source name and rendezvous point.

pub fn name(&self) -> &Source

The name of the stream’s source operator.

pub fn scope(&self) -> Scope<'scope, T>

The scope immediately containing the stream.

pub fn container<C2>(self) -> Stream<'scope, T, C2>

where Self: AsStream<'scope, T, C2>,

Allows the assertion of a container type, for the benefit of type inference.

This method can be needed when the container type of a stream is unconstrained, most commonly after creating an input, or bracking wholly generic operators.

Examples

use timely::dataflow::operators::{ToStream, Inspect};

timely::example(|scope| {
    (0..10).to_stream(scope)
           .container::<Vec<_>>()
           .inspect(|x| println!("seen: {:?}", x));
});

Examples found in repository

examples/capture_send.rs (line 14)

fn main() {
    timely::execute_from_args(std::env::args(), |worker| {

        let addr = format!("127.0.0.1:{}", 8000 + worker.index());
        let send = TcpStream::connect(addr).unwrap();

        worker.dataflow::<u64,_,_>(|scope|
            (0..10u64)
                .to_stream(scope)
                .container::<Vec<_>>()
                .capture_into(EventWriter::new(send))
        );
    }).unwrap();
}

examples/unordered_input.rs (line 8)

fn main() {
    timely::execute(Config::thread(), |worker| {
        let (mut input, mut cap) = worker.dataflow::<usize,_,_>(|scope| {
            let (input, stream) = scope.new_unordered_input();
            stream.container::<Vec<_>>().inspect_batch(|t, x| println!("{:?} -> {:?}", t, x));
            input
        });

        for round in 0..10 {
            input.activate().session(&cap).give(round);
            cap = cap.delayed(&(round + 1));
            worker.step();
        }
    }).unwrap();
}

examples/threadless.rs (line 21)

fn main() {

    // create a naked single-threaded worker.
    let allocator = timely::communication::Allocator::Thread(
        timely::communication::allocator::Thread::default(),
    );
    let mut worker = timely::worker::Worker::new(WorkerConfig::default(), allocator, None);

    // create input and probe handles.
    let mut input = InputHandle::new();
    let probe = ProbeHandle::new();

    // directly build a dataflow.
    worker.dataflow(|scope| {
        input
            .to_stream(scope)
            .container::<Vec<_>>()
            .inspect(|x| println!("{:?}", x))
            .probe_with(&probe);
    });

    // manage inputs.
    for i in 0 .. 10 {
        input.send(i);
        input.advance_to(i);
        while probe.less_than(input.time()) {
            worker.step();
        }
    }
}

examples/rc.rs (line 19)

fn main() {
    // initializes and runs a timely dataflow.
    timely::execute_from_args(std::env::args(), |worker| {
        // create a new input, exchange data, and inspect its output
        let index = worker.index();
        let mut input = InputHandle::new();
        let probe = ProbeHandle::new();
        worker.dataflow(|scope| {
            scope.input_from(&mut input)
                 .container::<Vec<_>>()
                 //.exchange(|x| *x) // <-- cannot exchange this; Rc is not Send.
                 .inspect(move |x| println!("worker {}:\thello {:?}", index, x))
                 .probe_with(&probe);
        });

        // introduce data and watch!
        for round in 0..10 {
            input.send(Test { _field: Rc::new(round) } );
            input.advance_to(round + 1);
            worker.step_while(|| probe.less_than(input.time()));
        }
    }).unwrap();
}

examples/hello.rs (line 15)

fn main() {
    // initializes and runs a timely dataflow.
    timely::execute_from_args(std::env::args(), |worker| {

        let index = worker.index();
        let mut input = InputHandle::new();
        let probe = ProbeHandle::new();

        // create a new input, exchange data, and inspect its output
        worker.dataflow(|scope| {
            scope.input_from(&mut input)
                 .container::<Vec<_>>()
                 .exchange(|x| *x)
                 .inspect(move |x| println!("worker {}:\thello {}", index, x))
                 .probe_with(&probe);
        });

        // introduce data and watch!
        for round in 0..10 {
            if index == 0 {
                input.send(round);
            }
            input.advance_to(round + 1);
            while probe.less_than(input.time()) {
                worker.step();
            }
        }
    }).unwrap();
}

examples/exchange.rs (line 16)

fn main() {
    // initializes and runs a timely dataflow.
    timely::execute_from_args(std::env::args(), |worker| {

        let batch = std::env::args().nth(1).unwrap().parse::<usize>().unwrap();
        let rounds = std::env::args().nth(2).unwrap().parse::<usize>().unwrap();
        let mut input = InputHandle::new();

        // create a new input, exchange data, and inspect its output
        let probe = worker.dataflow(|scope|
            scope
                .input_from(&mut input)
                .container::<Vec<_>>()
                .exchange(|&x| x as u64)
                .probe()
                .0
        );


        let timer = std::time::Instant::now();

        for round in 0 .. rounds {

            for i in 0 .. batch {
                input.send(i);
            }
            input.advance_to(round);

            while probe.less_than(input.time()) {
                worker.step();
            }

        }

        let volume = (rounds * batch) as f64;
        let elapsed = timer.elapsed();
        let seconds = elapsed.as_secs() as f64 + (f64::from(elapsed.subsec_nanos())/1000000000.0);

        println!("{:?}\tworker {} complete; rate: {:?}", timer.elapsed(), worker.index(), volume / seconds);

    }).unwrap();
}

Additional examples can be found in:

• examples/distinct.rs
• examples/wordcount.rs
• examples/logging_replay.rs
• examples/hashjoin.rs
• examples/logging-send.rs
• examples/columnar.rs
• examples/bfs.rs

Trait Implementations

impl<'scope, T: Timestamp, C> AsStream<'scope, T, C> for Stream<'scope, T, C>

fn as_stream(self) -> Self

Translate self to a Stream.

impl<'scope, T: Timestamp, C: Container> BranchWhen<T> for Stream<'scope, T, C>

fn branch_when(self, condition: impl Fn(&T) -> bool + 'static) -> (Self, Self)

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.

impl<T: Timestamp, C: Container> Capture<T, C> for Stream<'_, T, C>

fn capture_into<P: EventPusher<T, C> + 'static>(self, event_pusher: P)

Captures a stream of timestamped data for later replay.

fn capture(self) -> Receiver<Event<T, C>>

Captures a stream using Rust’s MPSC channels.

impl<'scope, T: Timestamp, C: Clone + 'static> Clone for Stream<'scope, T, C>

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<'scope, T: Timestamp, C: Container> Concat for Stream<'scope, T, C>

fn concat(self, other: Self) -> Self

Merge the contents of two streams.

impl<'scope, T: Timestamp, C: Container> ConnectLoop<'scope, T, C> for Stream<'scope, T, C>

fn connect_loop(self, handle: Handle<'scope, T, C>)

Connect a Stream to be the input of a loop variable.

impl<'scope, T: Timestamp, C> Debug for Stream<'scope, T, C>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<'outer, TOuter, TInner, C> Enter<'outer, TOuter, TInner, C> for Stream<'outer, TOuter, C>

where TOuter: Timestamp, TInner: Timestamp + Refines<TOuter>, C: Container,

fn enter<'inner>( self, inner: Scope<'inner, TInner>, ) -> Stream<'inner, TInner, C>

Moves the Stream argument into a child of its current Scope.

impl<T: Timestamp, C> Exchange<C> for Stream<'_, T, C>

where C: Container + SizableContainer + DrainContainer + Send + ContainerBytes + for<'a> PushInto<C::Item<'a>>,

fn exchange<F>(self, route: F) -> Self

where for<'a> F: FnMut(&C::Item<'a>) -> u64 + 'static,

Exchange records between workers.

impl<T: Timestamp, C> Filter<C> for Stream<'_, T, C>

where for<'a> C: PushInto<C::Item<'a>> + Container + SizableContainer + DrainContainer,

fn filter<P: FnMut(&C::Item<'_>) -> bool + 'static>(self, predicate: P) -> Self

Returns a new instance of self containing only records satisfying predicate.

impl<T: Timestamp, C: Container> Inspect<T, C> for Stream<'_, T, C>

where for<'a> &'a C: IntoIterator,

fn inspect_core<F>(self, func: F) -> Self

where F: FnMut(Result<(&T, &C), &[T]>) + 'static,

Runs a supplied closure on each observed data batch, and each frontier advancement.

fn inspect<F>(self, func: F) -> Self

where F: for<'a> FnMut(<&'a C as IntoIterator>::Item) + 'static,

Runs a supplied closure on each observed data element.

fn inspect_time<F>(self, func: F) -> Self

where F: for<'a> FnMut(&T, <&'a C as IntoIterator>::Item) + 'static,

Runs a supplied closure on each observed data element and associated time.

fn inspect_batch(self, func: impl FnMut(&T, &C) + 'static) -> Self

Runs a supplied closure on each observed data batch (time and data slice).

impl<T: Timestamp, C: Container> InspectCore<T, C> for Stream<'_, T, C>

fn inspect_container<F>(self, func: F) -> Self

where F: FnMut(Result<(&T, &C), &[T]>) + 'static,

Runs a supplied closure on each observed container, and each frontier advancement.

impl<'inner, TInner, C> Leave<'inner, TInner, C> for Stream<'inner, TInner, C>

where TInner: Timestamp, C: Container,

fn leave<'outer, TOuter: Timestamp>( self, outer: Scope<'outer, TOuter>, ) -> Stream<'outer, TOuter, C>

where TInner: Refines<TOuter>,

Moves a Stream to the parent of its current Scope.

impl<'scope, T: Timestamp, C: Container + DrainContainer> Map<'scope, T, C> for Stream<'scope, T, C>

fn flat_map<C2, I, L>(self, logic: L) -> Stream<'scope, T, C2>

where I: IntoIterator, C2: Container + SizableContainer + PushInto<I::Item>, L: FnMut(C::Item<'_>) -> I + 'static,

Consumes each element of the stream and yields some number of new elements.

fn map<C2, D2, L>(self, logic: L) -> Stream<'scope, T, C2>

where C2: Container + SizableContainer + PushInto<D2>, L: FnMut(C::Item<'_>) -> D2 + 'static,

Consumes each element of the stream and yields a new element.

fn flat_map_builder<I, L>(self, logic: L) -> FlatMapBuilder<Self, C, L, I>

where L: for<'a> Fn(C::Item<'a>) -> I,

Creates a FlatMapBuilder, which allows chaining of iterator logic before finalization into a stream.

impl<'scope, T: Timestamp, C: Container + DrainContainer> OkErr<'scope, T, C> for Stream<'scope, T, C>

fn ok_err<C1, D1, C2, D2, L>( self, logic: L, ) -> (Stream<'scope, T, C1>, Stream<'scope, T, C2>)

where C1: Container + SizableContainer + PushInto<D1>, C2: Container + SizableContainer + PushInto<D2>, L: FnMut(C::Item<'_>) -> Result<D1, D2> + 'static,

Takes one input stream and splits it into two output streams. For each record, the supplied closure is called with the data. If it returns Ok(x), then x will be sent to the first returned stream; otherwise, if it returns Err(e), then e will be sent to the second.

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>,

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.

fn unary_notify<CB: ContainerBuilder, L: FnMut(&mut InputHandleCore<T, C1, P::Puller>, &mut OutputBuilderSession<'_, T, CB>, &mut Notificator<'_, T>) + 'static, P: ParallelizationContract<T, C1>>( self, pact: P, name: &str, init: impl IntoIterator<Item = T>, logic: L, ) -> Stream<'scope, T, CB::Container>

Creates a new dataflow operator that partitions its input stream by a parallelization strategy pact, and repeatedly invokes the closure supplied as logic, which can read from the input stream, write to the output stream, and inspect the frontier at the input.

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 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.

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.

fn binary_notify<C2: Container, CB: ContainerBuilder, L: FnMut(&mut InputHandleCore<T, C1, P1::Puller>, &mut InputHandleCore<T, C2, P2::Puller>, &mut OutputBuilderSession<'_, T, CB>, &mut Notificator<'_, T>) + 'static, P1: ParallelizationContract<T, C1>, P2: ParallelizationContract<T, C2>>( self, other: Stream<'scope, T, C2>, pact1: P1, pact2: P2, name: &str, init: impl IntoIterator<Item = T>, logic: L, ) -> Stream<'scope, T, CB::Container>

Creates a new dataflow operator that partitions its input stream by a parallelization strategy pact, and repeatedly invokes the closure supplied as logic, which can read from the input streams, write to the output stream, and inspect the frontier at the inputs.

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 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.

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>,

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.

impl<'scope, T: Timestamp, C: Container + DrainContainer> Partition<'scope, T, C> for Stream<'scope, T, C>

fn partition<CB, D2, F>( self, parts: u64, route: F, ) -> Vec<Stream<'scope, T, CB::Container>>

where CB: ContainerBuilder + PushInto<D2>, F: FnMut(C::Item<'_>) -> (u64, D2) + 'static,

Produces parts output streams, containing records produced and assigned by route.

impl<T: Timestamp, C: Container> Probe<T, C> for Stream<'_, T, C>

fn probe(self) -> (Handle<T>, Self)

Constructs a progress probe which indicates which timestamps have elapsed at the operator.

fn probe_with(self, handle: &Handle<T>) -> Self

Inserts a progress probe in a stream.

impl<'scope, T: Timestamp, C: Container> Reclock<'scope, T> for Stream<'scope, T, C>

fn reclock<TC: Container>(self, clock: Stream<'scope, T, TC>) -> Self

Delays records until an input is observed on the clock input.

impl<'scope, T: Timestamp, C: Container> SharedStream<'scope, T, C> for Stream<'scope, T, C>

fn shared(self) -> Stream<'scope, T, Rc<C>>

Convert a stream into a stream of shared data