Remove worker identifier from logging (#533)

* 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]>

* Log a communication setup message

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

* rebase

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

---------

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: Arc<dyn Fn(CommunicationSetup)->Option<Logger<CommunicationEvent, CommunicationSetup>>+Send+Sync>,
+    log_sender: Arc<dyn Fn(CommunicationSetup)->Option<Logger<CommunicationEvent>>+Send+Sync>,
 )
 -> ::std::io::Result<(Vec<TcpBuilder<ProcessBuilder>>, CommsGuard)>
 {
@@ -58,7 +58,7 @@ pub fn initialize_networking_from_sockets<S: Stream + 'static>(
     mut sockets: Vec<Option<S>>,
     my_index: usize,
     threads: usize,
-    log_sender: Arc<dyn Fn(CommunicationSetup)->Option<Logger<CommunicationEvent, CommunicationSetup>>+Send+Sync>,
+    log_sender: Arc<dyn Fn(CommunicationSetup)->Option<Logger<CommunicationEvent>>+Send+Sync>,
 )
 -> ::std::io::Result<(Vec<TcpBuilder<ProcessBuilder>>, CommsGuard)>
 {

communication/src/allocator/zero_copy/tcp.rs

@@ -11,7 +11,7 @@ use super::stream::Stream;
 
 use timely_logging::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

@@ -39,7 +39,7 @@ pub enum Config {
         /// Verbosely report connection process
         report: bool,
         /// Closure to create a new logger for a communication thread
-        log_fn: Arc<dyn Fn(CommunicationSetup) -> Option<Logger<CommunicationEvent, CommunicationSetup>> + Send + Sync>,
+        log_fn: Arc<dyn Fn(CommunicationSetup) -> Option<Logger<CommunicationEvent>> + Send + Sync>,
     }
 }
 
@@ -183,62 +183,62 @@ impl Config {
 /// # Examples
 /// ```
 /// use timely_communication::{Allocate, Bytesable};
-/// 
+///
 /// /// A wrapper that indicates the serialization/deserialization strategy.
 /// pub struct Message {
 ///     /// Text contents.
 ///     pub payload: String,
 /// }
-/// 
+///
 /// impl Bytesable for Message {
 ///     fn from_bytes(bytes: timely_bytes::arc::Bytes) -> Self {
 ///         Message { payload: std::str::from_utf8(&bytes[..]).unwrap().to_string() }
 ///     }
-/// 
+///
 ///     fn length_in_bytes(&self) -> usize {
 ///         self.payload.len()
 ///     }
-/// 
+///
 ///     fn into_bytes<W: ::std::io::Write>(&self, writer: &mut W) {
 ///         writer.write_all(self.payload.as_bytes()).unwrap();
 ///     }
 /// }
-/// 
+///
 /// fn main() {
-/// 
+///
 ///     // extract the configuration from user-supplied arguments, initialize the computation.
 ///     let config = timely_communication::Config::from_args(std::env::args()).unwrap();
 ///     let guards = timely_communication::initialize(config, |mut allocator| {
-/// 
+///
 ///         println!("worker {} of {} started", allocator.index(), allocator.peers());
-/// 
+///
 ///         // allocates a pair of senders list and one receiver.
 ///         let (mut senders, mut receiver) = allocator.allocate(0);
-/// 
+///
 ///         // send typed data along each channel
 ///         for i in 0 .. allocator.peers() {
 ///             senders[i].send(Message { payload: format!("hello, {}", i)});
 ///             senders[i].done();
 ///         }
-/// 
+///
 ///         // no support for termination notification,
 ///         // we have to count down ourselves.
 ///         let mut received = 0;
 ///         while received < allocator.peers() {
-/// 
+///
 ///             allocator.receive();
-/// 
+///
 ///             if let Some(message) = receiver.recv() {
 ///                 println!("worker {}: received: <{}>", allocator.index(), message.payload);
 ///                 received += 1;
 ///             }
-/// 
+///
 ///             allocator.release();
 ///         }
-/// 
+///
 ///         allocator.index()
 ///     });
-/// 
+///
 ///     // computation runs until guards are joined or dropped.
 ///     if let Ok(guards) = guards {
 ///         for guard in guards.join() {
@@ -279,62 +279,62 @@ pub fn initialize<T:Send+'static, F: Fn(Generic)->T+Send+Sync+'static>(
 /// # Examples
 /// ```
 /// use timely_communication::{Allocate, Bytesable};
-/// 
+///
 /// /// A wrapper that indicates `bincode` as the serialization/deserialization strategy.
 /// pub struct Message {
 ///     /// Text contents.
 ///     pub payload: String,
 /// }
-/// 
+///
 /// impl Bytesable for Message {
 ///     fn from_bytes(bytes: timely_bytes::arc::Bytes) -> Self {
 ///         Message { payload: std::str::from_utf8(&bytes[..]).unwrap().to_string() }
 ///     }
-/// 
+///
 ///     fn length_in_bytes(&self) -> usize {
 ///         self.payload.len()
 ///     }
-/// 
+///
 ///     fn into_bytes<W: ::std::io::Write>(&self, writer: &mut W) {
 ///         writer.write_all(self.payload.as_bytes()).unwrap();
 ///     }
 /// }
-/// 
+///
 /// fn main() {
-/// 
+///
 ///     // extract the configuration from user-supplied arguments, initialize the computation.
 ///     let config = timely_communication::Config::from_args(std::env::args()).unwrap();
 ///     let guards = timely_communication::initialize(config, |mut allocator| {
-/// 
+///
 ///         println!("worker {} of {} started", allocator.index(), allocator.peers());
-/// 
+///
 ///         // allocates a pair of senders list and one receiver.
 ///         let (mut senders, mut receiver) = allocator.allocate(0);
-/// 
+///
 ///         // send typed data along each channel
 ///         for i in 0 .. allocator.peers() {
 ///             senders[i].send(Message { payload: format!("hello, {}", i)});
 ///             senders[i].done();
 ///         }
-/// 
+///
 ///         // no support for termination notification,
 ///         // we have to count down ourselves.
 ///         let mut received = 0;
 ///         while received < allocator.peers() {
-/// 
+///
 ///             allocator.receive();
-/// 
+///
 ///             if let Some(message) = receiver.recv() {
 ///                 println!("worker {}: received: <{}>", allocator.index(), message.payload);
 ///                 received += 1;
 ///             }
-/// 
+///
 ///             allocator.release();
 ///         }
-/// 
+///
 ///         allocator.index()
 ///     });
-/// 
+///
 ///     // computation runs until guards are joined or dropped.
 ///     if let Ok(guards) = guards {
 ///         for guard in guards.join() {

communication/src/logging.rs

@@ -21,6 +21,8 @@ pub enum CommunicationEvent {
     Message(MessageEvent),
     /// A state transition.
     State(StateEvent),
+    /// Setup event
+    Setup(CommunicationSetup)
 }
 
 /// An observed message.