Non blocking single producer/consumer queue with C++11 atomics

Question

I've recently got interested in lockless programming, mainly lockless queues for job repartitions, so here I undertook a challenge of creating lockless queue for single producer and consumer.

This is a queue with fixed size ring buffer.

Empty -> head = -1 and maybe tail = -1

[_][_][x][x][x][_][_]
       ^     ^
     head   tail

Sometimes front() (or pop()) skip a value and for the rest of execution I get sequentially offset values.

Old code full of bugs:

#pragma once
#ifndef ASYNC_SINGLE_FIFO_H
#define ASYNC_SINGLE_FIFO_H

#include <atomic>
#include <vector>
#include <condition_variable>
#include <thread>

using namespace std;

template<typename T>
class async_single_fifo
{
public:
    async_single_fifo(size_t n);

    bool inline empty();
    bool inline full();

    T front();
    bool try_front(T&);
    bool pop();
    void push(T);
    bool try_push(T);
    int size();

protected:

    int next(int);
    atomic<int> head_;
    atomic<int> tail_;
    vector<T> data;
    int size_;
};


template<typename T> async_single_fifo<T>::async_single_fifo(size_t n)
{
    head_ = -1;
    tail_ = -1;
    size_ = n;
    data = vector<T>(size_);
}

template<typename T>
bool async_single_fifo<T>::empty()
{
    return head_.load(memory_order_acquire) == -1;
}

template<typename T>
bool async_single_fifo<T>::full()
{
    return next(tail_.load(memory_order_acquire)) == head_.load(memory_order_acquire);
}

template<typename T>
T async_single_fifo<T>::front()
{
    while (empty())
        this_thread::yield();
    T res = data[head_.load(memory_order_acquire)];
    return res;
}

template<typename T>
inline bool async_single_fifo<T>::try_front(T &res)
{
    if (empty())
        return false;
    res = data[head_.load(memory_order_acquire)];
    return true;
}



template<typename T>
bool async_single_fifo<T>::pop()
{
    auto head = head_.load(memory_order_acquire);
    auto tail = tail_.load(memory_order_acquire);
    if (head == tail)// queue will be empty after pop
    {
        if (tail == -1) // queue was empty from beginning
            return false;

        if (tail_.compare_exchange_strong(tail, -1, memory_order_seq_cst)) // try to set tail to -1 if new item hadn't been added
        {
            head_.store(-1, memory_order_release); //tail modified, we can safely reset head
            return true;
        }
    }
    // otherwise queue isn't empty
    head_.store(next(head), memory_order_release);
    return true;
}

template<typename T>
inline void async_single_fifo<T>::push(T _arg)
{
    while (full()) // queue full case, wait for next element
        this_thread::yield();
    auto tail = tail_.load(memory_order_acquire);
    auto head = head_.load(memory_order_acquire);
    int new_tail;
    if (tail == -1)
        new_tail = 0;
    else
        new_tail = next(tail); // otherwise store in next cell

    data[new_tail] = _arg;
    tail_.store(new_tail, memory_order_release);
    head = -1;
    head_.compare_exchange_strong(head, tail, memory_order_seq_cst);
}

template<typename T>
bool async_single_fifo<T>::try_push(T _arg)
{
    auto tail = tail_.load(memory_order_acquire);
    auto head = head_.load(memory_order_acquire);
    if (next(tail) == head) // queue full case, do nothing
        return false;

    int new_tail;
    if (tail == -1)
        new_tail = 0;
    else
        new_tail = next(tail); // otherwise store in next cell

    data[new_tail] = _arg;
    tail_.store(new_tail, memory_order_release);
    head = -1;
    head_.compare_exchange_strong(head, tail, memory_order_seq_cst);
    return true;
}

template<typename T>
int async_single_fifo<T>::size()
{
    auto head = head_.load(memory_order_acquire);
    auto tail = tail_.load(memory_order_acquire);
    if (head > tail)
        tail += size;
    return tail - head;
}


template<typename T>
int async_single_fifo<T>::next(int i)
{
    auto res = i == size_ - 1 ? 0 : i + 1;
    return res;
}


#endif //ASYNC_SINGLE_FIFO_H

