Method build

src/timely-util/src/builder_async.rs:571–647

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`]

(self, constructor: B)

    /// be drained. The button can be converted into a token by using
    /// [`Button::press_on_drop`]
    pub fn build<B, L>(self, constructor: B) -> Button
    where
        B: FnOnce(Vec<Capability<T>>) -> L,
        L: Future + 'static,
    {
        let operator_waker = self.operator_waker;
        let mut input_frontiers = self.input_frontiers;
        let mut input_queues = self.input_queues;
        let mut output_flushes = self.output_flushes;
        let mut shutdown_handle = self.shutdown_handle;
        self.builder.build_reschedule(move |caps| {
            let mut logic_fut = Some(Box::pin(constructor(caps)));
            move |new_frontiers| {
                operator_waker.active.store(true, Ordering::SeqCst);
                for (i, queue) in input_queues.iter_mut().enumerate() {
                    // First, discover if there are any frontier notifications
                    let cur = &mut input_frontiers[i];
                    let new = new_frontiers[i].frontier();
                    if PartialOrder::less_than(&cur.borrow(), &new) {
                        queue.notify_progress(new.to_owned());
                        *cur = new.to_owned();
                    }
                    // Then accept all input into local queues. This step registers the received
                    // messages with progress tracking.
                    queue.accept_input();
                }
                operator_waker.active.store(false, Ordering::SeqCst);

                // If our worker pressed the button we stop scheduling the logic future and/or
                // draining the input handles to stop producing data and frontier updates
                // downstream.
                if shutdown_handle.local_pressed() {
                    // When all workers press their buttons we drop the logic future and start
                    // draining the input handles.
                    if shutdown_handle.all_pressed() {
                        logic_fut = None;
                        for queue in input_queues.iter_mut() {
                            queue.drain_input();
                        }
                        false
                    } else {
                        true
                    }
                } else {
                    // Schedule the logic future if any of the wakers above marked the task as ready
                    if let Some(fut) = logic_fut.as_mut() {
                        if operator_waker.task_ready.load(Ordering::SeqCst) {
                            let waker = futures_util::task::waker_ref(&operator_waker);
                            let mut cx = Context::from_waker(&waker);
                            operator_waker.task_ready.store(false, Ordering::SeqCst);
                            if Pin::new(fut).poll(&mut cx).is_ready() {
                                // We're done with logic so deallocate the task
                                logic_fut = None;
                            }
                            // Flush all the outputs before exiting
                            for flush in output_flushes.iter_mut() {
                                (flush)();
                            }