/* * ===================================================================================== * * Filename: msg.h * * Description: * * Version: 1.0 * Created: 2021年03月24日 16时49分20秒 * Revision: none * Compiler: gcc * * Author: Li Chao (), * Organization: * * ===================================================================================== */ #ifndef MSG_5BILLZET #define MSG_5BILLZET #include "bh_util.h" #include "proto.h" #include "shm.h" #include #include class ShmSocket; namespace bhome_msg { using namespace bhome_shm; // ShmMsg is safe to be stored in shared memory, so POD data or offset_ptr is required. // message content layout: (meta) / header_size + header + data_size + data class ShmMsg { private: // store ref count, msgs shareing the same data should also hold a pointer of the same RefCount object. class RefCount : private boost::noncopyable { std::atomic num_; public: RefCount() : num_(1) { } // num_(1) { static_assert(std::is_pod::value); } int Inc() { return ++num_; } int Dec() { return --num_; } int Get() { return num_.load(); } }; typedef int64_t OffsetType; static OffsetType Addr(void *ptr) { return reinterpret_cast(ptr); } static void *Ptr(const OffsetType offset) { return reinterpret_cast(offset); } OffsetType BaseAddr() const { return Addr(shm().get_address()); } static const uint32_t kMsgTag = 0xf1e2d3c4; struct Meta { static int64_t NewId() { static std::atomic id(0); return ++id; } RefCount count_; const uint32_t tag_ = kMsgTag; const uint32_t capacity_ = 0; const int64_t id_ = 0; std::atomic timestamp_; bool managed_ = false; uint32_t size_ = 0; Meta(uint32_t size) : capacity_(size), id_(NewId()), timestamp_(NowSec()) {} }; OffsetType offset_; SharedMemory *pshm_; void *Alloc(const size_t size, const void *src = nullptr) { void *p = shm().Alloc(sizeof(Meta) + size); if (p) { auto pmeta = new (p) Meta(size); p = pmeta + 1; if (src) { memcpy(p, src, size); } } return p; } private: Meta *meta() const { return get() - 1; } template void *Pack(const BHMsgHead &head, const uint32_t head_len, const Body &body, const uint32_t body_len) { void *addr = get(); if (addr) { auto p = static_cast(addr); auto Pack1 = [&p](auto len, auto &&writer) { Put32(p, len); p += sizeof(len); writer(p, len); p += len; }; Pack1(head_len, [&](void *p, int len) { head.SerializeToArray(p, len); }); Pack1(body_len, [&](void *p, int len) { body.SerializeToArray(p, len); }); meta()->size_ = 4 + head_len + 4 + body_len; } return addr; } void *Pack(const BHMsgHead &head, const uint32_t head_len, const std::string &body_content) { void *addr = get(); if (addr) { auto p = static_cast(addr); auto Pack1 = [&p](uint32_t len, auto &&writer) { Put32(p, len); p += sizeof(len); writer(p, len); p += len; }; Pack1(head_len, [&](void *p, int len) { head.SerializeToArray(p, len); }); Pack1(body_content.size(), [&](void *p, int len) { memcpy(p, body_content.data(), len); }); meta()->size_ = 4 + head_len + 4 + body_content.size(); } return addr; } void *Pack(const void *src, const size_t size) { void *addr = get(); if (addr && src) { memcpy(addr, src, size); meta()->size_ = size; } return addr; } void *Pack(const std::string &content) { return Pack(content.data(), content.size()); } bool Make(void *addr) { if (!addr) { return false; } offset_ = Addr(addr) - BaseAddr(); return true; } template T *get() const { return offset_ != 0 ? static_cast(Ptr(offset_ + BaseAddr())) : nullptr; } public: explicit ShmMsg(SharedMemory &shm) : offset_(0), pshm_(&shm) {} ShmMsg(const OffsetType offset, SharedMemory &shm) : offset_(offset), pshm_(&shm) {} OffsetType Offset() const { return offset_; } OffsetType &OffsetRef() { return offset_; } SharedMemory &shm() const { return *pshm_; } void swap(ShmMsg &a) { std::swap(offset_, a.offset_); std::swap(pshm_, a.pshm_); } bool valid() const { return offset_ != 0 && meta()->tag_ == kMsgTag; } int64_t id() const { return valid() ? meta()->id_ : 0; } int64_t timestamp() const { return valid() ? meta()->timestamp_.load() : 0; } size_t Size() const { return valid() ? meta()->size_ : 0; } int Count() const { return valid() ? meta()->count_.Get() : 1; } int AddRef() const { return valid() ? meta()->count_.Inc() : 1; } int Release(); void Free(); void reset_managed(const bool val) const { if (valid()) { meta()->managed_ = val; } } template inline bool Make(const BHMsgHead &head, const Body &body) { uint32_t head_len = head.ByteSizeLong(); uint32_t body_len = body.ByteSizeLong(); uint32_t size = sizeof(head_len) + head_len + sizeof(body_len) + body_len; return Make(size) && Pack(head, head_len, body, body_len); } inline bool Make(const BHMsgHead &head, const std::string &body_content) { uint32_t head_len = head.ByteSizeLong(); uint32_t size = sizeof(head_len) + head_len + sizeof(uint32_t) + body_content.size(); return Make(size) && Pack(head, head_len, body_content); } template inline bool Fill(const BHMsgHead &head, const Body &body) { uint32_t head_len = head.ByteSizeLong(); uint32_t body_len = body.ByteSizeLong(); uint32_t size = sizeof(head_len) + head_len + sizeof(body_len) + body_len; return valid() && (meta()->capacity_ >= size) && Pack(head, head_len, body, body_len); } inline bool Make(const void *src, const size_t size) { return Make(Alloc(size, src)); } inline bool Fill(const void *src, const size_t size) { return valid() && (meta()->capacity_ >= size) && Pack(src, size); } inline bool Make(const std::string &content) { return Make(content.data(), content.size()); } inline bool Fill(const std::string &content) { return Fill(content.data(), content.size()); } inline bool Make(const size_t size) { return Make(Alloc(size)); } template static inline std::string Serialize(const BHMsgHead &head, const Body &body) { uint32_t head_len = head.ByteSizeLong(); uint32_t body_len = body.ByteSizeLong(); std::string s(4 + head_len + 4 + body_len, '\0'); size_t pos = 0; auto add1 = [&](auto &&msg, auto &&size) { Put32(&s[pos], size); pos += 4; msg.SerializeToArray(&s[pos], size); pos += size; }; add1(head, head_len); add1(body, body_len); assert(pos == s.size()); return s; } bool ParseHead(BHMsgHead &head) const { auto p = get(); assert(p); uint32_t msg_size = Get32(p); p += 4; return head.ParseFromArray(p, msg_size); } std::string content() const { auto p = get(); return p ? std::string(p, meta()->size_) : std::string(); } std::string body() const { auto p = get(); assert(p); uint32_t size = Get32(p); p += 4; p += size; size = Get32(p); p += 4; return std::string(p, size); } template bool ParseBody(Body &body) const { auto p = get(); assert(p); uint32_t size = Get32(p); p += 4; p += size; size = Get32(p); p += 4; return body.ParseFromArray(p, size); } }; inline void swap(ShmMsg &m1, ShmMsg &m2) { m1.swap(m2); } typedef ShmMsg MsgI; constexpr inline int EncodeCmd(int cmd) { return ((cmd & MaskBits(3)) << 1) | 1; } constexpr inline int DecodeCmd(int64_t msg) { return (msg >> 1) & MaskBits(3); } constexpr inline bool IsCmd(int64_t msg) { return (msg & 1) != 0; } // int64_t pack format: cmd data ,3bit cmd, 1bit flag. enum MsgCmd { eCmdNodeInit = 0, // upto 56bit ssn id eCmdNodeInitReply = 1, // 31bit proc index, eCmdAllocRequest0 = 2, // 8bit size, 4bit socket index, 16bit proc index, 28bit id eCmdAllocReply0 = 3, // 31bit ptr, 28bit id, eCmdFree = 4, // upto 59bit msg id, }; } // namespace bhome_msg #endif // end of include guard: MSG_5BILLZET