wip

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

Author: antiguru

communication/Cargo.toml

@@ -17,7 +17,6 @@ license = "MIT"
 default = ["getopts"]
 
 [dependencies]
-anyhow = "1.0.68"
 getopts = { version = "0.2.14", optional = true }
 bincode = { version = "1.0", optional = true }
 serde_derive = "1.0"

communication/examples/comm_hello.rs

@@ -35,7 +35,7 @@ fn main() {
             allocator.release()?;
         }
 
-        Result::Ok(allocator.index())
+        Ok(allocator.index())
     });
 
     // computation runs until guards are joined or dropped.

communication/src/allocator/counters.rs

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

communication/src/allocator/process.rs

@@ -182,7 +182,7 @@ impl Allocate for Process {
         self.inner.await_events(duration);
     }
 
-    fn receive(&mut self) -> crate::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>) -> crate::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

@@ -81,7 +81,7 @@ pub struct Pusher<T> {
 
 impl<T> Push<T> for Pusher<T> {
     #[inline]
-    fn push(&mut self, element: &mut Option<T>) -> crate::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/bytes_exchange.rs

@@ -2,10 +2,10 @@
 
 use std::sync::{Arc, Mutex};
 use std::collections::VecDeque;
-use anyhow::bail;
 
 use bytes::arc::Bytes;
 use super::bytes_slab::BytesSlab;
+use crate::err::CommError;
 
 /// A target for `Bytes`.
 pub trait BytesPush {
@@ -45,8 +45,8 @@ impl MergeQueue {
     }
     /// Indicates that all input handles to the queue have dropped.
     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(|e| anyhow::anyhow!("MergeQueue mutex poisoned: {e}"))?.is_empty())
+        if self.panic.load(Ordering::SeqCst) { return Err(CommError::Poison); }
+        Ok(Arc::strong_count(&self.queue) == 1 && self.queue.lock()?.is_empty())
     }
 
     /// Mark self as poisoned, which causes all subsequent operations to error.
@@ -58,15 +58,14 @@ impl MergeQueue {
 impl BytesPush for MergeQueue {
     fn extend<I: IntoIterator<Item=Bytes>>(&mut self, iterator: I) -> crate::Result<()> {
 
-        if self.panic.load(Ordering::SeqCst) { bail!("MergeQueue poisoned."); }
+        if self.panic.load(Ordering::SeqCst) { return Err(CommError::Poison); }
 
         // try to acquire lock without going to sleep (Rust's lock() might yield)
         let mut lock_ok = self.queue.try_lock();
         while let Err(::std::sync::TryLockError::WouldBlock) = lock_ok {
             lock_ok = self.queue.try_lock();
         }
-        let mut queue = lock_ok
-            .map_err(|e| anyhow::anyhow!("MergeQueue mutex poisoned: {e}"))?;
+        let mut queue = lock_ok?;
 
         let mut iterator = iterator.into_iter();
         let mut should_ping = false;
@@ -101,15 +100,14 @@ impl BytesPush for MergeQueue {
 
 impl BytesPull for MergeQueue {
     fn drain_into(&mut self, vec: &mut Vec<Bytes>) -> crate::Result<()> {
-        if self.panic.load(Ordering::SeqCst) { bail!("MergeQueue poisoned."); }
+        if self.panic.load(Ordering::SeqCst) { return Err(CommError::Poison); }
 
         // try to acquire lock without going to sleep (Rust's lock() might yield)
         let mut lock_ok = self.queue.try_lock();
         while let Err(::std::sync::TryLockError::WouldBlock) = lock_ok {
             lock_ok = self.queue.try_lock();
         }
-        let mut queue = lock_ok
-            .map_err(|e| anyhow::anyhow!("MergeQueue mutex poisoned: {e}"))?;
+        let mut queue = lock_ok?;
 
         vec.extend(queue.drain(..));
         Ok(())

communication/src/allocator/zero_copy/initialize.rs

@@ -4,9 +4,9 @@ use std::sync::Arc;
 // use crate::allocator::Process;
 use crate::allocator::process::ProcessBuilder;
 use crate::networking::create_sockets;
+use super::tcp::{send_loop, recv_loop};
 use super::allocator::{TcpBuilder, new_vector};
 use super::stream::Stream;
-use super::tcp::{send_loop, recv_loop};
 
 /// Join handles for send and receive threads.
 ///

communication/src/allocator/zero_copy/push_pull.rs

@@ -3,7 +3,6 @@
 use std::rc::Rc;
 use std::cell::RefCell;
 use std::collections::VecDeque;
-use anyhow::Context;
 
 use bytes::arc::Bytes;
 
@@ -38,7 +37,7 @@ impl<T, P: BytesPush> Pusher<T, P> {
 
 impl<T:Data, P: BytesPush> Push<Message<T>> for Pusher<T, P> {
     #[inline]
-    fn push(&mut self, element: &mut Option<Message<T>>) -> crate::Result<()>{
+    fn push(&mut self, element: &mut Option<Message<T>>) -> crate::Result<()> {
         if let Some(ref mut element) = *element {
 
             // determine byte lengths and build header.
@@ -53,7 +52,7 @@ impl<T:Data, P: BytesPush> Push<Message<T>> for Pusher<T, P> {
                 let mut bytes = borrow.reserve(header.required_bytes())?;
                 assert!(bytes.len() >= header.required_bytes());
                 let writer = &mut bytes;
-                header.write_to(writer).context("writing header")?;
+                header.write_to(writer)?;
                 element.into_bytes(writer);
             }
             borrow.make_valid(header.required_bytes())?;

communication/src/allocator/zero_copy/tcp.rs

@@ -1,7 +1,6 @@
 //!
 
 use std::io::Write;
-use anyhow::{bail, Context};
 use crossbeam_channel::{Sender, Receiver};
 
 use crate::networking::MessageHeader;
@@ -12,6 +11,7 @@ use super::stream::Stream;
 
 use logging_core::Logger;
 
+use crate::err::CommError;
 use crate::logging::{CommunicationEvent, CommunicationSetup, MessageEvent, StateEvent};
 
 /// Repeatedly reads from a TcpStream and carves out messages.
@@ -83,8 +83,8 @@ where
         assert!(!buffer.empty().is_empty());
 
         // Attempt to read some more bytes into self.buffer.
-        let read = match reader.read(&mut buffer.empty()).context("reading data")? {
-            0 => bail!("reading data: Unexpected EOF"),
+        let read = match reader.read(&mut buffer.empty())? {
+            0 => { return Err(std::io::Error::from(std::io::ErrorKind::UnexpectedEof).into()); },
             n => n,
         };
 
@@ -109,11 +109,11 @@ where
                 // Shutting down; confirm absence of subsequent data.
                 active = false;
                 if !buffer.valid().is_empty() {
-                    bail!("Clean shutdown followed by data.");
+                    return Err(CommError::UnexpectedData);
                 }
                 buffer.ensure_capacity(1);
-                if reader.read(&mut buffer.empty()).context("reading data")? > 0 {
-                    bail!("Clean shutdown followed by data.");
+                if reader.read(&mut buffer.empty())? > 0 {
+                    return Err(CommError::UnexpectedData);
                 }
             }
         }
@@ -174,7 +174,7 @@ pub fn send_loop<S: Stream>(
             // still be a signal incoming.
             //
             // We could get awoken by more data, a channel closing, or spuriously perhaps.
-            writer.flush().context("Flushing writer")?;
+            writer.flush()?;
             for source in sources.iter_mut() {
                 if let Some(s) = source {
                     if s.is_complete()? {
@@ -200,7 +200,7 @@ pub fn send_loop<S: Stream>(
                     }
                 });
 
-                writer.write_all(&bytes[..]).context("writing data")?;
+                writer.write_all(&bytes[..])?;
             }
         }
     }
@@ -215,9 +215,9 @@ pub fn send_loop<S: Stream>(
         length:     0,
         seqno:      0,
     };
-    header.write_to(&mut writer).context("writing data")?;
-    writer.flush().context("flushing writer")?;
-    writer.get_mut().shutdown(::std::net::Shutdown::Write).context("Write shutdown failed")?;
+    header.write_to(&mut writer)?;
+    writer.flush()?;
+    writer.get_mut().shutdown(::std::net::Shutdown::Write)?;
     logger.as_mut().map(|logger| logger.log(MessageEvent { is_send: true, header }));
 
     // Log the send thread's end.

communication/src/initialize.rs

@@ -17,7 +17,6 @@ use crate::allocator::zero_copy::initialize::initialize_networking;
 use crate::logging::{CommunicationSetup, CommunicationEvent};
 use logging_core::Logger;
 use std::fmt::{Debug, Formatter};
-use anyhow::{bail, Context};
 
 
 /// Possible configurations for the communication infrastructure.
@@ -90,23 +89,23 @@ impl Config {
     /// This method is only available if the `getopts` feature is enabled, which
     /// it is by default.
     #[cfg(feature = "getopts")]
-    pub fn from_matches(matches: &getopts::Matches) -> crate::Result<Config> {
-        let threads = matches.opt_get_default("w", 1_usize)?;
-        let process = matches.opt_get_default("p", 0_usize)?;
-        let processes = matches.opt_get_default("n", 1_usize)?;
+    pub fn from_matches(matches: &getopts::Matches) -> Result<Config, String> {
+        let threads = matches.opt_get_default("w", 1_usize).map_err(|e| e.to_string())?;
+        let process = matches.opt_get_default("p", 0_usize).map_err(|e| e.to_string())?;
+        let processes = matches.opt_get_default("n", 1_usize).map_err(|e| e.to_string())?;
         let report = matches.opt_present("report");
         let zerocopy = matches.opt_present("zerocopy");
 
         if processes > 1 {
             let mut addresses = Vec::new();
             if let Some(hosts) = matches.opt_str("h") {
-                let file = ::std::fs::File::open(hosts.clone())?;
+                let file = ::std::fs::File::open(hosts.clone()).map_err(|e| e.to_string())?;
                 let reader = ::std::io::BufReader::new(file);
                 for line in reader.lines().take(processes) {
-                    addresses.push(line?);
+                    addresses.push(line.map_err(|e| e.to_string())?);
                 }
                 if addresses.len() < processes {
-                    bail!("could only read {} addresses from {hosts}, but -n: {processes}", addresses.len());
+                    return Err(format!("could only read {} addresses from {}, but -n: {}", addresses.len(), hosts, processes));
                 }
             }
             else {
@@ -141,10 +140,10 @@ impl Config {
     /// This method is only available if the `getopts` feature is enabled, which
     /// it is by default.
     #[cfg(feature = "getopts")]
-    pub fn from_args<I: Iterator<Item=String>>(args: I) -> crate::Result<Config> {
+    pub fn from_args<I: Iterator<Item=String>>(args: I) -> Result<Config, String> {
         let mut opts = getopts::Options::new();
         Config::install_options(&mut opts);
-        let matches = opts.parse(args)?;
+        let matches = opts.parse(args).map_err(|e| e.to_string())?;
         Config::from_matches(&matches)
     }
 
@@ -161,8 +160,7 @@ impl Config {
                 Ok((ProcessBuilder::new_vector(threads).into_iter().map(|x| GenericBuilder::ProcessBinary(x)).collect(), Box::new(())))
             },
             Config::Cluster { threads, process, addresses, report, log_fn } => {
-                let (stuff, guard) = initialize_networking(addresses, process, threads, report, log_fn)
-                    .context("initializing network")?;
+                let (stuff, guard) = initialize_networking(addresses, process, threads, report, log_fn)?;
                 Ok((stuff.into_iter().map(|x| GenericBuilder::ZeroCopy(x)).collect(), Box::new(guard)))
             },
         }
@@ -337,10 +335,10 @@ impl<T:Send+'static> WorkerGuards<T> {
     }
 
     /// Waits on the worker threads and returns the results they produce.
-    pub fn join(mut self) -> Vec<crate::Result<T>> {
+    pub fn join(mut self) -> Vec<Result<T, String>> {
         self.guards
             .drain(..)
-            .map(|guard| guard.join().map_err(|e| anyhow::anyhow!("{e:?}")))
+            .map(|guard| guard.join().map_err(|e| format!("{:?}", e)))
             .collect()
     }
 }