From a76a94009d78a6a41654335dcb9202fb31659de0 Mon Sep 17 00:00:00 2001 From: lichao <lichao@aiotlink.com> Date: 星期四, 25 三月 2021 13:23:16 +0800 Subject: [PATCH] bulid msg, refactor. --- src/shm.h | 7 ++ src/msg.h | 19 ++++-- src/shm_queue.h | 1 utest/utest.cpp | 28 ++++++-- src/shm_queue.cpp | 46 ++++++++------ src/msg.cpp | 32 ++++++++++ 6 files changed, 98 insertions(+), 35 deletions(-) diff --git a/src/msg.cpp b/src/msg.cpp index e8c6d26..4ddb726 100644 --- a/src/msg.cpp +++ b/src/msg.cpp @@ -32,4 +32,36 @@ *static_cast<MsgMetaV1*>(p) = *this; } +bool Msg::Build(SharedMemory &shm, const MQId &src_id, const void *data, const size_t size, const bool refcount) +{ + if (!data || !size) { + return false; + } + void *p = shm.Alloc(sizeof(MsgMetaV1) + size); + if (!p) { + return false; + } + RefCount *rc = 0; + if (refcount) { + rc = shm.New<RefCount>(); + if (!rc) { + shm.Dealloc(p); + return false; + } + } + MsgMetaV1 meta; + meta.data_size_ = size; + meta.src_id_ = src_id; + meta.Pack(p); + memcpy(static_cast<char *>(p) + sizeof(meta), data, size); + Msg(p, rc).swap(*this); + return true; + +} + +void Msg::FreeFrom(SharedMemory &shm) +{ + shm.Dealloc(ptr_); + shm.Delete(count_); +} } // namespace bhome_shm diff --git a/src/msg.h b/src/msg.h index 214bb72..8a820bd 100644 --- a/src/msg.h +++ b/src/msg.h @@ -21,6 +21,7 @@ #include <stdint.h> #include "shm.h" #include <boost/interprocess/offset_ptr.hpp> +#include <boost/uuid/uuid_generators.hpp> namespace bhome_shm { @@ -32,6 +33,8 @@ kMsgTypeMaxValue }; +typedef boost::uuids::uuid MQId; + const uint32_t kMsgMetaTag = 0xf1e2d3c4; struct MsgMetaV1 { @@ -39,10 +42,10 @@ uint16_t type_ = kMsgTypeNormal; // msg type. uint32_t tag_ = kMsgMetaTag; uint32_t data_size_ = 0; - unsigned char src_id_[16] = {0}; + MQId src_id_; // more fields add at end, must not change - MsgMetaV1(){} + MsgMetaV1():src_id_(boost::uuids::nil_uuid()){} bool Parse(const void *p); void Pack(void *p); }; @@ -63,7 +66,6 @@ private: offset_ptr<void> ptr_; offset_ptr<RefCount> count_; - void swap(Msg &a) { std::swap(ptr_, a.ptr_); std::swap(count_, a.count_); } public: class CountGuard : private boost::noncopyable { Msg &msg_; @@ -73,6 +75,7 @@ }; Msg(void *p=0, RefCount *c=0):ptr_(p), count_(c) {} + void swap(Msg &a) { std::swap(ptr_, a.ptr_); std::swap(count_, a.count_); } // ~Msg() { RemoveRef(); } // Msg(const Msg &a):ptr_(a.ptr_), count_(a.count_) { AddRef(); } @@ -81,11 +84,15 @@ // Msg & operator = (Msg &&a) { Msg(std::move(a)).swap(*this); } template <class T = void> T *get() { return static_cast<T*>(ptr_.get()); } - int AddRef() { return count_ ? count_->Inc() : 0; } - int RemoveRef() { return count_ ? count_->Dec() : 0; } - int Count() { return count_ ? count_->Get() : 0; } + int AddRef() const { return count_ ? count_->Inc() : 0; } + int RemoveRef() const{ return count_ ? count_->Dec() : 0; } + int Count() const { return count_ ? count_->Get() : 0; } + bool Build(SharedMemory &shm, const MQId &src_id, const void *p, const size_t size, const bool refcount); + void FreeFrom(SharedMemory &shm); }; +Msg BuildMsg(const MQId &src, const void *p, const size_t size); + } // namespace bhome_shm diff --git a/src/shm.h b/src/shm.h index 91a339d..808ed5d 100644 --- a/src/shm.h +++ b/src/shm.h @@ -54,6 +54,11 @@ {} std::string name() const { return name_; } bool Remove() { return Remove(name()); } + + void *Alloc(const size_t size) { return allocate(size, std::nothrow); } + void Dealloc(void *p) { if(p) { deallocate(p); } } + template<class T> void Dealloc(offset_ptr<T> ptr) { return Dealloc(ptr.get()); } + template <class T, class ...Params> T * New(Params const&...params) { return construct<T>(anonymous_instance, std::nothrow)(params...); } template <class T> void Delete(T *p) { if (p) { destroy_ptr<T>(p); }; } template <class T> void Delete(offset_ptr<T> p) { Delete(p.get()); } @@ -75,7 +80,7 @@ bool IsOk() const { return pdata_; } protected: - ShmType &shm() { return shm_; } + ShmType &shm() const { return shm_; } public: template <class...Params> ShmObject(ShmType &segment, const std::string &name, Params&&...t): diff --git a/src/shm_queue.cpp b/src/shm_queue.cpp index 263fd94..1446446 100644 --- a/src/shm_queue.cpp +++ b/src/shm_queue.cpp @@ -45,26 +45,32 @@ Remove(); } +bool ShmMsgQueue::Send(const MQId &remote_id, const Msg &msg, const int timeout_ms) +{ + Queue *remote = find(MsgQIdToName(remote_id)); + + return remote && remote->Write(msg, timeout_ms); + + if(!remote) { + return false; + } + msg.AddRef(); + if (remote->Write(msg, timeout_ms)) { + return true; + } else { + msg.RemoveRef(); + return false; + } +} + bool ShmMsgQueue::Send(const MQId &remote_id, const void *data, const size_t size, const int timeout_ms) { - if (data && size) { - Queue *remote = find(MsgQIdToName(remote_id)); - if (remote) { - void *p = shm().allocate(sizeof(MsgMetaV1) + size, std::nothrow); - bool r = false; - if (p) { - MsgMetaV1 meta; - meta.data_size_ = size; - memcpy(meta.src_id_, &Id(), sizeof(MQId)); - meta.Pack(p); - - memcpy(static_cast<char*>(p) + sizeof(meta), data, size); - if (remote->Write(p, timeout_ms)) { - return true; - } else { - shm().deallocate(p); - } - } + Msg msg; + if (msg.Build(shm(), Id(), data, size, false)) { + if (Send(remote_id, msg, timeout_ms)) { + return true; + } else { + msg.FreeFrom(shm()); } } return false; @@ -76,10 +82,10 @@ if (Read(msg, timeout_ms)) { auto ptr = msg.get<char>(); if (ptr) { - DEFER1(shm().deallocate(ptr);); + DEFER1(shm().Dealloc(ptr);); MsgMetaV1 meta; meta.Parse(ptr); - memcpy(&source_id, meta.src_id_, sizeof(MQId)); + source_id = meta.src_id_; size = meta.data_size_; data = malloc(size); if (data) { diff --git a/src/shm_queue.h b/src/shm_queue.h index a98fcf4..d0eb972 100644 --- a/src/shm_queue.h +++ b/src/shm_queue.h @@ -90,6 +90,7 @@ bool Send(const MQId &remote_id, const void *data, const size_t size, const int timeout_ms); bool Recv(MQId &source_id, void *&data, size_t &size, const int timeout_ms); const MQId &Id() const { return data()->Id(); } + bool Send(const MQId &remote_id, const Msg &msg, const int timeout_ms); }; } // namespace bhome_shm diff --git a/utest/utest.cpp b/utest/utest.cpp index 57b0e9c..61e6437 100644 --- a/utest/utest.cpp +++ b/utest/utest.cpp @@ -54,6 +54,21 @@ ShmRemover auto_remove(shm_name); SharedMemory shm(shm_name, 1024*1024*10); + offset_ptr<const void> p; + BOOST_CHECK(!p); + BOOST_CHECK(p.get() == 0); + p = 0; + BOOST_CHECK(!p); + BOOST_CHECK(p.get() == 0); + const char *str = "basic"; + p = str; + BOOST_CHECK(p); + BOOST_CHECK(p.get() == str); + p = 0; + BOOST_CHECK(!p); + BOOST_CHECK(p.get() == 0); + + auto Avail = [&]() { return shm.get_free_memory(); }; auto init_avail = Avail(); @@ -77,8 +92,8 @@ { auto old = Avail(); - void *p = shm.allocate(1024); - shm.deallocate(p); + void *p = shm.Alloc(1024); + shm.Dealloc(p); BOOST_CHECK_EQUAL(old, Avail()); } @@ -129,7 +144,7 @@ ShmRemover auto_remove(shm_name); SharedMemory shm(shm_name, 1024*1024); - Msg m0(shm.allocate(1000), shm.New<RefCount>()); + Msg m0(shm.Alloc(1000), shm.New<RefCount>()); BOOST_CHECK_EQUAL(m0.AddRef(), 1); Msg m1 = m0; BOOST_CHECK_EQUAL(m1.AddRef(), 2); @@ -155,14 +170,11 @@ BOOST_CHECK_EQUAL(head.type_, kMsgTypeNormal); BOOST_CHECK_EQUAL(head.tag_, kMsgMetaTag); BOOST_CHECK_EQUAL(head.data_size_, 0); - BOOST_CHECK_EQUAL(head.src_id_[5], 0); + BOOST_CHECK(head.src_id_ == boost::uuids::nil_uuid()); head.data_size_ = 100; - auto rand_id = boost::uuids::random_generator()(); - memcpy(head.src_id_, &rand_id, sizeof(rand_id)); + head.src_id_ = boost::uuids::random_generator()(); head.type_ = 123; - - BOOST_CHECK_EQUAL(sizeof(head.src_id_), sizeof(rand_id)); char buf[100] = {0}; head.Pack(buf); -- Gitblit v1.8.0