| | |
| | | using namespace bhome_shm; |
| | | using namespace bhome_msg; |
| | | |
| | | #define BH_USE_ATOMIC_Q |
| | | |
| | | class ShmMsgQueue : public StaticDataRef<std::atomic<uint64_t>, ShmMsgQueue> |
| | | class ShmMsgQueue |
| | | { |
| | | public: |
| | | typedef int64_t RawData; |
| | | |
| | | #ifdef BH_USE_ATOMIC_Q |
| | | typedef ShmObject<SharedQ63<0>> Shmq; |
| | | #else |
| | | typedef ShmObject<SharedQueue<RawData>> Shmq; |
| | | // typedef robust::FMutex Mutex; |
| | | // typedef robust::SemMutex Mutex; |
| | | typedef robust::NullMutex Mutex; |
| | | typedef robust::Guard<Mutex> Guard; |
| | | #endif |
| | | |
| | | typedef NamedShmObject<SharedQ63> Shmq; |
| | | typedef Shmq::Data Queue; |
| | | typedef Shmq::ShmType ShmType; |
| | | typedef uint64_t MQId; |
| | | |
| | | static MQId NewId(); |
| | | |
| | | ShmMsgQueue(ShmType &segment, const MQId id, const int len); |
| | | ShmMsgQueue(ShmType &segment, const bool create_or_else_find, const MQId id, const int len); |
| | | ShmMsgQueue(ShmType &segment, const MQId id, Mode mode); |
| | | ShmMsgQueue(const int64_t abs_addr, ShmType &segment, const MQId id); |
| | | ~ShmMsgQueue(); |
| | | static bool Remove(ShmType &shm, const MQId id); |
| | |
| | | ShmType &shm() const { return queue_.shm(); } |
| | | int64_t AbsAddr() const { return queue_.offset(); } |
| | | |
| | | bool Recv(RawData &val, const int timeout_ms) |
| | | { |
| | | #ifndef BH_USE_ATOMIC_Q |
| | | Guard lock(GetMutex(Id())); |
| | | #endif |
| | | return queue().Read(val, timeout_ms); |
| | | } |
| | | // bool Recv(RawData &val, const int timeout_ms) { return queue().Read(val, timeout_ms); } |
| | | bool TryRecv(RawData &val) { return queue().TryRead(val); } |
| | | |
| | | bool TryRecv(RawData &val) |
| | | { |
| | | #ifndef BH_USE_ATOMIC_Q |
| | | Guard lock(GetMutex(Id())); |
| | | #endif |
| | | return queue().TryRead(val); |
| | | } |
| | | |
| | | bool Recv(MsgI &msg, const int timeout_ms) { return Recv(msg.OffsetRef(), timeout_ms); } |
| | | // bool Recv(MsgI &msg, const int timeout_ms) { return Recv(msg.OffsetRef(), timeout_ms); } |
| | | bool TryRecv(MsgI &msg) { return TryRecv(msg.OffsetRef()); } |
| | | static Queue *Find(ShmType &shm, const MQId remote); |
| | | static bool TrySend(ShmType &shm, const MQInfo &remote, const RawData val); |
| | | bool TrySend(const MQInfo &remote, const RawData val) { return TrySend(shm(), remote, val); } |
| | | |
| | | private: |
| | | #ifndef BH_USE_ATOMIC_Q |
| | | static Mutex &GetMutex(const MQId id); |
| | | #endif |
| | | MQId id_; |
| | | Queue &queue() { return *queue_.data(); } |
| | | Shmq queue_; |