wip

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

Author: antiguru

communication/Cargo.toml

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

communication/examples/comm_hello.rs

@@ -1,7 +1,7 @@
 extern crate timely_communication;
 
 use std::ops::Deref;
-use timely_communication::{Message, Allocate};
+use timely_communication::{Message, Allocate, Result};
 
 fn main() {
 
@@ -35,7 +35,7 @@ fn main() {
             allocator.release()?;
         }
 
-        Result::<_, anyhow::Error>::Ok(allocator.index())
+        Result::Ok(allocator.index())
     });
 
     // computation runs until guards are joined or dropped.

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, Result};
+use crate::{Push, Pull};
 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>) -> Result<()>{
+    fn push(&mut self, element: &mut Option<T>) -> crate::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>) -> Result<()>{
+    fn push(&mut self, element: &mut Option<T>) -> crate::Result<()>{
         // if element.is_none() {
         //     if self.count != 0 {
         //         self.events

communication/src/allocator/generic.rs

@@ -57,7 +57,7 @@ impl Generic {
         }
     }
     /// Perform work before scheduling operators.
-    fn receive(&mut self) -> anyhow::Result<()> {
+    fn receive(&mut self) -> crate::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) -> anyhow::Result<()> {
+    pub fn release(&mut self) -> crate::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) -> anyhow::Result<()> { self.receive() }
-    fn release(&mut self) -> anyhow::Result<()> { self.release() }
+    fn receive(&mut self) -> crate::Result<()> { self.receive() }
+    fn release(&mut self) -> crate::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, Result};
+use crate::{Data, Push, Pull, Message};
 
 /// 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) -> Result<()> { Ok(()) }
+    fn receive(&mut self) -> crate::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) -> Result<()> { Ok(()) }
+    fn release(&mut self) -> crate::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, Result};
+use crate::{Push, Pull, Message};
 use crate::buzzer::Buzzer;
 
 /// An allocator for inter-thread, intra-process communication
@@ -182,7 +182,7 @@ impl Allocate for Process {
         self.inner.await_events(duration);
     }
 
