timely/

worker.rs

//! The root of each single-threaded worker.

use std::rc::Rc;
use std::cell::{RefCell, RefMut};
use std::any::Any;
use std::str::FromStr;
use std::time::{Instant, Duration};
use std::collections::HashMap;
use std::collections::hash_map::Entry;
use std::sync::Arc;

use crate::communication::{Allocator, Exchangeable, Push, Pull};
use crate::communication::allocator::thread::{ThreadPusher, ThreadPuller};
use crate::scheduling::{Schedule, Activations, Activator, SyncActivator};
use crate::progress::timestamp::{Refines};
use crate::progress::SubgraphBuilder;
use crate::progress::operate::Operate;
use crate::dataflow::scope::Scope;
use crate::logging::TimelyLogger;

/// Different ways in which timely's progress tracking can work.
///
/// These options drive some buffering and accumulation that timely
/// can do to try and trade volume of progress traffic against latency.
/// By accumulating updates longer, a smaller total volume of messages
/// are sent.
///
/// The `ProgressMode::Demand` variant is the most robust, and least
/// likely to lead to catastrophic performance. The `Eager` variant
/// is useful for getting the smallest latencies on systems with few
/// workers, but does risk saturating the system with progress messages
/// and should be used with care, or not at all.
///
/// If you are not certain which option to use, prefer `Demand`, and
/// perhaps monitor the progress messages through timely's logging
/// infrastructure to see if their volume is surprisingly high.
#[derive(Debug, Default, Clone, Copy, Eq, PartialEq)]
pub enum ProgressMode {
    /// Eagerly transmit all progress updates produced by a worker.
    ///
    /// Progress messages are transmitted without consideration for the
    /// possibility that they may unblock other workers. This can result
    /// in a substantial volume of messages that do not result in a
    /// change to the lower bound of outstanding work.
    Eager,
    /// Delay transmission of progress updates until any could advance
    /// the global frontier of timestamps.
    ///
    /// As timely executes, the progress messages inform each worker of
    /// the outstanding work remaining in the system. As workers work,
    /// they produce changes to this outstanding work. This option
    /// delays the communication of those changes until they might
    /// possibly cause a change in the lower bound of all outstanding
    /// work.
    ///
    /// The most common case this remedies is when one worker transmits
    /// messages to other workers, that worker holds a capability for the
    /// operator and timestamp. Other workers will receive messages, and
    /// with this option will not immediately acknowledge receiving the
    /// messages, because the held capability is strictly prior to what
    /// the messages can affect. Once the capability is released, the
    /// progress messages are unblocked and transmitted, in accumulated
    /// form.
    #[default]
    Demand,
}

impl FromStr for ProgressMode {
    type Err = String;

    fn from_str(s: &str) -> Result<ProgressMode, String> {
        match s {
            "eager" => Ok(ProgressMode::Eager),
            "demand" => Ok(ProgressMode::Demand),
            _ => Err(format!("unknown progress mode: {}", s)),
        }
    }
}

/// Worker configuration.
#[derive(Debug, Clone)]
pub struct Config {
    /// The progress mode to use.
    pub(crate) progress_mode: ProgressMode,
    /// How long an idle worker polls for new events before parking its thread.
    ///
    /// Parking a thread and waking it again costs several microseconds, which
    /// dominates the cost of fine-grained coordination among workers (for
    /// example, a barrier per iteration of a loop). A worker with nothing to do
    /// first polls its channels for this long, and parks only if nothing arrives.
    /// The polling occupies a core, so the duration bounds the CPU an idle worker
    /// burns each time it goes idle.
    pub(crate) idle_spin: Duration,
    /// A map from parameter name to typed parameter values.
    registry: HashMap<String, Arc<dyn Any + Send + Sync>>,
}

impl Default for Config {
    fn default() -> Self {
        Config {
            progress_mode: ProgressMode::default(),
            idle_spin: Duration::from_micros(Self::DEFAULT_IDLE_SPIN_MICROS),
            registry: HashMap::new(),
        }
    }
}

