Retry on EINTR in blocking zmq calls

To allow potential optimizations (e.g. huge pages)

.gitmodules

@@ -4,3 +4,6 @@
 [submodule "extern/asio"]
 	path = extern/asio
 	url = https://github.com/chriskohlhoff/asio
+[submodule "extern/PicoSHA2"]
+	path = extern/PicoSHA2
+	url = https://github.com/okdshin/PicoSHA2

CMakeLists.txt

@@ -94,7 +94,7 @@ endif()
 
 if(BUILD_FAIRMQ OR BUILD_SDK)
   find_package2(PUBLIC FairLogger REQUIRED
-    VERSION 1.2.0
+    VERSION 1.6.0
   )
 
   foreach(dep IN LISTS FairLogger_PACKAGE_DEPENDENCIES)
@@ -146,6 +146,8 @@ if(BUILD_FAIRMQ)
   find_package2(PRIVATE ZeroMQ REQUIRED
     VERSION 4.1.4
   )
+  build_bundled(PicoSHA2 extern/PicoSHA2)
+  find_package2(PRIVATE PicoSHA2 REQUIRED)
 endif()
 
 if(BUILD_TESTING)

COPYRIGHT

@@ -23,6 +23,10 @@ Files: extern/asio
 Copyright: 2003-2019, Christopher M. Kohlhoff (chris at kohlhoff dot com)
 License: BSL-1.0
 
+Files: extern/PicoSHA2
+Copyright: 2017 okdshin
+License: MIT
+
 License: LGPL-3.0-only
   [see LICENSE file]
 
@@ -102,3 +106,22 @@ License: BSL-1.0
   FOR ANY DAMAGES OR OTHER LIABILITY, WHETHER IN CONTRACT, TORT OR OTHERWISE,
   ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
   DEALINGS IN THE SOFTWARE.
+
+License: MIT
+  Permission is hereby granted, free of charge, to any person obtaining a copy
+  of this software and associated documentation files (the "Software"), to deal
+  in the Software without restriction, including without limitation the rights
+  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+  copies of the Software, and to permit persons to whom the Software is
+  furnished to do so, subject to the following conditions:
+
+  The above copyright notice and this permission notice shall be included in all
+  copies or substantial portions of the Software.
+
+  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+  SOFTWARE.

FairMQTest.cmake

@@ -32,6 +32,7 @@ Set(configure_options "${configure_options};-DBUILD_SDK=ON")
 Set(configure_options "${configure_options};-DBUILD_SDK_COMMANDS=ON")
 Set(configure_options "${configure_options};-DFAST_BUILD=ON")
 Set(configure_options "${configure_options};-DCOTIRE_MAXIMUM_NUMBER_OF_UNITY_INCLUDES=-j$ENV{number_of_processors}")
+Set(configure_options "${configure_options};-DBoost_NO_BOOST_CMAKE=ON")
 
 Set(EXTRA_FLAGS $ENV{EXTRA_FLAGS})
 If(EXTRA_FLAGS)

cmake/FairMQLib.cmake

@@ -145,7 +145,7 @@ macro(set_fairmq_defaults)
   # Configure build types
   set(CMAKE_CONFIGURATION_TYPES "Debug" "Release" "RelWithDebInfo" "Nightly" "Profile" "Experimental" "AddressSan" "ThreadSan")
   set(_warnings "-Wshadow -Wall -Wextra -Wpedantic")
-  set(CMAKE_CXX_FLAGS_DEBUG          "-g ${_warnings}")
+  set(CMAKE_CXX_FLAGS_DEBUG          "-Og -g ${_warnings}")
   set(CMAKE_CXX_FLAGS_RELEASE        "-O2 -DNDEBUG")
   set(CMAKE_CXX_FLAGS_RELWITHDEBINFO "-O2 -g ${_warnings} -DNDEBUG")
   set(CMAKE_CXX_FLAGS_NIGHTLY        "-O2 -g ${_warnings}")
@@ -498,6 +498,8 @@ function(build_bundled package bundle)
     set(${package}_BUILD_INCLUDE_DIR ${${package}_SOURCE_DIR}/asio/include CACHE PATH "Bundled ${package} build-interface include dir")
     set(${package}_INSTALL_INCLUDE_DIR ${PROJECT_INSTALL_INCDIR}/bundled CACHE PATH "Bundled ${package} install-interface include dir")
     set(${package}_ROOT ${${package}_SOURCE_DIR}/asio)
+  elseif(${package} STREQUAL PicoSHA2)
+    set(${package}_ROOT ${${package}_SOURCE_DIR})
   endif()
 
   string(TOUPPER ${package} package_upper)

cmake/FindPicoSHA2.cmake

@@ -0,0 +1,21 @@
+################################################################################
+#    Copyright (C) 2020 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH    #
+#                                                                              #
+#              This software is distributed under the terms of the             #
+#              GNU Lesser General Public Licence (LGPL) version 3,             #
+#                  copied verbatim in the file "LICENSE"                       #
+################################################################################
+
+find_path(PicoSHA2_INCLUDE_DIR NAMES picosha2.h)
+
+include(FindPackageHandleStandardArgs)
+find_package_handle_standard_args(PicoSHA2
+  REQUIRED_VARS PicoSHA2_INCLUDE_DIR
+)
+
+if(PicoSHA2_FOUND)
+  add_library(PicoSHA2 INTERFACE IMPORTED)
+  set_target_properties(PicoSHA2 PROPERTIES
+    INTERFACE_INCLUDE_DIRECTORIES "${PicoSHA2_INCLUDE_DIR}"
+  )
+endif()

examples/1-1/Sampler.cxx

@@ -17,8 +17,7 @@ Sampler::Sampler()
     : fText()
     , fMaxIterations(0)
     , fNumIterations(0)
-{
-}
+{}
 
 void Sampler::InitTask()
 {
@@ -32,25 +31,22 @@ bool Sampler::ConditionalRun()
     // create a copy of the data with new(), that will be deleted after the transfer is complete
     string* text = new string(fText);
 
-    // create message object with a pointer to the data buffer,
-    // its size,
+    // create message object with a pointer to the data buffer, its size,
     // custom deletion function (called when transfer is done),
     // and pointer to the object managing the data buffer
-    FairMQMessagePtr msg(NewMessage(const_cast<char*>(text->c_str()),
-                                    text->length(),
-                                    [](void* /*data*/, void* object) { delete static_cast<string*>(object); },
-                                    text));
+    FairMQMessagePtr msg(NewMessage(
+        const_cast<char*>(text->c_str()),
+        text->length(),
+        [](void* /*data*/, void* object) { delete static_cast<string*>(object); },
+        text));
 
     LOG(info) << "Sending \"" << fText << "\"";
 
-    // in case of error or transfer interruption, return false to go to IDLE state
+    // in case of error or transfer interruption, return false to go to the Ready state
     // successfull transfer will return number of bytes transfered (can be 0 if sending an empty message).
-    if (Send(msg, "data") < 0)
-    {
+    if (Send(msg, "data") < 0) {
         return false;
-    }
-    else if (fMaxIterations > 0 && ++fNumIterations >= fMaxIterations)
-    {
+    } else if (fMaxIterations > 0 && ++fNumIterations >= fMaxIterations) {
         LOG(info) << "Configured maximum number of iterations reached. Leaving RUNNING state.";
         return false;
     }
@@ -58,8 +54,6 @@ bool Sampler::ConditionalRun()
     return true;
 }
 
-Sampler::~Sampler()
-{
-}
+Sampler::~Sampler() {}
 
 } // namespace example_1_1

examples/1-1/Sink.cxx

@@ -33,23 +33,22 @@ void Sink::InitTask()
     fMaxIterations = fConfig->GetProperty<uint64_t>("max-iterations");
 }
 
-// handler is called whenever a message arrives on "data", with a reference to the message and a sub-channel index (here 0)
+// handler is called whenever a message arrives on "data", with a reference to the message and a
+// sub-channel index (here 0)
 bool Sink::HandleData(FairMQMessagePtr& msg, int /*index*/)
 {
     LOG(info) << "Received: \"" << string(static_cast<char*>(msg->GetData()), msg->GetSize()) << "\"";
 
-    if (fMaxIterations > 0 && ++fNumIterations >= fMaxIterations)
-    {
+    if (fMaxIterations > 0 && ++fNumIterations >= fMaxIterations) {
         LOG(info) << "Configured maximum number of iterations reached. Leaving RUNNING state.";
         return false;
     }
 
-    // return true if want to be called again (otherwise return false go to IDLE state)
+    // return true if you want the handler to be called again (otherwise return false go to the
+    // Ready state)
     return true;
 }
 
-Sink::~Sink()
-{
-}
+Sink::~Sink() {}
 
 } // namespace example_1_1

examples/n-m/ex-n-m-pair-topology.xml

@@ -10,7 +10,7 @@
         <exe reachable="true">fairmq-ex-n-m-synchronizer --id sync --rate 100 --color false -P dds --channel-config name=sync,type=pub,method=bind</exe>
         <env reachable="false">fairmq-ex-n-m-env.sh</env>
         <properties>
-            <name access="write">fmqchan_sync</id>
+            <name access="write">fmqchan_sync</name>
         </properties>
     </decltask>
 
@@ -18,8 +18,8 @@
         <exe reachable="true">fairmq-ex-n-m-sender --id sender%taskIndex% --timeframe-size 100000 --num-receivers ${numReceivers} --color false -P dds --channel-config name=sync,type=sub,method=connect name=data,type=pair,method=connect,numSockets=${numReceivers} --dds-i data:%taskIndex%</exe>
         <env reachable="false">fairmq-ex-n-m-env.sh</env>
         <properties>
-            <name access="read">fmqchan_sync</id>
-            <name access="read">fmqchan_data</id>
+            <name access="read">fmqchan_sync</name>
+            <name access="read">fmqchan_data</name>
         </properties>
     </decltask>
 
@@ -27,7 +27,7 @@
         <exe reachable="true">fairmq-ex-n-m-receiver --id receiver%taskIndex% --num-senders ${numSenders} --color false -P dds --max-timeframes 10 --channel-config name=data,type=pair,method=bind,numSockets=${numSenders}</exe>
         <env reachable="false">fairmq-ex-n-m-env.sh</env>
         <properties>
-            <name access="write">fmqchan_data</id>
+            <name access="write">fmqchan_data</name>
         </properties>
     </decltask>
 

examples/n-m/ex-n-m-topology.xml

@@ -10,7 +10,7 @@
         <exe reachable="true">fairmq-ex-n-m-synchronizer --id sync --rate 100 --color false -P dds --channel-config name=sync,type=pub,method=bind</exe>
         <env reachable="false">fairmq-ex-n-m-env.sh</env>
         <properties>
-            <name access="write">fmqchan_sync</id>
+            <name access="write">fmqchan_sync</name>
         </properties>
     </decltask>
 
@@ -18,8 +18,8 @@
         <exe reachable="true">fairmq-ex-n-m-sender --id sender%taskIndex% --timeframe-size 100000 --num-receivers ${numReceivers} --color false -P dds --channel-config name=sync,type=sub,method=connect name=data,type=push,method=connect,numSockets=${numReceivers}</exe>
         <env reachable="false">fairmq-ex-n-m-env.sh</env>
         <properties>
-            <name access="read">fmqchan_sync</id>
-            <name access="read">fmqchan_data</id>
+            <name access="read">fmqchan_sync</name>
+            <name access="read">fmqchan_data</name>
         </properties>
     </decltask>
 
@@ -27,7 +27,7 @@
         <exe reachable="true">fairmq-ex-n-m-receiver --id receiver%taskIndex% --num-senders ${numSenders} --color false -P dds --max-timeframes 10 --channel-config name=data,type=pull,method=bind</exe>
         <env reachable="false">fairmq-ex-n-m-env.sh</env>
         <properties>
-            <name access="write">fmqchan_data</id>
+            <name access="write">fmqchan_data</name>
         </properties>
     </decltask>
 

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

@@ -51,7 +51,7 @@ fairmq-dds-command-ui -c k
 fairmq-dds-command-ui -c b
 fairmq-dds-command-ui -c x
 fairmq-dds-command-ui -c j
-allexceptqctasks="main/(Sampler|QCDispatcher|Sink)"
+allexceptqctasks="main/(Sampler|QCDispatcher|Sink).*"
 fairmq-dds-command-ui -c r -p $allexceptqctasks
 qctask="main/QCTask.*"
 qcdispatcher="main/QCDispatcher.*"

examples/region/Sampler.cxx

@@ -23,36 +23,40 @@ namespace example_region
 
 Sampler::Sampler()
     : fMsgSize(10000)
+    , fLinger(100)
     , fMaxIterations(0)
     , fNumIterations(0)
     , fRegion(nullptr)
     , fNumUnackedMsgs(0)
-{
-}
+{}
 
 void Sampler::InitTask()
 {
     fMsgSize = fConfig->GetProperty<int>("msg-size");
+    fLinger = fConfig->GetProperty<uint32_t>("region-linger");
     fMaxIterations = fConfig->GetProperty<uint64_t>("max-iterations");
 
     fChannels.at("data").at(0).Transport()->SubscribeToRegionEvents([](FairMQRegionInfo info) {
-        LOG(warn) << ">>>" << info.event;
-        LOG(warn) << "id: " << info.id;
-        LOG(warn) << "ptr: " << info.ptr;
-        LOG(warn) << "size: " << info.size;
-        LOG(warn) << "flags: " << info.flags;
+        LOG(info) << "Region event: " << info.event
+                  << ", id: " << info.id
+                  << ", ptr: " << info.ptr
+                  << ", size: " << info.size
+                  << ", flags: " << info.flags;
     });
 
     fRegion = FairMQUnmanagedRegionPtr(NewUnmanagedRegionFor("data",
                                                              0,
                                                              10000000,
                                                              [this](const std::vector<fair::mq::RegionBlock>& blocks) { // callback to be called when message buffers no longer needed by transport
+                                                                 lock_guard<mutex> lock(fMtx);
                                                                  fNumUnackedMsgs -= blocks.size();
+
                                                                  if (fMaxIterations > 0) {
-                                                                    LOG(debug) << "Received " << blocks.size() << " acks";
+                                                                    LOG(info) << "Received " << blocks.size() << " acks";
                                                                  }
                                                              }
                                                              ));
+    fRegion->SetLinger(fLinger);
 }
 
 bool Sampler::ConditionalRun()
@@ -69,27 +73,30 @@ bool Sampler::ConditionalRun()
     // LOG(info) << "check: " << static_cast<char*>(fRegion->GetData())[3];
     // std::this_thread::sleep_for(std::chrono::seconds(1));
 
+    lock_guard<mutex> lock(fMtx);
     if (Send(msg, "data", 0) > 0) {
-        ++fNumUnackedMsgs;
-
         if (fMaxIterations > 0 && ++fNumIterations >= fMaxIterations) {
             LOG(info) << "Configured maximum number of iterations reached. Leaving RUNNING state.";
             return false;
         }
     }
+    ++fNumUnackedMsgs;
 
     return true;
 }
 
 void Sampler::ResetTask()
 {
-    // if not all messages acknowledged, wait for a bit. But only once, since receiver could be already dead.
-    if (fNumUnackedMsgs != 0) {
-        LOG(debug) << "waiting for all acknowledgements... (" << fNumUnackedMsgs << ")";
-        this_thread::sleep_for(chrono::milliseconds(500));
-        LOG(debug) << "done, still unacked: " << fNumUnackedMsgs;
-    }
+    // On destruction UnmanagedRegion will try to TODO
     fRegion.reset();
+    {
+        lock_guard<mutex> lock(fMtx);
+        if (fNumUnackedMsgs != 0) {
+            LOG(info) << "Done, still not acknowledged: " << fNumUnackedMsgs;
+        } else {
+            LOG(info) << "All acknowledgements received.";
+        }
+    }
     fChannels.at("data").at(0).Transport()->UnsubscribeFromRegionEvents();
 }
 

examples/region/Sampler.h

@@ -15,7 +15,8 @@
 #ifndef FAIRMQEXAMPLEREGIONSAMPLER_H
 #define FAIRMQEXAMPLEREGIONSAMPLER_H
 
-#include <atomic>
+#include <mutex>
+#include <cstdint>
 
 #include "FairMQDevice.h"
 
@@ -35,10 +36,12 @@ class Sampler : public FairMQDevice
 
   private:
     int fMsgSize;
+    uint32_t fLinger;
     uint64_t fMaxIterations;
     uint64_t fNumIterations;
     FairMQUnmanagedRegionPtr fRegion;
-    std::atomic<uint64_t> fNumUnackedMsgs;
+    std::mutex fMtx;
+    uint64_t fNumUnackedMsgs;
 };
 
 } // namespace example_region