-    fn receive(&mut self) -> anyhow::Result<()>{
+    fn receive(&mut self) -> crate::Result<()>{
         let mut events = self.inner.events().borrow_mut();
         while let Ok((index, event)) = self.counters_recv.try_recv() {
             events.push_back((index, event));
@@ -205,7 +205,7 @@ impl<T> Clone for Pusher<T> {
 }
 
 impl<T> Push<T> for Pusher<T> {
-    #[inline] fn push(&mut self, element: &mut Option<T>) -> Result<()>{
+    #[inline] fn push(&mut self, element: &mut Option<T>) -> crate::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.

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, Result};
+use crate::{Push, Pull, Message};
 
 /// Builder for single-threaded allocator.
 pub struct ThreadBuilder;
@@ -81,7 +81,7 @@ pub struct Pusher<T> {
 
 impl<T> Push<T> for Pusher<T> {
     #[inline]
-    fn push(&mut self, element: &mut Option<T>) -> Result<()>{
+    fn push(&mut self, element: &mut Option<T>) -> crate::Result<()>{
         let mut borrow = self.target.borrow_mut();
         if let Some(element) = element.take() {
             borrow.0.push_back(element);

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) -> anyhow::Result<()> {
+    fn receive(&mut self) -> crate::Result<()> {
 
         // Check for channels whose `Puller` has been dropped.
         let mut canaries = self.canaries.borrow_mut();
@@ -205,7 +205,7 @@ impl<A: Allocate> Allocate for TcpAllocator<A> {
             // on events from it.
             // assert!(dropped.borrow().is_empty());
         }
-        drop(canaries);
+        ::std::mem::drop(canaries);
 
         self.inner.receive()?;
 
@@ -255,7 +255,7 @@ impl<A: Allocate> Allocate for TcpAllocator<A> {
     }
 
     // Perform postparatory work, most likely sending un-full binary buffers.
-    fn release(&mut self) -> anyhow::Result<()> {
+    fn release(&mut self) -> crate::Result<()> {
         // Publish outgoing byte ledgers.
         for send in self.sends.iter_mut() {
             send.borrow_mut().publish()?;

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) -> anyhow::Result<()> {
+    fn receive(&mut self) -> crate::Result<()> {
 
         // Check for channels whose `Puller` has been dropped.
         let mut canaries = self.canaries.borrow_mut();
@@ -222,7 +222,7 @@ impl Allocate for ProcessAllocator {
     }
 
     // Perform postparatory work, most likely sending un-full binary buffers.
-    fn release(&mut self) -> anyhow::Result<()> {
+    fn release(&mut self) -> crate::Result<()> {
         // Publish outgoing byte ledgers.
         for send in self.sends.iter_mut() {
             send.borrow_mut().publish()?;

communication/src/allocator/zero_copy/bytes_exchange.rs

@@ -1,30 +1,28 @@
 //! Types and traits for sharing `Bytes`.
 
 use std::sync::{Arc, Mutex};
-use std::sync::atomic::{AtomicBool, Ordering};
 use std::collections::VecDeque;
 use anyhow::bail;
 
 use bytes::arc::Bytes;
 use super::bytes_slab::BytesSlab;
 
-use crate::Result;
-
 /// A target for `Bytes`.
 pub trait BytesPush {
     // /// Pushes bytes at the instance.
     // fn push(&mut self, bytes: Bytes);
     /// Pushes many bytes at the instance.
-    fn extend<I: IntoIterator<Item=Bytes>>(&mut self, iter: I) -> Result<()>;
+    fn extend<I: IntoIterator<Item=Bytes>>(&mut self, iter: I) -> crate::Result<()>;
 }
 /// A source for `Bytes`.
 pub trait BytesPull {
     // /// Pulls bytes from the instance.
     // fn pull(&mut self) -> Option<Bytes>;
     /// Drains many bytes from the instance.
-    fn drain_into(&mut self, vec: &mut Vec<Bytes>) -> Result<()>;
+    fn drain_into(&mut self, vec: &mut Vec<Bytes>) -> crate::Result<()>;
 }
 
+use std::sync::atomic::{AtomicBool, Ordering};
 /// An unbounded queue of bytes intended for point-to-point communication
 /// between threads. Cloning returns another handle to the same queue.
 ///
@@ -46,19 +44,19 @@ impl MergeQueue {
         }
     }
     /// Indicates that all input handles to the queue have dropped.
-    pub fn is_complete(&self) -> Result<bool> {
+    pub fn is_complete(&self) -> crate::Result<bool> {
         if self.panic.load(Ordering::SeqCst) { bail!("MergeQueue poisoned."); }
-        Ok(Arc::strong_count(&self.queue) == 1 && self.queue.lock().map_err(|_| anyhow::anyhow!("MergeQueue mutex poisoned."))?.is_empty())
+        Ok(Arc::strong_count(&self.queue) == 1 && self.queue.lock().map_err(|e| anyhow::anyhow!("MergeQueue mutex poisoned: {e}"))?.is_empty())
     }
 
-    /// TODO
+    /// Mark self as poisoned, which causes all subsequent operations to error.
     pub fn poison(&mut self) {
         self.panic.store(true, Ordering::SeqCst);
     }
 }
 
 impl BytesPush for MergeQueue {
-    fn extend<I: IntoIterator<Item=Bytes>>(&mut self, iterator: I) -> Result<()> {
+    fn extend<I: IntoIterator<Item=Bytes>>(&mut self, iterator: I) -> crate::Result<()> {
 
         if self.panic.load(Ordering::SeqCst) { bail!("MergeQueue poisoned."); }
 
@@ -68,7 +66,7 @@ impl BytesPush for MergeQueue {
             lock_ok = self.queue.try_lock();
         }
         let mut queue = lock_ok
-            .map_err(|_| anyhow::anyhow!("MergeQueue mutex poisoned."))?;
+            .map_err(|e| anyhow::anyhow!("MergeQueue mutex poisoned: {e}"))?;
 
         let mut iterator = iterator.into_iter();
         let mut should_ping = false;
@@ -102,7 +100,7 @@ impl BytesPush for MergeQueue {
 }
 
 impl BytesPull for MergeQueue {
-    fn drain_into(&mut self, vec: &mut Vec<Bytes>) -> Result<()> {
+    fn drain_into(&mut self, vec: &mut Vec<Bytes>) -> crate::Result<()> {
         if self.panic.load(Ordering::SeqCst) { bail!("MergeQueue poisoned."); }
 
         // try to acquire lock without going to sleep (Rust's lock() might yield)
@@ -111,7 +109,7 @@ impl BytesPull for MergeQueue {
             lock_ok = self.queue.try_lock();
         }
         let mut queue = lock_ok
-            .map_err(|_| anyhow::anyhow!("MergeQueue mutex poisoned."))?;
+            .map_err(|e| anyhow::anyhow!("MergeQueue mutex poisoned: {e}"))?;
 
         vec.extend(queue.drain(..));
         Ok(())
@@ -128,7 +126,7 @@ impl Drop for MergeQueue {
         }
         else {
             // TODO: Perhaps this aggressive ordering can relax orderings elsewhere.
-            // if self.panic.load(Ordering::SeqCst) { panic!("MergeQueue poisoned."); }
+            if self.panic.load(Ordering::SeqCst) { panic!("MergeQueue poisoned."); }
         }
         // Drop the queue before pinging.
         self.queue = Arc::new(Mutex::new(VecDeque::new()));
@@ -146,7 +144,7 @@ pub struct SendEndpoint<P: BytesPush> {
 impl<P: BytesPush> SendEndpoint<P> {
 
     /// Moves `self.buffer` into `self.send`, replaces with empty buffer.
-    fn send_buffer(&mut self) -> Result<()> {
+    fn send_buffer(&mut self) -> crate::Result<()> {
         let valid_len = self.buffer.valid().len();
         if valid_len > 0 {
             self.send.extend(Some(self.buffer.extract(valid_len)))?;
@@ -164,12 +162,12 @@ impl<P: BytesPush> SendEndpoint<P> {
     /// Makes the next `bytes` bytes valid.
     ///
     /// The current implementation also sends the bytes, to ensure early visibility.
-    pub fn make_valid(&mut self, bytes: usize) -> Result<()> {
+    pub fn make_valid(&mut self, bytes: usize) -> crate::Result<()> {
         self.buffer.make_valid(bytes);
         self.send_buffer()
     }
     /// Acquires a prefix of `self.empty()` of length at least `capacity`.
-    pub fn reserve(&mut self, capacity: usize) -> Result<&mut [u8]> {
+    pub fn reserve(&mut self, capacity: usize) -> crate::Result<&mut [u8]> {
 
         if self.buffer.empty().len() < capacity {
             self.send_buffer()?;
@@ -180,7 +178,7 @@ impl<P: BytesPush> SendEndpoint<P> {
         Ok(self.buffer.empty())
     }
     /// Marks all written data as valid, makes visible.
-    pub fn publish(&mut self) -> Result<()> {
+    pub fn publish(&mut self) -> crate::Result<()> {
         self.send_buffer()
     }
 }