Skip to content
CryoCryo home
Stdlibfuture

blocking

import std::future::blocking; · source

An off-executor home for work that cannot be polled. A Reactor lets a future sleep on something the OS can report readiness for; some work has no such report. getaddrinfo is a synchronous call with no non-blocking form on either platform, and a regular file is always "ready" as far as epoll is concerned, so neither can be driven by the readiness reactor. Running one directly inside a poll stalls an executor worker for the whole call — on a single-threaded Executor, the whole runtime.

The pool is the standard answer: a small set of OS threads that run such calls off the executor and fire a Waker when they finish. The call still blocks a thread; it blocks one that nobody is polling on.

Types

BlockingPool

type struct BlockingPool {
    mtx:      [u8;
    40];
    cv:       [u8;
    48];
    head:     BlockingJob*;
    tail:     BlockingJob*;
    shutdown: boolean;
    live:     u32;
    idle:     u32;
    max:      u32;
    tids:     u64*;

    static start() -> BlockingPool*;
    static with_max(max: u32) -> BlockingPool*;
    static stop(p: BlockingPool*) -> void;
    static free(p: BlockingPool*) -> void;
    static current() -> BlockingPool*;
    static set_current(p: BlockingPool*) -> void;
    submit(mut &this, job: BlockingJob*) -> boolean;
    take(mut &this) -> BlockingJob*;
}

An Executor starts one and its workers set it as the thread's ambient pool through set_current; current() is how a future finds it. submit queues a job and returns whether a thread will run it.

Properties

PropertyType
mtx[u8
40]
cv[u8
48]
headBlockingJob*
tailBlockingJob*
shutdownboolean
liveu32
idleu32
maxu32
tidsu64*

Constructors

static start() -> BlockingPool*;

Create an empty pool with the default thread ceiling.

static with_max(max: u32) -> BlockingPool*;

Create an empty pool that will grow to at most max threads (clamped to >= 1: a pool that could never create a thread would accept jobs and never reach them). No thread is started here — the first submit creates one, so a program that never offloads anything never pays for a thread, and max costs only its slot in the id table. Panics only if the OS refuses the lock primitives or the allocator refuses that table, either of which would make every later submit unsound.

static current() -> BlockingPool*;

The pool serving the current thread's polls, or null if this thread is not an executor worker.

Static functions

static stop(p: BlockingPool*) -> void;

Stop the pool and release everything it owns, leaving the block itself allocated (as Reactor::stop does, and for the same reason: a future dropped later still lets go of a job through this pool).

static free(p: BlockingPool*) -> void;

Release the pool block. Must run only after every future that could still let go of a job has been dropped — see stop.

static set_current(p: BlockingPool*) -> void;

Publish p as this thread's pool. Called by an executor worker as it starts, so every future it polls can reach the pool.

Methods

submit(mut &this, job: BlockingJob*) -> boolean;

Queue job and make sure a thread will reach it, acquiring the pool's reference to the block. Returns false when the job was NOT accepted — the pool is shutting down, or it has no thread and the OS refused to create one — in which case the caller still owns the block and must run the work itself rather than wait for a completion that will never come.

take(mut &this) -> BlockingJob*;

Block until a job is queued or shutdown is signaled. Returns the dequeued job, or null once shutdown is set — leaving anything still queued for stop to cancel.

BlockingJob

type struct BlockingJob {
    rc:     Atomic<u32>;
    status: Atomic<u8>;

    run_fn: (u8*) -> void;
    drop_fn: (u8*) -> void;

    waker: Waker;
    next:  BlockingJob*;
    pool:  BlockingPool*;

    static init(job: BlockingJob*, run_fn: (u8*) -> void, drop_fn: (u8*) -> void) -> void;
    arm(mut &this, w: Waker) -> u8;
    finish(mut &this, status: u8) -> void;
    retain(mut &this) -> void;
    release(mut &this) -> void;
}

Properties

PropertyType
rcAtomic<u32>
statusAtomic<u8>
wakerWaker
nextBlockingJob*
poolBlockingPool*

Static functions

static init(job: BlockingJob*, run_fn: (u8*) -> void, drop_fn: (u8*) -> void) -> void;

Prepare a freshly allocated block. The refcount starts at 1 — the submitter's own reference — and submit acquires the pool's second one only if it accepts the job, so a refused job needs no unwinding.

Methods

run_fn: (u8*) -> void;
drop_fn: (u8*) -> void;
arm(mut &this, w: Waker) -> u8;

Read this job's status and install w as the waker to fire when it finishes. Returns the status as of the moment w was stored.

finish(mut &this, status: u8) -> void;

Publish status and fire the waker the submitter armed, then release the pool's reference. Called by a pool thread when run_fn returns, and by BlockingPool::stop (with JOB_CANCELLED) for a job no thread reached.

retain(mut &this) -> void;

Acquire one reference to this block, so it cannot be freed while the acquirer still names it. init hands out the first; submit takes the pool's. Pairs one-for-one with release.

release(mut &this) -> void;

Drop one reference to this block. The last one releases any waker still stored and hands the block to its payload's drop_fn, which frees it.

The shape of a job

The pool's queue is homogeneous, so a job is type-erased the same way a task and a waker are — a manual vtable of non-capturing function pointers. A user of the pool declares a struct whose first field is a BlockingJob and hands that field's address to submit; run_fn casts the address back to the concrete block and finds its payload beside the header:

type struct ResolveJob {
    job:  BlockingJob;   // first: `submit` is handed &this.job
    host: String;        // the request, owned by the block
    out:  Result<..>;    // written by `run_fn`, read once it finishes
}

init fills the header with the run and drop functions; the future's poll calls arm(waker) to say who to wake and reads the status it returns; finish is what the pool thread calls when run_fn returns. retain and release are the reference count that lets the future and the pool thread disagree about who finishes last — a cancelled future releases its side and the pool frees the block when the call eventually returns.

Growth

Threads are created on demand, never in advance, and only when no thread is already idle: a pool that grew on every submit would reach its ceiling under a burst one thread could have absorbed. They are not reaped while the pool lives — a blocking pool's threads are asleep in a syscall, not consuming CPU, and an executor's lifetime is the natural bound. Past max threads, jobs queue. Executor::with_threads_and_blocking sets the cap.

net::dns::Resolve, process::ChildWait, and the pipe drain behind Command::collect are the in-tree users.

Constants

const JOB_PENDING:              u8  = 0;
const JOB_DONE:                 u8  = 1;
const JOB_CANCELLED:            u8  = 2;
const BLOCKING_DEFAULT_THREADS: u32 = 512;