Add test for externally (outside the session) created shmem region

And link to the GH workflow page instead of fair-software.eu

.github/workflows/fair-software.yml

@@ -0,0 +1,15 @@
+name: fair-software
+
+on: push
+
+jobs:
+  verify:
+    name: "fair-software"
+    runs-on: ubuntu-latest
+    steps:
+      - uses: fair-software/howfairis-github-action@0.2.1
+        name: Measure compliance with fair-software.eu recommendations
+        env:
+          PYCHARM_HOSTED: "Trick colorama into displaying colored output"
+        with:
+          MY_REPO_URL: "https://github.com/${{ github.repository }}"

.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

README.md

@@ -1,5 +1,10 @@
 <!-- {#mainpage} -->
-# FairMQ [![license](https://alfa-ci.gsi.de/shields/badge/license-LGPL--3.0-orange.svg)](COPYRIGHT)
+# FairMQ
+
+[![license](https://alfa-ci.gsi.de/shields/badge/license-LGPL--3.0-orange.svg)](COPYRIGHT)
+[![DOI](https://zenodo.org/badge/DOI/10.5281/zenodo.1689985.svg)](https://doi.org/10.5281/zenodo.1689985)
+[![OpenSSF Best Practices](https://bestpractices.coreinfrastructure.org/projects/6915/badge)](https://bestpractices.coreinfrastructure.org/projects/6915)
+[![fair-software.eu](https://img.shields.io/badge/fair--software.eu-%E2%97%8F%20%20%E2%97%8F%20%20%E2%97%8B%20%20%E2%97%8F%20%20%E2%97%8F-yellow)](https://github.com/FairRootGroup/FairMQ/actions/workflows/fair-software.yml)
 
 C++ Message Queuing Library and Framework
 

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,26 @@ 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 = !fExternalRegion; // mlock region after creation
+        regionCfg.zero = !fExternalRegion; // 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 +100,7 @@ struct Sampler : fair::mq::Device
     }
 
   private:
+    int fExternalRegion = false;
     int fMsgSize = 10000;
     uint32_t fLinger = 100;
     uint64_t fMaxIterations = 0;
@@ -105,7 +115,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

@@ -207,22 +207,22 @@ class Manager
 
             fEventCounter = fManagementSegment.find<EventCounter>(unique_instance).first;
             if (fEventCounter) {
-                LOG(debug) << "event counter found: " << fEventCounter->fCount;
+                LOG(trace) << "event counter found: " << fEventCounter->fCount;
             } else {
-                LOG(debug) << "no event counter found, creating one and initializing with 0";
+                LOG(trace) << "no event counter found, creating one and initializing with 0";
                 fEventCounter = fManagementSegment.construct<EventCounter>(unique_instance)(0);
-                LOG(debug) << "initialized event counter with: " << fEventCounter->fCount;
+                LOG(trace) << "initialized event counter with: " << fEventCounter->fCount;
             }
 
             fDeviceCounter = fManagementSegment.find<DeviceCounter>(unique_instance).first;
             if (fDeviceCounter) {
-                LOG(debug) << "device counter found, with value of " << fDeviceCounter->fCount << ". incrementing.";
+                LOG(trace) << "device counter found, with value of " << fDeviceCounter->fCount << ". incrementing.";
                 (fDeviceCounter->fCount)++;
-                LOG(debug) << "incremented device counter, now: " << fDeviceCounter->fCount;
+                LOG(trace) << "incremented device counter, now: " << fDeviceCounter->fCount;
             } else {
-                LOG(debug) << "no device counter found, creating one and initializing with 1";
+                LOG(trace) << "no device counter found, creating one and initializing with 1";
                 fDeviceCounter = fManagementSegment.construct<DeviceCounter>(unique_instance)(1);
-                LOG(debug) << "initialized device counter with: " << fDeviceCounter->fCount;
+                LOG(trace) << "initialized device counter with: " << fDeviceCounter->fCount;
             }
 
             fShmSegments = fManagementSegment.find_or_construct<Uint16SegmentInfoHashMap>(unique_instance)(fShmVoidAlloc);
@@ -265,10 +265,10 @@ class Manager
                         }
                     }
                 }
-                LOG(debug) << "Created/opened shared memory segment '" << "fmq_" << fShmId << "_m_" << fSegmentId << "'."
-                << " Size: " << boost::apply_visitor(SegmentSize(), fSegments.at(fSegmentId)) << " bytes."
-                << " Available: " << boost::apply_visitor(SegmentFreeMemory(), fSegments.at(fSegmentId)) << " bytes."
-                << " Allocation algorithm: " << allocationAlgorithm;
+                LOG(debug) << (createdSegment ? "Created" : "Opened") << " managed shared memory segment " << "fmq_" << fShmId << "_m_" << fSegmentId
+                    << ". Size: " << boost::apply_visitor(SegmentSize(), fSegments.at(fSegmentId)) << " bytes."
+                    << " Available: " << boost::apply_visitor(SegmentFreeMemory(), fSegments.at(fSegmentId)) << " bytes."
+                    << " Allocation algorithm: " << allocationAlgorithm;
             } catch (interprocess_exception& bie) {
                 LOG(error) << "Failed to create/open shared memory segment '" << "fmq_" << fShmId << "_m_" << fSegmentId << "': " << bie.what();
                 throw TransportError(tools::ToString("Failed to create/open shared memory segment '", "fmq_", fShmId, "_m_", fSegmentId, "': ", bie.what()));
@@ -395,17 +395,25 @@ 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();
-                // 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.
+                UnmanagedRegion* region = nullptr;
+
+                auto it = fRegions.find(id);
+                if (it != fRegions.end()) {
+                    region = it->second.get();
+                    if (region->fControlling) {
+                        LOG(error) << "Unmanaged Region with id " << id << " already exists. Only unique IDs per session are allowed.";
+                        throw TransportError(tools::ToString("Unmanaged Region with id ", id, " already exists. Only unique IDs per session are allowed."));
+                    }
+
+                    LOG(debug) << "Unmanaged region (view) already present, promoting to controller";
+                    region->BecomeController(cfg);
+                } else {
+                    auto res = fRegions.emplace(id, std::make_unique<UnmanagedRegion>(fShmId, size, true, cfg));
+                    region = res.first->second.get();
                 }
+                // LOG(debug) << "Created region with id '" << id << "', path: '" << cfg.path << "', flags: '" << cfg.creationFlags << "'";
 
                 // start ack receiver only if a callback has been provided.
                 if (callback || bulkCallback) {
@@ -455,7 +463,6 @@ class Manager
     UnmanagedRegion* GetRegion(uint16_t id)
     {
         std::lock_guard<std::mutex> lock(fLocalRegionsMtx);
-        // remote region could actually be a local one if a message originates from this device (has been sent out and returned)
         auto it = fRegions.find(id);
         if (it != fRegions.end()) {
             return it->second.get();
@@ -472,7 +479,7 @@ class Manager
                 }
                 // LOG(debug) << "Located remote region with id '" << id << "', path: '" << cfg.path << "', flags: '" << cfg.creationFlags << "'";
 
-                auto r = fRegions.emplace(id, std::make_unique<UnmanagedRegion>(fShmId, 0, true, std::move(cfg)));
+                auto r = fRegions.emplace(id, std::make_unique<UnmanagedRegion>(fShmId, 0, false, std::move(cfg)));
                 r.first->second->InitializeQueues();
                 r.first->second->StartAckSender();
                 return r.first->second.get();
@@ -540,7 +547,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
@@ -563,7 +570,7 @@ class Manager
                     if (it != fRegions.end()) {
                         region = it->second.get();
                     } else {
-                        auto r = fRegions.emplace(cfgIt->first, std::make_unique<UnmanagedRegion>(fShmId, 0, true, cfgIt->second));
+                        auto r = fRegions.emplace(cfgIt->first, std::make_unique<UnmanagedRegion>(fShmId, 0, false, cfgIt->second));
                         region = r.first->second.get();
                         region->InitializeQueues();
                         region->StartAckSender();

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

@@ -44,19 +44,19 @@ struct UnmanagedRegion
     friend class Monitor;
 
     UnmanagedRegion(const std::string& shmId, uint16_t id, uint64_t size)
-        : UnmanagedRegion(shmId,  size, false, makeRegionConfig(id))
+        : UnmanagedRegion(shmId,  size, true, makeRegionConfig(id))
     {}
 
     UnmanagedRegion(const std::string& shmId, uint64_t size, RegionConfig cfg)
-        : UnmanagedRegion(shmId, size, false, std::move(cfg))
+        : UnmanagedRegion(shmId, size, true, std::move(cfg))
     {}
 
     UnmanagedRegion(const std::string& shmId, RegionConfig cfg)
-        : UnmanagedRegion(shmId, cfg.size, false, std::move(cfg))
+        : UnmanagedRegion(shmId, cfg.size, true, std::move(cfg))
     {}
 
-    UnmanagedRegion(const std::string& shmId, uint64_t size, bool remote, RegionConfig cfg)
-        : fRemote(remote)
+    UnmanagedRegion(const std::string& shmId, uint64_t size, bool controlling, RegionConfig cfg)
+        : fControlling(controlling)
         , fRemoveOnDestruction(cfg.removeOnDestruction)
         , fLinger(cfg.linger)
         , fStopAcks(false)
@@ -73,11 +73,15 @@ struct UnmanagedRegion
 
         // TODO: refactor this
         cfg.size = size;
+        const uint16_t id = cfg.id.value();
+        bool created = false;
+
+        LOG(debug) << "UnmanagedRegion(): " << fName << " (" << (fControlling ? "controller" : "viewer") << ")";
 
         if (!cfg.path.empty()) {
             fName = std::string(cfg.path + fName);
 
-            if (!fRemote) {
+            if (fControlling) {
                 // create a file
                 std::filebuf fbuf;
                 if (fbuf.open(fName, std::ios_base::in | std::ios_base::out | std::ios_base::trunc | std::ios_base::binary)) {
@@ -92,23 +96,30 @@ struct UnmanagedRegion
             if (!fFile) {
                 LOG(error) << "Failed to initialize file: " << fName;
                 LOG(error) << "errno: " << errno << ": " << strerror(errno);
-                throw std::runtime_error(tools::ToString("Failed to initialize file for shared memory region: ", strerror(errno)));
+                throw TransportError(tools::ToString("Failed to initialize file for shared memory region: ", strerror(errno)));
             }
             fFileMapping = file_mapping(fName.c_str(), read_write);
-            LOG(debug) << "shmem: initialized file: " << fName;
+            LOG(debug) << "UnmanagedRegion(): initialized file: " << fName;
             fRegion = mapped_region(fFileMapping, read_write, 0, size, 0, cfg.creationFlags);
         } else {
             try {
                 // if opening fails, create
                 try {
                     fShmemObject = shared_memory_object(open_only, fName.c_str(), read_write);
+                    created = false;
                 } catch (interprocess_exception& e) {
-                    LOG(debug) << "Could not open " << (remote ? "remote" : "local") << " shared_memory_object for region id '" << cfg.id.value() << "': " << e.what() << ", creating...";
-                    fShmemObject = shared_memory_object(create_only, fName.c_str(), read_write);
-                    fShmemObject.truncate(size);
+                    if (fControlling) {
+                        LOG(debug) << "Could not open controlling shared_memory_object for region " << id << ": " << e.what() << ", creating...";
+                        fShmemObject = shared_memory_object(create_only, fName.c_str(), read_write);
+                        fShmemObject.truncate(size);
+                        created = true;
+                    } else {
+                        LOG(error) << "Could not open view for shared_memory_object for region " << id << ": " << e.what();
+                        throw TransportError(tools::ToString("Could not open view for shared_memory_object for region ", id, ": ", e.what()));
+                    }
                 }
             } catch (interprocess_exception& e) {
-                LOG(error) << "Failed " << (remote ? "opening" : "creating") << " shared_memory_object for region id '" << cfg.id.value() << "': " << e.what();
+                LOG(error) << "Failed initializing shared_memory_object for region id " << id << ": " << e.what();
                 throw;
             }
 
@@ -119,27 +130,27 @@ struct UnmanagedRegion
                     throw TransportError(tools::ToString("Created/opened region size (", fRegion.get_size(), ") does not match configured size (", size, ")"));
                 }
             } catch (interprocess_exception& e) {
-                LOG(error) << "Failed mapping shared_memory_object for region id '" << cfg.id.value() << "': " << e.what();
+                LOG(error) << "Failed mapping shared_memory_object for region id " << id << ": " << e.what();
                 throw;
             }
         }
 
         if (cfg.lock) {
-            LOG(debug) << "Locking region " << cfg.id.value() << "...";
+            LOG(debug) << "Locking region " << id << "...";
             Lock();
-            LOG(debug) << "Successfully locked region " << cfg.id.value() << ".";
+            LOG(debug) << "Successfully locked region " << id << ".";
         }
         if (cfg.zero) {
-            LOG(debug) << "Zeroing free memory of region " << cfg.id.value() << "...";
+            LOG(debug) << "Zeroing free memory of region " << id << "...";
             Zero();
-            LOG(debug) << "Successfully zeroed free memory of region " << cfg.id.value() << ".";
+            LOG(debug) << "Successfully zeroed free memory of region " << id << ".";
         }
 
-        if (!remote) {
+        if (fControlling && created) {
             Register(shmId, cfg);
         }
 
-        LOG(trace) << "shmem: initialized region: " << fName << " (" << (remote ? "remote" : "local") << ")";
+        LOG(debug) << (created ? "Created" : "Opened") << " unmanaged shared memory region: " << fName << " (" << (fControlling ? "controller" : "viewer") << ")";
     }
 
     UnmanagedRegion() = delete;
@@ -149,6 +160,13 @@ struct UnmanagedRegion
     UnmanagedRegion& operator=(const UnmanagedRegion&) = delete;
     UnmanagedRegion& operator=(UnmanagedRegion&&) = delete;
 
+    void BecomeController(RegionConfig& cfg)
+    {
+        fControlling = true;
+        fLinger = cfg.linger;
+        fRemoveOnDestruction = cfg.removeOnDestruction;
+    }
+
     void Zero()
     {
         memset(fRegion.get_address(), 0x00, fRegion.get_size());
@@ -171,6 +189,7 @@ struct UnmanagedRegion
 
     ~UnmanagedRegion()
     {
+        LOG(debug) << "~UnmanagedRegion(): " << fName << " (" << (fControlling ? "controller" : "viewer") << ")";
         fStopAcks = true;
 
         if (fAcksSender.joinable()) {
@@ -178,7 +197,7 @@ struct UnmanagedRegion
             fAcksSender.join();
         }
 
-        if (!fRemote) {
+        if (fControlling) {
             if (fAcksReceiver.joinable()) {
                 fAcksReceiver.join();
             }
@@ -204,14 +223,14 @@ struct UnmanagedRegion
                 fclose(fFile);
             }
         } else {
-            // LOG(debug) << "Region queue '" << fQueueName << "' is remote, no cleanup necessary";
+            // LOG(debug) << "Region queue '" << fQueueName << "' is viewer, no cleanup necessary";
         }
 
-        // LOG(debug) << "Region '" << fName << "' (" << (fRemote ? "remote" : "local") << ") destructed.";
+        // LOG(debug) << "Region '" << fName << "' (" << (fControlling ? "controller" : "viewer") << ") destructed.";
     }
 
   private:
-    bool fRemote;
+    bool fControlling;
     bool fRemoveOnDestruction;
     uint32_t fLinger;
     std::atomic<bool> fStopAcks;
@@ -243,6 +262,7 @@ struct UnmanagedRegion
     static void Register(const std::string& shmId, const RegionConfig& cfg)
     {
         using namespace boost::interprocess;
+        LOG(debug) << "Registering unmanaged shared memory region with id " << cfg.id.value();
         managed_shared_memory mngSegment(open_or_create, std::string("fmq_" + shmId + "_mng").c_str(), kManagementSegmentSize);
         VoidAlloc alloc(mngSegment.get_segment_manager());
 
@@ -250,10 +270,14 @@ struct UnmanagedRegion
 
         EventCounter* eventCounter = mngSegment.find_or_construct<EventCounter>(unique_instance)(0);
 
-        bool newShmRegionCreated = shmRegions->emplace(cfg.id.value(), RegionInfo(cfg.path.c_str(), cfg.creationFlags, cfg.userFlags, cfg.size, alloc)).second;
-        if (newShmRegionCreated) {
-            (eventCounter->fCount)++;
+        auto it = shmRegions->find(cfg.id.value());
+        if (it != shmRegions->end()) {
+            LOG(error) << "Unmanaged Region with id " << cfg.id.value() << " has already been registered. Only unique IDs per session are allowed.";
+            throw TransportError(tools::ToString("Unmanaged Region with id ", cfg.id.value(), " has already been registered. Only unique IDs per session are allowed."));
         }
+
+        shmRegions->emplace(cfg.id.value(), RegionInfo(cfg.path.c_str(), cfg.creationFlags, cfg.userFlags, cfg.size, alloc)).second;
+        (eventCounter->fCount)++;
     }
 
     void SetCallbacks(RegionCallback callback, RegionBulkCallback bulkCallback)

fairmq/shmem/UnmanagedRegionImpl.h

@@ -40,9 +40,9 @@ class UnmanagedRegionImpl final : public fair::mq::UnmanagedRegion
         , fRegion(nullptr)
         , fRegionId(0)
     {
-        auto result = fManager.CreateRegion(size, callback, bulkCallback, std::move(cfg));
-        fRegion = result.first;
-        fRegionId = result.second;
+        auto [regionPtr, regionId] = fManager.CreateRegion(size, callback, bulkCallback, std::move(cfg));
+        fRegion = regionPtr;
+        fRegionId = regionId;
     }
 
     UnmanagedRegionImpl(const UnmanagedRegionImpl&) = delete;

test/region/_region.cxx

@@ -6,6 +6,11 @@
  *                  copied verbatim in the file "LICENSE"                       *
  ********************************************************************************/
 
+#include <fairmq/shmem/Common.h>
+#include <fairmq/shmem/Monitor.h>
+#include <fairmq/shmem/Segment.h>
+#include <fairmq/shmem/UnmanagedRegion.h>
+
 #include <fairmq/TransportFactory.h>
 #include <fairmq/ProgOptions.h>
 #include <fairmq/tools/Unique.h>
@@ -16,8 +21,12 @@
 
 #include <gtest/gtest.h>
 
+#include <cstdint>
+#include <map>
 #include <memory> // make_unique
 #include <string>
+#include <utility> // pair
+#include <vector> // pair
 
 namespace
 {
@@ -25,6 +34,34 @@ namespace
 using namespace std;
 using namespace fair::mq;
 
+struct ShmOwner
+{
+    ShmOwner(const string& sessionId,
+             const vector<pair<uint16_t, size_t>>& segments,
+             const vector<pair<uint16_t, size_t>>& regions)
+        : fShmId(fair::mq::shmem::makeShmIdStr(sessionId))
+    {
+        LOG(info) << "ShmOwner: creating segments";
+        for (auto [id, size] : segments) {
+            fSegments.emplace(id, fair::mq::shmem::Segment(fShmId, id, size, fair::mq::shmem::rbTreeBestFit));
+        }
+        LOG(info) << "ShmOwner: creating regions";
+        for (auto [id, size] : regions) {
+            fRegions.emplace(id, make_unique<fair::mq::shmem::UnmanagedRegion>(fShmId, id, size));
+        }
+    }
+
+    ~ShmOwner()
+    {
+        LOG(info) << "ShmOwner: cleaning up";
+        fair::mq::shmem::Monitor::Cleanup(fair::mq::shmem::ShmId{fShmId});
+    }
+
+    string fShmId;
+    map<uint16_t, fair::mq::shmem::Segment> fSegments;
+    map<uint16_t, unique_ptr<fair::mq::shmem::UnmanagedRegion>> fRegions;
+};
+
 void RegionsSizeMismatch()
 {
     size_t session = tools::UuidHash();
@@ -108,31 +145,69 @@ void RegionsCache(const string& transport, const string& address)
     }
 }
 
-void RegionEventSubscriptions(const string& transport)
+void RegionEventSubscriptions(const string& transport, bool external)
 {
+    fair::Logger::SetConsoleSeverity(fair::Severity::debug);
+
+    unique_ptr<ShmOwner> shmOwner = nullptr;
+
     size_t session{tools::UuidHash()};
 
+    constexpr int sSize = 100000000;
+    constexpr int r1Size = 1000000;
+    constexpr int r2Size = 5000000;
+    constexpr uint16_t sId = 0;
+    constexpr uint16_t r1id = 100;
+    constexpr uint16_t r2id = 101;
+
+    if (external) {
+        shmOwner = make_unique<ShmOwner>(
+            to_string(session),
+            vector<pair<uint16_t, size_t>>{ { sId, sSize } },
+            vector<pair<uint16_t, size_t>>{ { r1id, r1Size }, { r2id, r2Size } }
+        );
+    }
+
     ProgOptions config;
     config.SetProperty<string>("session", to_string(session));
-    config.SetProperty<size_t>("shm-segment-size", 100000000);
+    config.SetProperty<size_t>("shm-segment-size", sSize);
+    if (external) {
+        config.SetProperty<bool>("shm-no-cleanup", true);
+        config.SetProperty<bool>("shm-monitor", false);
+    }
 
     auto factory = TransportFactory::CreateTransportFactory(transport, tools::Uuid(), &config);
 
-    constexpr int size1 = 1000000;
-    constexpr int size2 = 5000000;
     constexpr int64_t userFlags = 12345;
     tools::Semaphore blocker;
 
     {
-        auto region1 = factory->CreateUnmanagedRegion(size1, [](void*, size_t, void*) {});
+        fair::mq::RegionConfig r1Cfg;
+        if (external) {
+            r1Cfg.id = r1id;
+            r1Cfg.removeOnDestruction = false;
+        }
+        auto region1 = factory->CreateUnmanagedRegion(r1Size, [](void*, size_t, void*) {}, r1Cfg);
         void* ptr1 = region1->GetData();
         uint64_t id1 = region1->GetId();
-        ASSERT_EQ(region1->GetSize(), size1);
+        if (external) {
+            ASSERT_EQ(id1, r1id);
+        }
+        ASSERT_EQ(region1->GetSize(), r1Size);
 
-        auto region2 = factory->CreateUnmanagedRegion(size2, userFlags, [](void*, size_t, void*) {});
+        fair::mq::RegionConfig r2Cfg;
+        r2Cfg.userFlags = userFlags;
+        if (external) {
+            r2Cfg.id = r2id;
+            r2Cfg.removeOnDestruction = false;
+        }
+        auto region2 = factory->CreateUnmanagedRegion(r2Size, [](void*, size_t, void*) {}, r2Cfg);
         void* ptr2 = region2->GetData();
         uint64_t id2 = region2->GetId();
-        ASSERT_EQ(region2->GetSize(), size2);
+        if (external) {
+            ASSERT_EQ(id2, r2id);
+        }
+        ASSERT_EQ(region2->GetSize(), r2Size);
 
         ASSERT_EQ(factory->SubscribedToRegionEvents(), false);
         factory->SubscribeToRegionEvents([&, id1, id2, ptr1, ptr2](RegionInfo info) {
@@ -144,13 +219,15 @@ void RegionEventSubscriptions(const string& transport)
                       << ", flags: " << info.flags;
             if (info.event == RegionEvent::created) {
                 if (info.id == id1) {
-                    ASSERT_EQ(info.size, size1);
+                    ASSERT_EQ(info.size, r1Size);
                     ASSERT_EQ(info.ptr, ptr1);
                     blocker.Signal();
                 } else if (info.id == id2) {
-                    ASSERT_EQ(info.size, size2);
+                    ASSERT_EQ(info.size, r2Size);
                     ASSERT_EQ(info.ptr, ptr2);
-                    ASSERT_EQ(info.flags, userFlags);
+                    if (!external) {
+                        ASSERT_EQ(info.flags, userFlags);
+                    }
                     blocker.Signal();
                 }
             } else if (info.event == RegionEvent::destroyed) {
@@ -170,10 +247,12 @@ void RegionEventSubscriptions(const string& transport)
         LOG(info) << "2 done.";
     }
 
-    blocker.Wait();
-    LOG(info) << "3 done.";
-    blocker.Wait();
-    LOG(info) << "4 done.";
+    if (!external) {
+        blocker.Wait();
+        LOG(info) << "3 done.";
+        blocker.Wait();
+        LOG(info) << "4 done.";
+    }
     LOG(info) << "All done.";
 
     factory->UnsubscribeFromRegionEvents();
@@ -185,9 +264,13 @@ void RegionCallbacks(const string& transport, const string& _address)
     size_t session(tools::UuidHash());
     std::string address(tools::ToString(_address, "_", transport));
 
+    constexpr size_t sSize = 100000000;
+    constexpr size_t r1Size = 2000000;
+    constexpr size_t r2Size = 3000000;
+
     ProgOptions config;
     config.SetProperty<string>("session", to_string(session));
-    config.SetProperty<size_t>("shm-segment-size", 100000000);
+    config.SetProperty<size_t>("shm-segment-size", sSize);
 
     auto factory = TransportFactory::CreateTransportFactory(transport, tools::Uuid(), &config);
 
@@ -206,7 +289,7 @@ void RegionCallbacks(const string& transport, const string& _address)
     void* ptr2 = nullptr;
     size_t size2 = 200;
 
-    auto region1 = factory->CreateUnmanagedRegion(2000000, [&](void* ptr, size_t size, void* hint) {
+    auto region1 = factory->CreateUnmanagedRegion(r1Size, [&](void* ptr, size_t size, void* hint) {
         ASSERT_EQ(ptr, ptr1);
         ASSERT_EQ(size, size1);
         ASSERT_EQ(hint, intPtr1.get());
@@ -215,7 +298,7 @@ void RegionCallbacks(const string& transport, const string& _address)
     });
     ptr1 = region1->GetData();
 
-    auto region2 = factory->CreateUnmanagedRegion(3000000, [&](const std::vector<RegionBlock>& blocks) {
+    auto region2 = factory->CreateUnmanagedRegion(r2Size, [&](const std::vector<RegionBlock>& blocks) {
         ASSERT_EQ(blocks.size(), 1);
         ASSERT_EQ(blocks.at(0).ptr, ptr2);
         ASSERT_EQ(blocks.at(0).size, size2);
@@ -263,12 +346,12 @@ TEST(Cache, shmem)
 
 TEST(EventSubscriptions, zeromq)
 {
-    RegionEventSubscriptions("zeromq");
+    RegionEventSubscriptions("zeromq", false);
 }
 
 TEST(EventSubscriptions, shmem)
 {
-    RegionEventSubscriptions("shmem");
+    RegionEventSubscriptions("shmem", false);
 }
 
 TEST(Callbacks, zeromq)
@@ -281,4 +364,9 @@ TEST(Callbacks, shmem)
     RegionCallbacks("shmem", "ipc://test_region_callbacks");
 }
 
+TEST(EventSubscriptionsExternalRegion, shmem)
+{
+    RegionEventSubscriptions("shmem", true);
+}
+
 } // namespace