CircularBuffer.cpp

// CircularBuffer.cpp by AlSch092 @ Github
// Includes a lock-free circular buffer class + a multi-producer/single consumer class (uses a lock) which can push events produced by multiple thread channels to disc
// Also includes some test code in main()

#include <iostream>
#include <atomic>
#include <thread>
#include <chrono>
#include <unordered_map>
#include <mutex>
#include <sstream>
#include <fstream>

template <typename T, std::size_t N>
class CircularBuffer
{
private:
    static_assert((N& (N - 1)) == 0, "N must be power of 2");
    static constexpr size_t capacity = N;
    static constexpr size_t Mask = N - 1;

    std::aligned_storage_t<sizeof(T), alignof(T)> storage[N];

    alignas(64) std::atomic<size_t> ReadIndex{ 0 }; //head
    char _pad1[64 - sizeof(ReadIndex)]{};
    alignas(64) std::atomic<size_t> WriteIndex{ 0 }; //tail
    char _pad2[64 - sizeof(WriteIndex)]{};

public:

    bool Head(T& out)
    {
        size_t Writeindx = WriteIndex.load(std::memory_order_acquire);
        size_t Readindx = ReadIndex.load(std::memory_order_relaxed);
        if (Readindx == Writeindx)
            return false;

        T* p = std::launder(reinterpret_cast<const T*>(&storage[Readindx & Mask])); //Memory laundering is used to prevent the compiler from tracing where you got your object from, thus forcing it to avoid any optimizations that may no longer apply.
        out = *p;
        return true;
    }

    bool Tail(T& out)
    {
        size_t Writeindx = WriteIndex.load(std::memory_order_acquire);
        size_t Readindx = ReadIndex.load(std::memory_order_relaxed);
        if (Readindx == Writeindx)
            return false;

        T* p = std::launder(reinterpret_cast<const T*>(&storage[Writeindx & Mask]));
        out = *p;
        return true;
    }

    bool TryPush(T&& val) //put onto tail (producer)
    {
        size_t Writeindx = WriteIndex.load(std::memory_order_relaxed); 
        size_t ReadIndx = ReadIndex.load(std::memory_order_acquire);

        if (Writeindx - ReadIndx >= capacity) //full
        {
            return false;
        }

        ::new(&storage[Writeindx & Mask]) T(std::move(val));
        WriteIndex.store(Writeindx + 1, std::memory_order_release); //publish
        return true;
    }

    bool TryPop(T& val) //take from head (consumer)
    {
        size_t ReadIndx = ReadIndex.load(std::memory_order_relaxed);
        size_t Writeindx = WriteIndex.load(std::memory_order_acquire);

        if (ReadIndx == Writeindx) //empty
        {
            return false;
        }

        auto p = std::launder(reinterpret_cast<T*>(&storage[ReadIndx & Mask]));
        val = std::move(*p);
        p->~T();
        ReadIndex.store(ReadIndx + 1, std::memory_order_release); //publish
        return true;
    }

    bool Empty() const
    {
        return (ReadIndex.load(std::memory_order_acquire) == WriteIndex.load(std::memory_order_acquire));
    }

    bool Full() const
    {
        return (WriteIndex.load(std::memory_order_acquire) - ReadIndex.load(std::memory_order_acquire) >= capacity);
    }
};


template<typename T, size_t N>
class MPSC //multiple thread producer channels, single consumer -> each thread is identified as a producer, and events are combined from all threads when drained
{
private:

    std::mutex ChannelMutex;
    std::unordered_map<std::thread::id, std::unique_ptr<CircularBuffer<T, N>>> Channels;

    std::stringstream ss;
    std::ofstream ofs;

public:

    CircularBuffer<T,N>* GetChannel()
    {
        std::lock_guard<std::mutex> lock(ChannelMutex);

        auto& channel = Channels[std::this_thread::get_id()];

        if (channel == nullptr)
            channel = std::make_unique<CircularBuffer<T, N>>();

        return channel.get();
    }

    bool DrainVals(__in const std::string& fileName) //combine all thread channels and push their data to a file on disk
    {
        if (fileName.empty())
            return false;

        this->ofs = std::ofstream(fileName);

        std::unique_lock<std::mutex> lock(ChannelMutex);
        
        for (auto& channel : Channels)
        {
            while (!channel.second->Empty())
            {
                int val = 0;
                channel.second->TryPop(val);
                ss << val << std::endl;
            }
        }

        ofs << ss.str();
        ofs.close();
        return true;
    }

};


int main()
{
    //basic example: queue on its own
    std::unique_ptr<CircularBuffer<float, 256>> circBuffer = std::make_unique<CircularBuffer<float, 256>>();

    for (int i = 0; i < 512; i++)
    {
        if (!circBuffer->TryPush((float)(i + 0.125)))
        {
            float f = 0.0;
            if (circBuffer->TryPop(f)) //since buffer size = 256 and loop size = buffer size * 2, popped items = buffer size / (buffersize/loop size) / 2 = 128 items popped
            {
                std::cout << "Buffer was full -> Popped item: " << f << std::endl;
            }
        } //in cases where you need to push and the buffer is full, you can re-try if some other spot in code pops/drains from the buffer
    }

    //threaded examples using thread channel in MPSC:

    auto begin = std::chrono::steady_clock::now();

    std::unique_ptr<MPSC<int, 128>> manager = std::make_unique<MPSC<int, 128>>();

    std::thread t([&manager]()
        {
            CircularBuffer<int, 128>* buff = manager->GetChannel();

            if (!buff)
                return;

            for (int i = 0; i < 128; i++)
            {
                if (!buff->Full())
                {
                    if (!buff->TryPush(i + 1)) //using i by itself will require std::move(i) due to needing an r-value reference
                        i--;
                }
                else //if you increase the loop beyond 128, youll need to pop items from the queue, otherwise this will loop forever
                    i--;
            }
        });

    std::thread p([&manager]()
        {
            CircularBuffer<int, 128>* buff = manager->GetChannel();

            if (!buff)
                return;

            for (int i = 0; i < 256; i++)
            {
                int i_cpy = i; //suppress warnings about std::move on i

                if (!buff->Full())
                {
                    if (!buff->TryPush(std::move(i_cpy)))
                        i--;
                }
                else
                    while (!buff->Empty()) //drain the buffer if full
                    {
                        int val = -1;
                        if (buff->TryPop(val))
                        {
                            std::cout << "Popped: " << val << std::endl;
                        }
                    }
            }
        });

    if (t.joinable())
        t.join();
    if (p.joinable())
        p.join();

    if (!manager->DrainVals("vals.txt"))
    {
        std::cout << "Failed to drain queues!\n";
    }

    auto end = std::chrono::steady_clock::now();

    std::cout << "Ticks (nanoseconds): " << end - begin << std::endl;
    return 0;
}