tokio-rs/tokio tokio-0.2.5

Tokio v0.2.5

on GitHub

Includes new APIs, utilities, and fixes. Some highlights:

tokio::sync::broadcast

A multi-producer, multi-consumer channel where each sent value is sent to all consumers (fan-out). The channel is bounded and when consumers lag, they will receive an error indicating they have lagged too far behind.

use tokio::sync::broadcast;

#[tokio::main]
async fn main() {
    let (tx, mut rx1) = broadcast::channel(16);
    let mut rx2 = tx.subscribe();

    tokio::spawn(async move {
        assert_eq!(rx1.recv().await.unwrap(), 10);
        assert_eq!(rx1.recv().await.unwrap(), 20);
    });

    tokio::spawn(async move {
        assert_eq!(rx2.recv().await.unwrap(), 10);
        assert_eq!(rx2.recv().await.unwrap(), 20);
    });

    tx.send(10).unwrap();
    tx.send(20).unwrap();
}

Senders never block. When the channel is full, the oldest value still held by the channel is overwritten. This tends to be the desired behavior in order to prevent slow consumers from blocking the entire system. However, you can use a Semaphore (also added in this release) to ensure that all consumers see all messages.

tokio::sync::Semaphore

A counting synchronization primitive. It is used to limit a critical section to 1 or more concurrent tasks. For example, assume we wish to limit the number of in-flight database queries, we could do something like:

struct MyDbClient {
    db: MyDbHandle,
    semaphore: tokio::sync:Semaphore,
}

async fn query(client: &MyDbClient, query: Query) -> QueryResult {
     let _permit = client.semaphore.acquire().await;
     client.db.query(query).await
}

There may be any number of concurrent calls to query, but the semaphore will limit the number that are able to concurrently perform the query.

Added

• io::AsyncSeek trait (#1924).
• Mutex::try_lock (#1939)
• mpsc::Receiver::try_recv and mpsc::UnboundedReceiver::try_recv (#1939).
• writev support for TcpStream (#1956).
• time::throttle for throttling streams (#1949).
• implement Stream for time::DelayQueue (#1975).
• sync::broadcast provides a fan-out channel (#1943).
• sync::Semaphore provides an async semaphore (#1973).
• stream::StreamExt provides stream utilities (#1962).

Fixes

• deadlock risk while shutting down the runtime (#1972).
• panic while shutting down the runtime (#1978).
• sync::MutexGuard debug output (#1961).
• misc doc improvements (#1933, #1934, #1940, #1942).

Changes

• runtime threads are configured with runtime::Builder::core_threads and
  runtime::Builder::max_threads. runtime::Builder::num_threads is
  deprecated (#1977).