Fixes

• sync: broadcast closing the channel no longer requires capacity (#2448).
• rt: regression when configuring runtime with max_threads less than the number of CPUs (#2457).

Fixes

• docs: misc improvements (#2400, #2405, #2414, #2420, #2423, #2426, #2427, #2434, #2436, #2440).
• rt: support block_in_place in more contexts (#2409, #2410).
• stream: no panic in merge() and chain() when using size_hint() (#2430).
• task: include visibility modifier when defining a task-local (#2416).

Added

• rt: runtime::Handle::block_on (#2437).
• sync: owned Semaphore permit (#2421).
• tcp: owned split (#2270).

Fixes a regression with LocalSet that allowed !Send futures to cross threads.

This change makes LocalSet !Send. The Send implementation was accidentally added in v0.2.14. Removing the Send implementation is not considered a breaking change as it fixes a soundness bug and the implementation was accidental.

Fixes

• task: LocalSet was incorrectly marked as Send (#2398)
• io: correctly report WriteZero failure in write_int (#2334)

This release fixes a bug in the threaded scheduler that could result in panics under load (see #2382). Additionally, the default number of worker threads now uses the logical CPU count, so it will now respect scheduler affinity and cgroups CPU quotas.

Fixes

• rt: bug in work-stealing queue (#2387)

Changes

• rt: threadpool uses logical CPU count instead of physical by default (#2391)

This release fixes a regression in tokio::sync and a bug in tokio::fs::copy. It also
adds a new APIs to tokio::time and tokio::io.

Fixes

• sync: fix a regression where Mutex, Semaphore, and RwLock futures no
  longer implement Sync (#2375)
• fs: fix fs::copy not copying file permissions (#2354)

Added

• time: added deadline method to delay_queue::Expired (#2300)
• io: added StreamReader (#2052)

Fixes a queue regression and adds a new disarm fn to mpsc::Sender.

Fixes

• rt: fix queue regression (#2362).

Added

• sync: Add disarm to mpsc::Sender (#2358).

This release introduces automatic cooperative task yielding. This article describes the new functionality in more detail.

Fixes

• rt: concurrency bug in scheduler (#2273).
• rt: concurrency bug with shell runtime (#2333).
• test-util: correct pause/resume of time (#2253).
• time: DelayQueue correct wakeup after insert (#2285).

Added

• io: impl RawFd, AsRawHandle for std io types (#2335).
• rt: automatic cooperative task yielding (#2160, #2343, #2349).
• sync: RwLock::into_inner (#2321).

Changed

• sync: semaphore, mutex internals rewritten to avoid allocations (#2325).

Fixes a minor bug in the previous release that resulted in compilation errors using the new pin! form.

Fixes

• macros: unresolved import in pin! (#2281).

Polish, small additions, and fixes. The biggest additions in this release are StreamMap and Notify.

StreamMap

Similar to StreamExt::merge, StreamMap supports merging multiple source streams into a single stream, producing items as they become available in the source streams. However, StreamMap supports inserting and removing streams at run-time. This is useful for cases where a consumer wishes to subscribe to messages from multiple sources and dynamically manage those subscriptions.

As the name implies, StreamMap maps keys to streams. Streams are [inserted] or [removed] as needed and then the StreamMap is used as any other stream. Items are returned with their keys, enabling the caller to identify which source stream the item originated from.

Example

use tokio::stream::{StreamExt, StreamMap};
use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (mut tx1, rx1) = mpsc::channel(10);
    let (mut tx2, rx2) = mpsc::channel(10);
    // use `Sender` handles

    let mut map = StreamMap::new();

    // Insert both streams
    map.insert("one", rx1);
    map.insert("two", rx2);

    // Read twice
    for _ in 0..2 {
        let (key, val) = map.next().await.unwrap();

        println!("got {} from {}", val, key);

        // Remove the stream to prevent reading the next value
        map.remove(key);
    }
}

Notify

Notify is the next step in providing async / await based synchronization primitives. It is similar to how thread::park() / unpark() work, but for asynchronous tasks. Consumers await notifications and producers notify consumers. Notify is intended to be used as a building block for higher level synchronization primitives, such as channels.

Examples

Basic usage.

use tokio::sync::Notify;
use std::sync::Arc;

#[tokio::main]
async fn main() {
    let notify = Arc::new(Notify::new());
    let notify2 = notify.clone();

    tokio::spawn(async move {
        notify2.notified().await;
        println!("received notification");
    });

    println!("sending notification");
    notify.notify();
}

Here is how Notify can be used as a building block for an unbounded channel.

use tokio::sync::Notify;

use std::collections::VecDeque;
use std::sync::Mutex;

struct Channel<T> {
    values: Mutex<VecDeque<T>>,
    notify: Notify,
}

impl<T> Channel<T> {
    pub fn send(&self, value: T) {
        self.values.lock().unwrap()
            .push_back(value);

        // Notify the consumer a value is available
        self.notify.notify();
    }

    pub async fn recv(&self) -> T {
        loop {
            // Drain values
            if let Some(value) = self.values.lock().unwrap().pop_front() {
                return value;
            }

            // Wait for values to be available
            self.notify.notified().await;
        }
    }
}

Changes

Fixes

• net: UnixStream::poll_shutdown should call shutdown(Write) (#2245).
• process: Wake up read and write on EPOLLERR (#2218).
• rt: potential deadlock when using block_in_place and shutting down the
  runtime (#2119).
• rt: only detect number of CPUs if core_threads not specified (#2238).
• sync: reduce watch::Receiver struct size (#2191).
• time: succeed when setting delay of $MAX-1 (#2184).
• time: avoid having to poll DelayQueue after inserting new delay (#2217).

Added

• macros: pin! variant that assigns to identifier and pins (#2274).
• net: impl Stream for Listener types (#2275).
• rt: Runtime::shutdown_timeout waits for runtime to shutdown for specified
  duration (#2186).
• stream: StreamMap merges streams and can insert / remove streams at
  runtime (#2185).
• stream: StreamExt::skip() skips a fixed number of items (#2204).
• stream: StreamExt::skip_while() skips items based on a predicate (#2205).
• sync: Notify provides basic async / await task notification (#2210).
• sync: Mutex::into_inner retrieves guarded data (#2250).
• sync: mpsc::Sender::send_timeout sends, waiting for up to specified duration
  for channel capacity (#2227).
• time: impl Ord and Hash for Instant (#2239).

Introduces select!, join!, and try_join! macros for waiting on multiple async operations concurrently from the same task. These macros are implemented primarily as declarative macros, which works around the recursion limit issue. The select! macro works with any type that implements Future and does not require special FusedFuture traits.

Fixes

• docs: misc fixes and tweaks (#2155, #2103, #2027, #2167, #2175).
• macros: handle generics in #[tokio::main] method (#2177).
• sync: broadcast potential lost notifications (#2135).
• rt: improve "no runtime" panic messages (#2145).

Added

• optional support for using parking_lot internally (#2164).
• fs: fs::copy, an async version of std::fs::copy (#2079).
• macros: select! waits for the first branch to complete (#2152).
• macros: join! waits for all branches to complete (#2158).
• macros: try_join! waits for all branches to complete or the first error (#2169).
• macros: pin! pins a value to the stack (#2163).
• net: ReadHalf::poll() and ReadHalf::poll_peak (#2151)
• stream: StreamExt::timeout() sets a per-item max duration (#2149).
• stream: StreamExt::fold() applies a function, producing a single value. (#2122).
• sync: impl Eq, PartialEq for oneshot::RecvError (#2168).
• task: methods for inspecting the JoinError cause (#2051).