Return to Question

#ifndef MULTIQUEUE_HPP_
#define MULTIQUEUE_HPP_

#include <atomic>
#include <condition_variable>
#include <stdexcept>

template<size_t cachealign = 64>
class multiqueue_cacheline{

    const int producers, consumers;
    const size_t ringSize, unblockProducerThreshold, unblockConsumerThreshold;

    struct reference{
        std::atomic<int> count;
        bool eof;
        char padding[cachealign > sizeof(count) + sizeof(eof) ? cachealign - sizeof(count) - sizeof(eof) : 0];
        reference(int v): count(v), eof(false) {}
    };
    static_assert(cachealign, "cachealign must be non zero.");

    reference* refs;
    void* unalignedRing = 0;
    std::atomic<size_t> inRing = {0};
    struct blocker{
        std::condition_variable condition;
        std::mutex mutex;
        template<typename T>
        void sleep(T& ready){
            std::unique_lock<std::mutex> lk(mutex);
            if(!ready()) condition.wait(lk, ready);
        }
        void wake(){
            std::unique_lock<std::mutex> lk(mutex);
            condition.notify_all();
        }
    } consumerLock, producerLock;

public:

    multiqueue_cacheline(unsigned int producers, unsigned int consumers, size_t ringSize, size_t unblockProducerThreshold = 0, size_t unblockConsumerThreshold = 0)
    : producers(producers), consumers(consumers), ringSize(ringSize), unblockProducerThreshold(unblockProducerThreshold ? unblockProducerThreshold : ringSize / 5), unblockConsumerThreshold(unblockConsumerThreshold ? unblockConsumerThreshold : (ringSize * 4) / 5)
    {
        if(!producers || !consumers || !(ringSize > this->unblockConsumerThreshold && this->unblockConsumerThreshold > this->unblockProducerThreshold && this->unblockProducerThreshold > 0))
            throw std::invalid_argument("Bad ring parameters.");
        unalignedRing = malloc(sizeof(reference[ringSize]) + cachealign - 1);
        if(!unalignedRing) throw std::bad_alloc();
        refs = reinterpret_cast<reference*>((size_t(unalignedRing) + cachealign - 1) & ~(cachealign - 1));
        for(size_t i = 0; i < ringSize; i++) new(refs + i) reference(producers);
        std::atomic_thread_fence(std::memory_order_release);
    }

    ~multiqueue_cacheline(){
        free(unalignedRing);
    }

    ssize_t acquire_consumer(size_t i){
        auto ready = [=]{ return refs[i].count.load(std::memory_order_acquire) <= 0 || refs[i].eof; };
        if(!ready()) consumerLock.sleep(ready);
        if(refs[i].eof){
            producerLock.wake();
            return -1;
        }
        return i;
    };

    void release_consumer(size_t& i){
        if(refs[i].count.fetch_sub(1, std::memory_order_acq_rel) == 1 - consumers){
            refs[i].count.store(producers, std::memory_order_release);
            if(inRing.fetch_sub(1, std::memory_order_relaxed) == unblockProducerThreshold + 1) producerLock.wake();
        }
        i = (i + 1) % ringSize;
    };

    size_t acquire_producer(size_t i){
        auto ready = [=]{ return refs[i].count.load(std::memory_order_acquire) > 0; };
        if(!ready()) producerLock.sleep(ready);
        return i;
    };

    void release_producer(size_t& i){
        if(refs[i].count.fetch_sub(1, std::memory_order_acq_rel) == 1 && inRing.fetch_add(1, std::memory_order_relaxed) == unblockConsumerThreshold - 1)
            consumerLock.wake();
        i = (i + 1) % ringSize;
    };

    void eof(size_t i){
        acquire_producer(i);
        refs[i].eof = true;
        release_producer(i);
        consumerLock.wake();
    };

};

using multiqueue = multiqueue_cacheline<>;
using multiqueue_nocachealign = multiqueue_cacheline<1>;

#endif /* MULTIQUEUE_HPP_ */

The queue constructor is multiqueue::multiqueue(unsigned int producers, unsigned int consumers, size_t ringsize, size_t unblockProducerThreshold = 0, size_t unblockConsumerThreshold = 0), where besides the obvious parameters

