Remove the identifier from logging

This changes the data received from logging from (Duration, Id, Data) to
(Duration, Data), saving the space for Id. In most cases, the Id can be
derived from the context where the log data is processed.

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

Author: antiguru

communication/src/allocator/zero_copy/initialize.rs

@@ -39,7 +39,7 @@ pub fn initialize_networking(
     my_index: usize,
     threads: usize,
     noisy: bool,
-    log_sender: Box<dyn Fn(CommunicationSetup)->Option<Logger<CommunicationEvent, CommunicationSetup>>+Send+Sync>)
+    log_sender: Box<dyn Fn(CommunicationSetup)->Option<Logger<CommunicationEvent>>+Send+Sync>)
 -> ::std::io::Result<(Vec<TcpBuilder<ProcessBuilder>>, CommsGuard)>
 {
     let sockets = create_sockets(addresses, my_index, noisy)?;
@@ -57,7 +57,7 @@ pub fn initialize_networking_from_sockets<S: Stream + 'static>(
     mut sockets: Vec<Option<S>>,
     my_index: usize,
     threads: usize,
-    log_sender: Box<dyn Fn(CommunicationSetup)->Option<Logger<CommunicationEvent, CommunicationSetup>>+Send+Sync>)
+    log_sender: Box<dyn Fn(CommunicationSetup)->Option<Logger<CommunicationEvent>>+Send+Sync>)
 -> ::std::io::Result<(Vec<TcpBuilder<ProcessBuilder>>, CommsGuard)>
 {
     // Sockets are expected to be blocking,

communication/src/allocator/zero_copy/tcp.rs

@@ -11,7 +11,7 @@ use super::stream::Stream;
 
 use logging_core::Logger;
 
-use crate::logging::{CommunicationEvent, CommunicationSetup, MessageEvent, StateEvent};
+use crate::logging::{CommunicationEvent, MessageEvent, StateEvent};
 
 fn tcp_panic(context: &'static str, cause: io::Error) -> ! {
     // NOTE: some downstream crates sniff out "timely communication error:" from
@@ -35,7 +35,7 @@ pub fn recv_loop<S>(
     worker_offset: usize,
     process: usize,
     remote: usize,
-    mut logger: Option<Logger<CommunicationEvent, CommunicationSetup>>)
+    mut logger: Option<Logger<CommunicationEvent>>)
 where
     S: Stream,
 {
@@ -134,7 +134,7 @@ pub fn send_loop<S: Stream>(
     sources: Vec<Sender<MergeQueue>>,
     process: usize,
     remote: usize,
-    mut logger: Option<Logger<CommunicationEvent, CommunicationSetup>>)
+    mut logger: Option<Logger<CommunicationEvent>>)
 {
 
     // Log the send thread's start.

communication/src/initialize.rs

@@ -38,7 +38,7 @@ pub enum Config {
         /// Verbosely report connection process
         report: bool,
         /// Closure to create a new logger for a communication thread
-        log_fn: Box<dyn Fn(CommunicationSetup) -> Option<Logger<CommunicationEvent, CommunicationSetup>> + Send + Sync>,
+        log_fn: Box<dyn Fn(CommunicationSetup) -> Option<Logger<CommunicationEvent>> + Send + Sync>,
     }
 }
 

logging/src/lib.rs

@@ -6,16 +6,14 @@ use std::collections::HashMap;
 use std::time::{Instant, Duration};
 use std::fmt::{self, Debug};
 
-pub struct Registry<Id> {
-    /// A worker-specific identifier.
-    id: Id,
+pub struct Registry {
     /// A map from names to typed loggers.
     map: HashMap<String, (Box<dyn Any>, Box<dyn Flush>)>,
     /// An instant common to all logging statements.
     time: Instant,
 }
 
-impl<Id: Clone+'static> Registry<Id> {
+impl Registry {
     /// Binds a log name to an action on log event batches.
     ///
     /// This method also returns any pre-installed action, rather than overwriting it
@@ -27,20 +25,20 @@ impl<Id: Clone+'static> Registry<Id> {
     /// 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<T: 'static, F: FnMut(&Duration, &mut Vec<(Duration, Id, T)>)+'static>(
+    pub fn insert<T: 'static, F: FnMut(&Duration, &mut Vec<(Duration, T)>)+'static>(
         &mut self,
         name: &str,
         action: F) -> Option<Box<dyn Any>>
     {
-        let logger = Logger::<T, Id>::new(self.time, Duration::default(), self.id.clone(), action);
+        let logger = Logger::<T>::new(self.time, Duration::default(), action);
         self.insert_logger(name, logger)
     }
 
     /// Binds a log name to a logger.
     pub fn insert_logger<T: 'static>(
         &mut self,
         name: &str,
-        logger: Logger<T, Id>) -> Option<Box<dyn Any>>
+        logger: Logger<T>) -> Option<Box<dyn Any>>
     {
         self.map.insert(name.to_owned(), (Box::new(logger.clone()), Box::new(logger))).map(|x| x.0)
     }
@@ -56,17 +54,16 @@ impl<Id: Clone+'static> Registry<Id> {
     }
 
     /// Retrieves a shared logger, if one has been inserted.
