DDS plugin: Remove unused dependency

CMake: Support running static analysis
OFF by default (Imposes significant additional build time). Controlled with CMake variable RUN_STATIC_ANALYSIS. Supported analysers: * clang-tidy * iwyu * cpplint
2025-10-16 01:51:45 +00:00 · 2019-06-13 17:47:47 +02:00 · 2019-06-04 12:17:37 +02:00 · 2019-06-04 12:17:37 +02:00 · 2019-06-01 06:20:39 +02:00 · 2019-05-14 14:56:08 +02:00
65 changed files with 1891 additions and 1294 deletions
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -7,13 +7,12 @@
 ################################################################################

 cmake_minimum_required(VERSION 3.10 FATAL_ERROR)
-
+cmake_policy(VERSION 3.10...3.14)

 # Project ######################################################################
 set(CMAKE_MODULE_PATH ${CMAKE_SOURCE_DIR}/cmake)
 include(FairMQLib)

-set_fairmq_cmake_policies()
 get_git_version()

 project(FairMQ VERSION ${PROJECT_VERSION} LANGUAGES CXX)
@@ -61,14 +60,12 @@ endif()

 if(BUILD_OFI_TRANSPORT)
  find_package2(PRIVATE asiofi REQUIRED
-    VERSION 0.2.0
+    VERSION 0.3.1
  )
  find_package2(PRIVATE OFI REQUIRED
    VERSION ${asiofi_OFI_VERSION}
    COMPONENTS ${asiofi_OFI_COMPONENTS}
  )
-  find_package2(PRIVATE AZMQ REQUIRED)
-  set(PROJECT_AZMQ_VERSION 1.0.2)
 endif()

 if(BUILD_NANOMSG_TRANSPORT)
@@ -194,11 +191,6 @@ if(BUILD_DDS_PLUGIN)
    DESTINATION ${PROJECT_INSTALL_CMAKEMODDIR}
  )
 endif()
