A thread that blocks after the call to fetch_add() will block later producers and consumers that use the same slot. As objects are enqueued (and dequeued), the ring will wrap around and come back to earlier slots. For enqueue, such an event can be pushed far into the future as the queue capacity can be made arbitrarily large (as memory allows) but not for dequeue. Possibly blocking could be avoided if producers and consumers synchronise with each other (e.g. use CAS to update the slot and restart the operation (allocate new index) if the slot doesn't have the expected value.

Inspired by your try_consume PR I added a try_consume_until_current_head method that consumes all elements of the queue that were added before the current head.

My use case is a macro recorder. The queue captures commands from different threads. Then, when the user triggers the macro, all recorded commands are executed. However it's possible that while the commands belonging to the first recording are still being consumed, commands from a second recording are already being recorded, and they shouldn't be executed until next time the macro is triggered. With try_consume or try_pop I cannot guarantee this.

I'm not sure what the memory ordering should be for the head_.load at the beginning of the function. Can you help here?

I'm not sure what's you policy on new features, let me know if you think this doesn't belong to the project.

Hi

I tried a test pushing 1000000 times into the queue, and when the queue size is small like 64 or 1024, the latency is quite long. and when the queue size is big like 65535, the latency becomes low.

queue size 1024, push 10000 times, per use 0.3 us. queue size 1024, push 1000000 times, per use 13 us

    using namespace chrono;
   
    vector<thread> threads;
    rigtorp::MPMCQueue<int *> q(1024);
    atomic<int> push_times{0};
    atomic<int> pop_times{0};

    {
        steady_clock::time_point t1 = steady_clock::now();
        for(int i=0; i<Consumer_num; ++i) {
            threads.emplace_back(
                [&] {
                    while(pop_times.load()<test_size) {
                        int* temp;
                        q.pop(temp);
                        pop_times.fetch_add(1);
                    }
                }
            );
        }
        while(push_times<test_size) {
            int* temp;
            q.push(temp);
            push_times.fetch_add(1);
        }
        steady_clock::time_point t2 = steady_clock::now();
        duration<double, micro> time_span = (duration<double, micro>)(t2 - t1);

        this_thread::sleep_for(chrono::microseconds(500));
        cout<<"rigtorp::MPMCQueue: push " << test_size << " data, per cost us:" << (time_span).count()/test_size <<endl;
        if(push_times.load() != pop_times.load()) {
            cout<<"warning: push_times=" << push_times.load() << ", pop_times=" << pop_times.load() <<endl;
        }
    }
    exit(0);

For example, Linux open_shm, or MPI_win_allocate_shared, combined with placement new:

const int N = 10; // capacity
std::byte* buffer;
/* allocate buffer as shm on main proc*/
if (/*main proc*/) {
    new (buffer) MPMCQueue<int> q(N);
    // consume
} else {
    new (buffer) MPMCQueue<int> q(N);
   // produce
}

If given 'capacity' is < 1 you create a memory leak by throwing after memory allocation since '~MPMCQueue' will not be called. I suggest to move the check in a nested static member function called before 'slots_' initialization.

Hi, I am trying to use MPMCQueue with a custom allocator. But I am getting the two following errors:

/Path/To/Project/Debug/external/include/rigtorp/MPMCQueue.h:129:33: error: cannot convert ‘Message*’ to ‘rigtorp::mpmc::Slot<Message>*’ in assignment
  129 |     slots_ = allocator_.allocate(capacity_ + 1);
/Path/To/Project/Debug/external/include/rigtorp/MPMCQueue.h:134:29: error: cannot convert ‘rigtorp::mpmc::Slot<Message>*’ to ‘CommMemoryAllocator<Message>::pointer’ {aka ‘Message*’}
  134 |       allocator_.deallocate(slots_, capacity_ + 1);

This is how I define the allocator and pass it to MPMCQueue:

template <typename T>
class CommMemoryAllocator
{
    public:
        typedef size_t size_type;
        typedef ptrdiff_t difference_type;
        typedef T* pointer;
        typedef const T* const_pointer;
        typedef T& reference;
        typedef const T& const_reference;
        typedef T value_type;

        T* allocate(size_type num, const void* hint = nullptr);

        void deallocate(pointer ptr, size_type num);
};

template <typename T>
using comm_mqueue = rigtorp::MPMCQueue<T, CommMemoryAllocator<T>>;

Then inside one of my classes I actually use it like so:

comm_mqueue<Message> downward_queue{20};

I don't actually use the queues yet (so no pushing or popping), I am getting these errors directly from this simple use.

Is there something I am missing or is this a bug?I am using Linux, gcc/g++ 11.1.0, and this is being compiled as C++17.

I noticed that standard c++20 has add wait and notify function to atomic library occasionally:

• wait blocks the thread until notified and the atomic value changes

• notify_all notifies all threads blocked waiting on the atomic object

Thinking that there's an empty queue, one thread got blocked in a while(!is_it_changed) loop while popping an item, it would consume many cpu resources if there is no other thread(s) enqueue an item instantly, this case as same as pushing to a full queue.

In my opinion, the thread which got blocked while dequeueing from an empty queue (or enqueueing to a full queue) is better go for "sleep" after detecting whether changes has been made to it for several times, then if there are some thread enqueued (or dequeued), it will "wake-up" the sleeping thread(s) again to detect changes, I think using this form of change-detection is often more efficient than simple polling.

Through my poor implementation seems works and passed the MPMCQueueTest with no significant performance effect, but I may not considered all the different situations in synchronizing threads. Could you have some suggestion? Thanks!

Used ~30% CPU on wait in pop. Compiled in VS2017

sample code:

auto q = std::make_shared<rigtorp::MPMCQueue<int>>(10);
while(true) {
    int v;
    q->pop(v);
    switch(v) {
    case 4726:
        LOGV << "event=" << v << " user removed";
        break;
    case 4724:
        LOGV << "event=" << v << " user password manually reseted";
        break;
    default:
        LOGV << "event=" << v << " unknow";
        break;
    }
}

This is a very nice MPMC queue! Is it possible to add LIFO functionality? It looks as if pop() could be modified to use head_.fetch_sub(1) but I am uncertain if this breaks something?

According to the implementation.

Enqeue:

• Acquire next write ticket from head.
• Wait for our turn (true) to write slot (ticket % capacity).
• Set turn = false to inform the readers we are done writing.

Dequeue:

• Acquire next read ticket from tail.
• Wait for our turn (false) to read slot (ticket % capacity).
• Set turn = true to inform the writers we are done reading.

Dose this work?

Thank you.

I'm running into a compile issue using a struct containing a std::string as the type when constructing a new MPMCQueue.

For example, using a struct:

struct MyEvent {
	std::string MyString;
	long MyLong;
};

and trying to construct a MPMCQueue:

MPMCQueue<MyEvent> EventQueue(100);

I get a compiler message:

"T must be nothrow copy constructible".

If I remove the std::string from the struct and replace it with another long for example, it compiles and works perfectly. It also works perfectly if I use just a std::string in constructing the queue. But not when it's used inside the struct.

Any advice would be appreciated.

Hi, Erik: It's a great honor to see the lock free project you released on GitHub. The algorithm is very exquisite, and I've learned a lot. Thank you for your creation. In the code of MPMC queue, I found two points that can further improve the performance in the processing of array index and the number of turns. I was inspired by the kfifo queue in the Linux kernel code. And my brief description is as follows：

// first point:  
constexpr size_t idx(size_t i) const noexcept { 
    return i % capacity_;
}

// second point:
constexpr size_t turn(size_t i) const noexcept {
    return i / capacity_;
}

​ Every time you access the array index, you need a '%' operation, If the capacity of the queue is an integer power of 2, when the subscript is increasing in the ring queue, a "&" operator can be used to locate the specific array position, which should be a faster operation. Like this:

constexpr size_t idx(size_t i) const noexcept {
    return i & (capacity_ - 1);
}
/*
	if capacity is 8, i is 20 sometimes. Expressed in binary:
	---------------------
	 8:   0 0 0 0  1 0 0 0
   8-1:   0 0 0 0  0 1 1 1
	20:   0 0 0 1  0 1 0 0
	----------------------
	that, 20 & ( 8 - 1) is 4, same as 20 % 8. 
	      0 0 0 0  0 1 0 0  
*/

​ The processing of the number of turns is the same, Each time the turn of the queue loop is calculated, the division method needs to be used. If the capacity keeps the above characteristics, the bit operation can be used to replace the division operation every time the turn is calculated. The process is as follows:

// capacity_ is an integer power of 2，Logarithmic operation:
size_t power_of_capacity_ = ::log2(capacity_); 
constexpr size_t turn(size_t i) const noexcept {
    return  i >> power_of_capacity_;
}
/*
	if capacity is 8, i is 35 sometimes. Expressed in binary:
	---------------------
	 8:   0 0 0 0  1 0 0 0  (power_of_capacity: 3)
	35:   0 0 1 0  0 0 1 1
	----------------------
	that: 35 >> 3 is 4, same as 35 / 8. 
	      0 0 0 0  0 1 0 0
*/

​ The above two points are what I want to express. At the same time, I have carefully modified your code ^_^. I tried to remove the capacity variable and replace it with its mask value. At the same time, some modifications have been made to the constructor. If the passed capacity parameter is less than the integer power of 2, I will round it up. In this way, the assignment of capacity has to be removed from the formal parameter list, and the const attribute of the capacity variable in the class is also removed.

​ These changes will have an impact: the user's capacity parameter is likely not to be the actual number of memory space requests in the queue. Please evaluate.

​ I have passed the test program under the src directory with the above modifications, which may need to be improved in other aspects.

​ These are some of my considerations. I hope to see your guidance and reply.

​ Ps：The Linux kernel code version I read is : linux-4.4.293. Here is the location of the kfifo code, a lightweight SPSC queue using memory barrier.

kfifo.h:    linux-4.4.293\include\linux\kfifo.h
kfifo.c:    linux-4.4.293\lib\kfifo.c