| | |
| | | namespace bhome_shm |
| | | { |
| | | |
| | | template <class D> |
| | | using Circular = boost::circular_buffer<D, Allocator<D>>; |
| | | // using Circular = robust::CircularBuffer<D, Allocator<D>>; |
| | | |
| | | template <class D> |
| | | class SharedQueue |
| | | { |
| | | public: |
| | | SharedQueue(const uint32_t len, Allocator<D> const &alloc) : |
| | | queue_(len, alloc) {} |
| | | |
| | | bool Read(D &d, const int timeout_ms) |
| | | { |
| | | using namespace std::chrono; |
| | | auto end_time = steady_clock::now() + milliseconds(timeout_ms); |
| | | do { |
| | | if (TryRead(d)) { |
| | | return true; |
| | | } else { |
| | | robust::QuickSleep(); |
| | | } |
| | | } while (steady_clock::now() < end_time); |
| | | return false; |
| | | } |
| | | bool TryRead(D &d) |
| | | { |
| | | // bhome_shm::Guard lock(mutex_); |
| | | if (!queue_.empty()) { |
| | | d = queue_.front(); |
| | | queue_.pop_front(); |
| | | return true; |
| | | } else { |
| | | return false; |
| | | } |
| | | } |
| | | bool TryWrite(const D &d) |
| | | { |
| | | // bhome_shm::Guard lock(mutex_); |
| | | if (!queue_.full()) { |
| | | queue_.push_back(d); |
| | | return true; |
| | | } else { |
| | | return false; |
| | | } |
| | | } |
| | | |
| | | private: |
| | | Circular<D> queue_; |
| | | bhome_shm::Mutex mutex_; |
| | | }; |
| | | |
| | | template <int Power = 4> |
| | | // just wrap robust::AtomicQ63 |
| | | class SharedQ63 |
| | | { |
| | | public: |
| | | template <class... T> |
| | | explicit SharedQ63(T &&...t) {} // easy testing |
| | | |
| | | typedef int64_t Data; |
| | | bool Read(Data &d, const int timeout_ms) |
| | | { |
| | | using namespace std::chrono; |
| | | auto end_time = steady_clock::now() + milliseconds(timeout_ms); |
| | | do { |
| | | if (TryRead(d)) { |
| | | return true; |
| | | } else { |
| | | robust::QuickSleep(); |
| | | } |
| | | } while (steady_clock::now() < end_time); |
| | | return false; |
| | | } |
| | | 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: |
| | | robust::AtomicQueue<Power, Data> queue_; |
| | | AQ63 queue_; |
| | | }; |
| | | |
| | | } // namespace bhome_shm |