adopt namespaced fairmq api consistently

see FairRootGroup/FairMQ#423

Author: dennisklein

CMakeLists.txt

@@ -92,7 +92,7 @@ message(STATUS "CMAKE_CXX_STANDARD = ${CMAKE_CXX_STANDARD}")
 list(APPEND CMAKE_MODULE_PATH "${CMAKE_CURRENT_LIST_DIR}/cmake")
 
 find_package(Threads REQUIRED)
-find_package(FairMQ 1.4 REQUIRED)
+find_package(FairMQ 1.4.41 REQUIRED)
 find_package(FairLogger ${FairMQ_FairLogger_VERSION} REQUIRED)
 message(STATUS "Looking for FairLogger dependencies: ${FairLogger_PACKAGE_DEPENDENCIES}" )
 foreach(dep IN LISTS FairLogger_PACKAGE_DEPENDENCIES)

src/ReadoutEmulator/CruMemoryHandler.cxx

@@ -17,8 +17,8 @@
 
 #include <ConcurrentQueue.h>
 
-#include <fairmq/FairMQUnmanagedRegion.h>
-#include <fairmq/FairMQDevice.h> /* NewUnmanagedRegionFor */
+#include <fairmq/UnmanagedRegion.h>
+#include <fairmq/Device.h> /* NewUnmanagedRegionFor */
 #include <fairmq/ProgOptions.h>
 
 #include <chrono>

src/ReadoutEmulator/ReadoutDevice.h

@@ -63,7 +63,7 @@ class ReadoutDevice : public DataDistDevice
   void SendingThread();
 
   // data and Descriptor regions
-  // must be here because NewUnmanagedRegionFor() is a method of FairMQDevice...
+  // must be here because NewUnmanagedRegionFor() is a method of fair::mq::Device...
   std::unique_ptr<DataRegionAllocatorResource> mDataRegion;
 
   std::string mOutChannelName;

src/ReadoutEmulator/runReadoutEmulatorDevice.cxx

@@ -14,8 +14,7 @@
 #include "ReadoutDevice.h"
 
 #include <fairmq/ProgOptions.h>
-
-#include "runFairMQDevice.h"
+#include <fairmq/runDevice.h>
 
 namespace bpo = boost::program_options;
 template class std::basic_string<char, std::char_traits<char>, std::allocator<char> >; // Workaround for bug in CC7 devtoolset7
@@ -46,7 +45,7 @@ void addCustomOptions(bpo::options_description& options)
     "Input throughput per link (bits per second).");
 }
 
-FairMQDevicePtr getDevice(const fair::mq::ProgOptions& /*config*/)
+std::unique_ptr<fair::mq::Device> getDevice(const fair::mq::ProgOptions& /*config*/)
 {
-  return new o2::DataDistribution::ReadoutDevice();
+  return std::make_unique<o2::DataDistribution::ReadoutDevice>();
 }

src/StfBuilder/StfBuilderDevice.cxx

@@ -110,7 +110,7 @@ void StfBuilderDevice::InitTask()
   }
   if (I().mPartitionId.empty()) {
     EDDLOG("Partition id is not provided during InitTask(). Check command line or ECS parameters. Exiting.");
-    ChangeState(fair::mq::Transition::ErrorFound);
+    ChangeStateOrThrow(fair::mq::Transition::ErrorFound);
     return;
   }
 
@@ -143,7 +143,7 @@ void StfBuilderDevice::InitTask()
   // check run type
   if (ReadoutDataUtils::sRunType == ReadoutDataUtils::RunType::eInvalid) {
     EDDLOG("Run type paramter must be correctly set.");
-    ChangeState(fair::mq::Transition::ErrorFound);
+    ChangeStateOrThrow(fair::mq::Transition::ErrorFound);
     return;
   }
 
@@ -153,7 +153,7 @@ void StfBuilderDevice::InitTask()
       (ReadoutDataUtils::sSpecifiedDataOrigin == o2::header::gDataOriginMFT))) {
       EDDLOG("Run type paramter 'topology' is supported only for ITS and MFT. Please specify the detector option. detector={}",
         ReadoutDataUtils::sSpecifiedDataOrigin.as<std::string>());
-      ChangeState(fair::mq::Transition::ErrorFound);
+      ChangeStateOrThrow(fair::mq::Transition::ErrorFound);
       return;
     }
   }
