shm: allow monitor::ResetContent to cleanup after a crash

fairmq/UnmanagedRegion.h

@@ -133,6 +133,7 @@ struct RegionConfig
     bool removeOnDestruction = true; /// remove the region on object destruction
     int creationFlags = 0; /// flags passed to the underlying transport on region creation
     int64_t userFlags = 0; /// custom flags that have no effect on the transport, but can be retrieved from the region by the user
+    uint64_t size = 0; /// region size
     std::string path = ""; /// file path, if the region is backed by a file
     std::optional<uint16_t> id = std::nullopt; /// region id
     uint32_t linger = 100; /// delay in ms before region destruction to collect outstanding events

fairmq/shmem/Common.h

@@ -28,6 +28,8 @@
 namespace fair::mq::shmem
 {
 
+static constexpr uint64_t kManagementSegmentSize = 6553600;
+
 struct SharedMemoryError : std::runtime_error { using std::runtime_error::runtime_error; };
 
 using SimpleSeqFitSegment = boost::interprocess::basic_managed_shared_memory<char,
@@ -58,19 +60,22 @@ struct RegionInfo
         : fPath("", alloc)
         , fCreationFlags(0)
         , fUserFlags(0)
+        , fSize(0)
         , fDestroyed(false)
     {}
 
-    RegionInfo(const char* path, const int flags, const uint64_t userFlags, const VoidAlloc& alloc)
+    RegionInfo(const char* path, int flags, uint64_t userFlags, uint64_t size, const VoidAlloc& alloc)
         : fPath(path, alloc)
         , fCreationFlags(flags)
         , fUserFlags(userFlags)
+        , fSize(size)
         , fDestroyed(false)
     {}
 
     Str fPath;
     int fCreationFlags;
     uint64_t fUserFlags;
+    uint64_t fSize;
     bool fDestroyed;
 };
 

fairmq/shmem/Manager.h

@@ -132,7 +132,7 @@ class Manager
         : fShmId64(config ? config->GetProperty<uint64_t>("shmid", makeShmIdUint64(sessionName)) : makeShmIdUint64(sessionName))
         , fShmId(makeShmIdStr(fShmId64))
         , fSegmentId(config ? config->GetProperty<uint16_t>("shm-segment-id", 0) : 0)
-        , fManagementSegment(boost::interprocess::open_or_create, std::string("fmq_" + fShmId + "_mng").c_str(), 6553600)
+        , fManagementSegment(boost::interprocess::open_or_create, std::string("fmq_" + fShmId + "_mng").c_str(), kManagementSegmentSize)
         , fShmVoidAlloc(fManagementSegment.get_segment_manager())
         , fShmMtx(fManagementSegment.find_or_construct<boost::interprocess::interprocess_mutex>(boost::interprocess::unique_instance)())
         , fNumObservedEvents(0)

fairmq/shmem/Monitor.cxx

@@ -6,9 +6,10 @@
  *                  copied verbatim in the file "LICENSE"                       *
  ********************************************************************************/
 
-#include "Monitor.h"
 #include "Common.h"
-#include "UnmanagedRegion.h"
+#include "Monitor.h"
+#include "Segment.h"
+#include <fairmq/shmem/UnmanagedRegion.h>
 
 #include <fairmq/tools/IO.h>
 #include <fairmq/tools/Strings.h>
@@ -267,13 +268,14 @@ bool Monitor::PrintShm(const ShmId& shmId)
             ss << "\n   unmanaged regions:";
             for (const auto& r : *shmRegions) {
                 ss << "\n      [" << r.first << "]: " << (r.second.fDestroyed ? "destroyed" : "alive");
+                ss << ", size: " << r.second.fSize;
 
-                try {
+                // try {
-                    boost::interprocess::message_queue q(open_only, std::string("fmq_" + std::string(shmId) + "_rgq_" + to_string(r.first)).c_str());
+                //     boost::interprocess::message_queue q(open_only, std::string("fmq_" + std::string(shmId) + "_rgq_" + to_string(r.first)).c_str());
-                    ss << ", ack queue: " << q.get_num_msg() << " messages";
+                //     ss << ", ack queue: " << q.get_num_msg() << " messages";
-                } catch (bie&) {
+                // } catch (bie&) {
-                    ss << ", ack queue: not found";
+                //     ss << ", ack queue: not found";
-                }
+                // }
             }
         }
         LOGV(info, user1) << ss.str();
