Synchronization Primitives

Tasks that share state, wait for each other, or wait for an event outside Capy need a way to wait. That wait must suspend the task, not block its thread. Capy provides one serializing executor and three primitives for this.

Choosing a Primitive

Need Use Waiters Shown in

One task at a time touches the shared state

strand

Not applicable: the strand runs everything posted to it, one at a time

Executors, Strand Serialization

Exclusive access to the state, held across a co_await

async_mutex

Many, resumed one at a time in first-in, first-out order

Async Mutex

Resume every waiting task when a condition becomes true

async_event

Many, all resumed by set(); the event stays set until clear()

Producer-Consumer

Resume one task from another thread or library

async_waker

One at a time, plus one wakeup kept for the next wait

Timeout with Cancellation, Writing Your Own Task Type (Early Quitter)

A strand serializes everything that runs on it. A mutex serializes only its critical section, so other work on the same executor keeps running. Strand vs Async Mutex compares the two.

Holding a Lock Across a co_await

A task that finds the async_mutex held suspends instead of blocking its thread. The waiting tasks form a queue, and unlock() hands the mutex to the first waiter that a stop has not already claimed, then posts it.

capy::async_mutex mtx;
auto [ec, guard] = co_await mtx.scoped_lock();
if (ec)
{
    std::cout << "Worker " << id
              << " canceled: " << ec.message() << "\n";
    co_return capy::results<>{ec};
}

scoped_lock() completes with an error code and a guard. The error code is set when the lock is canceled, for example by a stop request that arrives while the task waits in the queue. The guard releases the mutex when it leaves its scope.

In this example, the critical section does not suspend. A real task often must, for example to write the shared state to a stream. The guard stays held across any co_await in its scope, and the other tasks stay queued until it is released.

A std::mutex cannot do this. It blocks its thread while it waits. It must also be unlocked on the thread that locked it, but a task on a thread pool can resume on another thread.

Resuming Many Tasks at Once

An async_event starts unset. Every task that awaits wait() on an unset event suspends.

capy::async_event data_ready;

The producer calls set(), which posts every waiting task to its executor:

auto producer = [&]() -> capy::task<capy::results<>> {
    std::cout << "Producer: preparing data...\n";
    shared_value = 42;
    std::cout << "Producer: data ready, signaling\n";
    data_ready.set();
    co_return capy::results<>{};
};

The consumer waits for that call:

auto consumer = [&]() -> capy::task<capy::results<>> {
    std::cout << "Consumer: waiting for data...\n";
    auto [ec] = co_await data_ready.wait();
    (void)ec;
    std::cout << "Consumer: received value " << shared_value << "\n";
    co_return capy::results<>{};
};

The event stays set until clear() is called. A wait() on a set event completes at once, so the order of set() and wait() does not matter.

Resuming a Task from Another Thread

An async_waker connects a task to an event that Capy does not see. That event can come from a timer thread, a callback from another library, or a thread that watches a flag set by a signal handler. One task suspends in wait(), and any thread calls wake().

struct fetch_channel
{
    capy::async_waker fetch_ready;
    std::atomic<bool> cancelled{false};
    std::string result;
};
capy::task<capy::results<std::string>> await_fetch(fetch_channel& ch)
{
    auto [ec] = co_await ch.fetch_ready.wait();
    if (ec)
    {
        ch.cancelled.store(true);
        co_return capy::results<std::string>{ec, {}};
    }
    co_return capy::results<std::string>{std::error_code(), std::move(ch.result)};
}

A worker thread stores the result in the channel, then calls wake() on fetch_ready. Timeout with Cancellation shows the whole program.

wake() never resumes the task inline. It posts the task to the executor that the task awaited from. A wakeup that arrives before the task waits is kept, and the next wait() consumes it at once. Several wakeups before a wait count as one. Only one task may wait on a waker at a time.

The Threading Model

The three primitives share one threading model. Every task that touches a given mutex or event must run on the same executor, and that executor must never run its continuations concurrently. Use a single-threaded executor, or one strand over a thread pool that all those tasks share. Two tasks on two different strands must not share a mutex or an event. Producer-Consumer and Async Mutex run all their tasks on one shared strand for this reason. Awaiting a primitive directly on a multi-threaded executor is undefined.

A waker is the exception. Only one task waits on it at a time, and that task’s executor follows the same rule. wake() may be called from any thread.

A primitive resumes each waiter by posting it to its executor. A stop callback can run on any thread, so it also only posts. Stop Callbacks Must Post, Not Resume explains why.

Cancellation

A waiter observes the stop token of the task that awaits it. When a stop request arrives while the task is suspended, the task resumes with error::canceled. Compare the result with cond::canceled, as The canceled Convention describes.

Two rules decide what an operation does when a stop is already requested:

  • An operation that can complete at once does so, and ignores the stop token. This covers a free mutex, a set event, and a waker that holds a wakeup.

  • An operation that would suspend completes at once with error::canceled. It does not join the queue.

A wakeup that loses the race against a stop request is kept for the next wait().

Destruction

The usual order is to request a stop, wait for the work to end, and then destroy the primitive. In this snippet, the pool has one thread, so it meets the rule in The Threading Model:

capy::thread_pool pool(1);
std::stop_source stop;
capy::run_async(pool.get_executor(), stop.get_token())(
    wait_until_stopped(never_set));

// Later, to shut down:
stop.request_stop(); (1)
pool.join();         (2)
(3)
1 The waiting task observes the stop token and finishes through its own code.
2 join waits for the outstanding work, then for the worker threads.
3 From here on, nothing waits on never_set, so the event and the pool can be destroyed safely.

Destroying a primitive that still has waiters cancels them. Each waiter is posted to its executor to resume with error::canceled. An executor that can no longer run the waiter abandons it instead, possibly before the destructor returns. A task that the mutex was handed to, but that has not resumed yet, resumes canceled. It holds nothing.

Two preconditions apply:

  • No task holds the async_mutex when it is destroyed.

  • No call to wake() is in progress on the async_waker, or made afterward.

The reference page of each destructor gives the exact wording. Shutdown Ordering explains the order in the general case.

Allocation and Lifetime

The primitives allocate nothing themselves. Resuming a waiter posts it to its executor, and that executor may allocate. Each waiter lives in the frame of the task that awaits.

The primitives can be neither copied nor moved, because suspended waiters point into them. Keep each one at a stable address that outlives its waiters. For example, Writing Your Own Task Type (Early Quitter) keeps its wakers in a std::array.

See Also

Edit this Page