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.
push
; thread C is allowed to callpop
. Who (if anyone) is allowed to callfront
,try_front
,empty
, andfull
? Also, what are the intended semantics ofsize
? \$\endgroup\$push
function are obviously wrong. Consider what happens if thread P tries topush
on a full queue, and yields, and during that yielded timeslice thread C callspop
enough times that the queue goes from "full" to "empty". Boom. \$\endgroup\$push
and non-blockingtry_push
. C is allowed to callfront
, non-blockingtry_front
to get value andpop
to remove it.pop
is always non-blocking.empty
andfull
theoretically can be called by anyone.size
gives the estimate of number of element currently contained. \$\endgroup\$