Struct Scope 

pub struct Scope<'scope, T: Timestamp> { /* private fields */ }

A Scope manages the creation of new dataflow scopes, of operators and edges between them.

This is a shared object that can be freely copied, subject to its lifetime requirements. It manages scope construction through a RefCell-wrapped subgraph builder, and all of this type’s methods use but do not hold write access through the RefCell.

Implementations

impl<'scope, T: Timestamp> Scope<'scope, T>

pub fn worker(&self) -> &'scope Worker

Access to the underlying worker.

pub fn index(&self) -> usize

This worker’s index out of 0 .. self.peers().

pub fn peers(&self) -> usize

The total number of workers in the computation.

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

Provides a shared handle to the activation scheduler.

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

Constructs an Activator tied to the specified operator address.

pub fn name(&self) -> String

A useful name describing the scope.

pub fn addr(&self) -> Rc<[usize]>

A sequence of scope identifiers describing the path from the worker root to this scope.

pub fn add_edge(&self, source: Source, target: Target)

Connects a source of data with a target of the data. This only links the two for the purposes of tracking progress, rather than effect any data movement itself.

pub fn reserve_operator(&self) -> OperatorSlot<'scope, T>

Reserves a slot for an operator in this scope.

The returned OperatorSlot carries the scope-local index and worker-unique identifier of the future operator. It must be consumed by OperatorSlot::install before being dropped; otherwise it will panic, since the scope expects every reserved slot to eventually be filled.

pub fn scoped<T2, R, F>(&self, name: &str, func: F) -> R

