Rust publish/subscribe channel library

Question

I've written a simple multiple-sender multiple-receiver channel, but I'm unsure if my approach is the best way to do this. I'm not using multi-threaded primitives but instead wrapping around std's builtin sync::mpsc. Performance is fine so far for my use case, but I'm unsure how well this will scale given the fact that it works by keeping track of a list of senders.

Documentation: https://creato.gitlab.io/pub-sub/pub_sub/index.html

Source code:

#![warn(missing_docs)]

//! A basic publish/subscribe channel.
//!
//! # Usage
//!
//! Add to crate dependencies:
//! 
//! ```toml
//! [dependencies]
//! pub-sub = "*"
//! ```
//! Import in crate root:
//!
//! ```
//! extern crate pub_sub;
//! ```
//!
//! # Example
//!
//! ```
//! extern crate pub_sub;
//! extern crate uuid;
//! 
//! use std::thread;
//! use uuid::Uuid;
//! 
//! fn main() {
//!     let (send, recv) = pub_sub::new();
//!     // send: pub_sub::Sender<Uuid>
//!     // recv: pub_sub::Receiver<Uuid>
//! 
//!     for _ in 0..16 {
//!         let recv = recv.clone();
//! 
//!         thread::spawn(move || {
//!             while let Ok(msg) = recv.recv() {
//!                 println!("recevied {}", msg);
//!             }
//!         });
//!     }
//! 
//!     for _ in 0..16 {
//!         let send = send.clone();
//! 
//!         thread::spawn(move || {
//!             let msg_id = Uuid::new_v4();
//!             println!("    sent {}", msg_id);
//!             send.send(msg_id);
//!         });
//!     }
//! }
//! ```

#[macro_use]
extern crate log;
extern crate uuid;

use std::sync::{mpsc, Arc, Mutex};
use std::collections::HashMap;


/// Sending component of a pub/sub channel.
#[derive(Clone)]
pub struct Sender<T: Clone> {
    senders: Arc<Mutex<HashMap<uuid::Uuid, mpsc::Sender<T>>>>,
}

/// Receiver component of a pub/sub channel.
pub struct Receiver<T: Clone> {
    receiver: mpsc::Receiver<T>,
    senders: Arc<Mutex<HashMap<uuid::Uuid, mpsc::Sender<T>>>>,
    id: uuid::Uuid,
}

impl<T: Clone> Sender<T> {
    /// Attempts to broadcast
    pub fn send(&self, it: T) -> Result<(), mpsc::SendError<T>> {
        let senders = self.senders.lock().unwrap();

        for (_, sender) in senders.iter() {
            match sender.send(it.clone()) {
                Ok(_) => {}
                Err(err) => return Err(err),
            }
        }

        Ok(())
    }
}

impl<T: Clone> Receiver<T> {
    /// Receives a single message. Blocks until a message is available.
    pub fn recv(&self) -> Result<T, mpsc::RecvError> {
        self.receiver.recv()
    }

    /// Tries to receive a single message, not blocking if one is not available.
    pub fn try_recv(&self) -> Result<T, mpsc::TryRecvError> {
        self.receiver.try_recv()
    }

    /// Creates an iterator that will block waiting for messages.
    pub fn iter(&self) -> mpsc::Iter<T> {
        self.receiver.iter()
    }
}


impl<T: Clone> Clone for Receiver<T> {
    /// Create a new receiver associated with the sender.
    fn clone(&self) -> Self {
        let id = uuid::Uuid::new_v4();
        let (send, recv) = mpsc::channel();

        {
            let mut senders = self.senders.lock().unwrap();
            senders.insert(id, send);
        }

        Receiver {
            receiver: recv,
            senders: self.senders.clone(),
            id: id,
        }
    }
}

impl<T: Clone> Drop for Receiver<T> {
    /// Remove our sender ID from the sender list.
    fn drop(&mut self) {
        let mut senders = self.senders.lock().unwrap();
        senders.remove(&self.id);
    }
}

/// Create a pub/sub channel
pub fn new<T: Clone>() -> (Sender<T>, Receiver<T>) {
    let mut senders = HashMap::new();

    let initial_id = uuid::Uuid::new_v4();
    let (send, recv) = mpsc::channel();

    senders.insert(initial_id, send);

    let senders = Arc::new(Mutex::new(senders));

    (Sender { senders: senders.clone() },
     Receiver {
        senders: senders.clone(),
        id: initial_id,
        receiver: recv,
    })
}

#[cfg(test)]
extern crate env_logger;

#[cfg(test)]
mod tests {
    use std;

    use super::*;

    fn pre() {
        use env_logger;
        env_logger::init().unwrap();
    }

    #[test]
    fn many_senders() {
        use std::sync::atomic::{AtomicUsize, Ordering};
        pre();

        let (send, recv) = new();

        let threads = 5;
        let pulses = 50;

        let received = std::sync::Arc::new(AtomicUsize::new(0));

        for _ in 0..threads {
            let recv = recv.clone();
            let received = received.clone();
            std::thread::spawn(move || {
                while let Ok(_) = recv.recv() {
                    received.fetch_add(1, Ordering::AcqRel);
                }
            });
        }


        let mut accum = 0;

        for _ in 0..pulses {
            accum += 1;
            debug!("pulse {}", accum);
            send.send(accum).unwrap();
        }

        std::thread::sleep(std::time::Duration::from_millis(75));
        assert_eq!(received.load(Ordering::Acquire), threads * pulses);
    }
}

Answer 1

I've found your crate looking for existing pub-sub solutions. Time has passed and there is now a good choice. async-broadcast lists a few. Review their comparisons to others. There are some situations to consider, like slow subscriber.

I think your solution will probably work on small scale. I would try to avoid cloning the items to each subscriber though.

However I'm still looking for a simpler solution. I figured a linked list is all we need, actually. Just have to attach items at the end and let the consumers eat the list by dropping references to the head item. This works with Arc or Rc, but then you cannot add that tail. Found a solution for that with arc_swap.

I believe this simple thing can also be made async, just have to figure out the waking. Not sure if AtomicWaker or Event and EventListener...