@@ -414,6 +416,7 @@ void Monitor::PrintDebugInfo(const ShmId& shmId __attribute__((unused)))
 
         size_t numMessages = 0;
 
+        if (debug) {
             for (const auto& e : *debug) {
                 numMessages += e.second.size();
             }
@@ -433,6 +436,9 @@ void Monitor::PrintDebugInfo(const ShmId& shmId __attribute__((unused)))
                         << ", at: " << setw(2) << tm->tm_hour << ":" << setw(2) << tm->tm_min << ":" << setw(2) << tm->tm_sec << "." << setw(6) << ms;
                 }
             }
+        } else {
+            LOG(info) << "no debug data found";
+        }
     } catch (bie&) {
         LOG(info) << "no segments found";
     }
@@ -462,12 +468,17 @@ unordered_map<uint16_t, std::vector<BufferDebugInfo>> Monitor::GetDebugInfo(cons
 
         result.reserve(debug->size());
 
+
+        if (debug) {
             for (const auto& s : *debug) {
                 result[s.first].reserve(s.second.size());
                 for (const auto& e : s.second) {
                     result[s.first][e.first] = BufferDebugInfo(e.first, e.second.fPid, e.second.fSize, e.second.fCreationTime);
                 }
             }
+        } else {
+            LOG(info) << "no debug data found";
+        }
     } catch (bie&) {
         LOG(info) << "no segments found";
     }
@@ -552,15 +563,16 @@ std::pair<std::string, bool> Remove(const std::string& name, bool verbose)
     }
 }
 
-std::vector<std::pair<std::string, bool>> Monitor::Cleanup(const ShmId& shmId, bool verbose /* = true */)
+std::vector<std::pair<std::string, bool>> Monitor::Cleanup(const ShmId& shmIdT, bool verbose /* = true */)
 {
+    std::string shmId = shmIdT.shmId;
     std::vector<std::pair<std::string, bool>> result;
 
     if (verbose) {
-        LOG(info) << "Cleaning up for shared memory id '" << shmId.shmId << "'...";
+        LOG(info) << "Cleaning up for shared memory id '" << shmId << "'...";
     }
 
-    string managementSegmentName("fmq_" + shmId.shmId + "_mng");
+    string managementSegmentName("fmq_" + shmId + "_mng");
     try {
         bipc::managed_shared_memory managementSegment(bipc::open_only, managementSegmentName.c_str());
 
@@ -578,22 +590,21 @@ std::vector<std::pair<std::string, bool>> Monitor::Cleanup(const ShmId& shmId, b
                     LOG(info) << "Found RegionInfo with path: '" << path << "', flags: " << flags << ", fDestroyed: " << info.fDestroyed << ".";
                 }
                 if (!path.empty()) {
-                    result.emplace_back(Remove<bipc::file_mapping>(path + "fmq_" + shmId.shmId + "_rg_" + to_string(id), verbose));
+                    result.emplace_back(Remove<bipc::file_mapping>(path + "fmq_" + shmId + "_rg_" + to_string(id), verbose));
                 } else {
-                    result.emplace_back(Remove<bipc::shared_memory_object>("fmq_" + shmId.shmId + "_rg_" + to_string(id), verbose));
+                    result.emplace_back(Remove<bipc::shared_memory_object>("fmq_" + shmId + "_rg_" + to_string(id), verbose));
                 }
-                result.emplace_back(Remove<bipc::message_queue>("fmq_" + shmId.shmId + "_rgq_" + to_string(id), verbose));
+                result.emplace_back(Remove<bipc::message_queue>("fmq_" + shmId + "_rgq_" + to_string(id), verbose));
             }
         }
 
         Uint16SegmentInfoHashMap* shmSegments = managementSegment.find<Uint16SegmentInfoHashMap>(bipc::unique_instance).first;