New code with better circular buffer logic

/*
    [_][_][_][_] empty
     ^head_tail

    [x][x][_][_] partially filled
 head^     ^tail

    [x][x][_][x] full
       tail^  ^head

    1 cell is always empty to not to create ambiguity
    like tail=head meaning both full and empty
*/

#pragma once
#ifndef ASYNC_SINGLE_FIFO_H
#define ASYNC_SINGLE_FIFO_H

#include <atomic>
#include <vector>
#include <condition_variable>
#include <thread>

using namespace std;

template<typename T>
class async_single_fifo
{
public:
    async_single_fifo(size_t n);

    bool inline empty();
    bool inline full();

    T front();
    bool try_front(T&);
    bool pop();
    bool try_push(T);
    void push(T);
    size_t size();

protected:

    size_t next(size_t);
    atomic<size_t> head_;
    atomic<size_t> tail_;
    vector<T> data;
    size_t size_;
};


template<typename T> async_single_fifo<T>::async_single_fifo(size_t n)
{
    head_ = 0;
    tail_ = 0;
    size_ = n+1;
    data = vector<T>(size_);
}

template<typename T>
bool async_single_fifo<T>::empty()
{
    return tail_.load(memory_order_acquire) == head_.load(memory_order_acquire);
}

template<typename T>
bool async_single_fifo<T>::full()
{
    return next(tail_.load(memory_order_acquire)) == head_.load(memory_order_acquire);
}

template<typename T>
T async_single_fifo<T>::front()
{
    while (empty())
        this_thread::yield();
    T res = data[head_.load(memory_order_acquire)];
    return res;
}

template<typename T>
inline bool async_single_fifo<T>::try_front(T &res)
{
    if (empty())
        return false;
    res = data[head_.load(memory_order_acquire)];
    return true;
}



template<typename T>
bool async_single_fifo<T>::pop()
{
    size_t head = head_.load(memory_order_acquire);
    size_t tail = tail_.load(memory_order_acquire);
    if (head == tail)
        return false;
    size_t new_head = next(head);
    head_.store(new_head, memory_order_release);
    return true;
}

template<typename T>
bool async_single_fifo<T>::try_push(T _arg)
{
    size_t tail = tail_.load(memory_order_acquire);
    size_t head = head_.load(memory_order_acquire);
    size_t new_tail = next(tail);
    if (new_tail == head) // queue full , do nothing
        return false;
    data[tail] = _arg;
    tail_.store(new_tail, memory_order_release);
    return true;
}

template<typename T>
inline void async_single_fifo<T>::push(T _arg)
{
    while (full()) // queue full case, wait for next element
        this_thread::yield();
    auto tail = tail_.load(memory_order_acquire);
    auto head = head_.load(memory_order_acquire);
    size_t new_tail = next(tail);
    if (new_tail == head) // queue full , do nothing
        return;
    data[tail] = _arg;
    tail_.store(new_tail, memory_order_release);
}

template<typename T>
size_t async_single_fifo<T>::size()
{
    auto head = head_.load(memory_order_acquire);
    auto tail = tail_.load(memory_order_acquire);
    if (head > tail)
        tail += size_;
    return tail - head;
}


template<typename T>
size_t async_single_fifo<T>::next(size_t i)
{
    size_t res = i == size_ - 1 ? 0 : i + 1;
    return res;
}


#endif //ASYNC_SINGLE_FIFO_H

I am mainly interested in feedback related to queue logic, but I would like to hear code-style and template design critique as well.

Answer 1

Bug

Unfortunately your code is not really close to working properly. I will point out the first bug I thought of but there are definitely many more. Consider the following situation:

Initial state:
Queue has one element in it
head = 0
tail = 0

Sequence of events, P = producer, C = consumer:
P calls push()
P stores data item in data[1] but doesn't yet modify tail
C calls pop()
C sets tail to -1
C sets head to -1
P sets tail to 1

End state:
Queue is in illegal state
head = -1
tail = 1

Suggestion

I have a suggestion that will greatly simplify things. Make a rule that only the producer can ever modify tail and only the consumer can ever modify head. By doing this, you avoid a whole class of problems dealing with simultaneous modification of the same variable.