From f52f2c2828047c2f30d30fc1fe2b54d8db146d49 Mon Sep 17 00:00:00 2001 From: wangzhengquan <wangzhengquan85@126.com> Date: 星期四, 25 二月 2021 15:56:35 +0800 Subject: [PATCH] update --- src/queue/lock_free_queue.h | 218 +++++++++++++++++++++++++++++++----------------------- 1 files changed, 125 insertions(+), 93 deletions(-) diff --git a/src/queue/lock_free_queue.h b/src/queue/lock_free_queue.h index 9f03437..54d0b04 100644 --- a/src/queue/lock_free_queue.h +++ b/src/queue/lock_free_queue.h @@ -6,15 +6,21 @@ #include <usg_common.h> #include <assert.h> // assert() -#include "mem_pool.h" +#include "shm_mm.h" #include "sem_util.h" #include "logger_factory.h" #include "shm_allocator.h" #include "psem.h" #include "bus_error.h" #include "bus_def.h" + // default Queue size #define LOCK_FREE_Q_DEFAULT_SIZE 16 + + +#define LOCK_FREE_Q_ST_OPENED 0 + +#define LOCK_FREE_Q_ST_CLOSED 1 // static Logger *logger = LoggerFactory::getLogger(); // define this macro if calls to "size" must return the real size of the @@ -72,9 +78,9 @@ /// ArrayLockFreeQueue are supported (single producer /// by default) template< - typename ELEM_T, - typename Allocator = SHM_Allocator, - template<typename T, typename AT> class Q_TYPE = ArrayLockFreeQueue + typename ELEM_T, + typename Allocator = SHM_Allocator, + template<typename T, typename AT> class Q_TYPE = ArrayLockFreeQueue > class LockFreeQueue { @@ -82,9 +88,11 @@ sem_t slots; sem_t items; + time_t createTime; + time_t closeTime; + int status; public: - sem_t mutex; LockFreeQueue(size_t qsize = LOCK_FREE_Q_DEFAULT_SIZE); @@ -93,7 +101,9 @@ /// template ~LockFreeQueue(); - std::atomic_uint reference; + inline void close(); + + // std::atomic_uint reference; /// @brief constructor of the class @@ -116,6 +126,20 @@ inline bool empty(); inline ELEM_T &operator[](unsigned i); + + + + time_t getCreateTime() { + return createTime; + } + + time_t getCloseTime() { + return closeTime; + } + + int getStatus() { + return status; + } /// @brief push an element at the tail of the queue /// @param the element to insert in the queue @@ -148,161 +172,169 @@ template< - typename ELEM_T, - typename Allocator, - template<typename T, typename AT> class Q_TYPE> -LockFreeQueue<ELEM_T, Allocator, Q_TYPE>::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"); - + typename ELEM_T, + typename Allocator, + template<typename T, typename AT> class Q_TYPE> +LockFreeQueue<ELEM_T, Allocator, Q_TYPE>::LockFreeQueue(size_t qsize): 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"); + + createTime = time(NULL); + status = LOCK_FREE_Q_ST_OPENED; } template< - typename ELEM_T, - typename Allocator, - template<typename T, typename AT> class Q_TYPE> + typename ELEM_T, + typename Allocator, + template<typename T, typename AT> class Q_TYPE> +inline void LockFreeQueue<ELEM_T, Allocator, Q_TYPE>::close() { + status = LOCK_FREE_Q_ST_CLOSED; + closeTime = time(NULL); +} + + +template< + typename ELEM_T, + typename Allocator, + template<typename T, typename AT> class Q_TYPE> LockFreeQueue<ELEM_T, Allocator, Q_TYPE>::~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"); - } + // 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"); + } + } template< - typename ELEM_T, - typename Allocator, - template<typename T, typename AT> class Q_TYPE> + typename ELEM_T, + typename Allocator, + template<typename T, typename AT> class Q_TYPE> inline uint32_t LockFreeQueue<ELEM_T, Allocator, Q_TYPE>::size() { - return m_qImpl.size(); + return m_qImpl.size(); } template< - typename ELEM_T, - typename Allocator, - template<typename T, typename AT> class Q_TYPE> + typename ELEM_T, + typename Allocator, + template<typename T, typename AT> class Q_TYPE> inline bool LockFreeQueue<ELEM_T, Allocator, Q_TYPE>::full() { - return m_qImpl.full(); + return m_qImpl.full(); } template< - typename ELEM_T, - typename Allocator, - template<typename T, typename AT> class Q_TYPE> + typename ELEM_T, + typename Allocator, + template<typename T, typename AT> class Q_TYPE> inline bool LockFreeQueue<ELEM_T, Allocator, Q_TYPE>::empty() { - return m_qImpl.empty(); + return m_qImpl.empty(); } template<typename ELEM_T, - typename Allocator, - template<typename T, typename AT> class Q_TYPE> + typename Allocator, + template<typename T, typename AT> class Q_TYPE> int LockFreeQueue<ELEM_T, Allocator, Q_TYPE>::push(const ELEM_T &a_data, const struct timespec *timeout, int flag) { // LoggerFactory::getLogger()->debug("==================LockFreeQueue push before\n"); - sigset_t mask_all, pre; - sigfillset(&mask_all); + // sigset_t mask_all, pre; + // sigfillset(&mask_all); - sigprocmask(SIG_BLOCK, &mask_all, &pre); + // sigprocmask(SIG_BLOCK, &mask_all, &pre); if ((flag & BUS_NOWAIT_FLAG) == BUS_NOWAIT_FLAG) { - if (psem_trywait(&slots) == -1) { - goto LABEL_FAILTURE; - } + if (psem_trywait(&slots) == -1) { + goto LABEL_FAILTURE; + } } else if ((flag & BUS_TIMEOUT_FLAG) == BUS_TIMEOUT_FLAG && timeout != NULL) { - if (psem_timedwait(&slots, timeout) == -1) { - goto LABEL_FAILTURE; - } + if (psem_timedwait(&slots, timeout) == -1) { + goto LABEL_FAILTURE; + } } else { - if (psem_wait(&slots) == -1) { - goto LABEL_FAILTURE; - } + if (psem_wait(&slots) == -1) { + goto LABEL_FAILTURE; + } } if (m_qImpl.push(a_data)) { psem_post(&items); - sigprocmask(SIG_SETMASK, &pre, NULL); - LoggerFactory::getLogger()->debug("==================LockFreeQueue push after\n"); + // sigprocmask(SIG_SETMASK, &pre, NULL); + // LoggerFactory::getLogger()->debug("==================LockFreeQueue push after\n"); return 0; } - -LABEL_FAILTURE: - sigprocmask(SIG_SETMASK, &pre, NULL); - return errno; + LABEL_FAILTURE: + // sigprocmask(SIG_SETMASK, &pre, NULL); + return errno; } template<typename ELEM_T, - typename Allocator, - template<typename T, typename AT> class Q_TYPE> + typename Allocator, + template<typename T, typename AT> class Q_TYPE> int LockFreeQueue<ELEM_T, Allocator, Q_TYPE>::pop(ELEM_T &a_data, const struct timespec *timeout, int flag) { // LoggerFactory::getLogger()->debug("==================LockFreeQueue pop before...."); - sigset_t mask_all, pre; - sigfillset(&mask_all); + // sigset_t mask_all, pre; + // sigfillset(&mask_all); - sigprocmask(SIG_BLOCK, &mask_all, &pre); + // sigprocmask(SIG_BLOCK, &mask_all, &pre); if ((flag & BUS_NOWAIT_FLAG) == BUS_NOWAIT_FLAG) { - if (psem_trywait(&items) == -1) { - goto LABEL_FAILTURE; - } + if (psem_trywait(&items) == -1) { + goto LABEL_FAILTURE; + } } else if ((flag & BUS_TIMEOUT_FLAG) == BUS_TIMEOUT_FLAG && timeout != NULL) { // LoggerFactory::getLogger()->debug("==================LockFreeQueue pop before. flag=%d , %d\n", flag, timeout->tv_sec); - if (psem_timedwait(&items, timeout) == -1) { - goto LABEL_FAILTURE; - } + if (psem_timedwait(&items, timeout) == -1) { + goto LABEL_FAILTURE; + } } else { - if (psem_wait(&items) == -1) { - goto LABEL_FAILTURE; - } + if (psem_wait(&items) == -1) { + goto LABEL_FAILTURE; + } } - if (m_qImpl.pop(a_data)) { - psem_post(&slots); - sigprocmask(SIG_SETMASK, &pre, NULL); - // LoggerFactory::getLogger()->debug("==================LockFreeQueue pop after\n"); - return 0; + psem_post(&slots); + // sigprocmask(SIG_SETMASK, &pre, NULL); + // LoggerFactory::getLogger()->debug("==================LockFreeQueue pop after\n"); + return 0; } -LABEL_FAILTURE: - sigprocmask(SIG_SETMASK, &pre, NULL); + + LABEL_FAILTURE: + // sigprocmask(SIG_SETMASK, &pre, NULL); return errno; } template<typename ELEM_T, - typename Allocator, - template<typename T, typename AT> class Q_TYPE> + typename Allocator, + template<typename T, typename AT> class Q_TYPE> ELEM_T &LockFreeQueue<ELEM_T, Allocator, Q_TYPE>::operator[](unsigned i) { - return m_qImpl.operator[](i); + return m_qImpl.operator[](i); } template<typename ELEM_T, - typename Allocator, - template<typename T, typename AT> class Q_TYPE> + typename Allocator, + template<typename T, typename AT> class Q_TYPE> void *LockFreeQueue<ELEM_T, Allocator, Q_TYPE>::operator new(size_t size) { - return Allocator::allocate(size); + return Allocator::allocate(size); } template<typename ELEM_T, - typename Allocator, - template<typename T, typename AT> class Q_TYPE> + typename Allocator, + template<typename T, typename AT> class Q_TYPE> void LockFreeQueue<ELEM_T, Allocator, Q_TYPE>::operator delete(void *p) { - return Allocator::deallocate(p); + LockFreeQueue<ELEM_T, Allocator, Q_TYPE> * _que = (LockFreeQueue<ELEM_T, Allocator, Q_TYPE> * )p; + Allocator::deallocate(p); } // include implementation files -- Gitblit v1.8.0