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

Question

My header file implementation:

#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 current_back = back.load(std::memory_order_relaxed);

    const auto next_back = increment(current_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;
}

template<typename Data, size_t Size>
bool CircularSpsc<Data, Size>::pop_front(Data& item)
{
    const auto currentfront = front.load(std::memory_order_relaxed);
    if(currentfront == back.load(std::memory_order_acquire))
        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
{
    return (front.load() == back.load());
}

template<typename Data, size_t Size>
bool CircularSpsc<Data, Size>::isFull() const
{
    const auto nextback = increment(back.load());
    return (nextback == front.load());
}

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 <thread>
#include <vector>
#include "circularspsc.hpp"
#include "dataobject.hpp"

#define QUEUESIZE 10000

void reader(CircularSpsc<int, QUEUESIZE>& cspsc_q)
{
    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;

    std::thread writer_thread(writer, std::ref(cspsc_q));
    std::thread reader_thread(reader, std::ref(cspsc_q));

    reader_thread.join();
    writer_thread.join();
}

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.

Answer 1

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.

Answer 2

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... :)

Answer 3

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.