examples/region/Sink.cxx

@@ -30,11 +30,11 @@ void Sink::InitTask()
     // Get the fMaxIterations value from the command line options (via fConfig)
     fMaxIterations = fConfig->GetProperty<uint64_t>("max-iterations");
     fChannels.at("data").at(0).Transport()->SubscribeToRegionEvents([](FairMQRegionInfo info) {
-        LOG(warn) << ">>>" << info.event;
-        LOG(warn) << "id: " << info.id;
-        LOG(warn) << "ptr: " << info.ptr;
-        LOG(warn) << "size: " << info.size;
-        LOG(warn) << "flags: " << info.flags;
+        LOG(info) << "Region event: " << info.event
+                  << ", id: " << info.id
+                  << ", ptr: " << info.ptr
+                  << ", size: " << info.size
+                  << ", flags: " << info.flags;
     });
 }
 

examples/region/runSampler.cxx

@@ -15,6 +15,7 @@ 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)");
 }
 

examples/region/test-ex-region.sh.in

@@ -23,6 +23,7 @@ SAMPLER+=" --verbosity veryhigh"
 SAMPLER+=" --control static --color false"
 SAMPLER+=" --max-iterations 1"
 SAMPLER+=" --msg-size $msgSize"
+SAMPLER+=" --region-linger 500"
 SAMPLER+=" --channel-config name=data,type=push,method=bind,address=tcp://127.0.0.1:7777"
 @CMAKE_CURRENT_BINARY_DIR@/$SAMPLER &
 SAMPLER_PID=$!

fairmq/CMakeLists.txt

@@ -170,6 +170,7 @@ if(BUILD_FAIRMQ)
     PluginManager.h
     PluginServices.h
     runFairMQDevice.h
+    shmem/Monitor.h
   )
 
   set(FAIRMQ_PRIVATE_HEADER_FILES
@@ -234,6 +235,7 @@ if(BUILD_FAIRMQ)
     plugins/config/Config.cxx
     plugins/Control.cxx
     MemoryResources.cxx
+    shmem/Monitor.cxx
   )
 
   if(BUILD_OFI_TRANSPORT)
@@ -325,6 +327,7 @@ if(BUILD_FAIRMQ)
 
     PRIVATE # only libFairMQ links against private dependencies
     libzmq
+    PicoSHA2
     ${OFI_DEPS}
   )
   set_target_properties(${_target} PROPERTIES
@@ -377,6 +380,8 @@ if(BUILD_FAIRMQ)
     Boost::boost
     Boost::date_time
     Boost::program_options
+    FairLogger::FairLogger
+    PicoSHA2
   )
   target_include_directories(fairmq-shmmonitor PUBLIC
     $<BUILD_INTERFACE:${CMAKE_SOURCE_DIR}>

fairmq/FairMQChannel.cxx

@@ -671,7 +671,6 @@ void FairMQChannel::Init()
 bool FairMQChannel::ConnectEndpoint(const string& endpoint)
 {
     lock_guard<mutex> lock(fMtx);
-
     return fSocket->Connect(endpoint);
 }
 
@@ -683,6 +682,13 @@ bool FairMQChannel::BindEndpoint(string& endpoint)
     if (fSocket->Bind(endpoint)) {
         return true;
     } else {
+        // auto-bind only implemented for TCP
+        size_t protocolPos = endpoint.find(':');
+        string protocol = endpoint.substr(0, protocolPos);
+        if (protocol != "tcp") {
+            return false;
+        }
+
         if (fAutoBind) {
             // number of attempts when choosing a random port
             int numAttempts = 0;

fairmq/FairMQMessage.h

@@ -11,6 +11,7 @@
 
 #include <cstddef> // for size_t
 #include <memory> // unique_ptr
+#include <stdexcept>
 
 #include <fairmq/Transports.h>
 

fairmq/FairMQSocket.h

@@ -9,9 +9,10 @@
 #ifndef FAIRMQSOCKET_H_
 #define FAIRMQSOCKET_H_
 
+#include <memory>
+#include <stdexcept>
 #include <string>
 #include <vector>
-#include <memory>
 
 #include "FairMQMessage.h"
 class FairMQTransportFactory;
@@ -37,6 +38,10 @@ class FairMQSocket
     virtual void SetOption(const std::string& option, const void* value, size_t valueSize) = 0;
     virtual void GetOption(const std::string& option, void* value, size_t* valueSize) = 0;
 
+    /// If the backend supports it, fills the unsigned integer @a events with the ZMQ_EVENTS value
+    /// DISCLAIMER: this API is experimental and unsupported and might be dropped / refactored in
+    /// the future.
+    virtual void Events(uint32_t* events) = 0;
     virtual void SetLinger(const int value) = 0;
     virtual int GetLinger() const = 0;
     virtual void SetSndBufSize(const int value) = 0;

fairmq/FairMQUnmanagedRegion.h

@@ -10,6 +10,7 @@
 #define FAIRMQUNMANAGEDREGION_H_
 
 #include <cstddef> // size_t
+#include <cstdint> // uint32_t
 #include <memory> // std::unique_ptr
 #include <functional> // std::function
 #include <ostream> // std::ostream
@@ -72,6 +73,8 @@ class FairMQUnmanagedRegion
     virtual void* GetData() const = 0;
     virtual size_t GetSize() const = 0;
     virtual uint64_t GetId() const = 0;
+    virtual void SetLinger(uint32_t linger) = 0;
+    virtual uint32_t GetLinger() const = 0;
 
     FairMQTransportFactory* GetTransport() { return fTransport; }
     void SetTransport(FairMQTransportFactory* transport) { fTransport = transport; }

fairmq/MemoryResources.cxx

@@ -15,8 +15,7 @@
 #include <fairmq/FairMQTransportFactory.h>
 #include <fairmq/MemoryResources.h>
 
-void *fair::mq::ChannelResource::do_allocate(std::size_t bytes, std::size_t /*alignment*/)
+void *fair::mq::ChannelResource::do_allocate(std::size_t bytes, std::size_t alignment)
 {
-    return setMessage(factory->CreateMessage(bytes));
+    return setMessage(factory->CreateMessage(bytes, fair::mq::Alignment{alignment}));
 }
-

fairmq/ProgOptions.cxx

@@ -120,8 +120,10 @@ void ProgOptions::ParseAll(const int argc, char const* const* argv, bool allowUn
 
     po::command_line_parser parser(argc, argv);
 
+    parser.options(fAllOptions);
+
     if (allowUnregistered) {
-        parser.options(fAllOptions).allow_unregistered();
+        parser.allow_unregistered();
     }
 
     using namespace po::command_line_style;

fairmq/States.cxx

@@ -18,9 +18,10 @@ namespace fair
 namespace mq
 {
 
-array<string, 15> stateNames =
+array<string, 16> stateNames =
 {
     {
+        "UNDEFINED",
         "OK",
         "ERROR",
         "IDLE",
@@ -41,6 +42,7 @@ array<string, 15> stateNames =
 
 unordered_map<string, State> states =
 {
+    { "UNDEFINED",           State::Undefined },
     { "OK",                  State::Ok },
     { "ERROR",               State::Error },
     { "IDLE",                State::Idle },

fairmq/States.h

@@ -20,6 +20,7 @@ namespace mq
 
 enum class State : int
 {
+    Undefined = 0,
     Ok,
     Error,
     Idle,
@@ -39,7 +40,7 @@ enum class State : int
 
 enum class Transition : int
 {
-    Auto,
+    Auto = 0,
     InitDevice,
     CompleteInit,
     Bind,

fairmq/devices/FairMQBenchmarkSampler.h

@@ -17,6 +17,7 @@
 #include <chrono>
 #include <cstddef>   // size_t
 #include <cstdint>   // uint64_t
+#include <cstring>   // memset
 #include <string>
 
 /**
@@ -28,8 +29,10 @@ class FairMQBenchmarkSampler : public FairMQDevice
   public:
     FairMQBenchmarkSampler()
         : fMultipart(false)
+        , fMemSet(false)
         , fNumParts(1)
         , fMsgSize(10000)
+        , fMsgAlignment(0)
         , fMsgRate(0)
         , fNumIterations(0)
         , fMaxIterations(0)
@@ -39,8 +42,10 @@ class FairMQBenchmarkSampler : public FairMQDevice
     void InitTask() override
     {
         fMultipart = fConfig->GetProperty<bool>("multipart");
+        fMemSet = fConfig->GetProperty<bool>("memset");
         fNumParts = fConfig->GetProperty<size_t>("num-parts");
         fMsgSize = fConfig->GetProperty<size_t>("msg-size");
+        fMsgAlignment = fConfig->GetProperty<size_t>("msg-alignment");
         fMsgRate = fConfig->GetProperty<float>("msg-rate");
         fMaxIterations = fConfig->GetProperty<uint64_t>("max-iterations");
         fOutChannelName = fConfig->GetProperty<std::string>("out-channel");
@@ -51,8 +56,6 @@ class FairMQBenchmarkSampler : public FairMQDevice
         // store the channel reference to avoid traversing the map on every loop iteration
         FairMQChannel& dataOutChannel = fChannels.at(fOutChannelName).at(0);
 
-        FairMQMessagePtr baseMsg(dataOutChannel.NewMessage(fMsgSize));
-
         LOG(info) << "Starting the benchmark with message size of " << fMsgSize << " and " << fMaxIterations << " iterations.";
         auto tStart = std::chrono::high_resolution_clock::now();
 
@@ -63,7 +66,10 @@ class FairMQBenchmarkSampler : public FairMQDevice
                 FairMQParts parts;
 
                 for (size_t i = 0; i < fNumParts; ++i) {
-                    parts.AddPart(dataOutChannel.NewMessage(fMsgSize));
+                    parts.AddPart(dataOutChannel.NewMessage(fMsgSize, fair::mq::Alignment{fMsgAlignment}));
+                    if (fMemSet) {
+                        std::memset(parts.At(i)->GetData(), 0, parts.At(i)->GetSize());
+                    }
                 }
 
                 if (dataOutChannel.Send(parts) >= 0) {
@@ -75,7 +81,10 @@ class FairMQBenchmarkSampler : public FairMQDevice
                     ++fNumIterations;
                 }
             } else {
-                FairMQMessagePtr msg(dataOutChannel.NewMessage(fMsgSize));
+                FairMQMessagePtr msg(dataOutChannel.NewMessage(fMsgSize, fair::mq::Alignment{fMsgAlignment}));
+                if (fMemSet) {
+                    std::memset(msg->GetData(), 0, msg->GetSize());
+                }
 
                 if (dataOutChannel.Send(msg) >= 0) {
                     if (fMaxIterations > 0) {
@@ -101,8 +110,10 @@ class FairMQBenchmarkSampler : public FairMQDevice
 
   protected:
     bool fMultipart;
+    bool fMemSet;
     size_t fNumParts;
     size_t fMsgSize;
+    size_t fMsgAlignment;
     std::atomic<int> fMsgCounter;
     float fMsgRate;
     uint64_t fNumIterations;

fairmq/ofi/Socket.h

@@ -45,6 +45,7 @@ class Socket final : public fair::mq::Socket
 
     auto GetId() const -> std::string override { return fId; }
 
+    auto Events(uint32_t *events) -> void override { *events = 0; }
     auto Bind(const std::string& address) -> bool override;
     auto Connect(const std::string& address) -> bool override;
 

fairmq/plugins/DDS/runDDSCommandUI.cxx

@@ -65,7 +65,7 @@ void handleCommand(const string& command, const string& path, unsigned int timeo
             cout << d.taskId << " : " << d.state << endl;
         }
     } else if (command == "o") {
-        cout << "> dumping config of the devices (" << path << ")" << endl;
+        cout << "> dumping config of " << (path == "" ? "all" : path) << endl;
         // TODO: extend this regex to return all properties, once command size limitation is removed.
         auto const result = topo.GetProperties("^(session|id)$", path, std::chrono::milliseconds(timeout));
         for (const auto& d : result.second.devices) {
@@ -79,39 +79,39 @@ void handleCommand(const string& command, const string& path, unsigned int timeo
             return;
         }
         const DeviceProperties props{{pKey, pVal}};
-        cout << "> sending property (" << path << ")" << endl;
+        cout << "> setting properties --> " << (path == "" ? "all" : path) << endl;
         topo.SetProperties(props, path);
         // give dds time to complete request
         this_thread::sleep_for(chrono::milliseconds(100));
     } else if (command == "i") {
-        cout << "> init devices (" << path << ")" << endl;
+        cout << "> initiating InitDevice transition --> " << (path == "" ? "all" : path) << endl;
         topo.ChangeState(TopologyTransition::InitDevice, path, std::chrono::milliseconds(timeout));
     } else if (command == "k") {
-        cout << "> complete init (" << path << ")" << endl;
+        cout << "> initiating CompleteInit transition --> " << (path == "" ? "all" : path) << endl;
         topo.ChangeState(TopologyTransition::CompleteInit, path, std::chrono::milliseconds(timeout));
     } else if (command == "b") {
-        cout << "> bind devices (" << path << ")" << endl;
+        cout << "> initiating Bind transition --> " << (path == "" ? "all" : path) << endl;
         topo.ChangeState(TopologyTransition::Bind, path, std::chrono::milliseconds(timeout));
     } else if (command == "x") {
-        cout << "> connect devices (" << path << ")" << endl;
+        cout << "> initiating Connect transition --> " << (path == "" ? "all" : path) << endl;
         topo.ChangeState(TopologyTransition::Connect, path, std::chrono::milliseconds(timeout));
     } else if (command == "j") {
-        cout << "> init tasks (" << path << ")" << endl;
+        cout << "> initiating InitTask transition --> " << (path == "" ? "all" : path) << endl;
         topo.ChangeState(TopologyTransition::InitTask, path, std::chrono::milliseconds(timeout));
     } else if (command == "r") {
-        cout << "> run tasks (" << path << ")" << endl;
+        cout << "> initiating Run transition --> " << (path == "" ? "all" : path) << endl;
         topo.ChangeState(TopologyTransition::Run, path, std::chrono::milliseconds(timeout));
     } else if (command == "s") {
-        cout << "> stop devices (" << path << ")" << endl;
+        cout << "> initiating Stop transition --> " << (path == "" ? "all" : path) << endl;
         topo.ChangeState(TopologyTransition::Stop, path, std::chrono::milliseconds(timeout));
     } else if (command == "t") {
-        cout << "> reset tasks (" << path << ")" << endl;
+        cout << "> initiating ResetTask transition --> " << (path == "" ? "all" : path) << endl;
         topo.ChangeState(TopologyTransition::ResetTask, path, std::chrono::milliseconds(timeout));
     } else if (command == "d") {
-        cout << "> reset devices (" << path << ")" << endl;
+        cout << "> initiating ResetDevice transition --> " << (path == "" ? "all" : path) << endl;
         topo.ChangeState(TopologyTransition::ResetDevice, path, std::chrono::milliseconds(timeout));
     } else if (command == "q") {
-        cout << "> end (" << path << ")" << endl;
+        cout << "> initiating End transition --> " << (path == "" ? "all" : path) << endl;
         topo.ChangeState(TopologyTransition::End, path, std::chrono::milliseconds(timeout));
     } else if (command == "h") {
         cout << "> help" << endl;
@@ -206,6 +206,7 @@ try {
             sendCommand(command, path, timeout, topo, pKey, pVal);
         }
         size_t pos = targetState.find("->");
+        cout << "> waiting for " << (path == "" ? "all" : path) << " to reach " << targetState << endl;
         if (pos == string::npos) {
             /* auto ec =  */topo.WaitForState(GetState(targetState), path, std::chrono::milliseconds(timeout));
             // cout << "WaitForState(" << targetState << ") result: " << ec.message() << endl;

fairmq/plugins/config/Config.cxx

@@ -68,7 +68,9 @@ Plugin::ProgOptions ConfigPluginProgramOptions()
         ("init-timeout",        po::value<int           >()->default_value(120),        "Timeout for the initialization in seconds (when expecting dynamic initialization).")
         ("max-run-time",        po::value<uint64_t      >()->default_value(0),          "Maximum runtime for the Running state handler, after which state will change to Ready (in seconds, 0 for no limit).")
         ("print-channels",      po::value<bool          >()->implicit_value(true),      "Print registered channel endpoints in a machine-readable format (<channel name>:<min num subchannels>:<max num subchannels>)")
-        ("shm-segment-size",    po::value<size_t        >()->default_value(2000000000), "Shared memory: size of the shared memory segment (in bytes).")
+        ("shm-segment-size",    po::value<size_t        >()->default_value(2ULL << 30), "Shared memory: size of the shared memory segment (in bytes).")
+        ("shm-mlock-segment",   po::value<bool          >()->default_value(false),      "Shared memory: mlock the shared memory segment after initialization.")
+        ("shm-zero-segment",    po::value<bool          >()->default_value(false),      "Shared memory: zero the shared memory segment memory after initialization.")
         ("shm-throw-bad-alloc", po::value<bool          >()->default_value(true),       "Throw a fair::mq::MessageBadAlloc if cannot allocate a message (retry if false).")
         ("shm-monitor",         po::value<bool          >()->default_value(true),       "Shared memory: run monitor daemon.")
         ("ofi-size-hint",       po::value<size_t        >()->default_value(0),          "EXPERIMENTAL: OFI size hint for the allocator.")

fairmq/run/runBenchmarkSampler.cxx

@@ -15,10 +15,11 @@ void addCustomOptions(bpo::options_description& options)
 {
     options.add_options()
         ("out-channel", bpo::value<std::string>()->default_value("data"), "Name of the output channel")
-        ("same-msg", bpo::value<bool>()->default_value(false), "Re-send the same message, or recreate for each iteration")
         ("multipart", bpo::value<bool>()->default_value(false), "Handle multipart payloads")
+        ("memset", bpo::value<bool>()->default_value(false), "Memset allocated buffers to 0")
         ("num-parts", bpo::value<size_t>()->default_value(1), "Number of parts to send. 1 will send single messages, not parts")
         ("msg-size", bpo::value<size_t>()->default_value(1000000), "Message size in bytes")
+        ("msg-alignment", bpo::value<size_t>()->default_value(0), "Message alignment")
         ("max-iterations", bpo::value<uint64_t>()->default_value(0), "Number of run iterations (0 - infinite)")
         ("msg-rate", bpo::value<float>()->default_value(0), "Msg rate limit in maximum number of messages per second");
 }

fairmq/sdk/Error.h

@@ -25,11 +25,6 @@ struct RuntimeError : ::std::runtime_error
     {}
 };
 
-struct MixedStateError : RuntimeError
-{
-    using RuntimeError::RuntimeError;
-};
-
 } /* namespace sdk */
 
 enum class ErrorCode

fairmq/sdk/Topology.h

@@ -70,6 +70,65 @@ const std::map<DeviceTransition, DeviceState> expectedState =
     { DeviceTransition::End,          DeviceState::Exiting }
 };
 
+// mirrors DeviceState, but adds a "Mixed" state that represents a topology where devices are currently not in the same state.
+enum class AggregatedTopologyState : int
+{
+    Undefined = static_cast<int>(fair::mq::State::Undefined),
+    Ok = static_cast<int>(fair::mq::State::Ok),
+    Error = static_cast<int>(fair::mq::State::Error),
+    Idle = static_cast<int>(fair::mq::State::Idle),
+    InitializingDevice = static_cast<int>(fair::mq::State::InitializingDevice),
+    Initialized = static_cast<int>(fair::mq::State::Initialized),
+    Binding = static_cast<int>(fair::mq::State::Binding),
+    Bound = static_cast<int>(fair::mq::State::Bound),
+    Connecting = static_cast<int>(fair::mq::State::Connecting),
+    DeviceReady = static_cast<int>(fair::mq::State::DeviceReady),
+    InitializingTask = static_cast<int>(fair::mq::State::InitializingTask),
+    Ready = static_cast<int>(fair::mq::State::Ready),
+    Running = static_cast<int>(fair::mq::State::Running),
+    ResettingTask = static_cast<int>(fair::mq::State::ResettingTask),
+    ResettingDevice = static_cast<int>(fair::mq::State::ResettingDevice),
+    Exiting = static_cast<int>(fair::mq::State::Exiting),
+    Mixed
+};
+
+inline auto operator==(DeviceState lhs, AggregatedTopologyState rhs) -> bool
+{
+    return static_cast<int>(lhs) == static_cast<int>(rhs);
+}
+
+inline auto operator==(AggregatedTopologyState lhs, DeviceState rhs) -> bool
+{
+    return static_cast<int>(lhs) == static_cast<int>(rhs);
+}
+
+inline std::ostream& operator<<(std::ostream& os, const AggregatedTopologyState& state)
+{
+    if (state == AggregatedTopologyState::Mixed) {
+        return os << "MIXED";
+    } else {
+        return os << static_cast<DeviceState>(state);
+    }
+}
+
+inline std::string GetAggregatedTopologyStateName(AggregatedTopologyState s)
+{
+    if (s == AggregatedTopologyState::Mixed) {
+        return "MIXED";
+    } else {
+        return GetStateName(static_cast<State>(s));
+    }
+}
+
+inline AggregatedTopologyState GetAggregatedTopologyState(const std::string& state)
+{
+    if (state == "MIXED") {
+        return AggregatedTopologyState::Mixed;
+    } else {
+        return static_cast<AggregatedTopologyState>(GetState(state));
+    }
+}
+
 struct DeviceStatus
 {
     bool subscribed_to_state_changes;
@@ -100,22 +159,22 @@ using TopologyStateByTask = std::unordered_map<DDSTask::Id, DeviceStatus>;
 using TopologyStateByCollection = std::unordered_map<DDSCollection::Id, std::vector<DeviceStatus>>;
 using TopologyTransition = fair::mq::Transition;
 
-inline DeviceState AggregateState(const TopologyState& topologyState)
+inline AggregatedTopologyState AggregateState(const TopologyState& topologyState)
 {
     DeviceState first = topologyState.begin()->state;
 
     if (std::all_of(topologyState.cbegin(), topologyState.cend(), [&](TopologyState::value_type i) {
             return i.state == first;
         })) {
-        return first;
+        return static_cast<AggregatedTopologyState>(first);
     }
 
-    throw MixedStateError("State is not uniform");
+    return AggregatedTopologyState::Mixed;
 }
 
 inline bool StateEqualsTo(const TopologyState& topologyState, DeviceState state)
 {
-    return AggregateState(topologyState) == state;
+    return AggregateState(topologyState) == static_cast<AggregatedTopologyState>(state);
 }
 
 inline TopologyStateByCollection GroupByCollectionId(const TopologyState& topologyState)
@@ -439,6 +498,9 @@ class BasicTopology : public AsioBase<Executor, Allocator>
                     }
                 });
             }
+            if (fTasks.empty()) {
+                FAIR_LOG(warn) << "ChangeState initiated on an empty set of tasks, check the path argument.";
+            }
         }
         ChangeStateOp() = delete;
         ChangeStateOp(const ChangeStateOp&) = delete;
@@ -741,6 +803,9 @@ class BasicTopology : public AsioBase<Executor, Allocator>
                     }
                 });
             }
