feat: Improve multiplexer ergonomics (#111)

pallas-multiplexer/src/agents.rs

@@ -1,13 +1,18 @@
 //! Interface to interact with the multiplexer as an agent
 
 use crate::Payload;
-
 use pallas_codec::{minicbor, Fragment};
+use thiserror::Error;
 
-#[derive(Debug)]
+#[derive(Debug, Error)]
 pub enum ChannelError {
+    #[error("channel is not connected, failed to send payload")]
     NotConnected(Option<Payload>),
+
+    #[error("failure encoding message into CBOR")]
     Encoding(String),
+
+    #[error("failure decoding message from CBOR")]
     Decoding(String),
 }
 

pallas-multiplexer/src/bearers.rs

@@ -1,28 +1,19 @@
 use byteorder::{ByteOrder, NetworkEndian, WriteBytesExt};
 use log::{debug, log_enabled, trace};
 use std::io::{Read, Write};
-#[cfg(target_family = "unix")]
-use std::os::unix::net::UnixStream;
+use std::net::{TcpListener, ToSocketAddrs};
 use std::{net::TcpStream, time::Instant};
 
 use crate::Payload;
 
+#[cfg(target_family = "unix")]
+use std::os::unix::net::UnixStream;
+
 pub struct Segment {
     pub protocol: u16,
     pub timestamp: u32,
     pub payload: Payload,
 }
-
-pub trait Bearer: Read + Write + Send + Sync + Sized {
-    type Error: std::error::Error;
-
-    fn read_segment(&mut self) -> Result<Option<Segment>, Self::Error>;
-
-    fn write_segment(&mut self, segment: Segment) -> Result<(), Self::Error>;
-
-    fn clone(&self) -> Self;
-}
-
 impl Segment {
     pub fn new(clock: Instant, protocol: u16, payload: Payload) -> Self {
         Segment {
@@ -104,35 +95,74 @@ fn read_segment_with_timeout(reader: &mut impl Read) -> Result<Option<Segment>,
     }
 }
 
-impl Bearer for TcpStream {
-    type Error = std::io::Error;
+pub enum Bearer {
+    Tcp(TcpStream),
 
-    fn clone(&self) -> Self {
-        self.try_clone().expect("error cloning tcp stream")
+    #[cfg(target_family = "unix")]
+    Unix(UnixStream),
+}
+
+impl Bearer {
+    pub fn connect_tcp<A: ToSocketAddrs>(addr: A) -> Result<Self, std::io::Error> {
+        let bearer = TcpStream::connect(addr)?;
+        bearer.set_nodelay(true)?;
+
+        Ok(Bearer::Tcp(bearer))
     }
 
-    fn read_segment(&mut self) -> Result<Option<Segment>, std::io::Error> {
-        read_segment_with_timeout(self)
+    pub fn accept_tcp(server: TcpListener) -> Result<Self, std::io::Error> {
+        let (bearer, _) = server.accept().unwrap();
+        bearer.set_nodelay(true)?;
+
+        Ok(Bearer::Tcp(bearer))
     }
 
-    fn write_segment(&mut self, segment: Segment) -> Result<(), std::io::Error> {
-        write_segment(self, segment)
+    #[cfg(target_family = "unix")]
+    pub fn connect_unix<P: AsRef<std::path::Path>>(path: P) -> Result<Self, std::io::Error> {
+        let bearer = UnixStream::connect(path)?;
+
+        Ok(Bearer::Unix(bearer))
+    }
+
+    pub fn read_segment(&mut self) -> Result<Option<Segment>, std::io::Error> {
+        match self {
+            Bearer::Tcp(s) => read_segment_with_timeout(s),
+
+            #[cfg(target_family = "unix")]
+            Bearer::Unix(s) => read_segment_with_timeout(s),
+        }
+    }
+
+    pub fn write_segment(&mut self, segment: Segment) -> Result<(), std::io::Error> {
+        match self {
+            Bearer::Tcp(s) => write_segment(s, segment),
+
+            #[cfg(target_family = "unix")]
+            Bearer::Unix(s) => write_segment(s, segment),
+        }
+    }
+}
+
+impl From<TcpStream> for Bearer {
+    fn from(stream: TcpStream) -> Self {
+        Bearer::Tcp(stream)
     }
 }
 
 #[cfg(target_family = "unix")]
-impl Bearer for UnixStream {
-    type Error = std::io::Error;
-
-    fn clone(&self) -> Self {
-        self.try_clone().expect("error cloning unix stream")
-    }
-
-    fn read_segment(&mut self) -> Result<Option<Segment>, std::io::Error> {
-        read_segment_with_timeout(self)
-    }
-
-    fn write_segment(&mut self, segment: Segment) -> Result<(), std::io::Error> {
-        write_segment(self, segment)
+impl From<UnixStream> for Bearer {
+    fn from(stream: UnixStream) -> Self {
+        Bearer::Unix(stream)
+    }
+}
+
+impl Clone for Bearer {
+    fn clone(&self) -> Self {
+        match self {
+            Bearer::Tcp(s) => Bearer::Tcp(s.try_clone().expect("error cloning tcp stream")),
+
+            #[cfg(target_family = "unix")]
+            Bearer::Unix(s) => Bearer::Unix(s.try_clone().expect("error cloning unix stream")),
+        }
     }
 }

pallas-multiplexer/src/demux.rs

@@ -1,6 +1,6 @@
 use std::collections::HashMap;
 
-use crate::{bearers::Bearer, std::Cancel, Payload};
+use crate::{bearers::Bearer, Payload};
 
 pub struct EgressError(pub Payload);
 
@@ -8,8 +8,8 @@ pub trait Egress {
     fn send(&self, payload: Payload) -> Result<(), EgressError>;
 }
 
-pub enum DemuxError<B: Bearer> {
-    BearerError(B::Error),
+pub enum DemuxError {
+    BearerError(std::io::Error),
     EgressDisconnected(u16, Payload),
     EgressUnknown(u16, Payload),
 }
@@ -20,17 +20,16 @@ pub enum TickOutcome {
 }
 
 /// A demuxer that reads from a bearer into the corresponding egress
-pub struct Demuxer<B, E> {
-    bearer: B,
+pub struct Demuxer<E> {
+    bearer: Bearer,
     egress: HashMap<u16, E>,
 }
 
-impl<B, E> Demuxer<B, E>
+impl<E> Demuxer<E>
 where
-    B: Bearer,
     E: Egress,
 {
-    pub fn new(bearer: B) -> Self {
+    pub fn new(bearer: Bearer) -> Self {
         Demuxer {
             bearer,
             egress: Default::default(),
@@ -41,7 +40,7 @@ where
         self.egress.insert(id, tx);
     }
 
-    fn dispatch(&self, protocol: u16, payload: Payload) -> Result<(), DemuxError<B>> {
+    fn dispatch(&self, protocol: u16, payload: Payload) -> Result<(), DemuxError> {
         match self.egress.get(&protocol) {
             Some(tx) => match tx.send(payload) {
                 Err(EgressError(p)) => Err(DemuxError::EgressDisconnected(protocol, p)),
@@ -51,7 +50,7 @@ where
         }
     }
 
-    pub fn tick(&mut self) -> Result<TickOutcome, DemuxError<B>> {
+    pub fn tick(&mut self) -> Result<TickOutcome, DemuxError> {
         match self.bearer.read_segment() {
             Err(err) => Err(DemuxError::BearerError(err)),
             Ok(None) => Ok(TickOutcome::Idle),
@@ -61,23 +60,4 @@ where
             },
         }
     }
-
-    pub fn block(&mut self, cancel: Cancel) -> Result<(), B::Error> {
-        loop {
-            match self.tick() {
-                Ok(TickOutcome::Busy) => (),
-                Ok(TickOutcome::Idle) => match cancel.is_set() {
-                    true => break Ok(()),
-                    false => (),
-                },
-                Err(DemuxError::BearerError(err)) => return Err(err),
-                Err(DemuxError::EgressDisconnected(id, _)) => {
-                    log::warn!("disconnected protocol {}", id)
-                }
-                Err(DemuxError::EgressUnknown(id, _)) => {
-                    log::warn!("unknown protocol {}", id)
-                }
-            }
-        }
-    }
 }

pallas-multiplexer/src/lib.rs

@@ -3,28 +3,34 @@ pub mod bearers;
 pub mod demux;
 pub mod mux;
 
+use bearers::Bearer;
+
+#[cfg(feature = "std")]
+mod std;
+
+#[cfg(feature = "std")]
+pub use crate::std::*;
+
 pub type Payload = Vec<u8>;
 
-pub struct Multiplexer<B, I, E>
+pub struct Multiplexer<I, E>
 where
-    B: bearers::Bearer,
     I: mux::Ingress,
     E: demux::Egress,
 {
-    pub muxer: mux::Muxer<B, I>,
-    pub demuxer: demux::Demuxer<B, E>,
+    pub muxer: mux::Muxer<I>,
+    pub demuxer: demux::Demuxer<E>,
 }
 
-impl<B, I, E> Multiplexer<B, I, E>
+impl<I, E> Multiplexer<I, E>
 where
-    B: bearers::Bearer,
     I: mux::Ingress,
     E: demux::Egress,
 {
-    pub fn new(bearer: B) -> Self {
+    pub fn new(bearer: Bearer) -> Self {
         Multiplexer {
             muxer: mux::Muxer::new(bearer.clone()),
-            demuxer: demux::Demuxer::new(bearer.clone()),
+            demuxer: demux::Demuxer::new(bearer),
         }
     }
 
@@ -33,9 +39,3 @@ where
         self.demuxer.register(protocol, egress);
     }
 }
-
-#[cfg(feature = "std")]
-mod std;
-
-#[cfg(feature = "std")]
-pub use crate::std::*;

pallas-multiplexer/src/mux.rs

@@ -5,7 +5,6 @@ use rand::thread_rng;
 
 use crate::{
     bearers::{Bearer, Segment},
-    std::Cancel,
     Payload,
 };
 
@@ -24,27 +23,23 @@ pub trait Ingress {
 
 type Message = (u16, Payload);
 
-pub enum TickOutcome<TBearer>
-where
-    TBearer: Bearer,
-{
-    BearerError(TBearer::Error),
+pub enum TickOutcome {
+    BearerError(std::io::Error),
     Idle,
     Busy,
 }
 
-pub struct Muxer<B, I> {
-    bearer: B,
+pub struct Muxer<I> {
+    bearer: Bearer,
     ingress: HashMap<u16, I>,
     clock: Instant,
 }
 
-impl<B, I> Muxer<B, I>
+impl<I> Muxer<I>
 where
-    B: Bearer,
     I: Ingress,
 {
-    pub fn new(bearer: B) -> Self {
+    pub fn new(bearer: Bearer) -> Self {
         Self {
             bearer,
             ingress: Default::default(),
@@ -93,7 +88,7 @@ where
         None
     }
 
-    pub fn tick(&mut self) -> TickOutcome<B> {
+    pub fn tick(&mut self) -> TickOutcome {
         match self.select() {
             Some((id, payload)) => {
                 let segment = Segment::new(self.clock, id, payload);
@@ -106,17 +101,4 @@ where
             None => TickOutcome::Idle,
         }
     }
-
-    pub fn block(&mut self, cancel: Cancel) -> Result<(), B::Error> {
-        loop {
-            match self.tick() {
-                TickOutcome::BearerError(err) => return Err(err),
-                TickOutcome::Idle => match cancel.is_set() {
-                    true => break Ok(()),
-                    false => std::thread::yield_now(),
-                },
-                TickOutcome::Busy => (),
-            }
-        }
-    }
 }

pallas-multiplexer/src/std.rs

@@ -1,4 +1,4 @@
-use crate::{agents, bearers::Bearer, demux, mux, Payload};
+use crate::{agents, demux, mux, Payload};
 
 use std::{
     sync::{
@@ -7,6 +7,7 @@ use std::{
         Arc,
     },
     thread::{spawn, JoinHandle},
+    time::Duration,
 };
 
 pub type StdIngress = Receiver<Payload>;
@@ -32,7 +33,73 @@ impl demux::Egress for StdEgress {
     }
 }
 
-pub type StdPlexer<B> = crate::Multiplexer<B, StdIngress, StdEgress>;
+pub type StdPlexer = crate::Multiplexer<StdIngress, StdEgress>;
+
+impl StdPlexer {
+    pub fn use_channel(&mut self, protocol: u16) -> StdChannel {
+        let (demux_tx, demux_rx) = channel::<Payload>();
+        let (mux_tx, mux_rx) = channel::<Payload>();
+
+        self.register_channel(protocol, mux_rx, demux_tx);
+
+        (mux_tx, demux_rx)
+    }
+}
+
+impl mux::Muxer<StdIngress> {
+    pub fn block(&mut self, cancel: Cancel) -> Result<(), std::io::Error> {
+        loop {
+            match self.tick() {
+                mux::TickOutcome::BearerError(err) => return Err(err),
+                mux::TickOutcome::Idle => match cancel.is_set() {
+                    true => break Ok(()),
+                    false => {
+                        // TODO: investigate why std::thread::yield_now() hogs the thread
+                        std::thread::sleep(Duration::from_millis(100))
+                    }
+                },
+                mux::TickOutcome::Busy => (),
+            }
+        }
+    }
+
+    pub fn spawn(mut self) -> Loop {
+        let cancel = Cancel::default();
+        let cancel2 = cancel.clone();
+        let thread = spawn(move || self.block(cancel2));
+
+        Loop { cancel, thread }
+    }
+}
+
+impl demux::Demuxer<StdEgress> {
+    pub fn block(&mut self, cancel: Cancel) -> Result<(), std::io::Error> {
+        loop {
+            match self.tick() {
+                Ok(demux::TickOutcome::Busy) => (),
+                Ok(demux::TickOutcome::Idle) => match cancel.is_set() {
+                    true => break Ok(()),
+                    false => (),
+                },
+                Err(demux::DemuxError::BearerError(err)) => return Err(err),
+                Err(demux::DemuxError::EgressDisconnected(id, _)) => {
+                    log::warn!("disconnected protocol {}", id)
+                }
+                Err(demux::DemuxError::EgressUnknown(id, _)) => {
+                    log::warn!("unknown protocol {}", id)
+                }
+            }
+        }
+    }
+
+    pub fn spawn(mut self) -> Loop {
+        let cancel = Cancel::default();
+        let cancel2 = cancel.clone();
+        let thread = spawn(move || self.block(cancel2));
+
+        Loop { cancel, thread }
+    }
+}
 
 pub type StdChannel = (Sender<Payload>, Receiver<Payload>);
 
@@ -52,15 +119,6 @@ impl agents::Channel for StdChannel {
     }
 }
 
-pub fn use_channel<B: Bearer>(plexer: &mut StdPlexer<B>, protocol: u16) -> StdChannel {
-    let (demux_tx, demux_rx) = channel::<Payload>();
-    let (mux_tx, mux_rx) = channel::<Payload>();
-
-    plexer.register_channel(protocol, mux_rx, demux_tx);
-
-    (mux_tx, demux_rx)
-}
-
 #[derive(Clone, Debug, Default)]
 pub struct Cancel(Arc<AtomicBool>);
 
@@ -75,49 +133,17 @@ impl Cancel {
 }
 
 #[derive(Debug)]
-pub struct Loop<B>
-where
-    B: Bearer,
-{
+pub struct Loop {
     cancel: Cancel,
-    thread: JoinHandle<Result<(), B::Error>>,
+    thread: JoinHandle<Result<(), std::io::Error>>,
 }
 
-impl<B> Loop<B>
-where
-    B: Bearer,
-{
+impl Loop {
     pub fn cancel(&self) {
         self.cancel.set();
     }
 
-    pub fn join(self) -> Result<(), B::Error> {
+    pub fn join(self) -> Result<(), std::io::Error> {
         self.thread.join().unwrap()
     }
 }
-
-pub fn spawn_muxer<B, I>(mut muxer: mux::Muxer<B, I>) -> Loop<B>
-where
-    B: Bearer + 'static,
-    B::Error: Send,
-    I: mux::Ingress + Send + 'static,
-{
-    let cancel = Cancel::default();
-    let cancel2 = cancel.clone();
-    let thread = spawn(move || muxer.block(cancel2));
-
-    Loop { cancel, thread }
-}
-
-pub fn spawn_demuxer<B, E>(mut demuxer: demux::Demuxer<B, E>) -> Loop<B>
-where
-    B: Bearer + 'static,
-    B::Error: Send,
-    E: demux::Egress + Send + 'static,
-{
-    let cancel = Cancel::default();
-    let cancel2 = cancel.clone();
-    let thread = spawn(move || demuxer.block(cancel2));
-
-    Loop { cancel, thread }
-}