Improved (?) error handling for communication and timely

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

Author: antiguru

README.md

@@ -82,9 +82,10 @@ fn main() {
             }
             input.advance_to(round + 1);
             while probe.less_than(input.time()) {
-                worker.step();
+                worker.step()?;
             }
         }
+        Ok(())
     }).unwrap();
 }
 ```

communication/Cargo.toml

@@ -17,12 +17,13 @@ license = "MIT"
 default = ["getopts"]
 
 [dependencies]
-getopts = { version = "0.2.14", optional = true }
-bincode = { version = "1.0", optional = true }
-serde_derive = "1.0"
-serde = "1.0"
+anyhow = "1.0.57"
 abomonation = "0.7"
 abomonation_derive = "0.5"
+bincode = { version = "1.0", optional = true }
+crossbeam-channel = "0.5.0"
+getopts = { version = "0.2.14", optional = true }
+serde = "1.0"
+serde_derive = "1.0"
 timely_bytes = { path = "../bytes", version = "0.12" }
 timely_logging = { path = "../logging", version = "0.12" }
-crossbeam-channel = "0.5.0"

communication/examples/comm_hello.rs

@@ -16,32 +16,32 @@ fn main() {
 
         // send typed data along each channel
         for i in 0 .. allocator.peers() {
-            senders[i].send(Message::from_typed(format!("hello, {}", i)));
-            senders[i].done();
+            senders[i].send(Message::from_typed(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();
+            allocator.receive()?;
 
             if let Some(message) = receiver.recv() {
                 println!("worker {}: received: <{}>", allocator.index(), message.deref());
                 received += 1;
             }
 
-            allocator.release();
+            allocator.release()?;
         }
 
-        allocator.index()
+        Result::<_, anyhow::Error>::Ok(allocator.index())
     });
 
     // computation runs until guards are joined or dropped.
     if let Ok(guards) = guards {
         for guard in guards.join() {
-            println!("result: {:?}", guard);
+            println!("result: {:?}", guard.unwrap().unwrap());
         }
     }
     else { println!("error in computation"); }

communication/src/allocator/counters.rs

@@ -4,7 +4,7 @@ use std::rc::Rc;
 use std::cell::RefCell;
 use std::collections::VecDeque;
 
-use crate::{Push, Pull};
+use crate::{Push, Pull, Result};
 use crate::allocator::Event;
 
 /// The push half of an intra-thread channel.
@@ -31,7 +31,7 @@ impl<T, P: Push<T>>  Pusher<T, P> {
 
 impl<T, P: Push<T>> Push<T> for Pusher<T, P> {
     #[inline]
-    fn push(&mut self, element: &mut Option<T>) {
+    fn push(&mut self, element: &mut Option<T>) -> Result<()>{
         // if element.is_none() {
         //     if self.count != 0 {
         //         self.events
@@ -81,7 +81,7 @@ impl<T, P: Push<T>>  ArcPusher<T, P> {
 
 impl<T, P: Push<T>> Push<T> for ArcPusher<T, P> {
     #[inline]
-    fn push(&mut self, element: &mut Option<T>) {
+    fn push(&mut self, element: &mut Option<T>) -> Result<()>{
         // if element.is_none() {
         //     if self.count != 0 {
         //         self.events
@@ -98,11 +98,12 @@ impl<T, P: Push<T>> Push<T> for ArcPusher<T, P> {
         // we first enqueue data, second enqueue interest in the channel,
         // and finally awaken the thread. Other orders are defective when
         // multiple threads are involved.
-        self.pusher.push(element);
+        self.pusher.push(element)?;
         let _ = self.events.send((self.index, Event::Pushed(1)));
             // TODO : Perhaps this shouldn't be a fatal error (e.g. in shutdown).
             // .expect("Failed to send message count");
         self.buzzer.buzz();
+        Ok(())
     }
 }
 

communication/src/allocator/generic.rs

@@ -57,7 +57,7 @@ impl Generic {
         }
     }
     /// Perform work before scheduling operators.
-    fn receive(&mut self) {
+    fn receive(&mut self) -> anyhow::Result<()> {
         match self {
             Generic::Thread(t) => t.receive(),
             Generic::Process(p) => p.receive(),
@@ -66,7 +66,7 @@ impl Generic {
         }
     }
     /// Perform work after scheduling operators.
-    pub fn release(&mut self) {
+    pub fn release(&mut self) -> anyhow::Result<()> {
         match self {
             Generic::Thread(t) => t.release(),
             Generic::Process(p) => p.release(),
@@ -91,8 +91,8 @@ impl Allocate for Generic {
         self.allocate(identifier)
     }
 
-    fn receive(&mut self) { self.receive(); }
-    fn release(&mut self) { self.release(); }
+    fn receive(&mut self) -> anyhow::Result<()> { self.receive() }
+    fn release(&mut self) -> anyhow::Result<()> { self.release() }
     fn events(&self) -> &Rc<RefCell<VecDeque<(usize, Event)>>> { self.events() }
     fn await_events(&self, _duration: Option<std::time::Duration>) {
         match self {

communication/src/allocator/mod.rs

@@ -18,7 +18,7 @@ pub mod counters;
 
 pub mod zero_copy;
 
-use crate::{Data, Push, Pull, Message};
+use crate::{Data, Push, Pull, Message, Result};
 
 /// A proto-allocator, which implements `Send` and can be completed with `build`.
 ///
@@ -70,7 +70,7 @@ pub trait Allocate {
     /// present messages contained in otherwise scarce resources (for example
     /// network buffers), under the premise that someone is about to consume
     /// the messages and release the resources.
-    fn receive(&mut self) { }
+    fn receive(&mut self) -> Result<()> { Ok(()) }
 
     /// Signal the completion of a batch of reads from channels.
     ///
@@ -79,7 +79,7 @@ pub trait Allocate {
     /// the fabric should consider re-acquiring scarce resources. This can
     /// lead to the fabric performing defensive copies out of un-consumed
     /// buffers, and can be a performance problem if invoked casually.
-    fn release(&mut self) { }
+    fn release(&mut self) -> Result<()> { Ok(()) }
 
     /// Constructs a pipeline channel from the worker to itself.
     ///

communication/src/allocator/process.rs

@@ -10,7 +10,7 @@ use crossbeam_channel::{Sender, Receiver};
 
 use crate::allocator::thread::{ThreadBuilder};
 use crate::allocator::{Allocate, AllocateBuilder, Event, Thread};
-use crate::{Push, Pull, Message};
+use crate::{Push, Pull, Message, Result};
 use crate::buzzer::Buzzer;
 
 /// An allocator for inter-thread, intra-process communication
@@ -182,11 +182,12 @@ impl Allocate for Process {
         self.inner.await_events(duration);
     }
 
-    fn receive(&mut self) {
+    fn receive(&mut self) -> anyhow::Result<()>{
         let mut events = self.inner.events().borrow_mut();
         while let Ok((index, event)) = self.counters_recv.try_recv() {
             events.push_back((index, event));
         }
+        Ok(())
     }
 }
 
@@ -204,12 +205,13 @@ impl<T> Clone for Pusher<T> {
 }
 
 impl<T> Push<T> for Pusher<T> {
-    #[inline] fn push(&mut self, element: &mut Option<T>) {
+    #[inline] fn push(&mut self, element: &mut Option<T>) -> Result<()>{
         if let Some(element) = element.take() {
             // The remote endpoint could be shut down, and so
             // it is not fundamentally an error to fail to send.
             let _ = self.target.send(element);
         }
+        Ok(())
     }
 }
 

communication/src/allocator/thread.rs

@@ -8,7 +8,7 @@ use std::collections::VecDeque;
 use crate::allocator::{Allocate, AllocateBuilder, Event};
 use crate::allocator::counters::Pusher as CountPusher;
 use crate::allocator::counters::Puller as CountPuller;
-use crate::{Push, Pull, Message};
+use crate::{Push, Pull, Message, Result};
 
 /// Builder for single-threaded allocator.
 pub struct ThreadBuilder;
@@ -81,12 +81,13 @@ pub struct Pusher<T> {
 
 impl<T> Push<T> for Pusher<T> {
     #[inline]
-    fn push(&mut self, element: &mut Option<T>) {
+    fn push(&mut self, element: &mut Option<T>) -> Result<()>{
         let mut borrow = self.target.borrow_mut();
         if let Some(element) = element.take() {
             borrow.0.push_back(element);
         }
         *element = borrow.1.pop_front();
+        Ok(())
     }
 }
 

communication/src/allocator/zero_copy/allocator.rs

@@ -190,7 +190,7 @@ impl<A: Allocate> Allocate for TcpAllocator<A> {
 
     // Perform preparatory work, most likely reading binary buffers from self.recv.
     #[inline(never)]
-    fn receive(&mut self) {
+    fn receive(&mut self) -> anyhow::Result<()> {
 
         // Check for channels whose `Puller` has been dropped.
         let mut canaries = self.canaries.borrow_mut();
@@ -205,12 +205,12 @@ impl<A: Allocate> Allocate for TcpAllocator<A> {
             // on events from it.
             // assert!(dropped.borrow().is_empty());
         }
-        ::std::mem::drop(canaries);
+        drop(canaries);
 
-        self.inner.receive();
+        self.inner.receive()?;
 
         for recv in self.recvs.iter_mut() {
-            recv.drain_into(&mut self.staged);
+            recv.drain_into(&mut self.staged)?;
         }
 
         let mut events = self.inner.events().borrow_mut();
@@ -251,13 +251,14 @@ impl<A: Allocate> Allocate for TcpAllocator<A> {
                 }
             }
         }
+        Ok(())
     }
 
     // Perform postparatory work, most likely sending un-full binary buffers.
-    fn release(&mut self) {
+    fn release(&mut self) -> anyhow::Result<()> {
         // Publish outgoing byte ledgers.
         for send in self.sends.iter_mut() {
-            send.borrow_mut().publish();
+            send.borrow_mut().publish()?;
         }
 
         // OPTIONAL: Tattle on channels sitting on borrowed data.
@@ -268,11 +269,12 @@ impl<A: Allocate> Allocate for TcpAllocator<A> {
         //         eprintln!("Warning: worker {}, undrained channel[{}].len() = {}", self.index, index, len);
         //     }
         // }
+        Ok(())
     }
     fn events(&self) -> &Rc<RefCell<VecDeque<(usize, Event)>>> {
         self.inner.events()
     }
     fn await_events(&self, duration: Option<std::time::Duration>) {
         self.inner.await_events(duration);
     }
-}
+}

communication/src/allocator/zero_copy/allocator_process.rs

@@ -159,7 +159,7 @@ impl Allocate for ProcessAllocator {
 
     // Perform preparatory work, most likely reading binary buffers from self.recv.
     #[inline(never)]
-    fn receive(&mut self) {
+    fn receive(&mut self) -> anyhow::Result<()> {
 
         // Check for channels whose `Puller` has been dropped.
         let mut canaries = self.canaries.borrow_mut();
@@ -179,7 +179,7 @@ impl Allocate for ProcessAllocator {
         let mut events = self.events.borrow_mut();
 
         for recv in self.recvs.iter_mut() {
-            recv.drain_into(&mut self.staged);
+            recv.drain_into(&mut self.staged)?;
         }
 
         for mut bytes in self.staged.drain(..) {
@@ -218,13 +218,14 @@ impl Allocate for ProcessAllocator {
                 }
             }
         }
+        Ok(())
     }
 
     // Perform postparatory work, most likely sending un-full binary buffers.
-    fn release(&mut self) {
+    fn release(&mut self) -> anyhow::Result<()> {
         // Publish outgoing byte ledgers.
         for send in self.sends.iter_mut() {
-            send.borrow_mut().publish();
+            send.borrow_mut().publish()?;
         }
 
         // OPTIONAL: Tattle on channels sitting on borrowed data.
@@ -235,6 +236,7 @@ impl Allocate for ProcessAllocator {
         //         eprintln!("Warning: worker {}, undrained channel[{}].len() = {}", self.index, index, len);
         //     }
         // }
+        Ok(())
     }
 
     fn events(&self) -> &Rc<RefCell<VecDeque<(usize, Event)>>> {
@@ -250,4 +252,4 @@ impl Allocate for ProcessAllocator {
             }
         }
     }
-}
+}