Struct differential_dataflow::operators::arrange::TraceAgent

pub struct TraceAgent<K, V, T, R, Tr> where
    T: Lattice + Ord + Clone + 'static,
    Tr: TraceReader<K, V, T, R>,  { /* fields omitted */ }

A TraceReader wrapper which can be imported into other dataflows.

The TraceAgent is the default trace type produced by arranged, and it can be extracted from the dataflow in which it was defined, and imported into other dataflows.

Methods

impl<K, V, T, R, Tr> TraceAgent<K, V, T, R, Tr> where
    T: Lattice + Ord + Clone + 'static,
    Tr: TraceReader<K, V, T, R>, 

fn new(trace: Tr) -> (Self, TraceWriter<K, V, T, R, Tr>) where
    Tr: Trace<K, V, T, R>,
    Tr::Batch: Batch<K, V, T, R>, 

Creates a new agent from a trace reader.

fn new_listener(
    &mut self
) -> Rc<RefCell<VecDeque<(Vec<T>, Option<(T, <Tr as TraceReader<K, V, T, R>>::Batch)>)>>> where
    T: Default, 

Attaches a new shared queue to the trace.

The queue will be immediately populated with existing historical batches from the trace, and until the reference is dropped the queue will receive new batches as produced by the source arrange operator.

impl<K, V, T, R, Tr> TraceAgent<K, V, T, R, Tr> where
    T: Lattice + Ord + Clone + 'static,
    Tr: TraceReader<K, V, T, R>, 

fn import<G: Scope<Timestamp = T>>(
    &mut self,
    scope: &G
) -> Arranged<G, K, V, R, TraceAgent<K, V, T, R, Tr>> where
    T: Timestamp, 

Copies an existing collection into the supplied scope.

This method creates an Arranged collection that should appear indistinguishable from applying arrange directly to the source collection brought into the local scope. The only caveat is that the initial state of the collection is its current state, and updates occur from this point forward. The historical changes the collection experienced in the past are accumulated, and the distinctions from the initial collection are no longer evident.

The current behavior is that the introduced collection accumulates updates to some times less or equal to self.advance_frontier(). There is not currently a guarantee that the updates are accumulated to the frontier, and the resulting collection history may be weirdly partial until this point. In particular, the historical collection may move through configurations that did not actually occur, even if eventually arriving at the correct collection. This is probably a bug; although we get to the right place in the end, the intermediate computation could do something that the original computation did not, like diverge.

I would expect the semantics to improve to "updates are advanced to self.advance_frontier()", which means the computation will run as if starting from exactly this frontier. It is not currently clear whose responsibility this should be (the trace/batch should only reveal these times, or an operator should know to advance times before using them).

Examples

extern crate timely;
extern crate timely_communication;
extern crate differential_dataflow;

use timely_communication::Configuration;
use differential_dataflow::input::Input;
use differential_dataflow::operators::arrange::Arrange;
use differential_dataflow::operators::group::GroupArranged;
use differential_dataflow::trace::Trace;
use differential_dataflow::trace::implementations::ord::OrdValSpine;
use differential_dataflow::hashable::OrdWrapper;

fn main() {
    ::timely::execute(Configuration::Thread, |worker| {

        // create a first dataflow
        let mut trace = worker.dataflow::<u32,_,_>(|scope| {
            // create input handle and collection.
            scope.new_collection_from(0 .. 10).1
                 .map(|x| (OrdWrapper { item: x }, x))
                 .arrange(OrdValSpine::new())
                 .trace
        });

        // do some work.
        worker.step();
        worker.step();

        // create a second dataflow
        worker.dataflow(move |scope| {
            trace.import(scope)
                 .group_arranged(
                     move |_key, src, dst| dst.push((*src[0].0, 1)),
                     OrdValSpine::new()
                 );
        });

    }).unwrap();
}

Trait Implementations

impl<K, V, T, R, Tr> TraceReader<K, V, T, R> for TraceAgent<K, V, T, R, Tr> where
    T: Lattice + Ord + Clone + 'static,
    Tr: TraceReader<K, V, T, R>, 

type Batch = Tr::Batch

The type of an immutable collection of updates.

type Cursor = Tr::Cursor

The type used to enumerate the collections contents.

fn advance_by(&mut self, frontier: &[T])

Advances the frontier of times the collection must be correctly accumulable through.

fn advance_frontier(&mut self) -> &[T]

Reports the frontier from which all time comparisions should be accurate.

fn distinguish_since(&mut self, frontier: &[T])

Advances the frontier that may be used in cursor_through.

fn distinguish_frontier(&mut self) -> &[T]

Reports the frontier from which the collection may be subsetted.

fn cursor_through(
    &mut self,
    frontier: &[T]
) -> Option<(Tr::Cursor, <Tr::Cursor as Cursor<K, V, T, R>>::Storage)>

Acquires a cursor to the restriction of the collection's contents to updates at times not greater or equal to an element of upper.

fn map_batches<F: FnMut(&Self::Batch)>(&mut self, f: F)

Maps some logic across the batches the collection manages.

fn cursor(
    &mut self
) -> (Self::Cursor, <Self::Cursor as Cursor<Key, Val, Time, R>>::Storage)

Provides a cursor over updates contained in the trace.

impl<K, V, T, R, Tr> Clone for TraceAgent<K, V, T, R, Tr> where
    T: Lattice + Ord + Clone + 'static,
    Tr: TraceReader<K, V, T, R>, 

fn clone(&self) -> Self

Returns a copy of the value.

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

Performs copy-assignment from source.

impl<K, V, T, R, Tr> Drop for TraceAgent<K, V, T, R, Tr> where
    T: Lattice + Ord + Clone + 'static,
    Tr: TraceReader<K, V, T, R>, 

fn drop(&mut self)

Executes the destructor for this type.