Queues and stacks

The library's collections are small, lock-free building blocks. None of them waits: BoundedPointerQueue is a fixed-size FIFO of pointers, ConcurrentStack a lock-free intrusive LIFO over the Windows SList, BasicStack its unsynchronized counterpart, and Padded places an object on cache lines of its own. To make a task wait for an element, combine them with a Semaphore. The examples come from guide-queues-stacks.cpp: producer tasks pass jobs to consumer tasks through a bounded buffer, and the job nodes are recycled through a lock-free free list.

BoundedPointerQueue

BoundedPointerQueue<T> holds T* pointers in a ring. Any number of threads and tasks may call Enqueue and Dequeue at the same time; the scheduler's own task queues are built on it.

• Construct it, then call Initialize once with the capacity, before the queue is shared. The capacity is a power of two and at least 2. Initialize returns false if the memory could not be allocated. Capacity returns it (0 before Initialize).
• Enqueue takes a pointer that is not nullptr.
• Dequeue returns the oldest pointer, or nullptr if the queue is empty. Under contention the queue can look empty for a moment while another consumer is taking an element, so nullptr means "try again later", not proof that the queue is empty.
• The queue does not own the pointers; destroying it does not touch them.

A bounded buffer

Neither Enqueue nor Dequeue waits, so two semaphores do the waiting: one counts free slots and keeps the queue from overflowing, one counts jobs and lets consumers sleep while there are none. The example uses these sizes:

constexpr std::size_t Capacity  = 64; // queue capacity, a power of two
constexpr int         Producers = 4;  // tasks that submit jobs
constexpr int         Consumers = 4;  // tasks that handle them

// A bounded buffer for any number of producer and consumer tasks. The queue never
// waits, so two semaphores do the waiting: one counts free slots, one counts jobs.
class Buffer
{
public:
    bool Initialize() noexcept { return queue_.Initialize(Capacity); }

    void Put(Job* job) noexcept
    {
        free_.Wait(); // never more than Capacity jobs in the queue
        queue_.Enqueue(job);
        jobs_.Release();
    }

    Job* Take() noexcept
    {
        jobs_.Wait(); // a job is in the queue
        Job* job;
        while ((job = queue_.Dequeue()) == nullptr)
            YieldProcessor(); // nullptr under contention means "try again"
        free_.Release();
        return job;
    }

private:
    BoundedPointerQueue<Job> queue_;
    Semaphore                free_{Capacity};
    Semaphore                jobs_{0};
};

Take retries Dequeue after jobs_.Wait(): the semaphore guarantees that a job is in the queue, but a concurrent Dequeue may hide it for a moment. With several producers and consumers the semaphores must be the multi-waiter Semaphore; with one producer and one consumer Semaphore_s is enough.

ConcurrentStack

ConcurrentStack<T> is a lock-free LIFO for items derived from SLIST_ENTRY. The items are linked through that base, so Push never allocates. Push and Pop may be called from any number of threads and tasks at once; Pop returns nullptr when the stack is empty. Size is a snapshot (QueryDepthSList) that may be outdated at once and wraps above 65535.

The Windows SList requires every item, and the stack object itself, to be aligned to MEMORY_ALLOCATION_ALIGNMENT (16 bytes on x64):

// A job that can sit on a ConcurrentStack: derived from SLIST_ENTRY and aligned to
// 16 bytes (MEMORY_ALLOCATION_ALIGNMENT on x64), as the Windows SList requires.
struct alignas(MEMORY_ALLOCATION_ALIGNMENT) Job : SLIST_ENTRY {
    int orderId = 0;
    int amount  = 0;
};

A typical use is a free list: preallocated nodes are popped by whoever needs one and pushed back when done, without a lock and without allocating:

// A free list of nodes shared by all tasks: Push and Pop are lock-free and
// never allocate.
static Job nodes[Capacity + Producers
                 + Consumers]; // queue full, one per producer and consumer
ConcurrentStack<Job> pool;
for (Job& node : nodes)
    pool.Push(&node);

A producer pops a node, fills it and puts it into the buffer; a consumer takes a job, handles it, counts it in its own statistics (see Padded) and pushes the node back:

// Producer side: take a node from the free list, fill it, put it into the buffer.
void Submit(Buffer& buffer, ConcurrentStack<Job>& pool, int orderId,
            int amount) noexcept
{
    Job* job
        = pool.Pop(); // never empty: enough nodes for every place a node can be
    job->orderId = orderId;
    job->amount  = amount;
    buffer.Put(job);
}

// Consumer side: take a job, handle it, return the node to the free list.
void HandleNext(Buffer& buffer, ConcurrentStack<Job>& pool,
                ConsumerStats& myStats) noexcept
{
    Job* job = buffer.Take();
    ++myStats.jobs;
    myStats.amount += job->amount; // ... handle the job here ...
    pool.Push(job);                // back to the free list
}

The suspended tasks of the multi-waiter synchronization primitives are kept on a ConcurrentStack too, which is why those primitives wake waiters in LIFO order.

BasicStack

BasicStack<T> has the same intrusive interface without synchronization. Use it where only one task or thread touches the stack, for example for a local list:

// One task only: take nodes for a batch onto a local stack, return them when
// done.
BasicStack<Job> batch;
for (int i = 0; i < 8; ++i)
    if (Job* job = pool.Pop())
        batch.Push(job);
// ... fill and use the batch ...
while (Job* job = batch.Pop())
    pool.Push(job);

Destroying a non-empty BasicStack or ConcurrentStack just forgets the items.

Padded

Two variables that different threads write often should not share a cache line: every write invalidates the line in the other cores' caches (false sharing). Padded<T> derives from T, aligns it to the cache line size and pads it to a multiple of it, also in arrays:

// Statistics per consumer, each on its own cache line: consumers on different
// threads do not slow each other down by writing to the same line (false sharing).
struct ConsumerStats {
    std::uint64_t jobs   = 0;
    std::uint64_t amount = 0;
};
TC::Cpp::Padded<ConsumerStats> stats[Consumers];

A Padded<T> is a T and is constructed and used like one. T must be a class that can be derived from (not final). The optional second parameter is the padding size, a power of two; BoundedPointerQueue is itself a Padded type.

The program prints:

jobs:       40000 handled, amounts match
free nodes: 72 of 72
OK

See also

• BoundedPointerQueue, ConcurrentStack, BasicStack, Padded, Semaphore
• Performance first, Lifetimes and sizing, The scheduler
• Semaphores and locks, Task graphs, Custom primitives
• Samples: Producer/consumer