where T2: Timestamp + Refines<T>, F: FnOnce(Scope<'_, T2>) -> R,

Creates a dataflow subgraph.

This method allows the user to create a nested scope with any timestamp that “refines” the enclosing timestamp (informally: extends it in a reversible way).

This is most commonly used to create new iterative contexts, and the provided method iterative for this task demonstrates the use of this method.

Examples

use timely::dataflow::operators::{Input, Enter, Leave};
use timely::order::Product;

timely::execute_from_args(std::env::args(), |worker| {
    // must specify types as nothing else drives inference.
    let input = worker.dataflow::<u64,_,_>(|child1| {
        let (input, stream) = child1.new_input::<Vec<String>>();
        let output = child1.scoped::<Product<u64,u32>,_,_>("ScopeName", |child2| {
            stream.enter(child2).leave(child1)
        });
        input
    });
});

pub fn scoped_raw<T2, R, F>( &self, name: &str, func: F, ) -> (R, Subgraph<T, T2>, OperatorSlot<'scope, T>)

where T2: Timestamp + Refines<T>, F: FnOnce(Scope<'_, T2>) -> R,

Creates a dataflow subgraph, runs a user closure, and returns a result and the to-be-assembled parts.

The returned subgraph must be registered in the operator slot.

pub fn iterative<T2, R, F>(&self, func: F) -> R

where T2: Timestamp, F: FnOnce(Scope<'_, Product<T, T2>>) -> R,

Creates an iterative dataflow subgraph.

This method is a specialization of scoped which uses the Product timestamp combinator, suitable for iterative computations in which iterative development at some time cannot influence prior iterations at a future time.

Examples

use timely::dataflow::operators::{Input, Enter, Leave};

timely::execute_from_args(std::env::args(), |worker| {
    // must specify types as nothing else drives inference.
    let input = worker.dataflow::<u64,_,_>(|child1| {
        let (input, stream) = child1.new_input::<Vec<String>>();
        let output = child1.iterative::<u32,_,_>(|child2| {
            stream.enter(child2).leave(child1)
        });
        input
    });
});

pub fn region<R, F>(&self, func: F) -> R

where F: FnOnce(Scope<'_, T>) -> R,

Creates a dataflow region with the same timestamp.

This method is a specialization of scoped which uses the same timestamp as the containing scope. It is used mainly to group regions of a dataflow computation, and provides some computational benefits by abstracting the specifics of the region.

Examples

use timely::dataflow::operators::{Input, Enter, Leave};

timely::execute_from_args(std::env::args(), |worker| {
    // must specify types as nothing else drives inference.
    let input = worker.dataflow::<u64,_,_>(|child1| {
        let (input, stream) = child1.new_input::<Vec<String>>();
        let output = child1.region(|child2| {
            stream.enter(child2).leave(child1)
        });
        input
    });
});

pub fn region_named<R, F>(&self, name: &str, func: F) -> R

where F: FnOnce(Scope<'_, T>) -> R,

Creates a dataflow region with the same timestamp and a supplied name.

This method is a specialization of scoped which uses the same timestamp as the containing scope. It is used mainly to group regions of a dataflow computation, and provides some computational benefits by abstracting the specifics of the region.

This variant allows you to specify a name for the region, which can be read out in the timely logging streams.

Examples

use timely::dataflow::operators::{Input, Enter, Leave};

timely::execute_from_args(std::env::args(), |worker| {
    // must specify types as nothing else drives inference.
    let input = worker.dataflow::<u64,_,_>(|child1| {
        let (input, stream) = child1.new_input::<Vec<String>>();
        let output = child1.region_named("region", |child2| {
            stream.enter(child2).leave(child1)
        });
        input
    });
});

Trait Implementations

impl<'scope, T: Timestamp> Clone for Scope<'scope, T>

fn clone(&self) -> Self

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl<'scope, T: Timestamp> Concatenate<'scope, T> for Scope<'scope, T>

fn concatenate<I, C: Container>(&self, sources: I) -> Stream<'scope, T, C>

where I: IntoIterator<Item = Stream<'scope, T, C>>,

Merge the contents of multiple streams.

impl<'scope, T: Timestamp> Debug for Scope<'scope, T>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<'scope, T: Timestamp> Feedback<'scope, T> for Scope<'scope, T>

fn feedback<C: Container>( &self, summary: <T as Timestamp>::Summary, ) -> (Handle<'scope, T, C>, Stream<'scope, T, C>)

Creates a Stream and a Handle to later bind the source of that Stream.

impl<'scope, T: Timestamp + TotalOrder> Input<'scope> for Scope<'scope, T>

type Timestamp = T

The timestamp at which this input scope operates.

fn new_input<C: Container + Clone>( &self, ) -> (Handle<T, CapacityContainerBuilder<C>>, Stream<'scope, T, C>)

Create a new Stream and Handle through which to supply input.

fn new_input_with_builder<CB: ContainerBuilder<Container: Clone>>( &self, ) -> (Handle<T, CB>, Stream<'scope, T, CB::Container>)

Create a new Stream and Handle through which to supply input.

fn input_from<CB: ContainerBuilder<Container: Clone>>( &self, handle: &mut Handle<T, CB>, ) -> Stream<'scope, T, CB::Container>

Create a new stream from a supplied interactive handle.

impl<'scope, T: Timestamp + TotalOrder> Input<'scope> for Scope<'scope, T>

type Timestamp = T

The timestamp at which this input scope operates.

fn new_input<D: Clone + 'static>( &self, ) -> (Handle<T, D>, StreamVec<'scope, T, D>)

Create a new StreamVec and Handle through which to supply input.

fn input_from<D: Clone + 'static>( &self, handle: &mut Handle<T, D>, ) -> StreamVec<'scope, T, D>

Create a new stream from a supplied interactive handle.

impl<'scope, T: Timestamp> UnorderedInput<'scope, T> for Scope<'scope, T>

fn new_unordered_input<CB: ContainerBuilder>( &self, ) -> ((UnorderedHandle<T, CB>, ActivateCapability<T>), Stream<'scope, T, CB::Container>)

Create a new capability-based Stream and UnorderedHandle through which to supply input. This input supports multiple open epochs (timestamps) at the same time.

impl<'scope, T: Timestamp> UnorderedInput<'scope, T> for Scope<'scope, T>

fn new_unordered_input<D: 'static>( &self, ) -> ((UnorderedHandle<T, D>, ActivateCapability<T>), StreamVec<'scope, T, D>)

Create a new capability-based StreamVec and Handle through which to supply input. This input supports multiple open epochs (timestamps) at the same time.

impl<'scope, T: Timestamp> Copy for Scope<'scope, T>