-    pub fn get<T: 'static>(&self, name: &str) -> Option<Logger<T, Id>> {
+    pub fn get<T: 'static>(&self, name: &str) -> Option<Logger<T>> {
         self.map
             .get(name)
-            .and_then(|entry| entry.0.downcast_ref::<Logger<T, Id>>())
+            .and_then(|entry| entry.0.downcast_ref::<Logger<T>>())
             .map(|x| (*x).clone())
     }
 
     /// Creates a new logger registry.
-    pub fn new(time: Instant, id: Id) -> Self {
+    pub fn new(time: Instant) -> Self {
         Registry {
-            id,
             time,
             map: HashMap::new(),
         }
@@ -78,7 +75,7 @@ impl<Id: Clone+'static> Registry<Id> {
     }
 }
 
-impl<Id> Flush for Registry<Id> {
+impl Flush for Registry {
     fn flush(&mut self) {
         for value in self.map.values_mut() {
             value.1.flush();
@@ -88,42 +85,40 @@ impl<Id> Flush for Registry<Id> {
 
 /// A buffering logger.
 #[derive(Debug)]
-pub struct Logger<T, E> {
-    inner: Rc<RefCell<LoggerInner<T, E, dyn FnMut(&Duration, &mut Vec<(Duration, E, T)>)>>>,
+pub struct Logger<T> {
+    inner: Rc<RefCell<LoggerInner<T, dyn FnMut(&Duration, &mut Vec<(Duration, T)>)>>>,
 }
 
-impl<T, E: Clone> Clone for Logger<T, E> {
+impl<T> Clone for Logger<T> {
     fn clone(&self) -> Self {
         Self {
             inner: self.inner.clone()
         }
     }
 }
 
-struct LoggerInner<T, E, A: ?Sized + FnMut(&Duration, &mut Vec<(Duration, E, T)>)> {
-    id:     E,
+struct LoggerInner<T, A: ?Sized + FnMut(&Duration, &mut Vec<(Duration, T)>)> {
     /// common instant used for all loggers.
     time:   Instant,
     /// offset to allow re-calibration.
     offset: Duration,
     /// shared buffer of accumulated log events
-    buffer: Vec<(Duration, E, T)>,
+    buffer: Vec<(Duration, T)>,
     /// action to take on full log buffers.
     action: A,
 }
 
-impl<T, E: Clone> Logger<T, E> {
+impl<T> Logger<T> {
     /// Allocates a new shareable logger bound to a write destination.
-    pub fn new<F>(time: Instant, offset: Duration, id: E, action: F) -> Self
+    pub fn new<F>(time: Instant, offset: Duration, action: F) -> Self
     where
-        F: FnMut(&Duration, &mut Vec<(Duration, E, T)>)+'static
+        F: FnMut(&Duration, &mut Vec<(Duration, T)>)+'static
     {
         let inner = LoggerInner {
-            id,
             time,
             offset,
             action,
-            buffer: Vec::with_capacity(LoggerInner::<T, E, F>::buffer_capacity()),
+            buffer: Vec::with_capacity(LoggerInner::<T, F>::buffer_capacity()),
         };
         let inner = Rc::new(RefCell::new(inner));
         Logger { inner }
@@ -167,7 +162,7 @@ impl<T, E: Clone> Logger<T, E> {
     }
 }
 
-impl<T, E: Clone, A: ?Sized + FnMut(&Duration, &mut Vec<(Duration, E, T)>)> LoggerInner<T, E, A> {
+impl<T, A: ?Sized + FnMut(&Duration, &mut Vec<(Duration, T)>)> LoggerInner<T, A> {
 
     /// The upper limit for buffers to allocate, size in bytes. [Self::buffer_capacity] converts
     /// this to size in elements.
@@ -177,7 +172,7 @@ impl<T, E: Clone, A: ?Sized + FnMut(&Duration, &mut Vec<(Duration, E, T)>)> Logg
     /// and 1, inclusively.
     // TODO: This fn is not const because it cannot depend on non-Sized generic parameters
     fn buffer_capacity() -> usize {
-        let size =  ::std::mem::size_of::<(Duration, E, T)>();
+        let size =  ::std::mem::size_of::<(Duration, T)>();
         if size == 0 {
             Self::BUFFER_SIZE_BYTES
         } else if size <= Self::BUFFER_SIZE_BYTES {
@@ -192,7 +187,7 @@ impl<T, E: Clone, A: ?Sized + FnMut(&Duration, &mut Vec<(Duration, E, T)>)> Logg
     {
         let elapsed = self.time.elapsed() + self.offset;
         for event in events {
-            self.buffer.push((elapsed, self.id.clone(), event.into()));
+            self.buffer.push((elapsed, event.into()));
             if self.buffer.len() == self.buffer.capacity() {
                 // Would call `self.flush()`, but for `RefCell` panic.
                 (self.action)(&elapsed, &mut self.buffer);
@@ -209,7 +204,7 @@ impl<T, E: Clone, A: ?Sized + FnMut(&Duration, &mut Vec<(Duration, E, T)>)> Logg
 }
 
 /// Flush on the *last* drop of a logger.
-impl<T, E, A: ?Sized + FnMut(&Duration, &mut Vec<(Duration, E, T)>)> Drop for LoggerInner<T, E, A> {
+impl<T, A: ?Sized + FnMut(&Duration, &mut Vec<(Duration, T)>)> Drop for LoggerInner<T, A> {
     fn drop(&mut self) {
         // Avoid sending out empty buffers just because of drops.
         if !self.buffer.is_empty() {
@@ -218,14 +213,12 @@ impl<T, E, A: ?Sized + FnMut(&Duration, &mut Vec<(Duration, E, T)>)> Drop for Lo
     }
 }
 
-impl<T, E, A: ?Sized + FnMut(&Duration, &mut Vec<(Duration, E, T)>)> Debug for LoggerInner<T, E, A>
+impl<T, A: ?Sized + FnMut(&Duration, &mut Vec<(Duration, T)>)> Debug for LoggerInner<T, A>
 where
-    E: Debug,
     T: Debug,
 {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("LoggerInner")
-            .field("id", &self.id)
             .field("time", &self.time)
             .field("offset", &self.offset)
             .field("action", &"FnMut")
@@ -240,13 +233,13 @@ trait Flush {
     fn flush(&mut self);
 }
 
-impl<T, E> Flush for Logger<T, E> {
+impl<T> Flush for Logger<T> {
     fn flush(&mut self) {
         self.inner.borrow_mut().flush()
     }
 }
 
-impl<T, E, A: ?Sized + FnMut(&Duration, &mut Vec<(Duration, E, T)>)> Flush for LoggerInner<T, E, A> {
+impl<T, A: ?Sized + FnMut(&Duration, &mut Vec<(Duration, T)>)> Flush for LoggerInner<T, A> {
     fn flush(&mut self) {
         let elapsed = self.time.elapsed() + self.offset;
         if !self.buffer.is_empty() {

timely/src/dataflow/scopes/child.rs

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

timely/src/execute.rs

@@ -245,7 +245,6 @@ where
                     result = Some(crate::logging_core::Logger::new(
                         ::std::time::Instant::now(),
                         ::std::time::Duration::default(),
-                        events_setup,
                         move |time, data| logger.publish_batch(time, data)
                     ));
                 }

timely/src/logging.rs

@@ -3,7 +3,7 @@
 /// Type alias for logging timely events.
 pub type WorkerIdentifier = usize;
 /// Logger type for worker-local logging.
-pub type Logger<Event> = crate::logging_core::Logger<Event, WorkerIdentifier>;
+pub type Logger<Event> = crate::logging_core::Logger<Event>;
 /// Logger for timely dataflow system events.
 pub type TimelyLogger = Logger<TimelyEvent>;
 /// Logger for timely dataflow progress events (the "timely/progress" log stream).
@@ -13,14 +13,14 @@ use std::time::Duration;
 use crate::dataflow::operators::capture::{Event, EventPusher};
 
 /// Logs events as a timely stream, with progress statements.
-pub struct BatchLogger<T, E, P> where P: EventPusher<Duration, (Duration, E, T)> {
+pub struct BatchLogger<T, P> where P: EventPusher<Duration, (Duration, T)> {
     // None when the logging stream is closed
     time: Duration,
     event_pusher: P,
-    _phantom: ::std::marker::PhantomData<(E, T)>,
+    _phantom: ::std::marker::PhantomData<T>,
 }
 
-impl<T, E, P> BatchLogger<T, E, P> where P: EventPusher<Duration, (Duration, E, T)> {
+impl<T, P> BatchLogger<T, P> where P: EventPusher<Duration, (Duration, T)> {
     /// Creates a new batch logger.
     pub fn new(event_pusher: P) -> Self {
         BatchLogger {
@@ -30,7 +30,7 @@ impl<T, E, P> BatchLogger<T, E, P> where P: EventPusher<Duration, (Duration, E,
         }
     }
     /// Publishes a batch of logged events and advances the capability.
-    pub fn publish_batch(&mut self, &time: &Duration, data: &mut Vec<(Duration, E, T)>) {
+    pub fn publish_batch(&mut self, &time: &Duration, data: &mut Vec<(Duration, T)>) {
         if !data.is_empty() {
             self.event_pusher.push(Event::Messages(self.time, data.drain(..).collect()));
         }
@@ -42,7 +42,7 @@ impl<T, E, P> BatchLogger<T, E, P> where P: EventPusher<Duration, (Duration, E,
         self.time = time;
     }
 }
-impl<T, E, P> Drop for BatchLogger<T, E, P> where P: EventPusher<Duration, (Duration, E, T)> {
+impl<T, P> Drop for BatchLogger<T, P> where P: EventPusher<Duration, (Duration, T)> {
     fn drop(&mut self) {
         self.event_pusher.push(Event::Progress(vec![(self.time, -1)]));
     }

timely/src/synchronization/sequence.rs

@@ -157,7 +157,7 @@ impl<T: ExchangeData> Sequencer<T> {
                         }
 
                         let mut activator_borrow = activator_source.borrow_mut();
-                        let mut activator = activator_borrow.as_mut().unwrap();
+                        let activator = activator_borrow.as_mut().unwrap();
 
                         if let Some(t) = activator.catchup_until {
                             if capability.time().less_than(&t) {
@@ -190,7 +190,7 @@ impl<T: ExchangeData> Sequencer<T> {
 
                     if let Some(last) = recvd.last() {
                         let mut activator_borrow = activator_sink.borrow_mut();
-                        let mut activator = activator_borrow.as_mut().unwrap();
+                        let activator = activator_borrow.as_mut().unwrap();
 
                         activator.catchup_until = Some((last.0).0);
                         activator.activate();

timely/src/worker.rs

@@ -201,7 +201,7 @@ pub trait AsWorker : Scheduler {
     /// Allocates a new worker-unique identifier.
     fn new_identifier(&mut 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) -> ::std::cell::RefMut<crate::logging_core::Registry>;
     /// Provides access to the timely logging stream.
     fn logging(&self) -> Option<crate::logging::TimelyLogger> { self.log_register().get("timely") }
 }
@@ -217,7 +217,7 @@ pub struct Worker<A: Allocate> {
     // 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: Rc<RefCell<crate::logging_core::Registry>>,
 
     activations: Rc<RefCell<Activations>>,
     active_dataflows: Vec<usize>,
@@ -247,7 +247,7 @@ impl<A: Allocate> AsWorker for Worker<A> {
     }
 
     fn new_identifier(&mut self) -> usize { self.new_identifier() }
-    fn log_register(&self) -> RefMut<crate::logging_core::Registry<crate::logging::WorkerIdentifier>> {
+    fn log_register(&self) -> RefMut<crate::logging_core::Registry> {
         self.log_register()
     }
 }
@@ -262,7 +262,6 @@ 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();
-        let index = c.index();
         Worker {
             config,
             timer: now,
@@ -271,7 +270,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: Rc::new(RefCell::new(crate::logging_core::Registry::new(now))),
             activations: Rc::new(RefCell::new(Activations::new(now))),
             active_dataflows: Default::default(),
             temp_channel_ids:  Default::default(),
@@ -538,7 +537,7 @@ impl<A: Allocate> Worker<A> {
     ///           );
     /// });
     /// ```
-    pub fn log_register(&self) -> ::std::cell::RefMut<crate::logging_core::Registry<crate::logging::WorkerIdentifier>> {
+    pub fn log_register(&self) -> ::std::cell::RefMut<crate::logging_core::Registry> {
         self.logging.borrow_mut()
     }