adopt namespaced fairmq api consistently

see FairRootGroup/FairMQ#423

Author: dennisklein

CMakeLists.txt

@@ -79,7 +79,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/CruEmulator.cxx

@@ -13,7 +13,7 @@
 
 #include <ConcurrentQueue.h>
 
-#include <options/FairMQProgOptions.h>
+#include <fairmq/ProgOptions.h>
 
 #include <chrono>
 #include <thread>

src/ReadoutEmulator/CruMemoryHandler.cxx

@@ -14,9 +14,9 @@
 
 #include <ConcurrentQueue.h>
 
-#include <fairmq/FairMQUnmanagedRegion.h>
-#include <fairmq/FairMQDevice.h> /* NewUnmanagedRegionFor */
-#include <options/FairMQProgOptions.h>
+#include <fairmq/UnmanagedRegion.h>
+#include <fairmq/Device.h> /* NewUnmanagedRegionFor */
+#include <fairmq/ProgOptions.h>
 
 #include <chrono>
 #include <thread>
@@ -40,7 +40,7 @@ void CruMemoryHandler::teardown()
   }
 }
 
-void CruMemoryHandler::init(FairMQUnmanagedRegion* pDataRegion, std::size_t pSuperPageSize)
+void CruMemoryHandler::init(fair::mq::UnmanagedRegion* pDataRegion, std::size_t pSuperPageSize)
 {
   mSuperpageSize = pSuperPageSize;
   mDataRegion = pDataRegion;

src/ReadoutEmulator/CruMemoryHandler.h

@@ -16,7 +16,7 @@
 
 #include "Headers/DataHeader.h"
 
-#include <fairmq/FairMQUnmanagedRegion.h>
+#include <fairmq/UnmanagedRegion.h>
 
 #include <stack>
 #include <map>
@@ -43,7 +43,7 @@ struct CRUSuperpage {
 };
 
 struct CruDmaPacket {
-  FairMQUnmanagedRegion* mDataSHMRegion = nullptr;
+  fair::mq::UnmanagedRegion* mDataSHMRegion = nullptr;
   char* mDataPtr = nullptr;
   size_t mDataSize = size_t(0);
 };
@@ -62,7 +62,7 @@ class CruMemoryHandler
     teardown();
   }
 
-  void init(FairMQUnmanagedRegion* pDataRegion, std::size_t pSuperPageSize);
+  void init(fair::mq::UnmanagedRegion* pDataRegion, std::size_t pSuperPageSize);
   void teardown();
 
   std::size_t getSuperpageSize() const
@@ -117,7 +117,7 @@ class CruMemoryHandler
   }
 
  private:
-  FairMQUnmanagedRegion* mDataRegion;
+  fair::mq::UnmanagedRegion* mDataRegion;
 
   std::size_t mSuperpageSize;
 

src/ReadoutEmulator/ReadoutDevice.cxx

@@ -13,7 +13,7 @@
 
 #include <ReadoutDataModel.h>
 
-#include <options/FairMQProgOptions.h>
+#include <fairmq/ProgOptions.h>
 
 #include <chrono>
 #include <thread>