impl Config {
    /// The default `idle_spin`, in microseconds.
    ///
    /// Parking and unparking a thread costs a few microseconds on common
    /// platforms, so a worker that polls for this long before parking avoids
    /// that cost for the short waits typical of tightly coupled workers, at
    /// the expense of at most this much CPU each time it goes idle.
    pub const DEFAULT_IDLE_SPIN_MICROS: u64 = 10;

    /// Installs options into a [getopts::Options] struct that correspond
    /// to the parameters in the configuration.
    ///
    /// It is the caller's responsibility to ensure that the installed options
    /// do not conflict with any other options that may exist in `opts`, or
    /// that may be installed into `opts` in the future.
    ///
    /// This method is only available if the `getopts` feature is enabled, which
    /// it is by default.
    #[cfg(feature = "getopts")]
    pub fn install_options(opts: &mut getopts::Options) {
        opts.optopt("", "progress-mode", "progress tracking mode (eager or demand)", "MODE");
        opts.optopt("", "idle-spin", "microseconds an idle worker polls before parking", "MICROS");
    }

    /// Instantiates a configuration based upon the parsed options in `matches`.
    ///
    /// The `matches` object must have been constructed from a
    /// [getopts::Options] which contained at least the options installed by
    /// [Self::install_options].
    ///
    /// This method is only available if the `getopts` feature is enabled, which
    /// it is by default.
    #[cfg(feature = "getopts")]
    pub fn from_matches(matches: &getopts::Matches) -> Result<Config, String> {
        let progress_mode = matches
            .opt_get_default("progress-mode", ProgressMode::Demand)?;
        let mut config = Config::default().progress_mode(progress_mode);
        if let Some(micros) = matches.opt_get::<u64>("idle-spin").map_err(|e| e.to_string())? {
            config = config.idle_spin(Duration::from_micros(micros));
        }
        Ok(config)
    }

    /// Sets the progress mode to `progress_mode`.
    pub fn progress_mode(mut self, progress_mode: ProgressMode) -> Self {
        self.progress_mode = progress_mode;
        self
    }

    /// Sets how long an idle worker polls for events before parking.
    pub fn idle_spin(mut self, idle_spin: Duration) -> Self {
        self.idle_spin = idle_spin;
        self
    }

    /// Sets a typed configuration parameter for the given `key`.
    ///
    /// It is recommended to install a single configuration struct using a key
    /// that uniquely identifies your project, to avoid clashes. For example,
    /// differential dataflow registers a configuration struct under the key
    /// "differential".
    ///
    /// # Examples
    /// ```rust
    /// let mut config = timely::Config::process(3);
    /// config.worker.set("example".to_string(), 7u64);
    /// timely::execute(config, |worker| {
    ///    assert_eq!(worker.config().get::<u64>("example"), Some(&7));
    /// }).unwrap();
    /// ```
    pub fn set<T>(&mut self, key: String, val: T) -> &mut Self
    where
        T: Send + Sync + 'static,
    {
        self.registry.insert(key, Arc::new(val));
        self
    }

    /// Gets the value for configured parameter `key`.
    ///
    /// Returns `None` if `key` has not previously been set with
    /// [Config::set], or if the specified `T` does not match the `T`
    /// from the call to `set`.
    ///
    /// # Examples
    /// ```rust
    /// let mut config = timely::Config::process(3);
    /// config.worker.set("example".to_string(), 7u64);
    /// timely::execute(config, |worker| {
    ///    assert_eq!(worker.config().get::<u64>("example"), Some(&7));
    /// }).unwrap();
    /// ```
    pub fn get<T: 'static>(&self, key: &str) -> Option<&T> {
        self.registry.get(key).and_then(|val| val.downcast_ref())
    }
}