@@ -180,13 +180,13 @@ void StfBuilderDevice::InitTask()
 
   // File sink
   if (!I().mFileSink->loadVerifyConfig(*(this->GetConfig()))) {
-    ChangeState(fair::mq::Transition::ErrorFound);
+    ChangeStateOrThrow(fair::mq::Transition::ErrorFound);
     return;
   }
 
   // File source
   if (!I().mFileSource->loadVerifyConfig(*(this->GetConfig()))) {
-    ChangeState(fair::mq::Transition::ErrorFound);
+    ChangeStateOrThrow(fair::mq::Transition::ErrorFound);
     return;
   }
 
@@ -214,15 +214,15 @@ void StfBuilderDevice::InitTask()
       (ReadoutDataUtils::sSpecifiedDataOrigin == o2::header::gDataOriginAny)) {
       EDDLOG("Detector string parameter must be specified when receiving the data from the "
         "readout and not using RDHv6 or greater.");
-      ChangeState(fair::mq::Transition::ErrorFound);
+      ChangeStateOrThrow(fair::mq::Transition::ErrorFound);
       return;
     } else {
       IDDLOG("READOUT INTERFACE: Configured detector: {}", ReadoutDataUtils::sSpecifiedDataOrigin.as<std::string>());
     }
 
     if (ReadoutDataUtils::sRdhVersion == ReadoutDataUtils::RdhVersion::eRdhInvalid) {
       EDDLOG("The RDH version must be specified when receiving data from readout.");
-      ChangeState(fair::mq::Transition::ErrorFound);
+      ChangeStateOrThrow(fair::mq::Transition::ErrorFound);
       return;
     } else {
       IDDLOG("READOUT INTERFACE: Configured RDHv{}", ReadoutDataUtils::sRdhVersion);
@@ -261,7 +261,7 @@ void StfBuilderDevice::InitTask()
         GetChannel(I().mInputChannelName);
       } catch(std::exception &) {
         EDDLOG("Input channel not configured (from o2-readout-exe) and not running with file source enabled.");
-        ChangeState(fair::mq::Transition::ErrorFound);
+        ChangeStateOrThrow(fair::mq::Transition::ErrorFound);
         return;
       }
     }
@@ -272,7 +272,7 @@ void StfBuilderDevice::InitTask()
       }
     } catch(std::exception &e) {
       EDDLOG("Output channel (to DPL or StfSender) must be configured if not running in stand-alone mode.");
-      ChangeState(fair::mq::Transition::ErrorFound);
+      ChangeStateOrThrow(fair::mq::Transition::ErrorFound);
       return;
     }
   }

src/StfBuilder/StfBuilderInput.cxx

@@ -21,7 +21,7 @@
 #include <DataDistLogger.h>
 #include <DataDistributionOptions.h>
 
-#include <FairMQDevice.h>
+#include <fairmq/Device.h>
 
 #include <vector>
 #include <queue>

src/StfSender/StfSenderDevice.cxx

@@ -92,7 +92,7 @@ void StfSenderDevice::Init()
   I().mPartitionId = Config::getPartitionOption(*GetConfig()).value_or("");
   if (I().mPartitionId.empty()) {
     WDDLOG("StfSender 'discovery-partition' parameter not set during Init(). Exiting.");
-    ChangeState(fair::mq::Transition::ErrorFound);
+    ChangeStateOrThrow(fair::mq::Transition::ErrorFound);
     return;
   }
 
@@ -130,7 +130,7 @@ void StfSenderDevice::Init()
   lBuffersAllocatedFuture.wait();
   if (!lBuffersAllocatedFuture.get()) {
     EDDLOG("Init::MemorySegment allocation failed. Exiting...");
-    ChangeState(fair::mq::Transition::ErrorFound);
+    ChangeStateOrThrow(fair::mq::Transition::ErrorFound);
     return;
   }
 
@@ -185,7 +185,7 @@ void StfSenderDevice::InitTask()
   // Not available in Init()
   if (fair::mq::Transport::SHM != Transport()->GetType()) {
     EDDLOG("Default transport parameter must be set to shm.");
-    ChangeState(fair::mq::Transition::ErrorFound);
+    ChangeStateOrThrow(fair::mq::Transition::ErrorFound);
     return;
   }
 
@@ -195,13 +195,13 @@ void StfSenderDevice::InitTask()
     GetChannel(I().mInputChannelName, 0);
   } catch (...) {
     EDDLOG("Requested input channel is not configured. input_chan={}", I().mInputChannelName);
-    ChangeState(fair::mq::Transition::ErrorFound);
+    ChangeStateOrThrow(fair::mq::Transition::ErrorFound);
     return;
   }
 
   // File sink
   if (!I().mFileSink->loadVerifyConfig(*GetConfig())) {
-    ChangeState(fair::mq::Transition::ErrorFound);
+    ChangeStateOrThrow(fair::mq::Transition::ErrorFound);
     return;
   }
 
