DPL: initial support for FairMQRegionInfo callback

This should allow users to register a callback and be notified
whenever FairMQ creates a new shared memory region.

Guaranteed to be called synchronously wrt the processing.

Author: ktf

Framework/Core/include/Framework/CallbackService.h

@@ -13,6 +13,8 @@
 #include "CallbackRegistry.h"
 #include <tuple>
 
+class FairMQRegionInfo;
+
 namespace o2
 {
 namespace framework
@@ -35,7 +37,23 @@ class CallbackService
     /// Invoked when we are notified that no further data will arrive.
     /// Notice that one could have more "EndOfData" notifications. Because
     /// we could be signaled by control that the data flow restarted.
-    EndOfStream
+    EndOfStream,
+
+    /// Invoked whenever FairMQ notifies us of a new region
+    ///
+    /// return AlgorithmSpec::InitCallback{[=](InitContext& ic) {
+    ///    auto& callbacks = ic.services().get<CallbackService>();
+    ///    callbacks.set(CallbackService::Id::RegionInfoCallback, [](FairMQRegionInfo const& info) {
+    ///    ... do GPU init ...
+    ///    });
+    ///  }
+    ///  ...
+    ///  return [task](ProcessingContext& pc) {
+    ///    // your processing loop. Guaranteed to be called synchronously
+    ///    // with the callback
+    ///  };
+    /// }};
+    RegionInfoCallback
   };
 
   template <typename T, T... v>
@@ -51,15 +69,17 @@ class CallbackService
   using IdleCallback = std::function<void()>;
   using ClockTickCallback = std::function<void()>;
   using EndOfStreamCallback = std::function<void(EndOfStreamContext&)>;
+  using RegionInfoCallback = std::function<void(FairMQRegionInfo const&)>;
 
-  using Callbacks = CallbackRegistry<Id,                                                    //
-                                     RegistryPair<Id, Id::Start, StartCallback>,            //
-                                     RegistryPair<Id, Id::Stop, StopCallback>,              //
-                                     RegistryPair<Id, Id::Reset, ResetCallback>,            //
-                                     RegistryPair<Id, Id::Idle, IdleCallback>,              //
-                                     RegistryPair<Id, Id::ClockTick, ClockTickCallback>,    //
-                                     RegistryPair<Id, Id::EndOfStream, EndOfStreamCallback> //
-                                     >;                                                     //
+  using Callbacks = CallbackRegistry<Id,                                                          //
+                                     RegistryPair<Id, Id::Start, StartCallback>,                  //
+                                     RegistryPair<Id, Id::Stop, StopCallback>,                    //
+                                     RegistryPair<Id, Id::Reset, ResetCallback>,                  //
+                                     RegistryPair<Id, Id::Idle, IdleCallback>,                    //
+                                     RegistryPair<Id, Id::ClockTick, ClockTickCallback>,          //
+                                     RegistryPair<Id, Id::EndOfStream, EndOfStreamCallback>,      //
+                                     RegistryPair<Id, Id::RegionInfoCallback, RegionInfoCallback> //
+                                     >;                                                           //
 
   // set callback for specified processing step
   template <typename U>

Framework/Core/include/Framework/DataProcessingDevice.h

@@ -45,6 +45,7 @@ class DataProcessingDevice : public FairMQDevice
  public:
   DataProcessingDevice(DeviceSpec const& spec, ServiceRegistry&, DeviceState& state);
   void Init() final;
+  void InitTask() final;
   void PreRun() final;
   void PostRun() final;
   void Reset() final;
@@ -84,6 +85,7 @@ class DataProcessingDevice : public FairMQDevice
   uint64_t mBeginIterationTimestamp = 0;     /// The timestamp of when the current ConditionalRun was started
   DataProcessingStats mStats;                /// Stats about the actual data processing.
   int mCurrentBackoff = 0;                   /// The current exponential backoff value.
+  std::vector<FairMQRegionInfo> mPendingRegionInfos; /// A list of the region infos not yet notified.
 };
 
 } // namespace o2::framework

Framework/Core/src/DataProcessingDevice.cxx

@@ -165,6 +165,20 @@ void DataProcessingDevice::Init()
   }
 }
 
+void DataProcessingDevice::InitTask()
+{
+  for (auto& channel : fChannels) {
+    channel.second.at(0).Transport()->SubscribeToRegionEvents([& pendingRegionInfos = mPendingRegionInfos](FairMQRegionInfo info) {
+      LOG(debug) << ">>> Region info event" << info.event;
+      LOG(debug) << "id: " << info.id;
+      LOG(debug) << "ptr: " << info.ptr;
+      LOG(debug) << "size: " << info.size;
+      LOG(debug) << "flags: " << info.flags;
+      pendingRegionInfos.push_back(info);
+    });
+  }
+}
+
 void DataProcessingDevice::PreRun() { mServiceRegistry.get<CallbackService>()(CallbackService::Id::Start); }
 
 void DataProcessingDevice::PostRun() { mServiceRegistry.get<CallbackService>()(CallbackService::Id::Stop); }
@@ -250,6 +264,13 @@ bool DataProcessingDevice::ConditionalRun()
   auto now = std::chrono::high_resolution_clock::now();
   mBeginIterationTimestamp = (uint64_t)std::chrono::duration<double, std::milli>(now.time_since_epoch()).count();
 
+  if (mPendingRegionInfos.empty() == false) {
+    std::vector<FairMQRegionInfo> toBeNotified;
+    toBeNotified.swap(mPendingRegionInfos); // avoid any MT issue.
+    for (auto& info : toBeNotified) {
+      mServiceRegistry.get<CallbackService>()(CallbackService::Id::RegionInfoCallback, info);
+    }
+  }
   mServiceRegistry.get<CallbackService>()(CallbackService::Id::ClockTick);
   // Whether or not we had something to do.
   bool active = false;