Common.hpublic header

Defines struct fair::mq::shmem::DeviceCounter, struct fair::mq::shmem::EventCounter, struct fair::mq::shmem::Heartbeat, struct fair::mq::shmem::MetaHeader, struct fair::mq::shmem::RefCount, struct fair::mq::shmem::RegionBlock, struct fair::mq::shmem::RegionCounter, struct fair::mq::shmem::RegionInfo, struct fair::mq::shmem::SegmentBufferShrink, struct fair::mq::shmem::SegmentInfo, struct fair::mq::shmem::SessionInfo, struct fair::mq::shmem::SharedMemoryError, struct fair::mq::shmem::ShmHeader, struct fair::mq::shmem::ShmHeader::Hdr.

Described in How shared memory works, Sessions and segments, Out-of-band buffers, Tune and debug.

fairmq/shmem/Common.h351 lines · C++API: DeviceCounterAPI: EventCounterAPI: HeartbeatAPI: MetaHeaderAPI: RefCountAPI: RegionBlockAPI: RegionCounterAPI: RegionInfoAPI: SegmentBufferShrinkAPI: SegmentInfoAPI: SessionInfoAPI: SharedMemoryErrorAPI: ShmHeaderAPI: HdrGitHub ↗

1/********************************************************************************2 *    Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH    *3 *                                                                              *4 *              This software is distributed under the terms of the             *5 *              GNU Lesser General Public Licence (LGPL) version 3,             *6 *                  copied verbatim in the file "LICENSE"                       *7 ********************************************************************************/8#ifndef FAIR_MQ_SHMEM_COMMON_H_9#define FAIR_MQ_SHMEM_COMMON_H_10 11#include <atomic>12#include <string>13#include <functional> // std::equal_to14 15#include <boost/functional/hash.hpp>16// #include <boost/interprocess/allocators/adaptive_pool.hpp>17#include <boost/interprocess/allocators/node_allocator.hpp>18#include <boost/interprocess/allocators/allocator.hpp>19#include <boost/interprocess/containers/map.hpp>20#include <boost/interprocess/containers/string.hpp>21#include <boost/interprocess/containers/vector.hpp>22#include <boost/interprocess/indexes/null_index.hpp>23#include <boost/interprocess/managed_shared_memory.hpp>24#include <boost/interprocess/mem_algo/simple_seq_fit.hpp>25#include <boost/unordered_map.hpp>26#include <variant>27 28#include <sys/types.h>29 30#include <fairmq/tools/Strings.h>31 32namespace fair::mq::shmem33{34 35static constexpr uint64_t kManagementSegmentSize = 6553600;36 37struct SharedMemoryError : std::runtime_error { using std::runtime_error::runtime_error; };38 39using SimpleSeqFitSegment = boost::interprocess::basic_managed_shared_memory<char,40    boost::interprocess::simple_seq_fit<boost::interprocess::mutex_family, boost::interprocess::offset_ptr<void>>,41    boost::interprocess::null_index>;42    // boost::interprocess::iset_index>;43using RBTreeBestFitSegment = boost::interprocess::basic_managed_shared_memory<char,44    boost::interprocess::rbtree_best_fit<boost::interprocess::mutex_family, boost::interprocess::offset_ptr<void>>,45    boost::interprocess::null_index>;46    // boost::interprocess::iset_index>;47 48inline std::string MakeShmName(const std::string& shmId, const std::string& type) {49    return std::string("fmq_" + shmId + "_" + type);50}51 52inline std::string MakeShmName(const std::string& shmId, const std::string& type, int index) {53    return std::string(MakeShmName(shmId, type) + "_" + std::to_string(index));54}55 56struct RefCount57{58    explicit RefCount(uint16_t c)59        : count(c)60    {}61 62    uint16_t Get() { return count.load(); }63    uint16_t Increment() { return count.fetch_add(1); }64    uint16_t Decrement() { return count.fetch_sub(1); }65 66    std::atomic<uint16_t> count;67};68 69// Number of nodes allocated at once when the allocator runs out of nodes.70static constexpr size_t numNodesPerBlock = 4096;71// Maximum number of totally free blocks that the adaptive node pool will hold.72// The rest of the totally free blocks will be deallocated with the segment manager.73// static constexpr size_t maxFreeBlocks = 2;74 75using RefCountPool = boost::interprocess::node_allocator<RefCount, boost::interprocess::managed_shared_memory::segment_manager, numNodesPerBlock>;76// using RefCountPool = boost::interprocess::adaptive_pool<RefCount, boost::interprocess::managed_shared_memory::segment_manager, numNodesPerBlock, maxFreeBlocks>;77 78using SegmentManager = boost::interprocess::managed_shared_memory::segment_manager;79using VoidAlloc      = boost::interprocess::allocator<void, SegmentManager>;80using CharAlloc      = boost::interprocess::allocator<char, SegmentManager>;81using Str            = boost::interprocess::basic_string<char, std::char_traits<char>, CharAlloc>;82using StrAlloc       = boost::interprocess::allocator<Str, SegmentManager>;83using StrVector      = boost::interprocess::vector<Str, StrAlloc>;84 85// ShmHeader stores user buffer alignment and the reference count in the following structure:86// [HdrOffset(uint16_t)][Hdr alignment][Hdr][user buffer alignment][user buffer]87// The alignment of Hdr depends on the alignment of std::atomic and is stored in the first entry88struct ShmHeader89{90    struct Hdr91    {92        uint16_t userOffset;93        std::atomic<uint16_t> refCount;94    };95 96    static Hdr* HdrPtr(char* ptr)97    {98        // [HdrOffset(uint16_t)][Hdr alignment][Hdr][user buffer alignment][user buffer]99        //                                     ^100        return reinterpret_cast<Hdr*>(ptr + sizeof(uint16_t) + *(reinterpret_cast<uint16_t*>(ptr)));101    }102 103    static uint16_t HdrPartSize() // [HdrOffset(uint16_t)][Hdr alignment][Hdr]104    {105        // [HdrOffset(uint16_t)][Hdr alignment][Hdr][user buffer alignment][user buffer]106        // <--------------------------------------->107        return sizeof(uint16_t) + alignof(Hdr) + sizeof(Hdr);108    }109 110    static std::atomic<uint16_t>& RefCountPtr(char* ptr)111    {112        // get the ref count ptr from the Hdr113        return HdrPtr(ptr)->refCount;114    }115 116    static uint16_t UserOffset(char* ptr)117    {118        return HdrPartSize() + HdrPtr(ptr)->userOffset;119    }120 121    static char* UserPtr(char* ptr)122    {123        // [HdrOffset(uint16_t)][Hdr alignment][Hdr][user buffer alignment][user buffer]124        //                                                                 ^125        return ptr + HdrPartSize() + HdrPtr(ptr)->userOffset;126    }127 128    static uint16_t RefCount(char* ptr) { return RefCountPtr(ptr).load(); }129    static uint16_t IncrementRefCount(char* ptr) { return RefCountPtr(ptr).fetch_add(1); }130    static uint16_t DecrementRefCount(char* ptr) { return RefCountPtr(ptr).fetch_sub(1); }131 132    static size_t FullSize(size_t size, size_t alignment)133    {134        // [HdrOffset(uint16_t)][Hdr alignment][Hdr][user buffer alignment][user buffer]135        // <--------------------------------------------------------------------------->136        return HdrPartSize() + alignment + size;137    }138 139    static void Construct(char* ptr, size_t alignment)140    {141        // place the Hdr in the aligned location, fill it and store its offset to HdrOffset142 143        // the address alignment should be at least 2144        assert(reinterpret_cast<uintptr_t>(ptr) % 2 == 0);145 146        // offset to the beginning of the Hdr. store it in the beginning147        uint16_t hdrOffset = alignof(Hdr) - ((reinterpret_cast<uintptr_t>(ptr) + sizeof(uint16_t)) % alignof(Hdr));148        memcpy(ptr, &hdrOffset, sizeof(hdrOffset));149 150        // offset to the beginning of the user buffer, store in Hdr together with the ref count151        uint16_t userOffset = alignment - ((reinterpret_cast<uintptr_t>(ptr) + HdrPartSize()) % alignment);152        new(ptr + sizeof(uint16_t) + hdrOffset) Hdr{ userOffset, std::atomic<uint16_t>(1) };153    }154 155    static void Destruct(char* ptr) { RefCountPtr(ptr).~atomic(); }156};157 158struct MetaHeader159{160    size_t fSize; // size of the shm buffer161    size_t fHint; // user-defined value, given by the user on message creation and returned to the user on "buffer no longer needed"-callbacks162    boost::interprocess::managed_shared_memory::handle_t fHandle; // handle to shm buffer, convertible to shm buffer ptr163    mutable boost::interprocess::managed_shared_memory::handle_t fShared; // handle to the buffer storing the ref count for shared buffers164    uint16_t fRegionId; // id of the unmanaged region165    mutable uint16_t fSegmentId; // id of the managed segment166    bool fManaged; // true = managed segment, false = unmanaged region167};168 169enum class AllocationAlgorithm : int170{171    rbtree_best_fit,172    simple_seq_fit173};174 175struct RegionInfo176{177    RegionInfo(const char* path, int flags, uint64_t userFlags, uint64_t size, uint64_t rcSegmentSize, const VoidAlloc& alloc)178        : fPath(path, alloc)179        , fCreationFlags(flags)180        , fUserFlags(userFlags)181        , fSize(size)182        , fRCSegmentSize(rcSegmentSize)183    {}184 185    Str fPath;186    int fCreationFlags;187    uint64_t fUserFlags;188    uint64_t fSize;189    uint64_t fRCSegmentSize;190    bool fDestroyed{false};191};192 193using Uint16RegionInfoPairAlloc = boost::interprocess::allocator<std::pair<const uint16_t, RegionInfo>, SegmentManager>;194using Uint16RegionInfoMap = boost::interprocess::map<uint16_t, RegionInfo, std::less<uint16_t>, Uint16RegionInfoPairAlloc>;195using Uint16RegionInfoHashMap = boost::unordered_map<uint16_t, RegionInfo, boost::hash<uint16_t>, std::equal_to<uint16_t>, Uint16RegionInfoPairAlloc>;196 197struct SegmentInfo198{199    SegmentInfo(AllocationAlgorithm aa)200        : fAllocationAlgorithm(aa)201    {}202 203    AllocationAlgorithm fAllocationAlgorithm;204};205 206struct SessionInfo207{208    SessionInfo(const char* sessionName, int creatorId, const VoidAlloc& alloc)209        : fSessionName(sessionName, alloc)210        , fCreatorId(creatorId)211    {}212 213    Str fSessionName;214    int fCreatorId;215};216 217using Uint16SegmentInfoPairAlloc = boost::interprocess::allocator<std::pair<const uint16_t, SegmentInfo>, SegmentManager>;218using Uint16SegmentInfoHashMap = boost::unordered_map<uint16_t, SegmentInfo, boost::hash<uint16_t>, std::equal_to<uint16_t>, Uint16SegmentInfoPairAlloc>;219// using Uint16SegmentInfoMap = boost::interprocess::map<uint16_t, SegmentInfo, std::less<uint16_t>, Uint16SegmentInfoPairAlloc>;220 221struct DeviceCounter222{223    DeviceCounter(unsigned int c)224        : fCount(c)225    {}226 227    std::atomic<unsigned int> fCount;228};229 230struct EventCounter231{232    EventCounter(uint64_t c)233        : fCount(c)234    {}235 236    std::atomic<uint64_t> fCount;237};238 239struct Heartbeat240{241    Heartbeat(uint64_t c)242        : fCount(c)243    {}244 245    std::atomic<uint64_t> fCount;246};247 248struct RegionCounter249{250    RegionCounter(uint16_t c)251        : fCount(c)252    {}253 254    std::atomic<uint16_t> fCount;255};256 257#ifdef FAIRMQ_DEBUG_MODE258struct MsgCounter259{260    MsgCounter()261        : fCount(0)262    {}263 264    MsgCounter(unsigned int c)265        : fCount(c)266    {}267 268    std::atomic<unsigned int> fCount;269};270 271using Uint16MsgCounterPairAlloc = boost::interprocess::allocator<std::pair<const uint16_t, MsgCounter>, SegmentManager>;272using Uint16MsgCounterHashMap = boost::unordered_map<uint16_t, MsgCounter, boost::hash<uint16_t>, std::equal_to<uint16_t>, Uint16MsgCounterPairAlloc>;273 274struct MsgDebug275{276    MsgDebug()277        : fPid(0)278        , fSize(0)279        , fCreationTime(0)280    {}281 282    MsgDebug(pid_t pid, size_t size, const uint64_t creationTime)283        : fPid(pid)284        , fSize(size)285        , fCreationTime(creationTime)286    {}287 288    pid_t fPid;289    size_t fSize;290    uint64_t fCreationTime;291};292 293using SizetMsgDebugPairAlloc = boost::interprocess::allocator<std::pair<const size_t, MsgDebug>, SegmentManager>;294// using SizetMsgDebugHashMap = boost::unordered_map<size_t, MsgDebug, boost::hash<size_t>, std::equal_to<size_t>, SizetMsgDebugPairAlloc>;295using SizetMsgDebugMap = boost::interprocess::map<size_t, MsgDebug, std::less<size_t>, SizetMsgDebugPairAlloc>;296using Uint16MsgDebugMapPairAlloc = boost::interprocess::allocator<std::pair<const uint16_t, SizetMsgDebugMap>, SegmentManager>;297using Uint16MsgDebugMapHashMap = boost::unordered_map<uint16_t, SizetMsgDebugMap, boost::hash<uint16_t>, std::equal_to<uint16_t>, Uint16MsgDebugMapPairAlloc>;298#endif299 300struct RegionBlock301{302    RegionBlock() = default;303 304    RegionBlock(boost::interprocess::managed_shared_memory::handle_t handle, size_t size, size_t hint)305        : fHandle(handle)306        , fSize(size)307        , fHint(hint)308    {}309 310    boost::interprocess::managed_shared_memory::handle_t fHandle{};311    size_t fSize = 0;312    size_t fHint = 0;313};314 315// find id for unique shmem name:316// a hash of user id + session id, truncated to 8 characters (to accommodate for name size limit on some systems (MacOS)).317std::string makeShmIdStr(const std::string& sessionId, const std::string& userId);318std::string makeShmIdStr(const std::string& sessionId);319std::string makeShmIdStr(uint64_t val);320uint64_t makeShmIdUint64(const std::string& sessionId);321 322struct SegmentBufferShrink323{324    SegmentBufferShrink(const size_t _new_size, char* _local_ptr)325        : new_size(_new_size)326        , local_ptr(_local_ptr)327    {}328 329    template<typename S>330    char* operator()(S& s) const331    {332        boost::interprocess::managed_shared_memory::size_type shrunk_size = new_size;333        return s.template allocation_command<char>(boost::interprocess::shrink_in_place, new_size + 128, shrunk_size, local_ptr);334    }335 336    const size_t new_size;337    mutable char* local_ptr;338};339 340} // namespace fair::mq::shmem341 342namespace fair::mq { class TransportFactory; }343 344namespace fair::mq::shmem {345// Resolve a MetaHeader (received over a side channel) to the local data pointer.346// The caller is responsible for ensuring the backing buffer remains alive for the347// duration of access; FairMQ provides no refcount protection for this path.348char* GetDataAddressFromHandle(fair::mq::TransportFactory& factory, const MetaHeader& meta);349} // namespace fair::mq::shmem350 351#endif /* FAIR_MQ_SHMEM_COMMON_H_ */