#2.18. task - M:N scheduler task API

The rok::core::task module provides a cooperative task API running on the M:N scheduler. Inspired by Rust Tokio and Go goroutines, tasks are implemented as stackless coroutines on a work-stealing scheduler.

Because use rok::core::**; applies implicitly, task::spawn, task::race, Task<T> and the rest are usable without any import; the module prefix task:: is still kept. Where code should be explicitly marked as asynchronous, use rok::core::task::*; is recommended, to make the asynchronous flow visibly separate.

#2.18.1. Core API

The signatures of the core API:

fn block_on<F: Future>(f: F) -> F::Output;              // sync entry point - attaches to the executor and runs f to completion
fn spawn<F: Future + Send>(f: F) -> Task<F::Output> where F::Output: Send;
fn spawn_local<F: Future>(f: F) -> Task<F::Output>;     // non-Send future - pinned to the current worker
fn spawn_blocking<T: Send>(f: fn() -> T + Send) -> Blocking<T>;   // blocking work on an OS thread outside the executor
fn workers() -> usize;                                  // pool size inside the executor, 1 outside
fn cpu_count() -> usize;                                // number of online CPUs - the pool size
async fn race<T>(tasks: List<Task<T>>) -> T;            // result of the first task to complete
fn select<F: Future>(futs: List<F>) -> SelectFuture<F>; // await - (index, result), does not spawn
fn id() -> TaskId;                                      // id of the current task - panics outside a task
fn try_id() -> TaskId | none;
async fn join<T>(tasks: List<Task<T>>) -> List<T>;      // after all complete
async fn timeout<T>(dur: Duration, task: Task<T>) throws (TimeoutError) -> T;   // throws in try await
async fn sleep(dur: Duration);
async fn sleep_until(when: Instant);
async fn yield_now();                                   // await - yields once to other ready tasks
fn registrations() -> usize;                            // diagnostic - parked wakers + pending timers of the running pool (implementation-defined)
fn readable(fd: i32) -> IoReady;                       // await - until fd becomes readable (refs-lang 7.7.2)
fn writable(fd: i32) -> IoReady;
fn wake_io(fd: i32);                                   // wakes the readable / writable waiters of fd (used after close / shutdown; low-level)

sleep and timeout measure their duration from the first poll.

#Example

use rok::core::task::*;

let t = task::spawn(async { compute_heavy() });
let result = await t;

let results = await task::join([
    task::spawn(async { fetch("a") }),
    task::spawn(async { fetch("b") }),
    task::spawn(async { fetch("c") }),
]);

let first = await task::race([
    task::spawn(async { source_a() }),
    task::spawn(async { source_b() }),
]);

await task::sleep(Duration::ms(100));
await task::yield_now();

spawn takes a future, not a closure, so it is passed an async block or the result of an async fn call, as in task::spawn(async { ... }). A future that is not Send runs on the current thread through spawn_local. join / race / timeout are awaited to obtain their results.

#spawn_blocking - blocking work

struct Blocking<T> { /* result slot */ }
impl<T> Future for Blocking<T> { type Output = T; ... }

spawn_blocking(f) runs f on an OS thread of the blocking pool rather than on an executor worker, and returns a future that completes with its result. It is used to wait inside a task on blocking syscalls that cannot be awaited by readiness (regular file I/O, etc., stdlib 3.1.4a).

• f starts immediately on the call, not at the first poll (if the pool is full, it is queued first).
• There is one blocking pool per process. It adds threads as needed (up to a limit) and ends threads that stay idle for a while. Jobs are processed in arrival order (FIFO). The limit is 512 threads; once reached, jobs queue up. A thread idle for 10 seconds ends. Each job is handed to one idle thread.
• Even if the awaiting task is cancelled, f runs to completion; only its result is discarded.

let n = await task::spawn_blocking(|| heavy_checksum(path));

#Send check for spawn

task::spawn requires the result type to be Send as well (where F::Output: Send), and rejects a future whose frame is not Send (2.13.9). This holds wherever the future came from: not only async blocks and async fn calls but also parameters, fields of structs / tuples / sums, results of helper functions, list elements, futures held by closures, and futures inside other coroutine frames are checked. A future's frame consists of:

• captured values;
• local variables and temporaries live across an await;
• other futures the frame holds (their frames too, recursively).

let c = Cell::new(0);                      // Cell: !Send
task::spawn(async { c.set(1); });          // ERROR - !Send capture
task::spawn_local(async { c.set(1); });    // OK - pinned to the current worker

