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 | 163 +++-------------------------------------------------- 1 files changed, 11 insertions(+), 152 deletions(-) diff --git a/src/shm_queue.h b/src/shm_queue.h index 70039b5..952dae9 100644 --- a/src/shm_queue.h +++ b/src/shm_queue.h @@ -19,170 +19,29 @@ #ifndef SHM_QUEUE_JE0OEUP3 #define SHM_QUEUE_JE0OEUP3 -#include "msg.h" +#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, class OnData> - int ReadAllOnCond(Pred const &pred, OnData const &onData) - { - Guard lock(this->mutex()); - int n = 0; - while (pred(lock)) { - ++n; - onData(this->front()); - this->pop_front(); - this->cond_write_.notify_one(); - } - return n; - } - - template <class Pred> - bool ReadOnCond(D &buf, Pred const &pred) - { - int flag = 0; - auto only_once = [&](Guard &lock) { return flag++ == 0 && pred(lock); }; - auto onData = [&buf](D &d) { - using std::swap; - swap(buf, d); - }; - return ReadAllOnCond(only_once, onData); - } - - 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); - this->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) {} - using Super::capacity; - using Super::size; + 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); - } - - 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()); } -}; - -using namespace bhome_msg; - -class ShmMsgQueue : private ShmObject<SharedQueue<MsgI>>, public StaticDataRef<std::atomic<uint64_t>, ShmMsgQueue> -{ - typedef ShmObject<SharedQueue<MsgI>> Super; - typedef Super::Data Queue; - typedef std::function<void()> OnSend; - -public: - typedef uint64_t MQId; - - static MQId NewId(); - - ShmMsgQueue(const MQId id, ShmType &segment, const int len); - ShmMsgQueue(ShmType &segment, const int len); - ~ShmMsgQueue(); - static bool Remove(SharedMemory &shm, const MQId id); - MQId Id() const { return id_; } - using Super::shm; - - bool Recv(MsgI &msg, const int timeout_ms) { return data()->Read(msg, timeout_ms); } - bool TryRecv(MsgI &msg) { return data()->TryRead(msg); } - template <class OnData> - int TryRecvAll(OnData const &onData) { return data()->TryReadAll(onData); } - static Queue *Find(SharedMemory &shm, const MQId remote_id); - static bool TrySend(SharedMemory &shm, const MQId remote_id, const MsgI &msg, OnSend const &onsend = OnSend()); - template <class Iter> - static int TrySendAll(SharedMemory &shm, const MQId remote_id, const Iter begin, const Iter end, OnSend const &onsend = OnSend()) - { - Queue *remote = Find(shm, remote_id); - if (remote) { - if (onsend) { - return remote->TryWrite(begin, end, [&onsend](const MsgI &msg) { onsend(); msg.AddRef(); }); - } else { - return remote->TryWrite(begin, end, [](const MsgI &msg) { msg.AddRef(); }); - } - } else { - // SetLestError(eNotFound); - return 0; - } - } - - template <class... Rest> - bool TrySend(const MQId remote_id, Rest const &...rest) { return TrySend(shm(), remote_id, rest...); } - template <class... Rest> - int TrySendAll(const MQId remote_id, Rest const &...rest) { return TrySendAll(shm(), remote_id, rest...); } + 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); } private: - MQId id_; + AQ63 queue_; }; } // namespace bhome_shm -- Gitblit v1.8.0