Make std::time::Instant optional

communication/src/allocator/generic.rs

@@ -93,14 +93,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),
-        }
-    }
 }
 
 

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 crossbeam_channel::{Sender, Receiver};
 
@@ -178,10 +177,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

@@ -271,7 +271,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

@@ -240,14 +240,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/examples/logging-send.rs

@@ -15,14 +15,14 @@ fn main() {
         let mut probe = ProbeHandle::new();
 
         // Register timely worker logging.
-        worker.log_register().insert::<TimelyEvent,_>("timely", |_time, data|
+        worker.log_register().unwrap().insert::<TimelyEvent,_>("timely", |_time, data|
             data.iter().for_each(|x| println!("LOG1: {:?}", x))
         );
 
         // Register timely progress logging.
         // Less generally useful: intended for debugging advanced custom operators or timely
         // internals.
-        worker.log_register().insert::<TimelyProgressEvent,_>("timely/progress", |_time, data|
+        worker.log_register().unwrap().insert::<TimelyProgressEvent,_>("timely/progress", |_time, data|
             data.iter().for_each(|x| {
                 println!("PROGRESS: {:?}", x);
                 let (_, _, ev) = x;
@@ -48,7 +48,7 @@ fn main() {
         });
 
         // Register timely worker logging.
-        worker.log_register().insert::<TimelyEvent,_>("timely", |_time, data|
+        worker.log_register().unwrap().insert::<TimelyEvent,_>("timely", |_time, data|
             data.iter().for_each(|x| println!("LOG2: {:?}", x))
         );
 
@@ -61,13 +61,13 @@ fn main() {
         });
 
         // Register user-level logging.
-        worker.log_register().insert::<(),_>("input", |_time, data|
+        worker.log_register().unwrap().insert::<(),_>("input", |_time, data|
             for element in data.iter() {
                 println!("Round tick at: {:?}", element.0);
             }
         );
 
-        let input_logger = worker.log_register().get::<()>("input").expect("Input logger absent");
+        let input_logger = worker.log_register().unwrap().get::<()>("input").expect("Input logger absent");
 
         let timer = std::time::Instant::now();
 

timely/examples/threadless.rs

@@ -6,7 +6,7 @@ fn main() {
 
     // create a naked single-threaded worker.
     let allocator = timely::communication::allocator::Thread::default();
-    let mut worker = timely::worker::Worker::new(WorkerConfig::default(), allocator);
+    let mut worker = timely::worker::Worker::new(WorkerConfig::default(), allocator, None);
 
     // create input and probe handles.
     let mut input = InputHandle::new();

timely/src/dataflow/scopes/child.rs

@@ -70,7 +70,7 @@ where
     fn peek_identifier(&self) -> usize {
         self.parent.peek_identifier()
     }
-    fn log_register(&self) -> ::std::cell::RefMut<crate::logging_core::Registry<crate::logging::WorkerIdentifier>> {
+    fn log_register(&self) -> Option<::std::cell::RefMut<crate::logging_core::Registry<crate::logging::WorkerIdentifier>>> {
         self.parent.log_register()
     }
 }

timely/src/execute.rs

@@ -154,7 +154,7 @@ where
     F: FnOnce(&mut Worker<crate::communication::allocator::thread::Thread>)->T+Send+Sync+'static
 {
     let alloc = crate::communication::allocator::thread::Thread::default();
-    let mut worker = crate::worker::Worker::new(WorkerConfig::default(), alloc);
+    let mut worker = crate::worker::Worker::new(WorkerConfig::default(), alloc, Some(std::time::Instant::now()));
     let result = func(&mut worker);
     while worker.has_dataflows() {
         worker.step_or_park(None);
@@ -320,7 +320,7 @@ where
     T: Send+'static,
     F: Fn(&mut Worker<<A as AllocateBuilder>::Allocator>)->T+Send+Sync+'static {
     initialize_from(builders, others, move |allocator| {
-        let mut worker = Worker::new(worker_config.clone(), allocator);
+        let mut worker = Worker::new(worker_config.clone(), allocator, Some(std::time::Instant::now()));
         let result = func(&mut worker);
         while worker.has_dataflows() {
             worker.step_or_park(None);

timely/src/progress/subgraph.rs

@@ -179,8 +179,11 @@ where
         let path = self.path.clone();
         let reachability_logging =
         worker.log_register()
-            .get::<reachability::logging::TrackerEvent>("timely/reachability")
-            .map(|logger| reachability::logging::TrackerLogger::new(path, logger));
+            .as_ref()
+            .and_then(|l| 
+                l.get::<reachability::logging::TrackerEvent>("timely/reachability")
+                .map(|logger| reachability::logging::TrackerLogger::new(path, logger))
+            );
         let (tracker, scope_summary) = builder.build(reachability_logging);
 
         let progcaster = Progcaster::new(worker, self.path.clone(), self.logging.clone(), self.progress_logging.clone());
@@ -296,10 +299,11 @@ where
         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.