/// A `Worker` is the entry point to a timely dataflow computation. It wraps an `Allocator`
/// and has a list of dataflows that it manages.
pub struct Worker {
    config: Config,
    /// An optional instant from which the start of the computation should be reckoned.
    ///
    /// If this is set to none, system time-based functionality will be unavailable or work badly.
    /// For example, logging will be unavailable, and activation after a delay will be unavailable.
    timer: Option<Instant>,
    paths: Rc<RefCell<HashMap<usize, Rc<[usize]>>>>,
    allocator: Rc<RefCell<Allocator>>,
    identifiers: Rc<RefCell<usize>>,
    // dataflows: Rc<RefCell<Vec<Wrapper>>>,
    dataflows: Rc<RefCell<HashMap<usize, Wrapper>>>,
    dataflow_counter: Rc<RefCell<usize>>,
    logging: Option<Rc<RefCell<crate::logging_core::Registry>>>,

    activations: Rc<RefCell<Activations>>,
    active_dataflows: Vec<usize>,

    // Temporary storage for channel identifiers during dataflow construction.
    // These are then associated with a dataflow once constructed.
    temp_channel_ids: Rc<RefCell<Vec<usize>>>,
}



impl Worker {
    /// Allocates a new `Worker` bound to a channel allocator.
    pub fn new(config: Config, c: Allocator, now: Option<std::time::Instant>) -> Worker {
        Worker {
            config,
            timer: now,
            paths:  Default::default(),
            allocator: Rc::new(RefCell::new(c)),
            identifiers:  Default::default(),
            dataflows: Default::default(),
            dataflow_counter:  Default::default(),
            logging: now.map(|now| Rc::new(RefCell::new(crate::logging_core::Registry::new(now)))),
            activations: Rc::new(RefCell::new(Activations::new(now))),
            active_dataflows: Default::default(),
            temp_channel_ids:  Default::default(),
        }
    }

    /// 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)
    ///             .container::<Vec<_>>()
    ///             .inspect(|x| println!("{:?}", x));
    ///     });
    ///
    ///     worker.step();
    /// });
    /// ```
    pub fn step(&mut self) -> bool {
        self.step_or_park(Some(Duration::from_secs(0)))
    }

    /// 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)
    ///             .container::<Vec<_>>()
    ///             .inspect(|x| println!("{:?}", x));
    ///     });
    ///
    ///     worker.step_or_park(Some(Duration::from_secs(1)));
    /// });
    /// ```
    pub fn step_or_park(&mut self, duration: Option<Duration>) -> bool {

        // Determine the minimum park duration, where `None` are an absence of a constraint.
        let mut delay = self.poll_events(duration);

        // An idle worker polls for a while before parking, as parking and
        // unparking a thread costs more than a short wait usually lasts.
        if delay != Some(Duration::new(0,0)) {
            let budget = match delay {
                Some(delay) => std::cmp::min(delay, self.config.idle_spin),
                None => self.config.idle_spin,
            };
            if budget > Duration::new(0,0) {
                let start = Instant::now();
                let mut polls = 0u32;
                loop {
                    std::hint::spin_loop();
                    delay = self.poll_events(duration);
                    if delay == Some(Duration::new(0,0)) { break; }
                    // Consult the clock only occasionally, as it is not free.
                    polls = polls.wrapping_add(1);
                    if polls % 16 == 0 && start.elapsed() >= budget { break; }
                }
            }
        }

        if delay != Some(Duration::new(0,0)) {

            // Log parking and flush log.
            if let Some(l) = self.logging().as_mut() {
                l.log(crate::logging::ParkEvent::park(delay));
                l.flush();
            }

            self.allocator
                .borrow()
                .await_events(delay);

            // Log return from unpark.
            self.logging().as_mut().map(|l| l.log(crate::logging::ParkEvent::unpark()));
        }
        else {   // Schedule active dataflows.

            let active_dataflows = &mut self.active_dataflows;
            self.activations
                .borrow_mut()
                .for_extensions(&[], |index| active_dataflows.push(index));

            let mut dataflows = self.dataflows.borrow_mut();
            for index in active_dataflows.drain(..) {
                // Step dataflow if it exists, remove if not incomplete.
                if let Entry::Occupied(mut entry) = dataflows.entry(index) {
                    // TODO: This is a moment at which a scheduling decision is being made.
                    let incomplete = entry.get_mut().step();
                    if !incomplete {
                        let mut paths = self.paths.borrow_mut();
                        for channel in entry.get_mut().channel_ids.drain(..) {
                            paths.remove(&channel);
                        }
                        entry.remove_entry();
                    }
                }
            }
        }

        // Clean up, indicate if dataflows remain.
        self.logging.as_ref().map(|l| l.borrow_mut().flush());
        self.allocator.borrow_mut().release();
        !self.dataflows.borrow().is_empty()
    }

