#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" #include "px_sem_util.h" #include "bus_error.h" // default Queue size #define LOCK_FREE_Q_DEFAULT_SIZE 16 // static Logger *logger = LoggerFactory::getLogger(); // 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: sem_t slots; sem_t items; public: sem_t 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 int push(const ELEM_T &a_data); int push_nowait(const ELEM_T &a_data); int 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 int pop(ELEM_T &a_data); int pop_nowait(ELEM_T &a_data); int 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; if (sem_init(&slots, 1, qsize) == -1) err_exit(errno, "LockFreeQueue sem_init"); if (sem_init(&items, 1, 0) == -1) err_exit(errno, "LockFreeQueue sem_init"); if (sem_init(&mutex, 1, 1) == -1) err_exit(errno, "LockFreeQueue sem_init"); } template < typename ELEM_T, typename Allocator, template class Q_TYPE> LockFreeQueue::~LockFreeQueue() { // LoggerFactory::getLogger()->debug("LockFreeQueue desctroy"); if(sem_destroy(&slots) == -1) { err_exit(errno, "LockFreeQueue sem_destroy"); } if(sem_destroy(&items) == -1) { err_exit(errno, "LockFreeQueue sem_destroy"); } if(sem_destroy(&mutex) == -1) { err_exit(errno, "LockFreeQueue sem_destroy"); } } 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> int LockFreeQueue::push(const ELEM_T &a_data) { // LoggerFactory::getLogger()->debug("==================LockFreeQueue push before\n"); if (sem_wait(&slots) == -1) { err_msg(errno, "LockFreeQueue push"); return errno; } if ( m_qImpl.push(a_data) ) { sem_post(&items); // LoggerFactory::getLogger()->debug("==================LockFreeQueue push after\n"); return 0; } return -1; } template < typename ELEM_T, typename Allocator, template class Q_TYPE> int LockFreeQueue::push_nowait(const ELEM_T &a_data) { if (sem_trywait(&slots) == -1) { if (errno == EAGAIN) return EAGAIN; else { err_msg(errno, "LockFreeQueue push_nowait"); return errno; } } if ( m_qImpl.push(a_data)) { sem_post(&items); return 0; } return -1; } template < typename ELEM_T, typename Allocator, template class Q_TYPE> int LockFreeQueue::push_timeout(const ELEM_T &a_data, const struct timespec * ts) { int rv; struct timespec timeout = PXSemUtil::calc_sem_timeout(ts); // LoggerFactory::getLogger()->debug("==================LockFreeQueue push_timeout before tv_sec=%d, tv_nsec=%ld", // timeout.tv_sec, timeout.tv_nsec); while ( sem_timedwait(&slots, &timeout) == -1) { // LoggerFactory::getLogger()->debug("==================LockFreeQueue push_timeout before tv_sec=%d, tv_nsec=%ld, ETIMEDOUT=%d, errno=%d\n", // timeout.tv_sec, timeout.tv_nsec, ETIMEDOUT, errno); if(errno == ETIMEDOUT) return EBUS_TIMEOUT; else if(errno == EINTR) continue; else { LoggerFactory::getLogger()->error(errno, "LockFreeQueue push_timeout"); return errno; } } if (m_qImpl.push(a_data)){ sem_post(&items); // LoggerFactory::getLogger()->debug("==================LockFreeQueue push_timeout after\n"); return 0; } return -1; } template < typename ELEM_T, typename Allocator, template class Q_TYPE> int LockFreeQueue::pop(ELEM_T &a_data) { // LoggerFactory::getLogger()->debug("==================LockFreeQueue pop before\n"); if (sem_wait(&items) == -1) { LoggerFactory::getLogger()->error(errno, "LockFreeQueue pop"); return errno; } if (m_qImpl.pop(a_data)) { sem_post(&slots); // LoggerFactory::getLogger()->debug("==================LockFreeQueue pop after\n"); return 0; } return -1; } template < typename ELEM_T, typename Allocator, template class Q_TYPE> int LockFreeQueue::pop_nowait(ELEM_T &a_data) { if (sem_trywait(&items) == -1) { if (errno == EAGAIN) return errno; else { LoggerFactory::getLogger()->error(errno, "LockFreeQueue pop_nowait"); return errno; } } if (m_qImpl.pop(a_data)) { sem_post(&slots); return 0; } return -1; } template < typename ELEM_T, typename Allocator, template class Q_TYPE> int LockFreeQueue::pop_timeout(ELEM_T &a_data, struct timespec * ts) { // LoggerFactory::getLogger()->debug("=================ts sec = %d, nsec = %ld \n", ts->tv_sec, ts->tv_nsec ); LoggerFactory::getLogger()->debug("==================LockFreeQueue pop_timeout before\n"); struct timespec timeout = PXSemUtil::calc_sem_timeout(ts); // LoggerFactory::getLogger()->debug("================== timeout before sec = %d, nsec = %ld \n", timeout.tv_sec, timeout.tv_nsec ); while (sem_timedwait(&items, &timeout) == -1) { if (errno == ETIMEDOUT) return EBUS_TIMEOUT; else if(errno == EINTR) continue; else { LoggerFactory::getLogger()->error(errno, "LockFreeQueue pop_timeout %d", errno); return errno; } } if (m_qImpl.pop(a_data)) { sem_post(&slots); // LoggerFactory::getLogger()->debug("==================LockFreeQueue pop_timeout after\n"); return 0; } return -1; } 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__