Log action can distinguish data from flush

Previously, the log action would receive an empty container when we'd like
to report progress but no new data. We'd only send the empty container when
we didn't have any other data to send. This makes it hard for users to
distinguish between flushing and just data updates. It might be important
to distinguish the two because only on flush we might not be called again
for a while, but otherwise it's very likely that the logger might receive
more data, or sees a flush.

This change alters the signature of the action to accept a `&mut Option<C>`
(where `C` is `CB::Container`), and we pass `Some(container)` on data, and
`None` on flush.

Clients using the logging API need to change their implementation, as both
vectors and `Option` offer a `iter` function, but with different results.

Signed-off-by: Moritz Hoffmann <[email protected]>

Author: antiguru

logging/src/lib.rs

@@ -7,8 +7,9 @@ use std::time::{Instant, Duration};
 use std::fmt::{self, Debug};
 use std::marker::PhantomData;
 
-use timely_container::{ContainerBuilder, PushInto};
+use timely_container::{Container, ContainerBuilder, PushInto};
 
+/// A registry binding names to typed loggers.
 pub struct Registry {
     /// A map from names to typed loggers.
     map: HashMap<String, (Box<dyn Any>, Box<dyn Flush>)>,
@@ -28,7 +29,9 @@ impl Registry {
     /// seen (likely greater or equal to the timestamp of the last event). The end of a
     /// logging stream is indicated only by dropping the associated action, which can be
     /// accomplished with `remove` (or a call to insert, though this is not recommended).
-    pub fn insert<CB: ContainerBuilder, F: FnMut(&Duration, &mut CB::Container)+'static>(
+    ///
+    /// Passing a `&mut None` container to an action indicates a flush.
+    pub fn insert<CB: ContainerBuilder, F: FnMut(&Duration, &mut Option<CB::Container>)+'static>(
         &mut self,
         name: &str,
         action: F) -> Option<Box<dyn Any>>
@@ -84,7 +87,7 @@ impl Flush for Registry {
 
 /// A buffering logger.
 pub struct Logger<CB: ContainerBuilder> {
-    inner: Rc<RefCell<LoggerInner<CB, dyn FnMut(&Duration, &mut CB::Container)>>>,
+    inner: Rc<RefCell<LoggerInner<CB, dyn FnMut(&Duration, &mut Option<CB::Container>)>>>,
 }
 
 impl<CB: ContainerBuilder> Clone for Logger<CB> {
@@ -103,25 +106,25 @@ impl<CB: ContainerBuilder + Debug> Debug for Logger<CB> {
     }
 }
 
-struct LoggerInner<CB: ContainerBuilder, A: ?Sized + FnMut(&Duration, &mut CB::Container)> {
+struct LoggerInner<CB: ContainerBuilder, A: ?Sized + FnMut(&Duration, &mut Option<CB::Container>)> {
     /// common instant used for all loggers.
     time:   Instant,
     /// offset to allow re-calibration.
     offset: Duration,
     /// container builder to produce buffers of accumulated log events
     builder: CB,
     /// True if we logged an event since the last flush.
-    /// Used to avoid sending empty buffers on drop.
+    /// Used to avoid flushing buffers on drop.
     dirty: bool,
-    /// action to take on full log buffers.
+    /// action to take on full log buffers, or on flush.
     action: A,
 }
 
 impl<CB: ContainerBuilder> Logger<CB> {
     /// Allocates a new shareable logger bound to a write destination.
     pub fn new<F>(time: Instant, offset: Duration, action: F) -> Self
     where
-        F: FnMut(&Duration, &mut CB::Container)+'static
+        F: FnMut(&Duration, &mut Option<CB::Container>)+'static
     {
         let inner = LoggerInner {
             time,
@@ -234,7 +237,7 @@ impl<CB: ContainerBuilder, T> std::ops::Deref for TypedLogger<CB, T> {
     }
 }
 
-impl<CB: ContainerBuilder, A: ?Sized + FnMut(&Duration, &mut CB::Container)> LoggerInner<CB, A> {
+impl<CB: ContainerBuilder, A: ?Sized + FnMut(&Duration, &mut Option<CB::Container>)> LoggerInner<CB, A> {
     fn log_many<I>(&mut self, events: I)
         where I: IntoIterator, CB: PushInto<(Duration, I::Item)>,
     {
@@ -243,31 +246,37 @@ impl<CB: ContainerBuilder, A: ?Sized + FnMut(&Duration, &mut CB::Container)> Log
             self.dirty = true;
             self.builder.push_into((elapsed, event.into()));
             while let Some(container) = self.builder.extract() {
-                (self.action)(&elapsed, container);
+                let mut c = Some(std::mem::take(container));
+                (self.action)(&elapsed, &mut c);
+                if let Some(mut c) = c {
+                    c.clear();
+                    *container = c;
+                }
             }
         }
     }
 
     fn flush(&mut self) {
         let elapsed = self.time.elapsed() + self.offset;
 
-        let mut action_ran = false;
-        while let Some(buffer) = self.builder.finish() {
-            (self.action)(&elapsed, buffer);
-            action_ran = true;
+        while let Some(container) = self.builder.finish() {
+            let mut c = Some(std::mem::take(container));
+            (self.action)(&elapsed, &mut c);
+            if let Some(mut c) = c {
+                c.clear();
+                *container = c;
+            }
         }
 
-        if !action_ran {
-            // Send an empty container to indicate progress.
-            (self.action)(&elapsed, &mut CB::Container::default());
-        }
+        // Send no container to indicate flush.
+        (self.action)(&elapsed, &mut None);
 
         self.dirty = false;
     }
 }
 
 /// Flush on the *last* drop of a logger.
-impl<CB: ContainerBuilder, A: ?Sized + FnMut(&Duration, &mut CB::Container)> Drop for LoggerInner<CB, A> {
+impl<CB: ContainerBuilder, A: ?Sized + FnMut(&Duration, &mut Option<CB::Container>)> Drop for LoggerInner<CB, A> {
     fn drop(&mut self) {
         // Avoid sending out empty buffers just because of drops.
         if self.dirty {
@@ -276,7 +285,7 @@ impl<CB: ContainerBuilder, A: ?Sized + FnMut(&Duration, &mut CB::Container)> Dro
     }
 }
 
-impl<CB, A: ?Sized + FnMut(&Duration, &mut CB::Container)> Debug for LoggerInner<CB, A>
+impl<CB, A: ?Sized + FnMut(&Duration, &mut Option<CB::Container>)> Debug for LoggerInner<CB, A>
 where
     CB: ContainerBuilder + Debug,
 {

timely/examples/logging-send.rs

@@ -16,27 +16,37 @@ fn main() {
 
         // Register timely worker logging.
         worker.log_register().insert::<TimelyEventBuilder,_>("timely", |_time, data|
-            data.iter().for_each(|x| println!("LOG1: {:?}", x))
+            if let Some(data) = data {
+                data.iter().for_each(|x| println!("LOG1: {:?}", x))
+            }
+            else {
+                println!("LOG1: Flush");
+            }
         );
 
         // Register timely progress logging.
         // Less generally useful: intended for debugging advanced custom operators or timely
         // internals.
         worker.log_register().insert::<TimelyProgressEventBuilder,_>("timely/progress", |_time, data|
-            data.iter().for_each(|x| {
-                println!("PROGRESS: {:?}", x);
-                let (_, ev) = x;
-                print!("PROGRESS: TYPED MESSAGES: ");
-                for (n, p, t, d) in ev.messages.iter() {
-                    print!("{:?}, ", (n, p, t.as_any().downcast_ref::<usize>(), d));
-                }
-                println!();
-                print!("PROGRESS: TYPED INTERNAL: ");
-                for (n, p, t, d) in ev.internal.iter() {
-                    print!("{:?}, ", (n, p, t.as_any().downcast_ref::<usize>(), d));
-                }
-                println!();
-            })
+            if let Some(data) = data {
+                data.iter().for_each(|x| {
+                    println!("PROGRESS: {:?}", x);
+                    let (_, ev) = x;
+                    print!("PROGRESS: TYPED MESSAGES: ");
+                    for (n, p, t, d) in ev.messages.iter() {
+                        print!("{:?}, ", (n, p, t.as_any().downcast_ref::<usize>(), d));
+                    }
+                    println!();
+                    print!("PROGRESS: TYPED INTERNAL: ");
+                    for (n, p, t, d) in ev.internal.iter() {
+                        print!("{:?}, ", (n, p, t.as_any().downcast_ref::<usize>(), d));
+                    }
+                    println!();
+                })
+            }
+            else {
+                println!("PROGRESS: Flush");
+            }
         );
 
         // create a new input, exchange data, and inspect its output
@@ -49,7 +59,12 @@ fn main() {
 
         // Register timely worker logging.
         worker.log_register().insert::<TimelyEventBuilder,_>("timely", |_time, data|
-            data.iter().for_each(|x| println!("LOG2: {:?}", x))
+            if let Some(data) = data {
+                data.iter().for_each(|x| println!("LOG2: {:?}", x))
+            }
+            else {
+                println!("LOG2: Flush");
+            }
         );
 
         // create a new input, exchange data, and inspect its output
@@ -62,9 +77,14 @@ fn main() {
 
         // Register user-level logging.
         type MyBuilder = CapacityContainerBuilder<Vec<(Duration, ())>>;
-        worker.log_register().insert::<MyBuilder,_>("input", |_time, data|
-            for element in data.iter() {
-                println!("Round tick at: {:?}", element.0);
+        worker.log_register().insert::<MyBuilder,_>("input", |time, data|
+            if let Some(data) = data {
+                for element in data.iter() {
+                    println!("Round tick at: {:?}", element.0);
+                }
+            }
+            else {
+                println!("Round flush at: {time:?}");
             }
         );