Custom primitives

Every synchronization primitive of ConcurrentTasks is ordinary C++ on top of three functions of Core.h: SuspendTask takes the calling task off the processor, ResumeTask makes it runnable again, and SuspendTaskFor does the same with a timeout. Your own primitive uses the same functions, or the library's building blocks SuspendableBase, Suspendable and Waitable. This guide builds one primitive each way. The code comes from guide-custom-primitives.cpp, which checks its own results.

The suspend protocol

A task that has to wait calls SuspendTask(callback). The core switches the task off its thread and then calls ISuspendCallback::OnSuspended with the task's handle, a SuspendedTask. The callback publishes the handle, for example by storing it where a signaling thread will find it, or, if the condition the task waits for has come true in the meantime, passes it straight to ResumeTask. Whoever later calls ResumeTask with the handle makes the task runnable again on the scheduler that suspended it; SuspendTask then returns. ResumeTask accepts only a SuspendedTask, never a plain TaskHandle such as the one GetCurrentTask returns, so resuming a task that is not suspended does not compile.

Publishing after the task is off the processor is what makes the protocol safe: no resumer can see the handle while the task is still running.

A primitive on Core.h

The mailbox below holds one int for one waiting task. Put may be called from any thread; Take suspends until a value is there:

// Holds one int (0 means empty) for one waiting task. Put may be called from any
// thread.
class Mailbox : private TC::ISuspendCallback
{
public:
    void Put(int value) noexcept
    {
        value_.store(value, std::memory_order_seq_cst); // 1. state
        if (auto waiter
            = waiter_.exchange(nullptr, std::memory_order_acq_rel)) // 2. waiter
            TC::ResumeTask(waiter); // whoever takes the handle resumes
    }

    // Task or converted thread only; at most one waiter at a time.
    int Take() noexcept
    {
        int value;
        while ((value = value_.exchange(0, std::memory_order_acq_rel)) == 0)
            TC::SuspendTask(
                this); // calls OnSuspended once this task is off the processor
        return value;
    }

    // Returns false if nothing arrived in time (needs EnableTimers, else no limit).
    bool Take(int& value, TC::Duration timeout) noexcept
    {
        while ((value = value_.exchange(0, std::memory_order_acq_rel)) == 0)
            if (!TC::SuspendTaskFor(this, timeout))
                return false; // OnTimeout took the task back
        return true;
    }

private:
    // Runs on the scheduler thread, outside any task: must not suspend, yield or
    // wait.
    void OnSuspended(TC::SuspendedTask task) noexcept override
    {
        waiter_.store(task, std::memory_order_seq_cst);  // 1. publish the waiter
        if (value_.load(std::memory_order_seq_cst) != 0) // 2. re-check the state:
            if (auto waiter = waiter_.exchange(
                    nullptr, std::memory_order_acq_rel)) // a Put may have
                TC::ResumeTask(waiter);                  // missed the waiter
    }

    // Runs on the timer thread while the task is still suspended: take the handle
    // back, or report that a resumer has it already.
    bool OnTimeout(TC::SuspendedTask task) noexcept override
    {
        TC::SuspendedTask expected = task;
        return waiter_.compare_exchange_strong(expected, nullptr,
                                               std::memory_order_acq_rel);
    }

    std::atomic<int>               value_{0};
    std::atomic<TC::SuspendedTask> waiter_{nullptr};
};

Lost wake-ups

Take checks the value, finds it empty and suspends. If Put stores a value between that check and the moment the waiter is published, Put finds no waiter and the task would sleep forever. Two measures close that gap:

• OnSuspended re-checks the condition after publishing the waiter.
• Put writes the state before it looks for a waiter.

With sequentially consistent operations on both sides, at least one of them sees the other: either Put finds the waiter, or the re-check finds the value. If both do, both try to take the handle with an exchange, and exactly one of them resumes the task. A primitive must never resume the same handle twice.

The re-check reads the mailbox after the waiter was published, so the mailbox must still exist then. The library's single-waiter events avoid that window by keeping state and waiter in one atomic word: one compare-exchange publishes and checks at once and is the last access, which is why an AutoResetEvent_s may be destroyed right after its Wait returns.

Timeouts

SuspendTaskFor suspends like SuspendTask, but ends the wait after the timeout at the latest. It returns true if a ResumeTask ended the wait and false if it timed out. Without EnableTimers, or with Duration::Infinite, it waits without a timeout.

When the time is up, the core's timer thread calls ISuspendCallback::OnTimeout while the task is still suspended. The callback races with the resumers for the handle:

• It returns true if it took the task back from where OnSuspended published it. The core then resumes the task as timed out, and SuspendTaskFor returns false.
• It returns false if a resumer took the task first. The core then waits for that resumer's ResumeTask, and SuspendTaskFor returns true.

OnTimeout must be short and must not resume the task itself. The default implementation returns false, which is right only for primitives that never time out. Once the task runs again, the core does not touch the callback any more.

Using the mailbox

Two mailboxes make a request/reply channel between a client and a service task:

