Struct timely::worker::Worker

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

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

Implementations

impl<A: Allocate> Worker<A>

pub fn new(config: Config, 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_or_park(&mut self, duration: Option<Duration>) -> 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.

This method takes an optional timeout and may park the thread until there is work to perform or until this timeout expires. A value of None allows the worker to park indefinitely, whereas a value of Some(Duration::new(0, 0)) will return without parking the thread.

Examples

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

    use std::time::Duration;
    use timely::dataflow::operators::{ToStream, Inspect};

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

    worker.step_or_park(Some(Duration::from_secs(1)));
});

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

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

This method will continually execute even if there is not work for the worker to perform. Consider using the similar method Self::step_or_park_while(duration) to allow the worker to yield control if that is appropriate.

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 step_or_park_while<F: FnMut() -> bool>( &mut self, duration: Option<Duration>, func: F, )

Calls self.step_or_park(duration) as long as func evaluates to true.

This method may yield whenever there is no work to perform, as performed by Self::step_or_park(). Please consult the documentation for further information about that method and its behavior. In particular, the method can park the worker indefinitely, if no new work re-awakens the worker.

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_or_park_while(None, || 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 peek_identifier(&self) -> usize

The next worker-unique identifier to be allocated.

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_named<T, R, F>(&mut self, name: &str, func: F) -> R

where T: Refines<()>, F: FnOnce(&mut Child<'_, Self, T>) -> R,

Construct a new dataflow with a (purely cosmetic) name.

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_named::<usize,_,_>("Some Dataflow", |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

        }
    );
});

pub fn drop_dataflow(&mut self, dataflow_identifier: usize)

Drops an identified dataflow.

This method removes the identified dataflow, which will no longer be scheduled. Various other resources will be cleaned up, though the method is currently in public beta rather than expected to work. Please report all crashes and unmet expectations!

pub fn next_dataflow_index(&self) -> usize

Returns the next index to be used for dataflow construction.

This identifier will appear in the address of contained operators, and can be used to drop the dataflow using self.drop_dataflow().

pub fn installed_dataflows(&self) -> Vec<usize>

List the current dataflow indices.

pub fn has_dataflows(&self) -> bool

Returns true if there is at least one dataflow under management.

Trait Implementations

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

fn config(&self) -> &Config

Returns the worker configuration parameters.

fn index(&self) -> usize

Index of the worker among its peers.

fn peers(&self) -> usize

Number of peer workers.

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

Allocates a new channel from a supplied identifier and address.

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

Constructs a pipeline channel from the worker to itself.

fn new_identifier(&mut self) -> usize

Allocates a new worker-unique identifier.

fn peek_identifier(&self) -> usize

The next worker-unique identifier to be allocated.

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> 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.

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: Rc<[usize]>) -> Activator

Constructs an Activator tied to the specified operator address.

fn sync_activator_for(&self, path: Vec<usize>) -> SyncActivator

Constructs a SyncActivator tied to the specified operator address.

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

type Timestamp = ()

The timestamp associated with data in this scope.