From 58d904a328c0d849769b483e901a0be9426b8209 Mon Sep 17 00:00:00 2001 From: liuxiaolong <liuxiaolong@aiotlink.com> Date: 星期二, 20 七月 2021 20:20:44 +0800 Subject: [PATCH] 调整Request C.BHFree的位置 --- src/shm_queue.h | 102 ++++++--------------------------------------------- 1 files changed, 12 insertions(+), 90 deletions(-) diff --git a/src/shm_queue.h b/src/shm_queue.h index 5dbda96..952dae9 100644 --- a/src/shm_queue.h +++ b/src/shm_queue.h @@ -19,107 +19,29 @@ #ifndef SHM_QUEUE_JE0OEUP3 #define SHM_QUEUE_JE0OEUP3 +#include "robust.h" #include "shm.h" #include <atomic> #include <boost/circular_buffer.hpp> -#include <boost/date_time/posix_time/posix_time.hpp> +#include <chrono> namespace bhome_shm { -template <class D> -using Circular = boost::circular_buffer<D, Allocator<D>>; - -template <class D> -class SharedQueue : private Circular<D> +// just wrap robust::AtomicQ63 +class SharedQ63 { - typedef Circular<D> Super; - Mutex mutex_; - Cond cond_read_; - Cond cond_write_; - Mutex &mutex() { return mutex_; } - - static boost::posix_time::ptime MSFromNow(const int ms) - { - using namespace boost::posix_time; - ptime cur = boost::posix_time::microsec_clock::universal_time(); - return cur + millisec(ms); - } - - auto TimedReadPred(const int timeout_ms) - { - auto endtime = MSFromNow(timeout_ms); - return [this, endtime](Guard &lock) { - return (cond_read_.timed_wait(lock, endtime, [&]() { return !this->empty(); })); - }; - } - auto TryReadPred() - { - return [this](Guard &lock) { return !this->empty(); }; - } - - template <class Pred> - bool ReadOnCond(D &buf, Pred const &pred) - { - auto Read = [&]() { - Guard lock(this->mutex()); - if (pred(lock)) { - using std::swap; - swap(buf, Super::front()); - Super::pop_front(); - return true; - } else { - return false; - } - }; - return Read() ? (this->cond_write_.notify_one(), true) : false; - } - - template <class Iter, class Pred, class OnWrite> - int WriteAllOnCond(Iter begin, Iter end, Pred const &pred, OnWrite const &onWrite) - { - if (begin == end) { return 0; } - - int n = 0; - Guard lock(mutex()); - while (pred(lock)) { - onWrite(*begin); - Super::push_back(*begin); - ++n; - cond_read_.notify_one(); - if (++begin == end) { - break; - } - } - return n; - } - public: - SharedQueue(const uint32_t len, Allocator<D> const &alloc) : - Super(len, alloc) {} + template <class... T> + explicit SharedQ63(T &&...t) {} // easy testing - template <class Iter, class OnWrite> - int TryWrite(Iter begin, Iter end, const OnWrite &onWrite) - { - auto tryWritePred = [this](Guard &lock) { return !this->full(); }; - return WriteAllOnCond(begin, end, tryWritePred, onWrite); - } + typedef robust::AtomicQ63 AQ63; + typedef AQ63::Data Data; + bool TryRead(Data &d, const bool try_more = true) { return queue_.pop(d, try_more); } + bool TryWrite(const Data d, const bool try_more = true) { return queue_.push(d, try_more); } - template <class OnWrite> - bool TryWrite(const D &buf, const OnWrite &onWrite) { return TryWrite(&buf, (&buf) + 1, onWrite); } - - bool TryWrite(const D &buf) - { - return TryWrite(buf, [](const D &buf) {}); - } - - template <class OnData> - int ReadAll(const int timeout_ms, OnData const &onData) { return ReadAllOnCond(TimedReadPred(timeout_ms), onData); } - template <class OnData> - int TryReadAll(OnData const &onData) { return ReadAllOnCond(TryReadPred(), onData); } - - bool Read(D &buf, const int timeout_ms) { return ReadOnCond(buf, TimedReadPred(timeout_ms)); } - bool TryRead(D &buf) { return ReadOnCond(buf, TryReadPred()); } +private: + AQ63 queue_; }; } // namespace bhome_shm -- Gitblit v1.8.0