#channel #broadcast #lock-free

bin+lib bus

A lock-free, bounded, single-producer, multi-consumer, broadcast channel

17 stable releases

✓ Uses Rust 2018 edition

2.2.1 Mar 8, 2019
2.0.1 Dec 4, 2018
2.0.0 May 6, 2018
1.4.1 Jan 10, 2018
0.2.1 Jun 3, 2016

#18 in Concurrency

Download history 98/week @ 2019-01-20 159/week @ 2019-01-27 141/week @ 2019-02-03 128/week @ 2019-02-10 122/week @ 2019-02-17 208/week @ 2019-02-24 267/week @ 2019-03-03 335/week @ 2019-03-10 436/week @ 2019-03-17 693/week @ 2019-03-24 620/week @ 2019-03-31 380/week @ 2019-04-07 439/week @ 2019-04-14 489/week @ 2019-04-21 312/week @ 2019-04-28

1,076 downloads per month
Used in 9 crates

MIT license

41KB
458 lines

bus

Crates.io Documentation Build Status Codecov

Bus provides a lock-free, bounded, single-producer, multi-consumer, broadcast channel.

It uses a circular buffer and atomic instructions to implement a lock-free single-producer, multi-consumer channel. The interface is similar to that of the std::sync::mpsc channels, except that multiple consumers (readers of the channel) can be produced, whereas only a single sender can exist. Furthermore, in contrast to most multi-consumer FIFO queues, bus is broadcast; every send goes to every consumer.

I haven't seen this particular implementation in literature (some extra bookkeeping is necessary to allow multiple consumers), but a lot of related reading can be found in Ross Bencina's blog post "Some notes on lock-free and wait-free algorithms".

Bus achieves broadcast by cloning the element in question, which is why T must implement Clone. However, Bus is clever about only cloning when necessary. Specifically, the last consumer to see a given value will move it instead of cloning, which means no cloning is happening for the single-consumer case. For cases where cloning is expensive, Arc should be used instead.

In a single-producer, single-consumer setup, Bus is ~7x faster then std::sync::mpsc::sync_channel on my machine. It's ~2x slower than crossbeam-channel. YMMV. You can check your performance on Nightly using

$ cargo bench --features bench

To see multi-consumer results, run the benchmark utility instead (should work on stable too)

$ cargo build --bin bench --release
$ target/release/bench

Examples

Single-send, multi-consumer example

use bus::Bus;
let mut bus = Bus::new(10);
let mut rx1 = bus.add_rx();
let mut rx2 = bus.add_rx();

bus.broadcast("Hello");
assert_eq!(rx1.recv(), Ok("Hello"));
assert_eq!(rx2.recv(), Ok("Hello"));

Multi-send, multi-consumer example

use bus::Bus;
use std::thread;

let mut bus = Bus::new(10);
let mut rx1 = bus.add_rx();
let mut rx2 = bus.add_rx();

// start a thread that sends 1..100
let j = thread::spawn(move || {
    for i in 1..100 {
        bus.broadcast(i);
    }
});

// every value should be received by both receivers
for i in 1..100 {
    // rx1
    assert_eq!(rx1.recv(), Ok(i));
    // and rx2
    assert_eq!(rx2.recv(), Ok(i));
}

j.join().unwrap();

Many-to-many channel using a dispatcher

use bus::Bus;

use std::thread;
use std::sync::mpsc;

// set up fan-in
let (tx1, mix_rx) = mpsc::sync_channel(100);
let tx2 = tx1.clone();
// set up fan-out
let mut mix_tx = Bus::new(100);
let mut rx1 = mix_tx.add_rx();
let mut rx2 = mix_tx.add_rx();
// start dispatcher
thread::spawn(move || {
    for m in mix_rx.iter() {
        mix_tx.broadcast(m);
    }
});

// sends on tx1 are received ...
tx1.send("Hello").unwrap();

// ... by both receiver rx1 ...
assert_eq!(rx1.recv(), Ok("Hello"));
// ... and receiver rx2
assert_eq!(rx2.recv(), Ok("Hello"));

// same with sends on tx2
tx2.send("world").unwrap();
assert_eq!(rx1.recv(), Ok("world"));
assert_eq!(rx2.recv(), Ok("world"));

Dependencies

~1.5MB
~23K SLoC