Instrumentation overhaul, fixes #91

Cargo.toml

@@ -17,12 +17,13 @@ license = "MIT"
 
 [features]
 default=[]
-logging=[]
 
 [dependencies]
 abomonation = { git = "https://github.com/frankmcsherry/abomonation" }
+abomonation_derive = { git = "https://github.com/mystor/abomonation_derive.git" }
 timely_sort="0.1.6"
-timely_communication = { git = "https://github.com/frankmcsherry/timely-dataflow/" }
+timely_communication = { version = "0.1.8", path = "./communication" }
+byteorder="0.4.2"
 time="0.1.34"
 
 [dev-dependencies]

communication/Cargo.toml

@@ -12,13 +12,15 @@ repository = "https://github.com/frankmcsherry/timely-dataflow.git"
 keywords = ["timely", "dataflow"]
 license = "MIT"
 
-[dependencies.abomonation]
-git="https://github.com/frankmcsherry/abomonation"
+[features]
+default=[] # ["logging"]
 
 [dependencies]
-#abomonation="0.4.4"
 byteorder="0.4.2"
 getopts="0.2.14"
+time="0.1.34"
+abomonation = { git = "https://github.com/frankmcsherry/abomonation" }
+abomonation_derive = { git = "https://github.com/mystor/abomonation_derive.git" }
 
 [profile.release]
 opt-level = 3

communication/src/allocator/binary.rs

@@ -1,4 +1,5 @@
 use std::sync::mpsc::{Sender, Receiver, channel};
+use std::sync::Arc;
 
 use {Allocate, Data, Push, Pull, Serialize};
 use allocator::Process;