    /// Surfaces channel events as activations, and reports how long the worker may idle.
    ///
    /// Returns the minimum of `duration` and the time until the next scheduled
    /// activation, where `None` is an absence of any constraint, and `Some(0)`
    /// means there is work to do now.
    fn poll_events(&self, duration: Option<Duration>) -> Option<Duration> {

        {   // Process channel events. Activate responders.
            let mut allocator = self.allocator.borrow_mut();
            allocator.receive();
            let events = allocator.events();
            let mut borrow = events.borrow_mut();
            let paths = self.paths.borrow();
            borrow.sort_unstable();
            borrow.dedup();
            for channel in borrow.drain(..) {
                // Consider tracking whether a channel
                // in non-empty, and only activating
                // on the basis of non-empty channels.
                // TODO: This is a sloppy way to deal
                // with channels that may not be alloc'd.
                if let Some(path) = paths.get(&channel) {
                    self.activations
                        .borrow_mut()
                        .activate(&path[..]);
                }
            }
        }

        // Organize activations.
        self.activations
            .borrow_mut()
            .advance();

        // Consider parking only if we have no pending events, some dataflows, and a non-zero duration.
        let empty_for = self.activations.borrow().empty_for();
        match (duration, empty_for) {
            (Some(x), Some(y)) => Some(std::cmp::min(x,y)),
            (x, y) => x.or(y),
        }
    }

    /// 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)
    ///             .container::<Vec<_>>()
    ///             .inspect(|x| println!("{:?}", x))
    ///             .probe()
    ///             .0
    ///     });
    ///
    ///     worker.step_while(|| probe.less_than(&0));
    /// });
    /// ```
    pub fn step_while<F: FnMut()->bool>(&mut self, func: F) {
        self.step_or_park_while(Some(Duration::from_secs(0)), func)
    }

    /// 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)
    ///             .container::<Vec<_>>()
    ///             .inspect(|x| println!("{:?}", x))
    ///             .probe()
    ///             .0
    ///     });
    ///
    ///     worker.step_or_park_while(None, || probe.less_than(&0));
    /// });
    /// ```
    pub fn step_or_park_while<F: FnMut()->bool>(&mut self, duration: Option<Duration>, mut func: F) {
        while func() { self.step_or_park(duration); }
    }

    /// 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().unwrap();
    ///
    ///     println!("{:?}\tWorker {} of {}", timer.elapsed(), index, peers);
    ///
    /// });
    /// ```
    pub fn index(&self) -> usize { self.allocator.borrow().index() }
    /// 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().unwrap();
    ///
    ///     println!("{:?}\tWorker {} of {}", timer.elapsed(), index, peers);
    ///
    /// });
    /// ```
    pub fn peers(&self) -> usize { self.allocator.borrow().peers() }

    /// 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().unwrap();
    ///
    ///     println!("{:?}\tWorker {} of {}", timer.elapsed(), index, peers);
    ///
    /// });
    /// ```
    pub fn timer(&self) -> Option<Instant> { self.timer }

