| | |
| | | |
| | | } // namespace |
| | | |
| | | ShmMsgQueue::MQId ShmMsgQueue::NewId() |
| | | { |
| | | static auto &id = GetData("Must init shared memory before use! Please make sure center is running."); |
| | | return (++id) * 10; |
| | | } |
| | | |
| | | ShmMsgQueue::ShmMsgQueue(ShmType &segment, const MQId id, const int len) : |
| | | ShmMsgQueue::ShmMsgQueue(ShmType &segment, const MQId id, Mode mode) : |
| | | id_(id), |
| | | queue_(segment, MsgQIdToName(id_), len, segment.get_segment_manager()) |
| | | queue_(segment, MsgQIdToName(id_), mode) |
| | | { |
| | | } |
| | | |
| | | ShmMsgQueue::ShmMsgQueue(ShmType &segment, const bool create_or_else_find, const MQId id, const int len) : |
| | | id_(id), |
| | | queue_(segment, create_or_else_find, MsgQIdToName(id_), len, segment.get_segment_manager()) |
| | | { |
| | | if (!queue_.IsOk()) { |
| | | throw("error create/find msgq " + std::to_string(id_)); |
| | | } |
| | | } |
| | | ShmMsgQueue::ShmMsgQueue(const int64_t abs_addr, ShmType &segment, const MQId id) : |
| | | id_(id), queue_(abs_addr, segment, MsgQIdToName(id_)) |
| | | { |
| | | //TODO check some tag. |
| | | } |
| | | id_(id), queue_(abs_addr, segment, MsgQIdToName(id_)) {} |
| | | |
| | | ShmMsgQueue::~ShmMsgQueue() {} |
| | | |
| | | #ifndef BH_USE_ATOMIC_Q |
| | | ShmMsgQueue::Mutex &ShmMsgQueue::GetMutex(const MQId id) |
| | | { |
| | | static std::unordered_map<MQId, std::shared_ptr<Mutex>> imm; |
| | | |
| | | static std::mutex mtx; |
| | | std::lock_guard<std::mutex> lock(mtx); |
| | | auto pos = imm.find(id); |
| | | if (pos == imm.end()) { |
| | | pos = imm.emplace(id, new Mutex(id)).first; |
| | | // pos = imm.emplace(id, new Mutex()).first; |
| | | } |
| | | return *pos->second; |
| | | } |
| | | #endif |
| | | |
| | | bool ShmMsgQueue::Remove(SharedMemory &shm, const MQId id) |
| | | { |
| | |
| | | if (IsCmd(val)) { |
| | | LOG_DEBUG() << "clsing queue " << id << ", has a cmd" << DecodeCmd(val); |
| | | } else { |
| | | MsgI(val).Release(); |
| | | MsgI(val, shm).Release(); |
| | | } |
| | | } |
| | | } |
| | |
| | | bool ShmMsgQueue::TrySend(SharedMemory &shm, const MQInfo &remote, const RawData val) |
| | | { |
| | | try { |
| | | //TODO find from center, or use offset. |
| | | ShmMsgQueue dest(remote.offset_, shm, remote.id_); |
| | | #ifndef BH_USE_ATOMIC_Q |
| | | Guard lock(GetMutex(remote_id)); |
| | | #endif |
| | | return dest.queue().TryWrite(val); |
| | | } catch (...) { |
| | | // SetLastError(eNotFound, "remote not found"); |
| | | return false; |
| | | } |
| | | } |
| | | |
| | | // Test shows that in the 2 cases: |
| | | // 1) build msg first, then find remote queue; |
| | | // 2) find remote queue first, then build msg; |
| | | // 1 is about 50% faster than 2, maybe cache related. |