Struct OperatorBuilder

pub struct OperatorBuilder<G: Scope> {
    builder: OperatorBuilder<G>,
    activator: Activator,
    operator_waker: Arc<TimelyWaker>,
    input_frontiers: Vec<Antichain<G::Timestamp>>,
    input_queues: Vec<Box<dyn InputQueue<G::Timestamp>>>,
    output_flushes: Vec<Box<dyn FnMut()>>,
    shutdown_handle: ButtonHandle,
    shutdown_button: Button,
}

Builds async operators with generic shape.

Fields

builder: OperatorBuilder<G>

activator: Activator

The activator for this operator

operator_waker: Arc<TimelyWaker>

The waker set up to activate this timely operator when woken

input_frontiers: Vec<Antichain<G::Timestamp>>

The currently known upper frontier of each of the input handles.

input_queues: Vec<Box<dyn InputQueue<G::Timestamp>>>

Input queues for each of the declared inputs of the operator.

output_flushes: Vec<Box<dyn FnMut()>>

Holds type erased closures that flush an output handle when called. These handles will be automatically drained when the operator is scheduled after the logic future has been polled

shutdown_handle: ButtonHandle

A handle to check whether all workers have pressed the shutdown button.

shutdown_button: Button

A button to coordinate shutdown of this operator among workers.

Implementations

impl<G: Scope> OperatorBuilder<G>

pub fn new(name: String, scope: G) -> Self

Allocates a new generic async operator builder from its containing scope.

pub fn new_input_for<D, P>( &mut self, stream: &StreamCore<G, D>, pact: P, output: &dyn OutputIndex, ) -> AsyncInputHandle<G::Timestamp, D, ConnectedToOne>

where D: Container + 'static, P: ParallelizationContract<G::Timestamp, D>,

Adds a new input that is connected to the specified output, returning the async input handle to use.

pub fn new_input_for_many<const N: usize, D, P>( &mut self, stream: &StreamCore<G, D>, pact: P, outputs: [&dyn OutputIndex; N], ) -> AsyncInputHandle<G::Timestamp, D, ConnectedToMany<N>>

where D: Container + 'static, P: ParallelizationContract<G::Timestamp, D>,

Adds a new input that is connected to the specified outputs, returning the async input handle to use.

pub fn new_disconnected_input<D, P>( &mut self, stream: &StreamCore<G, D>, pact: P, ) -> AsyncInputHandle<G::Timestamp, D, Disconnected>

where D: Container + 'static, P: ParallelizationContract<G::Timestamp, D>,

Adds a new input that is not connected to any output, returning the async input handle to use.

pub fn new_input_connection<D, P, C>( &mut self, stream: &StreamCore<G, D>, pact: P, connection: C, ) -> AsyncInputHandle<G::Timestamp, D, C>

where D: Container + 'static, P: ParallelizationContract<G::Timestamp, D>, C: InputConnection<G::Timestamp> + 'static,

Adds a new input with connection information, returning the async input handle to use.

pub fn new_output<CB: ContainerBuilder>( &mut self, ) -> (AsyncOutputHandle<G::Timestamp, CB, Tee<G::Timestamp, CB::Container>>, StreamCore<G, CB::Container>)

Adds a new output, returning the output handle and stream.

pub fn build<B, L>(self, constructor: B) -> Button

where B: FnOnce(Vec<Capability<G::Timestamp>>) -> L, L: Future + 'static,

Creates an operator implementation from supplied logic constructor. It returns a shutdown button that when pressed it will cause the logic future to be dropped and input handles to be drained. The button can be converted into a token by using Button::press_on_drop

pub fn build_fallible<E: 'static, F>( self, constructor: F, ) -> (Button, StreamCore<G, Vec<Rc<E>>>)

where F: for<'a> FnOnce(&'a mut [CapabilitySet<G::Timestamp>]) -> Pin<Box<dyn Future<Output = Result<(), E>> + 'a>> + 'static,

Creates a fallible operator implementation from supplied logic constructor. If the Future resolves to an error it will be emitted in the returned error stream and then the operator will wait indefinitely until the shutdown button is pressed.

Capability handling

Unlike OperatorBuilder::build, this method does not give owned capabilities to the constructor. All initial capabilities are wrapped in a CapabilitySet and a mutable reference to them is given instead. This is done to avoid storing owned capabilities in the state of the logic future which would make using the ? operator unsafe, since the frontiers would incorrectly advance, potentially causing incorrect actions downstream.

builder.build_fallible(|caps| Box::pin(async move {
    // Assert that we have the number of capabilities we expect
    // `cap` will be a `&mut Option<Capability<T>>`:
    let [cap_set]: &mut [_; 1] = caps.try_into().unwrap();

    // Using cap to send data:
    output.give(&cap_set[0], 42);

    // Using cap_set to downgrade it:
    cap_set.downgrade([]);

    // Explicitly dropping the capability:
    // Simply running `drop(cap_set)` will only drop the reference and not the capability set itself!
    *cap_set = CapabilitySet::new();

    // !! BIG WARNING !!:
    // It is tempting to `take` the capability out of the set for convenience. This will
    // move the capability into the future state, tying its lifetime to it, which will get
    // dropped when an error is hit, causing incorrect progress statements.
    let cap = cap_set.delayed(&Timestamp::minimum());
    *cap_set = CapabilitySet::new(); // DO NOT DO THIS
}));

pub fn operator_info(&self) -> OperatorInfo

Creates operator info for the operator.

pub fn activator(&self) -> &Activator

Returns the activator for the operator.