Trait timely::dataflow::operators::input::Input [−][src]
Create a new Stream
and Handle
through which to supply input.
Required methods
fn new_input<D: Data>(
&mut self
) -> (Handle<Self::Timestamp, D>, Stream<Self, D>)
[src]
&mut self
) -> (Handle<Self::Timestamp, D>, Stream<Self, D>)
Create a new Stream
and Handle
through which to supply input.
The new_input
method returns a pair (Handle, Stream)
where the Stream
can be used
immediately for timely dataflow construction, and the Handle
is later used to introduce
data into the timely dataflow computation.
The Handle
also provides a means to indicate
to timely dataflow that the input has advanced beyond certain timestamps, allowing timely
to issue progress notifications.
Examples
use timely::*; use timely::dataflow::operators::{Input, Inspect}; // construct and execute a timely dataflow timely::execute(Config::thread(), |worker| { // add an input and base computation off of it let mut input = worker.dataflow(|scope| { let (input, stream) = scope.new_input(); stream.inspect(|x| println!("hello {:?}", x)); input }); // introduce input, advance computation for round in 0..10 { input.send(round); input.advance_to(round + 1); worker.step(); } });
fn input_from<D: Data>(
&mut self,
handle: &mut Handle<Self::Timestamp, D>
) -> Stream<Self, D>
[src]
&mut self,
handle: &mut Handle<Self::Timestamp, D>
) -> Stream<Self, D>
Create a new stream from a supplied interactive handle.
This method creates a new timely stream whose data are supplied interactively through the handle
argument. Each handle may be used multiple times (or not at all), and will clone data as appropriate
if it as attached to more than one stream.
Examples
use timely::*; use timely::dataflow::operators::{Input, Inspect}; use timely::dataflow::operators::input::Handle; // construct and execute a timely dataflow timely::execute(Config::thread(), |worker| { // add an input and base computation off of it let mut input = Handle::new(); worker.dataflow(|scope| { scope.input_from(&mut input) .inspect(|x| println!("hello {:?}", x)); }); // introduce input, advance computation for round in 0..10 { input.send(round); input.advance_to(round + 1); worker.step(); } });
Implementors
impl<G: Scope> Input for G where
<G as ScopeParent>::Timestamp: TotalOrder,
[src]
<G as ScopeParent>::Timestamp: TotalOrder,