+            if (fTasks.empty()) {
+                FAIR_LOG(warn) << "WaitForState initiated on an empty set of tasks, check the path argument.";
+            }
         }
         WaitForStateOp() = delete;
         WaitForStateOp(const WaitForStateOp&) = delete;
@@ -754,9 +819,8 @@ class BasicTopology : public AsioBase<Executor, Allocator>
         {
             fCount = std::count_if(stateIndex.cbegin(), stateIndex.cend(), [=](const auto& s) {
                 if (ContainsTask(stateData.at(s.second).taskId)) {
-                    return stateData.at(s.second).state == fTargetCurrentState
-                           &&
-                           (stateData.at(s.second).lastState == fTargetLastState || fTargetLastState == DeviceState::Ok);
+                    return stateData.at(s.second).state == fTargetCurrentState &&
+                          (stateData.at(s.second).lastState == fTargetLastState || fTargetLastState == DeviceState::Undefined);
                 } else {
                     return false;
                 }
@@ -768,8 +832,7 @@ class BasicTopology : public AsioBase<Executor, Allocator>
         {
             if (!fOp.IsCompleted() && ContainsTask(taskId)) {
                 if (currentState == fTargetCurrentState &&
-                    (lastState == fTargetLastState ||
-                     fTargetLastState == DeviceState::Ok)) {
+                    (lastState == fTargetLastState || fTargetLastState == DeviceState::Undefined)) {
                     ++fCount;
                 }
                 TryCompletion();
@@ -873,7 +936,7 @@ class BasicTopology : public AsioBase<Executor, Allocator>
     template<typename CompletionToken>
     auto AsyncWaitForState(const DeviceState targetCurrentState, CompletionToken&& token)
     {
-        return AsyncWaitForState(DeviceState::Ok, targetCurrentState, "", Duration(0), std::move(token));
+        return AsyncWaitForState(DeviceState::Undefined, targetCurrentState, "", Duration(0), std::move(token));
     }
 
     /// @brief Wait for selected FairMQ devices to reach given last & current state in this topology
@@ -903,7 +966,7 @@ class BasicTopology : public AsioBase<Executor, Allocator>
     auto WaitForState(const DeviceState targetCurrentState, const std::string& path = "", Duration timeout = Duration(0))
         -> std::error_code
     {
-        return WaitForState(DeviceState::Ok, targetCurrentState, path, timeout);
+        return WaitForState(DeviceState::Undefined, targetCurrentState, path, timeout);
     }
 
     using GetPropertiesCompletionSignature = void(std::error_code, GetPropertiesResult);
@@ -938,6 +1001,9 @@ class BasicTopology : public AsioBase<Executor, Allocator>
                     }
                 });
             }
+            if (expectedCount == 0) {
+                FAIR_LOG(warn) << "GetProperties initiated on an empty set of tasks, check the path argument.";
+            }
             // FAIR_LOG(debug) << "GetProperties " << fId << " with expected count of " << fExpectedCount << " started.";
         }
         GetPropertiesOp() = delete;
@@ -1093,6 +1159,9 @@ class BasicTopology : public AsioBase<Executor, Allocator>
                     }
                 });
             }
+            if (expectedCount == 0) {
+                FAIR_LOG(warn) << "SetProperties initiated on an empty set of tasks, check the path argument.";
+            }
             // FAIR_LOG(debug) << "SetProperties " << fId << " with expected count of " << fExpectedCount << " started.";
         }
         SetPropertiesOp() = delete;
@@ -1242,7 +1311,7 @@ class BasicTopology : public AsioBase<Executor, Allocator>
         int index = 0;
 
         for (const auto& task : fDDSTopo.GetTasks()) {
-            fStateData.push_back(DeviceStatus{false, DeviceState::Ok, DeviceState::Ok, task.GetId(), task.GetCollectionId()});
+            fStateData.push_back(DeviceStatus{false, DeviceState::Undefined, DeviceState::Undefined, task.GetId(), task.GetCollectionId()});
             fStateIndex.emplace(task.GetId(), index);
             index++;
         }

fairmq/sdk/commands/Commands.cxx

@@ -70,9 +70,10 @@ array<string, 17> typeNames =
     }
 };
 
-array<fair::mq::State, 15> fbStateToMQState =
+array<fair::mq::State, 16> fbStateToMQState =
 {
     {
+        fair::mq::State::Undefined,
         fair::mq::State::Ok,
         fair::mq::State::Error,
         fair::mq::State::Idle,
@@ -91,9 +92,10 @@ array<fair::mq::State, 15> fbStateToMQState =
     }
 };
 
