Source · fairmq/shmem/
Monitor.cxxsource
Described in Monitor and clean up, fairmq-shmmonitor.
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 9#include "Common.h"10#include "Monitor.h"11#include "Segment.h"12#include <fairmq/shmem/UnmanagedRegion.h>13 14#include <fairmq/tools/IO.h>15#include <fairmq/tools/Strings.h>16 17#include <boost/interprocess/managed_shared_memory.hpp>18#include <boost/interprocess/file_mapping.hpp>19 20#include <boost/interprocess/sync/named_mutex.hpp>21#include <boost/interprocess/sync/interprocess_mutex.hpp>22#include <boost/interprocess/sync/named_condition.hpp>23#include <boost/interprocess/ipc/message_queue.hpp>24 25#include <csignal>26#include <cstdio>27#include <iostream>28#include <iomanip>29#include <chrono>30#include <ctime>31#include <iomanip>32#include <sstream>33#include <variant>34 35#include <poll.h>36 37#if FAIRMQ_HAS_STD_FILESYSTEM38#include <filesystem>39namespace fs = std::filesystem;40#else41#include <boost/filesystem.hpp>42namespace fs = ::boost::filesystem;43#endif44 45using namespace std;46using bie = ::boost::interprocess::interprocess_exception;47namespace bipc = ::boost::interprocess;48 49namespace50{51 volatile sig_atomic_t gSignalStatus = 0; // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)52}53 54namespace fair::mq::shmem55{56 57void signalHandler(int signal)58{59 gSignalStatus = signal;60}61 62Monitor::Monitor(string shmId,63 bool selfDestruct,64 bool interactive,65 bool viewOnly,66 unsigned int timeoutInMS,67 unsigned int intervalInMS,68 bool monitor,69 bool cleanOnExit)70 : fSelfDestruct(selfDestruct)71 , fInteractive(interactive)72 , fViewOnly(viewOnly)73 , fMonitor(monitor)74 , fCleanOnExit(cleanOnExit)75 , fTimeoutInMS(timeoutInMS)76 , fIntervalInMS(intervalInMS)77 , fShmId(std::move(shmId))78 , fTerminating(false)79 , fHeartbeatTriggered(false)80 , fLastHeartbeat(chrono::high_resolution_clock::now())81{82 if (!fViewOnly) {83 try {84 bipc::named_mutex monitorStatus(bipc::create_only, MakeShmName(fShmId, "ms").c_str());85 } catch (bie&) {86 if (fInteractive) {87 LOG(error) << "fairmq-shmmonitor for shm id " << fShmId << " is already running. Try `fairmq-shmmonitor --cleanup --shmid " << fShmId << "`, or run in view-only mode (-v)";88 } else {89 LOG(error) << "fairmq-shmmonitor for shm id " << fShmId << " is already running. Try `fairmq-shmmonitor --cleanup --shmid " << fShmId << "`";90 }91 throw DaemonPresent(tools::ToString("fairmq-shmmonitor (in monitoring mode) for shm id ", fShmId, " is already running."));92 }93 }94}95 96void Monitor::CatchSignals()97{98 signal(SIGINT, signalHandler);99 signal(SIGTERM, signalHandler);100 fSignalThread = thread(&Monitor::SignalMonitor, this);101}102 103void Monitor::SignalMonitor()104{105 while (true) {106 if (gSignalStatus != 0) {107 fTerminating = true;108 LOG(info) << "signal: " << gSignalStatus;109 break;110 } else if (fTerminating) {111 break;112 }113 114 this_thread::sleep_for(chrono::milliseconds(100));115 }116}117 118void Monitor::Run()119{120 thread heartbeatThread;121 if (!fViewOnly) {122 heartbeatThread = thread(&Monitor::CheckHeartbeats, this);123 }124 125 if (fInteractive) {126 Interactive();127 } else {128 Watch();129 }130 131 if (!fViewOnly) {132 heartbeatThread.join();133 }134}135 136void Monitor::Watch()137{138 while (!fTerminating) {139 using namespace boost::interprocess;140 141 try {142 managed_shared_memory managementSegment(open_read_only, MakeShmName(fShmId, "mng").c_str());143 144 fSeenOnce = true;145 146 auto now = chrono::high_resolution_clock::now();147 unsigned int duration = chrono::duration_cast<chrono::milliseconds>(now - fLastHeartbeat.load()).count();148 149 if (fHeartbeatTriggered && duration > fTimeoutInMS) {150 // memory is present, but no heartbeats since timeout duration151 LOG(info) << "no heartbeats since over " << fTimeoutInMS << " milliseconds, cleaning...";152 Cleanup(ShmId{fShmId});153 fHeartbeatTriggered = false;154 if (fSelfDestruct) {155 LOG(info) << "self destructing (segment has been observed and cleaned up by the monitor)";156 fTerminating = true;157 }158 }159 } catch (bie&) {160 fHeartbeatTriggered = false;161 162 if (fSelfDestruct) {163 if (fSeenOnce) {164 // segment has been observed at least once, can self-destruct165 LOG(info) << "self destructing (segment has been observed and cleaned up orderly)";166 fTerminating = true;167 } else {168 // if self-destruct is requested, and no segment has ever been observed, quit after double timeout duration169 auto now = chrono::high_resolution_clock::now();170 unsigned int duration = chrono::duration_cast<chrono::milliseconds>(now - fLastHeartbeat.load()).count();171 172 if (duration > fTimeoutInMS * 2) {173 Cleanup(ShmId{fShmId});174 LOG(info) << "self destructing (no segments observed within (timeout * 2) since start)";175 fTerminating = true;176 }177 }178 }179 }180 181 this_thread::sleep_for(chrono::milliseconds(100));182 }183}184 185bool Monitor::PrintShm(const ShmId& shmId)186{187 using namespace boost::interprocess;188 189 try {190 managed_shared_memory managementSegment(open_read_only, MakeShmName(shmId.shmId, "mng").c_str());191 VoidAlloc allocInstance(managementSegment.get_segment_manager());192 193 Uint16SegmentInfoHashMap* shmSegments = managementSegment.find<Uint16SegmentInfoHashMap>(unique_instance).first;194 std::unordered_map<uint16_t, std::variant<RBTreeBestFitSegment, SimpleSeqFitSegment>> segments;195 196 Uint16RegionInfoHashMap* shmRegions = managementSegment.find<Uint16RegionInfoHashMap>(unique_instance).first;197 198 if (!shmSegments) {199 LOG(error) << "Found management segment, but cannot locate segment info, something went wrong...";200 return false;201 }202 203 if (!shmRegions) {204 LOG(error) << "Found management segment, but cannot locate region info...";205 }206 207 for (const auto& s : *shmSegments) {208 if (s.second.fAllocationAlgorithm == AllocationAlgorithm::rbtree_best_fit) {209 segments.emplace(s.first, RBTreeBestFitSegment(open_read_only, MakeShmName(shmId.shmId, "m", s.first).c_str()));210 } else {211 segments.emplace(s.first, SimpleSeqFitSegment(open_read_only, MakeShmName(shmId.shmId, "m", s.first).c_str()));212 }213 }214 215 unsigned int numDevices = 0;216 int creatorId = -1;217 std::string sessionName;218 219 DeviceCounter* deviceCounter = managementSegment.find<DeviceCounter>(unique_instance).first;220 if (deviceCounter) {221 numDevices = deviceCounter->fCount;222 }223 SessionInfo* sessionInfo = managementSegment.find<SessionInfo>(unique_instance).first;224 if (sessionInfo) {225 creatorId = sessionInfo->fCreatorId;226 sessionName = sessionInfo->fSessionName;227 }228#ifdef FAIRMQ_DEBUG_MODE229 Uint16MsgCounterHashMap* msgCounters = managementSegment.find<Uint16MsgCounterHashMap>(unique_instance).first;230#endif231 232 stringstream ss;233 size_t mfree = managementSegment.get_free_memory();234 size_t mtotal = managementSegment.get_size();235 size_t mused = mtotal - mfree;236 237 ss << "shm id: " << shmId.shmId238 << ", session: " << sessionName239 << ", creator id: " << creatorId240 << ", devices: " << numDevices241 << ", managed segments:\n";242 243 for (const auto& s : segments) {244 size_t free = std::visit([](auto& seg){ return seg.get_free_memory(); }, s.second);245 size_t total = std::visit([](auto& seg){ return seg.get_size(); }, s.second);246 size_t used = total - free;247 248 std::string msgCount;249#ifdef FAIRMQ_DEBUG_MODE250 if (msgCounters) {251 auto it = msgCounters->find(s.first);252 if (it != msgCounters->end()) {253 msgCount = to_string(it->second.fCount.load());254 } else {255 msgCount = "0";256 }257 }258#else259 msgCount = "NODEBUG";260#endif261 262 ss << " [" << s.first << "]"263 << ": total: " << total264 << ", msgs: " << msgCount265 << ", free: " << free266 << ", used: " << used267 << "\n";268 }269 270 ss << " [m]: "271 << "total: " << mtotal272 << ", free: " << mfree273 << ", used: " << mused;274 275 if (shmRegions && !shmRegions->empty()) {276 ss << "\n unmanaged regions:";277 for (const auto& [id, info] : *shmRegions) {278 ss << "\n [" << id << "]: " << (info.fDestroyed ? "destroyed" : "alive");279 ss << ", size: " << info.fSize;280 281 try {282 managed_shared_memory rcCountSegment(open_read_only, MakeShmName(shmId.shmId, "rrc", id).c_str());283 auto size = rcCountSegment.get_size();284 auto free = rcCountSegment.get_free_memory();285 ss << ", rcCountSegment size: " << size << ", free: " << free << ", used: " << size - free;286 } catch (bie&) {287 ss << ", rcCountSegment: not found";288 }289 290 // try {291 // boost::interprocess::message_queue q(open_only, std::string("fmq_" + std::string(shmId) + "_rgq_" + to_string(id)).c_str());292 // ss << ", ack queue: " << q.get_num_msg() << " messages";293 // } catch (bie&) {294 // ss << ", ack queue: not found";295 // }296 }297 }298 LOGV(info, user1) << ss.str();299 } catch (bie&) {300 return false;301 }302 303 return true;304}305 306void Monitor::ListAll(const std::string& path)307{308 try {309 if (fs::is_empty(path)) {310 LOG(info) << "directory " << fs::path(path) << " is empty.";311 return;312 }313 314 for (const auto& entry : fs::directory_iterator(path)) {315 string filename = entry.path().filename().string();316 // LOG(info) << filename << ", size: " << entry.file_size() << " bytes";317 if (tools::StrStartsWith(filename, "fmq_") || tools::StrStartsWith(filename, "sem.fmq_")) {318 // LOG(info) << "The file '" << filename << "' belongs to FairMQ.";319 if (tools::StrEndsWith(filename, "_mng")) {320 string shmId = filename.substr(4, 8);321 LOG(info) << "\nFound shmid '" << shmId << "':\n";322 if (!PrintShm(ShmId{shmId})) {323 LOG(info) << "could not open file for shmid '" << shmId << "'";324 }325 }326 } else {327 LOG(info) << "The file '" << filename << "' does not belong to FairMQ, skipping...";328 }329 }330 } catch (fs::filesystem_error& fse) {331 LOG(error) << "error: " << fse.what();332 }333}334 335void Monitor::CheckHeartbeats()336{337 using namespace boost::interprocess;338 339 uint64_t localHb = 0;340 341 while (!fTerminating) {342 std::this_thread::sleep_for(std::chrono::milliseconds(200));343 try {344 managed_shared_memory managementSegment(open_read_only, MakeShmName(fShmId, "mng").c_str());345 Heartbeat* hb = managementSegment.find<Heartbeat>(unique_instance).first;346 347 if (hb) {348 uint64_t globalHb = hb->fCount;349 if (localHb != globalHb) {350 fHeartbeatTriggered = true;351 fLastHeartbeat.store(chrono::high_resolution_clock::now());352 localHb = globalHb;353 }354 }355 } catch (bie&) {356 // management segment not found, simply retry.357 }358 }359}360 361void Monitor::Interactive()362{363 char c = 0;364 pollfd cinfd[1];365 cinfd[0].fd = fileno(stdin);366 cinfd[0].events = POLLIN;367 368 tools::NonCanonicalInput nci;369 370 LOG(info) << "\n";371 PrintHelp();372 373 while (!fTerminating) {374 if (poll(cinfd, 1, fIntervalInMS)) {375 if (fTerminating || gSignalStatus != 0) {376 break;377 }378 379 c = getchar();380 381 switch (c) {382 case 'q':383 LOG(info) << "\n[q] --> quitting.";384 fTerminating = true;385 break;386 case 'x':387 LOG(info) << "\n[x] --> closing shared memory:";388 if (!fViewOnly) {389 Cleanup(ShmId{fShmId});390 } else {391 LOG(info) << "cannot close because in view only mode";392 }393 break;394 case 'h':395 LOG(info) << "\n[h] --> help:\n";396 PrintHelp();397 LOG(info);398 break;399 case '\n':400 LOG(info) << "\n[\\n] --> invalid input.";401 break;402 case 'b':403 PrintDebugInfo(ShmId{fShmId});404 break;405 default:406 LOG(info) << "\n[" << c << "] --> invalid input.";407 break;408 }409 410 if (fTerminating) {411 break;412 }413 }414 415 if (fTerminating) {416 break;417 }418 419 PrintShm(ShmId{fShmId});420 }421}422 423void Monitor::PrintDebugInfo(const ShmId& shmId __attribute__((unused)))424{425#ifdef FAIRMQ_DEBUG_MODE426 string managementSegmentName = MakeShmName(shmId.shmId, "mng");427 try {428 bipc::managed_shared_memory managementSegment(bipc::open_only, managementSegmentName.c_str());429 bipc::interprocess_mutex* mtx(managementSegment.find_or_construct<bipc::interprocess_mutex>(bipc::unique_instance)());430 bipc::scoped_lock<bipc::interprocess_mutex> lock(*mtx);431 432 Uint16MsgDebugMapHashMap* debug = managementSegment.find<Uint16MsgDebugMapHashMap>(bipc::unique_instance).first;433 434 size_t numMessages = 0;435 436 if (debug) {437 for (const auto& e : *debug) {438 numMessages += e.second.size();439 }440 LOG(info) << endl << "found " << numMessages << " messages.";441 442 for (const auto& s : *debug) {443 for (const auto& e : s.second) {444 using time_point = chrono::system_clock::time_point;445 time_point tmpt{chrono::duration_cast<time_point::duration>(chrono::nanoseconds(e.second.fCreationTime))};446 time_t t = chrono::system_clock::to_time_t(tmpt);447 uint64_t ms = e.second.fCreationTime % 1000000;448 auto tm = localtime(&t);449 LOG(info) << "segment: " << setw(3) << setfill(' ') << s.first450 << ", offset: " << setw(12) << setfill(' ') << e.first451 << ", size: " << setw(10) << setfill(' ') << e.second.fSize452 << ", creator PID: " << e.second.fPid << setfill('0')453 << ", at: " << setw(2) << tm->tm_hour << ":" << setw(2) << tm->tm_min << ":" << setw(2) << tm->tm_sec << "." << setw(6) << ms;454 }455 }456 } else {457 LOG(info) << "no debug data found";458 }459 } catch (bie&) {460 LOG(info) << "no segments found";461 }462#else463 LOG(info) << "FairMQ was not compiled in debug mode (FAIRMQ_DEBUG_MODE)";464#endif465}466 467void Monitor::PrintDebugInfo(const SessionId& sessionId)468{469 ShmId shmId{makeShmIdStr(sessionId.sessionId)};470 PrintDebugInfo(shmId);471}472 473unordered_map<uint16_t, std::vector<BufferDebugInfo>> Monitor::GetDebugInfo(const ShmId& shmId __attribute__((unused)))474{475 unordered_map<uint16_t, std::vector<BufferDebugInfo>> result;476 477#ifdef FAIRMQ_DEBUG_MODE478 string managementSegmentName = MakeShmName(shmId.shmId, "mng");479 try {480 bipc::managed_shared_memory managementSegment(bipc::open_only, managementSegmentName.c_str());481 bipc::interprocess_mutex* mtx(managementSegment.find_or_construct<bipc::interprocess_mutex>(bipc::unique_instance)());482 bipc::scoped_lock<bipc::interprocess_mutex> lock(*mtx);483 484 Uint16MsgDebugMapHashMap* debug = managementSegment.find<Uint16MsgDebugMapHashMap>(bipc::unique_instance).first;485 486 result.reserve(debug->size());487 488 489 if (debug) {490 for (const auto& s : *debug) {491 result[s.first].reserve(s.second.size());492 for (const auto& e : s.second) {493 result[s.first][e.first] = BufferDebugInfo(e.first, e.second.fPid, e.second.fSize, e.second.fCreationTime);494 }495 }496 } else {497 LOG(info) << "no debug data found";498 }499 } catch (bie&) {500 LOG(info) << "no segments found";501 }502#else503 LOG(info) << "FairMQ was not compiled in debug mode (FAIRMQ_DEBUG_MODE)";504#endif505 506 return result;507}508unordered_map<uint16_t, std::vector<BufferDebugInfo>> Monitor::GetDebugInfo(const SessionId& sessionId)509{510 ShmId shmId{makeShmIdStr(sessionId.sessionId)};511 return GetDebugInfo(shmId);512}513 514unsigned long Monitor::GetFreeMemory(const ShmId& shmId, uint16_t segmentId)515{516 using namespace boost::interprocess;517 try {518 bipc::managed_shared_memory managementSegment(bipc::open_only, MakeShmName(shmId.shmId, "mng").c_str());519 boost::interprocess::interprocess_mutex* mtx(managementSegment.find_or_construct<bipc::interprocess_mutex>(bipc::unique_instance)());520 boost::interprocess::scoped_lock<bipc::interprocess_mutex> lock(*mtx);521 522 Uint16SegmentInfoHashMap* shmSegments = managementSegment.find<Uint16SegmentInfoHashMap>(unique_instance).first;523 524 if (!shmSegments) {525 LOG(error) << "Found management segment, but could not locate segment info";526 throw MonitorError("Found management segment, but could not locate segment info");527 }528 529 auto it = shmSegments->find(segmentId);530 if (it != shmSegments->end()) {531 if (it->second.fAllocationAlgorithm == AllocationAlgorithm::rbtree_best_fit) {532 RBTreeBestFitSegment segment(open_read_only, MakeShmName(shmId.shmId, "m", segmentId).c_str());533 return segment.get_free_memory();534 } else {535 SimpleSeqFitSegment segment(open_read_only, MakeShmName(shmId.shmId, "m", segmentId).c_str());536 return segment.get_free_memory();537 }538 } else {539 LOG(error) << "Could not find segment id '" << segmentId << "'";540 throw MonitorError(tools::ToString("Could not find segment id '", segmentId, "'"));541 }542 } catch (bie&) {543 LOG(error) << "Could not find management segment for shmid '" << shmId.shmId << "'";544 throw MonitorError(tools::ToString("Could not find management segment for shmid '", shmId.shmId, "'"));545 }546}547unsigned long Monitor::GetFreeMemory(const SessionId& sessionId, uint16_t segmentId)548{549 ShmId shmId{makeShmIdStr(sessionId.sessionId)};550 return GetFreeMemory(shmId, segmentId);551}552 553bool Monitor::SegmentIsPresent(const ShmId& shmId, uint16_t segmentId)554{555 using namespace boost::interprocess;556 try {557 bipc::managed_shared_memory managementSegment(bipc::open_read_only, MakeShmName(shmId.shmId, "mng").c_str());558 Uint16SegmentInfoHashMap* shmSegments = managementSegment.find<Uint16SegmentInfoHashMap>(unique_instance).first;559 560 if (!shmSegments) {561 LOG(error) << "Found management segment, but could not locate segment info";562 return false;563 }564 565 auto it = shmSegments->find(segmentId);566 if (it != shmSegments->end()) {567 try {568 if (it->second.fAllocationAlgorithm == AllocationAlgorithm::rbtree_best_fit) {569 RBTreeBestFitSegment segment(open_read_only, MakeShmName(shmId.shmId, "m", segmentId).c_str());570 } else {571 SimpleSeqFitSegment segment(open_read_only, MakeShmName(shmId.shmId, "m", segmentId).c_str());572 }573 } catch (bie&) {574 LOG(error) << "Could not find segment with id '" << segmentId << "' for shmId '" << shmId.shmId << "'";575 return false;576 }577 } else {578 LOG(error) << "Could not find segment info for segment id '" << segmentId << "' for shmId '" << shmId.shmId << "'";579 return false;580 }581 } catch (bie&) {582 LOG(error) << "Could not find management segment for shmid '" << shmId.shmId << "'";583 return false;584 }585 586 return true;587}588bool Monitor::SegmentIsPresent(const SessionId& sessionId, uint16_t segmentId)589{590 ShmId shmId{makeShmIdStr(sessionId.sessionId)};591 return SegmentIsPresent(shmId, segmentId);592}593 594bool Monitor::RegionIsPresent(const ShmId& shmId, uint16_t regionId)595{596 using namespace boost::interprocess;597 try {598 bipc::managed_shared_memory managementSegment(bipc::open_read_only, MakeShmName(shmId.shmId, "mng").c_str());599 Uint16RegionInfoHashMap* shmRegions = managementSegment.find<Uint16RegionInfoHashMap>(bipc::unique_instance).first;600 601 if (!shmRegions) {602 LOG(error) << "Found management segment, but could not locate region info";603 return false;604 }605 606 std::string regionFileName(MakeShmName(shmId.shmId, "rg", regionId));607 608 auto it = shmRegions->find(regionId);609 if (it != shmRegions->end()) {610 try {611 if (it->second.fPath.empty()) {612 shared_memory_object object(open_only, regionFileName.c_str(), read_only);613 }614 } catch (bie&) {615 LOG(error) << "Could not find region with id '" << regionId << "' for shmId '" << shmId.shmId << "'";616 return false;617 }618 } else {619 LOG(error) << "Could not find region info for region id '" << regionId << "' for shmId '" << shmId.shmId << "'";620 return false;621 }622 } catch (bie&) {623 LOG(error) << "Could not find management segment for shmid '" << shmId.shmId << "'";624 return false;625 }626 627 return true;628}629bool Monitor::RegionIsPresent(const SessionId& sessionId, uint16_t regionId)630{631 ShmId shmId{makeShmIdStr(sessionId.sessionId)};632 return RegionIsPresent(shmId, regionId);633}634 635void Monitor::PrintHelp()636{637 LOG(info) << "controls: [x] close memory, "638 << "[b] print a list of allocated messages (only available when compiled with FAIMQ_DEBUG_MODE=ON), "639 << "[h] help, "640 << "[q] quit.";641}642 643bool Monitor::RemoveObject(const string& name) { return bipc::shared_memory_object::remove(name.c_str()); }644bool Monitor::RemoveFileMapping(const string& name) { return bipc::file_mapping::remove(name.c_str()); }645bool Monitor::RemoveQueue(const string& name) { return bipc::message_queue::remove(name.c_str()); }646bool Monitor::RemoveMutex(const string& name) { return bipc::named_mutex::remove(name.c_str()); }647bool Monitor::RemoveCondition(const string& name) { return bipc::named_condition::remove(name.c_str()); }648 649template<typename T>650std::pair<std::string, bool> Remove(const std::string& name, bool verbose)651{652 if (T::remove(name.c_str())) {653 if (verbose) {654 LOG(info) << "Successfully removed '" << name << "'.";655 }656 return {name, true};657 } else {658 if (verbose) {659 LOG(debug) << "Did not remove '" << name << "'. Already removed?";660 }661 return {name, false};662 }663}664 665std::vector<std::pair<std::string, bool>> Monitor::Cleanup(const ShmId& shmIdT, bool verbose /* = true */)666{667 std::string shmId = shmIdT.shmId;668 std::vector<std::pair<std::string, bool>> result;669 670 if (verbose) {671 LOG(info) << "Cleaning up for shared memory id '" << shmId << "'...";672 }673 674 string managementSegmentName = MakeShmName(shmId, "mng");675 try {676 bipc::managed_shared_memory managementSegment(bipc::open_read_only, managementSegmentName.c_str());677 678 Uint16RegionInfoHashMap* shmRegions = managementSegment.find<Uint16RegionInfoHashMap>(bipc::unique_instance).first;679 if (shmRegions) {680 if (verbose) {681 LOG(info) << "Found " << shmRegions->size() << " unmanaged regions...";682 }683 for (const auto& region : *shmRegions) {684 uint16_t id = region.first;685 RegionInfo info = region.second;686 string path = info.fPath.c_str();687 int flags = info.fCreationFlags;688 if (verbose) {689 LOG(info) << "Found UnmanagedRegion with id: " << id << ", path: '" << path << "', flags: " << flags << ", fDestroyed: " << info.fDestroyed << ".";690 }691 if (!path.empty()) {692 result.emplace_back(Remove<bipc::file_mapping>(path + MakeShmName(shmId, "rg", id), verbose));693 } else {694 result.emplace_back(Remove<bipc::shared_memory_object>(MakeShmName(shmId, "rg", id), verbose));695 }696 result.emplace_back(Remove<bipc::message_queue>(MakeShmName(shmId, "rgq", id), verbose));697 result.emplace_back(Remove<bipc::shared_memory_object>(MakeShmName(shmId, "rrc", id), verbose));698 }699 }700 701 Uint16SegmentInfoHashMap* shmSegments = managementSegment.find<Uint16SegmentInfoHashMap>(bipc::unique_instance).first;702 if (shmSegments) {703 if (verbose) {704 LOG(info) << "Found " << shmSegments->size() << " managed segments...";705 }706 for (const auto& segment : *shmSegments) {707 result.emplace_back(Remove<bipc::shared_memory_object>(MakeShmName(shmId, "m", segment.first), verbose));708 }709 } else {710 if (verbose) {711 LOG(info) << "Cannot find information about managed segments";712 }713 }714 715 result.emplace_back(Remove<bipc::shared_memory_object>(managementSegmentName, verbose));716 } catch (bie&) {717 if (verbose) {718 LOG(info) << "Did not find '" << managementSegmentName << "' management segment. No regions to cleanup.";719 }720 }721 722 return result;723}724 725std::vector<std::pair<std::string, bool>> Monitor::Cleanup(const SessionId& sessionId, bool verbose /* = true */)726{727 ShmId shmId{makeShmIdStr(sessionId.sessionId)};728 if (verbose) {729 LOG(info) << "Cleanup called with session id '" << sessionId.sessionId << "', translating to shared memory id '" << shmId.shmId << "'";730 }731 return Cleanup(shmId, verbose);732}733 734std::vector<std::pair<std::string, bool>> Monitor::CleanupFull(const ShmId& shmId, bool verbose /* = true */)735{736 auto result = Cleanup(shmId, verbose);737 result.emplace_back(Remove<bipc::named_mutex>(MakeShmName(shmId.shmId, "ms"), verbose));738 return result;739}740 741std::vector<std::pair<std::string, bool>> Monitor::CleanupFull(const SessionId& sessionId, bool verbose /* = true */)742{743 ShmId shmId{makeShmIdStr(sessionId.sessionId)};744 if (verbose) {745 LOG(info) << "Cleanup called with session id '" << sessionId.sessionId << "', translating to shared memory id '" << shmId.shmId << "'";746 }747 return CleanupFull(shmId, verbose);748}749 750void Monitor::ResetContent(const ShmId& shmIdT, bool verbose /* = true */)751{752 std::string shmId = shmIdT.shmId;753 if (verbose) {754 cout << "Resetting segments content for shared memory id '" << shmId << "'..." << endl;755 }756 757 string managementSegmentName = MakeShmName(shmId, "mng");758 try {759 using namespace boost::interprocess;760 managed_shared_memory managementSegment(open_only, managementSegmentName.c_str());761 762 Uint16SegmentInfoHashMap* segmentInfos = managementSegment.find<Uint16SegmentInfoHashMap>(unique_instance).first;763 if (segmentInfos) {764 cout << "Found info for " << segmentInfos->size() << " managed segments" << endl;765 for (const auto& [id, info] : *segmentInfos) {766 if (verbose) {767 cout << "Resetting content of segment '" << MakeShmName(shmId, "m", id) << "'..." << endl;768 }769 try {770 if (info.fAllocationAlgorithm == AllocationAlgorithm::rbtree_best_fit) {771 RBTreeBestFitSegment segment(open_only, MakeShmName(shmId, "m", id).c_str());772 void* ptr = segment.get_segment_manager();773 size_t size = segment.get_segment_manager()->get_size();774 new(ptr) segment_manager<char, rbtree_best_fit<mutex_family, offset_ptr<void>>, null_index>(size);775 } else {776 SimpleSeqFitSegment segment(open_only, MakeShmName(shmId, "m", id).c_str());777 void* ptr = segment.get_segment_manager();778 size_t size = segment.get_segment_manager()->get_size();779 new(ptr) segment_manager<char, simple_seq_fit<mutex_family, offset_ptr<void>>, null_index>(size);780 }781 if (verbose) {782 cout << "Done." << endl;783 }784 } catch (bie& e) {785 if (verbose) {786 cout << "Error resetting content of segment '" << MakeShmName(shmId, "m", id) << "': " << e.what() << endl;787 }788 }789 }790 } else {791 cout << "Found management segment, but cannot locate segment info, something went wrong..." << endl;792 }793 794 Uint16RegionInfoHashMap* shmRegions = managementSegment.find<Uint16RegionInfoHashMap>(bipc::unique_instance).first;795 if (shmRegions) {796 for (const auto& region : *shmRegions) {797 uint16_t id = region.first;798 Remove<bipc::message_queue>(MakeShmName(shmId, "rgq", id), verbose);799 Remove<bipc::shared_memory_object>(MakeShmName(shmId, "rrc", id), verbose);800 }801 }802 } catch (bie& e) {803 if (verbose) {804 cout << "Could not find '" << managementSegmentName << "' segment. Nothing to cleanup." << endl;805 cout << e.what() << endl;806 }807 }808 809 if (verbose) {810 cout << "Done resetting segment content for shared memory id '" << shmId << "'." << endl;811 }812}813 814void Monitor::ResetContent(const SessionId& sessionId, bool verbose /* = true */)815{816 ShmId shmId{makeShmIdStr(sessionId.sessionId)};817 if (verbose) {818 cout << "ResetContent called with session id '" << sessionId.sessionId << "', translating to shared memory id '" << shmId.shmId << "'" << endl;819 }820 ResetContent(shmId, verbose);821}822 823void Monitor::ResetContent(const ShmId& shmIdT, const std::vector<SegmentConfig>& segmentCfgs, const std::vector<RegionConfig>& regionCfgs, bool verbose /* = true */)824{825 using namespace boost::interprocess;826 827 std::string shmId = shmIdT.shmId;828 std::string managementSegmentName = MakeShmName(shmId, "mng");829 // delete management segment830 cout << "deleting management segment" << endl;831 Remove<bipc::shared_memory_object>(managementSegmentName, verbose);832 // recreate management segment833 cout << "recreating management segment..." << endl;834 managed_shared_memory mngSegment(create_only, managementSegmentName.c_str(), kManagementSegmentSize);835 cout << "done." << endl;836 // fill management segment with segment & region infos837 cout << "filling management segment with managed segment configs..." << endl;838 for (const auto& s : segmentCfgs) {839 if (s.allocationAlgorithm == "rbtree_best_fit") {840 Segment::Register(shmId, s.id, AllocationAlgorithm::rbtree_best_fit);841 } else if (s.allocationAlgorithm == "simple_seq_fit") {842 Segment::Register(shmId, s.id, AllocationAlgorithm::simple_seq_fit);843 } else {844 LOG(error) << "Unknown allocation algorithm provided: " << s.allocationAlgorithm;845 throw MonitorError("Unknown allocation algorithm provided: " + s.allocationAlgorithm);846 }847 }848 cout << "done." << endl;849 cout << "filling management segment with unmanaged region configs..." << endl;850 for (const auto& r : regionCfgs) {851 fair::mq::shmem::UnmanagedRegion::Register(shmId, r);852 }853 cout << "done." << endl;854 // reset managed segments855 ResetContent(shmIdT, verbose);856}857 858void Monitor::ResetContent(const SessionId& sessionId, const std::vector<SegmentConfig>& segmentCfgs, const std::vector<RegionConfig>& regionCfgs, bool verbose /* = true */)859{860 ShmId shmId{makeShmIdStr(sessionId.sessionId)};861 if (verbose) {862 cout << "ResetContent called with session id '" << sessionId.sessionId << "', translating to shared memory id '" << shmId.shmId << "'" << endl;863 }864 ResetContent(shmId, segmentCfgs, regionCfgs, verbose);865}866 867Monitor::~Monitor()868{869 if (fSignalThread.joinable()) {870 fSignalThread.join();871 }872 if (fCleanOnExit) {873 Cleanup(ShmId{fShmId});874 }875 if (!fViewOnly) {876 RemoveMutex(MakeShmName(fShmId, "ms"));877 }878}879 880} // namespace fair::mq::shmem