My program is structured like this: there are M producer threads, each of which computes a section of an object O with index i O[i]. When a specific O[i] is completed by all producers, it is put in a ring buffer, which is consumed by N threads. When all consumers are done with a specific O[i] can be discarded.

There must be no intra-group synchronization, that is producers need not to synchronize work on a single O[i], as well as consumers. The objects must appear in order in the ring though. Also, the speed of production and consumption can vary wildly, so a consumer that cannot acquire an element from the ring must block, and if a producer is about to start working on an element but the ring is full, it shall block. In all the other cases, insertion and removal should be lock free.

Here's what I came up with C++11. I seek for advice especially on the usage of the various relaxed memory models. I have checked the assembly and my code seems not to produce any mfence instruction on x86, only some lock when incrementing reference counting. Is that correct and expected?

#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_ */

An actual usage with two queues can be found here (the schema is main thread => leviqueue => N * producer => numthreads * consumer, where N is std::thread::hardware_concurrency()).

The class doesn't try to manage the actual objects on the behalf of the user, but just handles the indexes i of each object.

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.

Each producer has a thread local index in the ring, and must acquire and release it, with acquire_producer(), release_producer(), acquire_consumer(), release_consumer(). EOF is signaled by the producer with the eof() member function (which is an acquire - release operation), and when a consumer encounters EOF, acquire_consumer() returns -1.