@@ -224,7 +224,7 @@ void StfSenderDevice::InitTask()
         if (NewStatePending()) {
           IDDLOG("InitTask: The control system requested abort.");
           AbortInitTask();
-          ChangeState(fair::mq::Transition::ErrorFound);
+          ChangeStateOrThrow(fair::mq::Transition::ErrorFound);
           return;
         }
 
@@ -234,7 +234,7 @@ void StfSenderDevice::InitTask()
       // Did we fail to connect to the TfScheduler?
       if (!I().mTfSchedulerRpcClient.started()) {
         EDDLOG("InitTask: Failed to connect to TfScheduler. Exiting.");
-        ChangeState(fair::mq::Transition::ErrorFound);
+        ChangeStateOrThrow(fair::mq::Transition::ErrorFound);
         return;
       }
     }

src/StfSender/StfSenderOutput.h

@@ -147,8 +147,8 @@ class StfSenderOutput
   ConcurrentFifo<StfCopyInfo> mCopyQueue;
 
   std::mutex mStfOrderingLock;
-    std::condition_variable mStfOrderingCv;
-    std::queue<std::uint64_t> mStfOrderingQueue;
+  std::condition_variable mStfOrderingCv;
+  std::queue<std::uint64_t> mStfOrderingQueue;
 
   /// Scheduler threads
   struct StfSchedInfo {

src/StfSender/StfSenderOutputFairMQ.cxx

@@ -255,4 +255,4 @@ void StfSenderOutputFairMQ::DataHandlerThread(const std::string pTfBuilderId)
 }
 
 
-} /* o2::DataDistribution */
+} /* o2::DataDistribution */

src/TfBuilder/TfBuilderDevice.cxx

@@ -49,7 +49,7 @@ void TfBuilderDevice::Init()
   mPartitionId = Config::getPartitionOption(*GetConfig()).value_or("");
   if (mPartitionId.empty()) {
     WDDLOG("TfBuilder 'discovery-partition' parameter not set during Init(). Exiting.");
-    ChangeState(fair::mq::Transition::ErrorFound);
+    ChangeStateOrThrow(fair::mq::Transition::ErrorFound);
     return;
   }
 
@@ -145,7 +145,7 @@ void TfBuilderDevice::InitTask()
   } catch (...) {
     lBuffersAllocatedFuture.wait();
     EDDLOG("Consul Initialization failed. Exiting...");
-    ChangeState(fair::mq::Transition::ErrorFound);
+    ChangeStateOrThrow(fair::mq::Transition::ErrorFound);
     return;
   }
 
@@ -160,7 +160,7 @@ void TfBuilderDevice::InitTask()
   if (!mFileSink.loadVerifyConfig(*(this->GetConfig()))) {
     lBuffersAllocatedFuture.wait();
     AbortInitTask();
-    ChangeState(fair::mq::Transition::ErrorFound);
+    ChangeStateOrThrow(fair::mq::Transition::ErrorFound);
     return;
   }
 
@@ -179,7 +179,7 @@ void TfBuilderDevice::InitTask()
       EDDLOG("Can not start TfBuilder. id={}", lStatus.info().process_id());
       EDDLOG("Process with the same id already running? If not, clear the key manually.");
       EDDLOG("Discovery database error: this TfBuilder was/is already present. Exiting.");
-      ChangeState(fair::mq::Transition::ErrorFound);
+      ChangeStateOrThrow(fair::mq::Transition::ErrorFound);
       return;
     }
   }
@@ -218,7 +218,7 @@ void TfBuilderDevice::InitTask()
   if (!mFlpInputHandler->start()) {
     AbortInitTask();
     EDDLOG("Could not initialize input connections. Exiting.");
-    ChangeState(fair::mq::Transition::ErrorFound);
+    ChangeStateOrThrow(fair::mq::Transition::ErrorFound);
     return;
   }
 
@@ -227,15 +227,15 @@ void TfBuilderDevice::InitTask()
   if (!lBuffersAllocatedFuture.get()) {
     AbortInitTask();
     EDDLOG("InitTask::MemorySegment allocation failed. Exiting...");
-    ChangeState(fair::mq::Transition::ErrorFound);
+    ChangeStateOrThrow(fair::mq::Transition::ErrorFound);
     return;
   }
 
   // Start input handlers after the memory is finished allocating
   if (!mFlpInputHandler->map_data_region()) {
     AbortInitTask();
     EDDLOG("Could not initialize input connections. Exiting.");
-    ChangeState(fair::mq::Transition::ErrorFound);
+    ChangeStateOrThrow(fair::mq::Transition::ErrorFound);
     return;
   }