// queue.h -- interface for a queue
|
#ifndef SLinkedLockFreeQueue_H_
|
#define SLinkedLockFreeQueue_H_
|
#include "mm.h"
|
#include "sem_util.h"
|
#include "SAbstractQueue.h"
|
|
|
template <typename T> class Node;
|
|
template <typename T>
|
class Pointer {
|
public:
|
|
Node<T> *ptr;
|
unsigned long count;
|
Pointer( Node<T> *node = NULL, int c=0) noexcept : ptr(node), count(c) {}
|
|
bool operator == (const Pointer<T> o) const {
|
return (o.ptr == ptr) && (o.count == count);
|
}
|
bool operator != (const Pointer<T> o) const {
|
return !((o.ptr == ptr) && (o.count == count));
|
}
|
|
|
|
};
|
|
template <typename T>
|
class Node {
|
public:
|
alignas(16) std::atomic<Pointer<T> > next;
|
T value;
|
|
Node() {
|
}
|
|
void *operator new(size_t size){
|
return mm_malloc(size);
|
}
|
|
void operator delete(void *p) {
|
return mm_free(p);
|
}
|
};
|
|
|
|
|
|
template <typename T>
|
class SLinkedLockFreeQueue
|
{
|
private:
|
// class scope definitions
|
enum {Q_SIZE = 10};
|
|
int slots;
|
int items;
|
// private class members
|
std::atomic<Pointer<T> > Head; // pointer to front of Queue
|
std::atomic<Pointer<T> > Tail; // pointer to rear of Queue
|
//std::atomic_uint count; // current number of size in Queue
|
std::atomic_uint count;
|
const size_t qsize; // maximum number of size in Queue
|
// preemptive definitions to prevent public copying
|
SLinkedLockFreeQueue(const SLinkedLockFreeQueue & q) : qsize(0) { }
|
SLinkedLockFreeQueue & operator=(const SLinkedLockFreeQueue & q) { return *this;}
|
bool _add(const T &item); // add item to end
|
bool _remove(T &item); // remove item from front
|
public:
|
SLinkedLockFreeQueue(size_t qs = Q_SIZE); // create queue with a qs limit
|
~SLinkedLockFreeQueue();
|
bool isempty() const;
|
bool isfull() const;
|
unsigned int size() const;
|
bool add(const T &item); // add item to end
|
bool add_nowait(const T &item);
|
bool add_timeout(const T &item, struct timespec *timeout);
|
bool remove(T &item);
|
bool remove_nowait(T &item);
|
bool remove_timeout(T &item, struct timespec * timeout);
|
|
|
T& operator[](unsigned i);
|
|
void *operator new(size_t size);
|
void operator delete(void *p);
|
};
|
|
|
|
|
// Queue methods
|
template <typename T>
|
SLinkedLockFreeQueue<T>::SLinkedLockFreeQueue(size_t qs) : count(0), qsize(qs)
|
{
|
Node<T> *node = new Node<T>;
|
Pointer<T> pointer(node, 0);
|
|
Head.store(pointer, std::memory_order_relaxed);
|
Tail.store(pointer, std::memory_order_relaxed);
|
|
slots = SemUtil::get(IPC_PRIVATE, qsize);
|
items = SemUtil::get(IPC_PRIVATE, 0);
|
|
}
|
|
template <typename T>
|
SLinkedLockFreeQueue<T>::~SLinkedLockFreeQueue()
|
{
|
std::cerr << "SLinkedLockFreeQueue destory" << std::endl;
|
SemUtil::remove(slots);
|
SemUtil::remove(items);
|
|
|
Node<T> * nodeptr;
|
Pointer<T> tmp = Head.load(std::memory_order_relaxed);
|
while((nodeptr = tmp.ptr) != NULL) {
|
tmp = (tmp.ptr->next).load(std::memory_order_relaxed);
|
//std::cerr << "delete " << nodeptr << std::endl;
|
delete nodeptr;
|
|
}
|
|
}
|
|
template <typename T>
|
bool SLinkedLockFreeQueue<T>::isempty() const
|
{
|
return count == 0;
|
}
|
|
template <typename T>
|
bool SLinkedLockFreeQueue<T>::isfull() const
|
{
|
return count == qsize;
|
}
|
|
template <typename T>
|
unsigned int SLinkedLockFreeQueue<T>::size() const
|
{
|
return count;
|
}
|
|
// Add item to queue
|
template <typename T>
|
bool SLinkedLockFreeQueue<T>::_add(const T & item)
|
{
|
if (isfull())
|
return false;
|
|
Node<T> * node = new Node<T>;
|
node->value = item;
|
|
|
Pointer<T> tail ;
|
Pointer<T> next ;
|
|
|
while(true) {
|
tail = Tail.load(std::memory_order_relaxed);
|
next = (tail.ptr->next).load(std::memory_order_relaxed);
|
if (tail == Tail.load(std::memory_order_relaxed)) {
|
if (next.ptr == NULL) {
|
if ((tail.ptr->next).compare_exchange_weak(next,
|
Pointer<T>(node, next.count+1),
|
std::memory_order_release,
|
std::memory_order_relaxed) )
|
break;
|
else
|
Tail.compare_exchange_weak(tail,
|
Pointer<T>(next.ptr, tail.count+1),
|
std::memory_order_release,
|
std::memory_order_relaxed);
|
}
|
|
}
|
}
|
|
Tail.compare_exchange_weak(tail, Pointer<T>(node, tail.count+1),
|
std::memory_order_release,
|
std::memory_order_relaxed);
|
count++;
|
return true;
|
}
|
|
template <typename T>
|
bool SLinkedLockFreeQueue<T>::add(const T & item)
|
{
|
if (SemUtil::dec(slots) == -1) {
|
err_exit(errno, "add");
|
}
|
|
if (SLinkedLockFreeQueue<T>::_add(item)) {
|
SemUtil::inc(items);
|
return true;
|
}
|
return false;
|
|
}
|
|
template <typename T>
|
bool SLinkedLockFreeQueue<T>::add_nowait(const T & item)
|
{
|
if (SemUtil::dec_nowait(slots) == -1) {
|
if (errno == EAGAIN)
|
return false;
|
else
|
err_exit(errno, "add_nowait");
|
}
|
|
if (SLinkedLockFreeQueue<T>::_add(item)) {
|
SemUtil::inc(items);
|
return true;
|
}
|
return false;
|
|
}
|
|
template <typename T>
|
bool SLinkedLockFreeQueue<T>::add_timeout(const T & item, struct timespec * timeout)
|
{
|
if (SemUtil::dec_timeout(slots, timeout) == -1) {
|
if (errno == EAGAIN)
|
return false;
|
else
|
err_exit(errno, "add_timeout");
|
}
|
|
if (SLinkedLockFreeQueue<T>::_add(item)){
|
SemUtil::inc(items);
|
return true;
|
}
|
return false;
|
|
}
|
|
|
// Place front item into item variable and remove from queue
|
template <typename T>
|
bool SLinkedLockFreeQueue<T>::_remove(T & item)
|
{
|
if (isempty())
|
return false;
|
|
Pointer<T> head;
|
Pointer<T> tail;
|
Pointer<T> next;
|
|
while(true) {
|
head = Head.load(std::memory_order_relaxed);
|
tail = Tail.load(std::memory_order_relaxed);
|
next = (head.ptr->next).load();
|
if (head == Head.load(std::memory_order_relaxed)) {
|
if(head.ptr == tail.ptr) {
|
if (next.ptr == NULL)
|
return false;
|
// Tail is falling behind. Try to advance it
|
Tail.compare_exchange_weak(tail,
|
Pointer<T>(next.ptr, tail.count+1),
|
std::memory_order_release,
|
std::memory_order_relaxed);
|
} else {
|
item = next.ptr->value;
|
if (Head.compare_exchange_weak(head,
|
Pointer<T>(next.ptr, head.count+1),
|
std::memory_order_release,
|
std::memory_order_relaxed)) {
|
delete head.ptr;
|
break;
|
}
|
|
}
|
}
|
|
}
|
|
count--;
|
return true;
|
|
}
|
|
template <typename T>
|
bool SLinkedLockFreeQueue<T>::remove(T & item)
|
{
|
if (SemUtil::dec(items) == -1) {
|
err_exit(errno, "remove");
|
}
|
|
if (SLinkedLockFreeQueue<T>::_remove(item)) {
|
SemUtil::inc(slots);
|
return true;
|
}
|
return false;
|
|
}
|
|
template <typename T>
|
bool SLinkedLockFreeQueue<T>::remove_nowait(T & item)
|
{
|
if (SemUtil::dec_nowait(items) == -1) {
|
if (errno == EAGAIN)
|
return false;
|
else
|
err_exit(errno, "remove_nowait");
|
}
|
|
if (SLinkedLockFreeQueue<T>::_remove(item)) {
|
SemUtil::inc(slots);
|
return true;
|
}
|
return false;
|
|
}
|
|
template <typename T>
|
bool SLinkedLockFreeQueue<T>::remove_timeout(T & item, struct timespec * timeout)
|
{
|
if (SemUtil::dec_timeout(items, timeout) == -1) {
|
if (errno == EAGAIN)
|
return false;
|
else
|
err_exit(errno, "remove_timeout");
|
}
|
|
if (SLinkedLockFreeQueue<T>::_remove(item)) {
|
SemUtil::inc(slots);
|
return true;
|
}
|
return false;
|
|
}
|
|
|
template <class T>
|
T& SLinkedLockFreeQueue<T>::operator[](unsigned int i)
|
{
|
if (i < 0 || i >= count)
|
{
|
std::cerr << "Error in array limits: " << i << " is out of range\n";
|
std::exit(EXIT_FAILURE);
|
}
|
|
|
Pointer<T> tmp = Head.load(std::memory_order_relaxed);
|
//Pointer<T> tail = Tail.load(std::memory_order_relaxed);
|
|
while(i > 0) {
|
//std::cout << i << ":" << std::endl;
|
tmp = (tmp.ptr->next).load(std::memory_order_relaxed);
|
i--;
|
}
|
|
tmp = (tmp.ptr->next).load(std::memory_order_relaxed);
|
return tmp.ptr->value;
|
}
|
|
template <class T>
|
void * SLinkedLockFreeQueue<T>::operator new(size_t size){
|
return mm_malloc(size);
|
}
|
|
template <class T>
|
void SLinkedLockFreeQueue<T>::operator delete(void *p) {
|
return mm_free(p);
|
}
|
|
#endif
|