    /// 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 new_identifier(&self) -> usize {
        *self.identifiers.borrow_mut() += 1;
        *self.identifiers.borrow() - 1
    }

    /// The next worker-unique identifier to be allocated.
    pub fn peek_identifier(&self) -> usize {
        *self.identifiers.borrow()
    }

    /// Access to named loggers.
    ///
    /// # Examples
    ///
    /// ```
    /// timely::execute_from_args(::std::env::args(), |worker| {
    ///
    ///     worker.log_register()
    ///           .unwrap()
    ///           .insert::<timely::logging::TimelyEventBuilder,_>("timely", |time, data|
    ///               println!("{:?}\t{:?}", time, data)
    ///           );
    /// });
    /// ```
    pub fn log_register(&self) -> Option<RefMut<'_, crate::logging_core::Registry>> {
        self.logging.as_ref().map(|l| l.borrow_mut())
    }

    /// Returns the worker configuration parameters.
    pub fn config(&self) -> &Config { &self.config }

    /// Provides a shared handle to the activation scheduler.
    pub fn activations(&self) -> Rc<RefCell<Activations>> {
        Rc::clone(&self.activations)
    }

    /// Constructs an `Activator` tied to the specified operator address.
    pub fn activator_for(&self, path: Rc<[usize]>) -> Activator {
        Activator::new(path, self.activations())
    }

    /// Constructs a `SyncActivator` tied to the specified operator address.
    pub fn sync_activator_for(&self, path: Vec<usize>) -> SyncActivator {
        SyncActivator::new(path, self.activations().borrow().sync())
    }

    /// Acquires a logger by name, if the log register exists and the name is registered.
    pub fn logger_for<CB: crate::ContainerBuilder>(&self, name: &str) -> Option<timely_logging::Logger<CB>> {
        self.log_register().and_then(|l| l.get(name))
    }

    /// Provides access to the timely logging stream.
    pub fn logging(&self) -> Option<crate::logging::TimelyLogger> { self.logger_for("timely").map(Into::into) }

    /// Allocates a new channel from a supplied identifier and address.
    pub fn allocate<D: Exchangeable>(&self, identifier: usize, address: Rc<[usize]>) -> (Vec<Box<dyn Push<D>>>, Box<dyn Pull<D>>) {
        if address.is_empty() { panic!("Unacceptable address: Length zero"); }
        let mut paths = self.paths.borrow_mut();
        paths.insert(identifier, address);
        self.temp_channel_ids.borrow_mut().push(identifier);
        self.allocator.borrow_mut().allocate(identifier)
    }

    /// Constructs a pipeline channel from the worker to itself.
    pub fn pipeline<T: 'static>(&self, identifier: usize, address: Rc<[usize]>) -> (ThreadPusher<T>, ThreadPuller<T>) {
        if address.is_empty() { panic!("Unacceptable address: Length zero"); }
        let mut paths = self.paths.borrow_mut();
        paths.insert(identifier, address);
        self.temp_channel_ids.borrow_mut().push(identifier);
        self.allocator.borrow_mut().pipeline(identifier)
    }

    /// Allocates a broadcast channel, where each pushed message is received by all.
    pub fn broadcast<T: Exchangeable + Clone>(&self, identifier: usize, address: Rc<[usize]>) -> (Box<dyn Push<T>>, Box<dyn Pull<T>>) {
        if address.is_empty() { panic!("Unacceptable address: Length zero"); }
        let mut paths = self.paths.borrow_mut();
        paths.insert(identifier, address);
        self.temp_channel_ids.borrow_mut().push(identifier);
        self.allocator.borrow_mut().broadcast(identifier)
    }

    /// 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<T, R, F>(&mut self, func: F) -> R
    where
        T: Refines<()>,
        F: FnOnce(Scope<T>)->R,
    {
        self.dataflow_core("Dataflow", self.logging(), Box::new(()), |_, child| func(child))
    }

    /// 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_named<T, R, F>(&mut self, name: &str, func: F) -> R
    where
        T: Refines<()>,
        F: FnOnce(Scope<T>)->R,
    {
        self.dataflow_core(name, self.logging(), Box::new(()), |_, child| func(child))
    }

    /// 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 dataflow_core<T, R, F, V>(&mut self, name: &str, mut logging: Option<TimelyLogger>, mut resources: V, func: F) -> R
    where
        T: Refines<()>,
        F: FnOnce(&mut V, Scope<T>)->R,
        V: Any+'static,
    {
        let dataflow_index = self.allocate_dataflow_index();
        let addr = vec![dataflow_index].into();
        let identifier = self.new_identifier();

        let subscope = SubgraphBuilder::new_from(addr, identifier, name);
        let subscope = RefCell::new(subscope);

        let result = {
            let builder = Scope {
                subgraph: &subscope,
                worker: self,
            };
            func(&mut resources, builder)
        };

        let operator = subscope.into_inner().build(self);

        if let Some(l) = logging.as_mut() {
            l.log(crate::logging::OperatesEvent {
                id: identifier,
                addr: operator.path().to_vec(),
                name: operator.name().to_string(),
            });
            l.flush();
        }

        let (_, _, operator) = Box::new(operator).initialize();

        let mut temp_channel_ids = self.temp_channel_ids.borrow_mut();
        let channel_ids = temp_channel_ids.drain(..).collect::<Vec<_>>();

        let wrapper = Wrapper {
            logging,
            identifier,
            operate: Some(operator),
            resources: Some(Box::new(resources)),
            channel_ids,
        };
        self.dataflows.borrow_mut().insert(dataflow_index, wrapper);

        result

    }

    /// 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 drop_dataflow(&mut self, dataflow_identifier: usize) {
        if let Some(mut entry) = self.dataflows.borrow_mut().remove(&dataflow_identifier) {
            // Garbage collect channel_id to path information.
            let mut paths = self.paths.borrow_mut();
            for channel in entry.channel_ids.drain(..) {
                paths.remove(&channel);
            }
        }
    }

    /// 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 next_dataflow_index(&self) -> usize {
        *self.dataflow_counter.borrow()
    }

    /// List the current dataflow indices.
    pub fn installed_dataflows(&self) -> Vec<usize> {
        self.dataflows.borrow().keys().cloned().collect()
    }

    /// Returns `true` if there is at least one dataflow under management.
    pub fn has_dataflows(&self) -> bool {
        !self.dataflows.borrow().is_empty()
    }

    // Acquire a new distinct dataflow identifier.
    fn allocate_dataflow_index(&self) -> usize {
        *self.dataflow_counter.borrow_mut() += 1;
        *self.dataflow_counter.borrow() - 1
    }
}

