Source · fairmq/shmem/
Manager.hpublic header
Defines class fair::mq::shmem::Manager.
Described in Run a benchmark, Handle errors and timeouts, Test your devices, How shared memory works, Monitor and clean up, Unmanaged regions, Sessions and segments, Out-of-band buffers, Tune and debug.
1/********************************************************************************2 * Copyright (C) 2014-2023 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 9#ifndef FAIR_MQ_SHMEM_MANAGER_H_10#define FAIR_MQ_SHMEM_MANAGER_H_11 12#include "Common.h"13#include "Monitor.h"14#include "UnmanagedRegion.h"15#include <fairmq/Message.h>16#include <fairmq/ProgOptions.h>17#include <fairmq/tools/Strings.h>18#include <fairmq/Transports.h>19 20#include <fairlogger/Logger.h>21 22#include <boost/interprocess/ipc/message_queue.hpp>23#include <boost/interprocess/managed_shared_memory.hpp>24#include <boost/interprocess/sync/interprocess_condition.hpp>25#include <boost/interprocess/sync/interprocess_mutex.hpp>26#include <boost/interprocess/sync/named_mutex.hpp>27 28#include <algorithm> // max29#include <chrono>30#include <condition_variable>31#include <cstddef> // max_align_t, std::size_t32#include <cstdlib> // getenv33#include <cstring> // memcpy34#include <memory> // make_unique35#include <mutex>36#include <set>37#include <sstream>38#include <stdexcept>39#include <string>40#include <thread>41#include <tuple>42#include <unordered_map>43#include <utility> // pair44#include <variant>45#include <vector>46 47#include <unistd.h> // getuid48#include <sys/types.h> // getuid49#include <sys/mman.h> // mlock50 51namespace fair::mq::shmem52{53 54class Manager55{56 public:57 Manager(const std::string& sessionName, size_t size, const ProgOptions* config)58 : fShmId64(config ? config->GetProperty<uint64_t>("shmid", makeShmIdUint64(sessionName)) : makeShmIdUint64(sessionName))59 , fShmId(makeShmIdStr(fShmId64))60 , fSegmentId(config ? config->GetProperty<uint16_t>("shm-segment-id", 0) : 0)61 , fManagementSegment(boost::interprocess::open_or_create, MakeShmName(fShmId, "mng").c_str(), kManagementSegmentSize)62 , fShmVoidAlloc(fManagementSegment.get_segment_manager())63 , fShmMtx(fManagementSegment.find_or_construct<boost::interprocess::interprocess_mutex>(boost::interprocess::unique_instance)())64 , fNumObservedEvents(0)65 , fDeviceCounter(nullptr)66 , fEventCounter(nullptr)67 , fShmSegments(nullptr)68 , fShmRegions(nullptr)69#ifdef FAIRMQ_DEBUG_MODE70 , fMsgDebug(nullptr)71 , fShmMsgCounters(nullptr)72 , fMsgCounterNew(0)73 , fMsgCounterDelete(0)74#endif75 , fBeatTheHeart(true)76 , fRegionEventsSubscriptionActive(false)77 , fInterrupted(false)78 , fBadAllocMaxAttempts(1)79 , fBadAllocAttemptIntervalInMs(config ? config->GetProperty<int>("bad-alloc-attempt-interval", 50) : 50)80 , fNoCleanup(config ? config->GetProperty<bool>("shm-no-cleanup", false) : false)81 , fMetadataMsgSize(config ? config->GetProperty<std::size_t>("shm-metadata-msg-size", 0) : 0)82 {83 using namespace boost::interprocess;84 85 LOG(debug) << "Generated shmid '" << fShmId << "' out of session id '" << sessionName << "'.";86 87 if (config) {88 // if 'shm-throw-bad-alloc' is explicitly set to false (true is default), ignore other settings89 if (config->Count("shm-throw-bad-alloc") && config->GetProperty<bool>("shm-throw-bad-alloc") == false) {90 fBadAllocMaxAttempts = -1;91 } else if (config->Count("bad-alloc-max-attempts")) {92 // if 'bad-alloc-max-attempts' is provided, use it93 // this can override the default 'shm-throw-bad-alloc'==true if the value of 'bad-alloc-max-attempts' is -194 fBadAllocMaxAttempts = config->GetProperty<int>("bad-alloc-max-attempts");95 }96 // otherwise leave fBadAllocMaxAttempts at 1 (the original default, set in the initializer list)97 }98 99 bool mlockSegment = false;100 bool mlockSegmentOnCreation = false;101 bool zeroSegment = false;102 bool zeroSegmentOnCreation = false;103 bool autolaunchMonitor = false;104 std::string allocationAlgorithm("rbtree_best_fit");105 if (config) {106 mlockSegment = config->GetProperty<bool>("shm-mlock-segment", mlockSegment);107 mlockSegmentOnCreation = config->GetProperty<bool>("shm-mlock-segment-on-creation", mlockSegmentOnCreation);108 zeroSegment = config->GetProperty<bool>("shm-zero-segment", zeroSegment);109 zeroSegmentOnCreation = config->GetProperty<bool>("shm-zero-segment-on-creation", zeroSegmentOnCreation);110 autolaunchMonitor = config->GetProperty<bool>("shm-monitor", autolaunchMonitor);111 allocationAlgorithm = config->GetProperty<std::string>("shm-allocation", allocationAlgorithm);112 } else {113 LOG(debug) << "ProgOptions not available! Using defaults.";114 }115 116 if (autolaunchMonitor) {117 boost::interprocess::scoped_lock<boost::interprocess::interprocess_mutex> lock(*fShmMtx);118 StartMonitor(fShmId);119 }120 121 fHeartbeatThread = std::thread(&Manager::Heartbeats, this);122 123 try {124 boost::interprocess::scoped_lock<boost::interprocess::interprocess_mutex> lock(*fShmMtx);125 126 SessionInfo* sessionInfo = fManagementSegment.find<SessionInfo>(unique_instance).first;127 if (sessionInfo) {128 LOG(debug) << "session info found, name: " << sessionInfo->fSessionName << ", creator id: " << sessionInfo->fCreatorId;129 } else {130 LOG(debug) << "no session info found, creating and initializing";131 sessionInfo = fManagementSegment.construct<SessionInfo>(unique_instance)(sessionName.c_str(), geteuid(), fShmVoidAlloc);132 LOG(debug) << "initialized session info, name: " << sessionInfo->fSessionName << ", creator id: " << sessionInfo->fCreatorId;133 }134 135 fEventCounter = fManagementSegment.find<EventCounter>(unique_instance).first;136 if (fEventCounter) {137 LOG(trace) << "event counter found: " << fEventCounter->fCount;138 } else {139 LOG(trace) << "no event counter found, creating one and initializing with 0";140 fEventCounter = fManagementSegment.construct<EventCounter>(unique_instance)(0);141 LOG(trace) << "initialized event counter with: " << fEventCounter->fCount;142 }143 144 fDeviceCounter = fManagementSegment.find<DeviceCounter>(unique_instance).first;145 if (fDeviceCounter) {146 LOG(trace) << "device counter found, with value of " << fDeviceCounter->fCount << ". incrementing.";147 (fDeviceCounter->fCount)++;148 LOG(trace) << "incremented device counter, now: " << fDeviceCounter->fCount;149 } else {150 LOG(trace) << "no device counter found, creating one and initializing with 1";151 fDeviceCounter = fManagementSegment.construct<DeviceCounter>(unique_instance)(1);152 LOG(trace) << "initialized device counter with: " << fDeviceCounter->fCount;153 }154 155 fShmSegments = fManagementSegment.find_or_construct<Uint16SegmentInfoHashMap>(unique_instance)(fShmVoidAlloc);156 fShmRegions = fManagementSegment.find_or_construct<Uint16RegionInfoHashMap>(unique_instance)(fShmVoidAlloc);157 158 bool createdSegment = false;159 160 try {161 std::string segmentName = MakeShmName(fShmId, "m", fSegmentId);162 auto it = fShmSegments->find(fSegmentId);163 if (it == fShmSegments->end()) {164 // no segment with given id exists, creating165 if (allocationAlgorithm == "rbtree_best_fit") {166 fSegments.emplace(fSegmentId, RBTreeBestFitSegment(open_or_create, segmentName.c_str(), size));167 fShmSegments->emplace(fSegmentId, AllocationAlgorithm::rbtree_best_fit);168 } else if (allocationAlgorithm == "simple_seq_fit") {169 fSegments.emplace(fSegmentId, SimpleSeqFitSegment(open_or_create, segmentName.c_str(), size));170 fShmSegments->emplace(fSegmentId, AllocationAlgorithm::simple_seq_fit);171 }172 if (mlockSegmentOnCreation) {173 MlockSegment(fSegmentId);174 }175 if (zeroSegmentOnCreation) {176 ZeroSegment(fSegmentId);177 }178 createdSegment = true;179 } else {180 // found segment with the given id, opening181 if (it->second.fAllocationAlgorithm == AllocationAlgorithm::rbtree_best_fit) {182 fSegments.emplace(fSegmentId, RBTreeBestFitSegment(open_or_create, segmentName.c_str(), size));183 if (allocationAlgorithm != "rbtree_best_fit") {184 LOG(warn) << "Allocation algorithm of the opened segment is rbtree_best_fit, but requested is " << allocationAlgorithm << ". Ignoring requested setting.";185 allocationAlgorithm = "rbtree_best_fit";186 }187 } else {188 fSegments.emplace(fSegmentId, SimpleSeqFitSegment(open_or_create, segmentName.c_str(), size));189 if (allocationAlgorithm != "simple_seq_fit") {190 LOG(warn) << "Allocation algorithm of the opened segment is simple_seq_fit, but requested is " << allocationAlgorithm << ". Ignoring requested setting.";191 allocationAlgorithm = "simple_seq_fit";192 }193 }194 }195 LOG(debug) << (createdSegment ? "Created" : "Opened") << " managed shared memory segment " << "fmq_" << fShmId << "_m_" << fSegmentId196 << ". Size: " << std::visit([](auto& s) { return s.get_size(); }, fSegments.at(fSegmentId)) << " bytes."197 << " Available: " << std::visit([](auto& s) { return s.get_free_memory(); }, fSegments.at(fSegmentId)) << " bytes."198 << " Allocation algorithm: " << allocationAlgorithm;199 } catch (interprocess_exception& bie) {200 LOG(error) << "Failed to create/open shared memory segment '" << "fmq_" << fShmId << "_m_" << fSegmentId << "': " << bie.what();201 throw TransportError(tools::ToString("Failed to create/open shared memory segment '", "fmq_", fShmId, "_m_", fSegmentId, "': ", bie.what()));202 }203 204 if (mlockSegment) {205 MlockSegment(fSegmentId);206 }207 if (zeroSegment) {208 ZeroSegment(fSegmentId);209 }210 211 if (createdSegment) {212 (fEventCounter->fCount)++;213 }214 215#ifdef FAIRMQ_DEBUG_MODE216 fMsgDebug = fManagementSegment.find_or_construct<Uint16MsgDebugMapHashMap>(unique_instance)(fShmVoidAlloc);217 fShmMsgCounters = fManagementSegment.find_or_construct<Uint16MsgCounterHashMap>(unique_instance)(fShmVoidAlloc);218#endif219 } catch (...) {220 StopHeartbeats();221 CleanupIfLast();222 throw;223 }224 }225 226 Manager() = delete;227 Manager(const Manager&) = delete;228 Manager(Manager&&) = delete;229 Manager& operator=(const Manager&) = delete;230 Manager& operator=(Manager&&) = delete;231 232 void ZeroSegment(uint16_t id)233 {234 LOG(debug) << "Zeroing the managed segment free memory...";235 std::visit([](auto& s) { return s.zero_free_memory(); }, fSegments.at(id));236 LOG(debug) << "Successfully zeroed the managed segment free memory.";237 }238 239 void MlockSegment(uint16_t id)240 {241 LOG(debug) << "Locking the managed segment memory pages...";242 if (mlock(243 std::visit([](auto& s) { return s.get_address(); }, fSegments.at(id)),244 std::visit([](auto& s) { return s.get_size(); }, fSegments.at(id))) == -1) {245 LOG(error) << "Could not lock the managed segment memory. Code: " << errno << ", reason: " << strerror(errno);246 throw TransportError(tools::ToString("Could not lock the managed segment memory: ", strerror(errno)));247 }248 LOG(debug) << "Successfully locked the managed segment memory pages.";249 }250 251 private:252 static bool SpawnShmMonitor(const std::string& id);253 254 public:255 static void StartMonitor(const std::string& id)256 {257 using namespace boost::interprocess;258 try {259 named_mutex monitorStatus(open_only, MakeShmName(id, "ms").c_str());260 LOG(debug) << "Found fairmq-shmmonitor for shared memory id " << id;261 } catch (interprocess_exception&) {262 LOG(debug) << "no fairmq-shmmonitor found for shared memory id " << id << ", starting...";263 264 if (SpawnShmMonitor(id)) {265 int numTries = 0;266 do {267 try {268 named_mutex monitorStatus(open_only, MakeShmName(id, "ms").c_str());269 LOG(debug) << "Started fairmq-shmmonitor for shared memory id " << id;270 break;271 } catch (interprocess_exception&) {272 std::this_thread::sleep_for(std::chrono::milliseconds(10));273 if (++numTries > 1000) {274 LOG(error) << "Did not get response from fairmq-shmmonitor after " << 10 * 1000 << " milliseconds. Exiting.";275 throw std::runtime_error(tools::ToString("Did not get response from fairmq-shmmonitor after ", 10 * 1000, " milliseconds. Exiting."));276 }277 }278 } while (true);279 }280 }281 }282 283 void Interrupt() { fInterrupted.store(true); }284 void Resume() { fInterrupted.store(false); }285 void Reset()286 {287#ifdef FAIRMQ_DEBUG_MODE288 auto diff = fMsgCounterNew.load() - fMsgCounterDelete.load();289 if (diff != 0) {290 LOG(error) << "Message counter during Reset expected to be 0, found: " << diff;291 throw MessageError(tools::ToString("Message counter during Reset expected to be 0, found: ", diff));292 }293#endif294 }295 bool Interrupted() { return fInterrupted.load(); }296 297 std::pair<UnmanagedRegion*, uint16_t> CreateRegion(size_t size,298 RegionCallback callback,299 RegionBulkCallback bulkCallback,300 RegionConfig cfg)301 {302 using namespace boost::interprocess;303 try {304 std::pair<UnmanagedRegion*, uint16_t> result;305 306 {307 // lock order: fLocalRegionsMtx before fShmMtx (see GetRegion())308 std::lock_guard<std::mutex> lock(fLocalRegionsMtx);309 boost::interprocess::scoped_lock<boost::interprocess::interprocess_mutex> shmLock(*fShmMtx);310 311 if (!cfg.id.has_value()) {312 RegionCounter* rc = fManagementSegment.find<RegionCounter>(unique_instance).first;313 314 if (rc) {315 LOG(trace) << "region counter found, with value of " << rc->fCount << ". incrementing.";316 (rc->fCount)++;317 LOG(trace) << "incremented region counter, now: " << rc->fCount;318 } else {319 LOG(trace) << "no region counter found, creating one and initializing with 1024";320 rc = fManagementSegment.construct<RegionCounter>(unique_instance)(1024);321 LOG(trace) << "initialized region counter with: " << rc->fCount;322 }323 324 cfg.id = rc->fCount;325 }326 327 const uint16_t id = cfg.id.value();328 329 UnmanagedRegion* region = nullptr;330 331 auto it = fRegions.find(id);332 if (it != fRegions.end()) {333 region = it->second.get();334 if (region->fControlling) {335 LOG(error) << "Unmanaged Region with id " << id << " already exists. Only unique IDs per session are allowed.";336 throw TransportError(tools::ToString("Unmanaged Region with id ", id, " already exists. Only unique IDs per session are allowed."));337 }338 339 LOG(debug) << "Unmanaged region (view) already present, promoting to controller";340 region->BecomeController(cfg);341 } else {342 // we need to update local config, if the region information already exists343 auto info = fShmRegions->find(id);344 if (info != fShmRegions->end()) {345 cfg.rcSegmentSize = info->second.fRCSegmentSize;346 }347 348 auto res = fRegions.emplace(id, std::make_unique<UnmanagedRegion>(fShmId, size, true, cfg));349 region = res.first->second.get();350 }351 // LOG(debug) << "Created region with id '" << id << "', path: '" << cfg.path << "', flags: '" << cfg.creationFlags << "'";352 353 // start ack receiver only if a callback has been provided.354 if (callback || bulkCallback) {355 region->SetCallbacks(callback, bulkCallback);356 region->InitializeQueues();357 region->StartAckSender();358 region->StartAckReceiver();359 }360 result.first = region;361 result.second = id;362 }363 fRegionsGen += 1; // signal TL cache invalidation364 365 return result;366 } catch (interprocess_exception& e) {367 LOG(error) << "cannot create region. Already created/not cleaned up?";368 LOG(error) << e.what();369 throw;370 }371 }372 373 UnmanagedRegion* GetRegionFromCache(uint16_t id)374 {375 // NOTE: gcc optimizations. Prevent loading tls addresses many times in the fast path376 const auto &lTlCache = fTlRegionCache;377 const auto &lTlCacheVec = lTlCache.fRegionsTLCache;378 const auto lTlCacheGen = lTlCache.fRegionsTLCacheGen;379 380 // fast path381 for (const auto &lRegion : lTlCacheVec) {382 if ((std::get<1>(lRegion) == id) && (lTlCacheGen == fRegionsGen) && (std::get<2>(lRegion) == fShmId64)) {383 return std::get<0>(lRegion);384 }385 }386 387 // slow path: check invalidation388 if (lTlCacheGen != fRegionsGen) {389 fTlRegionCache.fRegionsTLCache.clear();390 }391 392 auto* lRegion = GetRegion(id);393 if (lRegion) {394 fTlRegionCache.fRegionsTLCache.emplace_back(lRegion, id, fShmId64);395 fTlRegionCache.fRegionsTLCacheGen = fRegionsGen;396 }397 return lRegion;398 }399 400 UnmanagedRegion* GetRegion(uint16_t id)401 {402 std::lock_guard<std::mutex> lock(fLocalRegionsMtx);403 auto it = fRegions.find(id);404 if (it != fRegions.end()) {405 return it->second.get();406 }407 408 try {409 RegionConfig cfg;410 // get region info (lock order: fLocalRegionsMtx before fShmMtx)411 {412 boost::interprocess::scoped_lock<boost::interprocess::interprocess_mutex> shmLock(*fShmMtx);413 RegionInfo regionInfo = fShmRegions->at(id);414 cfg.id = id;415 cfg.creationFlags = regionInfo.fCreationFlags;416 cfg.rcSegmentSize = regionInfo.fRCSegmentSize;417 cfg.path = regionInfo.fPath.c_str();418 }419 // LOG(debug) << "Located remote region with id '" << id << "', path: '" << cfg.path << "', flags: '" << cfg.creationFlags << "'";420 421 auto r = fRegions.emplace(id, std::make_unique<UnmanagedRegion>(fShmId, 0, false, std::move(cfg)));422 r.first->second->InitializeQueues();423 r.first->second->StartAckSender();424 return r.first->second.get();425 } catch (std::out_of_range& oor) {426 LOG(error) << "Could not get remote region with id '" << id << "'. Does the region creator run with the same session id?";427 LOG(error) << oor.what();428 return nullptr;429 } catch (boost::interprocess::interprocess_exception& e) {430 LOG(error) << "Could not get remote region for id '" << id << "': " << e.what();431 return nullptr;432 }433 }434 435 void RemoveRegion(uint16_t id)436 {437 UnmanagedRegion* region = nullptr;438 {439 std::lock_guard<std::mutex> lock(fLocalRegionsMtx);440 auto it = fRegions.find(id);441 if (it == fRegions.end()) {442 LOG(debug) << "RemoveRegion() could not locate region with id '" << id << "'";443 return;444 }445 region = it->second.get();446 }447 448 // Do not hold a lock here: the ack receiver calls the user callbacks for the outstanding449 // acks, and a callback can call into the transport.450 region->StopAcks();451 452 if (region->RemoveOnDestruction()) {453 boost::interprocess::scoped_lock<boost::interprocess::interprocess_mutex> shmLock(*fShmMtx);454 try {455 fShmRegions->at(id).fDestroyed = true;456 (fEventCounter->fCount)++;457 } catch (std::out_of_range& oor) {458 LOG(debug) << "RemoveRegion() could not locate region info with id '" << id << "'";459 }460 }461 462 std::unique_ptr<UnmanagedRegion> removedRegion;463 {464 std::lock_guard<std::mutex> lock(fLocalRegionsMtx);465 auto it = fRegions.find(id);466 if (it != fRegions.end()) {467 removedRegion = std::move(it->second);468 fRegions.erase(it);469 }470 }471 fRegionsGen += 1; // signal TL cache invalidation472 // removedRegion is destroyed here, without a lock473 }474 475 std::vector<fair::mq::RegionInfo> GetRegionInfo()476 {477 std::vector<fair::mq::RegionInfo> result;478 std::map<uint64_t, RegionConfig> regionCfgs;479 480 {481 boost::interprocess::scoped_lock<boost::interprocess::interprocess_mutex> shmLock(*fShmMtx);482 483 for (const auto& [segmentId, segmentInfo] : *fShmSegments) {484 // make sure any segments in the session are found485 GetSegment(segmentId);486 try {487 fair::mq::RegionInfo info;488 info.managed = true;489 info.id = segmentId;490 info.event = RegionEvent::created;491 info.ptr = std::visit([](auto& s) { return s.get_address(); }, fSegments.at(segmentId));492 info.size = std::visit([](auto& s) { return s.get_size(); }, fSegments.at(segmentId));493 result.push_back(info);494 } catch (const std::out_of_range& oor) {495 LOG(error) << "could not find segment with id " << segmentId;496 LOG(error) << oor.what();497 }498 }499 500 for (const auto& [regionId, regionInfo] : *fShmRegions) {501 fair::mq::RegionInfo info;502 info.managed = false;503 info.id = regionId;504 info.flags = regionInfo.fUserFlags;505 info.event = regionInfo.fDestroyed ? RegionEvent::destroyed : RegionEvent::created;506 if (info.event == RegionEvent::created) {507 RegionConfig cfg;508 cfg.id = info.id;509 cfg.creationFlags = regionInfo.fCreationFlags;510 cfg.path = regionInfo.fPath.c_str();511 cfg.rcSegmentSize = regionInfo.fRCSegmentSize;512 regionCfgs.emplace(info.id, cfg);513 // fill the ptr+size info after shmLock is released, to avoid constructing local region under it514 } else {515 info.ptr = nullptr;516 info.size = 0;517 }518 result.push_back(info);519 }520 }521 522 // do another iteration outside of shm lock, to fill ptr+size of unmanaged regions523 for (auto& info : result) {524 if (!info.managed && info.event == RegionEvent::created) {525 auto cfgIt = regionCfgs.find(info.id);526 if (cfgIt != regionCfgs.end()) {527 UnmanagedRegion* region = nullptr;528 std::lock_guard<std::mutex> lock(fLocalRegionsMtx);529 auto it = fRegions.find(info.id);530 if (it != fRegions.end()) {531 region = it->second.get();532 } else {533 auto r = fRegions.emplace(cfgIt->first, std::make_unique<UnmanagedRegion>(fShmId, 0, false, cfgIt->second));534 region = r.first->second.get();535 region->InitializeQueues();536 region->StartAckSender();537 }538 539 info.ptr = region->GetData();540 info.size = region->GetSize();541 } else {542 info.ptr = nullptr;543 info.size = 0;544 }545 }546 }547 548 return result;549 }550 551 void SubscribeToRegionEvents(RegionEventCallback callback)552 {553 if (fRegionEventThread.joinable()) {554 LOG(debug) << "Already subscribed. Overwriting previous subscription.";555 std::unique_lock<std::mutex> lock(fRegionEventsMtx);556 fRegionEventsSubscriptionActive = false;557 lock.unlock();558 fRegionEventsCV.notify_one();559 fRegionEventThread.join();560 }561 std::lock_guard<std::mutex> lock(fRegionEventsMtx);562 fRegionEventCallback = callback;563 fRegionEventsSubscriptionActive = true;564 fRegionEventThread = std::thread(&Manager::RegionEventsSubscription, this);565 }566 567 bool SubscribedToRegionEvents() { return fRegionEventThread.joinable(); }568 569 void UnsubscribeFromRegionEvents()570 {571 if (fRegionEventThread.joinable()) {572 std::unique_lock<std::mutex> lock(fRegionEventsMtx);573 fRegionEventsSubscriptionActive = false;574 lock.unlock();575 fRegionEventsCV.notify_one();576 fRegionEventThread.join();577 lock.lock();578 fRegionEventCallback = nullptr;579 }580 }581 582 void RegionEventsSubscription()583 {584 std::unique_lock<std::mutex> lock(fRegionEventsMtx);585 586 while (fRegionEventsSubscriptionActive) {587 if (fNumObservedEvents != fEventCounter->fCount) {588 auto infos = GetRegionInfo();589 590 for (const auto& i : infos) {591 auto el = fObservedRegionEvents.find({i.id, i.managed});592 if (el == fObservedRegionEvents.end()) { // if event id has not been observed593 fObservedRegionEvents.emplace(std::make_pair(i.id, i.managed), i.event);594 // if a region has been created and destroyed rapidly, we could see 'destroyed' without ever seeing 'created'595 // TODO: do we care to show 'created' events if we know region is already destroyed?596 if (i.event == RegionEvent::created) {597 fRegionEventCallback(i);598 ++fNumObservedEvents;599 } else {600 fNumObservedEvents += 2;601 }602 } else { // if event id has been observed (expected - there are two events per id - created & destroyed)603 // fire a callback if we have observed 'created' event and incoming is 'destroyed'604 if (el->second == RegionEvent::created && i.event == RegionEvent::destroyed) {605 fRegionEventCallback(i);606 el->second = i.event;607 ++fNumObservedEvents;608 } else {609 // LOG(debug) << "ignoring event " << i.id << ": incoming: " << i.event << ", stored: " << el->second;610 }611 }612 }613 }614 // TODO: do better than polling here, without adding too much shmem contention615 fRegionEventsCV.wait_for(lock, std::chrono::milliseconds(50), [&] { return !fRegionEventsSubscriptionActive; });616 }617 }618 619 void IncrementMsgCounter()620 {621#ifdef FAIRMQ_DEBUG_MODE622 fMsgCounterNew.fetch_add(1, std::memory_order_relaxed);623#endif624 }625 void DecrementMsgCounter()626 {627#ifdef FAIRMQ_DEBUG_MODE628 fMsgCounterDelete.fetch_add(1, std::memory_order_relaxed);629#endif630 }631 632#ifdef FAIRMQ_DEBUG_MODE633 void IncrementShmMsgCounter(uint16_t segmentId) { ++((*fShmMsgCounters)[segmentId].fCount); }634 void DecrementShmMsgCounter(uint16_t segmentId) { --((*fShmMsgCounters)[segmentId].fCount); }635#endif636 637 void Heartbeats()638 {639 using namespace boost::interprocess;640 641 Heartbeat* hb = fManagementSegment.find_or_construct<Heartbeat>(unique_instance)(0);642 std::unique_lock<std::mutex> lock(fHeartbeatsMtx);643 while (fBeatTheHeart) {644 (hb->fCount)++;645 fHeartbeatsCV.wait_for(lock, std::chrono::milliseconds(100), [&]() { return !fBeatTheHeart; });646 }647 }648 649 void StopHeartbeats()650 {651 {652 std::unique_lock<std::mutex> lock(fHeartbeatsMtx);653 fBeatTheHeart = false;654 }655 fHeartbeatsCV.notify_one();656 if (fHeartbeatThread.joinable()) {657 fHeartbeatThread.join();658 }659 }660 661 void GetSegment(uint16_t id)662 {663 auto it = fSegments.find(id);664 if (it == fSegments.end()) {665 try {666 // get segment info667 SegmentInfo segmentInfo = fShmSegments->at(id);668 LOG(debug) << "Located segment with id '" << id << "'";669 670 using namespace boost::interprocess;671 672 if (segmentInfo.fAllocationAlgorithm == AllocationAlgorithm::rbtree_best_fit) {673 fSegments.emplace(id, RBTreeBestFitSegment(open_only, MakeShmName(fShmId, "m", id).c_str()));674 } else {675 fSegments.emplace(id, SimpleSeqFitSegment(open_only, MakeShmName(fShmId, "m", id).c_str()));676 }677 } catch (std::out_of_range& oor) {678 LOG(error) << "Could not get segment with id '" << id << "': " << oor.what();679 } catch (boost::interprocess::interprocess_exception& bie) {680 LOG(error) << "Could not get segment with id '" << id << "': " << bie.what();681 }682 }683 }684 685 boost::interprocess::managed_shared_memory::handle_t GetHandleFromAddress(const void* ptr, uint16_t segmentId) const686 {687 return std::visit([ptr](auto& s) { return s.get_handle_from_address(ptr); }, fSegments.at(segmentId));688 }689 char* GetAddressFromHandle(const boost::interprocess::managed_shared_memory::handle_t handle, uint16_t segmentId) const690 {691 return std::visit([handle](auto& s) { return reinterpret_cast<char*>(s.get_address_from_handle(handle)); }, fSegments.at(segmentId));692 }693 694 char* Allocate(size_t size, size_t alignment = 0)695 {696 alignment = std::max(alignment, alignof(std::max_align_t));697 698 char* ptr = nullptr;699 int numAttempts = 0;700 size_t fullSize = ShmHeader::FullSize(size, alignment);701 702 while (!ptr) {703 try {704 size_t segmentSize = std::visit([](auto& s) { return s.get_size(); }, fSegments.at(fSegmentId));705 if (fullSize > segmentSize) {706 throw MessageBadAlloc(tools::ToString("Requested message size (", fullSize, ") exceeds segment size (", segmentSize, ")"));707 }708 709 ptr = std::visit([fullSize](auto& s) { return reinterpret_cast<char*>(s.allocate(fullSize)); }, fSegments.at(fSegmentId));710 ShmHeader::Construct(ptr, alignment);711 } catch (boost::interprocess::bad_alloc& ba) {712 // LOG(warn) << "Shared memory full...";713 if (fBadAllocMaxAttempts >= 0 && ++numAttempts >= fBadAllocMaxAttempts) {714 throw MessageBadAlloc(tools::ToString("shmem: could not create a message of size ", size,715 ", alignment: ", (alignment != 0) ? std::to_string(alignment) : "default",716 ", free memory: ", std::visit([](auto& s) { return s.get_free_memory(); }, fSegments.at(fSegmentId))));717 }718 if (numAttempts == 1 && fBadAllocMaxAttempts > 1) {719 LOG(warn) << tools::ToString("shmem: could not create a message of size ", size,720 ", alignment: ", (alignment != 0) ? std::to_string(alignment) : "default",721 ", free memory: ", std::visit([](auto& s) { return s.get_free_memory(); }, fSegments.at(fSegmentId)),722 ". Will try ", (fBadAllocMaxAttempts > 1 ? (std::to_string(fBadAllocMaxAttempts - 1)) + " more times" : " until success"),723 ", in ", fBadAllocAttemptIntervalInMs, "ms intervals");724 }725 std::this_thread::sleep_for(std::chrono::milliseconds(fBadAllocAttemptIntervalInMs));726 if (Interrupted()) {727 throw MessageBadAlloc(tools::ToString("shmem: could not create a message of size ", size,728 ", alignment: ", (alignment != 0) ? std::to_string(alignment) : "default",729 ", free memory: ", std::visit([](auto& s) { return s.get_free_memory(); }, fSegments.at(fSegmentId))));730 } else {731 continue;732 }733 }734#ifdef FAIRMQ_DEBUG_MODE735 boost::interprocess::scoped_lock<boost::interprocess::interprocess_mutex> lock(*fShmMtx);736 IncrementShmMsgCounter(fSegmentId);737 if (fMsgDebug->count(fSegmentId) == 0) {738 fMsgDebug->emplace(fSegmentId, fShmVoidAlloc);739 }740 fMsgDebug->at(fSegmentId).emplace(741 static_cast<size_t>(GetHandleFromAddress(ShmHeader::UserPtr(ptr), fSegmentId)),742 MsgDebug(getpid(), size, std::chrono::system_clock::now().time_since_epoch().count())743 );744#endif745 }746 747 return ptr;748 }749 750 void Deallocate(boost::interprocess::managed_shared_memory::handle_t handle, uint16_t segmentId)751 {752 char* ptr = GetAddressFromHandle(handle, segmentId);753#ifdef FAIRMQ_DEBUG_MODE754 boost::interprocess::scoped_lock<boost::interprocess::interprocess_mutex> lock(*fShmMtx);755 DecrementShmMsgCounter(segmentId);756 try {757 fMsgDebug->at(segmentId).erase(GetHandleFromAddress(ShmHeader::UserPtr(ptr), fSegmentId));758 } catch (const std::out_of_range& oor) {759 LOG(debug) << "could not locate debug container for " << segmentId << ": " << oor.what();760 }761#endif762 ShmHeader::Destruct(ptr);763 std::visit([ptr](auto& s) { s.deallocate(ptr); }, fSegments.at(segmentId));764 }765 766 char* ShrinkInPlace(size_t newSize, char* localPtr, uint16_t segmentId)767 {768 return std::visit(SegmentBufferShrink(newSize, localPtr), fSegments.at(segmentId));769 }770 771 uint16_t GetSegmentId() const { return fSegmentId; }772 773 void CleanupIfLast()774 {775 using namespace boost::interprocess;776 777 bool lastRemoved = false;778 try {779 boost::interprocess::scoped_lock<interprocess_mutex> lock(*fShmMtx);780 781 (fDeviceCounter->fCount)--;782 783 if (fDeviceCounter->fCount == 0) {784 LOG(debug) << "Last segment user, " << (fNoCleanup ? "skipping removal (--shm-no-cleanup is true)." : "removing segment.");785 lastRemoved = true;786 } else {787 LOG(debug) << "Other segment users present (" << fDeviceCounter->fCount << "), skipping removal.";788 }789 } catch (interprocess_exception& e) {790 LOG(error) << "Manager could not acquire lock: " << e.what();791 }792 793 if (lastRemoved) {794 if (!fNoCleanup) {795 Monitor::Cleanup(ShmId{fShmId});796 }797 }798 }799 800 auto GetMetadataMsgSize() const noexcept { return fMetadataMsgSize; }801 802 // Resolve a MetaHeader (received over a side channel) to the local data pointer.803 // The caller is responsible for ensuring the backing buffer remains alive for the804 // duration of access; FairMQ provides no refcount protection for this path.805 char* GetDataAddressFromHandle(const MetaHeader& meta)806 {807 if (meta.fManaged) {808 if (meta.fSize == 0) {809 return nullptr;810 }811 GetSegment(meta.fSegmentId);812 auto it = fSegments.find(meta.fSegmentId);813 if (it == fSegments.end()) {814 throw SharedMemoryError(tools::ToString("GetDataAddressFromHandle: cannot open segment with id ", meta.fSegmentId));815 }816 return ShmHeader::UserPtr(GetAddressFromHandle(meta.fHandle, meta.fSegmentId));817 } else {818 UnmanagedRegion* region = GetRegionFromCache(meta.fRegionId);819 if (!region) {820 throw SharedMemoryError(tools::ToString("GetDataAddressFromHandle: cannot get unmanaged region with id ", meta.fRegionId));821 }822 return reinterpret_cast<char*>(region->GetData()) + meta.fHandle;823 }824 }825 826 ~Manager()827 {828 fRegionsGen += 1; // signal TL cache invalidation829 UnsubscribeFromRegionEvents();830 831 StopHeartbeats();832 833 CleanupIfLast();834 }835 836 private:837 uint64_t fShmId64;838 std::string fShmId;839 uint16_t fSegmentId;840 std::unordered_map<uint16_t, std::variant<RBTreeBestFitSegment, SimpleSeqFitSegment>> fSegments; // TODO: refactor to use Segment class841 boost::interprocess::managed_shared_memory fManagementSegment; // TODO: refactor to use ManagementSegment class842 VoidAlloc fShmVoidAlloc;843 boost::interprocess::interprocess_mutex* fShmMtx;844 845 std::mutex fLocalRegionsMtx;846 std::mutex fRegionEventsMtx;847 std::condition_variable fRegionEventsCV;848 std::thread fRegionEventThread;849 std::function<void(fair::mq::RegionInfo)> fRegionEventCallback;850 std::map<std::pair<uint16_t, bool>, RegionEvent> fObservedRegionEvents; // pair: <region id, managed>851 uint64_t fNumObservedEvents;852 853 DeviceCounter* fDeviceCounter;854 EventCounter* fEventCounter;855 Uint16SegmentInfoHashMap* fShmSegments;856 Uint16RegionInfoHashMap* fShmRegions;857 std::unordered_map<uint16_t, std::unique_ptr<UnmanagedRegion>> fRegions;858 inline static std::atomic<unsigned long> fRegionsGen = 0ul;859 inline static thread_local struct ManagerTLCache {860 unsigned long fRegionsTLCacheGen;861 std::vector<std::tuple<UnmanagedRegion*, uint16_t, uint64_t>> fRegionsTLCache;862 } fTlRegionCache;863 864#ifdef FAIRMQ_DEBUG_MODE865 Uint16MsgDebugMapHashMap* fMsgDebug;866 Uint16MsgCounterHashMap* fShmMsgCounters;867 std::atomic_uint64_t fMsgCounterNew;868 std::atomic_uint64_t fMsgCounterDelete;869#endif870 871 std::thread fHeartbeatThread;872 std::mutex fHeartbeatsMtx;873 std::condition_variable fHeartbeatsCV;874 bool fBeatTheHeart;875 876 bool fRegionEventsSubscriptionActive;877 std::atomic<bool> fInterrupted;878 879 int fBadAllocMaxAttempts;880 int fBadAllocAttemptIntervalInMs;881 bool fNoCleanup;882 883 std::size_t fMetadataMsgSize;884};885 886} // namespace fair::mq::shmem887 888#endif /* FAIR_MQ_SHMEM_MANAGER_H_ */