-
         if (shmSegments) {
             if (verbose) {
                 LOG(info) << "Found " << shmSegments->size() << " managed segments...";
             }
             for (const auto& segment : *shmSegments) {
-                result.emplace_back(Remove<bipc::shared_memory_object>("fmq_" + shmId.shmId + "_m_" + to_string(segment.first), verbose));
+                result.emplace_back(Remove<bipc::shared_memory_object>("fmq_" + shmId + "_m_" + to_string(segment.first), verbose));
             }
         } else {
             if (verbose) {
@@ -636,41 +647,49 @@ std::vector<std::pair<std::string, bool>> Monitor::CleanupFull(const SessionId&
     return CleanupFull(shmId, verbose);
 }
 
-void Monitor::ResetContent(const ShmId& shmId, bool verbose /* = true */)
+void Monitor::ResetContent(const ShmId& shmIdT, bool verbose /* = true */)
 {
+    std::string shmId = shmIdT.shmId;
     if (verbose) {
-        cout << "Resetting segments content for shared memory id '" << shmId.shmId << "'..." << endl;
+        cout << "Resetting segments content for shared memory id '" << shmId << "'..." << endl;
     }
 
-    string managementSegmentName("fmq_" + shmId.shmId + "_mng");
+    string managementSegmentName("fmq_" + shmId + "_mng");
     try {
         using namespace boost::interprocess;
         managed_shared_memory managementSegment(open_only, managementSegmentName.c_str());
 
         Uint16SegmentInfoHashMap* segmentInfos = managementSegment.find<Uint16SegmentInfoHashMap>(unique_instance).first;
-
         for (const auto& s : *segmentInfos) {
             if (verbose) {
-                cout << "Resetting content of segment '" << "fmq_" << shmId.shmId << "_m_" << s.first << "'..." << endl;
+                cout << "Resetting content of segment '" << "fmq_" << shmId << "_m_" << s.first << "'..." << endl;
             }
             try {
                 if (s.second.fAllocationAlgorithm == AllocationAlgorithm::rbtree_best_fit) {
-                    RBTreeBestFitSegment segment(open_only, std::string("fmq_" + shmId.shmId + "_m_" + to_string(s.first)).c_str());
+                    RBTreeBestFitSegment segment(open_only, std::string("fmq_" + shmId + "_m_" + to_string(s.first)).c_str());
                     void* ptr = segment.get_segment_manager();
                     size_t size = segment.get_segment_manager()->get_size();
                     new(ptr) segment_manager<char, rbtree_best_fit<mutex_family, offset_ptr<void>>, null_index>(size);
                 } else {
-                    SimpleSeqFitSegment segment(open_only, std::string("fmq_" + shmId.shmId + "_m_" + to_string(s.first)).c_str());
+                    SimpleSeqFitSegment segment(open_only, std::string("fmq_" + shmId + "_m_" + to_string(s.first)).c_str());
                     void* ptr = segment.get_segment_manager();
                     size_t size = segment.get_segment_manager()->get_size();
                     new(ptr) segment_manager<char, simple_seq_fit<mutex_family, offset_ptr<void>>, null_index>(size);
                 }
             } catch (bie& e) {
                 if (verbose) {
-                    cout << "Error resetting content of segment '" << std::string("fmq_" + shmId.shmId + "_m_" + to_string(s.first)) << "': " << e.what() << endl;
+                    cout << "Error resetting content of segment '" << std::string("fmq_" + shmId + "_m_" + to_string(s.first)) << "': " << e.what() << endl;
                 }
             }
         }
+
+        Uint16RegionInfoHashMap* shmRegions = managementSegment.find<Uint16RegionInfoHashMap>(bipc::unique_instance).first;
+        if (shmRegions) {
+            for (const auto& region : *shmRegions) {
+                uint16_t id = region.first;
+                Remove<bipc::message_queue>("fmq_" + shmId + "_rgq_" + to_string(id), verbose);
+            }
+        }
     } catch (bie& e) {
         if (verbose) {
             cout << "Could not find '" << managementSegmentName << "' segment. Nothing to cleanup." << endl;
@@ -679,7 +698,7 @@ void Monitor::ResetContent(const ShmId& shmId, bool verbose /* = true */)
     }
 
     if (verbose) {
-        cout << "Done resetting segment content for shared memory id '" << shmId.shmId << "'." << endl;
+        cout << "Done resetting segment content for shared memory id '" << shmId << "'." << endl;
     }
 }
 
@@ -692,6 +711,43 @@ void Monitor::ResetContent(const SessionId& sessionId, bool verbose /* = true */
     ResetContent(shmId, verbose);
 }
 
+void Monitor::ResetContent(const ShmId& shmIdT, const std::vector<SegmentConfig>& segmentCfgs, const std::vector<RegionConfig>& regionCfgs, bool verbose /* = true */)
+{
+    using namespace boost::interprocess;
+
+    std::string shmId = shmIdT.shmId;
+    std::string managementSegmentName("fmq_" + shmId + "_mng");
+    // reset managed segments
+    ResetContent(shmIdT, verbose);
+    // delete management segment
+    Remove<bipc::shared_memory_object>(managementSegmentName, verbose);
+    // recreate management segment
+    managed_shared_memory mngSegment(create_only, managementSegmentName.c_str(), kManagementSegmentSize);
+    // fill management segment with segment & region infos
+    for (const auto& s : segmentCfgs) {
+        if (s.allocationAlgorithm == "rbtree_best_fit") {
+            Segment::Register(shmId, s.id, AllocationAlgorithm::rbtree_best_fit);
+        } else if (s.allocationAlgorithm == "simple_seq_fit") {
+            Segment::Register(shmId, s.id, AllocationAlgorithm::simple_seq_fit);
+        } else {
+            LOG(error) << "Unknown allocation algorithm provided: " << s.allocationAlgorithm;
+            throw MonitorError("Unknown allocation algorithm provided: " + s.allocationAlgorithm);
+        }
+    }
+    for (const auto& r : regionCfgs) {
+        fair::mq::shmem::UnmanagedRegion::Register(shmId, r);
+    }
+}
+
+void Monitor::ResetContent(const SessionId& sessionId, const std::vector<SegmentConfig>& segmentCfgs, const std::vector<RegionConfig>& regionCfgs, bool verbose /* = true */)
+{
+    ShmId shmId{makeShmIdStr(sessionId.sessionId)};
+    if (verbose) {
+        cout << "ResetContent called with session id '" << sessionId.sessionId << "', translating to shared memory id '" << shmId.shmId << "'" << endl;
+    }
+    ResetContent(shmId, segmentCfgs, regionCfgs, verbose);
+}
+
 Monitor::~Monitor()
 {
     if (fSignalThread.joinable()) {

fairmq/shmem/Monitor.h

@@ -8,6 +8,8 @@
 #ifndef FAIR_MQ_SHMEM_MONITOR_H_
 #define FAIR_MQ_SHMEM_MONITOR_H_
 
+#include <fairmq/UnmanagedRegion.h>
+
 #include <fairlogger/Logger.h>
 
 #include <thread>
@@ -49,6 +51,13 @@ struct BufferDebugInfo
     uint64_t fCreationTime;
 };
 
+struct SegmentConfig
+{
+    uint16_t id;
+    uint64_t size;
+    std::string allocationAlgorithm;
+};
+
 class Monitor
 {
   public:
@@ -88,6 +97,14 @@ class Monitor
     /// @param sessionId session id
     /// Only call this when segment is not in use
     static void ResetContent(const SessionId& sessionId, bool verbose = true);
+    /// @brief [EXPERIMENTAL] cleanup the content of the shem segment, without recreating it
+    /// @param shmId shared memory id
+    /// Only call this when segment is not in use
+    static void ResetContent(const ShmId& shmId, const std::vector<SegmentConfig>& segmentCfgs, const std::vector<RegionConfig>& regionCfgs, bool verbose = true);
+    /// @brief [EXPERIMENTAL] cleanup the content of the shem segment, without recreating it
+    /// @param sessionId session id
+    /// Only call this when segment is not in use
+    static void ResetContent(const SessionId& sessionId, const std::vector<SegmentConfig>& segmentCfgs, const std::vector<RegionConfig>& regionCfgs, bool verbose = true);
 
     /// @brief Outputs list of messages in shmem (if compiled with FAIRMQ_DEBUG_MODE=ON)
     /// @param shmId shmem id

fairmq/shmem/Segment.h

@@ -26,6 +26,8 @@ static const RBTreeBestFit rbTreeBestFit = RBTreeBestFit();
 
 struct Segment
 {
+    friend class Monitor;
+
     Segment(const std::string& shmId, uint16_t id, size_t size, SimpleSeqFit)
         : fSegment(SimpleSeqFitSegment(boost::interprocess::open_or_create,
                                        std::string("fmq_" + shmId + "_m_" + std::to_string(id)).c_str(),
@@ -66,15 +68,12 @@ struct Segment
     static void Register(const std::string& shmId, uint16_t id, AllocationAlgorithm allocAlgo)
     {
         using namespace boost::interprocess;
-        managed_shared_memory mngSegment(open_or_create, std::string("fmq_" + shmId + "_mng").c_str(), 6553600);
+        managed_shared_memory mngSegment(open_or_create, std::string("fmq_" + shmId + "_mng").c_str(), kManagementSegmentSize);
         VoidAlloc alloc(mngSegment.get_segment_manager());
 
         Uint16SegmentInfoHashMap* shmSegments = mngSegment.find_or_construct<Uint16SegmentInfoHashMap>(unique_instance)(alloc);
 
-        EventCounter* eventCounter = mngSegment.find<EventCounter>(unique_instance).first;
+        EventCounter* eventCounter = mngSegment.find_or_construct<EventCounter>(unique_instance)(0);
-        if (!eventCounter) {
-            eventCounter = mngSegment.construct<EventCounter>(unique_instance)(0);
-        }
 
         bool newSegmentRegistered = shmSegments->emplace(id, allocAlgo).second;
         if (newSegmentRegistered) {

fairmq/shmem/UnmanagedRegion.h

@@ -41,6 +41,7 @@ struct UnmanagedRegion
 {
     friend class Message;
     friend class Manager;
+    friend class Monitor;
 
     UnmanagedRegion(const std::string& shmId, uint16_t id, uint64_t size)
         : UnmanagedRegion(shmId,  size, false, makeRegionConfig(id))
@@ -50,6 +51,10 @@ struct UnmanagedRegion
         : UnmanagedRegion(shmId, size, false, std::move(cfg))
     {}
 
+    UnmanagedRegion(const std::string& shmId, RegionConfig cfg)
+        : UnmanagedRegion(shmId, cfg.size, false, std::move(cfg))
+    {}
+
     UnmanagedRegion(const std::string& shmId, uint64_t size, bool remote, RegionConfig cfg)
         : fRemote(remote)
         , fRemoveOnDestruction(cfg.removeOnDestruction)
@@ -66,6 +71,9 @@ struct UnmanagedRegion
     {
         using namespace boost::interprocess;
 
+        // TODO: refactor this
+        cfg.size = size;
+
         if (!cfg.path.empty()) {
             fName = std::string(cfg.path + fName);
 
@@ -223,20 +231,17 @@ struct UnmanagedRegion
         return regionCfg;
     }
 
-    static void Register(const std::string& shmId, RegionConfig& cfg)
+    static void Register(const std::string& shmId, const RegionConfig& cfg)
     {
         using namespace boost::interprocess;
-        managed_shared_memory mngSegment(open_or_create, std::string("fmq_" + shmId + "_mng").c_str(), 6553600);
+        managed_shared_memory mngSegment(open_or_create, std::string("fmq_" + shmId + "_mng").c_str(), kManagementSegmentSize);
         VoidAlloc alloc(mngSegment.get_segment_manager());
 
         Uint16RegionInfoHashMap* shmRegions = mngSegment.find_or_construct<Uint16RegionInfoHashMap>(unique_instance)(alloc);
 
-        EventCounter* eventCounter = mngSegment.find<EventCounter>(unique_instance).first;
+        EventCounter* eventCounter = mngSegment.find_or_construct<EventCounter>(unique_instance)(0);
-        if (!eventCounter) {
-            eventCounter = mngSegment.construct<EventCounter>(unique_instance)(0);
-        }
 
-        bool newShmRegionCreated = shmRegions->emplace(cfg.id.value(), RegionInfo(cfg.path.c_str(), cfg.creationFlags, cfg.userFlags, alloc)).second;
+        bool newShmRegionCreated = shmRegions->emplace(cfg.id.value(), RegionInfo(cfg.path.c_str(), cfg.creationFlags, cfg.userFlags, cfg.size, alloc)).second;
         if (newShmRegionCreated) {
             (eventCounter->fCount)++;
         }