find_package2: Fix list sub-command REMOVE_DUPLICATES requires list to be present

Fixes #175
CMake: Support FairLogger v1.5.0
2025-10-15 17:41:45 +00:00 · 2019-06-27 14:02:15 +02:00 · 2019-06-27 14:02:15 +02:00 · 2019-06-27 14:02:15 +02:00 · 2019-06-26 20:42:20 +02:00 · 2019-06-26 20:42:20 +02:00
47 changed files with 970 additions and 555 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)
@@ -64,8 +63,7 @@ if(BUILD_OFI_TRANSPORT)
    VERSION 0.3.1
  )
  find_package2(PRIVATE OFI REQUIRED
-    VERSION ${asiofi_OFI_VERSION}
-    COMPONENTS ${asiofi_OFI_COMPONENTS}
+    ADD_REQUIREMENTS_OF asiofi
  )
 endif()

@@ -75,9 +73,33 @@ if(BUILD_NANOMSG_TRANSPORT)
  )
 endif()

+if(BUILD_DDS_PLUGIN)
+  find_package2(PRIVATE DDS REQUIRED
+    VERSION 2.4
+  )
+  set(DDS_Boost_COMPONENTS system log log_setup)
+  set(DDS_Boost_VERSION 1.67)
+endif()
+
+if(BUILD_PMIX_PLUGIN)
+  find_package2(PRIVATE PMIx REQUIRED
+    VERSION 2.1.4
+  )
+endif()
+
 if(BUILD_FAIRMQ)
+  find_package2(PUBLIC FairLogger REQUIRED
+    VERSION 1.2.0
+  )
+
+  if(NOT DEFINED Boost_NO_BOOST_CMAKE AND CMAKE_VERSION VERSION_LESS 3.15)
+    # Since Boost 1.70 a CMake package is shipped by default. Unfortunately, it has a number
+    # of problems that are only fixed in Boost 1.71 or CMake 3.15. By default we skip the
+    # BoostConfig lookup. This can be overridden on the command line via -DBoost_NO_BOOST_CMAKE=OFF
+    set(Boost_NO_BOOST_CMAKE ON)
+  endif()
  find_package2(PUBLIC Boost REQUIRED
-    VERSION 1.64 ${asiofi_Boost_VERSION}
+    VERSION 1.64

    COMPONENTS
    container
@@ -85,24 +107,17 @@ if(BUILD_FAIRMQ)
    filesystem
    date_time
    regex
-    ${asiofi_Boost_COMPONENTS}
-  )
-  find_package2(PUBLIC FairLogger REQUIRED
-    VERSION 1.2.0
+
+    ADD_REQUIREMENTS_OF
+    asiofi
+    DDS
+    FairLogger
  )
  find_package2(PRIVATE ZeroMQ REQUIRED
    VERSION 4.1.5
  )
 endif()

-if(BUILD_DDS_PLUGIN)
-  find_package2(PRIVATE DDS VERSION 2.2 REQUIRED)
-endif()
-
-if(BUILD_PMIX_PLUGIN)
-  find_package2(PRIVATE PMIx VERSION 2.1.4 REQUIRED)
-endif()
-
 if(BUILD_TESTING)
  find_package2(PRIVATE GTest REQUIRED
    VERSION 1.7.0
@@ -187,11 +202,6 @@ if(BUILD_FAIRMQ)
    DESTINATION ${PROJECT_INSTALL_CMAKEMODDIR}
  )
 endif()
