Communication primitives for concurrent tasks

Rust makes it very difficult to share data among tasks to prevent race conditions and to improve parallelism, but there is often a need for communication between concurrent tasks. The primitives defined in this module are the building blocks for synchronization in rust.

This module provides message-based communication over channels, concretely defined among three types:

• Sender
• SyncSender
• Receiver

A Sender or SyncSender is used to send data to a Receiver. Both senders are clone-able such that many tasks can send simultaneously to one receiver. These channels are task blocking, not thread blocking. This means that if one task is blocked on a channel, other tasks can continue to make progress.

Rust channels come in one of two flavors:

1. An asynchronous, infinitely buffered channel. The channel() function will return a (Sender, Receiver) tuple where all sends will be asynchronous (they never block). The channel conceptually has an infinite buffer.

2. A synchronous, bounded channel. The sync_channel() function will return a (SyncSender, Receiver) tuple where the storage for pending messages is a pre-allocated buffer of a fixed size. All sends will be synchronous by blocking until there is buffer space available. Note that a bound of 0 is allowed, causing the channel to become a "rendezvous" channel where each sender atomically hands off a message to a receiver.

Failure Propagation

In addition to being a core primitive for communicating in rust, channels are the points at which failure is propagated among tasks. Whenever the one half of channel is closed, the other half will have its next operation fail!. The purpose of this is to allow propagation of failure among tasks that are linked to one another via channels.

There are methods on both of senders and receivers to perform their respective operations without failing, however.

Runtime Requirements

The channel types defined in this module generally have very few runtime requirements in order to operate. The major requirement they have is for a local rust Task to be available if any blocking operation is performed.

If a local Task is not available (for example an FFI callback), then the send operation is safe on a Sender (as well as a send_opt) as well as the try_send method on a SyncSender, but no other operations are guaranteed to be safe.

Additionally, channels can interoperate between runtimes. If one task in a program is running on libnative and another is running on libgreen, they can still communicate with one another using channels.

Example

Simple usage:

// Create a simple streaming channel
let (tx, rx) = channel();
spawn(proc() {
    tx.send(10i);
});
assert_eq!(rx.recv(), 10i);

Shared usage:

// Create a shared channel which can be sent along from many tasks
let (tx, rx) = channel();
for i in range(0i, 10i) {
    let tx = tx.clone();
    spawn(proc() {
        tx.send(i);
    })
}

for _ in range(0i, 10i) {
    let j = rx.recv();
    assert!(0 <= j && j < 10);
}

Propagating failure:

// The call to recv() will fail!() because the channel has already hung
// up (or been deallocated)
let (tx, rx) = channel::<int>();
drop(tx);
rx.recv();

Synchronous channels:

let (tx, rx) = sync_channel::<int>(0);
spawn(proc() {
    // This will wait for the parent task to start receiving
    tx.send(53);
});
rx.recv();

Reading from a channel with a timeout requires to use a Timer together with the channel. You can use the select! macro to select either and handle the timeout case. This first example will break out of the loop after 10 seconds no matter what:

use std::io::timer::Timer;

let (tx, rx) = channel::<int>();
let mut timer = Timer::new().unwrap();
let timeout = timer.oneshot(10000);

loop {
    select! {
        val = rx.recv() => println!("Received {}", val),
        () = timeout.recv() => {
            println!("timed out, total time was more than 10 seconds")
            break;
        }
    }
}

This second example is more costly since it allocates a new timer every time a message is received, but it allows you to timeout after the channel has been inactive for 5 seconds:

use std::io::timer::Timer;

let (tx, rx) = channel::<int>();
let mut timer = Timer::new().unwrap();

loop {
    let timeout = timer.oneshot(5000);

    select! {
        val = rx.recv() => println!("Received {}", val),
        () = timeout.recv() => {
            println!("timed out, no message received in 5 seconds")
            break;
        }
    }
}