Future/Stream implementation for channel

[rivertam]

I suppose I'm reopening an issue on the archived dedicated channels repo (otherwise I'd just make a comment).

@stjepang mentioned here that it would be easier to reimplement channels from scratch with Future support in mind ahead of time than to add Future support to the extant library.

I have two questions:

1. Is this on any public roadmap? I don't know nearly enough about async at the moment in Rust to valuably contribute, I don't believe.
2. If one were to implement a Stream naiively using a crossbeam channel, would it not be efficient? I see an implementation exists as a benchmark, at least for some usages of the channel.

---

The primary reason I ask this for my own needs is that there appear to be no stable, working MPMC channels that also implement Stream. My application essentially wants a stream of events that occur at a rate of approximately 5 events per millisecond with a couple dozen listeners to that same stream of events. I've created a mini benchmark (not good; no need to share in my opinion but if anyone is curious I can make it public) for using a bunch of futures::sync::mpsc::channels, one for each listener, and publishing on each of those Senders. It seems, given the benchmarks, that that might actually not be terribly inefficient (I can add 100 listeners then publish 10 messages per millisecond for 10 milliseconds in about 12 ms), but I'd rather do things the right way with a Receiver: Clone + Stream, if I can.

[rivertam]

In an attempt to benchmark a naiive conversion to Stream, I ran into a SendError that I could use help debugging:

#[macro_use]
extern crate criterion;
extern crate crossbeam_channel;
extern crate futures;
extern crate tokio;

use criterion::Criterion;
use crossbeam_channel::bounded;
use futures::future::lazy;
use futures::stream::Stream;
use futures::sync::mpsc::{channel, Receiver, Sender};
use futures::{Async, Poll};
use futures::{Future, Sink};
use std::sync::{Arc, Mutex};
use tokio::timer::Interval;

struct Handler {
    total: Arc<Mutex<u32>>,
}

impl Handler {
    pub fn add_to(&self, num: u32) {
        let mut total = self.total.lock().unwrap();
        *total += num;
    }
}

#[derive(Clone)]
struct FutureReceiver<T> {
    receiver: crossbeam_channel::Receiver<T>,
}

impl<T> Stream for FutureReceiver<T> {
    type Item = T;
    type Error = crossbeam_channel::TryRecvError;

    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
        match self.receiver.try_recv() {
            Ok(i) => Ok(Async::Ready(Some(i))),
            Err(crossbeam_channel::TryRecvError::Empty) => Ok(Async::NotReady),
            Err(e) => Err(e),
        }
    }
}

fn go_crossbeam() {
    let handler = Arc::new(Handler {
        total: Arc::new(Mutex::new(0)),
    });

    let handler_clone = handler.clone();
    tokio::run(lazy(move || {
        let (tx, rx) = bounded(1_024);
        let future_rx = FutureReceiver { receiver: rx };
        for _ in 0..100 {
            let handler = handler_clone.clone();
            let rx = future_rx.clone();
            tokio::spawn(
                rx.for_each(move |num| -> Result<(), crossbeam_channel::TryRecvError> {
                    handler.add_to(num);
                    Ok(())
                })
                .map_err(|_| ()),
            );
        }

        let tx = Arc::new(tx);
        tokio::spawn(
            Interval::new(
                std::time::Instant::now(),
                std::time::Duration::from_micros(100),
            )
            .take(100)
            .map(|_| 100)
            .map_err(|e| {
                eprintln!("Got interval error = {:?}", e);
            })
            .for_each(move |num| tx.send(num).map_err(|e| {
                println!("send err = {:?}", e));
                match e {
                    SendError
                }
            })
            .map(|_| ()),
        );

        Ok(())
    }));
}

I'm getting a SendError on the tx.send(num) line, which doesn't make a lot of sense to me as I don't believe the Receivers should ever get dropped. Unrelated to the major issue, but would love an explanation.

Is this on any public roadmap?

Yes, there's a prototype of futures-based channels that take inspiration from crossbeam-channel, but it's in a private repo. The prototype will be completed, though.

If one were to implement a Stream naiively using a crossbeam channel, would it not be efficient? I see an implementation exists as a benchmark, at least for some usages of the channel.

That's a benchmark for the futures-channel crate, which has almost nothing in common with crossbeam-channel.

It should be efficient, but this naive implementation is not complete because we don't wake up sleeping receiver tasks when a message is sent into the channel.

In an attempt to benchmark a naiive conversion to Stream, I ran into a SendError that I could use help debugging:

The problem is that in poll(), if the receiver is not ready, we're not registering the task anywhere for notification. Since the current task is not moved anywhere else, the only reference to it is dropped and this way all receivers get dropped, thus disconnecting the channel.

[rivertam]

@stjepang Thanks! This explains a lot not just about the issue at hand, but about how Tokio and Futures work in Rust!

In the meantime, I would appreciate some quick advice: Do you think the multiple futures::sync::mpsc implementation with a bunch of Receivers as the "subscribers" being distributed out and the relevant Senders as the "publishers" being kept in a data structure such as a Vec is a valid approach? It's kind of the only way I can think of efficiently and relatively easy writing an event emitter (in this case, the event emitter is not generic and the production occurs on one thread).

Yes, that sounds like a good idea!

I'm curious: when sending an event message, do you pick one of the Senders randomly, or do you send a clone of the event to each Sender, or something else?

[rivertam]

In this case, each event is actually a type that is typically encoded by 8 bytes (it's a CAN v1 message), so right now I'm envisioning Clone but I think even Copy would be fair.

However, as the messages are const across all time, I think a reference approach would work if they were bigger.

So you're actually looking for a broadcast channel (like the bus crate)? A typical MPMC channel delivers each message to only one sender.

[rivertam]

Ah, yes, I am. Sorry, I forgot the terminologies were different and they meant different things. I hadn't seen the bus crate, but as it doesn't integrate with Futures, I don't know that I'll be able to use it efficiently.

I believe we can close this. A future-based version of crossbeam-channel is on my todo list.

[najamelan]

@stjpang Has any progress been made for a high perf async channel?

[cynecx]

@najamelan There are other implementations, which you can already use, like the one included in tokio: https://github.com/tokio-rs/tokio/tree/master/tokio-sync (Though the one included in tokio-sync is 'just' mpsc not mpmc.)

[najamelan]

@cynecx I didn't know about tokio-sync. I'll check it out soon! Thanks for pointing me to it.

[najamelan]

A new crate just got published last night: https://crates.io/crates/channel-async

[jnicholls]

That new crate is a nice, thin layer on top of crossbeam-channel if you ask me. It was almost perfect until I saw the errors use failure instead of just std::error::Error or something like snafu :)

[vulpyne]

@jnicholls It seems like channel-async just runs a loop polling try_recv or try_send in the underlying channel and sleeping if the call fails. That will chew CPU in each "blocked" thread if the sleep delay is low or introduce latency if it's higher.