From 68d23225a38a35f1325eb39fa4ed5a005d5de473 Mon Sep 17 00:00:00 2001 From: fujuntang <fujuntang@aiot.com> Date: 星期三, 11 八月 2021 09:50:20 +0800 Subject: [PATCH] fix from 3.1 first commit --- src/queue/shm_queue.h | 238 +++++++++++++++++++++++++--------------------------------- 1 files changed, 103 insertions(+), 135 deletions(-) diff --git a/src/queue/shm_queue.h b/src/queue/shm_queue.h index 7c7b89b..74b9b33 100644 --- a/src/queue/shm_queue.h +++ b/src/queue/shm_queue.h @@ -1,52 +1,56 @@ +/** + * encapsulate lock_free_queue, populate in userspace + */ + #ifndef __SHM_QUEUE_H__ #define __SHM_QUEUE_H__ #include "hashtable.h" -#include "lock_free_queue.h" + #include "logger_factory.h" #include "sem_util.h" #include "shm_allocator.h" #include "usg_common.h" -// default Queue size -// #define LOCK_FREE_Q_DEFAULT_SIZE 16 +#include "array_lock_free_sem_queue.h" +#include "lock_free_queue.h" +#include "bus_error.h" template <typename ELEM_T> class SHMQueue { private: - const int KEY; + const int mkey; + hashtable_t * hashtable; + // 鏄惁鏄痥ey瀵瑰簲鐨勫叡浜槦鍒楃殑鐪熸鎷ユ湁鑰咃紝涔熷氨鏄鏄痓ind鍒発ey涓婄殑锛屼笉鏄痑ttach鍒発ey涓婄殑瀵硅薄 + bool owner; + size_t mqsize; public: /// @brief constructor of the class - SHMQueue(int key = 0, size_t qsize = LOCK_FREE_Q_DEFAULT_SIZE); + SHMQueue(size_t qsize = 16); ~SHMQueue(); - void force_destroy(); + bool bind(int key, bool force) ; + bool attach(int key); - inline uint32_t size(); + int get_key(); - inline bool full(); - inline bool empty(); + uint32_t size(); - inline bool push(const ELEM_T &a_data); - inline bool push_nowait(const ELEM_T &a_data); - inline bool push_timeout(const ELEM_T &a_data, - const struct timespec *timeout); - inline bool pop(ELEM_T &a_data); - inline bool pop_nowait(ELEM_T &a_data); - inline bool pop_timeout(ELEM_T &a_data, struct timespec *timeout); + bool full(); + bool empty(); - inline ELEM_T &operator[](unsigned i); + 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); - // @deprecate - static size_t remove_queues_exclude(int keys[], size_t length); - static size_t remove_queues(int keys[], size_t length); - static size_t remove_queue(int key); + ELEM_T &operator[](unsigned i); + private: protected: /// @brief the actual queue-> methods are forwarded into the real /// implementation + LockFreeQueue<ELEM_T, SHM_Allocator> *queue; private: @@ -54,156 +58,120 @@ SHMQueue<ELEM_T>(const SHMQueue<ELEM_T> &a_src); }; -// @deprecate -template <typename ELEM_T> -size_t SHMQueue<ELEM_T>::remove_queues_exclude(int keys[], size_t length) { - hashtable_t *hashtable = mm_get_hashtable(); - std::set<int> *keyset = hashtable_keyset(hashtable); - std::set<int>::iterator keyItr; - LockFreeQueue<ELEM_T, SHM_Allocator> *mqueue; - bool found; - size_t count = 0; - for (keyItr = keyset->begin(); keyItr != keyset->end(); keyItr++) { - found = false; - for (size_t i = 0; i < length; i++) { - if (*keyItr == keys[i]) { - found = true; - break; - } - } - if (!found) { - // 閿�姣佸叡浜唴瀛樼殑queue - mqueue = (LockFreeQueue<ELEM_T, SHM_Allocator> *)hashtable_get(hashtable, *keyItr); - delete mqueue; - hashtable_remove(hashtable, *keyItr); - count++; - } - } - delete keyset; - return count; -} + + template <typename ELEM_T> -size_t SHMQueue<ELEM_T>::remove_queues(int keys[], size_t length) { - hashtable_t *hashtable = mm_get_hashtable(); - LockFreeQueue<ELEM_T, SHM_Allocator> *mqueue; - size_t count = 0; - for(int i = 0; i< length; i++) { - // 閿�姣佸叡浜唴瀛樼殑queue - mqueue = (LockFreeQueue<ELEM_T, SHM_Allocator> *)mm_get_by_key(keys[i]); - delete mqueue; - hashtable_remove(hashtable, keys[i]); - count++; - } - return count; -} +SHMQueue<ELEM_T>::SHMQueue(size_t qsize): mqsize(qsize) { -template <typename ELEM_T> -size_t SHMQueue<ELEM_T>::remove_queue(int key) { - int keys[] = {key}; - return remove_queues(keys, 1); -} - -template <typename ELEM_T> -SHMQueue<ELEM_T>::SHMQueue(int key, size_t qsize) : KEY(key) { - - hashtable_t *hashtable = mm_get_hashtable(); - queue = (LockFreeQueue<ELEM_T, SHM_Allocator> *)hashtable_get(hashtable, key); - if (queue == NULL || (void *)queue == (void *)1) { - queue = new LockFreeQueue<ELEM_T, SHM_Allocator>(qsize); - hashtable_put(hashtable, key, (void *)queue); - } - queue->reference++; + hashtable = mm_get_hashtable(); + owner = false; + mkey = 0; + // queue = (LockFreeQueue<ELEM_T, SHM_Allocator> *)hashtable_get(hashtable, key); + // if (queue == NULL || (void *)queue == (void *)1) { + // queue = new LockFreeQueue<ELEM_T, SHM_Allocator>(qsize); + // hashtable_put(hashtable, key, (void *)queue); + // } + // queue->reference++; // LoggerFactory::getLogger()->debug("SHMQueue constructor reference===%d", queue->reference.load()); } template <typename ELEM_T> SHMQueue<ELEM_T>::~SHMQueue() { - if(queue == NULL) { - // queue宸茬粡閿�姣� - return; - } + LoggerFactory::getLogger()->debug("SHMQueue destroy"); + - SemUtil::dec(queue->mutex); - queue->reference--; - // LoggerFactory::getLogger()->debug("SHMQueue destructor reference===%d", - if (queue->reference.load() == 0) { - delete queue; - queue = NULL; - hashtable_t *hashtable = mm_get_hashtable(); - hashtable_remove(hashtable, KEY); - // 姝ゆ椂queue宸茬粡閿�姣侊紝鏃犻渶 SemUtil::inc(queue->mutex) - // printf("SHMQueue destructor delete queue\n"); - } else { - SemUtil::inc(queue->mutex); - } } -template <typename ELEM_T> void SHMQueue<ELEM_T>::force_destroy() { - if(queue == NULL) { - // queue宸茬粡閿�姣� - return; - } +template <typename ELEM_T> +bool SHMQueue<ELEM_T>::bind(int key, bool force) { - SemUtil::dec(queue->mutex); - delete queue; - queue = NULL; - hashtable_t *hashtable = mm_get_hashtable(); - hashtable_remove(hashtable, KEY); - // 姝ゆ椂queue宸茬粡閿�姣侊紝鏃犻渶 SemUtil::inc(queue->mutex) + void *tmp_ptr = hashtable_get(hashtable, key); + if (tmp_ptr == NULL || tmp_ptr == (void *)1 || force) { + queue = new LockFreeQueue<ELEM_T, SHM_Allocator>(mqsize); + hashtable_put(hashtable, key, (void *)queue); + mkey = key; + owner = true; + return true; + } + + return false; } -template <typename ELEM_T> inline uint32_t SHMQueue<ELEM_T>::size() { +template <typename ELEM_T> +bool SHMQueue<ELEM_T>::attach(int key) { + void *tmp_ptr = hashtable_get(hashtable, key); + if (tmp_ptr == NULL || tmp_ptr == (void *)1) { + return false; + } + mkey = key; + queue = (LockFreeQueue<ELEM_T, SHM_Allocator> *)tmp_ptr; + return true; +} + +template <typename ELEM_T> int SHMQueue<ELEM_T>::get_key() { + return mkey; +} + + +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 bool SHMQueue<ELEM_T>::push(const ELEM_T &a_data) { - return queue->push(a_data); -} + + template <typename ELEM_T> -inline bool SHMQueue<ELEM_T>::push_nowait(const ELEM_T &a_data) { - return queue->push_nowait(a_data); +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 bool SHMQueue<ELEM_T>::push_timeout(const ELEM_T &a_data, - const struct timespec *timeout) { +int SHMQueue<ELEM_T>::pop(ELEM_T &a_data, const struct timespec *timeout, int flag) { - return queue->push_timeout(a_data, timeout); -} + int rv = queue->pop(a_data, timeout, flag); + if(rv == 0) { + return 0; + } -template <typename ELEM_T> inline bool 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 == ETIMEDOUT) + return EBUS_TIMEOUT; + else { + LoggerFactory::getLogger()->error("LockFreeQueue pop_timeout: %s", bus_strerror(rv)); + return rv; + } return rv; + } template <typename ELEM_T> -inline bool SHMQueue<ELEM_T>::pop_nowait(ELEM_T &a_data) { - return queue->pop_nowait(a_data); -} - -template <typename ELEM_T> -inline bool SHMQueue<ELEM_T>::pop_timeout(ELEM_T &a_data, - struct timespec *timeout) { - return queue->pop_timeout(a_data, timeout); -} - -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); } + + + #endif -- Gitblit v1.8.0