Lockless, blocking, non synchronized multiple producers and consumers ring buffer

Question

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.

Answer 1

1. You are trying to align each ref object to a cache line by padding the object to fill an entire cache line. However your refs vector is embedded as a member in the multiqueue class hence it will start at an offset relative to the beginning of the object which is probably either aligned on a 4 or 8 byte boundary. So your refs vector might start in the middle of a cache line and your padding to cache line size ill not gain you what you wanted.

   Aligning a member within a class to a cacheline seems to be a non trivial exercise. This question on SO might be helpful.

2. Your code is really hard to understand from just reading mostly due to not very helpful names for parameters to the class. It's not very obvious what the inthread, outthread, unblockconsumer and unblockproducer parameters mean and in how far they relate to the workings of the queue. I assume inthread is numberOfProducers and outthread is numberOfConsumers - the others are bit more convoluted. Adding some documentation comments would help.