Make std::time::Instant optional

Author: frankmcsherry

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());

timely/src/scheduling/activate.rs

@@ -48,14 +48,14 @@ pub struct Activations {
     rx: Receiver<Vec<usize>>,
 
     // Delayed activations.
-    timer: Instant,
+    timer: Option<Instant>,
     queue: BinaryHeap<Reverse<(Duration, Vec<usize>)>>,
 }
 
 impl Activations {
 
     /// Creates a new activation tracker.
-    pub fn new(timer: Instant) -> Self {
+    pub fn new(timer: Option<Instant>) -> Self {
         let (tx, rx) = crossbeam_channel::unbounded();
         Self {
             clean: 0,
@@ -77,13 +77,18 @@ impl Activations {
 
     /// Schedules a future activation for the task addressed by `path`.
     pub fn activate_after(&mut self, path: &[usize], delay: Duration) {
-        // TODO: We could have a minimum delay and immediately schedule anything less than that delay.
-        if delay == Duration::new(0, 0) {
-            self.activate(path);
-        }
+        if let Some(timer) = self.timer {
+            // TODO: We could have a minimum delay and immediately schedule anything less than that delay.
+            if delay == Duration::new(0, 0) {
+                self.activate(path);
+            }
+            else {
+                let moment = timer.elapsed() + delay;
+                self.queue.push(Reverse((moment, path.to_vec())));
+            }
+        } 
         else {
-            let moment = self.timer.elapsed() + delay;
-            self.queue.push(Reverse((moment, path.to_vec())));
+            self.activate(path);
         }
     }
 
@@ -96,10 +101,12 @@ impl Activations {
         }
 
         // Drain timer-based activations.
-        let now = self.timer.elapsed();
-        while self.queue.peek().map(|Reverse((t,_))| t <= &now) == Some(true) {
-            let Reverse((_time, path)) = self.queue.pop().unwrap();
-            self.activate(&path[..]);
+        if let Some(timer) = self.timer {
+            let now = timer.elapsed();
+            while self.queue.peek().map(|Reverse((t,_))| t <= &now) == Some(true) {
+                let Reverse((_time, path)) = self.queue.pop().unwrap();
+                self.activate(&path[..]);
+            }
         }
 
         self.bounds.drain(.. self.clean);
@@ -171,12 +178,12 @@ impl Activations {
     /// indicates the amount of time before the thread should be unparked for the
     /// next scheduled activation.
     pub fn empty_for(&self) -> Option<Duration> {
-        if !self.bounds.is_empty() {
+        if !self.bounds.is_empty() || self.timer.is_none() {
             Some(Duration::new(0,0))
         }
         else {
             self.queue.peek().map(|Reverse((t,_a))| {
-                let elapsed = self.timer.elapsed();
+                let elapsed = self.timer.unwrap().elapsed();
                 if t < &elapsed { Duration::new(0,0) }
                 else { *t - elapsed }
             })

timely/src/worker.rs

@@ -198,23 +198,27 @@ pub trait AsWorker : Scheduler {
     /// The next worker-unique identifier to be allocated.
     fn peek_identifier(&self) -> usize;
     /// Provides access to named logging streams.
-    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>>>;
     /// Provides access to the timely logging stream.
-    fn logging(&self) -> Option<crate::logging::TimelyLogger> { self.log_register().get("timely") }
+    fn logging(&self) -> Option<crate::logging::TimelyLogger> { self.log_register().and_then(|l| l.get("timely")) }
 }
 
 /// A `Worker` is the entry point to a timely dataflow computation. It wraps a `Allocate`,
 /// and has a list of dataflows that it manages.
 pub struct Worker<A: Allocate> {
     config: Config,
-    timer: Instant,
+    /// 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<A>>,
     identifiers: Rc<RefCell<usize>>,
     // dataflows: Rc<RefCell<Vec<Wrapper>>>,
     dataflows: Rc<RefCell<HashMap<usize, Wrapper>>>,
     dataflow_counter: Rc<RefCell<usize>>,
-    logging: Rc<RefCell<crate::logging_core::Registry<crate::logging::WorkerIdentifier>>>,
+    logging: Option<Rc<RefCell<crate::logging_core::Registry<crate::logging::WorkerIdentifier>>>>,
 
     activations: Rc<RefCell<Activations>>,
     active_dataflows: Vec<usize>,
@@ -245,7 +249,7 @@ impl<A: Allocate> AsWorker for Worker<A> {
 
     fn new_identifier(&mut self) -> usize { self.new_identifier() }
     fn peek_identifier(&self) -> usize { self.peek_identifier() }
-    fn log_register(&self) -> RefMut<crate::logging_core::Registry<crate::logging::WorkerIdentifier>> {
+    fn log_register(&self) -> Option<RefMut<crate::logging_core::Registry<crate::logging::WorkerIdentifier>>> {
         self.log_register()
     }
 }
@@ -258,8 +262,7 @@ impl<A: Allocate> Scheduler for Worker<A> {
 
 impl<A: Allocate> Worker<A> {
     /// Allocates a new `Worker` bound to a channel allocator.
-    pub fn new(config: Config, c: A) -> Worker<A> {
-        let now = Instant::now();
+    pub fn new(config: Config, c: A, now: Option<std::time::Instant>) -> Worker<A> {
         let index = c.index();
         Worker {
             config,
@@ -269,7 +272,7 @@ impl<A: Allocate> Worker<A> {
             identifiers:  Default::default(),
             dataflows: Default::default(),
             dataflow_counter:  Default::default(),
-            logging: Rc::new(RefCell::new(crate::logging_core::Registry::new(now, index))),
+            logging: now.map(|now| Rc::new(RefCell::new(crate::logging_core::Registry::new(now, index)))),
             activations: Rc::new(RefCell::new(Activations::new(now))),
             active_dataflows: Default::default(),
             temp_channel_ids:  Default::default(),
@@ -405,7 +408,7 @@ impl<A: Allocate> Worker<A> {
         }
 
         // Clean up, indicate if dataflows remain.
-        self.logging.borrow_mut().flush();
+        self.logging.as_ref().map(|l| l.borrow_mut().flush());
         self.allocator.borrow_mut().release();
         !self.dataflows.borrow().is_empty()
     }
@@ -476,7 +479,7 @@ impl<A: Allocate> Worker<A> {
     ///
     ///     let index = worker.index();
     ///     let peers = worker.peers();
-    ///     let timer = worker.timer();
+    ///     let timer = worker.timer().unwrap();
     ///
     ///     println!("{:?}\tWorker {} of {}", timer.elapsed(), index, peers);
     ///
@@ -491,7 +494,7 @@ impl<A: Allocate> Worker<A> {
     ///
     ///     let index = worker.index();
     ///     let peers = worker.peers();
-    ///     let timer = worker.timer();
+    ///     let timer = worker.timer().unwrap();
     ///
     ///     println!("{:?}\tWorker {} of {}", timer.elapsed(), index, peers);
     ///
@@ -507,13 +510,13 @@ impl<A: Allocate> Worker<A> {
     ///
     ///     let index = worker.index();
     ///     let peers = worker.peers();
-    ///     let timer = worker.timer();
+    ///     let timer = worker.timer().unwrap();
     ///
     ///     println!("{:?}\tWorker {} of {}", timer.elapsed(), index, peers);
     ///
     /// });
     /// ```
-    pub fn timer(&self) -> Instant { self.timer }
+    pub fn timer(&self) -> Option<Instant> { self.timer }
 
     /// Allocate a new worker-unique identifier.
     ///
@@ -537,13 +540,14 @@ impl<A: Allocate> Worker<A> {
     /// timely::execute_from_args(::std::env::args(), |worker| {
     ///
     ///     worker.log_register()
+    ///           .unwrap()
     ///           .insert::<timely::logging::TimelyEvent,_>("timely", |time, data|
     ///               println!("{:?}\t{:?}", time, data)
     ///           );
     /// });
     /// ```
-    pub fn log_register(&self) -> ::std::cell::RefMut<crate::logging_core::Registry<crate::logging::WorkerIdentifier>> {
-        self.logging.borrow_mut()
+    pub fn log_register(&self) -> Option<::std::cell::RefMut<crate::logging_core::Registry<crate::logging::WorkerIdentifier>>> {
+        self.logging.as_ref().map(|l| l.borrow_mut())
     }
 
     /// Construct a new dataflow.
@@ -566,7 +570,7 @@ impl<A: Allocate> Worker<A> {
         T: Refines<()>,
         F: FnOnce(&mut Child<Self, T>)->R,
     {
-        let logging = self.logging.borrow_mut().get("timely");
+        let logging = self.logging.as_ref().map(|l| l.borrow_mut()).and_then(|l| l.get("timely"));
         self.dataflow_core("Dataflow", logging, Box::new(()), |_, child| func(child))
     }
 
@@ -590,7 +594,7 @@ impl<A: Allocate> Worker<A> {
         T: Refines<()>,
         F: FnOnce(&mut Child<Self, T>)->R,
     {
-        let logging = self.logging.borrow_mut().get("timely");
+        let logging = self.logging.as_ref().map(|l| l.borrow_mut()).and_then(|l| l.get("timely"));
         self.dataflow_core(name, logging, Box::new(()), |_, child| func(child))
     }
 
@@ -629,7 +633,7 @@ impl<A: Allocate> Worker<A> {
         let addr = vec![dataflow_index].into();
         let identifier = self.new_identifier();
 
-        let progress_logging = self.logging.borrow_mut().get("timely/progress");
+        let progress_logging = self.logging.as_ref().map(|l| l.borrow_mut()).and_then(|l| l.get("timely/progress"));
         let subscope = SubgraphBuilder::new_from(addr, logging.clone(), progress_logging.clone(), name);
         let subscope = RefCell::new(subscope);