Shm: fix region init with external regions

.zenodo.json

@@ -0,0 +1,86 @@
+{
+    "creators": [
+        {
+            "name": "Al-Turany, Mohammad"
+        },
+        {
+            "orcid": "0000-0003-3787-1910",
+            "name": "Klein, Dennis"
+        },
+        {
+            "name": "Kollegger, Thorsten"
+        },
+        {
+            "name": "Rybalchenko, Alexey"
+        },
+        {
+            "name": "Winckler, Nicolas"
+        }
+    ],
+    "contributors": [
+        {
+            "type": "Other",
+            "name": "Aphecetche, Laurent"
+        },
+        {
+            "type": "Other",
+            "name": "Binet, Sebastien"
+        },
+        {
+            "type": "Other",
+            "name": "Eulisse, Giulio"
+        },
+        {
+            "type": "Other",
+            "name": "Karabowicz, Radoslaw"
+        },
+        {
+            "type": "Other",
+            "name": "Kretz, Matthias"
+        },
+        {
+            "type": "Other",
+            "name": "Krzewicki, Mikolaj"
+        },
+        {
+            "type": "Other",
+            "name": "Lebedev, Andrey"
+        },
+        {
+            "type": "Other",
+            "name": "Mrnjavac, Teo"
+        },
+        {
+            "type": "Other",
+            "name": "Neskovic, Gvozden"
+        },
+        {
+            "type": "Other",
+            "name": "Richter, Matthias"
+        },
+        {
+            "type": "Other",
+            "orcid": "0000-0002-5321-8404",
+            "name": "Tacke, Christian"
+        },
+        {
+            "type": "Other",
+            "name": "Uhlig, Florian"
+        },
+        {
+            "type": "Other",
+            "name": "Wenzel, Sandro"
+        }
+    ],
+    "description": "<p>C++ Message Queuing Library and Framework</p>",
+    "related_identifiers": [
+        {
+            "identifier": "https://github.com/FairRootGroup/FairMQ/",
+            "relation": "isSupplementTo",
+            "resource_type": "software",
+            "scheme": "url"
+        }
+    ],
+    "title": "FairMQ",
+    "license": "LGPL-3.0-only"
+}

AUTHORS

@@ -1,5 +1,5 @@
 Al-Turany, Mohammad
