Crate takeaway

Expand description

High-performance task queues.

This is a Rust library providing a high-performance, work-stealing task queue data structure. Consider using it when:

There may be thousands of tasks ready to run at any time.
Tasks are relatively small, synchronous, and CPU-bound.
All available CPUs should be used to maximum capacity.
(Optional) Some tasks have a higher priority than others.
(Optional, coming soon) Grouping certain tasks lets them be executed efficiently.

§Usage

Most of the complexity in setting up takeaway is inherent to using any multi-threaded task architecture. If you’re transitioning from a different task queue system into takeaway, you’ll notice you’ve already done most of the setup described below.

First, you need to define a task type, to represent the tasks being executed. This type must implement the Task trait.

/// The task type.
struct MyTask;

impl takeaway::Task for MyTask {
    // These will be automatic defaults in the future.
    type Priority = ();
    fn priority(&self) -> Self::Priority {}
}

At program startup, you need to initialize a Queue. The easiest way to do so is use Config, which provides a builder pattern API. Start with Config::default() (or Config::new() without the std feature), set any required parameters, and end with Config::build().

During configuration, a number of workers is selected (by default, it will be the estimated from the system resources). All of these workers must be initialized.

let queue = takeaway::Config::default()
//  .set_batch_size(64.try_into().unwrap())
//  .set_oneshot(false)
    .build();
let num_workers = queue.config().num_workers();

Then, start up all your worker threads. takeaway provides a convenient async API, so the main worker thread code should be async. To run an async fn on a new thread, you need to wrap it in an async runtime. If you’re going to write your own async code, you can use a fully-fledged executor like tokio; otherwise util::block_on() is sufficient.

Within the main body of each worker thread, create a Worker. Note that you need a unique index for every worker, which you can use to identify the thread as a whole. Then, you can enqueue any initial tasks you have using Worker::enqueue(). The main loop is very simple: call Worker::next() and execute the returned task.

// Spawn the worker threads.
std::thread::scope(|s| {
    for id in 0..num_workers.get() {
        let queue = &queue;
        s.spawn(move || block_on(worker(queue, id)));
    }
});

// The body of each worker thread.
async fn worker(queue: &Queue<MyTask>, id: usize) {
    // Set up the thread-local queue.
    let mut worker = Worker::new(queue, id);

    // Seed the worker with initial tasks.
    worker.enqueue_one(MyTask);

    // The main loop.
    while let Some(task) = worker.next().await {
        // Inspect and execute the task.
        let MyTask = task;
        //...

        // Optionally, spawn new tasks here.
        //worker.enqueue(...);
    }
}

While this program works well enough, there’s only problem: it won’t terminate. By default, takeaway assumes there are infinite tasks, so Worker::next() will block until a task is available. This is perfect if your program operates like a job server. takeaway also supports a one-shot mode, where it will automatically shut down when it runs out of tasks; this can be enabled with Config::with_oneshot(). In any case, a manual shutdown can be initiated at any time via Queue::shutdown().

That’s it! You now have a complete, up-and-running system using takeaway.

See the examples directory for some complex programs using takeaway to distribute tasks across threads.

§Prioritization

By default, takeaway will execute tasks in any order. This is perfectly fine if every task is equal. Most of the time, however, some tasks take longer to execute than others. Some tasks might lead to many new sub-tasks being enqueued. In either case, finding and executing these tasks before others can improve the overall runtime of your program.

takeaway is capable of ordering tasks by a user-defined priority metric, so that (on a best-effort basis) higher-priority tasks are executed before lower-priority ones. Don’t assume that tasks will be executed in order of descending priority, but if you have especially uneven tasks in your system, task prioritization will probably yield a performance improvement.

To make use of this, pick a suitable priority type in the implementation of Task. This is usually a simple non-zero integer type like NonZeroU32, but you can implement TaskPriority for a custom type. Tasks with greater priority values will be executed first.

// A task type involving prioritization.
struct MyTask {
    /// The priority of this task.
    priority: NonZeroU32,
}

