Producer/consumer

Two producer tasks pass 400,000 work items to four consumer tasks. The items travel through a BoundedPointerQueue, a lock-free queue of pointers that never waits: it neither blocks on an empty queue nor makes room in a full one. A Semaphore next to the queue counts its items, so a consumer suspends until there is one. The queue's size is chosen so that it cannot overflow, by construction.

The channel

// A fixed pool of items circulates between two queues: free items and items with
// work. There are never more than PoolSize items, so a queue with PoolSize slots
// can never overflow - by construction, not by luck. Each queue has a semaphore
// that counts its items.
constexpr std::size_t PoolSize
    = 256; // a power of two: it is also the queue capacity

class Channel
{
public:
    bool Initialize() noexcept
    {
        if (!free_.Initialize(PoolSize) || !work_.Initialize(PoolSize))
            return false;
        for (WorkItem& item : pool_)
            free_.Enqueue(&item);
        freeCount_.Release(PoolSize);
        return true;
    }

    WorkItem* AcquireFree() noexcept { return Take(freeCount_, free_); }
    void      Post(WorkItem* item) noexcept { Put(workCount_, work_, item); }
    WorkItem* Receive() noexcept { return Take(workCount_, work_); }
    void      Recycle(WorkItem* item) noexcept { Put(freeCount_, free_, item); }

private:
    static void Put(Semaphore& count, BoundedPointerQueue<WorkItem>& queue,
                    WorkItem* item) noexcept
    {
        queue.Enqueue(item); // first the item, then the permit that announces it
        count.Release();
    }

    static WorkItem* Take(Semaphore&                     count,
                          BoundedPointerQueue<WorkItem>& queue) noexcept
    {
        count.Wait(); // suspends the task until an item is there
        // The permit guarantees an item, but Dequeue can see the queue as empty for
        // a moment while another task takes the head item: just try again.
        WorkItem* item;
        while ((item = queue.Dequeue()) == nullptr)
            YieldProcessor();
        return item;
    }

    std::vector<WorkItem>         pool_ = std::vector<WorkItem>(PoolSize);
    BoundedPointerQueue<WorkItem> free_;
    BoundedPointerQueue<WorkItem> work_;
    Semaphore freeCount_; // several tasks wait on each semaphore: the multi-waiter
                          // variant
    Semaphore workCount_;
};

A fixed pool of 256 items circulates between two queues: free items and items with work. A producer takes a free item, fills it and posts it; a consumer receives it, uses it and recycles it. Each queue has 256 slots and there are only 256 items, so no queue can ever hold more than it has room for.

Put enqueues first and releases the permit afterwards: when a task gets the permit, its item is already in the queue. Take waits for a permit and then dequeues. Dequeue may still return nullptr for a moment while another consumer is taking the head item, so Take tries again – the permit guarantees that an item is there.

Producers and consumers

class Producer : public TC::Tasks::Task
{
public:
    Channel*      channel = nullptr;
    std::uint64_t first   = 0; // this producer posts first, first + 1, ...

    void Run() noexcept override
    {
        for (std::uint64_t i = 0; i < ItemsPerProducer; ++i) {
            WorkItem* item = channel->AcquireFree();
            item->value    = first + i;
            item->stop     = false;
            channel->Post(item);
        }
    }
};

class Consumer : public TC::Tasks::Task
{
public:
    Channel*      channel = nullptr;
    std::uint64_t sum     = 0;
    std::uint64_t count   = 0;

    void Run() noexcept override
    {
        for (;;) {
            WorkItem*  item = channel->Receive();
            const bool stop = item->stop;
            if (!stop) {
                sum += item->value;
                ++count;
            }
            channel->Recycle(item);
            if (stop)
                break;
        }
    }
};

Running them

// Everything the tasks use lives in this scope and is destroyed only after
// Stop.
Channel  channel;
Producer producers[Producers];
Consumer consumers[Consumers];
TC::Synchronization::TaskGroup_s
    producerGroup; // main is the only waiter of each group
TC::Synchronization::TaskGroup_s consumerGroup;

