Module ergo_sync::ch

Module for working with channels. Rexport of crossbeam_channel

Examples

Several of these examples are copies of the chan and crossbeam_channel crates.

Example: unbounded (async) channel

#[macro_use] extern crate ergo_sync;
use ergo_sync::*;

let (tx, rx) = ch::unbounded();

// Can send an arbitrarily large number of messages.
for i in 0..1000 {
    ch!(tx <- i);
}

Example: bounded (sync) channel

#[macro_use] extern crate ergo_sync;
use ergo_sync::*;

// Create a channel that can hold at most 5 messages at a time.
let (tx, rx) = ch::bounded(5);

// Can send only 5 messages.
for i in 0..5 {
    ch!(tx <- i);
}

// An attempt to send one more message will fail.
assert!(tx.try_send(5).is_err());

Example: rendevous channel

#[macro_use] extern crate ergo_sync;
use ergo_sync::*;

let (send, recv) = ch::bounded(0);
spawn(move || ch!(send <- 5));
assert_eq!(ch!(<- recv), 5); // blocks until the previous send occurs

Example: the sentinel channel idiom

When writing concurrent programs with ergo, you will often find that you need to somehow "wait" until some operation is done. For example, let's say you want to run a function in a separate thread, but wait until it completes. Here's one way to do it:

#[macro_use] extern crate ergo_sync;
use ergo_sync::*;

fn do_work(done: ch::Sender<()>) {
    // do something

    // signal that we're done.
    ch!(done <- ());
}

fn main() {
    let (sdone, rdone) = ch::bounded(0);
    spawn(move || do_work(sdone));
    // block until work is done, and then quit the program.
    ch!(<- rdone);
}

In effect, we've created a new channel that sends unit values. When we're done doing work, we send a unit value and main waits for it to be delivered.

Another way of achieving the same thing is to simply close the channel. Once the channel is closed, any previously blocked receive operations become immediately unblocked. What's even cooler is that channels are closed automatically when all senders are dropped. So the new program looks something like this:

#[macro_use] extern crate ergo_sync;
use ergo_sync::*;

fn do_work(_done: ch::Sender<()>) {
    // do something
}

fn main() {
    let (sdone, rdone) = ch::bounded(0);
    spawn(move || do_work(sdone));
    // Block until the channel is closed.
    //
    // Note: this _expects_ the error that
    // all senders have been dropped and will
    // panic if a value is sent instead.
    ch!(! <- rdone);
}

We no longer need to explicitly do anything with the _done channel. We give do_work ownership of the channel, but as soon as the function stops executing, _done is dropped, the channel is closed and rdone.recv() unblocks returning an error, which we expect with ch!(! <- rdone).

Example: non-blocking sends/receives

#[macro_use] extern crate ergo_sync;
use ergo_sync::*;

let (send, recv) = ch::bounded(1);
let data = "send data".to_string();
match ch!(send <-? data) {
    Some(data) => {
        println!("didn't send data, but got it back: {}", data);
        unreachable!(); // in this case we don't expect it
    }
    None => println!("message sent successfully"),
}

// attempting to send additional data fails
let data = "more data".to_string();
assert_eq!(Some(data.clone()), ch!(send <-? data));

match ch!(<-? recv) {
    Some(data) => println!("received data: {}", data),
    None => {
        println!("didn't receive any data yet");
        unreachable!(); // in this case we don't expect it
    }
}

Example: using select_loop

#[macro_use] extern crate ergo_sync;
use ergo_sync::*;

let (tx1, rx1) = ch::unbounded();
let (tx2, rx2) = ch::unbounded();

spawn(move || ch!(tx1 <- "foo"));
spawn(move || ch!(tx2 <- "bar"));

select_loop! {
    recv(rx1, msg) => {
        println!("Received a message from the first channel: {}", msg);
    }
    recv(rx2, msg) => {
        println!("Received a message from the second channel: {}", msg);
    }
}

Structs

IntoIter	An owning iterator that waits for messages until the channel is disconnected.
Iter	An iterator that waits for messages until the channel is disconnected.
Receiver	The receiving half of a channel.
RecvError	An error returned from the Receiver::recv method.
Select	The dynamic selection interface.
SelectRecvError	An error returned from the Select::recv method.
SelectSendError	An error returned from the Select::send method.
SendError	An error returned from the Sender::send method.
Sender	The sending half of a channel.
TryIter	An iterator that receives messages until the channel is empty or disconnected.

Enums

RecvTimeoutError	This enumeration is the list of possible errors that made recv_timeout unable to return data when called. This can occur with both bounded and unbounded channels.
SendTimeoutError	This enumeration is the list of possible errors that made send_timeout unable to return data when called. This can occur with bounded channels only.
TryRecvError	This enumeration is the list of the possible reasons that try_recv could not return data when called. This can occur with both bounded and unbounded channels.
TrySendError	This enumeration is the list of the possible error outcomes for the try_send method.

Functions

bounded	Creates a new channel of bounded capacity, returning the sender/receiver halves.
unbounded	Creates a new channel of unbounded capacity, returning the sender/receiver halves.