Merge pull request #188 from frankmcsherry/logging_revamp

Logging revamp

Author: frankmcsherry

CHANGELOG.md

@@ -0,0 +1,21 @@
+# Changelog
+All notable changes to this project will be documented in this file.
+
+## Unreleased
+- Many logging events have been rationalized. Operators and Channels should all have a worker-unique identifier that can be used to connect their metadata with events involving them. Previously this was a bit of a shambles.
+
+## 0.7.0
+
+### Added
+- You can now construct your own vector of allocator builders and supply them directly to `timely::execute::execute_from`. Previously one was restricted to whatever a `Configuration` could provide for you. This should allow more pleasant construction of custom allocators, or custom construction of existing allocators.
+- Each timely worker now has a log registry, `worker.log_registry()`, from which you can register and acquire typed loggers for named log streams. This supports user-level logging, as well as user-configurable timely logging. Timely logging is under the name `"timely"`.
+
+### Changed
+- The `Root` type has been renamed `Worker` and is found in the `::worker` module. The methods of the `ScopeParent` trait are now in the `::worker::AsWorker` trait.
+- The communication `Allocate` trait's main method `allocate` now takes a worker-unique identifier to use for the channel. The allocator may or may not use the information (most often for logging), but they are allowed to be incorrect if one allocates two channels with the same identifier.
+- A `CapabilityRef<T>` now supports `retain_for(usize)` which indicates a specific output port the capability should be retain for use with. The `retain()` method still exists for now and is equivalent to `retain(0)`. This change also comes with the *inability* to use an arbitrary `Capability<T>` with any output; using a capability bound to the wrong output will result in a run-time error.
+- The `unary` and `binary` operators now provide `data` as a `RefOrMut`, which does not implement `DerefMut`. More information on how to port methods can be found [here](https://github.com/frankmcsherry/timely-dataflow/pull/135#issuecomment-418355284).
+
+
+### Removed
+- The deprecated `Unary` and `Binary` operator extension traits have been removed in favor of the `Operator` trait that supports both of them, as well as their `_notify` variants.

Cargo.toml

@@ -18,9 +18,9 @@ license = "MIT"
 [dependencies]
 abomonation = { git = "https://github.com/frankmcsherry/abomonation" }
 abomonation_derive = "0.3"
-bytes = { path = "./bytes" }
+timely_bytes = { path = "./bytes" }
+timely_logging = { path = "./logging" }
 timely_communication = { path = "./communication"}
-time="0.1.34"
 
 [dev-dependencies]
 timely_sort="0.1.6"

bytes/Cargo.toml

@@ -1,5 +1,5 @@
 [package]
-name = "bytes"
+name = "timely_bytes"
 version = "0.1.0"
 authors = ["Frank McSherry <fmcsherry@me.com>"]
 

communication/Cargo.toml

@@ -17,12 +17,11 @@ default=["arg_parse"]
 arg_parse=["getopts"]
 
 [dependencies]
-#byteorder="1"
 getopts={version="0.2.14", optional=true}
-time="0.1.34"
 abomonation = { git = "https://github.com/frankmcsherry/abomonation" }
 abomonation_derive = "0.3"
-bytes = { path = "../bytes" }
+timely_bytes = { path = "../bytes" }
+timely_logging = { path = "../logging" }
 
 [profile.release]
 opt-level = 3

communication/examples/hello.rs

@@ -1,19 +1,18 @@
 extern crate timely_communication;
 
 use std::ops::Deref;
-use timely_communication::Message;
+use timely_communication::{Message, Allocate};
 
 fn main() {
 
     // extract the configuration from user-supplied arguments, initialize the computation.
     let config = timely_communication::Configuration::from_args(std::env::args()).unwrap();
-    let logger = ::std::sync::Arc::new(|_| timely_communication::logging::BufferingLogger::new_inactive());
-    let guards = timely_communication::initialize(config, logger, |mut allocator| {
+    let guards = timely_communication::initialize(config, |mut allocator| {
 
         println!("worker {} of {} started", allocator.index(), allocator.peers());
 
         // allocates pair of senders list and one receiver.
-        let (mut senders, mut receiver, _) = allocator.allocate();
+        let (mut senders, mut receiver) = allocator.allocate(0);
 
         // send typed data along each channel
         for i in 0 .. allocator.peers() {

communication/src/allocator/binary.rs

@@ -10,7 +10,6 @@ pub struct Binary {
     pub inner:      Process,    // inner Process (use for process-local channels)
     pub index:      usize,      // index of this worker
     pub peers:      usize,      // number of peer workers
-    pub allocated:  usize,      // indicates how many channels have been allocated (locally).
 
     // for loading up state in the networking threads.
     pub readers:    Vec<Sender<((usize, usize), Sender<Vec<u8>>)>>,
@@ -26,12 +25,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<Message<T>>>>, Box<Pull<Message<T>>>, Option<usize>) {
+    fn allocate<T: Data>(&mut self, identifier: usize) -> (Vec<Box<Push<Message<T>>>>, Box<Pull<Message<T>>>) {
         let mut pushers: Vec<Box<Push<Message<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(identifier);
 
         // prep a pushable for each endpoint, multiplied by inner_peers
         for index in 0..self.readers.len() {
@@ -42,7 +41,7 @@ impl Allocate for Binary {
                 if index >= self.index / inner_peers { target_index += inner_peers; }
 
                 let header = MessageHeader {
-                    channel:    self.allocated,
+                    channel:    identifier,
                     source:     self.index,
                     target:     target_index,
                     length:     0,
@@ -75,9 +74,7 @@ impl Allocate for Binary {
         });
         let pullable = Box::new(Puller::<T>::new(inner_recv, recv, logger)) as Box<Pull<Message<T>>>;
 
-        self.allocated += 1;
-
-        (pushers, pullable, Some(self.allocated - 1))
+        (pushers, pullable)
     }
 }
 

communication/src/allocator/generic.rs

@@ -42,12 +42,12 @@ impl Generic {
         }
     }
     /// Constructs several send endpoints and one receive endpoint.
-    pub fn allocate<T: Data>(&mut self) -> (Vec<Box<Push<Message<T>>>>, Box<Pull<Message<T>>>, Option<usize>) {
+    fn allocate<T: Data>(&mut self, identifier: usize) -> (Vec<Box<Push<Message<T>>>>, Box<Pull<Message<T>>>) {
         match self {
-            &mut Generic::Thread(ref mut t) => t.allocate(),
-            &mut Generic::Process(ref mut p) => p.allocate(),
-            &mut Generic::ProcessBinary(ref mut pb) => pb.allocate(),
-            &mut Generic::ZeroCopy(ref mut z) => z.allocate(),
+            &mut Generic::Thread(ref mut t) => t.allocate(identifier),
+            &mut Generic::Process(ref mut p) => p.allocate(identifier),
+            &mut Generic::ProcessBinary(ref mut pb) => pb.allocate(identifier),
+            &mut Generic::ZeroCopy(ref mut z) => z.allocate(identifier),
         }
     }
     /// Perform work before scheduling operators.
@@ -73,8 +73,8 @@ 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<Message<T>>>>, Box<Pull<Message<T>>>, Option<usize>) {
-        self.allocate()
+    fn allocate<T: Data>(&mut self, identifier: usize) -> (Vec<Box<Push<Message<T>>>>, Box<Pull<Message<T>>>) {
+        self.allocate(identifier)
     }
 
     fn pre_work(&mut self) { self.pre_work(); }

communication/src/allocator/mod.rs

@@ -36,7 +36,7 @@ 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<Message<T>>>>, Box<Pull<Message<T>>>, Option<usize>);
+    fn allocate<T: Data>(&mut self, identifier: usize) -> (Vec<Box<Push<Message<T>>>>, Box<Pull<Message<T>>>);
     /// Work performed before scheduling dataflows.
     fn pre_work(&mut self) { }
     /// Work performed after scheduling dataflows.

communication/src/allocator/process.rs

@@ -3,30 +3,30 @@
 use std::sync::{Arc, Mutex};
 use std::any::Any;
 use std::sync::mpsc::{Sender, Receiver, channel};
+use std::collections::HashMap;
 
 use allocator::{Allocate, AllocateBuilder, Message, Thread};
 use {Push, Pull};
 
 /// An allocater for inter-thread, intra-process communication
 pub struct Process {
-    inner:      Thread,                         // inner Thread
-    index:      usize,                          // number out of peers
-    peers:      usize,                          // number of peer allocators (for typed channel allocation).
-    allocated:  usize,                          // indicates how many have been allocated (locally).
-    channels:   Arc<Mutex<Vec<Box<Any+Send>>>>, // Box<Any+Send> -> Box<Vec<Option<(Vec<Sender<T>>, Receiver<T>)>>>
+    inner: Thread,
+    index: usize,
+    peers: usize,
+    // below: `Box<Any+Send>` is a `Box<Vec<Option<(Vec<Sender<T>>, Receiver<T>)>>>`
+    channels: Arc<Mutex<HashMap<usize, Box<Any+Send>>>>,
 }
 
 impl Process {
     /// Access the wrapped inner allocator.
     pub fn inner<'a>(&'a mut self) -> &'a mut Thread { &mut self.inner }
     /// Allocate a list of connected intra-process allocators.
     pub fn new_vector(count: usize) -> Vec<Process> {
-        let channels = Arc::new(Mutex::new(Vec::new()));
+        let channels = Arc::new(Mutex::new(HashMap::new()));
         (0 .. count).map(|index| Process {
             inner:      Thread,
             index:      index,
             peers:      count,
-            allocated:  0,
             channels:   channels.clone(),
         }).collect()
     }
@@ -35,43 +35,52 @@ 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<Message<T>>>>, Box<Pull<Message<T>>>, Option<usize>) {
+    fn allocate<T: Any+Send+'static>(&mut self, identifier: usize) -> (Vec<Box<Push<Message<T>>>>, Box<Pull<Message<T>>>) {
 
         // ensure exclusive access to shared list of channels
         let mut channels = self.channels.lock().ok().expect("mutex error?");
 
-        // we may need to alloc a new channel ...
-        if self.allocated == channels.len() {
-            let mut pushers = Vec::new();
-            let mut pullers = Vec::new();
-            for _ in 0..self.peers {
-                let (s, r): (Sender<Message<T>>, Receiver<Message<T>>) = channel();
-                pushers.push(Pusher { target: s });
-                pullers.push(Puller { source: r, current: None });
-            }
-
-            let mut to_box = Vec::new();
-            for recv in pullers.into_iter() {
-                to_box.push(Some((pushers.clone(), recv)));
-            }
-
-            channels.push(Box::new(to_box));
-        }
+        let (send, recv, empty) = {
+
+            // we may need to alloc a new channel ...
+            let mut entry = channels.entry(identifier).or_insert_with(|| {
+
+                let mut pushers = Vec::new();
+                let mut pullers = Vec::new();
+                for _ in 0..self.peers {
+                    let (s, r): (Sender<Message<T>>, Receiver<Message<T>>) = channel();
+                    pushers.push(Pusher { target: s });
+                    pullers.push(Puller { source: r, current: None });
+                }
+
+                let mut to_box = Vec::new();
+                for recv in pullers.into_iter() {
+                    to_box.push(Some((pushers.clone(), recv)));
+                }
+
+                Box::new(to_box)
+            });
+
+            let vector =
+            entry
+                .downcast_mut::<(Vec<Option<(Vec<Pusher<Message<T>>>, Puller<Message<T>>)>>)>()
+                .expect("failed to correctly cast channel");
+
+            let (send, recv) =
+            vector[self.index]
+                .take()
+                .expect("channel already consumed");
+
+            let empty = vector.iter().all(|x| x.is_none());
 
-        let vector =
-        channels[self.allocated]
-            .downcast_mut::<(Vec<Option<(Vec<Pusher<Message<T>>>, Puller<Message<T>>)>>)>()
-            .expect("failed to correctly cast channel");
+            (send, recv, empty)
+        };
 
-        let (send, recv) =
-        vector[self.index]
-            .take()
-            .expect("channel already consumed");
+        if empty { channels.remove(&identifier); }
 
-        self.allocated += 1;
         let mut temp = Vec::new();
         for s in send.into_iter() { temp.push(Box::new(s) as Box<Push<Message<T>>>); }
-        (temp, Box::new(recv) as Box<Pull<super::Message<T>>>, None)
+        (temp, Box::new(recv) as Box<Pull<super::Message<T>>>)
     }
 }
 

communication/src/allocator/thread.rs

@@ -13,9 +13,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<Message<T>>>>, Box<Pull<Message<T>>>, Option<usize>) {
+    fn allocate<T: 'static>(&mut self, _identifier: usize) -> (Vec<Box<Push<Message<T>>>>, Box<Pull<Message<T>>>) {
         let (pusher, puller) = Thread::new();
-        (vec![Box::new(pusher)], Box::new(puller), None)
+        (vec![Box::new(pusher)], Box::new(puller))
     }
 }
 

communication/src/allocator/zero_copy/allocator.rs

@@ -1,7 +1,7 @@
 //! Zero-copy allocator based on TCP.
 use std::rc::Rc;
 use std::cell::RefCell;
-use std::collections::VecDeque;
+use std::collections::{VecDeque, HashMap};
 // use std::sync::mpsc::{channel, Sender, Receiver};
 
 use bytes::arc::Bytes;
@@ -87,12 +87,12 @@ impl<A: Allocate> TcpBuilder<A> {
             inner: self.inner,
             index: self.index,
             peers: self.peers,
-            allocated: 0,
+            // allocated: 0,
             _signal: self.signal,
             staged: Vec::new(),
             sends,
             recvs: self.recvs,
-            to_local: Vec::new(),
+            to_local: HashMap::new(),
         }
     }
 }
@@ -104,32 +104,29 @@ pub struct TcpAllocator<A: Allocate> {
 
     index:      usize,                              // number out of peers
     peers:      usize,                              // number of peer allocators (for typed channel allocation).
-    allocated:  usize,                              // indicates how many channels have been allocated (locally).
+    // allocated:  usize,                              // indicates how many channels have been allocated (locally).
 
     _signal:     Signal,
 
     staged:     Vec<Bytes>,
 
     // sending, receiving, and responding to binary buffers.
-    sends:      Vec<Rc<RefCell<SendEndpoint<MergeQueue>>>>,         // sends[x] -> goes to process x.
-    recvs:      Vec<MergeQueue>,                    // recvs[x] <- from process x?.
-    to_local:   Vec<Rc<RefCell<VecDeque<Bytes>>>>,  // to worker-local typed pullers.
+    sends:      Vec<Rc<RefCell<SendEndpoint<MergeQueue>>>>,     // sends[x] -> goes to process x.
+    recvs:      Vec<MergeQueue>,                                // recvs[x] <- from process x?.
+    to_local:   HashMap<usize, Rc<RefCell<VecDeque<Bytes>>>>,   // to worker-local typed pullers.
 }
 
 impl<A: Allocate> Allocate for TcpAllocator<A> {
     fn index(&self) -> usize { self.index }
     fn peers(&self) -> usize { self.peers }
-    fn allocate<T: Data>(&mut self) -> (Vec<Box<Push<Message<T>>>>, Box<Pull<Message<T>>>, Option<usize>) {
-
-        let channel_id = self.allocated;
-        self.allocated += 1;
+    fn allocate<T: Data>(&mut self, identifier: usize) -> (Vec<Box<Push<Message<T>>>>, Box<Pull<Message<T>>>) {
 
         // Result list of boxed pushers.
         let mut pushes = Vec::<Box<Push<Message<T>>>>::new();
 
         // Inner exchange allocations.
         let inner_peers = self.inner.peers();
-        let (mut inner_sends, inner_recv, _) = self.inner.allocate();
+        let (mut inner_sends, inner_recv) = self.inner.allocate(identifier);
 
         for target_index in 0 .. self.peers() {
 
@@ -142,7 +139,7 @@ impl<A: Allocate> Allocate for TcpAllocator<A> {
             else {
                 // message header template.
                 let header = MessageHeader {
-                    channel:    channel_id,
+                    channel:    identifier,
                     source:     self.index,
                     target:     target_index,
                     length:     0,
@@ -155,13 +152,11 @@ impl<A: Allocate> Allocate for TcpAllocator<A> {
             }
         }
 
-        while self.to_local.len() <= channel_id {
-            self.to_local.push(Rc::new(RefCell::new(VecDeque::new())));
-        }
+        self.to_local.insert(identifier, Rc::new(RefCell::new(VecDeque::new())));
 
-        let puller = Box::new(PullerInner::new(inner_recv, self.to_local[channel_id].clone()));
+        let puller = Box::new(PullerInner::new(inner_recv, self.to_local[&identifier].clone()));
 
-        (pushes, puller, None)
+        (pushes, puller, )
     }
 
     // Perform preparatory work, most likely reading binary buffers from self.recv.
@@ -186,12 +181,11 @@ impl<A: Allocate> Allocate for TcpAllocator<A> {
 
                     // Ensure that a queue exists.
                     // We may receive data before allocating, and shouldn't block.
-                    while self.to_local.len() <= header.channel {
-                        self.to_local.push(Rc::new(RefCell::new(VecDeque::new())));
-                    }
-
-                    // Introduce the binary slice into the operator input queue.
-                    self.to_local[header.channel].borrow_mut().push_back(peel);
+                    self.to_local
+                        .entry(header.channel)
+                        .or_insert_with(|| Rc::new(RefCell::new(VecDeque::new())))
+                        .borrow_mut()
+                        .push_back(peel);
                 }
                 else {
                     println!("failed to read full header!");