-if(BUILD_DDS_PLUGIN)
-  install(FILES cmake/FindDDS.cmake
-    DESTINATION ${PROJECT_INSTALL_CMAKEMODDIR}
-  )
-endif()
 if(BUILD_DOCS)
  install(DIRECTORY ${CMAKE_BINARY_DIR}/doxygen/html
    DESTINATION ${PROJECT_INSTALL_DATADIR}/docs
@@ -234,7 +244,11 @@ if(PROJECT_PACKAGE_DEPENDENCIES)
  foreach(dep IN LISTS PROJECT_PACKAGE_DEPENDENCIES)
    if(${dep}_VERSION)
      if(${dep} STREQUAL Boost)
-        set(version_str "${BGreen}${${dep}_MAJOR_VERSION}.${${dep}_MINOR_VERSION}${CR}")
+        if(Boost_VERSION_MAJOR)
+          set(version_str "${BGreen}${${dep}_VERSION_MAJOR}.${${dep}_VERSION_MINOR}${CR}")
+        else()
+          set(version_str "${BGreen}${${dep}_MAJOR_VERSION}.${${dep}_MINOR_VERSION}${CR}")
+        endif()
      else()
        set(version_str "${BGreen}${${dep}_VERSION}${CR}")
      endif()
@@ -251,7 +265,11 @@ if(PROJECT_PACKAGE_DEPENDENCIES)
      pad("${version_str}${version_req_str}" 25 " " version_padded COLOR 1)
    endif()
    if(${dep} STREQUAL FairLogger)
-      set(prefix ${FairLogger_ROOT})
+      if(FairLogger_PREFIX)
+        set(prefix ${FairLogger_PREFIX})
+      else()
+        set(prefix ${FairLogger_ROOT})
+      endif()
    elseif(${dep} STREQUAL GTest)
      get_filename_component(prefix ${GTEST_INCLUDE_DIRS}/.. ABSOLUTE)
    elseif(${dep} STREQUAL msgpack)
@@ -264,6 +282,16 @@ if(PROJECT_PACKAGE_DEPENDENCIES)
    elseif(${dep} STREQUAL nanomsg)
      get_target_property(nn_include nanomsg INTERFACE_INCLUDE_DIRECTORIES)
      get_filename_component(prefix ${nn_include}/.. ABSOLUTE)
+    elseif(${dep} STREQUAL DDS)
+      get_target_property(dds_include DDS::dds_intercom_lib INTERFACE_INCLUDE_DIRECTORIES)
+      get_filename_component(prefix ${dds_include}/.. ABSOLUTE)
+    elseif(${dep} STREQUAL Boost)
+      if(TARGET Boost::headers)
+        get_target_property(boost_include Boost::headers INTERFACE_INCLUDE_DIRECTORIES)
+      else()
+        get_target_property(boost_include Boost::boost INTERFACE_INCLUDE_DIRECTORIES)
+      endif()
+      get_filename_component(prefix ${boost_include}/.. ABSOLUTE)
    elseif(${dep} STREQUAL Doxygen)
      get_target_property(doxygen_bin Doxygen::doxygen INTERFACE_LOCATION)
      get_filename_component(prefix ${doxygen_bin} DIRECTORY)
@@ -328,4 +356,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
@@ -126,9 +126,11 @@ After the `find_package(FairMQ)` call the following CMake variables are defined:
 | `${FairMQ_LIBDIR}` | the installation lib directory |
 | `${FairMQ_DATADIR}` | the installation data directory (`../share/fairmq`) |
 | `${FairMQ_CMAKEMODDIR}` | the installation directory of shipped CMake find modules |
-| `${FairMQ_CXX_STANDARD_REQUIRED}` | the value of `CMAKE_CXX_STANDARD_REQUIRED` at built-time |
-| `${FairMQ_CXX_STANDARD}` | the value of `CMAKE_CXX_STANDARD` at built-time |
-| `${FairMQ_CXX_EXTENSIONS}` | the values of `CMAKE_CXX_EXTENSIONS` at built-time |
+| `${FairMQ_CXX_STANDARD_REQUIRED}` | the value of `CMAKE_CXX_STANDARD_REQUIRED` at build-time |
+| `${FairMQ_CXX_STANDARD}` | the value of `CMAKE_CXX_STANDARD` at build-time |
+| `${FairMQ_CXX_EXTENSIONS}` | the values of `CMAKE_CXX_EXTENSIONS` at build-time |
+| `${FairMQ_BUILD_TYPE}` | the value of `CMAKE_BUILD_TYPE` at build-time |
+| `${FairMQ_CXX_FLAGS}` | the values of `CMAKE_CXX_FLAGS` and `CMAKE_CXX_FLAGS_${CMAKE_BUILD_TYPE}` at build-time |

 ## Documentation

@@ -157,3 +159,10 @@ After the `find_package(FairMQ)` call the following CMake variables are defined:
   4. [File output](docs/Logging.md#54-file-output)
   5. [Custom sinks](docs/Logging.md#55-custom-sinks)
 6. [Examples](docs/Examples.md#6-examples)
+7. [Plugins](docs/Plugins.md#7-plugins)
+   1. [Usage](docs/Plugins.md#71-usage)
+   2. [Development](docs/Plugins.md#72-development)
+   3. [Provided Plugins](docs/Plugins.md#73-provided-plugins)
+       1. [DDS](docs/Plugins.md#731-dds)
+       2. [PMIx](docs/Plugins.md#732-pmix)
+
--- a/cmake/FairMQConfig.cmake.in
+++ b/cmake/FairMQConfig.cmake.in
@@ -24,6 +24,9 @@ set(@PROJECT_NAME@_CXX_STANDARD_REQUIRED @CMAKE_CXX_STANDARD_REQUIRED@)
 set(@PROJECT_NAME@_CXX_STANDARD @CMAKE_CXX_STANDARD@)
 set(@PROJECT_NAME@_CXX_EXTENSIONS @CMAKE_CXX_EXTENSIONS@)
 set(@PROJECT_NAME@_VERSION_HOTFIX @PROJECT_VERSION_HOTFIX@)
+set(@PROJECT_NAME@_BUILD_TYPE @CMAKE_BUILD_TYPE@)
+set(@PROJECT_NAME@_BUILD_TYPE_UPPER @PROJECT_BUILD_TYPE_UPPER@)
+set(@PROJECT_NAME@_CXX_FLAGS @PROJECT_CXX_FLAGS@)

 ### Import cmake modules
 set(CMAKE_MODULE_PATH ${@PROJECT_NAME@_CMAKEMODDIR} ${CMAKE_MODULE_PATH})
--- 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)
@@ -280,6 +288,8 @@ macro(install_cmake_package)
  )
  generate_package_dependencies() # fills ${PACKAGE_DEPENDENCIES}
  generate_package_components() # fills ${PACKAGE_COMPONENTS}
+  string(TOUPPER ${CMAKE_BUILD_TYPE} PROJECT_BUILD_TYPE_UPPER)
+  set(PROJECT_CXX_FLAGS ${CMAKE_CXX_FLAGS} ${CMAKE_CXX_FLAGS_${PROJECT_BUILD_TYPE_UPPER}})
  configure_package_config_file(
    ${CMAKE_SOURCE_DIR}/cmake/${PROJECT_NAME}Config.cmake.in
    ${CMAKE_CURRENT_BINARY_DIR}/${PROJECT_NAME}Config.cmake
@@ -293,30 +303,74 @@ macro(install_cmake_package)
  )
 endmacro()

+#
+# find_package2(PRIVATE|PUBLIC|INTERFACE <pkgname>
+#               [VERSION <version>]
+#               [COMPONENTS <list of components>]
+#               [ADD_REQUIREMENTS_OF <list of dep_pgkname>]
+#               [any other option the native find_package supports]...)
+#
+# Wrapper around CMake's native find_package command to add some features and bookkeeping.
+#
+# The qualifier (PRIVATE|PUBLIC|INTERFACE) to the package to populate
+# the variables PROJECT_[INTERFACE]_<pkgname>_([VERSION]|[COMPONENTS]|PACKAGE_DEPENDENCIES)
+# accordingly. This bookkeeping information is used to print our dependency found summary
+# table and to generate a part of our CMake package.
+#
+# When a dependending package is listed with ADD_REQUIREMENTS_OF the variables
+# <dep_pkgname>_<pkgname>_VERSION|COMPONENTS are looked up to and added to the native
+# VERSION (selected highest version) and COMPONENTS (deduplicated) args.
+#
+# COMPONENTS and VERSION args are then just passed to the native find_package.
+#
 macro(find_package2 qualifier pkgname)
-  cmake_parse_arguments(ARGS "" "" "VERSION;COMPONENTS" ${ARGN})
+  cmake_parse_arguments(ARGS "" "VERSION" "COMPONENTS;ADD_REQUIREMENTS_OF" ${ARGN})

  string(TOUPPER ${pkgname} pkgname_upper)
-  set(old_CPP ${CMAKE_PREFIX_PATH})
+  set(__old_cpp__ ${CMAKE_PREFIX_PATH})
  set(CMAKE_PREFIX_PATH ${${pkgname_upper}_ROOT} $ENV{${pkgname_upper}_ROOT} ${CMAKE_PREFIX_PATH})
-  unset(__version__)
+
+  # build lists of required versions and components
+  unset(__required_versions__)
+  unset(__components__)
  if(ARGS_VERSION)
-    list(GET ARGS_VERSION 0 __version__)
-    list(LENGTH ARGS_VERSION __length__)
-    foreach(v IN LISTS ARGS_VERSION)
+    list(APPEND __required_versions__ ${ARGS_VERSION})
+  endif()
+  if(ARGS_COMPONENTS)
+    list(APPEND __components__ ${ARGS_COMPONENTS})
+  endif()
+  if(ARGS_ADD_REQUIREMENTS_OF)
+    foreach(dep_pkgname IN LISTS ARGS_ADD_REQUIREMENTS_OF)
+      if(${dep_pkgname}_${pkgname}_VERSION)
+        list(APPEND __required_versions__ ${${dep_pkgname}_${pkgname}_VERSION})
+      endif()
+      if(${dep_pkgname}_${pkgname}_COMPONENTS)
+        list(APPEND __components__ ${${dep_pkgname}_${pkgname}_COMPONENTS})
+      endif()
+    endforeach()
+  endif()
+
+  # select highest required version
+  unset(__version__)
+  if(__required_versions__)
+    list(GET __required_versions__ 0 __version__)
+    foreach(v IN LISTS __required_versions__)
      if(${v} VERSION_GREATER ${__version__})
        set(__version__ ${v})
      endif()
    endforeach()
  endif()
-  if(ARGS_COMPONENTS)
-    list(REMOVE_DUPLICATES ARGS_COMPONENTS)
-    find_package(${pkgname} ${__version__} QUIET COMPONENTS ${ARGS_COMPONENTS} ${ARGS_UNPARSED_ARGUMENTS})
+  # deduplicate required component list
+  if(__components__)
+    list(REMOVE_DUPLICATES __components__)
+  endif()
+
+  # call native find_package
+  if(__components__)
+    find_package(${pkgname} ${__version__} QUIET COMPONENTS ${__components__} ${ARGS_UNPARSED_ARGUMENTS})
  else()
    find_package(${pkgname} ${__version__} QUIET ${ARGS_UNPARSED_ARGUMENTS})
  endif()
-  set(CMAKE_PREFIX_PATH ${old_CPP})
-  unset(old_CPP)

  if(${pkgname}_FOUND)
    if(${qualifier} STREQUAL PRIVATE)
@@ -338,4 +392,8 @@ macro(find_package2 qualifier pkgname)
  endif()

  unset(__version__)
+  unset(__components__)
+  unset(__required_versions__)
+  set(CMAKE_PREFIX_PATH ${__old_cpp__})
+  unset(__old_cpp__)
 endmacro()
--- a/cmake/FindDDS.cmake
+++ b/cmake/FindDDS.cmake
@@ -1,88 +0,0 @@
-################################################################################
-# Copyright (C) 2014-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"                       #
-################################################################################
-
-find_path(DDS_INCLUDE_DIR
-  NAMES dds_intercom.h
-  HINTS ${DDS_ROOT} $ENV{DDS_ROOT}
-  PATH_SUFFIXES include
-)
-
-find_path(DDS_LIBRARY_DIR
-  NAMES libdds_intercom_lib.dylib libdds_intercom_lib.so
-  HINTS ${DDS_ROOT} $ENV{DDS_ROOT}
-  PATH_SUFFIXES lib
-)
-
-find_library(DDS_INTERCOM_LIBRARY_SHARED
-  NAMES libdds_intercom_lib.dylib libdds_intercom_lib.so
-  HINTS ${DDS_ROOT} $ENV{DDS_ROOT}
-  PATH_SUFFIXES lib
-  DOC "Path to libdds_intercom_lib.dylib libdds_intercom_lib.so."
-)
-
-find_library(DDS_PROTOCOL_LIBRARY_SHARED
-  NAMES libdds_protocol_lib.dylib libdds_protocol_lib.so
-  HINTS ${DDS_ROOT} $ENV{DDS_ROOT}
-  PATH_SUFFIXES lib
-  DOC "Path to libdds_protocol_lib.dylib libdds_protocol_lib.so."
-)
-
-find_library(DDS_USER_DEFAULTS_LIBRARY_SHARED
-  NAMES libdds-user-defaults.dylib libdds-user-defaults.so
-  HINTS ${DDS_ROOT} $ENV{DDS_ROOT}
-  PATH_SUFFIXES lib
-  DOC "Path to libdds-user-defaults.dylib libdds-user-defaults.so."
-)
-
-find_file(DDS_VERSION_FILE
-  NAMES version
-  HINTS ${DDS_ROOT} $ENV{DDS_ROOT}
-  PATH_SUFFIXES etc
-)
-
-if(DDS_VERSION_FILE AND NOT DDS_VERSION)
-  file(READ ${DDS_VERSION_FILE} DDS_VERSION)
-  string(STRIP "${DDS_VERSION}" DDS_VERSION)
-  set(DDS_VERSION ${DDS_VERSION} CACHE string "DDS version.")
-endif()
-
-include(FindPackageHandleStandardArgs)
-find_package_handle_standard_args(DDS
-  REQUIRED_VARS
-  DDS_INCLUDE_DIR
-  DDS_LIBRARY_DIR
-  DDS_INTERCOM_LIBRARY_SHARED
-  DDS_PROTOCOL_LIBRARY_SHARED
-  DDS_USER_DEFAULTS_LIBRARY_SHARED
-
-  VERSION_VAR DDS_VERSION
-)
-
-if(NOT TARGET DDS::dds_intercom_lib AND DDS_FOUND)
-  add_library(DDS::dds_intercom_lib SHARED IMPORTED)
-  set_target_properties(DDS::dds_intercom_lib PROPERTIES
-    IMPORTED_LOCATION ${DDS_INTERCOM_LIBRARY_SHARED}
-    INTERFACE_INCLUDE_DIRECTORIES ${DDS_INCLUDE_DIR}
-  )
-endif()
-
-if(NOT TARGET DDS::dds_protocol_lib AND DDS_FOUND)
-  add_library(DDS::dds_protocol_lib SHARED IMPORTED)
-  set_target_properties(DDS::dds_protocol_lib PROPERTIES
-    IMPORTED_LOCATION ${DDS_PROTOCOL_LIBRARY_SHARED}
-    INTERFACE_INCLUDE_DIRECTORIES ${DDS_INCLUDE_DIR}
-  )
-endif()
-
-if(NOT TARGET DDS::dds-user-defaults AND DDS_FOUND)
-  add_library(DDS::dds-user-defaults SHARED IMPORTED)
-  set_target_properties(DDS::dds-user-defaults PROPERTIES
-    IMPORTED_LOCATION ${DDS_USER_DEFAULTS_LIBRARY_SHARED}
-    INTERFACE_INCLUDE_DIRECTORIES ${DDS_INCLUDE_DIR}
-  )
-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/Plugins.md
+++ b/docs/Plugins.md
@@ -0,0 +1,93 @@
+← [Back](../README.md)
+
+# 7. Plugins
+
+FairMQ devices can be integrated with external configuration and control systems through its plugin system. FairMQ plugins are special dynamic libraries that can be loaded at runtime. Plugins have access to the Plugin API which includes the capability to control/monitor the device [state machine](Device.md#13-state-machine) and change/monitor configuration properties.
+
+A simple plugin may add the feature to read configuration from a certain desired file format once at the start of a device. A more complex plugin may create a long-running thread that integrates a network client to an external API of a central experiment control system.
+
+Because plugins are loaded dynamically, they can be developed in separate repositories/projects and also have their own set of runtime dependency that are not needed to be known at compile-time of the FairMQ device.
+
+## 7.1 Usage
+
+To load a plugin pass the `-P <name>[,<name>]` (or long `--plugin`) command line option. Multiple plugins can be loaded at the same time. The load order is as specified at the command line. This determines the order in which the plugins are instantiated (ctor call order) and in which order they are notified, should they subscribe to any notifications.
+
+When passing `-h/--help` on the command line one can find more detailed information:
+
+```
+Plugin Manager:
+  -S [ --plugin-search-path ] arg       List of plugin search paths.
+                                        * Override default search path, e.g.
+                                          -S /home/user/lib /lib
+                                        * Append(>) or prepend(<) to default
+                                        search path, e.g.
+                                          -S >/lib </home/user/lib
+                                        * If you mix the overriding and
+                                        appending/prepending syntaxes, the
+                                        overriding paths act as default search
+                                        path, e.g.
+                                          -S /usr/lib >/lib </home/user/lib
+                                        /usr/local/lib results in
+                                        /home/user/lib,/usr/local/lib,/usr/lib/
+                                        ,/lib
+                                        If nothing is found, the default
+                                        dynamic library lookup is performed,
+                                        see man ld.so(8) for details.
+  -P [ --plugin ] arg                   List of plugin names to load in
+                                        order,e.g. if the file is called
+                                        'libFairMQPlugin_example.so', just list
+                                        'example' or 'd:example' here.To load a
+                                        prelinked plugin, list 'p:example'
+                                        here.
+```
+
+## 7.2 Development
+
+To develop a custom FairMQ plugin, one simply needs to inherit from the `fair::mq::Plugin` base class (`#include <fairmq/Plugin.h>`) and call the `REGISTER_FAIRMQ_PLUGIN` macro. It is possible to introduce new command line option together with a plugin.
+
+The Plugin API includes:
+  * `Take/Steal/ReleaseDeviceControl()`/`GetCurrent/ChangeDeviceState()`/`SubscribeTo/UnsubscribeFromDeviceStateChange()` APIs enable controlling the device state machine. Only one plugin is authorized to control at the same time. Which one is determined by which plugin calls `TakeDeviceControl()` first.
+  * `Set/GetProperty()`/`GetPropertyKeys()`/`SubscribeTo/UnsubscribeFromPropertyChange()` APIs enable configuration of device properties.
+See [`<fairmq/Plugin.h>`](/fairmq/Plugin.h) for the full API.
+
+A more complete example which may serve as a start including example CMake code can be found here: [FairRootGroup/FairMQPlugin_example](https://github.com/FairRootGroup/FairMQPlugin_example).
+
+## 7.3 Provided Plugins
+
+### 7.3.1 DDS
+
+When launching a FairMQ topology via [DDS](http://dds.gsi.de/) the DDS plugin enables FairMQ devices to interact with DDS' custom command and property subsystems - enable the plugin by passing `-P dds` on the command line.
+
+Via the property subsystem a FairMQ topology may exchange channel connection data (essentially to do service discovery) needed to connect/bind all FairMQ channels appropriately. DDS is highly optimized for this use case. See [examples/dds](examples/dds/README.md) for more details.
+
+Via the custom command subsystem a FairMQ device can receive a number of commands. FairMQ provides a convenient command line tool `fairmq-dds-command-ui` that allows interactive or scripted control of a running FairMQ topology managed via DDS. If one develops directly against the custom command DDS API, the following table lists all the commands the DDS plugin currently understands:
+
+| Custom Command | Response | Error | Description |
+| --- | --- | --- | --- |
+| `check-state` | `<ID>: <STATE> (pid: <PID>)` | n/a | Query current device state, see state machine for possible states |
+| `dump-config` | `(<ID>: <PKEY> -> <PVALUE>\n)+` | n/a | Query current device config (list property key/value pairs) |
+| `INIT DEVICE` | `<ID>: queued <CMD> transition` | `<ID>: could not queue <CMD> transition` | Initiate state transition |
+| `BIND` | `<ID>: queued <CMD> transition` | `<ID>: could not queue <CMD> transition` | Initiate state transition |
+| `CONNECT` | `<ID>: queued <CMD> transition` | `<ID>: could not queue <CMD> transition` | Initiate state transition |
+| `INIT TASK` | `<ID>: queued <CMD> transition` | `<ID>: could not queue <CMD> transition` | Initiate state transition |
+| `RUN` | `<ID>: queued <CMD> transition` | `<ID>: could not queue <CMD> transition` | Initiate state transition |
+| `STOP` | `<ID>: queued <CMD> transition` | `<ID>: could not queue <CMD> transition` | Initiate state transition |
+| `RESET TASK` | `<ID>: queued <CMD> transition` | `<ID>: could not queue <CMD> transition` | Initiate state transition |
+| `RESET DEVICE` | `<ID>: queued <CMD> transition` | `<ID>: could not queue <CMD> transition` | Initiate state transition |
+| `END` | `<ID>: queued <CMD> transition` | `<ID>: could not queue <CMD> transition` | Initiate state transition |
+| `subscribe-to-heartbeats` | `heartbeat-subscription: <ID>,OK` | n/a | Subscribe to heartbeats |
+| on heartbeat subscription | `heartbeat: <ID>,<PID>` | n/a | Heartbeat every 100ms |
+| `unsubscribe-from-heartbeats` | `heartbeat-unsubscription: <ID>,OK` | n/a | Unsubscribe from heartbeats |
+| `subscribe-to-state-changes` | `state-changes-subscription: <ID>,OK` | n/a | Subscribe to state changes |
+| on state changes subscription | `state-change: <ID>,<STATE>` | n/a | State change notification |
+| `unsubscribe-from-state-changes` | `state-changes-unsubscription: <ID>,OK` | n/a | Unsubscribe from state changes |
+
+If unknown commands are received the plugin will print a warning.
+
+### 7.3.2 PMIx
+
+The [PMIx](https://pmix.org/) plugin enables launching a FairMQ topology with any PMIx capable launcher, e.g. the [Open Run-Time Environment (ORTE) of OpenMPI](https://www.open-mpi.org/doc/v4.0/man1/mpirun.1.php) or the [Slurm workload manager](https://slurm.schedmd.com/srun.html). This plugin is not (yet) very mature and serves as a proof of concept at the moment.
+
+TODO example usage
+
+← [Back](../README.md)
--- 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/dds/README.md
+++ b/examples/dds/README.md
@@ -40,9 +40,9 @@ The configuration of the channel connection addresses is done by the DDS plugin

 **If you chose step 2b earlier**, then modify the provided `ex-dds-topology.xml` in the top that the following lines read as following:
 ```xml
-<declrequirement id="SamplerWorker" type="wnname" value=".*"/>
-<declrequirement id="ProcessorWorker" type="wnname" value=".*"/>
-<declrequirement id="SinkWorker" type="wnname" value=".*"/>
+<declrequirement name="SamplerWorker" type="wnname" value=".*"/>
+<declrequirement name="ProcessorWorker" type="wnname" value=".*"/>
+<declrequirement name="SinkWorker" type="wnname" value=".*"/>
 ```

 Note that the attributes `value` contain a different value.
--- a/examples/dds/ex-dds-topology.xml
+++ b/examples/dds/ex-dds-topology.xml
@@ -1,47 +1,47 @@
-<topology id="ExampleDDS">
+<topology name="ExampleDDS">

-    <property id="data1" />
-    <property id="data2" />
+    <property name="data1" />
+    <property name="data2" />

-    <declrequirement id="SamplerWorker" type="wnname" value="sampler"/>
-    <declrequirement id="ProcessorWorker" type="wnname" value="processor"/>
-    <declrequirement id="SinkWorker" type="wnname" value="sink"/>
+    <declrequirement name="SamplerWorker" type="wnname" value="sampler"/>
+    <declrequirement name="ProcessorWorker" type="wnname" value="processor"/>
+    <declrequirement name="SinkWorker" type="wnname" value="sink"/>

-    <decltask id="Sampler">
+    <decltask name="Sampler">
        <exe reachable="true">@EX_BIN_DIR@/fairmq-ex-dds-sampler --id sampler --color false --channel-config name=data1,type=push,method=bind -S "&lt;@DDS_PLUGIN_LIB_DIR@/" -P dds</exe>
        <requirements>
-            <id>SamplerWorker</id>
+            <name>SamplerWorker</name>
        </requirements>
        <properties>
-            <id access="write">data1</id>
+            <name access="write">data1</name>
        </properties>
    </decltask>

-    <decltask id="Processor">
+    <decltask name="Processor">
        <exe reachable="true">@EX_BIN_DIR@/fairmq-ex-dds-processor --id processor_%taskIndex% --config-key processor --color false --channel-config name=data1,type=pull,method=connect name=data2,type=push,method=connect -S "&lt;@DDS_PLUGIN_LIB_DIR@/" -P dds</exe>
        <requirements>
-            <id>ProcessorWorker</id>
+            <name>ProcessorWorker</name>
        </requirements>
        <properties>
-            <id access="read">data1</id>
-            <id access="read">data2</id>
+            <name access="read">data1</name>
+            <name access="read">data2</name>
        </properties>
    </decltask>

-    <decltask id="Sink">
+    <decltask name="Sink">
        <exe reachable="true">@EX_BIN_DIR@/fairmq-ex-dds-sink --id sink --color false --channel-config name=data2,type=pull,method=bind -S "&lt;@DDS_PLUGIN_LIB_DIR@/" -P dds</exe>
        <requirements>
-            <id>SinkWorker</id>
+            <name>SinkWorker</name>
        </requirements>
        <properties>
-            <id access="write">data2</id>
+            <name access="write">data2</name>
        </properties>
    </decltask>

-    <main id="main">
+    <main name="main">
        <task>Sampler</task>
        <task>Sink</task>
-        <group id="ProcessorGroup" n="10">
+        <group name="ProcessorGroup" n="10">
            <task>Processor</task>
        </group>
    </main>
--- 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/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
@@ -366,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;
    }

--- 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/Version.h.in
+++ b/fairmq/Version.h.in
@@ -21,7 +21,7 @@
 #define FAIRMQ_GIT_DATE "@PROJECT_GIT_DATE@"
 #define FAIRMQ_REPO_URL "https://github.com/FairRootGroup/FairMQ"
 #define FAIRMQ_LICENSE "LGPL-3.0"
-#define FAIRMQ_COPYRIGHT "2012-2018 GSI"
+#define FAIRMQ_COPYRIGHT "2012-2019 GSI"
 #define FAIRMQ_BUILD_TYPE "@CMAKE_BUILD_TYPE@"

 #endif // FAIR_MQ_VERSION_H
--- 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);
 }
--- 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/Socket.cxx
+++ b/fairmq/ofi/Socket.cxx
@@ -154,8 +154,17 @@ auto Socket::BindDataEndpoint() -> void
        fDataEndpoint->accept([&]() {
            LOG(debug) << "OFI transport (" << fId << "): data band connection accepted.";

-            boost::asio::post(fContext.GetIoContext(), std::bind(&Socket::SendQueueReader, this));
-            boost::asio::post(fContext.GetIoContext(), std::bind(&Socket::RecvControlQueueReader, this));
+            if (fContext.GetSizeHint()) {
+                boost::asio::post(fContext.GetIoContext(),
+                                  std::bind(&Socket::SendQueueReaderStatic, this));
+                boost::asio::post(fContext.GetIoContext(),
+                                  std::bind(&Socket::RecvQueueReaderStatic, this));
+            } else {
+                boost::asio::post(fContext.GetIoContext(),
+                                  std::bind(&Socket::SendQueueReader, this));
+                boost::asio::post(fContext.GetIoContext(),
+                                  std::bind(&Socket::RecvControlQueueReader, this));
+            }
        });
    });

@@ -174,8 +183,13 @@ try {
    ConnectEndpoint(fControlEndpoint, Band::Control);
    ConnectEndpoint(fDataEndpoint, Band::Data);

-    boost::asio::post(fContext.GetIoContext(), std::bind(&Socket::SendQueueReader, this));
-    boost::asio::post(fContext.GetIoContext(), std::bind(&Socket::RecvControlQueueReader, this));
+    if (fContext.GetSizeHint()) {
+        boost::asio::post(fContext.GetIoContext(), std::bind(&Socket::SendQueueReaderStatic, this));
+        boost::asio::post(fContext.GetIoContext(), std::bind(&Socket::RecvQueueReaderStatic, this));
+    } else {
+        boost::asio::post(fContext.GetIoContext(), std::bind(&Socket::SendQueueReader, this));
+        boost::asio::post(fContext.GetIoContext(), std::bind(&Socket::RecvControlQueueReader, this));
+    }

    return true;
 }
@@ -347,6 +361,57 @@ auto Socket::SendQueueReader() -> void
    });
 }

+auto Socket::SendQueueReaderStatic() -> void
+{
+    fSendPopSem.async_wait([&] {
+        // Read msg from send queue
+        std::unique_lock<std::mutex> lk(fSendQueueMutex);
+        std::vector<MessagePtr> msgVec(std::move(fSendQueue.front()));
+        fSendQueue.pop();
+        lk.unlock();
+
+        bool postMultiPartStartBuffer = msgVec.size() > 1;
+        if (postMultiPartStartBuffer) {
+            throw SocketError{tools::ToString("Multipart API not supported in static size mode.")};
+        }
+
+        MessagePtr& msg = msgVec[0];
+
+        // Send data message
+        const auto size = msg->GetSize();
+
+        if (size) {
+            boost::asio::mutable_buffer buffer(msg->GetData(), size);
+
+            if (fNeedOfiMemoryRegistration) {
+                asiofi::memory_region mr(*fOfiDomain, buffer, asiofi::mr::access::send);
+                auto desc = mr.desc();
+
+                fDataEndpoint->send(buffer,
+                                    desc,
+                                    [&, size, msg2 = std::move(msg), mr2 = std::move(mr)](
+                                        boost::asio::mutable_buffer) mutable {
+                                        fBytesTx += size;
+                                        fMessagesTx++;
+                                        fSendPushSem.signal();
+                                    });
+            } else {
+                fDataEndpoint->send(
+                    buffer, [&, size, msg2 = std::move(msg)](boost::asio::mutable_buffer) mutable {
+                        fBytesTx += size;
+                        fMessagesTx++;
+                        fSendPushSem.signal();
+                    });
+            }
+        } else {
+            ++fMessagesTx;
+            fSendPushSem.signal();
+        }
+
+        boost::asio::dispatch(fContext.GetIoContext(), std::bind(&Socket::SendQueueReaderStatic, this));
+    });
+}
+
 auto Socket::Receive(MessagePtr& msg, const int /*timeout*/) -> int
 try {
    // timeout argument not yet implemented
@@ -472,6 +537,42 @@ auto Socket::OnRecvControl(ofi::unique_ptr<ControlMessage> ctrl) -> void
                          std::bind(&Socket::RecvControlQueueReader, this));
 }

+auto Socket::RecvQueueReaderStatic() -> void
+{
+    fRecvPushSem.async_wait([&] {
+        static size_t size = fContext.GetSizeHint();
+        // Receive data
+        auto msg = fContext.MakeReceiveMessage(size);
+
+        if (size) {
+            boost::asio::mutable_buffer buffer(msg->GetData(), size);
+
+            if (fNeedOfiMemoryRegistration) {
+                asiofi::memory_region mr(*fOfiDomain, buffer, asiofi::mr::access::recv);
+                auto desc = mr.desc();
+
+                fDataEndpoint->recv(buffer,
+                                    desc,
+                                    [&, msg2 = std::move(msg), mr2 = std::move(mr)](
+                                        boost::asio::mutable_buffer) mutable {
+                                        DataMessageReceived(std::move(msg2));
+                                    });
+
+            } else {
+                fDataEndpoint->recv(
+                    buffer, [&, msg2 = std::move(msg)](boost::asio::mutable_buffer) mutable {
+                        DataMessageReceived(std::move(msg2));
+                    });
+            }
+        } else {
+            DataMessageReceived(std::move(msg));
+        }
+
+        boost::asio::dispatch(fContext.GetIoContext(),
+                              std::bind(&Socket::RecvQueueReaderStatic, this));
+    });
+}
+
 auto Socket::DataMessageReceived(MessagePtr msg) -> void
 {
    if (fMultiPartRecvCounter > 0) {
--- a/fairmq/ofi/Socket.h
+++ b/fairmq/ofi/Socket.h
@@ -110,7 +110,9 @@ class Socket final : public fair::mq::Socket
    enum class Band { Control, Data };
    auto ConnectEndpoint(std::unique_ptr<asiofi::connected_endpoint>& endpoint, Band type) -> 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 */
--- 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
@@ -203,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;
@@ -324,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"
@@ -362,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"
--- 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/plugins/DDS/DDS.h
+++ b/fairmq/plugins/DDS/DDS.h
@@ -11,7 +11,7 @@

 #include <fairmq/Plugin.h>

-#include <dds_intercom.h>
+#include <DDS/dds_intercom.h>

 #include <condition_variable>
 #include <mutex>
--- a/fairmq/plugins/DDS/runDDSCommandUI.cxx
+++ b/fairmq/plugins/DDS/runDDSCommandUI.cxx
@@ -6,7 +6,7 @@
 *                  copied verbatim in the file "LICENSE"                       *
 ********************************************************************************/

-#include <dds_intercom.h>
+#include <DDS/dds_intercom.h>

 #include <termios.h> // raw mode console input

--- 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
Author	SHA1	Message	Date
Dennis Klein	b1c8264123	find_package2: Fix list sub-command REMOVE_DUPLICATES requires list to be present Fixes #175	2019-06-27 14:02:15 +02:00
Dennis Klein	4d53b7c024	CMake: Support FairLogger v1.5.0	2019-06-27 14:02:15 +02:00
Dennis Klein	09853e3031	CMake: Support Boost 1.70 config mode Below CMake 3.15 we disable config mode by default, this can be overridden via -DBoost_NO_BOOST_CMAKE=OFF Resolves #174	2019-06-27 14:02:15 +02:00
Dennis Klein	a866c6d936	CMake: Add Boost version requirement of DDS * Fix dependency summary table entry for DDS	2019-06-26 20:42:20 +02:00
Dennis Klein	a30a6955ef	CMake: Apply new ADD_REQUIREMENTS_OF option	2019-06-26 20:42:20 +02:00
Dennis Klein	187a821f36	CMake: Revert VERSION option to single value arg This fixes a regression introduced in `0ff8eaf` that caused other options to be ineffective in certain cases. Resolves #164	2019-06-26 20:42:20 +02:00
Dennis Klein	465d90924b	CMake: Implement ADD_REQUIREMENTS_OF option for find_package2 Resolves #142	2019-06-26 20:42:20 +02:00
Dennis Klein	3358a2ba12	Require DDS 2.4 * Remove obsolete FindDDS.cmake * Adapt to new DDS version	2019-06-26 18:39:12 +02:00
Dennis Klein	479c16a8fa	CMake: Export build type and cxx flags Resolve #158	2019-06-24 19:06:38 +02:00
Dennis Klein	374eb84039	Docs: Add section about plugins Resolve #3	2019-06-24 18:18:29 +02:00
Dennis Klein	56dc91ab87	Update year	2019-06-13 18:04:44 +02:00
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