Make std::time::Instant optional

communication/src/allocator/generic.rs

@@ -113,15 +113,6 @@ impl Allocate for Generic {
     fn receive(&mut self) { self.receive(); }
     fn release(&mut self) { self.release(); }
     fn events(&self) -> &Rc<RefCell<Vec<usize>>> { self.events() }
-    fn await_events(&self, _duration: Option<std::time::Duration>) {
-        match self {
-            Generic::Thread(t) => t.await_events(_duration),
-            Generic::Process(p) => p.await_events(_duration),
-            Generic::ProcessBinary(pb) => pb.await_events(_duration),
-            Generic::ZeroCopy(z) => z.await_events(_duration),
-            Generic::ZeroCopyBinary(z) => z.await_events(_duration),
-        }
-    }
 }
 
 

communication/src/allocator/mod.rs

@@ -2,7 +2,6 @@
 
 use std::rc::Rc;
 use std::cell::RefCell;
-use std::time::Duration;
 
 pub use self::thread::Thread;
 pub use self::process::Process;
@@ -57,14 +56,6 @@ pub trait Allocate {
     /// into a performance problem.
     fn events(&self) -> &Rc<RefCell<Vec<usize>>>;
 
-    /// Awaits communication events.
-    ///
-    /// This method may park the current thread, for at most `duration`,
-    /// until new events arrive.
-    /// The method is not guaranteed to wait for any amount of time, but
-    /// good implementations should use this as a hint to park the thread.
-    fn await_events(&self, _duration: Option<Duration>) { }
-
     /// Ensure that received messages are surfaced in each channel.
     ///
     /// This method should be called to ensure that received messages are

communication/src/allocator/process.rs

@@ -4,7 +4,6 @@ use std::rc::Rc;
 use std::cell::RefCell;
 use std::sync::{Arc, Mutex};
 use std::any::Any;
-use std::time::Duration;
 use std::collections::{HashMap};
 use std::sync::mpsc::{Sender, Receiver};
 
@@ -182,10 +181,6 @@ impl Allocate for Process {
         self.inner.events()
     }
 
-    fn await_events(&self, duration: Option<Duration>) {
-        self.inner.await_events(duration);
-    }
-
     fn receive(&mut self) {
         let mut events = self.inner.events().borrow_mut();
         while let Ok(index) = self.counters_recv.try_recv() {

communication/src/allocator/thread.rs

@@ -2,7 +2,6 @@
 
 use std::rc::Rc;
 use std::cell::RefCell;
-use std::time::Duration;
 use std::collections::VecDeque;
 
 use crate::allocator::{Allocate, AllocateBuilder};
@@ -36,16 +35,6 @@ impl Allocate for Thread {
     fn events(&self) -> &Rc<RefCell<Vec<usize>>> {
         &self.events
     }
-    fn await_events(&self, duration: Option<Duration>) {
-        if self.events.borrow().is_empty() {
-            if let Some(duration) = duration {
-                std::thread::park_timeout(duration);
-            }
-            else {
-                std::thread::park();
-            }
-        }
-    }
 }
 
 /// Thread-local counting channel push endpoint.

communication/src/allocator/zero_copy/allocator.rs

@@ -314,7 +314,4 @@ impl<A: Allocate> Allocate for TcpAllocator<A> {
     fn events(&self) -> &Rc<RefCell<Vec<usize>>> {
         self.inner.events()
     }
-    fn await_events(&self, duration: Option<std::time::Duration>) {
-        self.inner.await_events(duration);
-    }
 }

communication/src/allocator/zero_copy/allocator_process.rs

@@ -242,14 +242,4 @@ impl Allocate for ProcessAllocator {
     fn events(&self) -> &Rc<RefCell<Vec<usize>>> {
         &self.events
     }
-    fn await_events(&self, duration: Option<std::time::Duration>) {
-        if self.events.borrow().is_empty() {
-            if let Some(duration) = duration {
-                std::thread::park_timeout(duration);
-            }
-            else {
-                std::thread::park();
-            }
-        }
-    }
 }

timely/src/progress/subgraph.rs

@@ -300,10 +300,11 @@
         self.propagate_pointstamps();
 
         {   // Enqueue active children; scoped to let borrow drop.
+            use crate::scheduling::activate::Scheduler;
             let temp_active = &mut self.temp_active;
             self.activations
                 .borrow_mut()
-                .for_extensions(&self.path[..], |index| temp_active.push(Reverse(index)));
+                .extensions(&self.path[..], temp_active);
         }
 
         // Schedule child operators.
@@ -784,7 +785,7 @@
             for (time, diff) in internal.iter() {
                 if *diff > 0 {
                     let consumed = shared_progress.consumeds.iter_mut().any(|x| x.iter().any(|(t,d)| *d > 0 && t.less_equal(time)));
                     let internal = child_state.sources[output].implications.less_equal(time);
                     if !consumed && !internal {
                         println!("Increment at {:?}, not supported by\n\tconsumed: {:?}\n\tinternal: {:?}", time, shared_progress.consumeds, child_state.sources[output].implications);
                         panic!("Progress error; internal {:?}", self.name);

timely/src/scheduling/activate.rs

@@ -8,30 +8,45 @@ use std::time::{Duration, Instant};
 use std::cmp::Reverse;
 use crossbeam_channel::{Sender, Receiver};
 
-/// Methods required to act as a timely scheduler.
+/// Methods required to act as a scheduler for timely operators.
 ///
-/// The core methods are the activation of "paths", sequences of integers, and
-/// the enumeration of active paths by prefix. A scheduler may delay the report
-/// of a path indefinitely, but it should report at least one extension for the
-/// empty path `&[]` or risk parking the worker thread without a certain unpark.
+/// Operators are described by "paths" of integers, indicating the path along
+/// a tree of regions, arriving at the the operator. Each path is either "idle"
+/// or "active", where the latter indicates that someone has requested that the
+/// operator be scheduled by the worker. Operators go from idle to active when
+/// the `activate(path)` method is called, and from active to idle when the path
+/// is returned through a call to `extensions(path, _)`.
 ///
-/// There is no known harm to "spurious wake-ups" where a not-active path is
-/// returned through `extensions()`.
+/// The worker will continually probe for extensions to the root empty path `[]`,
+/// and then follow all returned addresses, recursively. A scheduler need not
+/// schedule all active paths, but it should return *some* active path when the
+/// worker probes the empty path, or the worker may put the thread to sleep.
+///
+/// There is no known harm to scheduling an idle path.
+/// The worker may speculatively schedule paths of its own accord.
 pub trait Scheduler {
     /// Mark a path as immediately scheduleable.
+    ///
+    /// The scheduler is not required to immediately schedule the path, but it
+    /// should not signal that it has no work until the path has been scheduled.
     fn activate(&mut self, path: &[usize]);
     /// Populates `dest` with next identifiers on active extensions of `path`.
     ///
     /// This method is where a scheduler is allowed to exercise some discretion,
     /// in that it does not need to present *all* extensions, but it can instead
-    /// present only those that the runtime should schedule.
-    fn extensions(&mut self, path: &[usize], dest: &mut Vec<usize>);
+    /// present only those that the runtime should immediately schedule.
+    ///
+    /// The worker *will* schedule all extensions before probing new prefixes.
+    /// The scheduler is invited to rely on this, and to schedule in "batches",
+    /// where the next time the worker probes for extensions to the empty path
+    /// then all addresses in the batch have certainly been scheduled.
+    fn extensions(&mut self, path: &[usize], dest: &mut BinaryHeap<Reverse<usize>>);
 }
 
 // Trait objects can be schedulers too.
 impl Scheduler for Box<dyn Scheduler> {
     fn activate(&mut self, path: &[usize]) { (**self).activate(path) }
-    fn extensions(&mut self, path: &[usize], dest: &mut Vec<usize>) { (**self).extensions(path, dest) }
+    fn extensions(&mut self, path: &[usize], dest: &mut BinaryHeap<Reverse<usize>>) { (**self).extensions(path, dest) }
 }
 
 /// Allocation-free activation tracker.
@@ -93,7 +108,7 @@ impl Activations {
     }
 
     /// Discards the current active set and presents the next active set.
-    pub fn advance(&mut self) {
+    fn advance(&mut self) {
 
         // Drain inter-thread activations.
         while let Ok(path) = self.rx.try_recv() {
@@ -130,15 +145,15 @@ impl Activations {
         self.clean = self.bounds.len();
     }
 
-    /// Maps a function across activated paths.
-    pub fn map_active(&self, logic: impl Fn(&[usize])) {
-        for (offset, length) in self.bounds.iter() {
-            logic(&self.slices[*offset .. (*offset + *length)]);
-        }
-    }
-
     /// Sets as active any symbols that follow `path`.
-    pub fn for_extensions(&self, path: &[usize], mut action: impl FnMut(usize)) {
+    fn for_extensions(&mut self, path: &[usize], mut action: impl FnMut(usize)) {
+
+        // Each call for the root path is a moment where the worker has reset.
+        // This relies on a worker implementation that follows the scheduling
+        // instructions perfectly; if any offered paths are not explored, oops.
+        if path.is_empty() {
+            self.advance();
+        }
 
         let position =
         self.bounds[..self.clean]
@@ -179,7 +194,7 @@ impl Activations {
     /// This method should be used before putting a worker thread to sleep, as it
     /// indicates the amount of time before the thread should be unparked for the
     /// next scheduled activation.
-    pub fn empty_for(&self) -> Option<Duration> {
+    fn empty_for(&self) -> Option<Duration> {
         if !self.bounds.is_empty() || self.timer.is_none() {
             Some(Duration::new(0,0))
         }
@@ -191,6 +206,35 @@ impl Activations {
             })
         }
     }
+
+    /// Indicates that there is nothing to do for `timeout`, and that the scheduler
+    /// can allow the thread to sleep until then.
+    ///
+    /// The method does not *need* to park the thread, and indeed it may elect to
+    /// unpark earlier if there are deferred activations.
+    pub fn park_timeout(&self, timeout: Option<Duration>) {
+        let empty_for = self.empty_for();
+        let timeout = match (timeout, empty_for) {
+            (Some(x), Some(y)) => Some(std::cmp::min(x,y)),
+            (x, y) => x.or(y),
+        };
+
+        if let Some(timeout) = timeout {
+            std::thread::park_timeout(timeout);
+        }
+        else {
+            std::thread::park();
+        }
+    }
+}
+
+impl Scheduler for Activations {
+    fn activate(&mut self, path: &[usize]) {
+        self.activate(path);
+    }
+    fn extensions(&mut self, path: &[usize], dest: &mut BinaryHeap<Reverse<usize>>) {
+        self.for_extensions(path, |index| dest.push(Reverse(index)));
+    }
 }
 
 /// A thread-safe handle to an `Activations`.

timely/src/worker.rs

@@ -2,10 +2,11 @@
 
 use std::rc::Rc;
 use std::cell::{RefCell, RefMut};
+use std::cmp::Reverse;
 use std::any::Any;
 use std::str::FromStr;
 use std::time::{Instant, Duration};
-use std::collections::HashMap;
+use std::collections::{HashMap, BinaryHeap};
 use std::collections::hash_map::Entry;
 use std::sync::Arc;
 
@@ -230,7 +231,7 @@ pub struct Worker<A: Allocate> {
     logging: Option<Rc<RefCell<crate::logging_core::Registry>>>,
 
     activations: Rc<RefCell<Activations>>,
-    active_dataflows: Vec<usize>,
+    active_dataflows: BinaryHeap<Reverse<usize>>,
 
     // Temporary storage for channel identifiers during dataflow construction.
     // These are then associated with a dataflow once constructed.
@@ -346,7 +347,7 @@ impl<A: Allocate> Worker<A> {
     ///     worker.step_or_park(Some(Duration::from_secs(1)));
     /// });
     /// ```
-    pub fn step_or_park(&mut self, duration: Option<Duration>) -> bool {
+    pub fn step_or_park(&mut self, timeout: Option<Duration>) -> bool {
 
         {   // Process channel events. Activate responders.
             let mut allocator = self.allocator.borrow_mut();
@@ -370,43 +371,41 @@ impl<A: Allocate> Worker<A> {
             }
         }
 
-        // 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();
-        // Determine the minimum park duration, where `None` are an absence of a constraint.
-        let delay = match (duration, empty_for) {
-            (Some(x), Some(y)) => Some(std::cmp::min(x,y)),
-            (x, y) => x.or(y),
-        };
+        // Commence a new round of scheduling, starting with dataflows.
+        // We probe the scheduler for active prefixes, where an empty response
+        // indicates that the scheduler has no work for us at the moment.
+        {   // Scoped to let borrow of `self.active_dataflows` drop.
+            use crate::scheduling::activate::Scheduler;
+            let active_dataflows = &mut self.active_dataflows;
+            self.activations
+                .borrow_mut()
+                .extensions(&[], active_dataflows);
+        }
 
-        if delay != Some(Duration::new(0,0)) {
+        // If no dataflows are active, there is nothing to do. Consider parking.
+        if self.active_dataflows.is_empty() {
 
-            // Log parking and flush log.
-            if let Some(l) = self.logging().as_mut() {
-                l.log(crate::logging::ParkEvent::park(delay));
-                l.flush();
-            }
+            // If the timeout is zero, don't bother trying to park.
+            // More generally, we could put some threshold in here.
+            if timeout != Some(Duration::new(0, 0)) {
+                // Log parking and flush log.
+                if let Some(l) = self.logging().as_mut() {
+                    l.log(crate::logging::ParkEvent::park(timeout));
+                    l.flush();
+                }
 
-            self.allocator
-                .borrow()
-                .await_events(delay);
+                // We have just drained `allocator.events()` up above;
+                // otherwise we should first check it for emptiness.
+                self.activations.borrow().park_timeout(timeout);
 
-            // Log return from unpark.
-            self.logging().as_mut().map(|l| l.log(crate::logging::ParkEvent::unpark()));
+                // 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));
+        else {   // Schedule all active dataflows.
 
             let mut dataflows = self.dataflows.borrow_mut();
-            for index in active_dataflows.drain(..) {
+            for Reverse(index) in self.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.
@@ -745,7 +744,7 @@ impl<A: Allocate> Clone for Worker<A> {
             dataflow_counter: Rc::clone(&self.dataflow_counter),
             logging: self.logging.clone(),
             activations: Rc::clone(&self.activations),
-            active_dataflows: Vec::new(),
+            active_dataflows: Default::default(),
             temp_channel_ids: Rc::clone(&self.temp_channel_ids),
         }
     }