impl takeaway::Task for MyTask {
    type Priority = NonZeroU32;

    fn priority(&self) -> Self::Priority {
        self.priority
    }
}

That’s all. takeaway will now sort enqueued tasks by priority and steal higher-priority tasks from other threads.

§Crate Details

takeaway is somewhat top-heavy; its complexity and implementation effort comes from its own codebase rather than from its dependencies. It tries to be pretty minimal, and adds up to a few thousand lines of code in total; it won’t burden your compilation times or crate dependency graph.

takeaway may be no_std compatible, but it relies on non-portable OS functionality. It can only be used on Linux, FreeBSD, Windows, or macOS.

§Feature Flags

takeaway has the following feature flags:

std (default): Depends on the standard library in order to add useful trait implementations, particularly relating to thread wakers. Use this if you’re writing std-dependent code.
crossbeam-utils: Enables a dependency on the crossbeam-utils crate instead of using a vendored copy of the relevant code. Use this if you already have crossbeam-utils in your dependency graph.

§Dependencies

takeaway depends on the following crates. The Cargo manifest documents why these crates are necessary, alternative solutions to those crates, and how these crates are maintained.

atomic_wait provides a simple, portable, and efficient way to block a thread while waiting for an atomic variable to change. This is necessary because takeaway can block the thread while loading tasks in some very rare circumstances.
Task priorities are shared between worker threads atomically. atomig provides generics over atomic types, so that user-selected task priority types can be operated on atomically.
crossbeam_utils is used for a few concurrency-related utility types. By default, it is vendored in (i.e. the relevant source code has been copied into takeaway and the crate is not depended on).

§Implementation

Internally, a Worker operates on batches of tasks. It will maintain a single batch at any time, consisting of the highest-priority tasks it has, and will gradually drain it as the user requests tasks. When the batch is sufficiently depleted, it will be refreshed. This is a fairly expensive process, which is why a larger batch size amortizes the runtime overhead of using takeaway.

Workers divide their tasks into three lists: the local queue, the public queue, and the postponed queue. The current batch of tasks are split (evenly) between the local and public queues, while the remaining tasks are left in the postponed queue. The public queue is exposed to all other workers, for them to steal at any time. The batch is considered to be depleted once the maximum number of tasks have been read from it, or when the public queue is stolen.

Care is taken to distribute high-priority tasks among workers efficiently. During a refresh, the current batch is sorted by priority and every other task within it is moved to the local queue. This way, the local and public queues have approximately the same distribution of tasks by priority. In the worst case, where a single worker has all the highest-priority tasks, the tasks will be propagated in a binary tree fashion; a worker will steal half the high-priority tasks, both the thief and the victim will refresh their batches and publish new public queues, and the process will repeat.

When refreshing a batch, the following steps are taken:

The worker’s tasks are coalesced together into a single list, from the local, public (if not stolen), and the postponed queues.
If some other worker has published a public queue containing tasks of a higher priority than the local set, the public queue is stolen and its tasks are merged into the local list.
The tasks are sorted by priority, and the highest-priority tasks form the new batch. The tasks are then divided into the postponed, local, and public queues.
If the public queue contains at least one task, information about it will be published in the global Queue state, so that other workers can see and steal it. If some workers are sleeping due to a lack of tasks, one of them is woken up so it can try stealing the public queue.
If no tasks were available at all, the worker is put to sleep, and will mark itself as such in the global state so it can be woken by others. It could also be woken up when tasks are enqueued locally or globally.

Modules§

util: Utility functionality.

Structs§

Config: Configuration for a Queue.
Enqueuer: A handle to enqueue tasks for a Worker.
Queue: Shared state for a task queue.
Worker: A thread-local view of the task queue.

Traits§

Task: A task.
TaskPriority: The priority of a task.

Crate takeaway

Crate takeaway Copy item path

§Usage

§Prioritization

§Crate Details

§Feature Flags

§Dependencies

§Implementation

Modules§

Structs§

Traits§

Crate takeaway