Struct differential_dataflow::input::InputSession

pub struct InputSession<T: Timestamp + Clone, D: Data, R: Semigroup> { /* fields omitted */ }

An input session wrapping a single timely dataflow capability.

Each timely dataflow message has a corresponding capability, which is a logical time in the timely dataflow system. Differential dataflow updates can happen at a much higher rate than timely dataflow’s progress tracking infrastructure supports, because the logical times are promoted to data and updates are batched together. The InputSession type does this batching.

Examples

extern crate timely;
extern crate differential_dataflow;

use timely::Config;
use differential_dataflow::input::Input;

fn main() {
    ::timely::execute(Config::thread(), |worker| {

           let (mut handle, probe) = worker.dataflow(|scope| {
               // create input handle and collection.
               let (handle, data) = scope.new_collection_from(0 .. 10);
            let probe = data.map(|x| x * 2)
                               .inspect(|x| println!("{:?}", x))
                               .probe();
               (handle, probe)
        });

           handle.insert(3);
           handle.advance_to(1);
           handle.insert(5);
           handle.advance_to(2);
           handle.flush();

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

           handle.remove(5);
           handle.advance_to(3);
           handle.flush();

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

       }).unwrap();
}

Implementations

impl<T: Timestamp + Clone, D: Data> InputSession<T, D, isize>

pub fn insert(&mut self, element: D)

Adds an element to the collection.

pub fn remove(&mut self, element: D)

Removes an element from the collection.

impl<T: Timestamp + Clone, D: Data, R: Semigroup> InputSession<T, D, R>

pub fn to_collection<G: TimelyInput>(
    &mut self,
    scope: &mut G
) -> Collection<G, D, R> where
    G: ScopeParent<Timestamp = T>, 

Introduces a handle as collection.

pub fn new() -> Self

Allocates a new input handle.

pub fn from(handle: Handle<T, (D, T, R)>) -> Self

Creates a new session from a reference to an input handle.

pub fn update(&mut self, element: D, change: R)

Adds to the weight of an element in the collection.

pub fn update_at(&mut self, element: D, time: T, change: R)

Adds to the weight of an element in the collection at a future time.

pub fn flush(&mut self)

Forces buffered data into the timely dataflow input, and advances its time to match that of the session.

It is important to call flush before expecting timely dataflow to report progress. Until this method is called, all updates may still be in internal buffers and not exposed to timely dataflow. Once the method is called, all buffers are flushed and timely dataflow is advised that some logical times are no longer possible.

pub fn advance_to(&mut self, time: T)

Advances the logical time for future records.

Importantly, this method does not immediately inform timely dataflow of the change. This happens only when the session is dropped or flushed. It is not correct to use this time as a basis for a computation’s step_while method unless the session has just been flushed.

pub fn epoch(&self) -> &T

Reveals the current time of the session.

pub fn time(&self) -> &T

Reveals the current time of the session.

pub fn close(self)

Closes the input, flushing and sealing the wrapped timely input.

Trait Implementations

impl<T: Timestamp + Clone, D: Data, R: Semigroup> Drop for InputSession<T, D, R>

fn drop(&mut self)

Executes the destructor for this type.