From 9bf199a4770b08c03d553129757d960b605e598a Mon Sep 17 00:00:00 2001 From: lichao <lichao@aiotlink.com> Date: 星期五, 14 五月 2021 18:05:21 +0800 Subject: [PATCH] add center info at fixed address in shm. --- src/defs.cpp | 79 ++++++++++++++++++++++++++++++++++++++- 1 files changed, 76 insertions(+), 3 deletions(-) diff --git a/src/defs.cpp b/src/defs.cpp index 450349e..b812b65 100644 --- a/src/defs.cpp +++ b/src/defs.cpp @@ -18,6 +18,10 @@ #include "defs.h" #include "msg.h" #include "shm_msg_queue.h" +#include "socket.h" +#include <boost/uuid/random_generator.hpp> +#include <boost/uuid/string_generator.hpp> +#include <boost/uuid/uuid.hpp> namespace { @@ -70,7 +74,77 @@ const int kAllocIndexLen = sizeof(AllocSizeIndex) / sizeof(AllocSizeIndex[0]); static_assert(kAllocIndexLen == 256, "Make sure alloc 8 bit is enough."); static_assert(AllocSizeIndex[255] > uint32_t(-1), "Make sure alloc size correct."); + +const int64_t kCenterInfoFixedAddress = 1024 * 4; + +const boost::uuids::uuid kCenterInfoTag = boost::uuids::string_generator()("fc5007bd-0e62-4d91-95dc-948cf1f02e5a"); +struct CenterMetaInfo { + boost::uuids::uuid tag_; + CenterInfo info_; +}; + +int64_t Addr(void *ptr) { return reinterpret_cast<int64_t>(ptr); } +// void *Ptr(const int64_t offset) { return reinterpret_cast<void *>(offset); } +template <class T = void> +T *Ptr(const int64_t offset) { return reinterpret_cast<T *>(offset); } + } // namespace + +CenterInfo *GetCenterInfo(bhome_shm::SharedMemory &shm) +{ + auto pmeta = Ptr<CenterMetaInfo>(kCenterInfoFixedAddress + Addr(shm.get_address())); + if (pmeta->tag_ == kCenterInfoTag) { + return &pmeta->info_; + } + return nullptr; +} + +// put center info at fixed memory position. +// as boost shm find object (find socket/mq by id, etc...) also locks inside, +// which node might crash inside and cause deadlock. +bool CenterInit(bhome_shm::SharedMemory &shm) +{ + Mutex *mutex = shm.Create<Mutex>("shm_center_lock"); + if (!mutex || !mutex->try_lock()) { + return false; + } + DEFER1(mutex->unlock()); + + auto pmeta = Ptr<CenterMetaInfo>(kCenterInfoFixedAddress + Addr(shm.get_address())); + if (pmeta->tag_ == kCenterInfoTag) { + return true; + } else { + auto base = Addr(shm.get_address()); + auto offset = kCenterInfoFixedAddress; + void *p = shm.Alloc(offset * 2); + if (Addr(p) - base <= offset) { + pmeta = new (Ptr(offset + base)) CenterMetaInfo; + auto &info = pmeta->info_; + + auto InitMQ = [&](auto &mq, auto &&id) { + mq.id_ = id; + ShmSocket tmp(shm, id, 16); + mq.offset_ = tmp.AbsAddr(); + }; + + int id = 100; + auto NextId = [&]() { return ++id; }; + InitMQ(info.mq_sender_, NextId()); + InitMQ(info.mq_center_, NextId()); + InitMQ(info.mq_bus_, NextId()); + InitMQ(info.mq_init_, NextId()); + + pmeta->tag_ = kCenterInfoTag; + return true; + } + } + return false; +} + +uint64_t BHGlobalSenderAddress() { return GetCenterInfo(BHomeShm())->mq_sender_.id_; } +uint64_t BHTopicCenterAddress() { return GetCenterInfo(BHomeShm())->mq_center_.id_; } +uint64_t BHTopicBusAddress() { return GetCenterInfo(BHomeShm())->mq_bus_.id_; } +uint64_t BHCenterReplyAddress() { return GetCenterInfo(BHomeShm())->mq_init_.id_; } int64_t CalcAllocIndex(int64_t size) { @@ -93,9 +167,8 @@ bool GlobalInit(bhome_shm::SharedMemory &shm) { MsgI::BindShm(shm); - typedef std::atomic<MQId> IdSrc; - IdSrc *psrc = shm.FindOrCreate<IdSrc>("shmqIdSrc0", 100000); - return psrc && ShmMsgQueue::SetData(*psrc); + CenterInfo *pinfo = GetCenterInfo(shm); + return pinfo && ShmMsgQueue::SetData(pinfo->mqid_); } void SetLastError(const int ec, const std::string &msg) -- Gitblit v1.8.0