fn make() -> AsyncFuture<()> { let c = Cell::new(0); async { c.set(1); } }
task::spawn(make());                       // ERROR - the frame of a future made by a helper is checked too

The same check applies wherever a closure is sent to another thread: thread::spawn, thread::Builder::spawn, par's Scope::spawn, and task::spawn_blocking. If the closure holds a future whose frame is not Send, it is the error `X` would move a future to another thread: ....

The check may approximate conservatively. An implementation may reject a future that is actually Send, but must never accept a future that passes a non-Send value to another thread. The check does not descend into:

• values whose type cannot hold a function value (they cannot hide a future's frame);
• values of types with an explicit unsafe impl Send (e.g. Task<T>); that type is responsible for safety.

AsyncFuture<T>, the result type of an async fn, is Send when T: Send; its throwing counterpart TryAsyncFuture<T, E> is Send when both T and E are Send. Captures and frames are examined separately by the check above. A frame whose captures are Send but not Sync (a list an async fn takes, a list or a future an async block captures) is accepted: task::spawn hands the future over like a channel send (2.13.9). The task gets a frame of its own holding copies of what the future captured, made in one hand-off, so what the captures shared (one list twice, a list and a future over it) they still share; using the future or what it captured after the spawn is a compile error (E202), as after a send. Only a future nothing has polled can be handed over: handing over one that has started (polled once by select, a user's Future, or by hand) or finished, or whose frame or a captured future's frame has a [ref x] capture, panics, and nothing is shared. A coroutine future sent through a channel, put in a Mutex or a Frozen is handed over the same way, a frame of its own each time. Each await of a task whose Task was handed on (Frozen::get, a channel, a lock, a par item) gets its own copy of the result; copies of a Task on one thread share it.

#select - the first future to finish

fn select<F: Future>(futs: List<F>) -> SelectFuture<F>;   // Output = (usize, F::Output)

let (i, v) = await task::select([fetch_a(), fetch_b()]);

Unlike race, it does not spawn the futures; it runs them inside the waiting task.

• Futures are polled in list order; if two or more finish in the same poll, the lower index wins.
• When one finishes, all the others that have already started are cancelled. They are polled once more in the cancelled state so that finally / using run (2.18.4). A future never polled is never run.
• All wakeup registrations (timers, parking) made by the losing futures are released.
• If the waiting task is cancelled, the child futures are cancelled too.
• An empty list panics.
• The futures must have a single type. If the result types differ, wrap them: async { Msg::A(await a) }.

A timeout can also be built from select and sleep.

enum Ev { Line(str), Timeout }

let (_, ev) = await task::select([
    async { Ev::Line(await read_line()) },
    async { await task::sleep(Duration::sec(5)); Ev::Timeout },
]);

#TaskId

task::id() is the id of the running task; calling it outside a task panics (try_id() returns none). Task::id() is the id of that task. Ids come from a process-wide counter and are never reused.

#2.18.2. Task<T> type

The Task<T> type is defined as follows.

struct Task<T> {
    // internal: coroutine frame on the GC heap + result slot
}

impl<T> Task<T> {
    fn cancel(self);                      // requests cancellation - ends at the next await (2.18.4)
    fn is_cancelled(self) -> bool;
    fn is_finished(self) -> bool;
    fn join(self) -> Join<T>;             // try await t.join() - CancellationError if cancelled without a result
    fn id(self) -> TaskId;
}

struct TaskId { /* u64 */ }
impl TaskId {
    fn as_u64(self) -> u64;
}
// TaskId: PartialEq / Eq / Hash / Display / Debug

struct CancellationError {}

impl<T> Future for Task<T> {              // `await t` - suspend until completion
    type Output = T;
    ...
}

#IntoFuture integration

Task<T> implements Future<Output = T>, so it can be awaited with await t, and it is covered by the IntoFuture blanket impl of 2.20 future, so it is awaited the same way as Thread. The authoritative definitions of the IntoFuture trait and the lang slot future_trait are in 2.20 future; this section follows them as is.

#2.18.3. Scheduler model - work-stealing

A process has one executor. It starts on first use and consists of as many worker threads as there are CPUs (cpu_count()), a blocking lane, and an I/O driver. block_on attaches the calling thread to this executor as a guest worker and runs f to completion. Tasks still running when block_on returns keep running. block_on returns only after the spawn_local tasks pinned to its thread are done too; they cannot move to another thread. par::* and spawn_blocking use the same executor.

A task created with spawn can run on any worker, so it must be Future + Send and its result must be Send too. A task from spawn_local is pinned to the worker that created it. A spawn_local from a task that a block_on thread polls but does not own (a spawn task it picked up while waiting) is pinned to a pool worker instead, so block_on waits only for the locals of its own tasks. All the locals one such poll spawns go to the same pool worker, and they start only once that poll has ended, as on the worker that would have run them.

The scheduler skeleton is as follows.

Worker 0 ─┐      Worker 1 ─┐    ...    Worker N-1 ─┐
 Local     │      Local     │            Local      │
 deque     │      deque     │            deque      │
 (Chase-Lev)      (Chase-Lev)            (Chase-Lev)│
           └─ Global Injector Queue ──────┘
           └─ I/O Driver (portable) ──────┘
           └─ Timer (heap)  ───────────────┘
           └─ Parked Worker Stack   ──────┘

Key properties of this structure: the local queue is a lock-free Chase-Lev deque, and both the owning worker and thieves take the oldest task first (so a task that keeps re-waking itself does not starve the others, refs-lang 7.6.2). A busy worker processes expired timers every 32 tasks. For the I/O driver, one reactor thread per process waits on epoll / kqueue edge-triggered (io_uring / IOCP are future work, refs-lang 7.7); the timer structure is implementation-defined (binary heap or timing wheel, refs-lang 7.7.4).

Note that the standard library's authority is limited to the API signatures; the algorithm bodies belong to the runtime and standard library body, outside the spec.

#2.18.4. Cooperative cancellation - Task::cancel

Cooperative cancellation is used as follows.

let t = task::spawn(async { long_running() });
// ... other work ...
t.cancel();              // cooperative signal - the task ends at a cancellation point

cancel() requests cancellation. The rules (refs-lang 7.7.9):

• Every await is a cancellation point. A cancelled task leaves its body at the next await as if by return, running the finally and using cleanups it passes, innermost first.
• Cancellation is not an error that can be catched inside the task. No CancellationError is thrown into the task.
• A blocking syscall is not a cancellation point. The task leaves at the next await after the call finishes.
• A task waiting in socket I/O (*_async), readable / writable, sleep / timeout, lock_async, or recv_async / send_async wakes when cancelled and leaves at the next await.
• When a task ends, all its wakeup registrations (I/O readiness, timers, parked waits) are released.
• If cancelled while waiting on spawn_blocking or the fs::*_async built on it, work already started runs to completion in the background and only the result is discarded. Side effects such as writes do happen. A result that finished in the meantime is dropped too: a cancelled poll returns Pending.
• A poll in which the awaited future handed over a value (a message, a lock guard) keeps that value. The cancellation check looks at "whether the awaited coroutine left because of cancellation"; the value stays and the task leaves at the next await (refs-lang 7.7.9).
• A task cancelled after passing its last cancellation point finishes normally and has a result. await t / join return that result.
• CancellationError is received by whoever waits on a task that ended without a result. Waiting on such a task with try await t.join() throws CancellationError; waiting with await t panics.

let t = task::spawn(async {
    using log = open_log();                      // dropped here on cancellation
    loop {
        await task::sleep(Duration::sec(1));     // cancellation point
        tick(log);
    }
});
t.cancel();

let r = try await t.join() catch _: CancellationError { fallback() };

#2.18.5. thread::* - 1:1 OS threads

The separate thread module spawns OS threads directly rather than using the M:N model.

use rok::std::thread;

let t = thread::spawn { os_level_work() };
let r = t.join();              // OS-level synchronous wait

The two models compare as follows.

Use task::spawn by default, and choose thread::spawn only when an OS thread is explicitly required.

#2.18.6. Duration / Instant

Duration and Instant, for representing time, are defined as follows.

struct Duration { /* i64 in ns */ }
struct Instant { /* monotonic clock */ }

impl Duration {
    fn ns(n: i64) -> Duration;
    fn us(n: i64) -> Duration;
    fn ms(n: i64) -> Duration;
    fn sec(n: i64) -> Duration;
    fn min(n: i64) -> Duration;
    fn hours(n: i64) -> Duration;
}

impl Instant {
    fn now() -> Instant;
    fn elapsed(self) -> Duration;
}

Calls such as await task::sleep(Duration::ms(100)) or await task::sleep_until(t) are guaranteed accurate monotonic time. Both types live in rok::core::task::time (2.18.7), and rok::std::time re-exports them.

#2.18.7. Module mapping

The module where each item lives:

Edit this page on GitHub