Struct differential_dataflow::operators::arrange::agent::TraceAgent

source ·

pub struct TraceAgent<Tr>where
    Tr: TraceReader,
{ /* private fields */ }

Expand description

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.

Implementations§

source §

impl<Tr: TraceReader> TraceAgent<Tr>

source

pub fn new( trace: Tr, operator: OperatorInfo, logging: Option<Logger>, ) -> (Self, TraceWriter<Tr>)
where Tr: Trace, Tr::Batch: Batch,

Creates a new agent from a trace reader.

source

pub fn new_listener( &mut self, activator: Activator, ) -> Rc<(Activator, RefCell<VecDeque<TraceReplayInstruction<Tr>>>)>

Attaches a new shared queue to the trace.

The queue is first populated with existing batches from 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.

source

pub fn operator(&self) -> &OperatorInfo

The OperatorInfo of the underlying Timely operator

source

pub fn trace_box_unstable(&self) -> Rc<RefCell<TraceBox<Tr>>>

Obtain a reference to the inner TraceBox. It is the caller’s obligation to maintain the trace box and this trace agent’s invariants. Specifically, it is undefined behavior to mutate the trace box. Keeping strong references can prevent resource reclamation.

This method is subject to changes and removal and should not be considered part of a stable interface.

source §

impl<Tr> TraceAgent<Tr>
where Tr: TraceReader + 'static,

source

pub fn import<G>(&mut self, scope: &G) -> Arranged<G, TraceAgent<Tr>>
where G: Scope<Timestamp = Tr::Time>,

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.get_logical_compaction(). 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.get_logical_compaction()”, 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

use timely::Config;
use differential_dataflow::input::Input;
use differential_dataflow::operators::arrange::ArrangeBySelf;
use differential_dataflow::operators::reduce::Reduce;
use differential_dataflow::trace::Trace;

::timely::execute(Config::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
             .arrange_by_self()
             .trace
    });

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

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

}).unwrap();

source

pub fn import_named<G>( &mut self, scope: &G, name: &str, ) -> Arranged<G, TraceAgent<Tr>>
where G: Scope<Timestamp = Tr::Time>,

Same as import, but allows to name the source.

source

pub fn import_core<G>( &mut self, scope: &G, name: &str, ) -> (Arranged<G, TraceAgent<Tr>>, ShutdownButton<CapabilitySet<Tr::Time>>)
where G: Scope<Timestamp = Tr::Time>,

Imports an arrangement into the supplied scope.

§Examples

use timely::Config;
use timely::dataflow::ProbeHandle;
use timely::dataflow::operators::Probe;
use differential_dataflow::input::InputSession;
use differential_dataflow::operators::arrange::ArrangeBySelf;
use differential_dataflow::operators::reduce::Reduce;
use differential_dataflow::trace::Trace;

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

    let mut input = InputSession::<_,(),isize>::new();
    let mut probe = ProbeHandle::new();

    // create a first dataflow
    let mut trace = worker.dataflow::<u32,_,_>(|scope| {
        // create input handle and collection.
        input.to_collection(scope)
             .arrange_by_self()
             .trace
    });

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

    // create a second dataflow
    let mut shutdown = worker.dataflow(|scope| {
        let (arrange, button) = trace.import_core(scope, "Import");
        arrange.stream.probe_with(&mut probe);
        button
    });

    worker.step();
    worker.step();
    assert!(!probe.done());

    shutdown.press();

    worker.step();
    worker.step();
    assert!(probe.done());

}).unwrap();

source

pub fn import_frontier<G>( &mut self, scope: &G, name: &str, ) -> (Arranged<G, TraceFrontier<TraceAgent<Tr>>>, ShutdownButton<CapabilitySet<Tr::Time>>)
where G: Scope<Timestamp = Tr::Time>, Tr: TraceReader,

Imports an arrangement into the supplied scope.

This variant of import uses the get_logical_compaction to forcibly advance timestamps in updates.

§Examples

use timely::Config;
use timely::progress::frontier::AntichainRef;
use timely::dataflow::ProbeHandle;
use timely::dataflow::operators::Probe;
use timely::dataflow::operators::Inspect;
use differential_dataflow::input::InputSession;
use differential_dataflow::operators::arrange::ArrangeBySelf;
use differential_dataflow::operators::reduce::Reduce;
use differential_dataflow::trace::Trace;
use differential_dataflow::trace::TraceReader;
use differential_dataflow::input::Input;

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

    let mut probe = ProbeHandle::new();

    // create a first dataflow
    let (mut handle, mut trace) = worker.dataflow::<u32,_,_>(|scope| {
        // create input handle and collection.
        let (handle, stream) = scope.new_collection();
        let trace = stream.arrange_by_self().trace;
        (handle, trace)
    });

    handle.insert(0); handle.advance_to(1); handle.flush(); worker.step();
    handle.remove(0); handle.advance_to(2); handle.flush(); worker.step();
    handle.insert(1); handle.advance_to(3); handle.flush(); worker.step();
    handle.remove(1); handle.advance_to(4); handle.flush(); worker.step();
    handle.insert(0); handle.advance_to(5); handle.flush(); worker.step();

    trace.set_logical_compaction(AntichainRef::new(&[5]));

    // create a second dataflow
    let mut shutdown = worker.dataflow(|scope| {
        let (arrange, button) = trace.import_frontier(scope, "Import");
        arrange
            .as_collection(|k,v| (*k,*v))
            .inner
            .inspect(|(d,t,r)| {
                assert!(t >= &5);
            })
            .probe_with(&mut probe);

        button
    });

    worker.step();
    worker.step();
    assert!(!probe.done());

    shutdown.press();

    worker.step();
    worker.step();
    assert!(probe.done());

}).unwrap();

source

pub fn import_frontier_core<G>( &mut self, scope: &G, name: &str, since: Antichain<Tr::Time>, until: Antichain<Tr::Time>, ) -> (Arranged<G, TraceFrontier<TraceAgent<Tr>>>, ShutdownButton<CapabilitySet<Tr::Time>>)
where G: Scope<Timestamp = Tr::Time>, Tr: TraceReader,

Import a trace restricted to a specific time interval [since, until).

All updates present in the input trace will be first advanced to since, and then either emitted, or if greater or equal to until, suppressed. Once all times are certain to be greater or equal to until the operator capability will be dropped.

Invoking this method with an until of Antichain::new() will perform no filtering, as the empty frontier indicates the end of times.