#ifndef __LOCK_FREE_QUEUE_H__ #define __LOCK_FREE_QUEUE_H__ #include #include // assert() #include "mem_pool.h" #include "sem_util.h" #include "logger_factory.h" #include "shm_allocator.h" // default Queue size #define LOCK_FREE_Q_DEFAULT_SIZE 16 // define this macro if calls to "size" must return the real size of the // queue. If it is undefined that function will try to take a snapshot of // the queue, but returned value might be bogus // forward declarations for default template values // template class ArrayLockFreeQueue; // template // class LinkedLockFreeQueue; /// @brief Lock-free queue based on a circular array /// No allocation of extra memory for the nodes handling is needed, but it has /// to add extra overhead (extra CAS operation) when inserting to ensure the /// thread-safety of the queue when the queue type is not /// ArrayLockFreeQueueSingleProducer. /// /// examples of instantiation: /// ArrayLockFreeQueue q; // queue of ints of default size (65535 - 1) /// // and defaulted to single producer /// ArrayLockFreeQueue q; /// // queue of ints of size (16 - 1) and /// // defaulted to single producer /// ArrayLockFreeQueue q; /// // queue of ints of size (100 - 1) with support /// // for multiple producers /// /// ELEM_T represents the type of elementes pushed and popped from the queue /// Q_SIZE size of the queue. The actual size of the queue is (Q_SIZE-1) /// This number should be a power of 2 to ensure /// indexes in the circular queue keep stable when the uint32_t /// variable that holds the current position rolls over from FFFFFFFF /// to 0. For instance /// 2 -> 0x02 /// 4 -> 0x04 /// 8 -> 0x08 /// 16 -> 0x10 /// (...) /// 1024 -> 0x400 /// 2048 -> 0x800 /// /// if queue size is not defined as requested, let's say, for /// instance 100, when current position is FFFFFFFF (4,294,967,295) /// index in the circular array is 4,294,967,295 % 100 = 95. /// When that value is incremented it will be set to 0, that is the /// last 4 elements of the queue are not used when the counter rolls /// over to 0 /// Q_TYPE type of queue implementation. ArrayLockFreeQueueSingleProducer and /// ArrayLockFreeQueue are supported (single producer /// by default) template < typename ELEM_T, typename Allocator = SHM_Allocator, template class Q_TYPE = ArrayLockFreeQueue > class LockFreeQueue { private: int slots; int items; public: int mutex; LockFreeQueue(size_t qsize = LOCK_FREE_Q_DEFAULT_SIZE); /// @brief destructor of the class. /// Note it is not virtual since it is not expected to inherit from this /// template ~LockFreeQueue(); std::atomic_uint reference; /// @brief constructor of the class /// @brief returns the current number of items in the queue /// It tries to take a snapshot of the size of the queue, but in busy environments /// this function might return bogus values. /// /// If a reliable queue size must be kept you might want to have a look at /// the preprocessor variable in this header file called '_WITH_LOCK_FREE_Q_KEEP_REAL_SIZE' /// it enables a reliable size though it hits overall performance of the queue /// (when the reliable size variable is on it's got an impact of about 20% in time) inline uint32_t size(); /// @brief return true if the queue is full. False otherwise /// It tries to take a snapshot of the size of the queue, but in busy /// environments this function might return bogus values. See help in method /// LockFreeQueue::size inline bool full(); inline bool empty(); inline ELEM_T& operator[](unsigned i); /// @brief push an element at the tail of the queue /// @param the element to insert in the queue /// Note that the element is not a pointer or a reference, so if you are using large data /// structures to be inserted in the queue you should think of instantiate the template /// of the queue as a pointer to that large structure /// @return true if the element was inserted in the queue. False if the queue was full bool push(const ELEM_T &a_data); bool push_nowait(const ELEM_T &a_data); bool push_timeout(const ELEM_T &a_data, const struct timespec * timeout); /// @brief pop the element at the head of the queue /// @param a reference where the element in the head of the queue will be saved to /// Note that the a_data parameter might contain rubbish if the function returns false /// @return true if the element was successfully extracted from the queue. False if the queue was empty bool pop(ELEM_T &a_data); bool pop_nowait(ELEM_T &a_data); bool pop_timeout(ELEM_T &a_data, struct timespec * timeout); void *operator new(size_t size); void operator delete(void *p); protected: /// @brief the actual queue. methods are forwarded into the real /// implementation Q_TYPE m_qImpl; private: /// @brief disable copy constructor declaring it private LockFreeQueue(const LockFreeQueue &a_src); }; template < typename ELEM_T, typename Allocator, template class Q_TYPE> LockFreeQueue::LockFreeQueue(size_t qsize): reference(0), m_qImpl(qsize) { // std::cout << "LockFreeQueue init reference=" << reference << std::endl; slots = SemUtil::get(IPC_PRIVATE, qsize); items = SemUtil::get(IPC_PRIVATE, 0); mutex = SemUtil::get(IPC_PRIVATE, 1); } template < typename ELEM_T, typename Allocator, template class Q_TYPE> LockFreeQueue::~LockFreeQueue() { LoggerFactory::getLogger().debug("LockFreeQueue desctroy"); SemUtil::remove(slots); SemUtil::remove(items); SemUtil::remove(mutex); } template < typename ELEM_T, typename Allocator, template class Q_TYPE> inline uint32_t LockFreeQueue::size() { return m_qImpl.size(); } template < typename ELEM_T, typename Allocator, template class Q_TYPE> inline bool LockFreeQueue::full() { return m_qImpl.full(); } template < typename ELEM_T, typename Allocator, template class Q_TYPE> inline bool LockFreeQueue::empty() { return m_qImpl.empty(); } template < typename ELEM_T, typename Allocator, template class Q_TYPE> bool LockFreeQueue::push(const ELEM_T &a_data) { // printf("==================LockFreeQueue push before\n"); if (SemUtil::dec(slots) == -1) { err_msg(errno, "LockFreeQueue push"); return false; } if ( m_qImpl.push(a_data) ) { SemUtil::inc(items); // printf("==================LockFreeQueue push after\n"); return true; } return false; } template < typename ELEM_T, typename Allocator, template class Q_TYPE> bool LockFreeQueue::push_nowait(const ELEM_T &a_data) { if (SemUtil::dec_nowait(slots) == -1) { if (errno == EAGAIN) return false; else { err_msg(errno, "LockFreeQueue push_nowait"); return false; } } if ( m_qImpl.push(a_data)) { SemUtil::inc(items); return true; } return false; } template < typename ELEM_T, typename Allocator, template class Q_TYPE> bool LockFreeQueue::push_timeout(const ELEM_T &a_data, const struct timespec * timeout) { if (SemUtil::dec_timeout(slots, timeout) == -1) { if (errno == EAGAIN) return false; else { // err_msg(errno, "LockFreeQueue push_timeout"); return false; } } if (m_qImpl.push(a_data)){ SemUtil::inc(items); return true; } return false; } template < typename ELEM_T, typename Allocator, template class Q_TYPE> bool LockFreeQueue::pop(ELEM_T &a_data) { // printf("==================LockFreeQueue pop before\n"); if (SemUtil::dec(items) == -1) { err_msg(errno, "LockFreeQueue pop"); return false; } if (m_qImpl.pop(a_data)) { SemUtil::inc(slots); // printf("==================LockFreeQueue pop after\n"); return true; } return false; } template < typename ELEM_T, typename Allocator, template class Q_TYPE> bool LockFreeQueue::pop_nowait(ELEM_T &a_data) { if (SemUtil::dec_nowait(items) == -1) { if (errno == EAGAIN) return false; else { err_msg(errno, "LockFreeQueue pop_nowait"); return false; } } if (m_qImpl.pop(a_data)) { SemUtil::inc(slots); return true; } return false; } template < typename ELEM_T, typename Allocator, template class Q_TYPE> bool LockFreeQueue::pop_timeout(ELEM_T &a_data, struct timespec * timeout) { // printf("==================LockFreeQueue pop_timeout before\n"); if (SemUtil::dec_timeout(items, timeout) == -1) { if (errno == EAGAIN) return false; else { // err_msg(errno, "LockFreeQueue pop_timeout"); return false; } } if (m_qImpl.pop(a_data)) { SemUtil::inc(slots); // printf("==================LockFreeQueue pop_timeout after\n"); return true; } return false; } template < typename ELEM_T, typename Allocator, template class Q_TYPE> ELEM_T& LockFreeQueue::operator[](unsigned i) { return m_qImpl.operator[](i); } template < typename ELEM_T, typename Allocator, template class Q_TYPE> void * LockFreeQueue::operator new(size_t size){ return Allocator::allocate(size); } template < typename ELEM_T, typename Allocator, template class Q_TYPE> void LockFreeQueue::operator delete(void *p) { return Allocator::deallocate(p); } // include implementation files //#include "linked_lock_free_queue.h" #include "array_lock_free_queue.h" #endif // _LOCK_FREE_QUEUE_H__