| | |
| | | * |
| | | * ===================================================================================== |
| | | */ |
| | | #include "shm_queue.h" |
| | | #include <boost/uuid/uuid_io.hpp> |
| | | #include <boost/uuid/uuid_generators.hpp> |
| | | #include "bh_util.h" |
| | | |
| | | namespace bhome_shm { |
| | | |
| | | #include "shm_queue.h" |
| | | #include "bh_util.h" |
| | | #include <boost/uuid/uuid_generators.hpp> |
| | | #include <boost/uuid/uuid_io.hpp> |
| | | |
| | | namespace bhome_shm |
| | | { |
| | | using namespace bhome_msg; |
| | | using namespace boost::interprocess; |
| | | using namespace boost::uuids; |
| | | |
| | | namespace { |
| | | std::string MsgQIdToName(const MQId& id) { return "shmq" + to_string(id); } |
| | | MQId EmptyId() { return nil_uuid(); } |
| | | namespace |
| | | { |
| | | std::string MsgQIdToName(const MQId &id) { return "shmq" + to_string(id); } |
| | | // MQId EmptyId() { return nil_uuid(); } |
| | | MQId NewId() { return random_generator()(); } |
| | | } |
| | | |
| | | ShmMsgQueue::ShmMsgQueue(const MQId &id, ShmType &segment, const uint32_t len): |
| | | SharedQueue(segment, MsgQIdToName(id), id, len, segment.get_segment_manager()) |
| | | const int AdjustMQLength(const int len) |
| | | { |
| | | printf("queue size: %ld cap: %ld\n", data()->size(), data()->capacity()); |
| | | const int kMaxLength = 10000; |
| | | const int kDefaultLen = 12; |
| | | if (len <= 0) { |
| | | return kDefaultLen; |
| | | } else if (len < kMaxLength) { |
| | | return len; |
| | | } else { |
| | | return kMaxLength; |
| | | } |
| | | } |
| | | |
| | | ShmMsgQueue::ShmMsgQueue(ShmType &segment, const uint32_t len):ShmMsgQueue(NewId(), segment, len) |
| | | {} |
| | | } // namespace |
| | | |
| | | ShmMsgQueue::~ShmMsgQueue() |
| | | // ShmMsgQueue memory usage: (320 + 16*length) bytes, length >= 2 |
| | | ShmMsgQueue::ShmMsgQueue(const MQId &id, ShmType &segment, const int len) : |
| | | Super(segment, MsgQIdToName(id), AdjustMQLength(len), segment.get_segment_manager()), |
| | | id_(id) |
| | | { |
| | | Remove(); |
| | | } |
| | | |
| | | bool ShmMsgQueue::Send(const MQId &remote_id, const Msg &msg, const int timeout_ms) |
| | | ShmMsgQueue::ShmMsgQueue(ShmType &segment, const int len) : |
| | | ShmMsgQueue(NewId(), segment, len) {} |
| | | |
| | | ShmMsgQueue::~ShmMsgQueue() {} |
| | | |
| | | bool ShmMsgQueue::Remove(SharedMemory &shm, const MQId &id) |
| | | { |
| | | Queue *remote = find(MsgQIdToName(remote_id)); |
| | | return remote && remote->Write(msg, timeout_ms, [&](){msg.AddRef();}); |
| | | return Super::Remove(shm, MsgQIdToName(id)); |
| | | } |
| | | |
| | | bool ShmMsgQueue::Send(const MQId &remote_id, const void *data, const size_t size, const int timeout_ms) |
| | | ShmMsgQueue::Queue *ShmMsgQueue::FindRemote(SharedMemory &shm, const MQId &remote_id) |
| | | { |
| | | // 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. |
| | | |
| | | Msg msg; |
| | | if(msg.Build(shm(), Id(), data, size, false)) { |
| | | if(Send(remote_id, msg, timeout_ms)) { |
| | | return true; |
| | | } else { |
| | | msg.Release(shm()); |
| | | } |
| | | } |
| | | return false; |
| | | return Find(shm, MsgQIdToName(remote_id)); |
| | | } |
| | | |
| | | bool ShmMsgQueue::Recv(MQId &source_id, void *&data, size_t &size, const int timeout_ms) |
| | | bool ShmMsgQueue::Send(SharedMemory &shm, const MQId &remote_id, const MsgI &msg, const int timeout_ms, OnSend const &onsend) |
| | | { |
| | | Msg msg; |
| | | if (Read(msg, timeout_ms)) { |
| | | DEFER1(msg.Release(shm());); |
| | | |
| | | auto ptr = msg.get<char>(); |
| | | if (ptr) { |
| | | MsgMetaV1 meta; |
| | | meta.Parse(ptr); |
| | | source_id = meta.src_id_; |
| | | size = meta.data_size_; |
| | | data = malloc(size); |
| | | if (data) { |
| | | memcpy(data, ptr + meta.self_size_, size); |
| | | return true; |
| | | } |
| | | } |
| | | } |
| | | source_id = EmptyId(); |
| | | data = 0; |
| | | size = 0; |
| | | return false; |
| | | Queue *remote = FindRemote(shm, remote_id); |
| | | if (remote) { |
| | | if (onsend) { |
| | | return remote->Write(msg, timeout_ms, [&onsend](const MsgI &msg) { onsend(); msg.AddRef(); }); |
| | | } else { |
| | | return remote->Write(msg, timeout_ms, [](const MsgI &msg) { msg.AddRef(); }); |
| | | } |
| | | } else { |
| | | // SetLestError(eNotFound); |
| | | return false; |
| | | } |
| | | } |
| | | |
| | | bool ShmMsgQueue::TrySend(SharedMemory &shm, const MQId &remote_id, const MsgI &msg, OnSend const &onsend) |
| | | { |
| | | Queue *remote = FindRemote(shm, remote_id); |
| | | if (remote) { |
| | | if (onsend) { |
| | | return remote->TryWrite(msg, [&onsend](const MsgI &msg) { onsend(); msg.AddRef(); }); |
| | | } else { |
| | | return remote->TryWrite(msg, [](const MsgI &msg) { msg.AddRef(); }); |
| | | } |
| | | } else { |
| | | // SetLestError(eNotFound); |
| | | 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. |
| | | |
| | | } // namespace bhome_shm |
| | | |