@@ -13,7 +14,8 @@ pub struct Binary {
 
     // for loading up state in the networking threads.
     pub readers:    Vec<Sender<((usize, usize), Sender<Vec<u8>>)>>,
-    pub senders:    Vec<Sender<(MessageHeader, Vec<u8>)>>
+    pub senders:    Vec<Sender<(MessageHeader, Vec<u8>)>>,
+    pub log_sender: Arc<Fn(::logging::CommsSetup)->::logging::CommsLogger+Send+Sync>,
 }
 
 impl Binary {
@@ -24,12 +26,12 @@ impl Binary {
 impl Allocate for Binary {
     fn index(&self) -> usize { self.index }
     fn peers(&self) -> usize { self.peers }
-    fn allocate<T:Data>(&mut self) -> (Vec<Box<Push<T>>>, Box<Pull<T>>) {
+    fn allocate<T:Data>(&mut self) -> (Vec<Box<Push<T>>>, Box<Pull<T>>, Option<usize>) {
         let mut pushers: Vec<Box<Push<T>>> = Vec::new();
 
         // we'll need process-local channels as well (no self-loop binary connection in this design; perhaps should allow)
         let inner_peers = self.inner.peers();
-        let (inner_sends, inner_recv) = self.inner.allocate();
+        let (inner_sends, inner_recv, _) = self.inner.allocate();
 
         // prep a pushable for each endpoint, multiplied by inner_peers
         for index in 0..self.readers.len() {
@@ -44,8 +46,14 @@ impl Allocate for Binary {
                     source:     self.index,
                     target:     target_index,
                     length:     0,
+                    seqno:      0,
                 };
-                pushers.push(Box::new(Pusher::new(header, self.senders[index].clone())));
+                let logger = (self.log_sender)(::logging::CommsSetup {
+                    process: self.index,
+                    sender: true,
+                    remote: Some(target_index),
+                });
+                pushers.push(Box::new(Pusher::new(header, self.senders[index].clone(), logger)));
             }
         }
 
@@ -60,38 +68,54 @@ impl Allocate for Binary {
             reader.send(((self.index, self.allocated), send.clone())).unwrap();
         }
 
-        let pullable = Box::new(Puller::new(inner_recv, recv));
+        let logger = (self.log_sender)(::logging::CommsSetup {
+            process: self.index,
+            sender: false,
+            remote: None,
+        });
+        let pullable = Box::new(Puller::new(inner_recv, recv, logger));
 
         self.allocated += 1;
 
-        return (pushers, pullable);
+        return (pushers, pullable, Some(self.allocated - 1));
     }
 }
 
 struct Pusher<T> {
     header:     MessageHeader,
     sender:     Sender<(MessageHeader, Vec<u8>)>,   // targets for each remote destination
     phantom:    ::std::marker::PhantomData<T>,
+    log_sender: ::logging::CommsLogger,
 }
 
 impl<T> Pusher<T> {
-    pub fn new(header: MessageHeader, sender: Sender<(MessageHeader, Vec<u8>)>) -> Pusher<T> {
+    pub fn new(header: MessageHeader, sender: Sender<(MessageHeader, Vec<u8>)>, log_sender: ::logging::CommsLogger) -> Pusher<T> {
         Pusher {
             header:     header,
             sender:     sender,
             phantom:    ::std::marker::PhantomData,
+            log_sender: log_sender,
         }
     }
 }
 
 impl<T:Data> Push<T> for Pusher<T> {
     #[inline] fn push(&mut self, element: &mut Option<T>) {
         if let Some(ref mut element) = *element {
+            self.log_sender.when_enabled(|l| l.log(::logging::CommsEvent::Serialization(::logging::SerializationEvent {
+                    seq_no: Some(self.header.seqno),
+                    is_start: true,
+                })));
             let mut bytes = Vec::new();
             <T as Serialize>::into_bytes(element, &mut bytes);
             let mut header = self.header;
             header.length = bytes.len();
             self.sender.send((header, bytes)).ok();     // TODO : should be unwrap()?
+            self.log_sender.when_enabled(|l| l.log(::logging::CommsEvent::Serialization(::logging::SerializationEvent {
+                    seq_no: Some(self.header.seqno),
+                    is_start: true,
+                })));
+            self.header.seqno += 1;
         }
     }
 }
@@ -100,21 +124,34 @@ struct Puller<T> {
     inner: Box<Pull<T>>,            // inner pullable (e.g. intra-process typed queue)
     current: Option<T>,
     receiver: Receiver<Vec<u8>>,    // source of serialized buffers
+    log_sender: ::logging::CommsLogger,
 }
 impl<T:Data> Puller<T> {
-    fn new(inner: Box<Pull<T>>, receiver: Receiver<Vec<u8>>) -> Puller<T> {
-        Puller { inner: inner, receiver: receiver, current: None }
+    fn new(inner: Box<Pull<T>>, receiver: Receiver<Vec<u8>>, log_sender: ::logging::CommsLogger) -> Puller<T> {
+        Puller { inner: inner, receiver: receiver, current: None, log_sender: log_sender }
     }
 }
 
 impl<T:Data> Pull<T> for Puller<T> {
     #[inline]
     fn pull(&mut self) -> &mut Option<T> {
         let inner = self.inner.pull();
+        let log_sender = &self.log_sender;
         if inner.is_some() { inner }
         else {
             self.current = self.receiver.try_recv().ok().map(|mut bytes| {
-                <T as Serialize>::from_bytes(&mut bytes)
+                log_sender.when_enabled(|l| l.log(
+                    ::logging::CommsEvent::Serialization(::logging::SerializationEvent {
+                        seq_no: None,
+                        is_start: true,
+                    })));
+                let result = <T as Serialize>::from_bytes(&mut bytes);
+                log_sender.when_enabled(|l| l.log(
+                    ::logging::CommsEvent::Serialization(::logging::SerializationEvent {
+                        seq_no: None,
+                        is_start: false,
+                    })));
+                result
             });
             &mut self.current
         }

communication/src/allocator/generic.rs

@@ -32,7 +32,7 @@ impl Generic {
         }
     }
     /// Constructs several send endpoints and one receive endpoint.
-    pub fn allocate<T: Data>(&mut self) -> (Vec<Box<Push<T>>>, Box<Pull<T>>) {
+    pub fn allocate<T: Data>(&mut self) -> (Vec<Box<Push<T>>>, Box<Pull<T>>, Option<usize>) {
         match self {
             &mut Generic::Thread(ref mut t) => t.allocate(),
             &mut Generic::Process(ref mut p) => p.allocate(),
@@ -44,5 +44,7 @@ impl Generic {
 impl Allocate for Generic {
     fn index(&self) -> usize { self.index() }
     fn peers(&self) -> usize { self.peers() }
-    fn allocate<T: Data>(&mut self) -> (Vec<Box<Push<T>>>, Box<Pull<T>>) { self.allocate() }
+    fn allocate<T: Data>(&mut self) -> (Vec<Box<Push<T>>>, Box<Pull<T>>, Option<usize>) {
+        self.allocate()
+    }
 }

communication/src/allocator/mod.rs

@@ -21,5 +21,5 @@ pub trait Allocate {
     /// The number of workers.
     fn peers(&self) -> usize;
     /// Constructs several send endpoints and one receive endpoint.
-    fn allocate<T: Data>(&mut self) -> (Vec<Box<Push<T>>>, Box<Pull<T>>);
+    fn allocate<T: Data>(&mut self) -> (Vec<Box<Push<T>>>, Box<Pull<T>>, Option<usize>);
 }

communication/src/allocator/process.rs

@@ -31,7 +31,7 @@ impl Process {
 impl Allocate for Process {
     fn index(&self) -> usize { self.index }
     fn peers(&self) -> usize { self.peers }
-    fn allocate<T: Any+Send+'static>(&mut self) -> (Vec<Box<Push<T>>>, Box<Pull<T>>) {
+    fn allocate<T: Any+Send+'static>(&mut self) -> (Vec<Box<Push<T>>>, Box<Pull<T>>, Option<usize>) {
 
         // ensure exclusive access to shared list of channels
         let mut channels = self.channels.lock().ok().expect("mutex error?");
@@ -60,7 +60,7 @@ impl Allocate for Process {
                 self.allocated += 1;
                 let mut temp = Vec::new();
                 for s in send.into_iter() { temp.push(Box::new(s) as Box<Push<T>>); }
-                return (temp, Box::new(recv) as Box<Pull<T>>)
+                return (temp, Box::new(recv) as Box<Pull<T>>, None)
             }
             else {
                 panic!("channel already consumed");

communication/src/allocator/thread.rs

@@ -11,9 +11,9 @@ pub struct Thread;
 impl Allocate for Thread {
     fn index(&self) -> usize { 0 }
     fn peers(&self) -> usize { 1 }
-    fn allocate<T: 'static>(&mut self) -> (Vec<Box<Push<T>>>, Box<Pull<T>>) {
+    fn allocate<T: 'static>(&mut self) -> (Vec<Box<Push<T>>>, Box<Pull<T>>, Option<usize>) {
         let (pusher, puller) = Thread::new();
-        (vec![Box::new(pusher)], Box::new(puller))
+        (vec![Box::new(pusher)], Box::new(puller), None)
     }
 }
 

communication/src/initialize.rs

@@ -72,12 +72,12 @@ impl Configuration {
     }
 }
 
-fn create_allocators(config: Configuration) -> Result<Vec<Generic>,String> {
+fn create_allocators(config: Configuration, logger: Arc<Fn(::logging::CommsSetup)->::logging::CommsLogger+Send+Sync>) -> Result<Vec<Generic>,String> {
     match config {
         Configuration::Thread => Ok(vec![Generic::Thread(Thread)]),
         Configuration::Process(threads) => Ok(Process::new_vector(threads).into_iter().map(|x| Generic::Process(x)).collect()),
         Configuration::Cluster(threads, process, addresses, report) => {
-            if let Ok(stuff) = initialize_networking(addresses, process, threads, report) {
+            if let Ok(stuff) = initialize_networking(addresses, process, threads, report, logger) {
                 Ok(stuff.into_iter().map(|x| Generic::Binary(x)).collect())
             }
             else {
@@ -146,17 +146,19 @@ fn create_allocators(config: Configuration) -> Result<Vec<Generic>,String> {
 /// result: Ok(0)
 /// result: Ok(1)
 /// ```
-pub fn initialize<T:Send+'static, F: Fn(Generic)->T+Send+Sync+'static>(config: Configuration, func: F) -> Result<WorkerGuards<T>,String> {
+pub fn initialize<T:Send+'static, F: Fn(Generic)->T+Send+Sync+'static>(config: Configuration, func: F, log_sender: Arc<Fn(::logging::CommsSetup)->::logging::CommsLogger+Send+Sync>) -> Result<WorkerGuards<T>,String> {
 
-    let allocators = try!(create_allocators(config));
+    let allocators = try!(create_allocators(config, log_sender));
     let logic = Arc::new(func);
 
     let mut guards = Vec::new();
     for allocator in allocators.into_iter() {
         let clone = logic.clone();
         guards.push(try!(thread::Builder::new()
                             .name(format!("worker thread {}", allocator.index()))
-                            .spawn(move || (*clone)(allocator))
+                            .spawn(move || {
+                                (*clone)(allocator)
+                            })
                             .map_err(|e| format!("{:?}", e))));
     }
 

communication/src/lib.rs

@@ -69,11 +69,14 @@
 extern crate getopts;
 extern crate byteorder;
 extern crate abomonation;
+#[macro_use] extern crate abomonation_derive;
+extern crate time;
 
 pub mod allocator;
 mod networking;
 pub mod initialize;
 mod drain;
+pub mod logging;
 
 use std::any::Any;
 use abomonation::{Abomonation, encode, decode};