// A service task: answers requests until it receives a negative one. Values must
// not be 0, which is the mailbox's "empty".
void Serve(Mailbox& requests, Mailbox& replies) noexcept
{
    for (int request; (request = requests.Take()) > 0;)
        replies.Put(request * 2); // ... compute the answer here ...
}

// Called by one task at a time: asks and waits for the answer.
int Ask(Mailbox& requests, Mailbox& replies, int request) noexcept
{
    requests.Put(request);
    return replies.Take();
}

Each mailbox has one waiter at a time: the service takes requests, the one client takes replies. Take(value, timeout) gives up after the timeout if nothing arrives.

Chaining waiters without allocation

The handle points to a SuspendedTaskEntry, whose leading SLIST_ENTRY belongs to the primitive from the moment OnSuspended hands it over until the primitive passes the handle to ResumeTask. A primitive with several waiters can therefore chain them on a ConcurrentStack<SuspendedTaskEntry> or a BasicStack<SuspendedTaskEntry> (ConcurrentStack, BasicStack) without allocating, which is what the library's multi-waiter primitives do. Outside that window the field belongs to the core.

A primitive on Waitable

The library's own primitives are built from these layers:

• SuspensionKind chooses Single (one suspended task at a time, kept in one atomic slot; the only kind with timeouts) or Multiple (up to 65535 suspended tasks on a lock-free stack, woken in LIFO order, no timeouts).
• SuspendableBase keeps the suspended tasks of one object: Suspend (and SuspendFor for Single), Resume (all tasks suspended right now; for Multiple the task that constructed the object is resumed last), ResumeNext (one task), ResumeNonReentrant and IsSuspended. Destroy it only when no task is suspended on it.
• Suspendable<Kind, T> implements OnSuspended: it publishes the task and then calls T::TryResume(), the re-check. TryResume may run after another thread has already resumed the task, so it must not assume that the task is still suspended.
• Waitable<Kind, T> turns a non-blocking T::TryWait() into Wait (and Wait(Duration) for Single): it calls TryWait first, suspends only if that fails, and after publishing calls TryWait again and, if it now succeeds, T::Resume().

With Waitable, a countdown latch takes a few lines. T provides a public TryWait and a Resume that resumes the tasks a successful TryWait stands for:

// Opens once CountDown has been called count times; any number of tasks may wait.
// Waitable supplies Wait(): it calls TryWait, suspends if that fails, and re-checks
// after the task has been published (then it calls Resume).
class Latch : public TC::Synchronization::Waitable<
                  TC::Synchronization::SuspensionKind::Multiple, Latch>
{
public:
    explicit Latch(std::size_t count) noexcept : count_(count) {}

    void CountDown() noexcept
    {
        if (count_.fetch_sub(1, std::memory_order_acq_rel) == 1)
            Resume(); // the last one opens the latch
    }

    // Required by Waitable: non-blocking check of the condition.
    bool TryWait() noexcept { return count_.load(std::memory_order_acquire) == 0; }

    // Required by Waitable: resume the tasks the condition lets pass (all of them).
    void Resume() noexcept
    {
        TC::Synchronization::SuspendableBase<
            TC::Synchronization::SuspensionKind::Multiple>::Resume();
    }

private:
    std::atomic<std::size_t> count_;
};

CountDown changes the state first and then resumes, so it pairs with the re-check in the same way as the mailbox. The latch inherits the rules of the multi-waiter primitives: destroy it only when no task can wait on it or count it down any more, and the last Resume has returned.

A typical use: several workers each prepare a part, and the tasks that need all parts wait for the latch:

// Each worker prepares its part and counts the latch down.
void PreparePart(Latch& prepared, Part& part) noexcept
{
    Load(part);
    prepared.CountDown();
}

// Any number of tasks may wait until every part is prepared.
int TotalSize(Latch& prepared, std::span<const Part> parts) noexcept
{
    prepared.Wait();
    int total = 0;
    for (const Part& part : parts)
        total += part.size;
    return total;
}

The program uses both primitives from several tasks, checks the results and prints:

mailbox:  sum 100010000
timeout:  timed out
latch:    total 828, seen by all three waiters: yes
OK

Checklist

• Wait only from a task or a converted thread; resume from anywhere.
• Publish the waiter in OnSuspended, never before SuspendTask.
• Change the state before looking for waiters; re-check the state after publishing a waiter.
• Make every handle reach ResumeTask exactly once; claim it with an atomic exchange.
• Implement OnTimeout for timed waits, with the same claim as the resumers.
• Never suspend, yield or wait inside the callbacks.
• Decide and document when the primitive may be destroyed.

See also

• SuspendTask, SuspendTaskFor, ResumeTask, ISuspendCallback, SuspendedTask, TaskHandle, GetCurrentTask, SuspendableBase, Suspendable, Waitable, SuspensionKind
• Suspending and resuming, Single-waiter variants, Timers, Lifetimes and sizing
• Events, Queues and stacks
• Samples: Ping-pong