From e924ba2fc0f4e57aae032fade3a566c2019f4630 Mon Sep 17 00:00:00 2001 From: wangzhengquan <wangzhengquan85@126.com> Date: 星期五, 29 一月 2021 09:52:30 +0800 Subject: [PATCH] update --- src/queue/shm_queue.h | 132 ++++++++++++++------------------------------ 1 files changed, 42 insertions(+), 90 deletions(-) diff --git a/src/queue/shm_queue.h b/src/queue/shm_queue.h index 5d2d9b6..104ea73 100644 --- a/src/queue/shm_queue.h +++ b/src/queue/shm_queue.h @@ -12,6 +12,7 @@ #include "shm_allocator.h" #include "usg_common.h" #include "array_lock_free_sem_queue.h" +#include "lock_free_queue.h" #include "bus_error.h" template <typename ELEM_T> class SHMQueue { @@ -27,19 +28,15 @@ void force_destroy(); - inline uint32_t size(); + uint32_t size(); - inline bool full(); - inline bool empty(); + bool full(); + bool empty(); - inline int push(const ELEM_T &a_data); - inline int push_nowait(const ELEM_T &a_data); - inline int push_timeout(const ELEM_T &a_data, const struct timespec *timeout); - inline int pop(ELEM_T &a_data); - inline int pop_nowait(ELEM_T &a_data); - inline int pop_timeout(ELEM_T &a_data, struct timespec *timeout); + int push(const ELEM_T &a_data, const struct timespec *timeout=NULL, int flag=0); + int pop(ELEM_T &a_data, const struct timespec *timeout=NULL, int flag=0); - inline ELEM_T &operator[](unsigned i); + ELEM_T &operator[](unsigned i); // @deprecate static size_t remove_queues_exclude(int keys[], size_t length); @@ -51,7 +48,7 @@ /// @brief the actual queue-> methods are forwarded into the real /// implementation - ArrayLockFreeSemQueue<ELEM_T, SHM_Allocator> *queue; + LockFreeQueue<ELEM_T, SHM_Allocator> *queue; private: /// @brief disable copy constructor declaring it private @@ -64,7 +61,7 @@ hashtable_t *hashtable = mm_get_hashtable(); std::set<int> *keyset = hashtable_keyset(hashtable); std::set<int>::iterator keyItr; - ArrayLockFreeSemQueue<ELEM_T, SHM_Allocator> *mqueue; + LockFreeQueue<ELEM_T, SHM_Allocator> *mqueue; bool found; size_t count = 0; for (keyItr = keyset->begin(); keyItr != keyset->end(); keyItr++) { @@ -77,7 +74,7 @@ } if (!found) { // 閿�姣佸叡浜唴瀛樼殑queue - mqueue = (ArrayLockFreeSemQueue<ELEM_T, SHM_Allocator> *)hashtable_get(hashtable, *keyItr); + mqueue = (LockFreeQueue<ELEM_T, SHM_Allocator> *)hashtable_get(hashtable, *keyItr); delete mqueue; hashtable_remove(hashtable, *keyItr); count++; @@ -91,11 +88,11 @@ template <typename ELEM_T> size_t SHMQueue<ELEM_T>::remove_queues(int keys[], size_t length) { hashtable_t *hashtable = mm_get_hashtable(); - ArrayLockFreeSemQueue<ELEM_T, SHM_Allocator> *mqueue; + LockFreeQueue<ELEM_T, SHM_Allocator> *mqueue; size_t count = 0; for(int i = 0; i< length; i++) { // 閿�姣佸叡浜唴瀛樼殑queue - mqueue = (ArrayLockFreeSemQueue<ELEM_T, SHM_Allocator> *)mm_get_by_key(keys[i]); + mqueue = (LockFreeQueue<ELEM_T, SHM_Allocator> *)mm_get_by_key(keys[i]); delete mqueue; hashtable_remove(hashtable, keys[i]); count++; @@ -113,9 +110,9 @@ SHMQueue<ELEM_T>::SHMQueue(int key, size_t qsize) : KEY(key) { hashtable_t *hashtable = mm_get_hashtable(); - queue = (ArrayLockFreeSemQueue<ELEM_T, SHM_Allocator> *)hashtable_get(hashtable, key); + queue = (LockFreeQueue<ELEM_T, SHM_Allocator> *)hashtable_get(hashtable, key); if (queue == NULL || (void *)queue == (void *)1) { - queue = new ArrayLockFreeSemQueue<ELEM_T, SHM_Allocator>(qsize); + queue = new LockFreeQueue<ELEM_T, SHM_Allocator>(qsize); hashtable_put(hashtable, key, (void *)queue); } // queue->reference++; @@ -131,103 +128,58 @@ } -template <typename ELEM_T> inline uint32_t SHMQueue<ELEM_T>::size() { +template <typename ELEM_T> uint32_t SHMQueue<ELEM_T>::size() { return queue->size(); } -template <typename ELEM_T> inline bool SHMQueue<ELEM_T>::full() { +template <typename ELEM_T> bool SHMQueue<ELEM_T>::full() { return queue->full(); } -template <typename ELEM_T> inline bool SHMQueue<ELEM_T>::empty() { +template <typename ELEM_T> bool SHMQueue<ELEM_T>::empty() { return queue->empty(); } + + template <typename ELEM_T> -inline int SHMQueue<ELEM_T>::push(const ELEM_T &a_data) { - int rv = queue->push(a_data); - if(rv == -1) { - return errno; - } else { +int SHMQueue<ELEM_T>::push(const ELEM_T &a_data, const struct timespec *timeout, int flag) { + + + int rv = queue->push(a_data, timeout, flag); + if(rv == 0) { return 0; } + if(rv == ETIMEDOUT) + return EBUS_TIMEOUT; + else { + LoggerFactory::getLogger()->error("LockFreeQueue push_timeout: %s", bus_strerror(rv)); + return rv; + } } + template <typename ELEM_T> -inline int SHMQueue<ELEM_T>::push_nowait(const ELEM_T &a_data) { - int rv = queue->push(a_data, NULL, LOCK_FREE_QUEUE_NOWAIT); - if(rv == -1) { - if (errno == EAGAIN) - return EAGAIN; - else { - err_msg(errno, "LockFreeQueue push_nowait"); - return errno; - } - } - return 0; -} +int SHMQueue<ELEM_T>::pop(ELEM_T &a_data, const struct timespec *timeout, int flag) { -template <typename ELEM_T> -inline int SHMQueue<ELEM_T>::push_timeout(const ELEM_T &a_data, const struct timespec *timeout) { - - int rv = queue->push(a_data, timeout, LOCK_FREE_QUEUE_TIMEOUT); - if(rv == -1) { - if(errno == ETIMEDOUT) - return EBUS_TIMEOUT; - else { - LoggerFactory::getLogger()->error(errno, "LockFreeQueue push_timeout"); - return errno; - } - } - return 0; -} - -template <typename ELEM_T> inline int SHMQueue<ELEM_T>::pop(ELEM_T &a_data) { - // printf("SHMQueue pop before\n"); - int rv = queue->pop(a_data); - // printf("SHMQueue after before\n"); - if(rv == -1) { - return errno; - } else { + int rv = queue->pop(a_data, timeout, flag); + if(rv == 0) { return 0; } -} -template <typename ELEM_T> -inline int SHMQueue<ELEM_T>::pop_nowait(ELEM_T &a_data) { - int rv = queue->pop(a_data, NULL, LOCK_FREE_QUEUE_NOWAIT); - - if(rv == -1) { - if (errno == EAGAIN) - return errno; - else { - LoggerFactory::getLogger()->error(errno, " SHMQueue pop_nowait"); - return errno; - } + + if(rv == ETIMEDOUT) + return EBUS_TIMEOUT; + else { + LoggerFactory::getLogger()->error("LockFreeQueue pop_timeout: %s", bus_strerror(rv)); + return rv; } - return 0; + return rv; } template <typename ELEM_T> -inline int SHMQueue<ELEM_T>::pop_timeout(ELEM_T &a_data, struct timespec *timeout) { - - int rv; - rv = queue->pop(a_data, timeout, LOCK_FREE_QUEUE_TIMEOUT); - if(rv == -1) { - if (errno == ETIMEDOUT) { - return EBUS_TIMEOUT; - } else { - LoggerFactory::getLogger()->error(errno, " SHMQueue pop_timeout"); - return errno; - } - } - return 0; - -} - -template <typename ELEM_T> -inline ELEM_T &SHMQueue<ELEM_T>::operator[](unsigned i) { +ELEM_T &SHMQueue<ELEM_T>::operator[](unsigned i) { return queue->operator[](i); } -- Gitblit v1.8.0