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.

• Quuxplusone: I guarantee you've got bugs; but in order to find them I'll have to know what you mean by "producer" and "consumer". You have two threads P and C. Thread P is allowed to call push; thread C is allowed to call pop. Who (if anyone) is allowed to call front, try_front, empty, and full? Also, what are the intended semantics of size?
• Quuxplusone: Actually, the first four lines of your push function are obviously wrong. Consider what happens if thread P tries to push on a full queue, and yields, and during that yielded timeslice thread C calls pop enough times that the queue goes from "full" to "empty". Boom.
• FanciestBanana: P is allowed to call push and non-blocking try_push. C is allowed to call front, non-blocking try_front to get value and pop to remove it. pop is always non-blocking. empty and full theoretically can be called by anyone. size gives the estimate of number of element currently contained.

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.

• FanciestBanana: I do set head at the end of push, so your bug is not present; i do however always set it to 0 even if my tail may be > 0, making consumer re-read old cells.
• FanciestBanana: I've considered you advice on restricting write operations to the respective methods. In my case i needed to modify both because of the way i decided to mark empty queue (head = tail = -1), so i will rewrite code with different convention.
• JS1: @FanciestBanana You only set head at the end of push if it were -1. However in the scenario I presented, it was loaded as 0 and never reloaded. Therefore, the line that says if (head == -1) will be evaluated as false. Also, please don't revise the code after being answered. That is not how this site works.
• FanciestBanana: I've completely rewritten the code with your suggestion about not modifying head in producer and tail in consumer.