-if(BUILD_OFI_TRANSPORT)
-  install(FILES cmake/FindAZMQ.cmake
-    DESTINATION ${PROJECT_INSTALL_CMAKEMODDIR}
-  )
-endif()
 if(BUILD_DOCS)
  install(DIRECTORY ${CMAKE_BINARY_DIR}/doxygen/html
    DESTINATION ${PROJECT_INSTALL_DATADIR}/docs
@@ -335,4 +327,26 @@ else()
 endif()
 message(STATUS "  ${BWhite}docs${CR}               ${docs_summary}")
 message(STATUS "  ")
+if(RUN_STATIC_ANALYSIS)
+  list(LENGTH PROJECT_STATIC_ANALYSERS size)
+  unset(analyser_list)
+  set(count 0)
+  foreach(analyser IN LISTS PROJECT_STATIC_ANALYSERS)
+    if(${analyser}_FOUND)
+      set(${analyser}_status "${analyser} ${BGreen}YES${CR}")
+    else()
+      set(${analyser}_status "${analyser} ${BRed}NO${CR}")
+    endif()
+    math(EXPR count "${count} + 1")
+    string(APPEND analyser_list "${${analyser}_status}")
+    if(count LESS size)
+      string(APPEND analyser_list "${BWhite},${CR} ")
+    endif()
+  endforeach()
+  set(static_ana_summary "${BWhite}(${CR}${analyser_list}${BWhite})${CR}  (disable with ${BMagenta}-DRUN_STATIC_ANALYSIS=OFF${CR})")
+else()
+  set(static_ana_summary "${BRed}OFF${CR}  (default, enable with ${BMagenta}-DRUN_STATIC_ANALYSIS=ON${CR})")
+endif()
+message(STATUS "  ${Cyan}RUN STATIC ANALYSIS  ${static_ana_summary}")
+message(STATUS "  ")
 ################################################################################
--- a/README.md
+++ b/README.md
@@ -38,7 +38,7 @@ a simulation, reconstruction and analysis framework.
  * PUBLIC: [**Boost**](https://www.boost.org/), [**FairLogger**](https://github.com/FairRootGroup/FairLogger)
  * BUILD: [CMake](https://cmake.org/), [GTest](https://github.com/google/googletest), [Doxygen](http://www.doxygen.org/)
  * PRIVATE: [ZeroMQ](http://zeromq.org/), [Msgpack](https://msgpack.org/index.html), [nanomsg](http://nanomsg.org/),
-[asiofi](https://github.com/FairRootGroup/asiofi), [DDS](http://dds.gsi.de), [PMIx](https://pmix.org/), [AZMQ](https://github.com/zeromq/azmq), [asiofi](https://github.com/FairRootGroup/asiofi)
+[asiofi](https://github.com/FairRootGroup/asiofi), [DDS](http://dds.gsi.de), [PMIx](https://pmix.org/)

  Supported platforms: Linux and MacOS.

@@ -51,7 +51,7 @@ cmake -DCMAKE_INSTALL_PREFIX=./fairmq_install ../fairmq
 cmake --build . --target install
 ```

-If dependencies are not installed in standard system directories, you can hint the installation location via `-DCMAKE_PREFIX_PATH=...` or per dependency via `-D{DEPENDENCY}_ROOT=...`. `{DEPENDENCY}` can be `GTEST`, `BOOST`, `FAIRLOGGER`, `ZEROMQ`, `MSGPACK`, `NANOMSG`, `OFI`, `PMIX`, `ASIOFI`, `AZMQ` or `DDS` (`*_ROOT` variables can also be environment variables).
+If dependencies are not installed in standard system directories, you can hint the installation location via `-DCMAKE_PREFIX_PATH=...` or per dependency via `-D{DEPENDENCY}_ROOT=...`. `{DEPENDENCY}` can be `GTEST`, `BOOST`, `FAIRLOGGER`, `ZEROMQ`, `MSGPACK`, `NANOMSG`, `OFI`, `PMIX`, `ASIOFI` or `DDS` (`*_ROOT` variables can also be environment variables).

 ## Usage

--- a/cmake/FairMQLib.cmake
+++ b/cmake/FairMQLib.cmake
@@ -29,26 +29,6 @@ if(NOT WIN32 AND NOT DISABLE_COLOR)
  set(BWhite   "${Esc}[1;37m")
 endif()

-# set_fairmq_cmake_policies()
-#
-# Sets CMake policies.
-macro(set_fairmq_cmake_policies)
-  # Find more details to each policy with cmake --help-policy CMPXXXX
-  foreach(policy
-    CMP0025 # Compiler id for Apple Clang is now AppleClang.
-    CMP0028 # Double colon in target name means ALIAS or IMPORTED target.
-    CMP0042 # MACOSX_RPATH is enabled by default.
-    CMP0048 # The ``project()`` command manages VERSION variables.
-    CMP0054 # Only interpret ``if()`` arguments as variables or keywords when unquoted.
-    CMP0074 # ``find_package()`` uses ``<PackageName>_ROOT`` variables.
-  )
-    if(POLICY ${policy})
-      cmake_policy(SET ${policy} NEW)
-    endif()
-  endforeach()
-endmacro()
-
-
 find_package(Git)
 # get_git_version([DEFAULT_VERSION version] [DEFAULT_DATE date] [OUTVAR_PREFIX prefix])
 #
@@ -186,6 +166,34 @@ macro(set_fairmq_defaults)
  else()
    set(PROJECT_VERSION_HOTFIX ${PROJECT_VERSION_TWEAK})
  endif()
+
+  if(NOT DEFINED RUN_STATIC_ANALYSIS)
+    set(RUN_STATIC_ANALYSIS OFF)
+  endif()
+
+  unset(PROJECT_STATIC_ANALYSERS)
+  if(RUN_STATIC_ANALYSIS)
+    set(analyser "clang-tidy")
+    find_program(${analyser}_FOUND "${analyser}")
+    if(${analyser}_FOUND)
+      set(CMAKE_CXX_CLANG_TIDY "${${analyser}_FOUND}" "-color")
+    endif()
+    list(APPEND PROJECT_STATIC_ANALYSERS "${analyser}")
+
+    set(analyser "iwyu")
+    find_program(${analyser}_FOUND "${analyser}")
+    if(${analyser}_FOUND)
+      set(CMAKE_CXX_IWYU "${${analyser}_FOUND}")
+    endif()
+    list(APPEND PROJECT_STATIC_ANALYSERS "${analyser}")
+
+    set(analyser "cpplint")
+    find_program(${analyser}_FOUND "${analyser}")
+    if(${analyser}_FOUND)
+      set(CMAKE_CXX_CPPLINT "${${analyser}_FOUND}")
+    endif()
+    list(APPEND PROJECT_STATIC_ANALYSERS "${analyser}")
+  endif()
 endmacro()

 function(join VALUES GLUE OUTPUT)
--- a/cmake/FindAZMQ.cmake
+++ b/cmake/FindAZMQ.cmake
@@ -1,57 +0,0 @@
-################################################################################
-#    Copyright (C) 2018 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"                       #
-################################################################################
-#
-# ###########################
-# # Locate the AZMQ library #
-# ###########################
-#
-#
-# Usage:
-#
-#   find_package(AZMQ [version] [QUIET] [REQUIRED])
-#
-#
-# Defines the following variables:
-#
-#   AZMQ_FOUND - Found the ZeroMQ library
-#   AZMQ_INCLUDE_DIR (CMake cache) - Include directory
-#
-#
-# Accepts the following variables as hints for installation directories:
-#
-#   AZMQ_ROOT (CMake var, ENV var)
-#
-#
-# If the above variables are not defined, or if ZeroMQ could not be found there,
-# it will look for it in the system directories. Custom ZeroMQ installations
-# will always have priority over system ones.
-#
-
-if(NOT AZMQ_ROOT)
-  set(AZMQ_ROOT $ENV{AZMQ_ROOT})
-endif()
-
-find_path(AZMQ_INCLUDE_DIR
-  NAMES azmq/socket.hpp
-  HINTS ${AZMQ_ROOT} $ENV{AZMQ_ROOT}
-  PATH_SUFFIXES include
-  DOC "AZMQ include directory"
-)
-
-include(FindPackageHandleStandardArgs)
-find_package_handle_standard_args(AZMQ
-    REQUIRED_VARS AZMQ_INCLUDE_DIR
-)
-
-if(AZMQ_FOUND AND NOT TARGET AZMQ::AZMQ)
-  add_library(AZMQ::AZMQ INTERFACE IMPORTED)
-  set_target_properties(AZMQ::AZMQ PROPERTIES
-    INTERFACE_INCLUDE_DIRECTORIES ${AZMQ_INCLUDE_DIR}
-    INTERFACE_LINK_LIBRARIES "libzmq;Boost::boost;Boost::container;Boost::system"
-  )
-endif()
--- a/docs/Device.md
+++ b/docs/Device.md
@@ -24,7 +24,7 @@ FairMQ devices communicate via the communication patterns offered by ZeroMQ (or

 Each FairMQ device has an internal state machine:

-![FairMQ state machine](images/device_states.png?raw=true "FairMQ state machine")
+![FairMQ state machine](images/device_states.svg "FairMQ state machine")

 The state machine can be querried and controlled via `GetCurrentStateName()` and `ChangeState("<state name>")` methods. Only legal state transitions are allowed (see image above). Illegal transitions will fail with an error.

--- a/docs/images/device_states.png
+++ b/docs/images/device_states.png
--- a/docs/images/device_states.svg
+++ b/docs/images/device_states.svg
--- a/examples/multipart/CMakeLists.txt
+++ b/examples/multipart/CMakeLists.txt
@@ -32,15 +32,20 @@ configure_file(${CMAKE_CURRENT_SOURCE_DIR}/fairmq-start-ex-multipart.sh.in ${CMA
 configure_file(${CMAKE_CURRENT_SOURCE_DIR}/test-ex-multipart.sh.in ${CMAKE_CURRENT_BINARY_DIR}/test-ex-multipart.sh)

 add_test(NAME Example.Multipart.zeromq COMMAND ${CMAKE_CURRENT_BINARY_DIR}/test-ex-multipart.sh zeromq)
-set_tests_properties(Example.Multipart.zeromq PROPERTIES TIMEOUT "30" RUN_SERIAL true PASS_REGULAR_EXPRESSION "Received message with 2 parts")
+set_tests_properties(Example.Multipart.zeromq PROPERTIES TIMEOUT "30" RUN_SERIAL true PASS_REGULAR_EXPRESSION "Received message with 5 parts")

 if(BUILD_NANOMSG_TRANSPORT)
  add_test(NAME Example.Multipart.nanomsg COMMAND ${CMAKE_CURRENT_BINARY_DIR}/test-ex-multipart.sh nanomsg)
-  set_tests_properties(Example.Multipart.nanomsg PROPERTIES TIMEOUT "30" RUN_SERIAL true PASS_REGULAR_EXPRESSION "Received message with 2 parts")
+  set_tests_properties(Example.Multipart.nanomsg PROPERTIES TIMEOUT "30" RUN_SERIAL true PASS_REGULAR_EXPRESSION "Received message with 5 parts")
 endif()

 add_test(NAME Example.Multipart.shmem COMMAND ${CMAKE_CURRENT_BINARY_DIR}/test-ex-multipart.sh shmem)
-set_tests_properties(Example.Multipart.shmem PROPERTIES TIMEOUT "30" RUN_SERIAL true PASS_REGULAR_EXPRESSION "Received message with 2 parts")
+set_tests_properties(Example.Multipart.shmem PROPERTIES TIMEOUT "30" RUN_SERIAL true PASS_REGULAR_EXPRESSION "Received message with 5 parts")
+
+if(BUILD_OFI_TRANSPORT)
+  add_test(NAME Example.Multipart.ofi COMMAND ${CMAKE_CURRENT_BINARY_DIR}/test-ex-multipart.sh ofi)
+  set_tests_properties(Example.Multipart.ofi PROPERTIES TIMEOUT "30" RUN_SERIAL true PASS_REGULAR_EXPRESSION "Received message with 5 parts")
+endif()

 # install

--- a/examples/multipart/Sampler.cxx
+++ b/examples/multipart/Sampler.cxx
@@ -53,6 +53,13 @@ bool Sampler::ConditionalRun()
    parts.AddPart(NewSimpleMessage(header));
    parts.AddPart(NewMessage(1000));

+    // create more data parts, testing the FairMQParts in-place constructor
+    FairMQParts auxData{ NewMessage(500), NewMessage(600), NewMessage(700) };
+    assert(auxData.Size() == 3);
+    parts.AddPart(std::move(auxData));
+    assert(auxData.Size() == 0);
+    assert(parts.Size() == 5);
+
    LOG(info) << "Sending body of size: " << parts.At(1)->GetSize();

    Send(parts, "data");
@@ -74,4 +81,4 @@ Sampler::~Sampler()
 {
 }

-} // namespace example_multipart
+} // namespace example_multipart
--- a/examples/multipart/test-ex-multipart.sh.in
+++ b/examples/multipart/test-ex-multipart.sh.in
@@ -20,7 +20,7 @@ SAMPLER+=" --verbosity veryhigh"
 SAMPLER+=" --session $SESSION"
 SAMPLER+=" --max-iterations 1"
 SAMPLER+=" --control static --color false"
-SAMPLER+=" --channel-config name=data,type=push,method=connect,rateLogging=0,address=tcp://127.0.0.1:5555"
+SAMPLER+=" --channel-config name=data,type=pair,method=connect,rateLogging=0,address=tcp://127.0.0.1:5555"
@CMAKE_CURRENT_BINARY_DIR@/$SAMPLER &
 SAMPLER_PID=$!

@@ -30,7 +30,7 @@ SINK+=" --transport $transport"
 SINK+=" --verbosity veryhigh"
 SINK+=" --session $SESSION"
 SINK+=" --control static --color false"
-SINK+=" --channel-config name=data,type=pull,method=bind,rateLogging=0,address=tcp://127.0.0.1:5555"
+SINK+=" --channel-config name=data,type=pair,method=bind,rateLogging=0,address=tcp://127.0.0.1:5555"
@CMAKE_CURRENT_BINARY_DIR@/$SINK &
 SINK_PID=$!

--- a/examples/readout/Builder.h
+++ b/examples/readout/Builder.h
@@ -5,36 +5,42 @@
 *              GNU Lesser General Public Licence (LGPL) version 3,             *
 *                  copied verbatim in the file "LICENSE"                       *
 ********************************************************************************/
-
 #ifndef FAIRMQEXAMPLEREGIONBUILDER_H
 #define FAIRMQEXAMPLEREGIONBUILDER_H

-#include <atomic>
-
 #include "FairMQDevice.h"

-namespace example_region
+#include <string>
+
+namespace example_readout
 {

 class Builder : public FairMQDevice
 {
  public:
-    Builder() {
-        OnData("data1", &Builder::HandleData);
-    }
-    virtual ~Builder() {}
+    Builder()
+        : fOutputChannelName()
+    {}
+
+    void Init() override
+    {
+        fOutputChannelName = fConfig->GetValue<std::string>("output-name");
+        OnData("rb", &Builder::HandleData);
+    }

-  protected:
    bool HandleData(FairMQMessagePtr& msg, int /*index*/)
    {
-        if (Send(msg, "data2") < 0) {
+        if (Send(msg, fOutputChannelName) < 0) {
            return false;
        }

        return true;
    }
+
+  private:
+    std::string fOutputChannelName;
 };

-} // namespace example_region
+} // namespace example_readout

 #endif /* FAIRMQEXAMPLEREGIONBUILDER_H */
--- a/examples/readout/CMakeLists.txt
+++ b/examples/readout/CMakeLists.txt
@@ -6,37 +6,35 @@
 #                  copied verbatim in the file "LICENSE"                       #
 ################################################################################

-add_library(ExampleReadoutLib STATIC
-    "Sampler.cxx"
-    "Sampler.h"
-    "Builder.h"
-    "Sink.cxx"
-    "Sink.h"
-)
-
-target_link_libraries(ExampleReadoutLib PUBLIC FairMQ)
-
-add_executable(fairmq-ex-readout-sampler runSampler.cxx)
-target_link_libraries(fairmq-ex-readout-sampler PRIVATE ExampleReadoutLib)
+add_executable(fairmq-ex-readout-readout runReadout.cxx)
+target_link_libraries(fairmq-ex-readout-readout PRIVATE FairMQ)

 add_executable(fairmq-ex-readout-builder runBuilder.cxx)
-target_link_libraries(fairmq-ex-readout-builder PRIVATE ExampleReadoutLib)
+target_link_libraries(fairmq-ex-readout-builder PRIVATE FairMQ)

-add_executable(fairmq-ex-readout-sink runSink.cxx)
-target_link_libraries(fairmq-ex-readout-sink PRIVATE ExampleReadoutLib)
+add_executable(fairmq-ex-readout-processor runProcessor.cxx)
+target_link_libraries(fairmq-ex-readout-processor PRIVATE FairMQ)

-add_custom_target(Examplereadout DEPENDS fairmq-ex-readout-sampler fairmq-ex-readout-sink)
+add_executable(fairmq-ex-readout-sender runSender.cxx)
+target_link_libraries(fairmq-ex-readout-sender PRIVATE FairMQ)
+
+add_executable(fairmq-ex-readout-receiver runReceiver.cxx)
+target_link_libraries(fairmq-ex-readout-receiver PRIVATE FairMQ)

 set(EX_BIN_DIR ${CMAKE_CURRENT_BINARY_DIR})
 set(FAIRMQ_BIN_DIR ${CMAKE_BINARY_DIR}/fairmq)
 configure_file(${CMAKE_CURRENT_SOURCE_DIR}/fairmq-start-ex-readout.sh.in ${CMAKE_CURRENT_BINARY_DIR}/fairmq-start-ex-readout.sh)
+configure_file(${CMAKE_CURRENT_SOURCE_DIR}/fairmq-start-ex-readout-processing.sh.in ${CMAKE_CURRENT_BINARY_DIR}/fairmq-start-ex-readout-processing.sh)

 # install

 install(
    TARGETS
-    fairmq-ex-readout-sampler
-    fairmq-ex-readout-sink
+    fairmq-ex-readout-readout
+    fairmq-ex-readout-builder
+    fairmq-ex-readout-processor
+    fairmq-ex-readout-sender
+    fairmq-ex-readout-receiver

    LIBRARY DESTINATION ${PROJECT_INSTALL_LIBDIR}
    RUNTIME DESTINATION ${PROJECT_INSTALL_BINDIR}
@@ -46,9 +44,16 @@ install(
 set(EX_BIN_DIR ${CMAKE_INSTALL_PREFIX}/${PROJECT_INSTALL_BINDIR})
 set(FAIRMQ_BIN_DIR ${CMAKE_INSTALL_PREFIX}/${PROJECT_INSTALL_BINDIR}/fairmq)
 configure_file(${CMAKE_CURRENT_SOURCE_DIR}/fairmq-start-ex-readout.sh.in ${CMAKE_CURRENT_BINARY_DIR}/fairmq-start-ex-readout.sh_install)
+configure_file(${CMAKE_CURRENT_SOURCE_DIR}/fairmq-start-ex-readout-processing.sh.in ${CMAKE_CURRENT_BINARY_DIR}/fairmq-start-ex-readout-processing.sh_install)

 install(
    PROGRAMS ${CMAKE_CURRENT_BINARY_DIR}/fairmq-start-ex-readout.sh_install
    DESTINATION ${PROJECT_INSTALL_BINDIR}
    RENAME fairmq-start-ex-readout.sh
 )
+
+install(
+    PROGRAMS ${CMAKE_CURRENT_BINARY_DIR}/fairmq-start-ex-readout-processing.sh_install
+    DESTINATION ${PROJECT_INSTALL_BINDIR}
+    RENAME fairmq-start-ex-readout-processing.sh
+)
--- a/examples/readout/Processor.h
+++ b/examples/readout/Processor.h
@@ -5,38 +5,33 @@
 *              GNU Lesser General Public Licence (LGPL) version 3,             *
 *                  copied verbatim in the file "LICENSE"                       *
 ********************************************************************************/
-/**
- * Sink.h
- *
- * @since 2014-10-10
- * @author A. Rybalchenko
- */
-
-#ifndef FAIRMQEXAMPLEREGIONSINK_H
-#define FAIRMQEXAMPLEREGIONSINK_H
-
-#include <string>
+#ifndef FAIRMQEXAMPLEREGIONPROCESSOR_H
+#define FAIRMQEXAMPLEREGIONPROCESSOR_H

 #include "FairMQDevice.h"

-namespace example_region
+namespace example_readout
 {

-class Sink : public FairMQDevice
+class Processor : public FairMQDevice
 {
  public:
-    Sink();
-    virtual ~Sink();
+    Processor() {
+        OnData("bp", &Processor::HandleData);
+    }

  protected:
-    virtual void Run();
-    virtual void InitTask();
+    bool HandleData(FairMQMessagePtr& msg, int /*index*/)
+    {
+        FairMQMessagePtr msg2(NewMessageFor("ps", 0, msg->GetSize()));
+        if (Send(msg2, "ps") < 0) {
+            return false;
+        }

-  private:
-    uint64_t fMaxIterations;
-    uint64_t fNumIterations;
+        return true;
+    }
 };

-} // namespace example_region
+} // namespace example_readout

-#endif /* FAIRMQEXAMPLEREGIONSINK_H */
+#endif /* FAIRMQEXAMPLEREGIONPROCESSOR_H */
--- a/examples/readout/README.md
+++ b/examples/readout/README.md
@@ -1,5 +1,27 @@
-Region example
-==============
+# Readout example

-This example demonstrates the use of a more advanced feature - UnmanagedRegion, that can be used to create a buffer through one of FairMQ transports. The contents of this buffer are managed by the user, who can also create messages out of sub-buffers of the created buffer. Such feature can be interesting in environments that have special requirements by the hardware that writes the data, to keep the transfer efficient (e.g. shared memory).
+This examples shows two possible topologies (out of many) for a node connected to a detector readout (followed by a processing node).

+## Setup without new data generation
+
+```
+|------------------------------- Readout Node ---------------------------|     |- Processing Node -|
+| Readout            -->            Builder          -->          Sender | --> |     Receiver      |
+| [# shared memory segment (unused in this topology) ##################] | ofi |                   |
+| [# shmem unmanaged region (readout writes here, others read) ########] |     |                   |
+|------------------------------------------------------------------------|     |-------------------|
+```
+
+The devices one the Readout Node communicate via shared memory transport. Readout device writes into shared memory unmanaged region. The data is then forwarded through Builder to Sender process, which sends it out via OFI transport.
+
+## Setup with generating new data on the Readout node
+
+```
+|------------------------------- Readout Node ---------------------------|     |- Processing Node -|
+| Readout     -->     Builder      -->      Processor     -->     Sender | --> |     Receiver      |
+| [# shared memory segment (used between Proccessor and Sender) #######] | ofi |                   |
+| [# shmem unmanaged region (readout writes here, builder & proc read) ] |     |                   |
+|------------------------------------------------------------------------|     |-------------------|
+```
+
+In this topology one more device is added - Processor. It examines the arriving data and creates new data in shared memory. This data is not part of the unmanaged region, but lives in the general shared memory segment (unused in the previous setup). This new data is then forwarded to Sender and the Readout device is notified that the corresponding data piece in the unmanaged region is no longer used.
--- a/examples/readout/Readout.h
+++ b/examples/readout/Readout.h
@@ -0,0 +1,91 @@
+/********************************************************************************
+ *    Copyright (C) 2014 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"                       *
+ ********************************************************************************/
+#ifndef FAIRMQEXAMPLEREADOUTREADOUT_H
+#define FAIRMQEXAMPLEREADOUTREADOUT_H
+
+#include <atomic>
+#include <thread>
+#include <chrono>
+
+#include "FairMQDevice.h"
+
+namespace example_readout
+{
+
+class Readout : public FairMQDevice
+{
+  public:
+    Readout()
+        : fMsgSize(10000)
+        , fMaxIterations(0)
+        , fNumIterations(0)
+        , fRegion(nullptr)
+        , fNumUnackedMsgs(0)
+    {}
+
+  protected:
+    void InitTask() override
+    {
+        fMsgSize = fConfig->GetValue<int>("msg-size");
+        fMaxIterations = fConfig->GetValue<uint64_t>("max-iterations");
+
+        fRegion = FairMQUnmanagedRegionPtr(NewUnmanagedRegionFor("rb",
+                                                                0,
+                                                                10000000,
+                                                                [this](void* /*data*/, size_t /*size*/, void* /*hint*/) { // callback to be called when message buffers no longer needed by transport
+                                                                    --fNumUnackedMsgs;
+                                                                    if (fMaxIterations > 0) {
+                                                                        LOG(debug) << "Received ack";
+                                                                    }
+                                                                }
+                                                                ));
+    }
+
+    bool ConditionalRun() override
+    {
+        FairMQMessagePtr msg(NewMessageFor("rb", // channel
+                                            0, // sub-channel
+                                            fRegion, // region
+                                            fRegion->GetData(), // ptr within region
+                                            fMsgSize, // offset from ptr
+                                            nullptr // hint
+                                            ));
+
+        if (Send(msg, "rb", 0) > 0) {
+            ++fNumUnackedMsgs;
+
+            if (fMaxIterations > 0 && ++fNumIterations >= fMaxIterations) {
+                LOG(info) << "Configured maximum number of iterations reached. Leaving RUNNING state.";
+                return false;
+            }
+        }
+
+        return true;
+    }
+    void ResetTask() override
+    {
+        // 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 << ")";
+            std::this_thread::sleep_for(std::chrono::milliseconds(500));
+            LOG(debug) << "done, still unacked: " << fNumUnackedMsgs;
+        }
+        fRegion.reset();
+    }
+
+  private:
+    int fMsgSize;
+    uint64_t fMaxIterations;
+    uint64_t fNumIterations;
+    FairMQUnmanagedRegionPtr fRegion;
+    std::atomic<uint64_t> fNumUnackedMsgs;
+};
+
+} // namespace example_readout
+
+#endif /* FAIRMQEXAMPLEREADOUTREADOUT_H */
--- a/examples/readout/Receiver.h
+++ b/examples/readout/Receiver.h
@@ -0,0 +1,54 @@
+/********************************************************************************
+ *    Copyright (C) 2014 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"                       *
+ ********************************************************************************/
+#ifndef FAIRMQEXAMPLEREGIONRECEIVER_H
+#define FAIRMQEXAMPLEREGIONRECEIVER_H
+
+#include "FairMQDevice.h"
+
+namespace example_readout
+{
+
+class Receiver : public FairMQDevice
+{
+  public:
+    Receiver()
+        : fMaxIterations(0)
+        , fNumIterations(0)
+    {}
+
+  protected:
+    void InitTask() override
+    {
+        // Get the fMaxIterations value from the command line options (via fConfig)
+        fMaxIterations = fConfig->GetValue<uint64_t>("max-iterations");
+    }
+
+    void Run() override
+    {
+        FairMQChannel& dataInChannel = fChannels.at("sr").at(0);
+
+        while (!NewStatePending()) {
+            FairMQMessagePtr msg(dataInChannel.Transport()->CreateMessage());
+            dataInChannel.Receive(msg);
+            // void* ptr = msg->GetData();
+
+            if (fMaxIterations > 0 && ++fNumIterations >= fMaxIterations) {
+                LOG(info) << "Configured maximum number of iterations reached. Leaving RUNNING state.";
+                break;
+            }
+        }
+    }
+
+  private:
+    uint64_t fMaxIterations;
+    uint64_t fNumIterations;
+};
+
+} // namespace example_readout
+
+#endif /* FAIRMQEXAMPLEREGIONRECEIVER_H */
--- a/examples/readout/Sampler.cxx
+++ b/examples/readout/Sampler.cxx
@@ -1,91 +0,0 @@
-/********************************************************************************
- *    Copyright (C) 2014 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"                       *
- ********************************************************************************/
-/**
- * Sampler.cpp
- *
- * @since 2014-10-10
- * @author A. Rybalchenko
- */
-
-#include "Sampler.h"
-
-#include <thread>
-
-using namespace std;
-
-namespace example_region
-{
-
-Sampler::Sampler()
-    : fMsgSize(10000)
-    , fMaxIterations(0)
-    , fNumIterations(0)
-    , fRegion(nullptr)
-    , fNumUnackedMsgs(0)
-{
-}
-
-void Sampler::InitTask()
-{
-    fMsgSize = fConfig->GetValue<int>("msg-size");
-    fMaxIterations = fConfig->GetValue<uint64_t>("max-iterations");
-
-    fRegion = FairMQUnmanagedRegionPtr(NewUnmanagedRegionFor("data1",
-                                                             0,
-                                                             10000000,
-                                                             [this](void* /*data*/, size_t /*size*/, void* /*hint*/) { // callback to be called when message buffers no longer needed by transport
-                                                                 --fNumUnackedMsgs;
-                                                                 if (fMaxIterations > 0)
-                                                                 {
-                                                                     LOG(debug) << "Received ack";
-                                                                 }
-                                                             }
-                                                             ));
-}
-
-bool Sampler::ConditionalRun()
-{
-    FairMQMessagePtr msg(NewMessageFor("data1", // channel
-                                        0, // sub-channel
-                                        fRegion, // region
-                                        fRegion->GetData(), // ptr within region
-                                        fMsgSize, // offset from ptr
-                                        nullptr // hint
-                                        ));
-
-    if (Send(msg, "data1", 0) > 0)
-    {
-        ++fNumUnackedMsgs;
-
-        if (fMaxIterations > 0 && ++fNumIterations >= fMaxIterations)
-        {
-            LOG(info) << "Configured maximum number of iterations reached. Leaving RUNNING state.";
-            return false;
-        }
-    }
-
-    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;
-    }
-    fRegion.reset();
-}
-
-Sampler::~Sampler()
-{
-}
-
-} // namespace example_region
--- a/examples/readout/Sampler.h
+++ b/examples/readout/Sampler.h
@@ -5,42 +5,42 @@
 *              GNU Lesser General Public Licence (LGPL) version 3,             *
 *                  copied verbatim in the file "LICENSE"                       *
 ********************************************************************************/
-/**
- * Sampler.h
- *
- * @since 2014-10-10
- * @author A. Rybalchenko
- */
-
-#ifndef FAIRMQEXAMPLEREGIONSAMPLER_H
-#define FAIRMQEXAMPLEREGIONSAMPLER_H
-
-#include <atomic>
+#ifndef FAIRMQEXAMPLEREGIONSENDER_H
+#define FAIRMQEXAMPLEREGIONSENDER_H

 #include "FairMQDevice.h"

-namespace example_region
+#include <string>
+
+namespace example_readout
 {

-class Sampler : public FairMQDevice
+class Sender : public FairMQDevice
 {
  public:
-    Sampler();
-    virtual ~Sampler();
+    Sender()
+        : fInputChannelName()
+    {}

-  protected:
-    virtual void InitTask();
-    virtual bool ConditionalRun();
-    virtual void ResetTask();
+    void Init() override
+    {
+        fInputChannelName = fConfig->GetValue<std::string>("input-name");
+        OnData(fInputChannelName, &Sender::HandleData);
+    }
+
+    bool HandleData(FairMQMessagePtr& msg, int /*index*/)
+    {
+        if (Send(msg, "sr") < 0) {
+            return false;
+        }
+
+        return true;
+    }

  private:
-    int fMsgSize;
-    uint64_t fMaxIterations;
-    uint64_t fNumIterations;
-    FairMQUnmanagedRegionPtr fRegion;
-    std::atomic<uint64_t> fNumUnackedMsgs;
+    std::string fInputChannelName;
 };

-} // namespace example_region
+} // namespace example_readout

-#endif /* FAIRMQEXAMPLEREGIONSAMPLER_H */
+#endif /* FAIRMQEXAMPLEREGIONSENDER_H */
--- a/examples/readout/Sink.cxx
+++ b/examples/readout/Sink.cxx
@@ -1,56 +0,0 @@
-/********************************************************************************
- *    Copyright (C) 2014 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"                       *
- ********************************************************************************/
-/**
- * Sink.cxx
- *
- * @since 2014-10-10
- * @author A. Rybalchenko
- */
-
-#include "Sink.h"
-
-using namespace std;
-
-namespace example_region
-{
-
-Sink::Sink()
-    : fMaxIterations(0)
-    , fNumIterations(0)
-{
-}
-
-void Sink::InitTask()
-{
-    // Get the fMaxIterations value from the command line options (via fConfig)
-    fMaxIterations = fConfig->GetValue<uint64_t>("max-iterations");
-}
-
-void Sink::Run()
-{
-    FairMQChannel& dataInChannel = fChannels.at("data").at(0);
-
-    while (!NewStatePending())
-    {
-        FairMQMessagePtr msg(dataInChannel.Transport()->CreateMessage());
-        dataInChannel.Receive(msg);
-        // void* ptr = msg->GetData();
-
-        if (fMaxIterations > 0 && ++fNumIterations >= fMaxIterations)
-        {
-            LOG(info) << "Configured maximum number of iterations reached. Leaving RUNNING state.";
-            break;
-        }
-    }
-}
-
-Sink::~Sink()
-{
-}
-
-} // namespace example_region
--- a/examples/readout/fairmq-start-ex-readout-processing.sh.in
+++ b/examples/readout/fairmq-start-ex-readout-processing.sh.in
@@ -0,0 +1,40 @@
+#!/bin/bash
+
+export FAIRMQ_PATH=@FAIRMQ_BIN_DIR@
+
+msgSize="1000000"
+
+if [[ $1 =~ ^[0-9]+$ ]]; then
+    msgSize=$1
+fi
+
+READOUT="fairmq-ex-readout-readout"
+READOUT+=" --id readout1"
+READOUT+=" --msg-size $msgSize"
+READOUT+=" --channel-config name=rb,type=pair,method=bind,address=tcp://localhost:7777,transport=shmem"
+xterm -geometry 80x23+0+0 -hold -e @EX_BIN_DIR@/$READOUT &
+
+BUILDER="fairmq-ex-readout-builder"
+BUILDER+=" --id builder1"
+BUILDER+=" --output-name bp"
+BUILDER+=" --channel-config name=rb,type=pair,method=connect,address=tcp://localhost:7777,transport=shmem"
+BUILDER+="                  name=bp,type=pair,method=connect,address=tcp://localhost:7778,transport=shmem"
+xterm -geometry 80x23+500+0 -hold -e @EX_BIN_DIR@/$BUILDER &
+
+PROCESSOR="fairmq-ex-readout-processor"
+PROCESSOR+=" --id processor1"
+PROCESSOR+=" --channel-config name=bp,type=pair,method=bind,address=tcp://localhost:7778,transport=shmem"
+PROCESSOR+="                  name=ps,type=pair,method=connect,address=tcp://localhost:7779,transport=shmem"
+xterm -geometry 80x23+750+500 -hold -e @EX_BIN_DIR@/$PROCESSOR &
+
+SENDER="fairmq-ex-readout-sender"
+SENDER+=" --id sender1"
+SENDER+=" --input-name ps"
+SENDER+=" --channel-config name=ps,type=pair,method=bind,address=tcp://localhost:7779,transport=shmem"
+SENDER+="                  name=sr,type=pair,method=connect,address=tcp://localhost:7780,transport=ofi"
+xterm -geometry 80x23+1000+0 -hold -e @EX_BIN_DIR@/$SENDER &
+
+RECEIVER="fairmq-ex-readout-receiver"
+RECEIVER+=" --id receiver1"
+RECEIVER+=" --channel-config name=sr,type=pair,method=bind,address=tcp://localhost:7780,transport=ofi"
+xterm -geometry 80x23+1500+0 -hold -e @EX_BIN_DIR@/$RECEIVER &
--- a/examples/readout/fairmq-start-ex-readout.sh.in
+++ b/examples/readout/fairmq-start-ex-readout.sh.in
@@ -8,23 +8,27 @@ if [[ $1 =~ ^[0-9]+$ ]]; then
    msgSize=$1
 fi

-SAMPLER="fairmq-ex-readout-sampler"
-SAMPLER+=" --id sampler1"
-SAMPLER+=" --severity debug"
-SAMPLER+=" --msg-size $msgSize"
-# SAMPLER+=" --rate 10"
-SAMPLER+=" --channel-config name=data1,type=pair,method=bind,address=tcp://127.0.0.1:7777,transport=shmem"
-xterm -geometry 80x23+0+0 -hold -e @EX_BIN_DIR@/$SAMPLER &
+READOUT="fairmq-ex-readout-readout"
+READOUT+=" --id readout1"
+READOUT+=" --msg-size $msgSize"
+READOUT+=" --channel-config name=rb,type=pair,method=bind,address=tcp://localhost:7777,transport=shmem"
+xterm -geometry 80x23+0+0 -hold -e @EX_BIN_DIR@/$READOUT &

 BUILDER="fairmq-ex-readout-builder"
 BUILDER+=" --id builder1"
-BUILDER+=" --severity debug"
-BUILDER+=" --channel-config name=data1,type=pair,method=connect,address=tcp://127.0.0.1:7777,transport=shmem"
-BUILDER+="                  name=data2,type=pair,method=connect,address=tcp://127.0.0.1:7778,transport=ofi"
+BUILDER+=" --output-name bs"
+BUILDER+=" --channel-config name=rb,type=pair,method=connect,address=tcp://localhost:7777,transport=shmem"
+BUILDER+="                  name=bs,type=pair,method=connect,address=tcp://localhost:7778,transport=shmem"
 xterm -geometry 80x23+500+0 -hold -e @EX_BIN_DIR@/$BUILDER &

-SINK="fairmq-ex-readout-sink"
-SINK+=" --id sink1"
-SINK+=" --severity debug"
-SINK+=" --channel-config name=data,type=pair,method=bind,address=tcp://127.0.0.1:7778,transport=ofi"
-xterm -geometry 80x23+1000+0 -hold -e @EX_BIN_DIR@/$SINK &
+SENDER="fairmq-ex-readout-sender"
+SENDER+=" --id sender1"
+SENDER+=" --input-name bs"
+SENDER+=" --channel-config name=bs,type=pair,method=bind,address=tcp://localhost:7778,transport=shmem"
+SENDER+="                  name=sr,type=pair,method=connect,address=tcp://localhost:7779,transport=ofi"
+xterm -geometry 80x23+1000+0 -hold -e @EX_BIN_DIR@/$SENDER &
+
+RECEIVER="fairmq-ex-readout-receiver"
+RECEIVER+=" --id receiver1"
+RECEIVER+=" --channel-config name=sr,type=pair,method=bind,address=tcp://localhost:7779,transport=ofi"
+xterm -geometry 80x23+1500+0 -hold -e @EX_BIN_DIR@/$RECEIVER &
--- a/examples/readout/runBuilder.cxx
+++ b/examples/readout/runBuilder.cxx
@@ -11,10 +11,13 @@

 namespace bpo = boost::program_options;

-void addCustomOptions(bpo::options_description& /* options */)
-{}
+void addCustomOptions(bpo::options_description& options)
+{
+    options.add_options()
+        ("output-name", bpo::value<std::string>()->default_value("bs"), "Output channel name");
+}

 FairMQDevicePtr getDevice(const FairMQProgOptions& /*config*/)
 {
-    return new example_region::Builder();
+    return new example_readout::Builder();
 }
--- a/examples/readout/runProcessor.cxx
+++ b/examples/readout/runProcessor.cxx
@@ -0,0 +1,20 @@
+/********************************************************************************
+ *    Copyright (C) 2014 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"                       *
+ ********************************************************************************/
+
+#include "runFairMQDevice.h"
+#include "Processor.h"
+
+namespace bpo = boost::program_options;
+
+void addCustomOptions(bpo::options_description& /* options */)
+{}
+
+FairMQDevicePtr getDevice(const FairMQProgOptions& /*config*/)
+{
+    return new example_readout::Processor();
+}
--- a/examples/readout/runReadout.cxx
+++ b/examples/readout/runReadout.cxx
@@ -7,7 +7,7 @@
 ********************************************************************************/

 #include "runFairMQDevice.h"
-#include "Sampler.h"
+#include "Readout.h"

 namespace bpo = boost::program_options;

@@ -20,5 +20,5 @@ void addCustomOptions(bpo::options_description& options)

 FairMQDevicePtr getDevice(const FairMQProgOptions& /*config*/)
 {
-    return new example_region::Sampler();
+    return new example_readout::Readout();
 }
--- a/examples/readout/runReceiver.cxx
+++ b/examples/readout/runReceiver.cxx
@@ -7,7 +7,7 @@
 ********************************************************************************/

 #include "runFairMQDevice.h"
-#include "Sink.h"
+#include "Receiver.h"

 namespace bpo = boost::program_options;

@@ -19,5 +19,5 @@ void addCustomOptions(bpo::options_description& options)

 FairMQDevicePtr getDevice(const FairMQProgOptions& /*config*/)
 {
-    return new example_region::Sink();
+    return new example_readout::Receiver();
 }
--- a/examples/readout/runSender.cxx
+++ b/examples/readout/runSender.cxx
@@ -0,0 +1,23 @@
+/********************************************************************************
+ *    Copyright (C) 2014 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"                       *
+ ********************************************************************************/
+
+#include "runFairMQDevice.h"
+#include "Sender.h"
+
+namespace bpo = boost::program_options;
+
+void addCustomOptions(bpo::options_description& options)
+{
+    options.add_options()
+        ("input-name", bpo::value<std::string>()->default_value("bs"), "Input channel name");
+}
+
+FairMQDevicePtr getDevice(const FairMQProgOptions& /*config*/)
+{
+    return new example_readout::Sender();
+}
--- a/fairmq/CMakeLists.txt
+++ b/fairmq/CMakeLists.txt
@@ -224,7 +224,6 @@ if(BUILD_OFI_TRANSPORT)
  set(OFI_DEPS
    asiofi::asiofi
    Boost::container
-    AZMQ::AZMQ
  )
 endif()
 set(optional_deps ${NANOMSG_DEPS} ${OFI_DEPS})
--- a/fairmq/FairMQChannel.h
+++ b/fairmq/FairMQChannel.h
@@ -345,6 +345,11 @@ class FairMQChannel
        return Transport()->NewStaticMessage(data);
    }

+    FairMQUnmanagedRegionPtr NewUnmanagedRegion(const size_t size, FairMQRegionCallback callback = nullptr)
+    {
+        return Transport()->CreateUnmanagedRegion(size, callback);
+    }
+
  private:
    std::shared_ptr<FairMQTransportFactory> fTransportFactory;
    fair::mq::Transport fTransportType;
--- a/fairmq/FairMQDevice.cxx
+++ b/fairmq/FairMQDevice.cxx
@@ -26,33 +26,34 @@
 #include <algorithm> // std::max

 using namespace std;
+using namespace fair::mq;

-static map<fair::mq::Transition, fair::mq::State> backwardsCompatibilityWaitForEndOfStateHelper =
+static map<Transition, State> backwardsCompatibilityWaitForEndOfStateHelper =
 {
-    { fair::mq::Transition::InitDevice,   fair::mq::State::InitializingDevice },
-    { fair::mq::Transition::CompleteInit, fair::mq::State::Initialized },
-    { fair::mq::Transition::Bind,         fair::mq::State::Bound },
-    { fair::mq::Transition::Connect,      fair::mq::State::DeviceReady },
-    { fair::mq::Transition::InitTask,     fair::mq::State::Ready },
-    { fair::mq::Transition::Run,          fair::mq::State::Ready },
-    { fair::mq::Transition::Stop,         fair::mq::State::Ready },
-    { fair::mq::Transition::ResetTask,    fair::mq::State::DeviceReady },
-    { fair::mq::Transition::ResetDevice,  fair::mq::State::Idle }
+    { Transition::InitDevice,   State::InitializingDevice },
+    { Transition::CompleteInit, State::Initialized },
+    { Transition::Bind,         State::Bound },
+    { Transition::Connect,      State::DeviceReady },
+    { Transition::InitTask,     State::Ready },
+    { Transition::Run,          State::Ready },
+    { Transition::Stop,         State::Ready },
+    { Transition::ResetTask,    State::DeviceReady },
+    { Transition::ResetDevice,  State::Idle }
 };

-static map<int, fair::mq::Transition> backwardsCompatibilityChangeStateHelper =
+static map<int, Transition> backwardsCompatibilityChangeStateHelper =
 {
-    { FairMQDevice::Event::INIT_DEVICE, fair::mq::Transition::InitDevice },
-    { FairMQDevice::Event::internal_DEVICE_READY, fair::mq::Transition::Auto },
-    { FairMQDevice::Event::INIT_TASK, fair::mq::Transition::InitTask },
-    { FairMQDevice::Event::internal_READY, fair::mq::Transition::Auto },
-    { FairMQDevice::Event::RUN, fair::mq::Transition::Run },
-    { FairMQDevice::Event::STOP, fair::mq::Transition::Stop },
-    { FairMQDevice::Event::RESET_TASK, fair::mq::Transition::ResetTask },
-    { FairMQDevice::Event::RESET_DEVICE, fair::mq::Transition::ResetDevice },
-    { FairMQDevice::Event::internal_IDLE, fair::mq::Transition::Auto },
-    { FairMQDevice::Event::END, fair::mq::Transition::End },
-    { FairMQDevice::Event::ERROR_FOUND, fair::mq::Transition::ErrorFound }
+    { FairMQDevice::Event::INIT_DEVICE,           Transition::InitDevice },
+    { FairMQDevice::Event::internal_DEVICE_READY, Transition::Auto },
+    { FairMQDevice::Event::INIT_TASK,             Transition::InitTask },
+    { FairMQDevice::Event::internal_READY,        Transition::Auto },
+    { FairMQDevice::Event::RUN,                   Transition::Run },
+    { FairMQDevice::Event::STOP,                  Transition::Stop },
+    { FairMQDevice::Event::RESET_TASK,            Transition::ResetTask },
+    { FairMQDevice::Event::RESET_DEVICE,          Transition::ResetDevice },
+    { FairMQDevice::Event::internal_IDLE,         Transition::Auto },
+    { FairMQDevice::Event::END,                   Transition::End },
+    { FairMQDevice::Event::ERROR_FOUND,           Transition::ErrorFound }
 };

 FairMQDevice::FairMQDevice()
@@ -65,21 +66,21 @@ FairMQDevice::FairMQDevice(FairMQProgOptions& config)
 {
 }

-FairMQDevice::FairMQDevice(const fair::mq::tools::Version version)
+FairMQDevice::FairMQDevice(const tools::Version version)
    : FairMQDevice(nullptr, version)
 {
 }

-FairMQDevice::FairMQDevice(FairMQProgOptions& config, const fair::mq::tools::Version version)
+FairMQDevice::FairMQDevice(FairMQProgOptions& config, const tools::Version version)
    : FairMQDevice(&config, version)
 {
 }

-FairMQDevice::FairMQDevice(FairMQProgOptions* config, const fair::mq::tools::Version version)
+FairMQDevice::FairMQDevice(FairMQProgOptions* config, const tools::Version version)
    : fTransportFactory(nullptr)
    , fTransports()
    , fChannels()
-    , fInternalConfig(config ? nullptr : fair::mq::tools::make_unique<FairMQProgOptions>())
+    , fInternalConfig(config ? nullptr : tools::make_unique<FairMQProgOptions>())
    , fConfig(config ? config : fInternalConfig.get())
    , fId()
    , fDefaultTransportType(fair::mq::Transport::ZMQ)
@@ -99,11 +100,11 @@ FairMQDevice::FairMQDevice(FairMQProgOptions* config, const fair::mq::tools::Ver
    , fMaxRunRuntimeInS(0)
    , fRawCmdLineArgs()
 {
-    SubscribeToNewTransition("device", [&](fair::mq::Transition transition) {
+    SubscribeToNewTransition("device", [&](Transition transition) {
        LOG(trace) << "device notified on new transition: " << transition;

        switch (transition) {
-            case fair::mq::Transition::Stop:
+            case Transition::Stop:
                UnblockTransports();
                break;
            default:
@@ -118,7 +119,7 @@ FairMQDevice::FairMQDevice(FairMQProgOptions* config, const fair::mq::tools::Ver
            lock_guard<mutex> lock(fStatesMtx);
            fStates.push(state);
        }
-        fStatesCV.notify_one();
+        fStatesCV.notify_all();

        switch (state) {
            case fair::mq::State::InitializingDevice:
@@ -182,7 +183,7 @@ bool FairMQDevice::ChangeState(const int transition)
    return ChangeState(backwardsCompatibilityChangeStateHelper.at(transition));
 }

-void FairMQDevice::WaitForEndOfState(fair::mq::Transition transition)
+void FairMQDevice::WaitForEndOfState(Transition transition)
 {
    WaitForState(backwardsCompatibilityWaitForEndOfStateHelper.at(transition));
 }
@@ -226,7 +227,7 @@ void FairMQDevice::InitWrapper()
        int subChannelIndex = 0;
        for (auto& vi : mi.second) {
            // set channel name: name + vector index
-            vi.fName = fair::mq::tools::ToString(mi.first, "[", subChannelIndex, "]");
+            vi.fName = tools::ToString(mi.first, "[", subChannelIndex, "]");

            // set channel transport
            LOG(debug) << "Initializing transport for channel " << vi.fName << ": " << fair::mq::TransportNames.at(vi.fTransportType);
@@ -237,9 +238,9 @@ void FairMQDevice::InitWrapper()
                if (vi.fAddress == "unspecified" || vi.fAddress == "") {
                    // if the configured network interface is default, get its name from the default route
                    if (networkInterface == "default") {
-                        networkInterface = fair::mq::tools::getDefaultRouteNetworkInterface();
+                        networkInterface = tools::getDefaultRouteNetworkInterface();
                    }
-                    vi.fAddress = "tcp://" + fair::mq::tools::getInterfaceIP(networkInterface) + ":1";
+                    vi.fAddress = "tcp://" + tools::getInterfaceIP(networkInterface) + ":1";
                }
                // fill the uninitialized list
                fUninitializedBindingChannels.push_back(&vi);
@@ -251,14 +252,14 @@ void FairMQDevice::InitWrapper()
                fUninitializedConnectingChannels.push_back(&vi);
            } else {
                LOG(error) << "Cannot update configuration. Socket method (bind/connect) for channel '" << vi.fName << "' not specified.";
-                throw runtime_error(fair::mq::tools::ToString("Cannot update configuration. Socket method (bind/connect) for channel ", vi.fName, " not specified."));
+                throw runtime_error(tools::ToString("Cannot update configuration. Socket method (bind/connect) for channel ", vi.fName, " not specified."));
            }

            subChannelIndex++;
        }
    }

-    // ChangeState(fair::mq::Transition::Auto);
+    // ChangeState(Transition::Auto);
 }

 void FairMQDevice::BindWrapper()
@@ -269,12 +270,12 @@ void FairMQDevice::BindWrapper()

    if (!fUninitializedBindingChannels.empty()) {
        LOG(error) << fUninitializedBindingChannels.size() << " of the binding channels could not initialize. Initial configuration incomplete.";
-        throw runtime_error(fair::mq::tools::ToString(fUninitializedBindingChannels.size(), " of the binding channels could not initialize. Initial configuration incomplete."));
+        throw runtime_error(tools::ToString(fUninitializedBindingChannels.size(), " of the binding channels could not initialize. Initial configuration incomplete."));
    }

    Bind();

-    ChangeState(fair::mq::Transition::Auto);
+    ChangeState(Transition::Auto);
 }

 void FairMQDevice::ConnectWrapper()
@@ -301,7 +302,7 @@ void FairMQDevice::ConnectWrapper()

        if (numAttempts++ > maxAttempts) {
            LOG(error) << "could not connect all channels after " << initializationTimeoutInS << " attempts";
-            throw runtime_error(fair::mq::tools::ToString("could not connect all channels after ", initializationTimeoutInS, " attempts"));
+            throw runtime_error(tools::ToString("could not connect all channels after ", initializationTimeoutInS, " attempts"));
        }

        AttachChannels(fUninitializedConnectingChannels);
@@ -313,7 +314,7 @@ void FairMQDevice::ConnectWrapper()

    Connect();

-    ChangeState(fair::mq::Transition::Auto);
+    ChangeState(Transition::Auto);
 }

 void FairMQDevice::AttachChannels(vector<FairMQChannel*>& chans)
@@ -366,7 +367,7 @@ bool FairMQDevice::AttachChannel(FairMQChannel& chan)
            string hostPart = addressString.substr(0, pos);
            if (!(bind && hostPart == "*")) {
                string portPart = addressString.substr(pos + 1);
-                string resolvedHost = fair::mq::tools::getIpFromHostname(hostPart);
+                string resolvedHost = tools::getIpFromHostname(hostPart);
                if (resolvedHost == "") {
                    return false;
                }
@@ -414,7 +415,7 @@ void FairMQDevice::InitTaskWrapper()
 {
    InitTask();

-    ChangeState(fair::mq::Transition::Auto);
+    ChangeState(Transition::Auto);
 }

 bool FairMQDevice::SortSocketsByAddress(const FairMQChannel &lhs, const FairMQChannel &rhs)
@@ -433,7 +434,7 @@ void FairMQDevice::SortChannel(const string& name, const bool reindex)
            for (auto vi = fChannels.at(name).begin(); vi != fChannels.at(name).end(); ++vi)
            {
                // set channel name: name + vector index
-                vi->fName = fair::mq::tools::ToString(name, "[", vi - fChannels.at(name).begin(), "]");
+                vi->fName = tools::ToString(name, "[", vi - fChannels.at(name).begin(), "]");
            }
        }
    }
@@ -467,7 +468,7 @@ void FairMQDevice::RunWrapper()
                HandleMultipleChannelInput();
            }
        } else {
-            fair::mq::tools::RateLimiter rateLimiter(fRate);
+            tools::RateLimiter rateLimiter(fRate);

            while (!NewStatePending() && ConditionalRun()) {
                if (fRate > 0.001) {
@@ -481,7 +482,7 @@ void FairMQDevice::RunWrapper()
        // if Run() exited and the state is still RUNNING, transition to READY.
        if (!NewStatePending()) {
            UnblockTransports();
-            ChangeState(fair::mq::Transition::Stop);
+            ChangeState(Transition::Stop);
        }

        PostRun();
@@ -489,10 +490,10 @@ void FairMQDevice::RunWrapper()
    } catch (const out_of_range& oor) {
        LOG(error) << "out of range: " << oor.what();
        LOG(error) << "incorrect/incomplete channel configuration?";
-        ChangeState(fair::mq::Transition::ErrorFound);
+        ChangeState(Transition::ErrorFound);
        throw;
    } catch (...) {
-        ChangeState(fair::mq::Transition::ErrorFound);
+        ChangeState(Transition::ErrorFound);
        throw;
    }

@@ -664,7 +665,7 @@ void FairMQDevice::PollForTransport(const FairMQTransportFactory* factory, const
    catch (exception& e)
    {
        LOG(error) << "FairMQDevice::PollForTransport() failed: " << e.what() << ", going to ERROR state.";
-        throw runtime_error(fair::mq::tools::ToString("FairMQDevice::PollForTransport() failed: ", e.what(), ", going to ERROR state."));
+        throw runtime_error(tools::ToString("FairMQDevice::PollForTransport() failed: ", e.what(), ", going to ERROR state."));
    }
 }

@@ -738,7 +739,7 @@ void FairMQDevice::LogSocketRates()
                filteredSockets.push_back(vi->fSocket.get());
                logIntervals.push_back(vi->fRateLogging);
                intervalCounters.push_back(0);
-                filteredChannelNames.push_back(fair::mq::tools::ToString(mi.first, "[", vi - (mi.second).begin(), "]"));
+                filteredChannelNames.push_back(tools::ToString(mi.first, "[", vi - (mi.second).begin(), "]"));
                chanNameLen = max(chanNameLen, filteredChannelNames.back().length());
            }
        }
@@ -814,7 +815,7 @@ void FairMQDevice::LogSocketRates()

        t0 = t1;
        if (fMaxRunRuntimeInS > 0 && ++secondsElapsed >= fMaxRunRuntimeInS) {
-            ChangeState(fair::mq::Transition::Stop);
+            ChangeState(Transition::Stop);
        }
    }
 }
@@ -830,7 +831,7 @@ void FairMQDevice::ResetTaskWrapper()
 {
    ResetTask();

-    ChangeState(fair::mq::Transition::Auto);
+    ChangeState(Transition::Auto);
 }

 void FairMQDevice::ResetWrapper()
@@ -850,7 +851,7 @@ void FairMQDevice::ResetWrapper()

    Reset();

-    ChangeState(fair::mq::Transition::Auto);
+    ChangeState(Transition::Auto);
 }

 FairMQDevice::~FairMQDevice()
--- a/fairmq/FairMQDevice.h
+++ b/fairmq/FairMQDevice.h
@@ -191,50 +191,58 @@ class FairMQDevice
        return fTransportFactory.get();
    }

+    // creates message with the default device transport
    template<typename... Args>
    FairMQMessagePtr NewMessage(Args&&... args)
    {
        return Transport()->CreateMessage(std::forward<Args>(args)...);
    }

+    // creates message with the transport of the specified channel
    template<typename... Args>
    FairMQMessagePtr NewMessageFor(const std::string& channel, int index, Args&&... args)
    {
        return GetChannel(channel, index).NewMessage(std::forward<Args>(args)...);
    }

+    // creates a message that will not be cleaned up after transfer, with the default device transport
    template<typename T>
    FairMQMessagePtr NewStaticMessage(const T& data)
    {
        return Transport()->NewStaticMessage(data);
    }

+    // creates a message that will not be cleaned up after transfer, with the transport of the specified channel
    template<typename T>
    FairMQMessagePtr NewStaticMessageFor(const std::string& channel, int index, const T& data)
    {
        return GetChannel(channel, index).NewStaticMessage(data);
    }

+    // creates a message with a copy of the provided data, with the default device transport
    template<typename T>
    FairMQMessagePtr NewSimpleMessage(const T& data)
    {
        return Transport()->NewSimpleMessage(data);
    }

+    // creates a message with a copy of the provided data, with the transport of the specified channel
    template<typename T>
    FairMQMessagePtr NewSimpleMessageFor(const std::string& channel, int index, const T& data)
    {
        return GetChannel(channel, index).NewSimpleMessage(data);
    }

-    FairMQUnmanagedRegionPtr NewUnmanagedRegion(const size_t size)
+    // creates unamanaged region with the default device transport
+    FairMQUnmanagedRegionPtr NewUnmanagedRegion(const size_t size, FairMQRegionCallback callback = nullptr)
    {
-        return Transport()->CreateUnmanagedRegion(size);
+        return Transport()->CreateUnmanagedRegion(size, callback);
    }

+    // creates unmanaged region with the transport of the specified channel
    FairMQUnmanagedRegionPtr NewUnmanagedRegionFor(const std::string& channel, int index, const size_t size, FairMQRegionCallback callback = nullptr)
    {
-        return GetChannel(channel, index).Transport()->CreateUnmanagedRegion(size, callback);
+        return GetChannel(channel, index).NewUnmanagedRegion(size, callback);
    }

    template<typename ...Ts>
@@ -358,8 +366,9 @@ class FairMQDevice
    try {
        return fChannels.at(channelName).at(index);
    } catch (const std::out_of_range& oor) {
-        LOG(error) << "out of range: " << oor.what();
        LOG(error) << "requested channel has not been configured? check channel names/configuration.";
+        LOG(error) << "channel: " << channelName << ", index: " << index;
+        LOG(error) << "out of range: " << oor.what();
        throw;
    }

@@ -422,7 +431,7 @@ class FairMQDevice
    template<class Rep, class Period>
    bool WaitFor(std::chrono::duration<Rep, Period> const& duration)
    {
-        return fStateMachine.WaitForPendingStateFor(std::chrono::duration_cast<std::chrono::milliseconds>(duration).count());
+        return !fStateMachine.WaitForPendingStateFor(std::chrono::duration_cast<std::chrono::milliseconds>(duration).count());
    }

  protected:
--- a/fairmq/FairMQParts.h
+++ b/fairmq/FairMQParts.h
@@ -31,6 +31,9 @@ class FairMQParts
    FairMQParts(FairMQParts&& p) = default;
    /// Assignment operator
    FairMQParts& operator=(const FairMQParts&) = delete;
+    /// Constructor from argument pack of std::unique_ptr<FairMQMessage> rvalues
+    template <typename... Ts>
+    FairMQParts(Ts&&... messages) : fParts() {AddPart(std::forward<Ts>(messages)...);}
    /// Default destructor
    ~FairMQParts() {};

@@ -41,14 +44,6 @@ class FairMQParts
        fParts.push_back(std::unique_ptr<FairMQMessage>(msg));
    }

-    /// Adds part (std::unique_ptr<FairMQMessage>&) to the container (move)
-    /// @param msg unique pointer to FairMQMessage
-    /// lvalue ref (move not required when passing argument)
-    // inline void AddPart(std::unique_ptr<FairMQMessage>& msg)
-    // {
-    //     fParts.push_back(std::move(msg));
-    // }
-
    /// Adds part (std::unique_ptr<FairMQMessage>&) to the container (move)
    /// @param msg unique pointer to FairMQMessage
    /// rvalue ref (move required when passing argument)
@@ -57,6 +52,23 @@ class FairMQParts
        fParts.push_back(std::move(msg));
    }

+    /// Add variable list of parts to the container (move)
+    template <typename... Ts>
+    void AddPart(std::unique_ptr<FairMQMessage>&& first, Ts&&... remaining)
+    {
+        AddPart(std::move(first));
+        AddPart(std::forward<Ts>(remaining)...);
+    }
+
+    /// Add content of another object by move
+    void AddPart(FairMQParts&& other)
+    {
+      container parts = std::move(other.fParts);
+      for (auto& part : parts) {
+	fParts.push_back(std::move(part));
+      }
+    }
+
    /// Get reference to part in the container at index (without bounds check)
    /// @param index container index
    FairMQMessage& operator[](const int index) { return *(fParts[index]); }
--- a/fairmq/PluginServices.h
+++ b/fairmq/PluginServices.h
@@ -196,6 +196,7 @@ class PluginServices
            || (currentState == DeviceState::Binding)
            || (currentState == DeviceState::Bound)
            || (currentState == DeviceState::Connecting)
+            || (currentState == DeviceState::Ready)
            || (currentState == DeviceState::Idle && key == "channel-config")) {
            fConfig.SetValue(key, val);
        } else {
--- a/fairmq/StateMachine.cxx
+++ b/fairmq/StateMachine.cxx
@@ -28,6 +28,7 @@
 #include <array>
 #include <unordered_map>
 #include <mutex>
+#include <stdexcept>

 using namespace std;
 using namespace boost::msm;
@@ -167,7 +168,6 @@ struct Machine_ : public state_machine_def<Machine_>
    Machine_()
        : fLastTransitionResult(true)
        , fNewStatePending(false)
-        , fWorkOngoing(false)
    {}

    virtual ~Machine_() {}
@@ -258,9 +258,7 @@ struct Machine_ : public state_machine_def<Machine_>

    mutex fStateMtx;
    atomic<bool> fNewStatePending;
-    atomic<bool> fWorkOngoing;
    condition_variable fNewStatePendingCV;
-    condition_variable fWorkDoneCV;

    boost::signals2::signal<void(const State)> fStateChangeSignal;
    boost::signals2::signal<void(const State)> fStateHandleSignal;
@@ -283,7 +281,6 @@ struct Machine_ : public state_machine_def<Machine_>
                LOG(state) << fState << " ---> " << fNewState;
                fState = static_cast<State>(fNewState);
                fNewStatePending = false;
-                fWorkOngoing = true;

                if (fState == State::Exiting || fState == State::Error) {
                    stop = true;
@@ -292,12 +289,10 @@ struct Machine_ : public state_machine_def<Machine_>

            CallStateChangeCallbacks(fState);
            CallStateHandler(fState);
+        }

-            {
-                lock_guard<mutex> lock(fStateMtx);
-                fWorkOngoing = false;
-                fWorkDoneCV.notify_one();
-            }
+        if (fState == State::Error) {
+            throw StateMachine::ErrorStateException("Device transitioned to error state");
        }
    }

@@ -463,13 +458,15 @@ void StateMachine::ProcessWork()
    try {
        fsm->CallStateChangeCallbacks(State::Idle);
        fsm->ProcessWork();
+    } catch(ErrorStateException& ese) {
+        LOG(trace) << "ErrorStateException caught in ProcessWork(), rethrowing";
+        throw;
    } catch(...) {
+        LOG(debug) << "Exception caught in ProcessWork(), going to Error state and rethrowing";
        {
            lock_guard<mutex> lock(fsm->fStateMtx);
            fsm->fState = State::Error;
            fsm->CallStateChangeCallbacks(State::Error);
-            fsm->fWorkOngoing = false;
-            fsm->fWorkDoneCV.notify_one();
        }
        ChangeState(Transition::ErrorFound);
        throw;
@@ -480,3 +477,4 @@ string StateMachine::GetStateName(const State state) { return stateNames.at(stat
 string StateMachine::GetTransitionName(const Transition transition) { return transitionNames.at(static_cast<int>(transition)); }
 State StateMachine::GetState(const string& state) { return stateNumbers.at(state); }
 Transition StateMachine::GetTransition(const string& transition) { return transitionNumbers.at(transition); }
+
--- a/fairmq/StateMachine.h
+++ b/fairmq/StateMachine.h
@@ -94,6 +94,8 @@ class StateMachine
    static State GetState(const std::string& state);
    static Transition GetTransition(const std::string& transition);

+    struct ErrorStateException : std::runtime_error { using std::runtime_error::runtime_error; };
+
  private:
    std::shared_ptr<void> fFsm;
 };
--- a/fairmq/ofi/Context.cxx
+++ b/fairmq/ofi/Context.cxx
@@ -37,6 +37,7 @@ Context::Context(FairMQTransportFactory& sendFactory,
    : fIoWork(fIoContext)
    , fReceiveFactory(receiveFactory)
    , fSendFactory(sendFactory)
+    , fSizeHint(0)
 {
    InitThreadPool(numberIoThreads);
 }
@@ -47,15 +48,24 @@ auto Context::InitThreadPool(int numberIoThreads) -> void

    for (int i = 1; i <= numberIoThreads; ++i) {
        fThreadPool.emplace_back([&, i, numberIoThreads]{
-            LOG(debug) << "OFI transport: I/O thread #" << i << " of " << numberIoThreads << " started";
-            fIoContext.run();
-            LOG(debug) << "OFI transport: I/O thread #" << i << " of " << numberIoThreads << " stopped";
+            try {
+                LOG(debug) << "OFI transport: I/O thread #" << i << " of " << numberIoThreads << " started";
+                fIoContext.run();
+                LOG(debug) << "OFI transport: I/O thread #" << i << " of " << numberIoThreads << " stopped";
+            } catch (const std::exception& e) {
+                LOG(error) << "OFI transport: Uncaught exception in I/O thread #" << i << ": " << e.what();
+            } catch (...) {
+                LOG(error) << "OFI transport: Uncaught exception in I/O thread #" << i;
+            }
        });
    }
 }

 auto Context::Reset() -> void
 {
+    // TODO "Linger", rethink this
+    std::this_thread::sleep_for(std::chrono::milliseconds(1000));
+
    fIoContext.stop();
 }

--- a/fairmq/ofi/Context.h
+++ b/fairmq/ofi/Context.h
@@ -72,6 +72,8 @@ class Context
    auto Reset() -> void;
    auto MakeReceiveMessage(size_t size) -> MessagePtr;
    auto MakeSendMessage(size_t size) -> MessagePtr;
+    auto GetSizeHint() -> size_t { return fSizeHint; }
+    auto SetSizeHint(size_t size) -> void { fSizeHint = size; }

  private:
    boost::asio::io_context fIoContext;
@@ -79,6 +81,7 @@ class Context
    std::vector<std::thread> fThreadPool;
    FairMQTransportFactory& fReceiveFactory;
    FairMQTransportFactory& fSendFactory;
+    size_t fSizeHint;

    auto InitThreadPool(int numberIoThreads) -> void;
 }; /* class Context */
--- a/fairmq/ofi/ControlMessages.h
+++ b/fairmq/ofi/ControlMessages.h
@@ -35,59 +35,76 @@ namespace ofi {

 enum class ControlMessageType
 {
-    DataAddressAnnouncement = 1,
+    Empty = 1,
    PostBuffer,
-    PostBufferAcknowledgement
+    PostMultiPartStartBuffer
+};
+
+struct Empty
+{};
+
+struct PostBuffer
+{
+    uint64_t size;   // buffer size (size_t)
+};
+
+struct PostMultiPartStartBuffer
+{
+    uint32_t numParts;   // buffer size (size_t)
+    uint64_t size;       // buffer size (size_t)
+};
+
+union ControlMessageContent
+{
+    PostBuffer postBuffer;
+    PostMultiPartStartBuffer postMultiPartStartBuffer;
 };

 struct ControlMessage
 {
    ControlMessageType type;
-};
-
-struct DataAddressAnnouncement : ControlMessage
-{
-    uint32_t ipv4;   // in_addr_t from <netinet/in.h>
-    uint32_t port;   // in_port_t from <netinet/in.h>
-};
-
-struct PostBuffer : ControlMessage
-{
-    uint64_t size;   // buffer size (size_t)
+    ControlMessageContent msg;
 };

 template<typename T>
 using unique_ptr = std::unique_ptr<T, std::function<void(T*)>>;

 template<typename T, typename... Args>
-auto MakeControlMessageWithPmr(boost::container::pmr::memory_resource* pmr, Args&&... args)
-    -> ofi::unique_ptr<T>
+auto MakeControlMessageWithPmr(boost::container::pmr::memory_resource& pmr, Args&&... args)
+    -> ofi::unique_ptr<ControlMessage>
 {
-    void* mem = pmr->allocate(sizeof(T));
-    T* ctrl = new (mem) T(std::forward<Args>(args)...);
+    void* mem = pmr.allocate(sizeof(ControlMessage));
+    ControlMessage* ctrl = new (mem) ControlMessage();

-    if (std::is_same<T, DataAddressAnnouncement>::value) {
-        ctrl->type = ControlMessageType::DataAddressAnnouncement;
-    } else if (std::is_same<T, PostBuffer>::value) {
+    if (std::is_same<T, PostBuffer>::value) {
        ctrl->type = ControlMessageType::PostBuffer;
+        ctrl->msg.postBuffer = PostBuffer(std::forward<Args>(args)...);
+    } else if (std::is_same<T, PostMultiPartStartBuffer>::value) {
+        ctrl->type = ControlMessageType::PostMultiPartStartBuffer;
+        ctrl->msg.postMultiPartStartBuffer = PostMultiPartStartBuffer(std::forward<Args>(args)...);
+    } else if (std::is_same<T, Empty>::value) {
+        ctrl->type = ControlMessageType::Empty;
    }

-    return ofi::unique_ptr<T>(ctrl, [=](T* p) {
-        p->~T();
-        pmr->deallocate(p, sizeof(T));
+    return ofi::unique_ptr<ControlMessage>(ctrl, [&pmr](ControlMessage* p) {
+        p->~ControlMessage();
+        pmr.deallocate(p, sizeof(T));
    });
 }

 template<typename T, typename... Args>
-auto MakeControlMessage(Args&&... args) -> T
+auto MakeControlMessage(Args&&... args) -> ControlMessage
 {
-    T ctrl = T(std::forward<Args>(args)...);
+    ControlMessage ctrl;

-    if (std::is_same<T, DataAddressAnnouncement>::value) {
-        ctrl.type = ControlMessageType::DataAddressAnnouncement;
-    } else if (std::is_same<T, PostBuffer>::value) {
+    if (std::is_same<T, PostBuffer>::value) {
        ctrl.type = ControlMessageType::PostBuffer;
+    } else if (std::is_same<T, PostMultiPartStartBuffer>::value) {
+        ctrl.type = ControlMessageType::PostMultiPartStartBuffer;
+    } else if (std::is_same<T, Empty>::value) {
+        ctrl.type = ControlMessageType::Empty;
    }
+    ctrl.msg = T(std::forward<Args>(args)...);

    return ctrl;
 }
--- a/fairmq/ofi/Socket.cxx
+++ b/fairmq/ofi/Socket.cxx
--- a/fairmq/ofi/Socket.h
+++ b/fairmq/ofi/Socket.h
@@ -18,10 +18,10 @@
 #include <asiofi/memory_resources.hpp>
 #include <asiofi/passive_endpoint.hpp>
 #include <asiofi/semaphore.hpp>
-#include <azmq/socket.hpp>
 #include <boost/asio.hpp>
 #include <memory> // unique_ptr
-#include <netinet/in.h>
+#include <mutex>
+

 namespace fair
 {
@@ -97,28 +97,24 @@ class Socket final : public fair::mq::Socket
    Address fLocalAddr;
    int fSndTimeout;
    int fRcvTimeout;
-    azmq::socket fSendQueueWrite, fSendQueueRead;
-    azmq::socket fRecvQueueWrite, fRecvQueueRead;
-    asiofi::semaphore fSendSem, fRecvSem;
+    std::mutex fSendQueueMutex, fRecvQueueMutex;
+    std::queue<std::vector<MessagePtr>> fSendQueue, fRecvQueue;
+    std::vector<MessagePtr> fInflightMultiPartMessage;
+    int64_t fMultiPartRecvCounter;
+    asiofi::synchronized_semaphore fSendPushSem, fSendPopSem, fRecvPushSem, fRecvPopSem;
    std::atomic<bool> fNeedOfiMemoryRegistration;

-    auto SendQueueReader() -> void;
-    auto OnSend(azmq::message& msg, size_t bytes_transferred) -> void;
-    auto RecvControlQueueReader() -> void;
-    auto OnRecvControl(ofi::unique_ptr<PostBuffer> ctrl) -> void;
-    auto OnReceive() -> void;
-    auto ReceiveImpl(MessagePtr& msg, const int flags, const int timeout) -> int;
-    auto SendImpl(std::vector<MessagePtr>& msgVec, const int flags, const int timeout) -> int64_t;
-    auto ReceiveImpl(std::vector<MessagePtr>& msgVec, const int flags, const int timeout) -> int64_t;
-
-    // auto WaitForControlPeer() -> void;
-    // auto AnnounceDataAddress() -> void;
    auto InitOfi(Address addr) -> void;
    auto BindControlEndpoint() -> void;
    auto BindDataEndpoint() -> void;
    enum class Band { Control, Data };
    auto ConnectEndpoint(std::unique_ptr<asiofi::connected_endpoint>& endpoint, Band type) -> void;
-    // auto ReceiveDataAddressAnnouncement() -> void;
+    auto SendQueueReader() -> void;
+    auto SendQueueReaderStatic() -> void;
+    auto RecvControlQueueReader() -> void;
+    auto RecvQueueReaderStatic() -> void;
+    auto OnRecvControl(ofi::unique_ptr<ControlMessage> ctrl) -> void;
+    auto DataMessageReceived(MessagePtr msg) -> void;
 }; /* class Socket */

 struct SilentSocketError : SocketError { using SocketError::SocketError; };
--- a/fairmq/ofi/TransportFactory.cxx
+++ b/fairmq/ofi/TransportFactory.cxx
@@ -23,12 +23,15 @@ namespace ofi

 using namespace std;

-TransportFactory::TransportFactory(const string& id, const FairMQProgOptions* /*config*/)
+TransportFactory::TransportFactory(const string& id, const FairMQProgOptions* config)
 try : FairMQTransportFactory(id)
    , fContext(*this, *this, 1)
 {
-    LOG(debug) << "OFI transport: Using AZMQ & "
-               << "asiofi (" << fContext.GetAsiofiVersion() << ")";
+    LOG(debug) << "OFI transport: asiofi (" << fContext.GetAsiofiVersion() << ")";
+
+    if (config) {
+        fContext.SetSizeHint(config->GetValue<size_t>("ofi-size-hint"));
+    }
 } catch (ContextError& e) {
    throw TransportFactoryError{e.what()};
 }
--- a/fairmq/options/FairMQProgOptions.cxx
+++ b/fairmq/options/FairMQProgOptions.cxx
@@ -66,6 +66,7 @@ FairMQProgOptions::FairMQProgOptions()
        ("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-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.")
        ("rate",                   po::value<float   >()->default_value(0.),         "Rate for conditional run loop (Hz).")
        ("session",                po::value<string  >()->default_value("default"),  "Session name.");

--- a/fairmq/plugins/Control.cxx
+++ b/fairmq/plugins/Control.cxx
@@ -29,8 +29,7 @@ namespace
        ++gSignalCount;
        gLastSignal = signal;

-        if (gSignalCount > 1)
-        {
+        if (gSignalCount > 1) {
            std::abort();
        }
    }
@@ -57,11 +56,18 @@ Control::Control(const string& name, const Plugin::Version version, const string
 {
    SubscribeToDeviceStateChange([&](DeviceState newState) {
        LOG(trace) << "control plugin notified on new state: " << newState;
+
        {
            lock_guard<mutex> lock{fEventsMutex};
            fEvents.push(newState);
        }
        fNewEvent.notify_one();
+
+        if (newState == DeviceState::Error) {
+            fPluginShutdownRequested = true;
+            fDeviceShutdownRequested = true;
+            // throw DeviceErrorState("Controlled device transitioned to error state.");
+        }
    });

    try {
@@ -110,6 +116,25 @@ auto Control::RunStartupSequence() -> void
    while (WaitForNextState() != DeviceState::Running) {}
 }

+auto Control::WaitForNextState() -> DeviceState
+{
+    unique_lock<mutex> lock{fEventsMutex};
+    while (fEvents.empty()) {
+        fNewEvent.wait_for(lock, chrono::milliseconds(50));
+    }
+
+    auto result = fEvents.front();
+
+    if (result == DeviceState::Error) {
+        ReleaseDeviceControl();
+        throw DeviceErrorState("Controlled device transitioned to error state.");
+    }
+
+    fEvents.pop();
+
+    return result;
+}
+
 auto ControlPluginProgramOptions() -> Plugin::ProgOptions
 {
    namespace po = boost::program_options;
@@ -164,7 +189,7 @@ try {
    bool keepRunning = true;

    while (keepRunning) {
-        if (poll(cinfd, 1, 500)) {
+        if (poll(cinfd, 1, 100)) {
            if (fDeviceShutdownRequested) {
                break;
            }
@@ -178,9 +203,10 @@ try {
                break;
                case 'i':
                    cout << "\n --> [i] init device\n\n" << flush;
-                    ChangeDeviceState(DeviceStateTransition::InitDevice);
-                    while (WaitForNextState() != DeviceState::InitializingDevice) {}
-                    ChangeDeviceState(DeviceStateTransition::CompleteInit);
+                    if (ChangeDeviceState(DeviceStateTransition::InitDevice)) {
+                        while (WaitForNextState() != DeviceState::InitializingDevice) {}
+                        ChangeDeviceState(DeviceStateTransition::CompleteInit);
+                    }
                break;
                case 'b':
                    cout << "\n --> [b] bind\n\n" << flush;
@@ -248,6 +274,7 @@ try {
        }

        if (GetCurrentDeviceState() == DeviceState::Error) {
+            ReleaseDeviceControl();
            throw DeviceErrorState("Controlled device transitioned to error state.");
        }

@@ -298,13 +325,13 @@ void Control::PrintStateMachineColor()
       << " ║ \033[01;33mINITIALIZING DEVICE\033[0m ║ ║  \033[01;33mRESETTING DEVICE\033[0m  ║ \n"
       << " ╚══════════╦══════════╝ ╚════════ ▲ ═════════╝ \n"
       << " ┌───────── ▼ ─────────┐           │  ┌────────────────────────────┐ \n"
-       << " │     \033[01;36mINITIALIZED\033[0m     │           │  │ Legend:                    │ \n"
+       << " │     \033[01;36mINITIALIZED\033[0m    [\033[01;32md\033[0m]──────────┤  │ Legend:                    │ \n"
       << " └─────────[\033[01;32mb\033[0m]─────────┘           │  │----------------------------│ \n"
       << " ╔═════════ ▼ ═════════╗           │  │ [\033[01;32mk\033[0m] keyboard shortcut for  │ \n"
       << " ║       \033[01;33mBINDING\033[0m       ║           │  │     interactive controller │ \n"
       << " ╚══════════╦══════════╝           │  │ ┌────────────────────────┐ │ \n"
       << " ┌───────── ▼ ─────────┐           │  │ │      \033[01;36mIDLING STATE\033[0m      │ │ \n"
-       << " │        \033[01;36mBOUND\033[0m        │           │  │ └────────────────────────┘ │ \n"
+       << " │        \033[01;36mBOUND\033[0m       [\033[01;32md\033[0m]──────────┤  │ └────────────────────────┘ │ \n"
       << " └─────────[\033[01;32mx\033[0m]─────────┘           │  │ ╔════════════════════════╗ │ \n"
       << " ╔═════════ ▼ ═════════╗           │  │ ║      \033[01;33mWORKING STATE\033[0m     ║ │ \n"
       << " ║     \033[01;33mCONNECTING\033[0m      ║           │  │ ╚════════════════════════╝ │ \n"
@@ -336,13 +363,13 @@ void Control::PrintStateMachine()
       << " ║ INITIALIZING DEVICE ║ ║  RESETTING DEVICE  ║                      \n"
       << " ╚══════════╦══════════╝ ╚════════ ▲ ═════════╝                      \n"
       << " ┌───────── ▼ ─────────┐           │  ┌────────────────────────────┐ \n"
-       << " │     INITIALIZED     │           │  │ Legend:                    │ \n"
+       << " │     INITIALIZED    [d]──────────┤  │ Legend:                    │ \n"
       << " └─────────[b]─────────┘           │  │----------------------------│ \n"
       << " ╔═════════ ▼ ═════════╗           │  │ [k] keyboard shortcut for  │ \n"
       << " ║       BINDING       ║           │  │     interactive controller │ \n"
       << " ╚══════════╦══════════╝           │  │ ┌────────────────────────┐ │ \n"
       << " ┌───────── ▼ ─────────┐           │  │ │      IDLING STATE      │ │ \n"
-       << " │        BOUND        │           │  │ └────────────────────────┘ │ \n"
+       << " │        BOUND       [d]──────────┤  │ └────────────────────────┘ │ \n"
       << " └─────────[x]─────────┘           │  │ ╔════════════════════════╗ │ \n"
       << " ╔═════════ ▼ ═════════╗           │  │ ║      WORKING STATE     ║ │ \n"
       << " ║     CONNECTING      ║           │  │ ╚════════════════════════╝ │ \n"
@@ -363,64 +390,38 @@ void Control::PrintStateMachine()
    cout << ss.str() << flush;
 }

-auto Control::WaitForNextState() -> DeviceState
-{
-    unique_lock<mutex> lock{fEventsMutex};
-    while (fEvents.empty()) {
-        fNewEvent.wait_for(lock, chrono::milliseconds(50));
-    }
-
-    auto result = fEvents.front();
-
-    if (result == DeviceState::Error) {
-        throw DeviceErrorState("Controlled device transitioned to error state.");
-    }
-
-    fEvents.pop();
-
-    return result;
-}
-
 auto Control::StaticMode() -> void
-try
-{
+try {
    RunStartupSequence();

    {
        // Wait for next state, which is DeviceState::Ready,
        // or for device shutdown request (Ctrl-C)
        unique_lock<mutex> lock{fEventsMutex};
-        while (fEvents.empty() && !fDeviceShutdownRequested)
-        {
+        while (fEvents.empty() && !fDeviceShutdownRequested) {
            fNewEvent.wait_for(lock, chrono::milliseconds(50));
        }

-        if (fEvents.front() == DeviceState::Error)
-        {
+        if (fEvents.front() == DeviceState::Error) {
+            ReleaseDeviceControl();
            throw DeviceErrorState("Controlled device transitioned to error state.");
        }
    }

    RunShutdownSequence();
-}
-catch (PluginServices::DeviceControlError& e)
-{
+} catch (PluginServices::DeviceControlError& e) {
    // If we are here, it means another plugin has taken control. That's fine, just print the exception message and do nothing else.
    LOG(debug) << e.what();
-}
-catch (DeviceErrorState&)
-{
+} catch (DeviceErrorState&) {
 }

 auto Control::SignalHandler() -> void
 {
-    while (gSignalCount == 0 && !fPluginShutdownRequested)
-    {
+    while (gSignalCount == 0 && !fPluginShutdownRequested) {
        this_thread::sleep_for(chrono::milliseconds(100));
    }

-    if (!fPluginShutdownRequested)
-    {
+    if (!fPluginShutdownRequested) {
        LOG(info) << "Received device shutdown request (signal " << gLastSignal << ").";
        LOG(info) << "Waiting for graceful device shutdown. Hit Ctrl-C again to abort immediately.";

@@ -431,20 +432,14 @@ auto Control::SignalHandler() -> void
            if (fControllerThread.joinable()) fControllerThread.join();
        }

-        if (!fDeviceHasShutdown)
-        {
+        if (!fDeviceHasShutdown) {
            // Take over control and attempt graceful shutdown
            StealDeviceControl();
-            try
-            {
+            try {
                RunShutdownSequence();
-            }
-            catch (PluginServices::DeviceControlError& e)
-            {
+            } catch (PluginServices::DeviceControlError& e) {
                LOG(info) << "Graceful device shutdown failed: " << e.what() << " If hanging, hit Ctrl-C again to abort immediately.";
-            }
-            catch (...)
-            {
+            } catch (...) {
                LOG(info) << "Graceful device shutdown failed. If hanging, hit Ctrl-C again to abort immediately.";
            }
        }
--- a/fairmq/plugins/DDS/CMakeLists.txt
+++ b/fairmq/plugins/DDS/CMakeLists.txt
@@ -8,13 +8,13 @@

 set(plugin FairMQPlugin_dds)
 add_library(${plugin} SHARED ${CMAKE_CURRENT_SOURCE_DIR}/DDS.cxx ${CMAKE_CURRENT_SOURCE_DIR}/DDS.h)
-target_link_libraries(${plugin} FairMQ DDS::dds_intercom_lib DDS::dds_protocol_lib DDS::dds-user-defaults)
+target_link_libraries(${plugin} FairMQ DDS::dds_intercom_lib DDS::dds_protocol_lib)
 target_include_directories(${plugin} PRIVATE ${CMAKE_CURRENT_BINARY_DIR})
 set_target_properties(${plugin} PROPERTIES CXX_VISIBILITY_PRESET hidden)

 set(exe fairmq-dds-command-ui)
 add_executable(${exe} ${CMAKE_CURRENT_SOURCE_DIR}/runDDSCommandUI.cxx)
-target_link_libraries(${exe} FairMQ DDS::dds_intercom_lib DDS::dds_protocol_lib DDS::dds-user-defaults)
+target_link_libraries(${exe} FairMQ DDS::dds_intercom_lib DDS::dds_protocol_lib)
 target_include_directories(${exe} PRIVATE ${CMAKE_CURRENT_BINARY_DIR})

 install(TARGETS ${plugin} ${exe}
--- a/fairmq/shmem/FairMQMessageSHM.cxx
+++ b/fairmq/shmem/FairMQMessageSHM.cxx
@@ -327,50 +327,19 @@ void FairMQMessageSHM::CloseMessage()
        }
        else
        {
-            // send notification back to the receiver
-            // RegionBlock block(fHandle, fSize);
-            // if (fManager.GetRegionQueue(fRegionId).try_send(static_cast<void*>(&block), sizeof(RegionBlock), 0))
-            // {
-            //     // LOG(info) << "true";
-            // }
-            // // else
-            // // {
-            // //     LOG(debug) << "could not send ack";
-            // // }
-
-            // timed version
-            RegionBlock block(fHandle, fSize, fHint);
-            bool success = false;
-            do
+            if (!fRegionPtr)
            {
-                auto sndTill = bpt::microsec_clock::universal_time() + bpt::milliseconds(200);
-                if (!fRegionPtr)
-                {
-                    fRegionPtr = fManager.GetRemoteRegion(fRegionId);
-                }
-                if (fRegionPtr)
-                {
-                    // LOG(debug) << "sending ack";
-                    if (fRegionPtr->fQueue->timed_send(&block, sizeof(RegionBlock), 0, sndTill))
-                    {
-                        success = true;
-                    }
-                    else
-                    {
-                        if (fInterrupted)
-                        {
-                            break;
-                        }
-                        LOG(debug) << "region ack queue is full, retrying...";
-                    }
-                }
-                else
-                {
-                    // LOG(warn) << "region ack queue for id " << fRegionId << " no longer exist. Not sending ack";
-                    success = true;
-                }
+                fRegionPtr = fManager.GetRemoteRegion(fRegionId);
+            }
+
+            if (fRegionPtr)
+            {
+                fRegionPtr->ReleaseBlock({fHandle, fSize, fHint});
+            }
+            else
+            {
+                LOG(warn) << "region ack queue for id " << fRegionId << " no longer exist. Not sending ack";
            }
-            while (!success);
        }
    }

--- a/fairmq/shmem/Manager.cxx
+++ b/fairmq/shmem/Manager.cxx
@@ -19,7 +19,7 @@ namespace mq
 namespace shmem
 {

-std::unordered_map<uint64_t, Region> Manager::fRegions;
+std::unordered_map<uint64_t, std::unique_ptr<Region>> Manager::fRegions;

 Manager::Manager(const string& name, size_t size)
    : fSessionName(name)
@@ -43,7 +43,7 @@ void Manager::Resume()
    // close remote regions before processing new transfers
    for (auto it = fRegions.begin(); it != fRegions.end(); /**/)
    {
-        if (it->second.fRemote)
+        if (it->second->fRemote)
        {
           it = fRegions.erase(it);
        }
@@ -64,11 +64,11 @@ bipc::mapped_region* Manager::CreateRegion(const size_t size, const uint64_t id,
    }
    else
    {
-        auto r = fRegions.emplace(id, Region{*this, id, size, false, callback});
+        auto r = fRegions.emplace(id, fair::mq::tools::make_unique<Region>(*this, id, size, false, callback));

-        r.first->second.StartReceivingAcks();
+        r.first->second->StartReceivingAcks();

-        return &(r.first->second.fRegion);
+        return &(r.first->second->fRegion);
    }
 }

@@ -78,14 +78,14 @@ Region* Manager::GetRemoteRegion(const uint64_t id)
    auto it = fRegions.find(id);
    if (it != fRegions.end())
    {
-        return &(it->second);
+        return it->second.get();
    }
    else
    {
        try
        {
-            auto r = fRegions.emplace(id, Region{*this, id, 0, true, nullptr});
-            return &(r.first->second);
+            auto r = fRegions.emplace(id, fair::mq::tools::make_unique<Region>(*this, id, 0, true, nullptr));
+            return r.first->second.get();
        }
        catch (bipc::interprocess_exception& e)
        {
--- a/fairmq/shmem/Manager.h
+++ b/fairmq/shmem/Manager.h
@@ -66,7 +66,7 @@ class Manager
    std::string fManagementSegmentName;
    boost::interprocess::managed_shared_memory fSegment;
    boost::interprocess::managed_shared_memory fManagementSegment;
-    static std::unordered_map<uint64_t, Region> fRegions;
+    static std::unordered_map<uint64_t, std::unique_ptr<Region>> fRegions;
 };

 } // namespace shmem
--- a/fairmq/shmem/Region.cxx
+++ b/fairmq/shmem/Region.cxx
@@ -12,6 +12,8 @@

 #include <boost/date_time/posix_time/posix_time.hpp>

+#include <chrono>
+
 using namespace std;

 namespace bipc = ::boost::interprocess;
@@ -32,7 +34,8 @@ Region::Region(Manager& manager, uint64_t id, uint64_t size, bool remote, FairMQ
    , fQueueName("fmq_" + fManager.fSessionName +"_rgq_" + to_string(id))
    , fShmemObject()
    , fQueue(nullptr)
-    , fWorker()
+    , fReceiveAcksWorker()
+    , fSendAcksWorker()
    , fCallback(callback)
 {
    if (fRemote)
@@ -49,52 +52,118 @@ Region::Region(Manager& manager, uint64_t id, uint64_t size, bool remote, FairMQ
        LOG(debug) << "shmem: created region: " << fName;
        fShmemObject.truncate(size);

-        fQueue = fair::mq::tools::make_unique<bipc::message_queue>(bipc::create_only, fQueueName.c_str(), 10000, sizeof(RegionBlock));
+        fQueue = fair::mq::tools::make_unique<bipc::message_queue>(bipc::create_only, fQueueName.c_str(), 1024, fAckBunchSize * sizeof(RegionBlock));
        LOG(debug) << "shmem: created region queue: " << fQueueName;
    }
    fRegion = bipc::mapped_region(fShmemObject, bipc::read_write); // TODO: add HUGEPAGES flag here
    // fRegion = bipc::mapped_region(fShmemObject, bipc::read_write, 0, 0, 0, MAP_ANONYMOUS | MAP_HUGETLB);
+
+    fSendAcksWorker = std::thread(&Region::SendAcks, this);
 }

 void Region::StartReceivingAcks()
 {
-    fWorker = std::thread(&Region::ReceiveAcks, this);
+    fReceiveAcksWorker = std::thread(&Region::ReceiveAcks, this);
 }

 void Region::ReceiveAcks()
 {
    unsigned int priority;
    bipc::message_queue::size_type recvdSize;
+    std::unique_ptr<RegionBlock[]> blocks = fair::mq::tools::make_unique<RegionBlock[]>(fAckBunchSize);

    while (!fStop) // end thread condition (should exist until region is destroyed)
    {
-        auto rcvTill = bpt::microsec_clock::universal_time() + bpt::milliseconds(200);
-        RegionBlock block;
-        if (fQueue->timed_receive(&block, sizeof(RegionBlock), recvdSize, priority, rcvTill))
+        auto rcvTill = bpt::microsec_clock::universal_time() + bpt::milliseconds(500);
+
+        while (fQueue->timed_receive(blocks.get(), fAckBunchSize * sizeof(RegionBlock), recvdSize, priority, rcvTill))
        {
            // LOG(debug) << "received: " << block.fHandle << " " << block.fSize << " " << block.fMessageId;
            if (fCallback)
            {
-                fCallback(reinterpret_cast<char*>(fRegion.get_address()) + block.fHandle, block.fSize, reinterpret_cast<void*>(block.fHint));
+                const auto numBlocks = recvdSize / sizeof(RegionBlock);
+                for (size_t i = 0; i < numBlocks; i++)
+                {
+                    fCallback(reinterpret_cast<char*>(fRegion.get_address()) + blocks[i].fHandle, blocks[i].fSize, reinterpret_cast<void*>(blocks[i].fHint));
+                }
            }
        }
-        else
-        {
-            // LOG(debug) << "queue " << fQueueName << " timeout!";
-        }
    } // while !fStop

-    LOG(debug) << "worker for " << fName << " leaving.";
+    LOG(debug) << "receive ack worker for " << fName << " leaving.";
+}
+
+void Region::ReleaseBlock(const RegionBlock &block)
+{
+    std::unique_lock<std::mutex> lock(fBlockLock);
+
+    fBlocksToFree.emplace_back(block);
+
+    if (fBlocksToFree.size() >= fAckBunchSize)
+    {
+        lock.unlock(); // reduces contention on fBlockLock
+        fBlockSendCV.notify_one();
+    }
+}
+
+void Region::SendAcks()
+{
+    std::unique_ptr<RegionBlock[]> blocks = fair::mq::tools::make_unique<RegionBlock[]>(fAckBunchSize);
+
+    while (true) // we'll try to send all acks before stopping
+    {
+        size_t blocksToSend = 0;
+
+        {   // mutex locking block
+            std::unique_lock<std::mutex> lock(fBlockLock);
+
+            // 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
+                fBlockSendCV.wait_for(lock, std::chrono::milliseconds(500));
+            }
+
+            blocksToSend = std::min(fBlocksToFree.size(), fAckBunchSize);
+
+            std::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...
+                this_thread::yield();
+            }
+        }
+        else // blocksToSend == 0
+        {
+            if (fStop)
+            {
+                break;
+            }
+        }
+    }
+
+    LOG(debug) << "send ack worker for " << fName << " leaving.";
 }

 Region::~Region()
 {
+    fStop = true;
+
+    if (fSendAcksWorker.joinable())
+    {
+        fSendAcksWorker.join();
+    }
+
    if (!fRemote)
    {
-        fStop = true;
-        if (fWorker.joinable())
+        if (fReceiveAcksWorker.joinable())
        {
-            fWorker.join();
+            fReceiveAcksWorker.join();
        }

        if (bipc::shared_memory_object::remove(fName.c_str()))
--- a/fairmq/shmem/Region.h
+++ b/fairmq/shmem/Region.h
@@ -19,11 +19,14 @@
 #include "FairMQUnmanagedRegion.h"

 #include <fairmq/Tools.h>
+#include <fairmq/shmem/Common.h>

 #include <boost/interprocess/managed_shared_memory.hpp>
 #include <boost/interprocess/ipc/message_queue.hpp>

 #include <thread>
+#include <mutex>
+#include <condition_variable>
 #include <unordered_map>

 namespace fair
@@ -47,6 +50,9 @@ struct Region
    void StartReceivingAcks();
    void ReceiveAcks();

+    void ReleaseBlock(const RegionBlock &);
+    void SendAcks();
+
    ~Region();

    Manager& fManager;
@@ -56,8 +62,15 @@ struct Region
    std::string fQueueName;
    boost::interprocess::shared_memory_object fShmemObject;
    boost::interprocess::mapped_region fRegion;
+
+    std::mutex fBlockLock;
+    std::condition_variable fBlockSendCV;
+    std::vector<RegionBlock> fBlocksToFree;
+    const std::size_t fAckBunchSize = 256;
    std::unique_ptr<boost::interprocess::message_queue> fQueue;
-    std::thread fWorker;
+
+    std::thread fReceiveAcksWorker;
+    std::thread fSendAcksWorker;
    FairMQRegionCallback fCallback;
 };

--- a/fairmq/tools/CppSTL.h
+++ b/fairmq/tools/CppSTL.h
@@ -27,6 +27,13 @@ std::unique_ptr<T> make_unique(Args&& ...args)
    return std::unique_ptr<T>(new T(std::forward<Args>(args)...));
 }

+// make_unique implementation (array variant), until C++14 is default
+template<typename T>
+std::unique_ptr<T> make_unique(std::size_t size)
+{
+    return std::unique_ptr<T>(new typename std::remove_extent<T>::type[size]());
+}
+
 // provide an enum hasher to compensate std::hash not supporting enums in C++11
 template<typename Enum>
 struct HashEnum
--- a/fairmq/tools/Process.cxx
+++ b/fairmq/tools/Process.cxx
@@ -7,8 +7,13 @@
 ********************************************************************************/

 #include <fairmq/tools/Process.h>
+#include <fairmq/tools/Strings.h>

 #include <boost/process.hpp>
+#include <boost/asio.hpp>
+#include <boost/date_time/posix_time/posix_time.hpp>
+
+#include <signal.h> // kill, signals

 #include <iostream>
 #include <sstream>
@@ -17,6 +22,28 @@

 using namespace std;
 namespace bp = boost::process;
+namespace ba = boost::asio;
+namespace bs = boost::system;
+
+class LinePrinter
+{
+  public:
+    LinePrinter(stringstream& out, const string& prefix)
+        : fOut(out)
+        , fPrefix(prefix)
+    {}
+
+    // prints line with prefix on both cout (thread-safe) and output stream
+    void Print(const string& line)
+    {
+        cout << fair::mq::tools::ToString(fPrefix, line, "\n") << flush;
+        fOut << fPrefix << line << endl;
+    }
+
+  private:
+    stringstream& fOut;
+    const string fPrefix;
+};

 namespace fair
 {
@@ -33,57 +60,111 @@ namespace tools
 * @param[in] log_prefix How to prefix each captured output line with
 * @return Captured stdout output and exit code
 */
-execute_result execute(const string& cmd, const string& prefix, const string& input)
+execute_result execute(const string& cmd, const string& prefix, const string& input, int sig)
 {
    execute_result result;
    stringstream out;

-    // print full line thread-safe
-    stringstream printCmd;
-    printCmd << prefix << " " << cmd << "\n";
-    cout << printCmd.str() << flush;
+    LinePrinter p(out, prefix);

-    out << prefix << cmd << endl;
+    p.Print(cmd);

-    // Execute command and capture stdout, add prefix line by line
-    bp::ipstream c_stdout;
-    bp::opstream c_stdin;
-    bp::child c(cmd, bp::std_out > c_stdout, bp::std_in < c_stdin);
+    ba::io_service ios;

-    while (c.valid() && !c.running()) {
-        ;
-    }
+    // containers for std_in
+    ba::const_buffer inputBuffer(ba::buffer(input));
+    bp::async_pipe inputPipe(ios);
+    // containers for std_out
+    ba::streambuf outputBuffer;
+    bp::async_pipe outputPipe(ios);
+    // containers for std_err
+    ba::streambuf errorBuffer;
+    bp::async_pipe errorPipe(ios);

-    if (!c.valid()) {
-        throw runtime_error("Can't execute the given process.");
-    }
+    const string delimiter = "\n";
+    ba::deadline_timer inputTimer(ios, boost::posix_time::milliseconds(100));
+    ba::deadline_timer signalTimer(ios, boost::posix_time::milliseconds(100));

-    // Optionally, write to stdin of the child
+    // child process
+    bp::child c(cmd, bp::std_out > outputPipe, bp::std_err > errorPipe, bp::std_in < inputPipe);
+    int pid = c.id();
+    p.Print(ToString("fair::mq::tools::execute: pid: ", pid));
+
+    // handle std_in with a delay
    if (input != "") {
-        this_thread::sleep_for(chrono::milliseconds(100));
-        c_stdin << input;
-        c_stdin.flush();
+        inputTimer.async_wait([&](const bs::error_code& ec1) {
+            if (!ec1) {
+                ba::async_write(inputPipe, inputBuffer, [&](const bs::error_code& ec2, size_t /* n */) {
+                    if (!ec2) {
+                        // inputPipe.async_close();
+                    } else {
+                        p.Print(ToString("error in boost::asio::async_write: ", ec2.message()));
+                    }
+                });
+            } else {
+                p.Print(ToString("error in boost::asio::deadline_timer.async_wait: ", ec1.message()));
+            }
+        });
    }

-    string line;
-    while (c.running() && getline(c_stdout, line)) {
-        // print full line thread-safe
-        stringstream printLine;
-        printLine << prefix << line << "\n";
-        cout << printLine.str() << flush;
-
-        out << prefix << line << "\n";
+    if (sig != -1) {
+        signalTimer.async_wait([&](const bs::error_code& ec1) {
+            if (!ec1) {
+                kill(pid, sig);
+            } else {
+                p.Print(ToString("error in boost::asio::deadline_timer.async_wait: ", ec1.message()));
+            }
+        });
    }

+    // handle std_out line by line
+    function<void(const bs::error_code&, size_t)> onStdOut = [&](const bs::error_code& ec, size_t /* n */) {
+        if (!ec) {
+            istream is(&outputBuffer);
+            string line;
+            getline(is, line);
+
+            p.Print(line);
+
+            ba::async_read_until(outputPipe, outputBuffer, delimiter, onStdOut);
+        } else {
+            if (ec == ba::error::eof) {
+                // outputPipe.async_close();
+            } else {
+                p.Print(ec.message());
+            }
+        }
+    };
+    ba::async_read_until(outputPipe, outputBuffer, delimiter, onStdOut);
+
+    // handle std_err line by line
+    function<void(const bs::error_code&, size_t)> onStdErr = [&](const bs::error_code& ec, size_t /* n */) {
+        if (!ec) {
+            istream is(&errorBuffer);
+            string line;
+            getline(is, line);
+
+            p.Print(ToString("error: ", line));
+
+            ba::async_read_until(errorPipe, errorBuffer, delimiter, onStdErr);
+        } else {
+            if (ec == ba::error::eof) {
+                // errorPipe.async_close();
+            } else {
+                p.Print(ec.message());
+            }
+        }
+    };
+    ba::async_read_until(errorPipe, errorBuffer, delimiter, onStdErr);
+
+    ios.run();
    c.wait();

-    // Capture exit code
    result.exit_code = c.exit_code();
-    out << prefix << " Exit code: " << result.exit_code << endl;

+    p.Print(ToString("fair::mq::tools::execute: exit code: ", result.exit_code));
    result.console_out = out.str();

-    // Return result
    return result;
 }

--- a/fairmq/tools/Process.h
+++ b/fairmq/tools/Process.h
@@ -38,7 +38,8 @@ struct execute_result
 */
 execute_result execute(const std::string& cmd,
                       const std::string& prefix = "",
-                       const std::string& input = "");
+                       const std::string& input = "",
+                       int sig = -1);

 } /* namespace tools */
 } /* namespace mq */
--- a/test/CMakeLists.txt
+++ b/test/CMakeLists.txt
@@ -105,6 +105,8 @@ add_testsuite(Device
    device/_config.cxx
    device/_waitfor.cxx
    device/_exceptions.cxx
+    device/_error_state.cxx
+    device/_signals.cxx

    LINKS FairMQ
    DEPENDS testhelper_runTestDevice
--- a/test/device/TestReceiver.h
+++ b/test/device/TestReceiver.h
@@ -38,12 +38,9 @@ class Receiver : public FairMQDevice
    auto Run() -> void override
    {
        auto msg = FairMQMessagePtr{NewMessage()};
-        if (Receive(msg, fChannelName) >= 0)
-        {
+        if (Receive(msg, fChannelName) >= 0) {
            LOG(info) << "received empty message";
-        }
-        else
-        {
+        } else {
            LOG(error) << "fair::mq::test::Receiver::Run(): Receive(msg, fChannelName) < 0";
        }
    };
--- a/test/device/TestSender.h
+++ b/test/device/TestSender.h
@@ -38,12 +38,9 @@ class Sender : public FairMQDevice
    auto Run() -> void override
    {
        auto msg = FairMQMessagePtr{NewMessage()};
-        if (Send(msg, fChannelName) >= 0)
-        {
+        if (Send(msg, fChannelName) >= 0) {
            LOG(info) << "sent empty message";
-        }
-        else
-        {
+        } else {
            LOG(error) << "fair::mq::test::Sender::Run(): Send(msg, fChannelName) < 0";
        }
    };
--- a/test/device/_error_state.cxx
+++ b/test/device/_error_state.cxx
@@ -0,0 +1,121 @@
+/********************************************************************************
+ *    Copyright (C) 2018 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"                       *
+ ********************************************************************************/
+
+#include "runner.h"
+
+#include <gtest/gtest.h>
+#include <boost/process.hpp>
+#include <fairmq/Tools.h>
+
+#include <string>
+#include <thread>
+#include <iostream>
+
+namespace
+{
+
+using namespace std;
+using namespace fair::mq::test;
+using namespace fair::mq::tools;
+
+void RunErrorStateIn(const string& state, const string& control, const string& input = "")
+{
+    size_t session{fair::mq::tools::UuidHash()};
+
+    execute_result result{"", 100};
+    thread device_thread([&]() {
+        stringstream cmd;
+        cmd << runTestDevice
+            << " --id error_state_" << state << "_"
+            << " --control " << control
+            << " --session " << session
+            << " --color false";
+        result = execute(cmd.str(), "[ErrorFound IN " + state + "]", input);
+    });
+
+    device_thread.join();
+
+    ASSERT_NE(string::npos, result.console_out.find("going to change to Error state from " + state + "()"));
+
+    exit(result.exit_code);
+}
+
+TEST(ErrorState, static_InInit)
+{
+    EXPECT_EXIT(RunErrorStateIn("Init", "static"), ::testing::ExitedWithCode(1), "");
+}
+TEST(ErrorState, static_InBind)
+{
+    EXPECT_EXIT(RunErrorStateIn("Bind", "static"), ::testing::ExitedWithCode(1), "");
+}
+TEST(ErrorState, static_InConnect)
+{
+    EXPECT_EXIT(RunErrorStateIn("Connect", "static"), ::testing::ExitedWithCode(1), "");
+}
+TEST(ErrorState, static_InInitTask)
+{
+    EXPECT_EXIT(RunErrorStateIn("InitTask", "static"), ::testing::ExitedWithCode(1), "");
+}
+TEST(ErrorState, static_InPreRun)
+{
+    EXPECT_EXIT(RunErrorStateIn("PreRun", "static"), ::testing::ExitedWithCode(1), "");
+}
+TEST(ErrorState, static_InRun)
+{
+    EXPECT_EXIT(RunErrorStateIn("Run", "static"), ::testing::ExitedWithCode(1), "");
+}
+TEST(ErrorState, static_InPostRun)
+{
+    EXPECT_EXIT(RunErrorStateIn("PostRun", "static"), ::testing::ExitedWithCode(1), "");
+}
+TEST(ErrorState, static_InResetTask)
+{
+    EXPECT_EXIT(RunErrorStateIn("ResetTask", "static"), ::testing::ExitedWithCode(1), "");
+}
+TEST(ErrorState, static_InReset)
+{
+    EXPECT_EXIT(RunErrorStateIn("Reset", "static"), ::testing::ExitedWithCode(1), "");
+}
+TEST(ErrorState, interactive_InInit)
+{
+    EXPECT_EXIT(RunErrorStateIn("Init", "interactive"), ::testing::ExitedWithCode(1), "");
+}
+TEST(ErrorState, interactive_InBind)
+{
+    EXPECT_EXIT(RunErrorStateIn("Bind", "interactive"), ::testing::ExitedWithCode(1), "");
+}
+TEST(ErrorState, interactive_InConnect)
+{
+    EXPECT_EXIT(RunErrorStateIn("Connect", "interactive"), ::testing::ExitedWithCode(1), "");
+}
+TEST(ErrorState, interactive_InInitTask)
+{
+    EXPECT_EXIT(RunErrorStateIn("InitTask", "interactive"), ::testing::ExitedWithCode(1), "");
+}
+TEST(ErrorState, interactive_InPreRun)
+{
+    EXPECT_EXIT(RunErrorStateIn("PreRun", "interactive"), ::testing::ExitedWithCode(1), "");
+}
+TEST(ErrorState, interactive_InRun)
+{
+    EXPECT_EXIT(RunErrorStateIn("Run", "interactive"), ::testing::ExitedWithCode(1), "");
+}
+TEST(ErrorState, interactive_InPostRun)
+{
+    EXPECT_EXIT(RunErrorStateIn("PostRun", "interactive"), ::testing::ExitedWithCode(1), "");
+}
+TEST(ErrorState, interactive_InResetTask)
+{
+    EXPECT_EXIT(RunErrorStateIn("ResetTask", "interactive", "q"), ::testing::ExitedWithCode(1), "");
+}
+TEST(ErrorState, interactive_InReset)
+{
+    EXPECT_EXIT(RunErrorStateIn("Reset", "interactive", "q"), ::testing::ExitedWithCode(1), "");
+}
+
+} // namespace
--- a/test/device/_exceptions.cxx
+++ b/test/device/_exceptions.cxx
@@ -23,7 +23,7 @@ using namespace std;
 using namespace fair::mq::test;
 using namespace fair::mq::tools;

-void RunExceptionIn(const std::string& state, const std::string& input = "")
+void RunExceptionIn(const string& state, const string& control, const string& input = "")
 {
    size_t session{fair::mq::tools::UuidHash()};

@@ -32,7 +32,7 @@ void RunExceptionIn(const std::string& state, const std::string& input = "")
        stringstream cmd;
        cmd << runTestDevice
            << " --id exceptions_" << state << "_"
-            << " --control " << ((input == "") ? "static" : "interactive")
+            << " --control " << control
            << " --session " << session
            << " --color false";
        result = execute(cmd.str(), "[EXCEPTION IN " + state + "]", input);
@@ -40,86 +40,82 @@ void RunExceptionIn(const std::string& state, const std::string& input = "")

    device_thread.join();

-    ASSERT_NE(std::string::npos, result.console_out.find("exception in " + state + "()"));
+    ASSERT_NE(string::npos, result.console_out.find("exception in " + state + "()"));

    exit(result.exit_code);
 }

 TEST(Exceptions, static_InInit)
 {
-    EXPECT_EXIT(RunExceptionIn("Init"), ::testing::ExitedWithCode(1), "");
+    EXPECT_EXIT(RunExceptionIn("Init", "static"), ::testing::ExitedWithCode(1), "");
+}
+TEST(Exceptions, static_InBind)
+{
+    EXPECT_EXIT(RunExceptionIn("Bind", "static"), ::testing::ExitedWithCode(1), "");
+}
+TEST(Exceptions, static_InConnect)
+{
+    EXPECT_EXIT(RunExceptionIn("Connect", "static"), ::testing::ExitedWithCode(1), "");
 }
 TEST(Exceptions, static_InInitTask)
 {
-    EXPECT_EXIT(RunExceptionIn("InitTask"), ::testing::ExitedWithCode(1), "");
+    EXPECT_EXIT(RunExceptionIn("InitTask", "static"), ::testing::ExitedWithCode(1), "");
 }
 TEST(Exceptions, static_InPreRun)
 {
-    EXPECT_EXIT(RunExceptionIn("PreRun"), ::testing::ExitedWithCode(1), "");
+    EXPECT_EXIT(RunExceptionIn("PreRun", "static"), ::testing::ExitedWithCode(1), "");
 }
 TEST(Exceptions, static_InRun)
 {
-    EXPECT_EXIT(RunExceptionIn("Run"), ::testing::ExitedWithCode(1), "");
+    EXPECT_EXIT(RunExceptionIn("Run", "static"), ::testing::ExitedWithCode(1), "");
 }
 TEST(Exceptions, static_InPostRun)
 {
-    EXPECT_EXIT(RunExceptionIn("PostRun"), ::testing::ExitedWithCode(1), "");
+    EXPECT_EXIT(RunExceptionIn("PostRun", "static"), ::testing::ExitedWithCode(1), "");
 }
 TEST(Exceptions, static_InResetTask)
 {
-    EXPECT_EXIT(RunExceptionIn("ResetTask"), ::testing::ExitedWithCode(1), "");
+    EXPECT_EXIT(RunExceptionIn("ResetTask", "static"), ::testing::ExitedWithCode(1), "");
 }
 TEST(Exceptions, static_InReset)
 {
-    EXPECT_EXIT(RunExceptionIn("Reset"), ::testing::ExitedWithCode(1), "");
+    EXPECT_EXIT(RunExceptionIn("Reset", "static"), ::testing::ExitedWithCode(1), "");
 }
 TEST(Exceptions, interactive_InInit)
 {
-    EXPECT_EXIT(RunExceptionIn("Init", "q"), ::testing::ExitedWithCode(1), "");
+    EXPECT_EXIT(RunExceptionIn("Init", "interactive"), ::testing::ExitedWithCode(1), "");
+}
+TEST(Exceptions, interactive_InBind)
+{
+    EXPECT_EXIT(RunExceptionIn("Bind", "interactive"), ::testing::ExitedWithCode(1), "");
+}
+TEST(Exceptions, interactive_InConnect)
+{
+    EXPECT_EXIT(RunExceptionIn("Connect", "interactive"), ::testing::ExitedWithCode(1), "");
 }
 TEST(Exceptions, interactive_InInitTask)
 {
-    EXPECT_EXIT(RunExceptionIn("InitTask", "q"), ::testing::ExitedWithCode(1), "");
+    EXPECT_EXIT(RunExceptionIn("InitTask", "interactive"), ::testing::ExitedWithCode(1), "");
 }
 TEST(Exceptions, interactive_InPreRun)
 {
-    EXPECT_EXIT(RunExceptionIn("PreRun", "q"), ::testing::ExitedWithCode(1), "");
+    EXPECT_EXIT(RunExceptionIn("PreRun", "interactive"), ::testing::ExitedWithCode(1), "");
 }
 TEST(Exceptions, interactive_InRun)
 {
-    EXPECT_EXIT(RunExceptionIn("Run", "q"), ::testing::ExitedWithCode(1), "");
+    EXPECT_EXIT(RunExceptionIn("Run", "interactive"), ::testing::ExitedWithCode(1), "");
 }
 TEST(Exceptions, interactive_InPostRun)
 {
-    EXPECT_EXIT(RunExceptionIn("PostRun", "q"), ::testing::ExitedWithCode(1), "");
+    EXPECT_EXIT(RunExceptionIn("PostRun", "interactive"), ::testing::ExitedWithCode(1), "");
 }
 TEST(Exceptions, interactive_InResetTask)
 {
-    EXPECT_EXIT(RunExceptionIn("ResetTask", "q"), ::testing::ExitedWithCode(1), "");
+    EXPECT_EXIT(RunExceptionIn("ResetTask", "interactive", "q"), ::testing::ExitedWithCode(1), "");
 }
 TEST(Exceptions, interactive_InReset)
 {
-    EXPECT_EXIT(RunExceptionIn("Reset", "q"), ::testing::ExitedWithCode(1), "");
-}
-TEST(Exceptions, interactive_invalid_InInit)
-{
-    EXPECT_EXIT(RunExceptionIn("Init", "_"), ::testing::ExitedWithCode(1), "");
-}
-TEST(Exceptions, interactive_invalid_InInitTask)
-{
-    EXPECT_EXIT(RunExceptionIn("InitTask", "_"), ::testing::ExitedWithCode(1), "");
-}
-TEST(Exceptions, interactive_invalid_InPreRun)
-{
-    EXPECT_EXIT(RunExceptionIn("PreRun", "_"), ::testing::ExitedWithCode(1), "");
-}
-TEST(Exceptions, interactive_invalid_InRun)
-{
-    EXPECT_EXIT(RunExceptionIn("Run", "_"), ::testing::ExitedWithCode(1), "");
-}
-TEST(Exceptions, interactive_invalid_InPostRun)
-{
-    EXPECT_EXIT(RunExceptionIn("PostRun", "_"), ::testing::ExitedWithCode(1), "");
+    EXPECT_EXIT(RunExceptionIn("Reset", "interactive", "q"), ::testing::ExitedWithCode(1), "");
 }

 } // namespace
--- a/test/device/_signals.cxx
+++ b/test/device/_signals.cxx
@@ -0,0 +1,121 @@
+/********************************************************************************
+ *    Copyright (C) 2018 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"                       *
+ ********************************************************************************/
+
+#include "runner.h"
+
+#include <gtest/gtest.h>
+#include <boost/process.hpp>
+#include <fairmq/Tools.h>
+
+#include <string>
+#include <thread>
+#include <iostream>
+
+namespace
+{
+
+using namespace std;
+using namespace fair::mq::test;
+using namespace fair::mq::tools;
+
+void RunSignalIn(const string& state, const string& control, const string& input = "")
+{
+    size_t session{fair::mq::tools::UuidHash()};
+
+    execute_result result{"", 100};
+    thread device_thread([&]() {
+        stringstream cmd;
+        cmd << runTestDevice
+            << " --id signals_" << state << "_"
+            << " --control " << control
+            << " --session " << session
+            << " --color false";
+        result = execute(cmd.str(), "[SIGINT IN " + state + "]", input);
+    });
+
+    device_thread.join();
+
+    ASSERT_NE(string::npos, result.console_out.find("raising SIGINT from " + state + "()"));
+
+    exit(result.exit_code);
+}
+
+TEST(Signal_SIGINT, static_InInit)
+{
+    EXPECT_EXIT(RunSignalIn("Init", "static"), ::testing::ExitedWithCode(0), "");
+}
+TEST(Signal_SIGINT, static_InBind)
+{
+    EXPECT_EXIT(RunSignalIn("Bind", "static"), ::testing::ExitedWithCode(0), "");
+}
+TEST(Signal_SIGINT, static_InConnect)
+{
+    EXPECT_EXIT(RunSignalIn("Connect", "static"), ::testing::ExitedWithCode(0), "");
+}
+TEST(Signal_SIGINT, static_InInitTask)
+{
+    EXPECT_EXIT(RunSignalIn("InitTask", "static"), ::testing::ExitedWithCode(0), "");
+}
+TEST(Signal_SIGINT, static_InPreRun)
+{
+    EXPECT_EXIT(RunSignalIn("PreRun", "static"), ::testing::ExitedWithCode(0), "");
+}
+TEST(Signal_SIGINT, static_InRun)
+{
+    EXPECT_EXIT(RunSignalIn("Run", "static"), ::testing::ExitedWithCode(0), "");
+}
+TEST(Signal_SIGINT, static_InPostRun)
+{
+    EXPECT_EXIT(RunSignalIn("PostRun", "static"), ::testing::ExitedWithCode(0), "");
+}
+TEST(Signal_SIGINT, static_InResetTask)
+{
+    EXPECT_EXIT(RunSignalIn("ResetTask", "static"), ::testing::ExitedWithCode(0), "");
+}
+TEST(Signal_SIGINT, static_InReset)
+{
+    EXPECT_EXIT(RunSignalIn("Reset", "static"), ::testing::ExitedWithCode(0), "");
+}
+TEST(Signal_SIGINT, interactive_InInit)
+{
+    EXPECT_EXIT(RunSignalIn("Init", "interactive"), ::testing::ExitedWithCode(0), "");
+}
+TEST(Signal_SIGINT, interactive_InBind)
+{
+    EXPECT_EXIT(RunSignalIn("Bind", "interactive"), ::testing::ExitedWithCode(0), "");
+}
+TEST(Signal_SIGINT, interactive_InConnect)
+{
+    EXPECT_EXIT(RunSignalIn("Connect", "interactive"), ::testing::ExitedWithCode(0), "");
+}
+TEST(Signal_SIGINT, interactive_InInitTask)
+{
+    EXPECT_EXIT(RunSignalIn("InitTask", "interactive"), ::testing::ExitedWithCode(0), "");
+}
+TEST(Signal_SIGINT, interactive_InPreRun)
+{
+    EXPECT_EXIT(RunSignalIn("PreRun", "interactive"), ::testing::ExitedWithCode(0), "");
+}
+TEST(Signal_SIGINT, interactive_InRun)
+{
+    EXPECT_EXIT(RunSignalIn("Run", "interactive"), ::testing::ExitedWithCode(0), "");
+}
+TEST(Signal_SIGINT, interactive_InPostRun)
+{
+    EXPECT_EXIT(RunSignalIn("PostRun", "interactive"), ::testing::ExitedWithCode(0), "");
+}
+TEST(Signal_SIGINT, interactive_InResetTask)
+{
+    EXPECT_EXIT(RunSignalIn("ResetTask", "interactive", "q"), ::testing::ExitedWithCode(0), "");
+}
+TEST(Signal_SIGINT, interactive_InReset)
+{
+    EXPECT_EXIT(RunSignalIn("Reset", "interactive", "q"), ::testing::ExitedWithCode(0), "");
+}
+
+} // namespace
--- a/test/device/_waitfor.cxx
+++ b/test/device/_waitfor.cxx
@@ -9,66 +9,65 @@
 #include "runner.h"

 #include <gtest/gtest.h>
-#include <boost/process.hpp>
+#include <fairmq/Tools.h>

-#include <sys/types.h>
-#include <signal.h>
+#include <signal.h> // kill

 #include <string>
 #include <thread>
-#include <chrono>
 #include <iostream>
-#include <future> // std::async, std::future

 namespace
 {

 using namespace std;
 using namespace fair::mq::test;
+using namespace fair::mq::tools;

-void RunWaitFor()
+void RunWaitFor(const string& state, const string& control)
 {
-    std::mutex mtx;
-    std::condition_variable cv;
-
-    int pid = 0;
-    int exit_code = 0;
-
-    thread deviceThread([&]() {
+    execute_result result;
+    thread device_thread([&] {
        stringstream cmd;
-        cmd << runTestDevice << " --id waitfor_" << " --control static " << " --severity nolog";
-
-        boost::process::ipstream stdout;
-        boost::process::child c(cmd.str(), boost::process::std_out > stdout);
-        string line;
-        getline(stdout, line);
-
-        {
-            std::lock_guard<std::mutex> lock(mtx);
-            pid = c.id();
-        }
-        cv.notify_one();
-
-        c.wait();
-
-        exit_code = c.exit_code();
+        cmd << runTestDevice
+            << " --id waitfor_" << state
+            << " --control " << control
+            << " --session " << UuidHash()
+            << " --severity debug"
+            << " --color false";
+        result = execute(cmd.str(), "[WaitFor]", "", SIGINT);
    });

-    {
-        std::unique_lock<std::mutex> lock(mtx);
-        cv.wait(lock, [&pid]{ return pid != 0; });
-    }
+    device_thread.join();

-    kill(pid, SIGINT);
+    ASSERT_NE(string::npos, result.console_out.find("Sleeping Done. Interrupted."));

-    deviceThread.join();
-
-    exit(exit_code);
+    exit(result.exit_code);
 }

-TEST(Device, WaitFor)
+TEST(WaitFor, static_InPreRun)
 {
-    EXPECT_EXIT(RunWaitFor(), ::testing::ExitedWithCode(0), "");
+    EXPECT_EXIT(RunWaitFor("PreRun", "static"), ::testing::ExitedWithCode(0), "");
+}
+TEST(WaitFor, static_InRun)
+{
+    EXPECT_EXIT(RunWaitFor("Run", "static"), ::testing::ExitedWithCode(0), "");
+}
+TEST(WaitFor, static_InPostRun)
+{
+    EXPECT_EXIT(RunWaitFor("PostRun", "static"), ::testing::ExitedWithCode(0), "");
+}
+TEST(WaitFor, interactive_InPreRun)
+{
+    EXPECT_EXIT(RunWaitFor("PreRun", "interactive"), ::testing::ExitedWithCode(0), "");
+}
+TEST(WaitFor, interactive_InRun)
+{
+    EXPECT_EXIT(RunWaitFor("Run", "interactive"), ::testing::ExitedWithCode(0), "");
+}
+TEST(WaitFor, interactive_InPostRun)
+{
+    EXPECT_EXIT(RunWaitFor("PostRun", "interactive"), ::testing::ExitedWithCode(0), "");
 }

 } // namespace
--- a/test/helper/devices/TestErrorState.h
+++ b/test/helper/devices/TestErrorState.h
@@ -0,0 +1,113 @@
+/********************************************************************************
+ *   Copyright (C) 2018 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"                       *
+ ********************************************************************************/
+
+#ifndef FAIR_MQ_TEST_ERROR_STATE_H
+#define FAIR_MQ_TEST_ERROR_STATE_H
+
+#include <FairMQDevice.h>
+#include <FairMQLogger.h>
+
+#include <iostream>
+
+namespace fair
+{
+namespace mq
+{
+namespace test
+{
+
+class ErrorState : public FairMQDevice
+{
+  public:
+    void Init() override
+    {
+        std::string state("Init");
+        if (std::string::npos != GetId().find("_" + state + "_")) {
+            LOG(debug) << "going to change to Error state from " << state << "()";
+            ChangeState(fair::mq::Transition::ErrorFound);
+        }
+    }
+
+    void Bind() override
+    {
+        std::string state("Bind");
+        if (std::string::npos != GetId().find("_" + state + "_")) {
+            LOG(debug) << "going to change to Error state from " << state << "()";
+            ChangeState(fair::mq::Transition::ErrorFound);
+        }
+    }
+
+    void Connect() override
+    {
+        std::string state("Connect");
+        if (std::string::npos != GetId().find("_" + state + "_")) {
+            LOG(debug) << "going to change to Error state from " << state << "()";
+            ChangeState(fair::mq::Transition::ErrorFound);
+        }
+    }
+
+    void InitTask() override
+    {
+        std::string state("InitTask");
+        if (std::string::npos != GetId().find("_" + state + "_")) {
+            LOG(debug) << "going to change to Error state from " << state << "()";
+            ChangeState(fair::mq::Transition::ErrorFound);
+        }
+    }
+
+    void PreRun() override
+    {
+        std::string state("PreRun");
+        if (std::string::npos != GetId().find("_" + state + "_")) {
+            LOG(debug) << "going to change to Error state from " << state << "()";
+            ChangeState(fair::mq::Transition::ErrorFound);
+        }
+    }
+
+    void Run() override
+    {
+        std::string state("Run");
+        if (std::string::npos != GetId().find("_" + state + "_")) {
+            LOG(debug) << "going to change to Error state from " << state << "()";
+            ChangeState(fair::mq::Transition::ErrorFound);
+        }
+    }
+
+    void PostRun() override
+    {
+        std::string state("PostRun");
+        if (std::string::npos != GetId().find("_" + state + "_")) {
+            LOG(debug) << "going to change to Error state from " << state << "()";
+            ChangeState(fair::mq::Transition::ErrorFound);
+        }
+    }
+
+    void ResetTask() override
+    {
+        std::string state("ResetTask");
+        if (std::string::npos != GetId().find("_" + state + "_")) {
+            LOG(debug) << "going to change to Error state from " << state << "()";
+            ChangeState(fair::mq::Transition::ErrorFound);
+        }
+    }
+
+    void Reset() override
+    {
+        std::string state("Reset");
+        if (std::string::npos != GetId().find("_" + state + "_")) {
+            LOG(debug) << "going to change to Error state from " << state << "()";
+            ChangeState(fair::mq::Transition::ErrorFound);
+        }
+    }
+};
+
+} // namespace test
+} // namespace mq
+} // namespace fair
+
+#endif /* FAIR_MQ_TEST_ERROR_STATE_H */
--- a/test/helper/devices/TestExceptions.h
+++ b/test/helper/devices/TestExceptions.h
@@ -33,6 +33,22 @@ class Exceptions : public FairMQDevice
        }
    }

+    auto Bind() -> void override
+    {
+        std::string state("Bind");
+        if (std::string::npos != GetId().find("_" + state + "_")) {
+            throw std::runtime_error("exception in " + state + "()");
+        }
+    }
+
+    auto Connect() -> void override
+    {
+        std::string state("Connect");
+        if (std::string::npos != GetId().find("_" + state + "_")) {
+            throw std::runtime_error("exception in " + state + "()");
+        }
+    }
+
    auto InitTask() -> void override
    {
        std::string state("InitTask");
--- a/test/helper/devices/TestPairLeft.h
+++ b/test/helper/devices/TestPairLeft.h
@@ -57,6 +57,7 @@ class PairLeft : public FairMQDevice
        if (counter == 6) LOG(info) << "Simple message with short text data successfull";

        assert(counter == 6);
+        if (counter == 6) LOG(info) << "PAIR test successfull.";
    };
 };

--- a/test/helper/devices/TestPairRight.h
+++ b/test/helper/devices/TestPairRight.h
@@ -55,9 +55,6 @@ class PairRight : public FairMQDevice
        auto msg6(NewSimpleMessageFor("data", 0, "testdata1234"));
        if (Send(msg6, "data") >= 0) counter++;
        if (counter == 6) LOG(info) << "Simple message with short text data successfull";
-
-        std::this_thread::sleep_for(std::chrono::milliseconds(1000));
-        if (counter == 6) LOG(info) << "PAIR test successfull.";
    };
 };

@@ -65,4 +62,4 @@ class PairRight : public FairMQDevice
 } // namespace mq
 } // namespace fair

-#endif /* FAIR_MQ_TEST_PAIRRIGHT_H */
+#endif /* FAIR_MQ_TEST_PAIRRIGHT_H */
--- a/test/helper/devices/TestSignals.h
+++ b/test/helper/devices/TestSignals.h
@@ -0,0 +1,112 @@
+/********************************************************************************
+ *   Copyright (C) 2018 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"                       *
+ ********************************************************************************/
+
+#ifndef FAIR_MQ_TEST_SIGNALS_H
+#define FAIR_MQ_TEST_SIGNALS_H
+
+#include <FairMQDevice.h>
+#include <FairMQLogger.h>
+
+#include <iostream>
+#include <csignal>
+
+namespace fair
+{
+namespace mq
+{
+namespace test
+{
+
+class Signals : public FairMQDevice
+{
+  public:
+    void Init() override
+    {
+        std::string state("Init");
+        if (std::string::npos != GetId().find("_" + state + "_")) {
+            LOG(debug) << "raising SIGINT from " << state << "()";
+            raise(SIGINT);
+        }
+    }
+    void Bind() override
+    {
+        std::string state("Bind");
+        if (std::string::npos != GetId().find("_" + state + "_")) {
+            LOG(debug) << "raising SIGINT from " << state << "()";
+            raise(SIGINT);
+        }
+    }
+    void Connect() override
+    {
+        std::string state("Connect");
+        if (std::string::npos != GetId().find("_" + state + "_")) {
+            LOG(debug) << "raising SIGINT from " << state << "()";
+            raise(SIGINT);
+        }
+    }
+
+    void InitTask() override
+    {
+        std::string state("InitTask");
+        if (std::string::npos != GetId().find("_" + state + "_")) {
+            LOG(debug) << "raising SIGINT from " << state << "()";
+            raise(SIGINT);
+        }
+    }
+
+    void PreRun() override
+    {
+        std::string state("PreRun");
+        if (std::string::npos != GetId().find("_" + state + "_")) {
+            LOG(debug) << "raising SIGINT from " << state << "()";
+            raise(SIGINT);
+        }
+    }
+
+    void Run() override
+    {
+        std::string state("Run");
+        if (std::string::npos != GetId().find("_" + state + "_")) {
+            LOG(debug) << "raising SIGINT from " << state << "()";
+            raise(SIGINT);
+        }
+    }
+
+    void PostRun() override
+    {
+        std::string state("PostRun");
+        if (std::string::npos != GetId().find("_" + state + "_")) {
+            LOG(debug) << "raising SIGINT from " << state << "()";
+            raise(SIGINT);
+        }
+    }
+
+    void ResetTask() override
+    {
+        std::string state("ResetTask");
+        if (std::string::npos != GetId().find("_" + state + "_")) {
+            LOG(debug) << "raising SIGINT from " << state << "()";
+            raise(SIGINT);
+        }
+    }
+
+    void Reset() override
+    {
+        std::string state("Reset");
+        if (std::string::npos != GetId().find("_" + state + "_")) {
+            LOG(debug) << "raising SIGINT from " << state << "()";
+            raise(SIGINT);
+        }
+    }
+};
+
+} // namespace test
+} // namespace mq
+} // namespace fair
+
+#endif /* FAIR_MQ_TEST_SIGNALS_H */
--- a/test/helper/devices/TestWaitFor.h
+++ b/test/helper/devices/TestWaitFor.h
@@ -24,10 +24,34 @@ namespace test
 class TestWaitFor : public FairMQDevice
 {
  public:
-    void Run()
+    void PreRun() override
    {
-        std::cout << "hello" << std::endl;
-        WaitFor(std::chrono::seconds(60));
+        std::string state("PreRun");
+        if (std::string::npos != GetId().find("_" + state)) {
+            LOG(info) << "Going to sleep via WaitFor() in " << state << "...";
+            bool result = WaitFor(std::chrono::seconds(60));
+            LOG(info) << (result == true ? "Sleeping Done. Not interrupted." : "Sleeping Done. Interrupted.");
+        }
+    }
+
+    void Run() override
+    {
+        std::string state("Run");
+        if (std::string::npos != GetId().find("_" + state)) {
+            LOG(info) << "Going to sleep via WaitFor() in " << state << "...";
+            bool result = WaitFor(std::chrono::seconds(60));
+            LOG(info) << (result == true ? "Sleeping Done. Not interrupted." : "Sleeping Done. Interrupted.");
+        }
+    }
+
+    void PostRun() override
+    {
+        std::string state("PostRun");
+        if (std::string::npos != GetId().find("_" + state)) {
+            LOG(info) << "Going to sleep via WaitFor() in " << state << "...";
+            bool result = WaitFor(std::chrono::seconds(60));
+            LOG(info) << (result == true ? "Sleeping Done. Not interrupted." : "Sleeping Done. Interrupted.");
+        }
    }
 };

--- a/test/helper/runTestDevice.cxx
+++ b/test/helper/runTestDevice.cxx
@@ -19,6 +19,8 @@
 #include "devices/TestTransferTimeout.h"
 #include "devices/TestWaitFor.h"
 #include "devices/TestExceptions.h"
+#include "devices/TestErrorState.h"
+#include "devices/TestSignals.h"

 #include <runFairMQDevice.h>

@@ -40,60 +42,38 @@ auto getDevice(const FairMQProgOptions& config) -> FairMQDevicePtr
    using namespace fair::mq::test;

    auto id = config.GetValue<std::string>("id");
-    if (0 == id.find("pull_"))
-    {
+
+    if (0 == id.find("pull_")) {
        return new Pull;
-    }
-    else if (0 == id.find("push_"))
-    {
+    } else if (0 == id.find("push_")) {
        return new Push;
-    }
-    else if (0 == id.find("sub_"))
-    {
+    } else if (0 == id.find("sub_")) {
        return new Sub;
-    }
-    else if (0 == id.find("pub_"))
-    {
+    } else if (0 == id.find("pub_")) {
        return new Pub;
-    }
-    else if (0 == id.find("req_"))
-    {
+    } else if (0 == id.find("req_")) {
        return new Req;
-    }
-    else if (0 == id.find("rep_"))
-    {
+    } else if (0 == id.find("rep_")) {
        return new Rep;
-    }
-    else if (0 == id.find("transfer_timeout_"))
-    {
+    } else if (0 == id.find("transfer_timeout_")) {
        return new TransferTimeout;
-    }
-    else if (0 == id.find("pollout_"))
-    {
+    } else if (0 == id.find("pollout_")) {
        return new PollOut;
-    }
-    else if (0 == id.find("pollin_"))
-    {
+    } else if (0 == id.find("pollin_")) {
        return new PollIn;
-    }
-    else if (0 == id.find("pairleft_"))
-    {
+    } else if (0 == id.find("pairleft_")) {
        return new PairLeft;
-    }
-    else if (0 == id.find("pairright_"))
-    {
+    } else if (0 == id.find("pairright_")) {
        return new PairRight;
-    }
-    else if (0 == id.find("waitfor_"))
-    {
+    } else if (0 == id.find("waitfor_")) {
        return new TestWaitFor;
-    }
-    else if (0 == id.find("exceptions_"))
-    {
+    } else if (0 == id.find("exceptions_")) {
        return new Exceptions;
-    }
-    else
-    {
+    } else if (0 == id.find("error_state_")) {
+        return new ErrorState;
+    } else if (0 == id.find("signals_")) {
+        return new Signals;
+    } else {
        cerr << "Don't know id '" << id << "'" << endl;
        return nullptr;
    }
Author	SHA1	Message	Date
Dennis Klein	99c8d33191	DDS plugin: Remove unused dependency	2019-06-13 17:47:47 +02:00
Dennis Klein	660420e4f3	CMake: Support running static analysis OFF by default (Imposes significant additional build time). Controlled with CMake variable RUN_STATIC_ANALYSIS. Supported analysers: * clang-tidy * iwyu * cpplint	2019-06-04 12:17:37 +02:00
Dennis Klein	f8f997abe6	CMake: Set policies by version This is the recommended way of dealing with policies. Also, this patch fixes a problem with the policies being set at the wrong depth in the policy stack.	2019-06-04 12:17:37 +02:00
Gvozden Neskovic	40f6db430a	Print channel name in error log	2019-06-01 06:20:39 +02:00
Gvozden Neskovic	2ed2177555	Batch Region ack messages Reduce CPU utilization by batching release ack messages on the IPC queue.	2019-05-14 14:56:08 +02:00
Gvozden Neskovic	9b326c7a71	add array variant of fair::mq::tools::make_unique()	2019-05-14 14:56:08 +02:00
Alexey Rybalchenko	9b4c5deb0b	Handle invalid transition to init in the control plugin	2019-05-07 16:28:55 +02:00
Alexey Rybalchenko	7b16c33ccd	Allow PluginServices::SetProperty in Ready state	2019-05-07 15:51:59 +02:00
Alexey Rybalchenko	7e6eb382d5	Extend Readout example	2019-04-29 20:32:38 +02:00
Alexey Rybalchenko	35399ee039	Update state machine picture	2019-04-29 20:32:37 +02:00
Alexey Rybalchenko	2cc1117637	Notify all listeners on state changes	2019-04-29 20:32:37 +02:00
Dennis Klein	3582091b1c	Add experimental static size mode for ofi transport Whenever --ofi-size-hint > 0, the ofi transport does not use the control band. Multipart is not supported.	2019-04-29 20:28:40 +02:00
Dennis Klein	2457094b6c	Require correct asiofi version v0.3.1	2019-03-21 18:03:32 +01:00
Dennis Klein	54b7742d85	Drop obsolete dependency to AZMQ	2019-03-21 18:03:32 +01:00
Dennis Klein	195644f132	Add some debug output	2019-03-21 18:03:32 +01:00
Dennis Klein	f17dade8f8	Fix example after rebase	2019-03-21 18:03:32 +01:00
Dennis Klein	cc8fd73025	Fix recv logic	2019-03-21 18:03:32 +01:00
Dennis Klein	90fdcc26bb	Run multipart example with ofi	2019-03-21 18:03:32 +01:00
Dennis Klein	b45e4da2a9	Implement linger for ofi This reduces test runtime significantly for most transports	2019-03-21 18:03:32 +01:00
Dennis Klein	a1b7efa2f4	Unify implementation of multi part and single part message interfaces	2019-03-21 18:03:32 +01:00
Dennis Klein	6ee7e5fbf0	Improve error handling	2019-03-21 18:03:32 +01:00
Alexey Rybalchenko	99ffb732f4	Use process tools for WaitFor test	2019-03-19 18:09:01 +01:00
Alexey Rybalchenko	6809d60fad	Fix mismatch of docs/API, wrong return value	2019-03-19 18:09:01 +01:00
Alexey Rybalchenko	ef4d6a3310	Process tools: add print helper, support signals	2019-03-19 18:09:01 +01:00
Matthias Richter	696257fd4f	Extending FairMQParts by a constructor taking list of FairMQMessagePtr This introduces a little helper to create a FairMQParts object in place from a variable list of arguments. As a side effect also AddParts is extended to support more than one FairMQMessagePtr.	2019-03-19 12:48:02 +01:00
Alexey Rybalchenko	cdc1ba084c	Fix broken pipe errors in tools::execute	2019-03-15 15:51:50 +01:00
Alexey Rybalchenko	922f7e9a92	Use Asio to launch processes in fair::mq::tools::execute	2019-03-14 18:15:17 +01:00
Dennis Klein	a8f1a4dfdb	Try to reconnect on refused connection	2019-03-13 18:04:49 +01:00
Dennis Klein	fb42b1e2f0	Adapt to new asiofi release	2019-03-13 18:04:49 +01:00
Alexey Rybalchenko	1a00f3edbd	Remove controller input for several tests The affected tests have independent shutdown conditions that race with the controller inputs.	2019-03-12 17:51:33 +01:00
Alexey Rybalchenko	74881d27e3	Remove obsolete state machine code	2019-03-12 12:10:47 +01:00
Alexey Rybalchenko	dd02c01c36	Extend tests of error cases - test raising SIGINT in every state - test going to Error state from every state - add new states (bind/connect) to exception tests	2019-03-12 12:10:47 +01:00
Alexey Rybalchenko	44a9946ea6	Allow creating region with a callback with default transport	2019-03-11 20:20:22 +01:00