impl Clone for Worker {
    fn clone(&self) -> Self {
        Worker {
            config: self.config.clone(),
            timer: self.timer,
            paths: Rc::clone(&self.paths),
            allocator: Rc::clone(&self.allocator),
            identifiers: Rc::clone(&self.identifiers),
            dataflows: Rc::clone(&self.dataflows),
            dataflow_counter: Rc::clone(&self.dataflow_counter),
            logging: self.logging.clone(),
            activations: Rc::clone(&self.activations),
            active_dataflows: Vec::new(),
            temp_channel_ids: Rc::clone(&self.temp_channel_ids),
        }
    }
}

struct Wrapper {
    logging: Option<TimelyLogger>,
    identifier: usize,
    operate: Option<Box<dyn Schedule>>,
    resources: Option<Box<dyn Any>>,
    channel_ids: Vec<usize>,
}

impl Wrapper {
    /// Steps the dataflow, indicates if it remains incomplete.
    ///
    /// If the dataflow is incomplete, this call will drop it and its resources,
    /// dropping the dataflow first and then the resources (so that, e.g., shared
    /// library bindings will outlive the dataflow).
    fn step(&mut self) -> bool {

        // Perhaps log information about the start of the schedule call.
        if let Some(l) = self.logging.as_mut() {
            l.log(crate::logging::ScheduleEvent::start(self.identifier));
        }

        let incomplete = {
            let _current = crate::scheduling::current::enter(self.identifier);
            self.operate.as_mut().map(|op| op.schedule()).unwrap_or(false)
        };
        if !incomplete {
            self.operate = None;
            self.resources = None;
        }

        // Perhaps log information about the stop of the schedule call.
        if let Some(l) = self.logging.as_mut() {
            l.log(crate::logging::ScheduleEvent::stop(self.identifier));
        }

        incomplete
    }
}

impl Drop for Wrapper {
    fn drop(&mut self) {
        if let Some(l) = self.logging.as_mut() {
            l.log(crate::logging::ShutdownEvent { id: self.identifier });
        }
        // ensure drop order
        self.operate = None;
        self.resources = None;
    }
}