LockFree_ConsumerProducer.cpp

#define _ENABLE_ATOMIC_ALIGNMENT_FIX 1

#include <iostream>
#include <thread>
#include <array>
#include <atomic>
#include <memory>
#include <Windows.h>
#include <variant>
#include <vector>

using namespace std;

constexpr size_t SLOTS = 5;

class Semaphore
{
public:
    Semaphore()
    {
        hSemaphore = CreateSemaphore(
            nullptr,
            0,
            1,
            nullptr);
    }

    ~Semaphore()
    {
        CloseHandle(hSemaphore);
    }

    Semaphore(Semaphore const&) = delete;
    Semaphore& operator=(Semaphore const&) = delete;

    Semaphore(Semaphore&& old)
    {
        std::swap(hSemaphore, old.hSemaphore);
    }

    Semaphore& operator=(Semaphore&& rhs)
    {
        std::swap(this->hSemaphore, rhs.hSemaphore);
    }

    void Wait()
    {
        WaitForSingleObject(hSemaphore, 0L);
    }

    void Signal()
    {
        ReleaseSemaphore(hSemaphore, 1L, nullptr);
    }

private:
    HANDLE hSemaphore;
};

struct Done {};
struct NoTask {};
struct Workload { int value; };

using Task = variant<NoTask, Done, Workload>;

array<atomic<Task>, SLOTS> slots;
array<Semaphore, SLOTS> sems;

bool moveToNextSlot(int* pCurrent)
{
    *pCurrent = (*pCurrent + 1) % SLOTS;
    return true;
}

struct ProducerTaskVisitor
{
    ProducerTaskVisitor(int* pCurrent) : pCurrent(pCurrent) {}

    bool operator()(NoTask)
    {
        // we found an empty slot, stop here
        return false;
    }

    bool operator()(Done)
    {
        return moveToNextSlot(pCurrent);
    }

    bool operator()(Workload)
    {
        return moveToNextSlot(pCurrent);
    }

private:
    int* pCurrent;
};

struct ConsumerTaskVisitor
{
    ConsumerTaskVisitor(int mySlot) : mySlot(mySlot) {}

    bool operator()(NoTask)
    {
        sems[mySlot].Wait();
        return true;
    }

    bool operator()(Done)
    {
        sems[mySlot].Wait();
        return false;
    }

    bool operator()(Workload w)
    {
        std::cout << w.value << std::endl;
        slots[mySlot].store(NoTask{});
        return true;
    }

private:
    int mySlot;
};

void producerTask()
{
    int limit = 100;
    int current = 0;

    // Fill slots with workloads.
    for (int i = 0; i < limit; i++)
    {
        bool keepGoing = true;
        while (keepGoing)
        {
            keepGoing = std::visit(ProducerTaskVisitor{ &current }, slots[current].load());
        }
        slots[current].store(Workload{ i });
        sems[current].Signal();
    }

    // We're done. Fill all slots with Done
    int notified = 0;
    current = 0;
    while (notified < SLOTS)
    {
        bool keepGoing = true;
        while (keepGoing)
        {
            keepGoing = std::visit(ProducerTaskVisitor{ &current }, slots[current].load());
        }
        slots[current].store(Done{});
        sems[current].Signal();
        notified++;
    }
}

void consumerTask(int mySlot)
{
    bool keepGoing = true;
    while (keepGoing)
    {
        keepGoing = std::visit(ConsumerTaskVisitor{ mySlot }, slots[mySlot].load());
    }
}

int main()
{
    vector<thread> threads;
    threads.push_back(thread{ producerTask });
    for (int i = 0; i < SLOTS; i++)
    {
        threads.push_back(thread{ consumerTask, i });
    }

    for (auto&& t : threads)
    {
        t.join();
    }
}