for (std::size_t i = 0; i < Consumers; ++i) {
    consumers[i].channel = &channel;
    consumerGroup.AddTask(&consumers[i]);
    scheduler->AddTask(&consumers[i]);
}
for (std::size_t i = 0; i < Producers; ++i) {
    producers[i].channel = &channel;
    producers[i].first   = i * ItemsPerProducer;
    producerGroup.AddTask(&producers[i]);
    scheduler->AddTask(&producers[i]);
}

// Once all work is posted, one stop item per consumer. They come from
// the same pool, so they cannot overflow the queue either.
producerGroup.Wait();
for (std::size_t i = 0; i < Consumers; ++i) {
    WorkItem* item = channel.AcquireFree();
    item->stop     = true;
    channel.Post(item);
}
consumerGroup.Wait();

When all producers have finished, main sends one stop item per consumer. The stop items come from the same pool, so they cannot overflow the queue either.

Full program

// Producer/consumer: producer tasks pass work items to consumer tasks through a
// BoundedPointerQueue. Semaphores count the items, because the queue itself never
// waits.
#define NOMINMAX
#include <Windows.h>

#include <cstddef>
#include <cstdint>
#include <cstdio>
#include <vector>

#include <ThinkMeta.ConcurrentTasks/Core.h>
import ThinkMeta.ConcurrentTasks.Collections.BoundedPointerQueue;
import ThinkMeta.ConcurrentTasks.Synchronization.Semaphore;
import ThinkMeta.ConcurrentTasks.Synchronization.TaskGroup;
import ThinkMeta.ConcurrentTasks.Tasks.Task;

namespace TC = ThinkMeta::ConcurrentTasks;
using TC::Collections::BoundedPointerQueue;
using TC::Synchronization::Semaphore;

constexpr std::size_t Producers        = 2;
constexpr std::size_t Consumers        = 4;
constexpr std::size_t ItemsPerProducer = 200'000;

struct WorkItem {
    std::uint64_t value = 0;
    bool          stop  = false; // tells a consumer to end
};

// A fixed pool of items circulates between two queues: free items and items with
// work. There are never more than PoolSize items, so a queue with PoolSize slots
// can never overflow - by construction, not by luck. Each queue has a semaphore
// that counts its items.
constexpr std::size_t PoolSize
    = 256; // a power of two: it is also the queue capacity

class Channel
{
public:
    bool Initialize() noexcept
    {
        if (!free_.Initialize(PoolSize) || !work_.Initialize(PoolSize))
            return false;
        for (WorkItem& item : pool_)
            free_.Enqueue(&item);
        freeCount_.Release(PoolSize);
        return true;
    }

    WorkItem* AcquireFree() noexcept { return Take(freeCount_, free_); }
    void      Post(WorkItem* item) noexcept { Put(workCount_, work_, item); }
    WorkItem* Receive() noexcept { return Take(workCount_, work_); }
    void      Recycle(WorkItem* item) noexcept { Put(freeCount_, free_, item); }

private:
    static void Put(Semaphore& count, BoundedPointerQueue<WorkItem>& queue,
                    WorkItem* item) noexcept
    {
        queue.Enqueue(item); // first the item, then the permit that announces it
        count.Release();
    }

    static WorkItem* Take(Semaphore&                     count,
                          BoundedPointerQueue<WorkItem>& queue) noexcept
    {
        count.Wait(); // suspends the task until an item is there
        // The permit guarantees an item, but Dequeue can see the queue as empty for
        // a moment while another task takes the head item: just try again.
        WorkItem* item;
        while ((item = queue.Dequeue()) == nullptr)
            YieldProcessor();
        return item;
    }

    std::vector<WorkItem>         pool_ = std::vector<WorkItem>(PoolSize);
    BoundedPointerQueue<WorkItem> free_;
    BoundedPointerQueue<WorkItem> work_;
    Semaphore freeCount_; // several tasks wait on each semaphore: the multi-waiter
                          // variant
    Semaphore workCount_;
};

class Producer : public TC::Tasks::Task
{
public:
    Channel*      channel = nullptr;
    std::uint64_t first   = 0; // this producer posts first, first + 1, ...