-Klein, Dennis
+Klein, Dennis [https://orcid.org/0000-0003-3787-1910]
 Kollegger, Thorsten
 Rybalchenko, Alexey
 Winckler, Nicolas

CONTRIBUTORS

@@ -3,11 +3,11 @@ Binet, Sebastien
 Eulisse, Giulio
 Karabowicz, Radoslaw
 Kretz, Matthias <kretz@kde.org>
-Krzewicki, Mikolaj 
+Krzewicki, Mikolaj
 Lebedev, Andrey
 Mrnjavac, Teo <teo.m@cern.ch>
 Neskovic, Gvozden
 Richter, Matthias
-Tacke, Christian
+Tacke, Christian [https://orcid.org/0000-0002-5321-8404]
 Uhlig, Florian
 Wenzel, Sandro

codemeta.json

@@ -18,7 +18,8 @@
     {
       "@type": "Person",
       "givenName": "Dennis",
-      "familyName": "Klein"
+      "familyName": "Klein",
+      "@id": "https://orcid.org/0000-0003-3787-1910"
     },
     {
       "@type": "Person",
@@ -92,7 +93,8 @@
     {
       "@type": "Person",
       "givenName": "Christian",
-      "familyName": "Tacke"
+      "familyName": "Tacke",
+      "@id": "https://orcid.org/0000-0002-5321-8404"
     },
     {
       "@type": "Person",

examples/region/fairmq-start-ex-region.sh.in

@@ -19,6 +19,10 @@ SAMPLER+=" --severity debug"
 SAMPLER+=" --msg-size $msgSize"
 # SAMPLER+=" --rate 10"
 SAMPLER+=" --transport $transport"
+# SAMPLER+=" --external-region true"
+# SAMPLER+=" --shm-no-cleaup true"
+# SAMPLER+=" --shm-monitor false"
+# SAMPLER+=" --shmid 1"
 SAMPLER+=" --channel-config name=data,type=push,method=bind,address=tcp://127.0.0.1:7777,sndKernelSize=212992"
 xterm -geometry 120x60+0+0 -hold -e @EX_BIN_DIR@/$SAMPLER &
 
@@ -26,5 +30,8 @@ SINK="fairmq-ex-region-sink"
 SINK+=" --id sink1"
 SINK+=" --severity debug"
 SINK+=" --transport $transport"
+# SINK+=" --shm-no-cleaup true"
+# SINK+=" --shm-monitor false"
+# SINK+=" --shmid 1"
 SINK+=" --channel-config name=data,type=pull,method=connect,address=tcp://127.0.0.1:7777,rcvKernelSize=212992"
 xterm -geometry 120x60+750+0 -hold -e @EX_BIN_DIR@/$SINK &

examples/region/sampler.cxx

@@ -19,6 +19,7 @@ struct Sampler : fair::mq::Device
 {
     void InitTask() override
     {
+        fExternalRegion = fConfig->GetProperty<bool>("external-region");
         fMsgSize = fConfig->GetProperty<int>("msg-size");
         fLinger = fConfig->GetProperty<uint32_t>("region-linger");
         fMaxIterations = fConfig->GetProperty<uint64_t>("max-iterations");
@@ -34,18 +35,29 @@ struct Sampler : fair::mq::Device
 
         fair::mq::RegionConfig regionCfg;
         regionCfg.linger = fLinger; // delay in ms before region destruction to collect outstanding events
-        regionCfg.lock = true; // mlock region after creation
-        regionCfg.zero = true; // zero region content after creation
-        fRegion = fair::mq::UnmanagedRegionPtr(NewUnmanagedRegionFor("data", // region is created using the transport of this channel...
-                                                                0,      // ... and this sub-channel
-                                                                10000000, // region size
-                                                                [this](const std::vector<fair::mq::RegionBlock>& blocks) { // callback to be called when message buffers no longer needed by transport
-                                                                    std::lock_guard<std::mutex> lock(fMtx);
-                                                                    fNumUnackedMsgs -= blocks.size();
-                                                                    if (fMaxIterations > 0) {
-                                                                        LOG(info) << "Received " << blocks.size() << " acks";
-                                                                    }
-                                                                }, regionCfg));
+        // options for testing with an externally-created -region
+        if (fExternalRegion) {
+            regionCfg.id = 1;
+            regionCfg.removeOnDestruction = false;
+            regionCfg.lock = false; // mlock region after creation
+            regionCfg.lock = false; // mlock region after creation
+        } else {
+            regionCfg.lock = true; // mlock region after creation
+            regionCfg.zero = true; // zero region content after creation
+        }
+        fRegion = fair::mq::UnmanagedRegionPtr(NewUnmanagedRegionFor(
+            "data", // region is created using the transport of this channel...
+            0,      // ... and this sub-channel
+            10000000, // region size
+            [this](const std::vector<fair::mq::RegionBlock>& blocks) { // callback to be called when message buffers no longer needed by transport
+                std::lock_guard<std::mutex> lock(fMtx);
+                fNumUnackedMsgs -= blocks.size();
+                if (fMaxIterations > 0) {
+                    LOG(info) << "Received " << blocks.size() << " acks";
+                }
+            },
+            regionCfg
+        ));
     }
 
     bool ConditionalRun() override
@@ -91,6 +103,7 @@ struct Sampler : fair::mq::Device
     }
 
   private:
+    int fExternalRegion = false;
     int fMsgSize = 10000;
     uint32_t fLinger = 100;
     uint64_t fMaxIterations = 0;
@@ -105,7 +118,8 @@ void addCustomOptions(bpo::options_description& options)
     options.add_options()
         ("msg-size", bpo::value<int>()->default_value(1000), "Message size in bytes")
         ("region-linger", bpo::value<uint32_t>()->default_value(100), "Linger period for regions")
-        ("max-iterations", bpo::value<uint64_t>()->default_value(0), "Maximum number of iterations of Run/ConditionalRun/OnData (0 - infinite)");
+        ("max-iterations", bpo::value<uint64_t>()->default_value(0), "Maximum number of iterations of Run/ConditionalRun/OnData (0 - infinite)")
+        ("external-region", bpo::value<bool>()->default_value(false), "Use region created by another process");
 }
 
 std::unique_ptr<fair::mq::Device> getDevice(fair::mq::ProgOptions& /*config*/)

fairmq/shmem/Manager.h

@@ -395,18 +395,10 @@ class Manager
 
                 const uint16_t id = cfg.id.value();
 
-                UnmanagedRegion* region = nullptr;
-                bool newRegionCreated = false;
                 std::lock_guard<std::mutex> lock(fLocalRegionsMtx);
-                auto res = fRegions.emplace(id, std::make_unique<UnmanagedRegion>(fShmId, size, false, cfg));
-                newRegionCreated = res.second;
-                region = res.first->second.get();
+                auto& region = fRegions[id] = std::make_unique<UnmanagedRegion>(fShmId, size, false, cfg);
                 // LOG(debug) << "Created region with id '" << id << "', path: '" << cfg.path << "', flags: '" << cfg.creationFlags << "'";
 
-                if (!newRegionCreated) {
-                    region->fRemote = false; // TODO: this should be more clear, refactor it.
-                }
-
                 // start ack receiver only if a callback has been provided.
                 if (callback || bulkCallback) {
                     region->SetCallbacks(callback, bulkCallback);
@@ -414,7 +406,7 @@ class Manager
                     region->StartAckSender();
                     region->StartAckReceiver();
                 }
-                result.first = region;
+                result.first = region.get();
                 result.second = id;
             }
             fRegionsGen += 1; // signal TL cache invalidation
@@ -540,7 +532,7 @@ class Manager
                 if (info.event == RegionEvent::created) {
                     RegionConfig cfg;
                     cfg.id = info.id;
-                    cfg.creationFlags = info.id;
+                    cfg.creationFlags = regionInfo.fCreationFlags;
                     cfg.path = regionInfo.fPath.c_str();
                     regionCfgs.emplace(info.id, cfg);
                     // fill the ptr+size info after shmLock is released, to avoid constructing local region under it

fairmq/shmem/Monitor.cxx

@@ -670,7 +670,7 @@ std::vector<std::pair<std::string, bool>> Monitor::Cleanup(const ShmId& shmIdT,
                 string path = info.fPath.c_str();
                 int flags = info.fCreationFlags;
                 if (verbose) {
-                    LOG(info) << "Found RegionInfo with path: '" << path << "', flags: " << flags << ", fDestroyed: " << info.fDestroyed << ".";
+                    LOG(info) << "Found UnmanagedRegion with id: " << id << ", path: '" << path << "', flags: " << flags << ", fDestroyed: " << info.fDestroyed << ".";
                 }
                 if (!path.empty()) {
                     result.emplace_back(Remove<bipc::file_mapping>(path + "fmq_" + shmId + "_rg_" + to_string(id), verbose));

fairmq/shmem/UnmanagedRegion.h

@@ -74,6 +74,8 @@ struct UnmanagedRegion
         // TODO: refactor this
         cfg.size = size;
 
+        LOG(debug) << "UnmanagedRegion(): " << fName << " | remote: " << remote << ".";
+
         if (!cfg.path.empty()) {
             fName = std::string(cfg.path + fName);
 
@@ -171,6 +173,7 @@ struct UnmanagedRegion
 
     ~UnmanagedRegion()
     {
+        LOG(debug) << "~UnmanagedRegion(): " << fName << " | remote: " << fRemote << ".";
         fStopAcks = true;
 
         if (fAcksSender.joinable()) {