# A concurrent queue that supports exactly two threads and not more. One thread is allowed to push, the other to pop

#ifndef CIRCULARSPSC_H
#define CIRCULARSPSC_H

#include <atomic>
#include <cstddef>

template<typename Data, size_t Size>
class CircularSpsc
{
public:
enum { queue_size = Size + 1 };

CircularSpsc() : back(0), front(0){}
virtual ~CircularSpsc() {}

bool push_back(const Data& item)    ;
bool pop_front(Data& item);

bool isEmpty() const;
bool isFull() const;

private:
size_t increment(size_t idx) const;
std::atomic<size_t> back;
Data dataArray[queue_size];
std::atomic<size_t> front;
};

#include "circularspsc.cpp"

#endif


My source file implementation:

#ifndef CIRCULARSPSC_CPP
#define CIRCULARSPSC_CPP

#include "circularspsc.hpp"

template<typename Data, size_t Size>
bool CircularSpsc<Data, Size>::push_back(const Data& item)
{

const auto next_back = increment(current_back);
{
dataArray[current_back] = item;
back.store(next_back, std::memory_order_release);
return true;
}
return false;
}

template<typename Data, size_t Size>
bool CircularSpsc<Data, Size>::pop_front(Data& item)
{
return false;

item = dataArray[currentfront];
front.store(increment(currentfront), std::memory_order_release);
return true;
}

template<typename Data, size_t Size>
bool CircularSpsc<Data, Size>::isEmpty() const
{
}

template<typename Data, size_t Size>
bool CircularSpsc<Data, Size>::isFull() const
{
}

template<typename Data, size_t Size>
size_t CircularSpsc<Data, Size>::increment(size_t idx) const
{
return (idx + 1) % queue_size;
}

#endif


Client file implementation:

#include <iostream>
#include <vector>
#include "circularspsc.hpp"
#include "dataobject.hpp"

#define QUEUESIZE 10000

{
for(int i = 0; i < QUEUESIZE; i++)
{
int dataItem;
cspsc_q.pop_front(dataItem);
std::cout << "popped value: " << dataItem << std::endl;
}
}

void writer(CircularSpsc<int, QUEUESIZE>& cspsc_q)
{
for(int i = 0; i < QUEUESIZE; i++)
{
cspsc_q.push_back(i);
}
}

int main()
{
CircularSpsc<int, QUEUESIZE> cspsc_q;

}


This has the clear drawback that we can only create queues of very limited size (because with larger values of QUEUESIZE I am getting segfault (surely because the stack size is limited?)). In my next iteration I plan to implement a version with dynamic memory allocation inside the constructor of the queue for QUEUESIZE number of objects and then use placement new to move values inside the queue.

• I would take a look at this blog - these queues achieve hundreds of millions of op/s, the algorithms are far more advanced than those here; so it might we very informative. Jan 10, 2018 at 16:32

### Naming

I don't understand what CircularSpsc is supposed to mean, and how to even pronounce it (circular space?). I suggest you use a full name without omitting the vowels.

### Include guards

You were right to add an include guard to the header file (.hpp). It is supposed to prevent the header file to be included more than once in a single translation unit, which could cause "multiple definition" errors.

However, in the implementation file (.cpp) an include guard is not needed, because this file is not supposed to be included at all.

The implementation file is in fact included at the end of the header file, but this is not the right way to get it compiled. Instead, you should pass its name directly to the compiler in addition to the "main" .cpp file.

You can read more about how compilation works in C++ here, for example.

• SPSC was meant to convey Single Producer Single Consumer. I see your point about the naming though. It is certainly not clear. Jan 10, 2018 at 15:26
• I had to add the header guard in the cpp file because since I wanted to put the definitions of the templated class member functions in a separate source file, I was forced to add #include "circularspsc.cpp" inside the header file (at the end). This led to multiple definition errors for the cpp file. Because of which I then added header guards to the cpp file as well. I know this is not pretty. Jan 10, 2018 at 15:30
• @Chani, note that you can implement your methods directly in the header file. There is no need to create a .cpp file for every .h file. There are actually a lot of benefits to keeping all code for your class together in one file. Many C++ libraries are implemented as "header-only libraries", they don't have any .cpp files at all. Including your .cpp file in your header file is going to confuse your users! Jan 10, 2018 at 16:01
• Several organizations use .tcc, .icc, or .inl suffix for template instantiation header files. stackoverflow.com/questions/29264656/… Jan 10, 2018 at 17:46
• @MooingDuck I didn't know about those file types! Thank you so much. Jan 11, 2018 at 16:26

I don't know a whole lot about concurrency, I thought this was a pretty cool idea.

A minor style thing, that I think helps readability:

In code like

if X {
code path 1;
return;
}
code path 2;
return;


I find it beneficial to put the shortest of the code paths as "code path 1". This is especially true if one of the code paths is empty (i.e. an early return). In the case of your function CircularSpsc::push_back:

if(next_back != front.load(std::memory_order_acquire)) {
dataArray[current_back] = item;
back.store(next_back, std::memory_order_release);
return true;
}
return false;


I'd prefer to write:

if(next_back == front.load(std::memory_order_acquire)) {
return false;
}
dataArray[current_back] = item;
back.store(next_back, std::memory_order_release);
return true;


You do it this way in CircularSpsc::pop_front, which looks a lot clearer to me.

I would also recommend that you always use braces after if, even if there is a single statement there. This will avoid a lot of confusion and difficult to find bugs when someone comes along and adds a statement to the if code but forgets to add the braces. I've seen it happen multiple times... :)

You should always follow two rules:

1. use the standard library whenever you can
2. given the choice between high-level and low-level, choose high-level.

Building your queue upon an array or, worse, a custom dynamic memory container, breaks both rules. If you want to build a concurrent queue, build it upon std::queue and focus on the thread safety issues, that's enough already on your plate.

Choosing std::atomic as a synchronization primitive breaks only the second rule, but it's enough to derail your implementation. Unless I'm mistaken, is_empty doesn't synchronize its reading of two different atomic variables, meaning front may have changed when you read back.

I believe your attempt was bound to run into difficulties regardless of those two first points, because you thought in terms of implementation and not of interface. Queue is a renowned example in research about exception and thread safety; you should look it up. To put it in a nutshell, if you want thread safety you have to minimize the available operations; signatures for pop and frontare enough, whereas synchronizing is_empty is a nightmare, and an unnecessary one to that. If you wonder how you can get rid of is_empty without running into various bounds violations, it's time to get back to the standard library once more: std::condition_variable is what you're looking for.