    void Run() noexcept override
    {
        for (std::uint64_t i = 0; i < ItemsPerProducer; ++i) {
            WorkItem* item = channel->AcquireFree();
            item->value    = first + i;
            item->stop     = false;
            channel->Post(item);
        }
    }
};

class Consumer : public TC::Tasks::Task
{
public:
    Channel*      channel = nullptr;
    std::uint64_t sum     = 0;
    std::uint64_t count   = 0;

    void Run() noexcept override
    {
        for (;;) {
            WorkItem*  item = channel->Receive();
            const bool stop = item->stop;
            if (!stop) {
                sum += item->value;
                ++count;
            }
            channel->Recycle(item);
            if (stop)
                break;
        }
    }
};

int main()
{
    TC::ConvertThreadToTask();

    TC::SchedulerInfo   info{};
    TC::ITaskScheduler* scheduler
        = TC::CreateTaskScheduler(TC::SchedulerKind::Default, &info);
    if (scheduler == nullptr)
        return 1;
    scheduler->Start();

    bool ok = false;
    {
        // Everything the tasks use lives in this scope and is destroyed only after
        // Stop.
        Channel  channel;
        Producer producers[Producers];
        Consumer consumers[Consumers];
        TC::Synchronization::TaskGroup_s
            producerGroup; // main is the only waiter of each group
        TC::Synchronization::TaskGroup_s consumerGroup;

        if (channel.Initialize()) {
            for (std::size_t i = 0; i < Consumers; ++i) {
                consumers[i].channel = &channel;
                consumerGroup.AddTask(&consumers[i]);
                scheduler->AddTask(&consumers[i]);
            }
            for (std::size_t i = 0; i < Producers; ++i) {
                producers[i].channel = &channel;
                producers[i].first   = i * ItemsPerProducer;
                producerGroup.AddTask(&producers[i]);
                scheduler->AddTask(&producers[i]);
            }

            // Once all work is posted, one stop item per consumer. They come from
            // the same pool, so they cannot overflow the queue either.
            producerGroup.Wait();
            for (std::size_t i = 0; i < Consumers; ++i) {
                WorkItem* item = channel.AcquireFree();
                item->stop     = true;
                channel.Post(item);
            }
            consumerGroup.Wait();

            std::uint64_t sum   = 0;
            std::uint64_t count = 0;
            for (const Consumer& consumer : consumers) {
                std::printf("consumer: %llu items\n",
                            static_cast<unsigned long long>(consumer.count));
                sum += consumer.sum;
                count += consumer.count;
            }
            const std::uint64_t n = Producers * ItemsPerProducer;
            ok                    = count == n && sum == n * (n - 1) / 2;
            std::printf("%llu items, sum %llu (%s)\n",
                        static_cast<unsigned long long>(count),
                        static_cast<unsigned long long>(sum),
                        ok ? "ok" : "MISMATCH");
        }
        scheduler->Stop();
    }

    scheduler->Destroy();
    TC::ConvertTaskToThread();
    return ok ? 0 : 1;
}

Source and project on GitHub

Build and run

A console program. With the core's props file imported and the library referenced (see Set up a project) it needs nothing else. It finishes in a fraction of a second; how the items spread over the consumers varies from run to run:

consumer: 101744 items
consumer: 97185 items
consumer: 99917 items
consumer: 101154 items
400000 items, sum 79999800000 (ok)

What to notice

Important

A BoundedPointerQueue never grows. Its capacity, a power of two set by Initialize, must cover the most elements the program can ever hold in it at once; enqueueing into a full queue is invalid use and leaves the queue unusable. Nothing checks this in a release build. Here the bound is the size of the item pool. See Performance first.

The queue does not wait – the semaphore does. Enqueue and Dequeue are lock-free and callable from any thread; they never suspend. Waiting for an item is the semaphore's job.
Several waiters, so no _s. Several consumers wait on workCount_ and several producers on freeCount_ at the same time, so both are Semaphore, not Semaphore_s.
nullptr means "try again". Under contention Dequeue can report an empty queue for a moment. With a permit in hand the item is there; the retry loop is short.
Lifetimes. The channel, the tasks and the groups live until after Stop. See Lifetimes.