• unblockProducerThreshold indicates when the ring is "almost empty", so if too fast producers were previously blocked they should be awaken, and
• unblockConsumerThreshold the opposite situation in case that consumers were blocked for starvation.

#ifndef MULTIQUEUE_HPP_
#define MULTIQUEUE_HPP_

#include <atomic>
#include <condition_variable>
#include <stdexcept>
#include <vector>

template<size_t linesize = 64>
class multiqueue_cacheline{

    const int inthread, outthread;
    const size_t ringsize, unblockproducer, unblockconsumer;

    struct ref{
        std::atomic<int> v;
        bool eof;
        char padding[linesize > sizeof(v) + sizeof(eof) ? linesize - sizeof(v) - sizeof(eof) : 0];
        ref(int v): v(v), eof(false) {}
        ref(const ref& o): v(o.v.load(std::memory_order_relaxed)), eof(false) {}
    };

    std::vector<ref> refs;
    std::atomic<size_t> inring = {0};
    struct sleepobj{
        std::condition_variable cond;
        std::mutex m;
        template<typename T>
        void sleep(T& ready){
            std::unique_lock<std::mutex> lk(m);
            if(!ready()) cond.wait(lk, ready);
        }
        void wake(){
            std::unique_lock<std::mutex> lk(m);
            cond.notify_all();
        }
    } consumersleep, producersleep;

public:

    multiqueue_cacheline(unsigned int inthread, unsigned int outthread, size_t ringsize, size_t unblockproducer = 0, size_t unblockconsumer = 0)
    : inthread(inthread), outthread(outthread), ringsize(ringsize), unblockproducer(unblockproducer ? unblockproducer : ringsize / 5), unblockconsumer(unblockconsumer ? unblockconsumer : (ringsize * 4) / 5)
    {
        if(!inthread || !outthread || !(ringsize > this->unblockconsumer && this->unblockconsumer > this->unblockproducer && this->unblockproducer > 0))
            throw std::invalid_argument("Bad ring parameters.");
        refs = std::vector<ref>(ringsize, {int(inthread)});
        std::atomic_thread_fence(std::memory_order_release);
    }

    ssize_t acquire_consumer(size_t i){
        auto ready = [=]{ return refs[i].v.load(std::memory_order_acquire) <= 0 || refs[i].eof; };
        if(!ready()) consumersleep.sleep(ready);
        if(refs[i].eof){
            producersleep.wake();
            return -1;
        }
        return i;
    };

    void release_consumer(size_t& i){
        if(refs[i].v.fetch_sub(1, std::memory_order_acq_rel) == 1 - outthread){
            refs[i].v.store(inthread, std::memory_order_release);
            if(inring.fetch_sub(1, std::memory_order_relaxed) == unblockproducer + 1) producersleep.wake();
        }
        i = (i + 1) % ringsize;
    };

    size_t acquire_producer(size_t i){
        auto ready = [=]{ return refs[i].v.load(std::memory_order_acquire) > 0; };
        if(!ready()) producersleep.sleep(ready);
        return i;
    };

    void release_producer(size_t& i){
        if(refs[i].v.fetch_sub(1, std::memory_order_acq_rel) == 1 && inring.fetch_add(1, std::memory_order_relaxed) == unblockconsumer - 1)
            consumersleep.wake();
        i = (i + 1) % ringsize;
    };

    void eof(size_t i){
        acquire_producer(i);
        refs[i].eof = true;
        release_producer(i);
        consumersleep.wake();
    };

};

using multiqueue = multiqueue_cacheline<>;
using multiqueue_nocacheline = multiqueue_cacheline<0>;

#endif /* MULTIQUEUE_HPP_ */

The queue constructor is multiqueue::multiqueue(unsigned int inthread, unsigned int outthread, size_t ringsize, size_t unblockproducer = 0, size_t unblockconsumer = 0), where besides the obvious parameters

• unblockproducer indicates when the ring is "almost empty", so if too fast producers were previously blocked they should be awaken, and
• unblockconsumer the opposite situation in case that consumers were blocked for starvation.