DPL: process region callbacks as early as possible (#6553)

In case we are not running, we can afford processing the callbacks
as soon as possible.

Author: ktf

Framework/Core/src/DataProcessingDevice.cxx

@@ -343,17 +343,36 @@ void on_signal_callback(uv_signal_t* handle, int signum)
   context->stats->totalSigusr1 += 1;
 }
 
+/// Invoke the callbacks for the mPendingRegionInfos
+void handleRegionCallbacks(ServiceRegistry& registry, std::vector<FairMQRegionInfo>& infos)
+{
+  if (infos.empty() == false) {
+    std::vector<FairMQRegionInfo> toBeNotified;
+    toBeNotified.swap(infos); // avoid any MT issue.
+    for (auto const& info : toBeNotified) {
+      registry.get<CallbackService>()(CallbackService::Id::RegionInfoCallback, info);
+    }
+  }
+}
+
 void DataProcessingDevice::InitTask()
 {
   for (auto& channel : fChannels) {
-    channel.second.at(0).Transport()->SubscribeToRegionEvents([&pendingRegionInfos = mPendingRegionInfos, &regionInfoMutex = mRegionInfoMutex](FairMQRegionInfo info) {
+    channel.second.at(0).Transport()->SubscribeToRegionEvents([this,
+                                                               &registry = mServiceRegistry,
+                                                               &pendingRegionInfos = mPendingRegionInfos,
+                                                               &regionInfoMutex = mRegionInfoMutex](FairMQRegionInfo info) {
       std::lock_guard<std::mutex> lock(regionInfoMutex);
       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);
+      // When not running we can handle the callbacks synchronously.
+      if (this->GetCurrentState() != fair::mq::State::Running) {
+        handleRegionCallbacks(registry, pendingRegionInfos);
+      }
     });
   }
 
@@ -498,6 +517,12 @@ void DataProcessingDevice::Reset() { mServiceRegistry.get<CallbackService>()(Cal
 
 bool DataProcessingDevice::ConditionalRun()
 {
+  // Notify on the main thread the new region callbacks, making sure
+  // no callback is issued if there is something still processing.
+  {
+    std::lock_guard<std::mutex> lock(mRegionInfoMutex);
+    handleRegionCallbacks(mServiceRegistry, mPendingRegionInfos);
+  }
   // This will block for the correct delay (or until we get data
   // on a socket). We also do not block on the first iteration
   // so that devices which do not have a timer can still start an
@@ -529,15 +554,13 @@ bool DataProcessingDevice::ConditionalRun()
 
   // Notify on the main thread the new region callbacks, making sure
   // no callback is issued if there is something still processing.
+  // Notice that we still need to perform callbacks also after
+  // the socket epolled, because otherwise we would end up serving
+  // the callback after the first data arrives is the system is too
+  // fast to transition from Init to Run.
   {
     std::lock_guard<std::mutex> lock(mRegionInfoMutex);
-    if (mPendingRegionInfos.empty() == false) {
-      std::vector<FairMQRegionInfo> toBeNotified;
-      toBeNotified.swap(mPendingRegionInfos); // avoid any MT issue.
-      for (auto const& info : toBeNotified) {
-        mServiceRegistry.get<CallbackService>()(CallbackService::Id::RegionInfoCallback, info);
-      }
-    }
+    handleRegionCallbacks(mServiceRegistry, mPendingRegionInfos);
   }
 
   assert(mStreams.size() == mHandles.size());