Remove the serialization machinery from progress logging, provide dynamic type information instead

Author: utaal

timely/examples/logging-send.rs

@@ -22,8 +22,23 @@ fn main() {
         );
 
         // 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|
-            data.iter().for_each(|x| println!("PROGRESS: {:?}", x))
+            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!();
+            })
         );
 
         // create a new input, exchange data, and inspect its output

timely/src/logging.rs

@@ -72,28 +72,32 @@ pub struct ChannelsEvent {
     pub target: (usize, usize),
 }
 
-#[cfg(not(feature = "bincode"))]
-/// Encapsulates (de)serialization, Debug, Display, for dynamically typed timestamps in logs
-pub trait ProgressEventTimestamp: std::fmt::Debug + std::fmt::Display + std::any::Any {
-    /// Encodes a typed reference into a binary buffer.
+/// Encapsulates Any and Debug for dynamically typed timestamps in logs
+pub trait ProgressEventTimestamp: std::fmt::Debug + std::any::Any {
+    /// Upcasts this `ProgressEventTimestamp` to `Any`.
     ///
-    /// # Safety
+    /// NOTE: This is required until https://github.com/rust-lang/rfcs/issues/2765 is fixed
+    fn as_any(&self) -> &dyn std::any::Any;
+
+    /// Returns the name of the concrete type of this object.
+    ///
+    /// # Note
     ///
-    /// This method is unsafe because it is unsafe to transmute typed allocations to binary.
-    /// Furthermore, Rust currently indicates that it is undefined behavior to observe padding
-    /// bytes, which will happen when we `memmcpy` structs which contain padding bytes.
-    unsafe fn encode(&self, write: &mut dyn std::io::Write) -> std::io::Result<()>;
+    /// This is intended for diagnostic use. The exact contents and format of the
+    /// string returned are not specified, other than being a best-effort
+    /// description of the type. For example, amongst the strings
+    /// that `type_name::<Option<String>>()` might return are `"Option<String>"` and
+    /// `"std::option::Option<std::string::String>"`.
+    fn type_name(&self) -> &'static str;
 }
+impl<T: crate::Data + std::fmt::Debug + std::any::Any> ProgressEventTimestamp for T {
+    fn as_any(&self) -> &dyn std::any::Any { self }
 
-#[cfg(not(feature = "bincode"))]
-impl<T: crate::communication::Data + std::fmt::Debug + std::fmt::Display + std::any::Any> ProgressEventTimestamp for T {
-    unsafe fn encode(&self, mut write: &mut dyn std::io::Write) -> std::io::Result<()> {
-        abomonation::encode(self, &mut write)
-    }
+    fn type_name(&self) -> &'static str { std::any::type_name::<T>() }
 }
 
 /// A vector of progress updates in logs
-pub trait ProgressEventTimestampVec: std::fmt::Debug + std::any::Any{
+pub trait ProgressEventTimestampVec: std::fmt::Debug + std::any::Any {
     /// Iterate over the contents of the vector
     fn iter<'a>(&'a self) -> Box<dyn Iterator<Item=(&'a usize, &'a usize, &'a dyn ProgressEventTimestamp, &'a i64)>+'a>;
 }

timely/src/progress/broadcast.rs

@@ -59,13 +59,6 @@ impl<T:Timestamp+Send> Progcaster<T> {
         changes.compact();
         if !changes.is_empty() {
 
-            // This logging is relatively more expensive than other logging, as we
-            // have formatting and string allocations on the main path. We do have
-            // local type information about the timestamp, and we could log *that*
-            // explicitly, but the consumer would have to know what to look for and
-            // interpret appropriately. That's a big ask, so let's start with this,
-            // and as folks need more performant logging think about allowing users
-            // to select the more efficient variant.
             self.progress_logging.as_ref().map(|l| {
 
                 // Pre-allocate enough space; we transfer ownership, so there is not
@@ -74,14 +67,13 @@ impl<T:Timestamp+Send> Progcaster<T> {
                 let mut messages = Box::new(Vec::with_capacity(changes.len()));
                 let mut internal = Box::new(Vec::with_capacity(changes.len()));
 
-                // TODO: Reconsider `String` type or perhaps re-use allocation.
                 for ((location, time), diff) in changes.iter() {
                     match location.port {
                         Port::Target(port) => {
-                            messages.push((location.node, port, format!("{:?}", time), *diff))
+                            messages.push((location.node, port, time.clone(), *diff))
                         },
                         Port::Source(port) => {
-                            internal.push((location.node, port, format!("{:?}", time), *diff))
+                            internal.push((location.node, port, time.clone(), *diff))
                         }
                     }
                 }
@@ -144,15 +136,14 @@ impl<T:Timestamp+Send> Progcaster<T> {
                 let mut messages = Vec::with_capacity(changes.len());
                 let mut internal = Vec::with_capacity(changes.len());
 
-                // TODO: Reconsider `String` type or perhaps re-use allocation.
                 for ((location, time), diff) in recv_changes.iter() {
 
                     match location.port {
                         Port::Target(port) => {
-                            messages.push((location.node, port, format!("{:?}", time), *diff))
+                            messages.push((location.node, port, time.clone(), *diff))
                         },
                         Port::Source(port) => {
-                            internal.push((location.node, port, format!("{:?}", time), *diff))
+                            internal.push((location.node, port, time.clone(), *diff))
                         }
                     }
                 }