@@ -60,7 +60,7 @@ void ReadoutDevice::InitTask()
   mDataRegion = NewUnmanagedRegionFor(
     mOutChannelName, 0,
     mDataRegionSize + mSuperpageSize,
-    [this](const std::vector<FairMQRegionBlock>& pBlkVect) { // callback to be called when message buffers no longer needed by transport
+    [this](const std::vector<fair::mq::RegionBlock>& pBlkVect) { // callback to be called when message buffers no longer needed by transport
       for (const auto &lBlk : pBlkVect) {
           mCruMemoryHandler->put_data_buffer(static_cast<char*>(lBlk.ptr), lBlk.size);
       }

src/ReadoutEmulator/ReadoutDevice.h

@@ -57,8 +57,8 @@ class ReadoutDevice : public DataDistDevice
   void SendingThread();
 
   // data and Descriptor regions
-  // must be here because NewUnmanagedRegionFor() is a method of FairMQDevice...
-  FairMQUnmanagedRegionPtr mDataRegion;
+  // must be here because NewUnmanagedRegionFor() is a method of fair::mq::Device...
+  fair::mq::UnmanagedRegionPtr mDataRegion;
 
   std::string mOutChannelName;
   std::size_t mDataRegionSize;
@@ -75,7 +75,7 @@ class ReadoutDevice : public DataDistDevice
   std::vector<std::unique_ptr<CruLinkEmulator>> mCruLinks;
 
   // messages to send
-  std::vector<FairMQMessagePtr> mDataBlockMsgs;
+  std::vector<fair::mq::MessagePtr> mDataBlockMsgs;
   std::thread mSendingThread;
 
   /// Observables

src/ReadoutEmulator/runReadoutEmulatorDevice.cxx

@@ -10,9 +10,7 @@
 
 #include "ReadoutDevice.h"
 
-#include <options/FairMQProgOptions.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
@@ -40,7 +38,7 @@ void addCustomOptions(bpo::options_description& options)
     "Input throughput per link (bits per second).");
 }
 
-FairMQDevicePtr getDevice(const FairMQProgOptions& /*config*/)
+std::unqiue_ptr<fair::mq::Device> getDevice(FairMQProgOptions& /*config*/)
 {
-  return new o2::DataDistribution::ReadoutDevice();
+  return std::make_unique<o2::DataDistribution::ReadoutDevice>();
 }

src/StfBuilder/StfBuilderDevice.cxx

@@ -23,7 +23,7 @@
 
 #include <Headers/DataHeader.h>
 
-#include <options/FairMQProgOptions.h>
+#include <fairmq/ProgOptions.h>
 
 #include <chrono>
 #include <thread>

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>
@@ -37,7 +37,7 @@ void StfInputInterface::start(bool pBuildStf, std::shared_ptr<ConsulStfBuilder>
   mBuildStf = pBuildStf;
   mDiscoveryConfig = pStfBuilderConfig;
 
-  mBuilderInputQueue = std::make_unique<ConcurrentFifo<std::vector<FairMQMessagePtr>>>();
+  mBuilderInputQueue = std::make_unique<ConcurrentFifo<std::vector<fair::mq::MessagePtr>>>();
   mStfBuilder = std::make_unique<SubTimeFrameReadoutBuilder>(mDevice.MemI());
 
   // sequence thread only needed when building physics STFs
@@ -87,7 +87,7 @@ mRunning = false;
 void StfInputInterface::StfReceiverThread()
 {
   using namespace std::chrono_literals;
-  std::vector<FairMQMessagePtr> lReadoutMsgs;
+  std::vector<fair::mq::MessagePtr> lReadoutMsgs;
   lReadoutMsgs.reserve(4096);
 
   // Reference to the input channel
@@ -191,7 +191,7 @@ void StfInputInterface::StfBuilderThread()
   using namespace std::chrono_literals;
 
   bool lStarted = false;
-  std::vector<FairMQMessagePtr> lReadoutMsgs;
+  std::vector<fair::mq::MessagePtr> lReadoutMsgs;
   lReadoutMsgs.reserve(1U << 20);
 
   // support FEEID masking
@@ -217,7 +217,7 @@ void StfInputInterface::StfBuilderThread()
   // insert and mask the feeid
   auto lInsertWithFeeIdMasking = [&lStfBuilder, lFeeIdMask] (const header::DataOrigin &pDataOrigin,
     const header::DataHeader::SubSpecificationType &pSubSpec, const ReadoutSubTimeframeHeader &pRdoHeader,
-    const FairMQParts::iterator pStartHbf, const std::size_t pInsertCnt) {
+    const fair::mq::Parts::iterator pStartHbf, const std::size_t pInsertCnt) {
 
     // mask the subspecification if the fee mode is used
     auto lMaskedSubspec = pSubSpec;
@@ -402,7 +402,7 @@ void StfInputInterface::TopologicalStfBuilderThread()
   using namespace std::chrono_literals;
 
   bool lStarted = false;
-  std::vector<FairMQMessagePtr> lReadoutMsgs(1U << 20);
+  std::vector<fair::mq::MessagePtr> lReadoutMsgs(1U << 20);
   lReadoutMsgs.clear();
 
   // per equipment data cache
@@ -447,7 +447,7 @@ void StfInputInterface::TopologicalStfBuilderThread()
     const header::DataOrigin &pDataOrigin,
     const header::DataHeader::SubSpecificationType &pSubSpec,
     const ReadoutSubTimeframeHeader &pRdoHeader,
-    FairMQParts::iterator &pStartHbf /* in/out */,
+    fair::mq::Parts::iterator &pStartHbf /* in/out */,
     std::size_t &pInsertCnt /* in/out */) -> std::optional<std::unique_ptr<SubTimeFrame> > {
 
     // mask the subspecification if the fee mode is used

src/StfBuilder/StfBuilderInput.h

@@ -77,7 +77,7 @@ class StfInputInterface
   std::thread mInputThread;
 
   /// StfBuilding thread and queues
-  std::unique_ptr<ConcurrentFifo<std::vector<FairMQMessagePtr>>> mBuilderInputQueue = nullptr;
+  std::unique_ptr<ConcurrentFifo<std::vector<fair::mq::MessagePtr>>> mBuilderInputQueue = nullptr;
   std::unique_ptr<SubTimeFrameReadoutBuilder> mStfBuilder = nullptr;
   std::uint32_t mStfIdBuilding = 0;
   std::thread mBuilderThread;