Crate unicycle

A scheduler for driving a large number of futures.

Unicycle provides a collection of Unordered types:

• FuturesUnordered
• StreamsUnordered
• IndexedStreamsUnordered

These are async abstractions that runs a set of futures or streams which may complete in any order. Similarly to FuturesUnordered from the futures crate. But we aim to provide a stronger guarantee of fairness (see below), and better memory locality for the futures being pollled.

Note: This project is experimental. It involves some amount of unsafe and possibly bad assumptions which needs to be either vetted or removed before you should consider putting it in production.

Features

• parking-lot - To enable locking using the parking_lot crate (default).
• futures-rs - Enable the used of the Stream type from futures-rs. This is required to get access to StreamsUnordered and IndexedStreamsUnordered since these wrap over futures-rs types. (default)

Examples

use tokio::time;
use std::time::Duration;
use unicycle::FuturesUnordered;

#[tokio::main]
async fn main() {
    let mut futures = FuturesUnordered::new();

    futures.push(time::delay_for(Duration::from_secs(2)));
    futures.push(time::delay_for(Duration::from_secs(3)));
    futures.push(time::delay_for(Duration::from_secs(1)));

    while let Some(_) = futures.next().await {
        println!("tick");
    }

    println!("done!");
}

Unordered types can be created from iterators:

use tokio::time;
use std::time::Duration;
use unicycle::FuturesUnordered;

#[tokio::main]
async fn main() {
    let mut futures = Vec::new();

    futures.push(time::delay_for(Duration::from_secs(2)));
    futures.push(time::delay_for(Duration::from_secs(3)));
    futures.push(time::delay_for(Duration::from_secs(1)));

    let mut futures = futures.into_iter().collect::<FuturesUnordered<_>>();

    while let Some(_) = futures.next().await {
        println!("tick");
    }

    println!("done!");
}

Fairness

You can think of abstractions like Unicycle as schedulers. They are provided a set of child tasks, and try to do their best to drive them to completion. In this regard, it's interesting to talk about fairness in how the tasks are being driven.

The current implementation of FuturesUnordered maintains a queue of tasks interested in waking up. As a task is woken up, it's added to the head of this queue to signal its interest. When FuturesUnordered is being polled, it drains this queue in a loop and polls the associated task. This process has a side effect of tasks who aggressively signal interest in waking up will receive priority and be polled more frequently, since there is a higher chance that while the queue is being drained, their interest will be re-added to the queue. This can lead to instances where a small number of tasks can can cause the polling loop of FuturesUnordered to spin abnormally. This issue was reported by Jon Gjengset, and improved on by limiting the amount FuturesUnordered is allowed to spin.

Unicycle addresses this by limiting how frequently a child task may be polled per polling cycle. This is done by tracking polling interest in two separate sets. Once we are polled, we swap out the active set, then take the swapped out set and use as a basis for what to poll in order, but we limit ourselves to only poll once per child task. Additional wakeups are only registered in the swapped in set which will be polled the next cycle.

This way we hope to achieve a higher degree of fairness, never favoring the behavior of one particular task.

Architecture

The Unordered type stores all futures being polled in a PinSlab (Inspired by the slab crate). A slab is capable of utomatically reclaiming storage at low cost, and will maintain decent memory locality. A PinSlab is different from a Slab in how it allocates the memory regions it uses to store objects. While a regular Slab is simply backed by a vector which grows as appropriate, this approach is not viable for pinning, since it would cause the objects to move while being reallocated. Instead PinSlab maintains a growable collection of fixed-size memory regions, allowing it to store and reference immovable objects through the pin API. Each future inserted into the slab is assigned an index, which we will be using below. We now call the inserted future a task, and you can think of this index as a unique task identifier.

Next to the slab we maintain two bit sets, one active and one alternate. When a task registers interest in waking up, the bit associated with its index is set in the active set, and the latest waker passed into Unordered is called to wake it up. Once Unordered is polled, it atomically swaps the active and alternate bit sets, waits until it has exclusive access to the now alternate BitSet, and drains it from all the indexes which have been flagged to determine which tasks to poll. Each task is then polled once in order. If the task is Ready, its result is yielded. After we receive control again, we continue draining the alternate set in this manner, until it is empty. When this is done we yield once, then we start the cycle over again.

Modules

pin_slab

A slab-like, pre-allocated storage where the slab is divided into immovable slots. Each allocated slot doubles the capacity of the slab.

Structs

Futures	Sentinel type for futures.
IndexedStreams	Sentinel type for streams which are indexed - for each value they yield, they also yield the task identifier associated with them.
Streams	Sentinel type for streams.
Unordered	A container for an unordered collection of Futures or Streams.

Traits

PollNext	Trait for providing a poll_next implementation for various unordered set types.
Sentinel	Trait implemented by sentinels for the Unordered type.

Type Definitions

FuturesUnordered	A container for an unordered collection of Futures.
IndexedStreamsUnordered	A container for an unordered collection of Streams, which also yields the index that produced the next item.
StreamsUnordered	A container for an unordered collection of Streams.