Struct timely::worker::Worker

pub struct Worker<A: Allocate> { /* fields omitted */ }

A Worker is the entry point to a timely dataflow computation. It wraps a Allocate, and has a list of dataflows that it manages.

Methods

impl<A: Allocate> Worker<A>

pub fn new(c: A) -> Worker<A>

Allocates a new Worker bound to a channel allocator.

pub fn step(&mut self) -> bool

Performs one step of the computation.

A step gives each dataflow operator a chance to run, and is the main way to ensure that a computation proceeds.

Examples

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

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

    worker.dataflow::<usize,_,_>(|scope| {
        (0 .. 10)
            .to_stream(scope)
            .inspect(|x| println!("{:?}", x));
    });

    worker.step();
});

pub fn step_while<F: FnMut() -> bool>(&mut self, func: F)

Calls self.step() as long as func evaluates to true.

Examples

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

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

    let probe =
    worker.dataflow::<usize,_,_>(|scope| {
        (0 .. 10)
            .to_stream(scope)
            .inspect(|x| println!("{:?}", x))
            .probe()
    });

    worker.step_while(|| probe.less_than(&0));
});

pub fn index(&self) -> usize

The index of the worker out of its peers.

Examples

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

    let index = worker.index();
    let peers = worker.peers();
    let timer = worker.timer();

    println!("{:?}\tWorker {} of {}", timer.elapsed(), index, peers);

});

pub fn peers(&self) -> usize

The total number of peer workers.

Examples

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

    let index = worker.index();
    let peers = worker.peers();
    let timer = worker.timer();

    println!("{:?}\tWorker {} of {}", timer.elapsed(), index, peers);

});

pub fn timer(&self) -> Instant

A timer started at the initiation of the timely computation.

Examples

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

    let index = worker.index();
    let peers = worker.peers();
    let timer = worker.timer();

    println!("{:?}\tWorker {} of {}", timer.elapsed(), index, peers);

});

pub fn new_identifier(&mut self) -> usize

Allocate a new worker-unique identifier.

This method is public, though it is not expected to be widely used outside of the timely dataflow system.

pub fn log_register(&self) -> RefMut<Registry<WorkerIdentifier>>

Access to named loggers.

Examples

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

    worker.log_register()
          .insert::<timely::logging::TimelyEvent,_>("timely", |time, data|
              println!("{:?}\t{:?}", time, data)
          );
});

pub fn dataflow<T, R, F>(&mut self, func: F) -> R where
    T: Refines<()>,
    F: FnOnce(&mut Child<Self, T>) -> R, 

Construct a new dataflow.

Examples

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

    // We must supply the timestamp type here, although
    // it would generally be determined by type inference.
    worker.dataflow::<usize,_,_>(|scope| {

        // uses of `scope` to build dataflow

    });
});

pub fn dataflow_core<T, R, F, V>(
    &mut self,
    name: &str,
    logging: Option<TimelyLogger>,
    resources: V,
    func: F
) -> R where
    T: Refines<()>,
    F: FnOnce(&mut V, &mut Child<Self, T>) -> R,
    V: Any + 'static, 

Construct a new dataflow with specific configurations.

This method constructs a new dataflow, using a name, logger, and additional resources specified as argument. The name is cosmetic, the logger is used to handle events generated by the dataflow, and the additional resources are kept alive for as long as the dataflow is alive (use case: shared library bindings).

Examples

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

    // We must supply the timestamp type here, although
    // it would generally be determined by type inference.
    worker.dataflow_core::<usize,_,_,_>(
        "dataflow X",           // Dataflow name
        None,                   // Optional logger
        37,                     // Any resources
        |resources, scope| {    // Closure

            // uses of `resources`, `scope`to build dataflow

        }
    );
});

Trait Implementations

impl<A: Allocate> AsWorker for Worker<A>

fn index(&self) -> usize

Index of the worker among its peers.

fn peers(&self) -> usize

Number of peer workers.

fn allocate<D: Data>(
    &mut self,
    identifier: usize,
    address: &[usize]
) -> (Vec<Box<dyn Push<Message<D>>>>, Box<dyn Pull<Message<D>>>)

Allocates a new channel from a supplied identifier and address.

fn pipeline<T: 'static>(
    &mut self,
    identifier: usize,
    address: &[usize]
) -> (ThreadPusher<Message<T>>, ThreadPuller<Message<T>>)

Constructs a pipeline channel from the worker to itself.

fn new_identifier(&mut self) -> usize

Allocates a new worker-unique identifier.

fn log_register(&self) -> RefMut<Registry<WorkerIdentifier>>

Provides access to named logging streams.

fn logging(&self) -> Option<TimelyLogger>

Provides access to the timely logging stream.

impl<A: Allocate> ScopeParent for Worker<A>

type Timestamp = ()

The timestamp associated with data in this scope.

impl<A: Allocate> Scheduler for Worker<A>

fn activations(&self) -> Rc<RefCell<Activations>>

Provides a shared handle to the activation scheduler.

fn activator_for(&self, path: &[usize]) -> Activator

impl<A: Allocate> Clone for Worker<A>

fn clone(&self) -> Self

Returns a copy of the value.

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

Performs copy-assignment from source.