DPL: handle spurious messages when in READY state (AliceO2Group#11799)

noferini · Aug 23, 2023 · 44b5559 · 44b5559
1 parent 637b7d4
commit 44b5559
Show file tree

Hide file tree

Showing 2 changed files with 30 additions and 3 deletions.
diff --git a/Framework/Core/src/DataProcessingDevice.cxx b/Framework/Core/src/DataProcessingDevice.cxx
@@ -1481,6 +1481,10 @@ void DataProcessingDevice::doPrepare(ServiceRegistryRef ref)
     if (info.channel == nullptr) {
       continue;
     }
+    // Only poll DPL channels for now.
+    if (info.channelType != ChannelAccountingType::DPL) {
+      continue;
+    }
     auto& socket = info.channel->GetSocket();
     // If we have pending events from a previous iteration,
     // we do receive in any case.

diff --git a/Framework/Core/src/ExternalFairMQDeviceProxy.cxx b/Framework/Core/src/ExternalFairMQDeviceProxy.cxx
@@ -37,6 +37,7 @@
 
 #include <fairmq/Parts.h>
 #include <fairmq/Device.h>
+#include <uv.h>
 #include <cstring>
 #include <cassert>
 #include <memory>
@@ -466,11 +467,12 @@ DataProcessorSpec specifyExternalFairMQDeviceProxy(char const* name,
       for (auto& channel : channels) {
         LOGP(detail, "Injecting channel '{}' into DPL configuration", channel);
         // Converter should pump messages
+        auto& channelPtr = services.get<RawDeviceService>().device()->GetChannel(channel, 0);
         deviceState.inputChannelInfos.push_back(InputChannelInfo{
           .state = InputChannelState::Running,
           .hasPendingEvents = false,
           .readPolled = false,
-          .channel = nullptr,
+          .channel = &channelPtr,
           .id = {ChannelIndex::INVALID},
           .channelType = ChannelAccountingType::RAWFMQ,
         });
@@ -491,12 +493,31 @@ DataProcessorSpec specifyExternalFairMQDeviceProxy(char const* name,
       std::vector<bool> lastNewStatePending(deviceState.inputChannelInfos.size(), false);
 
       // Continue iterating until all channels have seen a new state.
-      while (std::all_of(lastNewStatePending.begin(), lastNewStatePending.end(), [](bool b) { return b; })) {
+      while (std::all_of(lastNewStatePending.begin(), lastNewStatePending.end(), [](bool b) { return b; }) != true) {
         fair::mq::Parts parts;
         for (size_t ci = 0; ci < deviceState.inputChannelInfos.size(); ++ci) {
           auto& info = deviceState.inputChannelInfos[ci];
+          // We only care about rawfmq channels.
+          if (info.channelType != ChannelAccountingType::RAWFMQ) {
+            lastNewStatePending[ci] = true;
+            continue;
+          }
+          // This means we have not set things up yet. I.e. the first iteration from
+          // ready to run has not happened yet.
+          if (info.channel == nullptr) {
+            lastNewStatePending[ci] = true;
+            continue;
+          }
           info.channel->Receive(parts, 10);
-          lastNewStatePending[ci] = device->NewStatePending();
+          // Handle both cases of state changes:
+          //
+          // - The state has been changed from the outside and FairMQ knows about it.
+          // - The state has been changed from the GUI, and deviceState.nextFairMQState knows about it.
+          //
+          // This latter case is probably better handled from DPL itself, after all it's fair to
+          // assume we need to switch state as soon as the GUI notifies us.
+          // For now we keep it here to avoid side effects.
+          lastNewStatePending[ci] = device->NewStatePending() || (deviceState.nextFairMQState.empty() == false);
           if (parts.Size() == 0) {
             continue;
           }
@@ -510,6 +531,8 @@ DataProcessorSpec specifyExternalFairMQDeviceProxy(char const* name,
             info.readPolled = true;
           }
         }
+        // Keep state transitions going also when running with the standalone GUI.
+        uv_run(deviceState.loop, UV_RUN_NOWAIT);
       }
     };