-array<sdk::cmd::FBState, 15> mqStateToFBState =
+array<sdk::cmd::FBState, 16> mqStateToFBState =
 {
     {
+        sdk::cmd::FBState_Undefined,
         sdk::cmd::FBState_Ok,
         sdk::cmd::FBState_Error,
         sdk::cmd::FBState_Idle,

fairmq/sdk/commands/CommandsFormat.fbs

@@ -6,6 +6,7 @@ enum FBResult:byte {
 }
 
 enum FBState:byte {
+    Undefined,
     Ok,
     Error,
     Idle,

fairmq/shmem/Common.h

@@ -8,9 +8,10 @@
 #ifndef FAIR_MQ_SHMEM_COMMON_H_
 #define FAIR_MQ_SHMEM_COMMON_H_
 
+#include <picosha2.h>
+
 #include <atomic>
 #include <string>
-#include <unordered_map>
 
 #include <boost/interprocess/managed_shared_memory.hpp>
 #include <boost/interprocess/allocators/allocator.hpp>
@@ -110,9 +111,12 @@ struct RegionBlock
 // a hash of user id + session id, truncated to 8 characters (to accommodate for name size limit on some systems (MacOS)).
 inline std::string buildShmIdFromSessionIdAndUserId(const std::string& sessionId)
 {
-    boost::hash<std::string> stringHash;
-    std::string shmId(std::to_string(stringHash(std::string((std::to_string(geteuid()) + sessionId)))));
-    shmId.resize(8, '_');
+    std::string seed((std::to_string(geteuid()) + sessionId));
+    // generate a 8-digit hex value out of sha256 hash
+    std::vector<unsigned char> hash(4);
+    picosha2::hash256(seed.begin(), seed.end(), hash.begin(), hash.end());
+    std::string shmId = picosha2::bytes_to_hex_string(hash.begin(), hash.end());
+
     return shmId;
 }
 

fairmq/shmem/Manager.h

@@ -17,21 +17,27 @@
 
 #include "Common.h"
 #include "Region.h"
+#include "Monitor.h"
 
 #include <FairMQLogger.h>
 #include <FairMQMessage.h>
+#include <fairmq/ProgOptions.h>
 #include <fairmq/tools/CppSTL.h>
 #include <fairmq/tools/Strings.h>
 
-#include <boost/process.hpp>
-#include <boost/filesystem.hpp>
 #include <boost/date_time/posix_time/posix_time.hpp>
+#include <boost/filesystem.hpp>
+#include <boost/interprocess/indexes/null_index.hpp>
+#include <boost/interprocess/ipc/message_queue.hpp>
 #include <boost/interprocess/managed_shared_memory.hpp>
 #include <boost/interprocess/sync/named_condition.hpp>
 #include <boost/interprocess/sync/named_mutex.hpp>
-#include <boost/interprocess/ipc/message_queue.hpp>
+#include <boost/interprocess/mem_algo/simple_seq_fit.hpp>
+#include <boost/process.hpp>
 
 #include <cstdlib> // getenv
+#include <condition_variable>
+#include <mutex>
 #include <set>
 #include <stdexcept>
 #include <string>
@@ -40,6 +46,8 @@
 #include <utility> // pair
 #include <vector>
 
+#include <sys/mman.h> // mlock
+
 namespace fair
 {
 namespace mq
@@ -49,16 +57,21 @@ namespace shmem
 
 struct SharedMemoryError : std::runtime_error { using std::runtime_error::runtime_error; };
 
+using SimpleSeqFitSegment = boost::interprocess::basic_managed_shared_memory<char,
+    boost::interprocess::simple_seq_fit<boost::interprocess::mutex_family>,
+    boost::interprocess::iset_index>;
+using RBTreeBestFitSegment = boost::interprocess::basic_managed_shared_memory<char,
+    boost::interprocess::rbtree_best_fit<boost::interprocess::mutex_family>,
+    boost::interprocess::iset_index>;
+
 class Manager
 {
   public:
-    Manager(std::string id, std::string deviceId, size_t size, bool throwOnBadAlloc)
-        : fShmId(std::move(id))
+    Manager(std::string shmId, std::string deviceId, size_t size, const ProgOptions* config)
+        : fShmId(std::move(shmId))
         , fDeviceId(std::move(deviceId))
-        , fSegmentName("fmq_" + fShmId + "_main")
-        , fManagementSegmentName("fmq_" + fShmId + "_mng")
-        , fSegment(boost::interprocess::open_or_create, fSegmentName.c_str(), size)
-        , fManagementSegment(boost::interprocess::open_or_create, fManagementSegmentName.c_str(), 655360)
+        , fSegment(boost::interprocess::open_or_create, std::string("fmq_" + fShmId + "_main").c_str(), size)
+        , fManagementSegment(boost::interprocess::open_or_create, std::string("fmq_" + fShmId + "_mng").c_str(), 655360)
         , fShmVoidAlloc(fManagementSegment.get_segment_manager())
         , fShmMtx(boost::interprocess::open_or_create, std::string("fmq_" + fShmId + "_mtx").c_str())
         , fRegionEventsCV(boost::interprocess::open_or_create, std::string("fmq_" + fShmId + "_cv").c_str())
@@ -69,10 +82,39 @@ class Manager
         , fMsgCounter(0)
         , fHeartbeatThread()
         , fSendHeartbeats(true)
-        , fThrowOnBadAlloc(throwOnBadAlloc)
+        , fThrowOnBadAlloc(true)
     {
         using namespace boost::interprocess;
-        LOG(debug) << "created/opened shared memory segment '" << "fmq_" << fShmId << "_main" << "' of " << size << " bytes. Available are " << fSegment.get_free_memory() << " bytes.";
+
+        bool mlockSegment = false;
+        bool zeroSegment = false;
+        bool autolaunchMonitor = false;
+        if (config) {
+            mlockSegment = config->GetProperty<bool>("shm-mlock-segment", mlockSegment);
+            zeroSegment = config->GetProperty<bool>("shm-zero-segment", zeroSegment);
+            autolaunchMonitor = config->GetProperty<bool>("shm-monitor", autolaunchMonitor);
+            fThrowOnBadAlloc = config->GetProperty<bool>("shm-throw-bad-alloc", fThrowOnBadAlloc);
+        } else {
+            LOG(debug) << "ProgOptions not available! Using defaults.";
+        }
+
+        if (autolaunchMonitor) {
+            StartMonitor(fShmId);
+        }
+
+        LOG(debug) << "created/opened shared memory segment '" << "fmq_" << fShmId << "_main" << "' of " << fSegment.get_size() << " bytes. Available are " << fSegment.get_free_memory() << " bytes.";
+        if (mlockSegment) {
+            LOG(debug) << "Locking the managed segment memory pages...";
+            if (mlock(fSegment.get_address(), fSegment.get_size()) == -1) {
+                LOG(error) << "Could not lock the managed segment memory. Code: " << errno << ", reason: " << strerror(errno);
+            }
+            LOG(debug) << "Successfully locked the managed segment memory pages.";
+        }
+        if (zeroSegment) {
+            LOG(debug) << "Zeroing the managed segment free memory...";
+            fSegment.zero_free_memory();
+            LOG(debug) << "Successfully zeroed the managed segment free memory.";
+        }
 
         fRegionInfos = fManagementSegment.find_or_construct<Uint64RegionInfoMap>(unique_instance)(fShmVoidAlloc);
         // store info about the managed segment as region with id 0
@@ -100,40 +142,7 @@ class Manager
     Manager(const Manager&) = delete;
     Manager operator=(const Manager&) = delete;
 
-    ~Manager()
-    {
-        using namespace boost::interprocess;
-        bool lastRemoved = false;
-
-        UnsubscribeFromRegionEvents();
-
-        fSendHeartbeats = false;
-        fHeartbeatThread.join();
-
-        try {
-            boost::interprocess::scoped_lock<named_mutex> lock(fShmMtx);
-
-            (fDeviceCounter->fCount)--;
-
-            if (fDeviceCounter->fCount == 0) {
-                LOG(debug) << "last segment user, removing segment.";
-
-                RemoveSegments();
-                lastRemoved = true;
-            } else {
-                LOG(debug) << "other segment users present (" << fDeviceCounter->fCount << "), not removing it.";
-            }
-        } catch(interprocess_exception& e) {
-            LOG(error) << "error while acquiring lock in Manager destructor: " << e.what();
-        }
-
-        if (lastRemoved) {
-            named_mutex::remove(std::string("fmq_" + fShmId + "_mtx").c_str());
-            named_condition::remove(std::string("fmq_" + fShmId + "_cv").c_str());
-        }
-    }
-
-    boost::interprocess::managed_shared_memory& Segment() { return fSegment; }
+    RBTreeBestFitSegment& Segment() { return fSegment; }
     boost::interprocess::managed_shared_memory& ManagementSegment() { return fManagementSegment; }
 
     static void StartMonitor(const std::string& id)
@@ -265,8 +274,12 @@ class Manager
 
                 auto r = fRegions.emplace(id, tools::make_unique<Region>(fShmId, id, 0, true, nullptr, nullptr, path, flags));
                 return r.first->second.get();
+            } catch (std::out_of_range& oor) {
+                LOG(error) << "Could not get remote region with id '" << id << "'. Does the region creator run with the same session id?";
+                LOG(error) << oor.what();
+                return nullptr;
             } catch (boost::interprocess::interprocess_exception& e) {
-                LOG(warn) << "Could not get remote region for id: " << id;
+                LOG(warn) << "Could not get remote region for id '" << id << "'";
                 return nullptr;
             }
         }
@@ -274,9 +287,9 @@ class Manager
 
     void RemoveRegion(const uint64_t id)
     {
+        fRegions.erase(id);
         {
             boost::interprocess::scoped_lock<boost::interprocess::named_mutex> lock(fShmMtx);
-            fRegions.erase(id);
             fRegionInfos->at(id).fDestroyed = true;
         }
         fRegionEventsCV.notify_all();
@@ -378,52 +391,72 @@ class Manager
         }
     }
 
-    void IncrementMsgCounter() { ++fMsgCounter; }
-    void DecrementMsgCounter() { --fMsgCounter; }
-
-    void RemoveSegments()
-    {
-        using namespace boost::interprocess;
-        if (shared_memory_object::remove(fSegmentName.c_str())) {
-            LOG(debug) << "successfully removed '" << fSegmentName << "' segment after the device has stopped.";
-        } else {
-            LOG(debug) << "did not remove " << fSegmentName << " segment after the device stopped. Already removed?";
-        }
-
-        if (shared_memory_object::remove(fManagementSegmentName.c_str())) {
-            LOG(debug) << "successfully removed '" << fManagementSegmentName << "' segment after the device has stopped.";
-        } else {
-            LOG(debug) << "did not remove '" << fManagementSegmentName << "' segment after the device stopped. Already removed?";
-        }
-    }
+    void IncrementMsgCounter() { fMsgCounter.fetch_add(1, std::memory_order_relaxed); }
+    void DecrementMsgCounter() { fMsgCounter.fetch_sub(1, std::memory_order_relaxed); }
 
     void SendHeartbeats()
     {
         std::string controlQueueName("fmq_" + fShmId + "_cq");
+        std::unique_lock<std::mutex> lock(fHeartbeatsMtx);
         while (fSendHeartbeats) {
             try {
                 boost::interprocess::message_queue mq(boost::interprocess::open_only, controlQueueName.c_str());
                 boost::posix_time::ptime sndTill = boost::posix_time::microsec_clock::universal_time() + boost::posix_time::milliseconds(100);
                 if (mq.timed_send(fDeviceId.c_str(), fDeviceId.size(), 0, sndTill)) {
-                    std::this_thread::sleep_for(std::chrono::milliseconds(100));
+                    fHeartbeatsCV.wait_for(lock, std::chrono::milliseconds(100), [&]() { return !fSendHeartbeats; });
                 } else {
                     LOG(debug) << "control queue timeout";
                 }
             } catch (boost::interprocess::interprocess_exception& ie) {
-                std::this_thread::sleep_for(std::chrono::milliseconds(500));
-                // LOG(warn) << "no " << controlQueueName << " found";
+                fHeartbeatsCV.wait_for(lock, std::chrono::milliseconds(500), [&]() { return !fSendHeartbeats; });
+                // LOG(debug) << "no " << controlQueueName << " found";
             }
         }
     }
 
     bool ThrowingOnBadAlloc() const { return fThrowOnBadAlloc; }
 
+    ~Manager()
+    {
+        using namespace boost::interprocess;
+        bool lastRemoved = false;
+
+        UnsubscribeFromRegionEvents();
+
+        {
+            std::unique_lock<std::mutex> lock(fHeartbeatsMtx);
+            fSendHeartbeats = false;
+        }
+        fHeartbeatsCV.notify_one();
+        if (fHeartbeatThread.joinable()) {
+            fHeartbeatThread.join();
+        }
+
+        try {
+            boost::interprocess::scoped_lock<named_mutex> lock(fShmMtx);
+
+            (fDeviceCounter->fCount)--;
+
+            if (fDeviceCounter->fCount == 0) {
+                LOG(debug) << "Last segment user, removing segment.";
+                lastRemoved = true;
+            } else {
+                LOG(debug) << "Other segment users present (" << fDeviceCounter->fCount << "), skipping removal.";
+            }
+        } catch (interprocess_exception& e) {
+            LOG(error) << "Manager could not acquire lock: " << e.what();
+        }
+
+        if (lastRemoved) {
+            Monitor::Cleanup(ShmId{fShmId});
+        }
+    }
+
   private:
     std::string fShmId;
     std::string fDeviceId;
-    std::string fSegmentName;
-    std::string fManagementSegmentName;
-    boost::interprocess::managed_shared_memory fSegment;
+    // boost::interprocess::managed_shared_memory fSegment;
+    RBTreeBestFitSegment fSegment;
     boost::interprocess::managed_shared_memory fManagementSegment;
     VoidAlloc fShmVoidAlloc;
     boost::interprocess::named_mutex fShmMtx;
@@ -442,7 +475,10 @@ class Manager
     std::atomic<int32_t> fMsgCounter; // TODO: find a better lifetime solution instead of the counter
 
     std::thread fHeartbeatThread;
-    std::atomic<bool> fSendHeartbeats;
+    bool fSendHeartbeats;
+    std::mutex fHeartbeatsMtx;
+    std::condition_variable fHeartbeatsCV;
+
     bool fThrowOnBadAlloc;
 };
 

fairmq/shmem/Message.h

@@ -109,7 +109,7 @@ class Message final : public fair::mq::Message
         , fRegionPtr(nullptr)
         , fLocalPtr(static_cast<char*>(data))
     {
-        if (reinterpret_cast<const char*>(data) >= reinterpret_cast<const char*>(region->GetData()) ||
+        if (reinterpret_cast<const char*>(data) >= reinterpret_cast<const char*>(region->GetData()) &&
             reinterpret_cast<const char*>(data) <= reinterpret_cast<const char*>(region->GetData()) + region->GetSize()) {
             fMeta.fHandle = (boost::interprocess::managed_shared_memory::handle_t)(reinterpret_cast<const char*>(data) - reinterpret_cast<const char*>(region->GetData()));
         } else {
@@ -244,13 +244,17 @@ class Message final : public fair::mq::Message
 
     bool InitializeChunk(const size_t size, size_t alignment = 0)
     {
-        tools::RateLimiter rateLimiter(20);
+        // tools::RateLimiter rateLimiter(20);
 
         while (fMeta.fHandle < 0) {
             try {
                 // boost::interprocess::managed_shared_memory::size_type actualSize = size;
                 // char* hint = 0; // unused for boost::interprocess::allocate_new
                 // fLocalPtr = fManager.Segment().allocation_command<char>(boost::interprocess::allocate_new, size, actualSize, hint);
+                size_t segmentSize = fManager.Segment().get_size();
+                if (size > segmentSize) {
+                    throw MessageBadAlloc(tools::ToString("Requested message size (", size, ") exceeds segment size (", segmentSize, ")"));
+                }
                 if (alignment == 0) {
                     fLocalPtr = reinterpret_cast<char*>(fManager.Segment().allocate(size));
                 } else {
@@ -259,9 +263,10 @@ class Message final : public fair::mq::Message
             } catch (boost::interprocess::bad_alloc& ba) {
                 // LOG(warn) << "Shared memory full...";
                 if (fManager.ThrowingOnBadAlloc()) {
-                    throw MessageBadAlloc(tools::ToString("shmem: could not create a message of size ", size, ", alignment: ", (alignment != 0) ? std::to_string(alignment) : "default"));
+                    throw MessageBadAlloc(tools::ToString("shmem: could not create a message of size ", size, ", alignment: ", (alignment != 0) ? std::to_string(alignment) : "default", ", free memory: ", fManager.Segment().get_free_memory()));
                 }
-                rateLimiter.maybe_sleep();
+                // rateLimiter.maybe_sleep();
+                std::this_thread::sleep_for(std::chrono::milliseconds(50));
                 if (fManager.Interrupted()) {
                     return false;
                 } else {

fairmq/shmem/Monitor.cxx

@@ -10,6 +10,7 @@
 #include "Common.h"
 
 #include <fairmq/Tools.h>
+#include <fairlogger/Logger.h>
 
 #include <boost/interprocess/managed_shared_memory.hpp>
 #include <boost/interprocess/file_mapping.hpp>
@@ -22,6 +23,10 @@
 #include <csignal>
 #include <iostream>
 #include <iomanip>
+#include <chrono>
+#include <ctime>
+#include <time.h>
+#include <iomanip>
 
 #include <termios.h>
 #include <poll.h>
@@ -48,7 +53,7 @@ void signalHandler(int signal)
     gSignalStatus = signal;
 }
 
-Monitor::Monitor(const string& shmId, bool selfDestruct, bool interactive, bool viewOnly, unsigned int timeoutInMS, bool runAsDaemon, bool cleanOnExit)
+Monitor::Monitor(const string& shmId, bool selfDestruct, bool interactive, bool viewOnly, unsigned int timeoutInMS, unsigned int intervalInMS, bool runAsDaemon, bool cleanOnExit)
     : fSelfDestruct(selfDestruct)
     , fInteractive(interactive)
     , fViewOnly(viewOnly)
@@ -56,6 +61,7 @@ Monitor::Monitor(const string& shmId, bool selfDestruct, bool interactive, bool
     , fSeenOnce(false)
     , fCleanOnExit(cleanOnExit)
     , fTimeoutInMS(timeoutInMS)
+    , fIntervalInMS(intervalInMS)
     , fShmId(shmId)
     , fSegmentName("fmq_" + fShmId + "_main")
     , fManagementSegmentName("fmq_" + fShmId + "_mng")
@@ -64,7 +70,6 @@ Monitor::Monitor(const string& shmId, bool selfDestruct, bool interactive, bool
     , fHeartbeatTriggered(false)
     , fLastHeartbeat(chrono::high_resolution_clock::now())
     , fSignalThread()
-    , fManagementSegment(bipc::open_or_create, fManagementSegmentName.c_str(), 65536)
     , fDeviceHeartbeats()
 {
     if (!fViewOnly) {
@@ -75,6 +80,10 @@ Monitor::Monitor(const string& shmId, bool selfDestruct, bool interactive, bool
             throw DaemonPresent(tools::ToString("fairmq-shmmonitor for shared memory id ", fShmId, " already started or not properly exited."));
         }
     }
+
+    Logger::SetConsoleColor(false);
+    Logger::DefineVerbosity(Verbosity::user1, VerbositySpec::Make(VerbositySpec::Info::timestamp_us));
+    Logger::SetVerbosity(Verbosity::verylow);
 }
 
 void Monitor::CatchSignals()
@@ -111,7 +120,7 @@ void Monitor::Run()
         Interactive();
     } else {
         while (!fTerminating) {
-            this_thread::sleep_for(chrono::milliseconds(100));
+            this_thread::sleep_for(chrono::milliseconds(fIntervalInMS));
             CheckSegment();
         }
     }
@@ -184,7 +193,7 @@ void Monitor::Interactive()
     PrintHeader();
 
     while (!fTerminating) {
-        if (poll(cinfd, 1, 100)) {
+        if (poll(cinfd, 1, fIntervalInMS)) {
             if (fTerminating || gSignalStatus != 0) {
                 break;
             }
@@ -203,7 +212,7 @@ void Monitor::Interactive()
                 case 'x':
                     cout << "\n[x] --> closing shared memory:" << endl;
                     if (!fViewOnly) {
-                        Cleanup(fShmId);
+                        Cleanup(ShmId{fShmId});
                     } else {
                         cout << "cannot close because in view only mode" << endl;
                     }
@@ -276,7 +285,7 @@ void Monitor::CheckSegment()
 
         unsigned int numDevices = 0;
 
-        if (fInteractive) {
+        if (fInteractive || fViewOnly) {
             DeviceCounter* dc = managementSegment.find<DeviceCounter>(bipc::unique_instance).first;
             if (dc) {
                 numDevices = dc->fCount;
@@ -288,7 +297,7 @@ void Monitor::CheckSegment()
 
         if (fHeartbeatTriggered && duration > fTimeoutInMS) {
             cout << "no heartbeats since over " << fTimeoutInMS << " milliseconds, cleaning..." << endl;
-            Cleanup(fShmId);
+            Cleanup(ShmId{fShmId});
             fHeartbeatTriggered = false;
             if (fSelfDestruct) {
                 cout << "\nself destructing" << endl;
@@ -304,6 +313,15 @@ void Monitor::CheckSegment()
                  << setw(8)  << numDevices                                      << " | "
                  << setw(10) << (fViewOnly ? "view only" : to_string(duration)) << " |"
                  << c << flush;
+        } else if (fViewOnly) {
+            size_t free = segment.get_free_memory();
+            size_t total = segment.get_size();
+            size_t used = total - free;
+            LOGV(info, user1) << "[" << fSegmentName
+                              << "] devices: " << numDevices
+                              << ", total: " << total
+                              << ", free: " << free
+                              << ", used: " << used;
         }
     } catch (bie&) {
         fHeartbeatTriggered = false;
@@ -321,7 +339,7 @@ void Monitor::CheckSegment()
         unsigned int duration = chrono::duration_cast<chrono::milliseconds>(now - fLastHeartbeat).count();
 
         if (fIsDaemon && duration > fTimeoutInMS * 2) {
-            Cleanup(fShmId);
+            Cleanup(ShmId{fShmId});
             fHeartbeatTriggered = false;
             if (fSelfDestruct) {
                 cout << "\nself destructing" << endl;
@@ -395,51 +413,52 @@ void Monitor::PrintHelp()
 void Monitor::RemoveObject(const string& name)
 {
     if (bipc::shared_memory_object::remove(name.c_str())) {
-        cout << "Successfully removed \"" << name << "\"." << endl;
+        cout << "Successfully removed '" << name << "'." << endl;
     } else {
-        cout << "Did not remove \"" << name << "\". Already removed?" << endl;
+        cout << "Did not remove '" << name << "'. Already removed?" << endl;
     }
 }
 
 void Monitor::RemoveFileMapping(const string& name)
 {
     if (bipc::file_mapping::remove(name.c_str())) {
-        cout << "Successfully removed \"" << name << "\"." << endl;
+        cout << "Successfully removed '" << name << "'." << endl;
     } else {
-        cout << "Did not remove \"" << name << "\". Already removed?" << endl;
+        cout << "Did not remove '" << name << "'. Already removed?" << endl;
     }
 }
 
 void Monitor::RemoveQueue(const string& name)
 {
     if (bipc::message_queue::remove(name.c_str())) {
-        cout << "Successfully removed \"" << name << "\"." << endl;
+        cout << "Successfully removed '" << name << "'." << endl;
     } else {
-        cout << "Did not remove \"" << name << "\". Already removed?" << endl;
+        cout << "Did not remove '" << name << "'. Already removed?" << endl;
     }
 }
 
 void Monitor::RemoveMutex(const string& name)
 {
     if (bipc::named_mutex::remove(name.c_str())) {
-        cout << "Successfully removed \"" << name << "\"." << endl;
+        cout << "Successfully removed '" << name << "'." << endl;
     } else {
-        cout << "Did not remove \"" << name << "\". Already removed?" << endl;
+        cout << "Did not remove '" << name << "'. Already removed?" << endl;
     }
 }
 
 void Monitor::RemoveCondition(const string& name)
 {
     if (bipc::named_condition::remove(name.c_str())) {
-        cout << "Successfully removed \"" << name << "\"." << endl;
+        cout << "Successfully removed '" << name << "'." << endl;
     } else {
-        cout << "Did not remove \"" << name << "\". Already removed?" << endl;
+        cout << "Did not remove '" << name << "'. Already removed?" << endl;
     }
 }
 
-void Monitor::Cleanup(const string& shmId)
+void Monitor::Cleanup(const ShmId& shmId)
 {
-    string managementSegmentName("fmq_" + shmId + "_mng");
+    cout << "Cleaning up for shared memory id '" << shmId.shmId << "'..." << endl;
+    string managementSegmentName("fmq_" + shmId.shmId + "_mng");
     try {
         bipc::managed_shared_memory managementSegment(bipc::open_only, managementSegmentName.c_str());
         RegionCounter* rc = managementSegment.find<RegionCounter>(bipc::unique_instance).first;
@@ -454,43 +473,64 @@ void Monitor::Cleanup(const string& shmId)
                     RegionInfo ri = m->at(i);
                     string path = ri.fPath.c_str();
                     int flags = ri.fFlags;
-                    cout << "Found RegionInfo with path: '" << path << "', flags: " << flags << "'." << endl;
+                    cout << "Found RegionInfo with path: '" << path << "', flags: " << flags << ", fDestroyed: " << ri.fDestroyed << "." << endl;
                     if (path != "") {
-                        RemoveFileMapping(tools::ToString(path, "fmq_" + shmId + "_rg_" + to_string(i)));
+                        RemoveFileMapping(tools::ToString(path, "fmq_" + shmId.shmId + "_rg_" + to_string(i)));
                     } else {
-                        RemoveObject("fmq_" + shmId + "_rg_" + to_string(i));
+                        RemoveObject("fmq_" + shmId.shmId + "_rg_" + to_string(i));
                     }
                 } else {
-                    RemoveObject("fmq_" + shmId + "_rg_" + to_string(i));
+                    RemoveObject("fmq_" + shmId.shmId + "_rg_" + to_string(i));
                 }
 
-                RemoveQueue(string("fmq_" + shmId + "_rgq_" + to_string(i)));
+                RemoveQueue(string("fmq_" + shmId.shmId + "_rgq_" + to_string(i)));
             }
         } else {
-            cout << "No region counter found. no regions to cleanup." << endl;
+            cout << "No region counter found. No regions to cleanup." << endl;
         }
 
         RemoveObject(managementSegmentName.c_str());
     } catch (bie&) {
         cout << "Did not find '" << managementSegmentName << "' shared memory segment. No regions to cleanup." << endl;
-    } catch(std::out_of_range& oor) {
+    } catch(out_of_range& oor) {
         cout << "Could not locate element in the region map, out of range: " << oor.what() << endl;
     }
 
-    RemoveObject("fmq_" + shmId + "_main");
-    RemoveMutex("fmq_" + shmId + "_mtx");
-    RemoveCondition("fmq_" + shmId + "_cv");
+    RemoveObject("fmq_" + shmId.shmId + "_main");
+    RemoveMutex("fmq_" + shmId.shmId + "_mtx");
+    RemoveCondition("fmq_" + shmId.shmId + "_cv");
 
     cout << endl;
 }
 
+void Monitor::Cleanup(const SessionId& sessionId)
+{
+    ShmId shmId{buildShmIdFromSessionIdAndUserId(sessionId.sessionId)};
+    cout << "Cleanup called with session id '" << sessionId.sessionId << "', translating to shared memory id '" << shmId.shmId << "'" << endl;
+    Cleanup(shmId);
+}
+
+void Monitor::CleanupFull(const ShmId& shmId)
+{
+    Cleanup(shmId);
+    RemoveMutex("fmq_" + shmId.shmId + "_ms");
+    RemoveQueue("fmq_" + shmId.shmId + "_cq");
+}
+
+void Monitor::CleanupFull(const SessionId& sessionId)
+{
+    ShmId shmId{buildShmIdFromSessionIdAndUserId(sessionId.sessionId)};
+    cout << "Cleanup called with session id '" << sessionId.sessionId << "', translating to shared memory id '" << shmId.shmId << "'" << endl;
+    CleanupFull(shmId);
+}
+
 Monitor::~Monitor()
 {
     if (fSignalThread.joinable()) {
         fSignalThread.join();
     }
     if (fCleanOnExit) {
-        Cleanup(fShmId);
+        Cleanup(ShmId{fShmId});
     }
     if (!fViewOnly) {
         RemoveMutex("fmq_" + fShmId + "_ms");

fairmq/shmem/Monitor.h

@@ -8,8 +8,6 @@
 #ifndef FAIR_MQ_SHMEM_MONITOR_H_
 #define FAIR_MQ_SHMEM_MONITOR_H_
 
-#include <boost/interprocess/managed_shared_memory.hpp>
-
 #include <thread>
 #include <chrono>
 #include <atomic>
@@ -24,10 +22,22 @@ namespace mq
 namespace shmem
 {
 
+struct SessionId
+{
+    std::string sessionId;
+    explicit operator std::string() const { return sessionId; }
+};
+
+struct ShmId
+{
+    std::string shmId;
+    explicit operator std::string() const { return shmId; }
+};
+
 class Monitor
 {
   public:
-    Monitor(const std::string& shmId, bool selfDestruct, bool interactive, bool viewOnly, unsigned int timeoutInMS, bool runAsDaemon, bool cleanOnExit);
+    Monitor(const std::string& shmId, bool selfDestruct, bool interactive, bool viewOnly, unsigned int timeoutInMS, unsigned int intervalInMS, bool runAsDaemon, bool cleanOnExit);
 
     Monitor(const Monitor&) = delete;
     Monitor operator=(const Monitor&) = delete;
@@ -37,7 +47,19 @@ class Monitor
     void CatchSignals();
     void Run();
 
-    static void Cleanup(const std::string& shmId);
+    /// @brief Cleanup all shared memory artifacts created by devices
+    /// @param shmId shared memory id
+    static void Cleanup(const ShmId& shmId);
+    /// @brief Cleanup all shared memory artifacts created by devices
+    /// @param sessionId session id
+    static void Cleanup(const SessionId& sessionId);
+    /// @brief Cleanup all shared memory artifacts created by devices and monitors
+    /// @param shmId shared memory id
+    static void CleanupFull(const ShmId& shmId);
+    /// @brief Cleanup all shared memory artifacts created by devices and monitors
+    /// @param sessionId session id
+    static void CleanupFull(const SessionId& sessionId);
+
     static void RemoveObject(const std::string&);
     static void RemoveFileMapping(const std::string&);
     static void RemoveQueue(const std::string&);
@@ -62,6 +84,7 @@ class Monitor
     bool fSeenOnce; // true is segment has been opened successfully at least once
     bool fCleanOnExit;
     unsigned int fTimeoutInMS;
+    unsigned int fIntervalInMS;
     std::string fShmId;
     std::string fSegmentName;
     std::string fManagementSegmentName;
@@ -70,7 +93,6 @@ class Monitor
     std::atomic<bool> fHeartbeatTriggered;
     std::chrono::high_resolution_clock::time_point fLastHeartbeat;
     std::thread fSignalThread;
-    boost::interprocess::managed_shared_memory fManagementSegment;
     std::unordered_map<std::string, std::chrono::high_resolution_clock::time_point> fDeviceHeartbeats;
 };
 

fairmq/shmem/Poller.h

@@ -127,13 +127,20 @@ class Poller final : public fair::mq::Poller
 
     void Poll(const int timeout) override
     {
-        if (zmq_poll(fItems, fNumItems, timeout) < 0) {
-            if (errno == ETERM) {
-                LOG(debug) << "polling exited, reason: " << zmq_strerror(errno);
-            } else {
-                LOG(error) << "polling failed, reason: " << zmq_strerror(errno);
-                throw fair::mq::PollerError(fair::mq::tools::ToString("Polling failed, reason: ", zmq_strerror(errno)));
+        while (true) {
+            if (zmq_poll(fItems, fNumItems, timeout) < 0) {
+                if (errno == ETERM) {
+                    LOG(debug) << "polling exited, reason: " << zmq_strerror(errno);
+                    return;
+                } else if (errno == EINTR) {
+                    LOG(debug) << "polling interrupted by system call";
+                    continue;
+                } else {
+                    LOG(error) << "polling failed, reason: " << zmq_strerror(errno);
+                    throw fair::mq::PollerError(fair::mq::tools::ToString("Polling failed, reason: ", zmq_strerror(errno)));
+                }
             }
+            break;
         }
     }
 

fairmq/shmem/README.md

@@ -6,6 +6,23 @@ The transport manages shared memory via boost::interprocess library. The transfe
 
 Devices track and cleanup shared memory on shutdown. For more information on the current shared memory segment and additional cleanup options, see following section.
 
+# Shared Memory objects / files
+
+FairMQ Shared Memory currently uses the following names to register shared memory on the system:
+
+| name                      | info                                           | created by         | used by                        |
+| ------------------------- | ---------------------------------------------- | ------------------ | ------------------------------ |
+| `fmq_<shmId>_main`        | main segment (user data)                       | one of the devices | devices                        |
+| `fmq_<shmId>_mng`         | management segment (management data)           | one of the devices | devices                        |
+| `fmq_<shmId>_mtx`         | mutex                                          | one of the devices | devices                        |
+| `fmq_<shmId>_cv`          | condition variable                             | one of the devices | devices with unmanaged regions |
+| `fmq_<shmId>_rg_<index>`  | unmanaged region(s)                            | one of the devices | devices with unmanaged regions |
+| `fmq_<shmId>_rgq_<index>` | unmanaged region queue(s)                      | one of the devices | devices with unmanaged regions |
+| `fmq_<shmId>_ms`          | shmmonitor status                              | shmmonitor         | devices, shmmonitor            |
+| `fmq_<shmId>_cq`          | message queue between transport and shmmonitor | shmmonitor         | devices, shmmonitor            |
+
+The shmId is generated out of session id and user id.
+
 ## Shared memory monitor
 
 The shared memory monitor tool, supplied with the shared memory transport can be used to monitor shared memory use and automatically cleanup shared memory in case of device crashes.
@@ -25,13 +42,3 @@ Without the `--self-destruct` option, the monitor will run continuously, moitori
 Possible further implementation would be to run the monitor with `--self-destruct` with each topology.
 
 The Monitor class can also be used independently from the supplied executable (built from `runMonitor.cxx`), allowing integration on any level. For example invoking the monitor could be a functionality that a device offers.
-
-FairMQ Shared Memory currently uses following names to register shared memory on the system:
-
-`fmq_<shmId>_main` - main segment name, used for user data (the shmId is generated out of session id and user id).
-`fmq_<shmId>_mng` - management segment name, used for storing management data.
-`fmq_<shmId>_cq` - message queue for communicating between shm transport and shm monitor (exists independent of above segments).
-`fmq_<shmId>_mtx` - boost::interprocess::named_mutex for management purposes (exists independent of above segments).
-`fmq_<shmId>_ms` - shmmonitor status used to signal if it is active or not (exists independent of above segments).
-`fmq_<shmId>_rg_<index>` - names of unmanaged regions.
-`fmq_<shmId>_rgq_<index>` - names of queues for the unmanaged regions.

fairmq/shmem/Region.h

@@ -30,6 +30,7 @@
 #include <boost/interprocess/ipc/message_queue.hpp>
 
 #include <algorithm> // min
+#include <atomic>
 #include <thread>
 #include <mutex>
 #include <condition_variable>
@@ -49,6 +50,7 @@ struct Region
 {
     Region(const std::string& shmId, uint64_t id, uint64_t size, bool remote, RegionCallback callback, RegionBulkCallback bulkCallback, const std::string& path, int flags)
         : fRemote(remote)
+        , fLinger(100)
         , fStop(false)
         , fName("fmq_" + shmId + "_rg_" + std::to_string(id))
         , fQueueName("fmq_" + shmId + "_rgq_" + std::to_string(id))
@@ -56,8 +58,8 @@ struct Region
         , fFile(nullptr)
         , fFileMapping()
         , fQueue(nullptr)
-        , fReceiveAcksWorker()
-        , fSendAcksWorker()
+        , fAcksReceiver()
+        , fAcksSender()
         , fCallback(callback)
         , fBulkCallback(bulkCallback)
     {
@@ -118,38 +120,35 @@ struct Region
         LOG(debug) << "shmem: initialized region queue: " << fQueueName;
     }
 
-    void StartSendingAcks()
-    {
-        fSendAcksWorker = std::thread(&Region::SendAcks, this);
-    }
-
+    void StartSendingAcks() { fAcksSender = std::thread(&Region::SendAcks, this); }
     void SendAcks()
     {
         std::unique_ptr<RegionBlock[]> blocks = tools::make_unique<RegionBlock[]>(fAckBunchSize);
+        size_t blocksToSend = 0;
 
-        while (true) { // we'll try to send all acks before stopping
-            size_t blocksToSend = 0;
-
-            {   // mutex locking block
+        while (true) {
+            blocksToSend = 0;
+            {
                 std::unique_lock<std::mutex> lock(fBlockMtx);
 
-                // try to get more blocks without waiting (we can miss a notify from CloseMessage())
-                if (!fStop && (fBlocksToFree.size() < fAckBunchSize)) {
-                    // cv.wait() timeout: send whatever blocks we have
+                // try to get <fAckBunchSize> blocks
+                if (fBlocksToFree.size() < fAckBunchSize) {
                     fBlockSendCV.wait_for(lock, std::chrono::milliseconds(500));
                 }
 
+                // send whatever blocks we have
                 blocksToSend = std::min(fBlocksToFree.size(), fAckBunchSize);
 
                 copy_n(fBlocksToFree.end() - blocksToSend, blocksToSend, blocks.get());
                 fBlocksToFree.resize(fBlocksToFree.size() - blocksToSend);
-            } // unlock the block mutex here while sending over IPC
+            }
 
             if (blocksToSend > 0) {
                 while (!fQueue->try_send(blocks.get(), blocksToSend * sizeof(RegionBlock), 0) && !fStop) {
                     // receiver slow? yield and try again...
                     std::this_thread::yield();
                 }
+                // LOG(debug) << "Sent " << blocksToSend << " blocks.";
             } else { // blocksToSend == 0
                 if (fStop) {
                     break;
@@ -157,14 +156,11 @@ struct Region
             }
         }
 
-        LOG(debug) << "send ack worker for " << fName << " leaving.";
-    }
-
-    void StartReceivingAcks()
-    {
-        fReceiveAcksWorker = std::thread(&Region::ReceiveAcks, this);
+        LOG(debug) << "AcksSender for " << fName << " leaving " << "(blocks left to free: " << fBlocksToFree.size() << ", "
+                                                                << " blocks left to send: " << blocksToSend << ").";
     }
 
+    void StartReceivingAcks() { fAcksReceiver = std::thread(&Region::ReceiveAcks, this); }
     void ReceiveAcks()
     {
         unsigned int priority;
@@ -173,12 +169,18 @@ struct Region
         std::vector<fair::mq::RegionBlock> result;
         result.reserve(fAckBunchSize);
 
-        while (!fStop) { // end thread condition (should exist until region is destroyed)
-            auto rcvTill = boost::posix_time::microsec_clock::universal_time() + boost::posix_time::milliseconds(500);
+        while (true) {
+            uint32_t timeout = 100;
+            bool leave = false;
+            if (fStop) {
+                timeout = fLinger;
+                leave = true;
+            }
+            auto rcvTill = boost::posix_time::microsec_clock::universal_time() + boost::posix_time::milliseconds(timeout);
 
             while (fQueue->timed_receive(blocks.get(), fAckBunchSize * sizeof(RegionBlock), recvdSize, priority, rcvTill)) {
-                // LOG(debug) << "received: " << block.fHandle << " " << block.fSize << " " << block.fMessageId;
                 const auto numBlocks = recvdSize / sizeof(RegionBlock);
+                // LOG(debug) << "Received " << numBlocks << " blocks (recvdSize: " << recvdSize << "). (remaining queue size: " << fQueue->get_num_msg() << ").";
                 if (fBulkCallback) {
                     result.clear();
                     for (size_t i = 0; i < numBlocks; i++) {
@@ -191,9 +193,13 @@ struct Region
                     }
                 }
             }
-        } // while !fStop
 
-        LOG(debug) << "ReceiveAcks() worker for " << fName << " leaving.";
+            if (leave) {
+                break;
+            }
+        }
+
+        LOG(debug) << "AcksReceiver for " << fName << " leaving (remaining queue size: " << fQueue->get_num_msg() << ").";
     }
 
     void ReleaseBlock(const RegionBlock& block)
@@ -203,31 +209,34 @@ struct Region
         fBlocksToFree.emplace_back(block);
 
         if (fBlocksToFree.size() >= fAckBunchSize) {
-            lock.unlock(); // reduces contention on fBlockMtx
+            lock.unlock();
             fBlockSendCV.notify_one();
         }
     }
 
+    void SetLinger(uint32_t linger) { fLinger = linger; }
+    uint32_t GetLinger() const { return fLinger; }
+
     ~Region()
     {
         fStop = true;
 
-        if (fSendAcksWorker.joinable()) {
+        if (fAcksSender.joinable()) {
             fBlockSendCV.notify_one();
-            fSendAcksWorker.join();
+            fAcksSender.join();
         }
 
         if (!fRemote) {
-            if (fReceiveAcksWorker.joinable()) {
-                fReceiveAcksWorker.join();
+            if (fAcksReceiver.joinable()) {
+                fAcksReceiver.join();
             }
 
             if (boost::interprocess::shared_memory_object::remove(fName.c_str())) {
-                LOG(debug) << "shmem: destroyed region " << fName;
+                LOG(debug) << "Region '" << fName << "' destroyed.";
             }
 
             if (boost::interprocess::file_mapping::remove(fName.c_str())) {
-                LOG(debug) << "shmem: destroyed file mapping " << fName;
+                LOG(debug) << "File mapping '" << fName << "' destroyed.";
             }
 
             if (fFile) {
@@ -235,16 +244,19 @@ struct Region
             }
 
             if (boost::interprocess::message_queue::remove(fQueueName.c_str())) {
-                LOG(debug) << "shmem: removed region queue " << fQueueName;
+                LOG(debug) << "Region queue '" << fQueueName << "' destroyed.";
             }
         } else {
             // LOG(debug) << "shmem: region '" << fName << "' is remote, no cleanup necessary.";
-            LOG(debug) << "shmem: region queue '" << fQueueName << "' is remote, no cleanup necessary";
+            LOG(debug) << "Region queue '" << fQueueName << "' is remote, no cleanup necessary";
         }
+
+        LOG(debug) << "Region '" << fName << "' (" << (fRemote ? "remote" : "local") << ") destructed.";
     }
 
     bool fRemote;
-    bool fStop;
+    uint32_t fLinger;
+    std::atomic<bool> fStop;
     std::string fName;
     std::string fQueueName;
     boost::interprocess::shared_memory_object fShmemObject;
@@ -258,8 +270,8 @@ struct Region
     const std::size_t fAckBunchSize = 256;
     std::unique_ptr<boost::interprocess::message_queue> fQueue;
 
-    std::thread fReceiveAcksWorker;
-    std::thread fSendAcksWorker;
+    std::thread fAcksReceiver;
+    std::thread fAcksSender;
     RegionCallback fCallback;
     RegionBulkCallback fBulkCallback;
 };

fairmq/shmem/Socket.h

@@ -55,8 +55,7 @@ class Socket final : public fair::mq::Socket
         , fBytesRx(0)
         , fMessagesTx(0)
         , fMessagesRx(0)
-        , fSndTimeout(100)
-        , fRcvTimeout(100)
+        , fTimeout(100)
     {
         assert(context);
         fSocket = zmq_socket(context, GetConstant(type));
@@ -77,11 +76,11 @@ class Socket final : public fair::mq::Socket
             LOG(error) << "Failed setting ZMQ_LINGER socket option, reason: " << zmq_strerror(errno);
         }
 
-        if (zmq_setsockopt(fSocket, ZMQ_SNDTIMEO, &fSndTimeout, sizeof(fSndTimeout)) != 0) {
+        if (zmq_setsockopt(fSocket, ZMQ_SNDTIMEO, &fTimeout, sizeof(fTimeout)) != 0) {
             LOG(error) << "Failed setting ZMQ_SNDTIMEO socket option, reason: " << zmq_strerror(errno);
         }
 
-        if (zmq_setsockopt(fSocket, ZMQ_RCVTIMEO, &fRcvTimeout, sizeof(fRcvTimeout)) != 0) {
+        if (zmq_setsockopt(fSocket, ZMQ_RCVTIMEO, &fTimeout, sizeof(fTimeout)) != 0) {
             LOG(error) << "Failed setting ZMQ_RCVTIMEO socket option, reason: " << zmq_strerror(errno);
         }
 
@@ -129,6 +128,32 @@ class Socket final : public fair::mq::Socket
         return true;
     }
 
+    bool ShouldRetry(int flags, int timeout, int& elapsed) const
+    {
+        if (!fManager.Interrupted() && ((flags & ZMQ_DONTWAIT) == 0)) {
+            if (timeout > 0) {
+                elapsed += fTimeout;
+                if (elapsed >= timeout) {
+                    return false;
+                }
+            }
+            return true;
+        } else {
+            return false;
+        }
+    }
+
+    int HandleErrors() const
+    {
+        if (zmq_errno() == ETERM) {
+            LOG(debug) << "Terminating socket " << fId;
+            return -1;
+        } else {
+            LOG(error) << "Failed transfer on socket " << fId << ", reason: " << zmq_strerror(errno);
+            return -1;
+        }
+    }
+
     int Send(MessagePtr& msg, const int timeout = -1) override
     {
         int flags = 0;
@@ -149,27 +174,14 @@ class Socket final : public fair::mq::Socket
                 size_t size = msg->GetSize();
                 fBytesTx += size;
                 return size;
-            } else if (zmq_errno() == EAGAIN) {
-                if (!fManager.Interrupted() && ((flags & ZMQ_DONTWAIT) == 0)) {
-                    if (timeout > 0) {
-                        elapsed += fSndTimeout;
-                        if (elapsed >= timeout) {
-                            return -2;
-                        }
-                    }
+            } else if (zmq_errno() == EAGAIN || zmq_errno() == EINTR) {
+                if (ShouldRetry(flags, timeout, elapsed)) {
                     continue;
                 } else {
                     return -2;
                 }
-            } else if (zmq_errno() == ETERM) {
-                LOG(info) << "terminating socket " << fId;
-                return -1;
-            } else if (zmq_errno() == EINTR) {
-                LOG(debug) << "Send interrupted by system call";
-                return nbytes;
-            }else {
-                LOG(error) << "Failed sending on socket " << fId << ", reason: " << zmq_strerror(errno) << ", nbytes = " << nbytes;
-                return nbytes;
+            } else {
+                return HandleErrors();
             }
         }
 
@@ -205,27 +217,14 @@ class Socket final : public fair::mq::Socket
                 fBytesRx += size;
                 ++fMessagesRx;
                 return size;
-            } else if (zmq_errno() == EAGAIN) {
-                if (!fManager.Interrupted() && ((flags & ZMQ_DONTWAIT) == 0)) {
-                    if (timeout > 0) {
-                        elapsed += fRcvTimeout;
-                        if (elapsed >= timeout) {
-                            return -2;
-                        }
-                    }
+            } else if (zmq_errno() == EAGAIN || zmq_errno() == EINTR) {
+                if (ShouldRetry(flags, timeout, elapsed)) {
                     continue;
                 } else {
                     return -2;
                 }
-            } else if (zmq_errno() == ETERM) {
-                LOG(info) << "terminating socket " << fId;
-                return -1;
-            } else if (zmq_errno() == EINTR) {
-                LOG(debug) << "Receive interrupted by system call";
-                return nbytes;
-            }else {
-                LOG(error) << "Failed receiving on socket " << fId << ", errno: " << errno << ", reason: " << zmq_strerror(errno) << ", nbytes = " << nbytes;
-                return nbytes;
+            } else {
+                return HandleErrors();
             }
         }
     }
@@ -267,27 +266,14 @@ class Socket final : public fair::mq::Socket
                 fBytesTx += totalSize;
 
                 return totalSize;
-            } else if (zmq_errno() == EAGAIN) {
-                if (!fManager.Interrupted() && ((flags & ZMQ_DONTWAIT) == 0)) {
-                    if (timeout > 0) {
-                        elapsed += fSndTimeout;
-                        if (elapsed >= timeout) {
-                            return -2;
-                        }
-                    }
+            } else if (zmq_errno() == EAGAIN || zmq_errno() == EINTR) {
+                if (ShouldRetry(flags, timeout, elapsed)) {
                     continue;
                 } else {
                     return -2;
                 }
-            } else if (zmq_errno() == ETERM) {
-                LOG(info) << "terminating socket " << fId;
-                return -1;
-            } else if (zmq_errno() == EINTR) {
-                LOG(debug) << "Send interrupted by system call";
-                return nbytes;
-            }else {
-                LOG(error) << "Failed sending on socket " << fId << ", reason: " << zmq_strerror(errno) << ", nbytes = " << nbytes;
-                return nbytes;
+            } else {
+                return HandleErrors();
             }
         }
 
@@ -334,24 +320,14 @@ class Socket final : public fair::mq::Socket
                 fBytesRx += totalSize;
 
                 return totalSize;
-            } else if (zmq_errno() == EAGAIN) {
-                if (!fManager.Interrupted() && ((flags & ZMQ_DONTWAIT) == 0)) {
-                    if (timeout > 0) {
-                        elapsed += fRcvTimeout;
-                        if (elapsed >= timeout) {
-                            return -2;
-                        }
-                    }
+            } else if (zmq_errno() == EAGAIN || zmq_errno() == EINTR) {
+                if (ShouldRetry(flags, timeout, elapsed)) {
                     continue;
                 } else {
                     return -2;
                 }
-            } else if (zmq_errno() == EINTR) {
-                LOG(debug) << "Receive interrupted by system call";
-                return nbytes;
             } else {
-                LOG(error) << "Failed receiving on socket " << fId << ", errno: " << errno << ", reason: " << zmq_strerror(errno) << ", nbytes = " << nbytes;
-                return nbytes;
+                return HandleErrors();
             }
         }
 
@@ -396,6 +372,15 @@ class Socket final : public fair::mq::Socket
         }
     }
 
+    void Events(uint32_t* events) override
+    {
+        size_t eventsSize = sizeof(uint32_t);
+        if (zmq_getsockopt(fSocket, ZMQ_EVENTS, events, &eventsSize) < 0) {
+            throw SocketError(
+                tools::ToString("failed setting ZMQ_EVENTS, reason: ", zmq_strerror(errno)));
+        }
+    }
+
     int GetLinger() const override
     {
         int value = 0;
@@ -505,6 +490,14 @@ class Socket final : public fair::mq::Socket
         if (constant == "no-block") return ZMQ_DONTWAIT;
         if (constant == "snd-more no-block") return ZMQ_DONTWAIT|ZMQ_SNDMORE;
 
+        if (constant == "fd") return ZMQ_FD;
+        if (constant == "events")
+            return ZMQ_EVENTS;
+        if (constant == "pollin")
+            return ZMQ_POLLIN;
+        if (constant == "pollout")
+            return ZMQ_POLLOUT;
+
         return -1;
     }
 
@@ -519,8 +512,7 @@ class Socket final : public fair::mq::Socket
     std::atomic<unsigned long> fMessagesTx;
     std::atomic<unsigned long> fMessagesRx;
 
-    int fSndTimeout;
-    int fRcvTimeout;
+    int fTimeout;
 };
 
 }

fairmq/shmem/TransportFactory.h

@@ -43,7 +43,7 @@ class TransportFactory final : public fair::mq::TransportFactory
         : fair::mq::TransportFactory(id)
         , fDeviceId(id)
         , fShmId()
-        , fZMQContext(zmq_ctx_new())
+        , fZmqCtx(zmq_ctx_new())
         , fManager(nullptr)
     {
         int major, minor, patch;
@@ -51,42 +51,35 @@ class TransportFactory final : public fair::mq::TransportFactory
         LOG(debug) << "Transport: Using ZeroMQ (" << major << "." << minor << "." << patch << ") & "
                 << "boost::interprocess (" << (BOOST_VERSION / 100000) << "." << (BOOST_VERSION / 100 % 1000) << "." << (BOOST_VERSION % 100) << ")";
 
-        if (!fZMQContext) {
+        if (!fZmqCtx) {
             throw std::runtime_error(tools::ToString("failed creating context, reason: ", zmq_strerror(errno)));
         }
 
         int numIoThreads = 1;
         std::string sessionName = "default";
-        size_t segmentSize = 2000000000;
-        bool autolaunchMonitor = false;
-        bool throwOnBadAlloc = true;
+        size_t segmentSize = 2ULL << 30;
         if (config) {
             numIoThreads = config->GetProperty<int>("io-threads", numIoThreads);
             sessionName = config->GetProperty<std::string>("session", sessionName);
             segmentSize = config->GetProperty<size_t>("shm-segment-size", segmentSize);
-            autolaunchMonitor = config->GetProperty<bool>("shm-monitor", autolaunchMonitor);
-            throwOnBadAlloc = config->GetProperty<bool>("shm-throw-bad-alloc", throwOnBadAlloc);
         } else {
             LOG(debug) << "ProgOptions not available! Using defaults.";
         }
 
         fShmId = buildShmIdFromSessionIdAndUserId(sessionName);
+        LOG(debug) << "Generated shmid '" << fShmId << "' out of session id '" << sessionName << "'.";
 
         try {
-            if (zmq_ctx_set(fZMQContext, ZMQ_IO_THREADS, numIoThreads) != 0) {
+            if (zmq_ctx_set(fZmqCtx, ZMQ_IO_THREADS, numIoThreads) != 0) {
                 LOG(error) << "failed configuring context, reason: " << zmq_strerror(errno);
             }
 
             // Set the maximum number of allowed sockets on the context.
-            if (zmq_ctx_set(fZMQContext, ZMQ_MAX_SOCKETS, 10000) != 0) {
+            if (zmq_ctx_set(fZmqCtx, ZMQ_MAX_SOCKETS, 10000) != 0) {
                 LOG(error) << "failed configuring context, reason: " << zmq_strerror(errno);
             }
 
-            if (autolaunchMonitor) {
-                Manager::StartMonitor(fShmId);
-            }
-
-            fManager = tools::make_unique<Manager>(fShmId, fDeviceId, segmentSize, throwOnBadAlloc);
+            fManager = tools::make_unique<Manager>(fShmId, fDeviceId, segmentSize, config);
         } catch (boost::interprocess::interprocess_exception& e) {
             LOG(error) << "Could not initialize shared memory transport: " << e.what();
             throw std::runtime_error(tools::ToString("Could not initialize shared memory transport: ", e.what()));
@@ -128,7 +121,7 @@ class TransportFactory final : public fair::mq::TransportFactory
 
     SocketPtr CreateSocket(const std::string& type, const std::string& name) override
     {
-        return tools::make_unique<Socket>(*fManager, type, name, GetId(), fZMQContext, this);
+        return tools::make_unique<Socket>(*fManager, type, name, GetId(), fZmqCtx, this);
     }
 
     PollerPtr CreatePoller(const std::vector<FairMQChannel>& channels) const override
@@ -186,14 +179,17 @@ class TransportFactory final : public fair::mq::TransportFactory
     {
         LOG(debug) << "Destroying Shared Memory transport...";
 
-        if (fZMQContext) {
-            if (zmq_ctx_term(fZMQContext) != 0) {
-                if (errno == EINTR) {
-                    LOG(error) << "failed closing context, reason: " << zmq_strerror(errno);
-                } else {
-                    fZMQContext = nullptr;
-                    return;
+        if (fZmqCtx) {
+            while (true) {
+                if (zmq_ctx_term(fZmqCtx) != 0) {
+                    if (errno == EINTR) {
+                        LOG(debug) << "zmq_ctx_term interrupted by system call, retrying";
+                        continue;
+                    } else {
+                        fZmqCtx = nullptr;
+                    }
                 }
+                break;
             }
         } else {
             LOG(error) << "context not available for shutdown";
@@ -203,7 +199,7 @@ class TransportFactory final : public fair::mq::TransportFactory
   private:
     std::string fDeviceId;
     std::string fShmId;
-    void* fZMQContext;
+    void* fZmqCtx;
     std::unique_ptr<Manager> fManager;
 };
 

fairmq/shmem/UnmanagedRegion.h

@@ -57,6 +57,8 @@ class UnmanagedRegion final : public fair::mq::UnmanagedRegion
     void* GetData() const override { return fRegion->get_address(); }
     size_t GetSize() const override { return fRegion->get_size(); }
     uint64_t GetId() const override { return fRegionId; }
+    void SetLinger(uint32_t linger) override { fManager.GetRegion(fRegionId)->SetLinger(linger); }
+    uint32_t GetLinger() const override { return fManager.GetRegion(fRegionId)->GetLinger(); }
 
     ~UnmanagedRegion() override { fManager.RemoveRegion(fRegionId); }
 

fairmq/shmem/runMonitor.cxx

@@ -76,6 +76,7 @@ int main(int argc, char** argv)
         bool interactive = false;
         bool viewOnly = false;
         unsigned int timeoutInMS = 5000;
+        unsigned int intervalInMS = 100;
         bool runAsDaemon = false;
         bool cleanOnExit = false;
 
@@ -90,6 +91,7 @@ int main(int argc, char** argv)
             ("timeout,t"      , value<unsigned int>(&timeoutInMS)->default_value(5000), "Heartbeat timeout in milliseconds")
             ("daemonize,d"    , value<bool>(&runAsDaemon)->implicit_value(true),        "Daemonize the monitor")
             ("clean-on-exit,e", value<bool>(&cleanOnExit)->implicit_value(true),        "Perform cleanup on exit")
+            ("interval"       , value<unsigned int>(&intervalInMS)->default_value(100), "Output interval for interactive/view-only mode")
             ("help,h", "Print help");
 
         variables_map vm;
@@ -111,15 +113,16 @@ int main(int argc, char** argv)
         }
 
         if (cleanup) {
-            cout << "Cleaning up \"" << shmId << "\"..." << endl;
-            Monitor::Cleanup(shmId);
-            Monitor::RemoveQueue("fmq_" + shmId + "_cq");
+            Monitor::CleanupFull(ShmId{shmId});
             return 0;
         }
 
         cout << "Starting shared memory monitor for session: \"" << sessionName << "\" (shmId: " << shmId << ")..." << endl;
+        if (viewOnly && !interactive) {
+            cout << "running in non-interactive view-only mode, outputting with interval of " << intervalInMS << "ms. (change with --interval), press ctrl+C to exit." << endl;
+        }
 
-        Monitor monitor(shmId, selfDestruct, interactive, viewOnly, timeoutInMS, runAsDaemon, cleanOnExit);
+        Monitor monitor(shmId, selfDestruct, interactive, viewOnly, timeoutInMS, intervalInMS, runAsDaemon, cleanOnExit);
 
         monitor.CatchSignals();
         monitor.Run();

fairmq/zeromq/Context.h

@@ -161,13 +161,16 @@ class Context
         UnsubscribeFromRegionEvents();
 
         if (fZmqCtx) {
-            if (zmq_ctx_term(fZmqCtx) != 0) {
-                if (errno == EINTR) {
-                    LOG(error) << " failed closing context, reason: " << zmq_strerror(errno);
-                } else {
-                    fZmqCtx = nullptr;
-                    return;
+            while (true) {
+                if (zmq_ctx_term(fZmqCtx) != 0) {
+                    if (errno == EINTR) {
+                        LOG(debug) << "zmq_ctx_term interrupted by system call, retrying";
+                        continue;
+                    } else {
+                        fZmqCtx = nullptr;
+                    }
                 }
+                break;
             }
         } else {
             LOG(error) << "context not available for shutdown";

fairmq/zeromq/Message.h

@@ -155,9 +155,17 @@ class Message final : public fair::mq::Message
     void* GetData() const override
     {
         if (!fViewMsg) {
-            return zmq_msg_data(fMsg.get());
+            if (zmq_msg_size(fMsg.get()) > 0) {
+                return zmq_msg_data(fMsg.get());
+            } else {
+                return nullptr;
+            }
         } else {
-            return zmq_msg_data(fViewMsg.get());
+            if (zmq_msg_size(fViewMsg.get()) > 0) {
+                return zmq_msg_data(fViewMsg.get());
+            } else {
+                return nullptr;
+            }
         }
     }
 

fairmq/zeromq/Poller.h

@@ -130,13 +130,20 @@ class Poller final : public fair::mq::Poller
 
     void Poll(const int timeout) override
     {
-        if (zmq_poll(fItems, fNumItems, timeout) < 0) {
-            if (errno == ETERM) {
-                LOG(debug) << "polling exited, reason: " << zmq_strerror(errno);
-            } else {
-                LOG(error) << "polling failed, reason: " << zmq_strerror(errno);
-                throw fair::mq::PollerError(fair::mq::tools::ToString("Polling failed, reason: ", zmq_strerror(errno)));
+        while (true) {
+            if (zmq_poll(fItems, fNumItems, timeout) < 0) {
+                if (errno == ETERM) {
+                    LOG(debug) << "polling exited, reason: " << zmq_strerror(errno);
+                    return;
+                } else if (errno == EINTR) {
+                    LOG(debug) << "polling interrupted by system call";
+                    continue;
+                } else {
+                    LOG(error) << "polling failed, reason: " << zmq_strerror(errno);
+                    throw fair::mq::PollerError(fair::mq::tools::ToString("Polling failed, reason: ", zmq_strerror(errno)));
+                }
             }
+            break;
         }
     }
 

fairmq/zeromq/Socket.h

@@ -12,13 +12,15 @@
 #include <FairMQLogger.h>
 #include <FairMQMessage.h>
 #include <FairMQSocket.h>
-#include <atomic>
 #include <fairmq/Tools.h>
 #include <fairmq/zeromq/Context.h>
 #include <fairmq/zeromq/Message.h>
-#include <memory>   // unique_ptr
+
 #include <zmq.h>
 
+#include <atomic>
+#include <memory> // unique_ptr
+
 namespace fair {
 namespace mq {
 namespace zmq {
@@ -35,8 +37,7 @@ class Socket final : public fair::mq::Socket
         , fBytesRx(0)
         , fMessagesTx(0)
         , fMessagesRx(0)
-        , fSndTimeout(100)
-        , fRcvTimeout(100)
+        , fTimeout(100)
     {
         if (fSocket == nullptr) {
             LOG(error) << "Failed creating socket " << fId << ", reason: " << zmq_strerror(errno);
@@ -54,11 +55,11 @@ class Socket final : public fair::mq::Socket
             LOG(error) << "Failed setting ZMQ_LINGER socket option, reason: " << zmq_strerror(errno);
         }
 
-        if (zmq_setsockopt(fSocket, ZMQ_SNDTIMEO, &fSndTimeout, sizeof(fSndTimeout)) != 0) {
+        if (zmq_setsockopt(fSocket, ZMQ_SNDTIMEO, &fTimeout, sizeof(fTimeout)) != 0) {
             LOG(error) << "Failed setting ZMQ_SNDTIMEO socket option, reason: " << zmq_strerror(errno);
         }
 
-        if (zmq_setsockopt(fSocket, ZMQ_RCVTIMEO, &fRcvTimeout, sizeof(fRcvTimeout)) != 0) {
+        if (zmq_setsockopt(fSocket, ZMQ_RCVTIMEO, &fTimeout, sizeof(fTimeout)) != 0) {
             LOG(error) << "Failed setting ZMQ_RCVTIMEO socket option, reason: " << zmq_strerror(errno);
         }
 
@@ -78,13 +79,12 @@ class Socket final : public fair::mq::Socket
 
     bool Bind(const std::string& address) override
     {
-        // LOG(info) << "bind socket " << fId << " on " << address;
+        // LOG(debug) << "Binding socket " << fId << " on " << address;
 
         if (zmq_bind(fSocket, address.c_str()) != 0) {
             if (errno == EADDRINUSE) {
                 // do not print error in this case, this is handled by FairMQDevice in case no
-                // connection could be established after trying a number of random ports from a
-                // range.
+                // connection could be established after trying a number of random ports from a range.
                 return false;
             }
             LOG(error) << "Failed binding socket " << fId << ", address: " << address << ", reason: " << zmq_strerror(errno);
@@ -96,7 +96,7 @@ class Socket final : public fair::mq::Socket
 
     bool Connect(const std::string& address) override
     {
-        // LOG(info) << "connect socket " << fId << " on " << address;
+        // LOG(debug) << "Connecting socket " << fId << " on " << address;
 
         if (zmq_connect(fSocket, address.c_str()) != 0) {
             LOG(error) << "Failed connecting socket " << fId << ", address: " << address << ", reason: " << zmq_strerror(errno);
@@ -106,6 +106,32 @@ class Socket final : public fair::mq::Socket
         return true;
     }
 
+    bool ShouldRetry(int flags, int timeout, int& elapsed) const
+    {
+        if (!fCtx.Interrupted() && ((flags & ZMQ_DONTWAIT) == 0)) {
+            if (timeout > 0) {
+                elapsed += fTimeout;
+                if (elapsed >= timeout) {
+                    return false;
+                }
+            }
+            return true;
+        } else {
+            return false;
+        }
+    }
+
+    int HandleErrors() const
+    {
+        if (zmq_errno() == ETERM) {
+            LOG(debug) << "Terminating socket " << fId;
+            return -1;
+        } else {
+            LOG(error) << "Failed transfer on socket " << fId << ", errno: " << errno << ", reason: " << zmq_strerror(errno);
+            return -1;
+        }
+    }
+
     int Send(MessagePtr& msg, const int timeout = -1) override
     {
         int flags = 0;
@@ -121,29 +147,15 @@ class Socket final : public fair::mq::Socket
             if (nbytes >= 0) {
                 fBytesTx += nbytes;
                 ++fMessagesTx;
-
                 return nbytes;
-            } else if (zmq_errno() == EAGAIN) {
-                if (!fCtx.Interrupted() && ((flags & ZMQ_DONTWAIT) == 0)) {
-                    if (timeout > 0) {
-                        elapsed += fSndTimeout;
-                        if (elapsed >= timeout) {
-                            return -2;
-                        }
-                    }
+            } else if (zmq_errno() == EAGAIN || zmq_errno() == EINTR) {
+                if (ShouldRetry(flags, timeout, elapsed)) {
                     continue;
                 } else {
                     return -2;
                 }
-            } else if (zmq_errno() == ETERM) {
-                LOG(info) << "terminating socket " << fId;
-                return -1;
-            } else if (zmq_errno() == EINTR) {
-                LOG(debug) << "Send interrupted by system call";
-                return nbytes;
             } else {
-                LOG(error) << "Failed sending on socket " << fId << ", reason: " << zmq_strerror(errno);
-                return nbytes;
+                return HandleErrors();
             }
         }
     }
@@ -162,27 +174,14 @@ class Socket final : public fair::mq::Socket
                 fBytesRx += nbytes;
                 ++fMessagesRx;
                 return nbytes;
-            } else if (zmq_errno() == EAGAIN) {
-                if (!fCtx.Interrupted() && ((flags & ZMQ_DONTWAIT) == 0)) {
-                    if (timeout > 0) {
-                        elapsed += fRcvTimeout;
-                        if (elapsed >= timeout) {
-                            return -2;
-                        }
-                    }
+            } else if (zmq_errno() == EAGAIN || zmq_errno() == EINTR) {
+                if (ShouldRetry(flags, timeout, elapsed)) {
                     continue;
                 } else {
                     return -2;
                 }
-            } else if (zmq_errno() == ETERM) {
-                LOG(info) << "terminating socket " << fId;
-                return -1;
-            } else if (zmq_errno() == EINTR) {
-                LOG(debug) << "Receive interrupted by system call";
-                return nbytes;
             } else {
-                LOG(error) << "Failed receiving on socket " << fId << ", reason: " << zmq_strerror(errno);
-                return nbytes;
+                return HandleErrors();
             }
         }
     }
@@ -210,32 +209,15 @@ class Socket final : public fair::mq::Socket
                     int nbytes = zmq_msg_send(static_cast<Message*>(msgVec[i].get())->GetMessage(), fSocket, (i < vecSize - 1) ? ZMQ_SNDMORE | flags : flags);
                     if (nbytes >= 0) {
                         totalSize += nbytes;
-                    } else {
-                        // according to ZMQ docs, this can only occur for the first part
-                        if (zmq_errno() == EAGAIN) {
-                            if (!fCtx.Interrupted() && ((flags & ZMQ_DONTWAIT) == 0)) {
-                                if (timeout > 0) {
-                                    elapsed += fSndTimeout;
-                                    if (elapsed >= timeout) {
-                                        return -2;
-                                    }
-                                }
-                                repeat = true;
-                                break;
-                            } else {
-                                return -2;
-                            }
-                        }
-                        if (zmq_errno() == ETERM) {
-                            LOG(info) << "terminating socket " << fId;
-                            return -1;
-                        } else if (zmq_errno() == EINTR) {
-                            LOG(debug) << "Receive interrupted by system call";
-                            return nbytes;
+                    } else if (zmq_errno() == EAGAIN || zmq_errno() == EINTR) {
+                        if (ShouldRetry(flags, timeout, elapsed)) {
+                            repeat = true;
+                            break;
                         } else {
-                            LOG(error) << "Failed sending on socket " << fId << ", reason: " << zmq_strerror(errno);
-                            return nbytes;
+                            return -2;
                         }
+                    } else {
+                        return HandleErrors();
                     }
                 }
 
@@ -243,16 +225,14 @@ class Socket final : public fair::mq::Socket
                     continue;
                 }
 
-                // store statistics on how many messages have been sent (handle all parts as a
-                // single message)
+                // store statistics on how many messages have been sent (handle all parts as a single message)
                 ++fMessagesTx;
                 fBytesTx += totalSize;
                 return totalSize;
             }
-        }   // If there's only one part, send it as a regular message
-        else if (vecSize == 1) {
+        } else if (vecSize == 1) { // If there's only one part, send it as a regular message
             return Send(msgVec.back(), timeout);
-        } else {   // if the vector is empty, something might be wrong
+        } else { // if the vector is empty, something might be wrong
             LOG(warn) << "Will not send empty vector";
             return -1;
         }
@@ -278,24 +258,15 @@ class Socket final : public fair::mq::Socket
                 if (nbytes >= 0) {
                     msgVec.push_back(move(part));
                     totalSize += nbytes;
-                } else if (zmq_errno() == EAGAIN) {
-                    if (!fCtx.Interrupted() && ((flags & ZMQ_DONTWAIT) == 0)) {
-                        if (timeout > 0) {
-                            elapsed += fRcvTimeout;
-                            if (elapsed >= timeout) {
-                                return -2;
-                            }
-                        }
+                } else if (zmq_errno() == EAGAIN || zmq_errno() == EINTR) {
+                    if (ShouldRetry(flags, timeout, elapsed)) {
                         repeat = true;
                         break;
                     } else {
                         return -2;
                     }
-                } else if (zmq_errno() == EINTR) {
-                    LOG(debug) << "Receive interrupted by system call";
-                    return nbytes;
                 } else {
-                    return nbytes;
+                    return HandleErrors();
                 }
 
                 size_t moreSize = sizeof(more);
@@ -306,8 +277,7 @@ class Socket final : public fair::mq::Socket
                 continue;
             }
 
-            // store statistics on how many messages have been received (handle all parts as a
-            // single message)
+            // store statistics on how many messages have been received (handle all parts as a single message)
             ++fMessagesRx;
             fBytesRx += totalSize;
             return totalSize;
@@ -345,6 +315,15 @@ class Socket final : public fair::mq::Socket
         }
     }
 
+    void Events(uint32_t* events) override
+    {
+        size_t eventsSize = sizeof(uint32_t);
+        if (zmq_getsockopt(fSocket, ZMQ_EVENTS, events, &eventsSize) < 0) {
+            throw SocketError(
+                tools::ToString("failed setting ZMQ_EVENTS, reason: ", zmq_strerror(errno)));
+        }
+    }
+
     void SetLinger(const int value) override
     {
         if (zmq_setsockopt(fSocket, ZMQ_LINGER, &value, sizeof(value)) < 0) {
@@ -459,6 +438,14 @@ class Socket final : public fair::mq::Socket
 
         if (constant == "linger") return ZMQ_LINGER;
 
+        if (constant == "fd") return ZMQ_FD;
+        if (constant == "events")
+            return ZMQ_EVENTS;
+        if (constant == "pollin")
+            return ZMQ_POLLIN;
+        if (constant == "pollout")
+            return ZMQ_POLLOUT;
+
         return -1;
     }
 
@@ -473,8 +460,7 @@ class Socket final : public fair::mq::Socket
     std::atomic<unsigned long> fMessagesTx;
     std::atomic<unsigned long> fMessagesRx;
 
-    int fSndTimeout;
-    int fRcvTimeout;
+    int fTimeout;
 };
 
 }   // namespace zmq

fairmq/zeromq/TransportFactory.h

@@ -55,7 +55,7 @@ class TransportFactory final : public FairMQTransportFactory
     {
         return tools::make_unique<Message>(this);
     }
-    
+
     MessagePtr CreateMessage(Alignment alignment) override
     {
         return tools::make_unique<Message>(alignment, this);

fairmq/zeromq/UnmanagedRegion.h

@@ -52,6 +52,8 @@ class UnmanagedRegion final : public fair::mq::UnmanagedRegion
     virtual size_t GetSize() const override { return fSize; }
     uint64_t GetId() const override { return fId; }
     int64_t GetUserFlags() const { return fUserFlags; }
+    void SetLinger(uint32_t /* linger */) override { LOG(debug) << "ZeroMQ UnmanagedRegion linger option not implemented. Acknowledgements are local."; }
+    uint32_t GetLinger() const override { LOG(debug) << "ZeroMQ UnmanagedRegion linger option not implemented. Acknowledgements are local."; return 0; }
 
     virtual ~UnmanagedRegion()
     {

test/message/_message.cxx

@@ -91,6 +91,31 @@ void Alignment(const string& transport)
     ASSERT_EQ(reinterpret_cast<uintptr_t>(msg->GetData()) % 64, 0);
 }
 
+void EmptyMessage(const string& transport, const string& _address)
+{
+    size_t session{fair::mq::tools::UuidHash()};
+    std::string address(fair::mq::tools::ToString(_address, "_", transport));
+
+    fair::mq::ProgOptions config;
+    config.SetProperty<string>("session", to_string(session));
+
+    auto factory = FairMQTransportFactory::CreateTransportFactory(transport, fair::mq::tools::Uuid(), &config);
+
+    FairMQChannel push{"Push", "push", factory};
+    push.Bind(address);
+
+    FairMQChannel pull{"Pull", "pull", factory};
+    pull.Connect(address);
+
+    FairMQMessagePtr outMsg(push.NewMessage());
+    ASSERT_EQ(outMsg->GetData(), nullptr);
+    ASSERT_EQ(push.Send(outMsg), 0);
+
+    FairMQMessagePtr inMsg(pull.NewMessage());
+    ASSERT_EQ(pull.Receive(inMsg), 0);
+    ASSERT_EQ(inMsg->GetData(), nullptr);
+}
+
 TEST(Resize, zeromq)
 {
     RunPushPullWithMsgResize("zeromq", "ipc://test_message_resize");
@@ -116,4 +141,14 @@ TEST(Alignment, shmem) // TODO: add test for ZeroMQ once it is implemented
     Alignment("shmem");
 }
 
+TEST(EmptyMessage, zeromq)
+{
+    EmptyMessage("zeromq", "ipc://test_empty_message");
+}
+
+TEST(EmptyMessage, shmem)
+{
+    EmptyMessage("shmem", "ipc://test_empty_message");
+}
+
 } // namespace

test/properties/_properties.cxx

@@ -70,6 +70,9 @@ TEST(ProgOptions, SetAndGet)
     set_and_get<vector<double>>(o, "_vector<double>", { 33.22, 33.23, 33.24 });
     set_and_get<vector<long double>>(o, "_vector<long double>", { 333.222, 333.223, 333.224 });
     set_and_get<vector<boost::filesystem::path>>(o, "_vector<boost::filesystem::path>", { boost::filesystem::path("C:\\Windows"), boost::filesystem::path("C:\\Windows\\System32") });
+
+    ASSERT_THROW(o.ParseAll({"cmd", "--unregistered", "option"}, false), boost::program_options::unknown_option);
+    ASSERT_NO_THROW(o.ParseAll({"cmd", "--unregistered", "option"}, true));
 }
 
 template<typename T>

test/protocols/_push_pull_multipart.cxx

@@ -163,9 +163,9 @@ TEST(PushPull, Multipart_ST_ipc_zeromq)
     RunSingleThreadedMultipart("zeromq", "ipc://test_Multipart_ST_ipc_zeromq_1", "ipc://test_Multipart_ST_ipc_zeromq_2");
 }
 
-TEST(PushPull, Multipart_ST_ipc_shmen)
+TEST(PushPull, Multipart_ST_ipc_shmem)
 {
-    RunSingleThreadedMultipart("shmem", "ipc://test_Multipart_ST_ipc_shmen_1", "ipc://test_Multipart_ST_ipc_shmen_2");
+    RunSingleThreadedMultipart("shmem", "ipc://test_Multipart_ST_ipc_shmem_1", "ipc://test_Multipart_ST_ipc_shmem_2");
 }
 
 TEST(PushPull, Multipart_MT_inproc_zeromq)

test/sdk/_topology.cxx

@@ -139,6 +139,32 @@ TEST_F(Topology, ChangeState)
     EXPECT_EQ(sdk::StateEqualsTo(currentState, sdk::DeviceState::InitializingDevice), true);
 }
 
+TEST_F(Topology, MixedState)
+{
+    using namespace fair::mq;
+
+    sdk::Topology topo(mDDSTopo, mDDSSession);
+    auto result1(topo.ChangeState(sdk::TopologyTransition::InitDevice, "main/Sampler.*"));
+    LOG(info) << result1.first;
+
+    EXPECT_EQ(result1.first, std::error_code());
+    EXPECT_EQ(sdk::AggregateState(result1.second), sdk::AggregatedTopologyState::Mixed);
+    EXPECT_EQ(sdk::StateEqualsTo(result1.second, sdk::DeviceState::InitializingDevice), false);
+    auto const currentState1 = topo.GetCurrentState();
+    EXPECT_EQ(sdk::AggregateState(currentState1), sdk::AggregatedTopologyState::Mixed);
+    EXPECT_EQ(sdk::StateEqualsTo(currentState1, sdk::DeviceState::InitializingDevice), false);
+
+    auto result2(topo.ChangeState(sdk::TopologyTransition::InitDevice, "main/SinkGroup/.*"));
+    LOG(info) << result2.first;
+
+    EXPECT_EQ(result2.first, std::error_code());
+    EXPECT_EQ(sdk::AggregateState(result2.second), sdk::AggregatedTopologyState::InitializingDevice);
+    EXPECT_EQ(sdk::StateEqualsTo(result2.second, sdk::DeviceState::InitializingDevice), true);
+    auto const currentState2 = topo.GetCurrentState();
+    EXPECT_EQ(sdk::AggregateState(currentState2), sdk::AggregatedTopologyState::InitializingDevice);
+    EXPECT_EQ(sdk::StateEqualsTo(currentState2, sdk::DeviceState::InitializingDevice), true);
+}
+
 TEST_F(Topology, AsyncChangeStateConcurrent)
 {
     using namespace fair::mq;
@@ -191,9 +217,9 @@ TEST_F(Topology, AsyncChangeStateCollectionView)
             ASSERT_EQ(cstate.size(), 5);
             for (const auto& c : cstate) {
                 LOG(debug) << "\t" << c.first;
-                State s;
+                sdk::AggregatedTopologyState s;
                 ASSERT_NO_THROW(s = sdk::AggregateState(c.second));
-                ASSERT_EQ(s, State::InitializingDevice);
+                ASSERT_EQ(s, static_cast<sdk::AggregatedTopologyState>(State::InitializingDevice));
                 LOG(debug) << "\tAggregated state: " << s;
                 for (const auto& ds : c.second) {
                     LOG(debug) << "\t\t" << ds.state;
@@ -426,4 +452,62 @@ TEST_F(Topology, SetAndGetProperties)
     ASSERT_EQ(topo.ChangeState(TopologyTransition::ResetDevice).first, std::error_code());
 }
 
+TEST(Topology2, AggregatedTopologyStateComparison)
+{
+    using namespace fair::mq::sdk;
+    ASSERT_TRUE(DeviceState::Undefined == AggregatedTopologyState::Undefined);
+    ASSERT_TRUE(AggregatedTopologyState::Undefined == DeviceState::Undefined);
+    ASSERT_TRUE(DeviceState::Ok == AggregatedTopologyState::Ok);
+    ASSERT_TRUE(DeviceState::Error == AggregatedTopologyState::Error);
+    ASSERT_TRUE(DeviceState::Idle == AggregatedTopologyState::Idle);
+    ASSERT_TRUE(DeviceState::InitializingDevice == AggregatedTopologyState::InitializingDevice);
+    ASSERT_TRUE(DeviceState::Initialized == AggregatedTopologyState::Initialized);
+    ASSERT_TRUE(DeviceState::Binding == AggregatedTopologyState::Binding);
+    ASSERT_TRUE(DeviceState::Bound == AggregatedTopologyState::Bound);
+    ASSERT_TRUE(DeviceState::Connecting == AggregatedTopologyState::Connecting);
+    ASSERT_TRUE(DeviceState::DeviceReady == AggregatedTopologyState::DeviceReady);
+    ASSERT_TRUE(DeviceState::InitializingTask == AggregatedTopologyState::InitializingTask);
+    ASSERT_TRUE(DeviceState::Ready == AggregatedTopologyState::Ready);
+    ASSERT_TRUE(DeviceState::Running == AggregatedTopologyState::Running);
+    ASSERT_TRUE(DeviceState::ResettingTask == AggregatedTopologyState::ResettingTask);
+    ASSERT_TRUE(DeviceState::ResettingDevice == AggregatedTopologyState::ResettingDevice);
+    ASSERT_TRUE(DeviceState::Exiting == AggregatedTopologyState::Exiting);
+
+    ASSERT_TRUE(GetAggregatedTopologyState("UNDEFINED") == AggregatedTopologyState::Undefined);
+    ASSERT_TRUE(GetAggregatedTopologyState("OK") == AggregatedTopologyState::Ok);
+    ASSERT_TRUE(GetAggregatedTopologyState("ERROR") == AggregatedTopologyState::Error);
+    ASSERT_TRUE(GetAggregatedTopologyState("IDLE") == AggregatedTopologyState::Idle);
+    ASSERT_TRUE(GetAggregatedTopologyState("INITIALIZING DEVICE") == AggregatedTopologyState::InitializingDevice);
+    ASSERT_TRUE(GetAggregatedTopologyState("INITIALIZED") == AggregatedTopologyState::Initialized);
+    ASSERT_TRUE(GetAggregatedTopologyState("BINDING") == AggregatedTopologyState::Binding);
+    ASSERT_TRUE(GetAggregatedTopologyState("BOUND") == AggregatedTopologyState::Bound);
+    ASSERT_TRUE(GetAggregatedTopologyState("CONNECTING") == AggregatedTopologyState::Connecting);
+    ASSERT_TRUE(GetAggregatedTopologyState("DEVICE READY") == AggregatedTopologyState::DeviceReady);
+    ASSERT_TRUE(GetAggregatedTopologyState("INITIALIZING TASK") == AggregatedTopologyState::InitializingTask);
+    ASSERT_TRUE(GetAggregatedTopologyState("READY") == AggregatedTopologyState::Ready);
+    ASSERT_TRUE(GetAggregatedTopologyState("RUNNING") == AggregatedTopologyState::Running);
+    ASSERT_TRUE(GetAggregatedTopologyState("RESETTING TASK") == AggregatedTopologyState::ResettingTask);
+    ASSERT_TRUE(GetAggregatedTopologyState("RESETTING DEVICE") == AggregatedTopologyState::ResettingDevice);
+    ASSERT_TRUE(GetAggregatedTopologyState("EXITING") == AggregatedTopologyState::Exiting);
+    ASSERT_TRUE(GetAggregatedTopologyState("MIXED") == AggregatedTopologyState::Mixed);
+
+    ASSERT_TRUE("UNDEFINED" == GetAggregatedTopologyStateName(AggregatedTopologyState::Undefined));
+    ASSERT_TRUE("OK" == GetAggregatedTopologyStateName(AggregatedTopologyState::Ok));
+    ASSERT_TRUE("ERROR" == GetAggregatedTopologyStateName(AggregatedTopologyState::Error));
+    ASSERT_TRUE("IDLE" == GetAggregatedTopologyStateName(AggregatedTopologyState::Idle));
+    ASSERT_TRUE("INITIALIZING DEVICE" == GetAggregatedTopologyStateName(AggregatedTopologyState::InitializingDevice));
+    ASSERT_TRUE("INITIALIZED" == GetAggregatedTopologyStateName(AggregatedTopologyState::Initialized));
+    ASSERT_TRUE("BINDING" == GetAggregatedTopologyStateName(AggregatedTopologyState::Binding));
+    ASSERT_TRUE("BOUND" == GetAggregatedTopologyStateName(AggregatedTopologyState::Bound));
+    ASSERT_TRUE("CONNECTING" == GetAggregatedTopologyStateName(AggregatedTopologyState::Connecting));
+    ASSERT_TRUE("DEVICE READY" == GetAggregatedTopologyStateName(AggregatedTopologyState::DeviceReady));
+    ASSERT_TRUE("INITIALIZING TASK" == GetAggregatedTopologyStateName(AggregatedTopologyState::InitializingTask));
+    ASSERT_TRUE("READY" == GetAggregatedTopologyStateName(AggregatedTopologyState::Ready));
+    ASSERT_TRUE("RUNNING" == GetAggregatedTopologyStateName(AggregatedTopologyState::Running));
+    ASSERT_TRUE("RESETTING TASK" == GetAggregatedTopologyStateName(AggregatedTopologyState::ResettingTask));
+    ASSERT_TRUE("RESETTING DEVICE" == GetAggregatedTopologyStateName(AggregatedTopologyState::ResettingDevice));
+    ASSERT_TRUE("EXITING" == GetAggregatedTopologyStateName(AggregatedTopologyState::Exiting));
+    ASSERT_TRUE("MIXED" == GetAggregatedTopologyStateName(AggregatedTopologyState::Mixed));
+}
+
 }   // namespace

test/transport/_options.cxx

@@ -66,6 +66,24 @@ void RunOptionsTest(const string& transport)
     ASSERT_EQ(channel.GetSocket().GetRcvKernelSize(), 8000);
 }
 
+void ZeroingAndMlock(const string& transport)
+{
+    size_t session{fair::mq::tools::UuidHash()};
+
+    fair::mq::ProgOptions config;
+    config.SetProperty<string>("session", to_string(session));
+    config.SetProperty<size_t>("shm-segment-size", 16384);
+    config.SetProperty<bool>("shm-zero-segment", true);
+    config.SetProperty<bool>("shm-mlock-segment", true);
+
+    auto factory = FairMQTransportFactory::CreateTransportFactory(transport, fair::mq::tools::Uuid(), &config);
+
+    FairMQMessagePtr outMsg(factory->CreateMessage(10000));
+    char test[10000];
+    memset(test, 0, sizeof(test));
+    ASSERT_EQ(memcmp(test, outMsg->GetData(), outMsg->GetSize()), 0);
+}
+
 TEST(Options, zeromq)
 {
     RunOptionsTest("zeromq");
@@ -76,4 +94,9 @@ TEST(Options, shmem)
     RunOptionsTest("shmem");
 }
 
+TEST(ZeroingAndMlock, shmem)
+{
+    ZeroingAndMlock("shmem");
+}
+
 } // namespace