DPL: have input proxy wait for EoS

ktf · davidrohr · commit d458a4a81bfc · 2022-06-30T13:52:24.000+02:00
A race condition is currently possible where if there is no data
still to be processed on the external and we quit too quickly
missing the EoS.
diff --git a/Framework/Core/src/DataProcessingDevice.cxx b/Framework/Core/src/DataProcessingDevice.cxx
@@ -965,7 +965,12 @@ void DataProcessingDevice::doPrepare(DataProcessorContext& context)
   // to be completed. In the case of data source devices, as they do not have
   // real data input channels, they have to signal EndOfStream themselves.
   context.allDone = std::any_of(context.deviceContext->state->inputChannelInfos.begin(), context.deviceContext->state->inputChannelInfos.end(), [](const auto& info) {
-    return info.parts.fParts.empty() == true && info.state != InputChannelState::Pull;
+    if (info.channel) {
+      LOGP(debug, "Input channel {}{} has {} parts left and is in state {}.", info.channel->GetName(), (info.id.value == ChannelIndex::INVALID ? " (non DPL)" : ""), info.parts.fParts.size(), (int)info.state);
+    } else {
+      LOGP(debug, "External channel {} is in state {}.", info.id.value, (int)info.state);
+    }
+    return (info.parts.fParts.empty() == true && info.state != InputChannelState::Pull);
   });
 
   // Whether or not all the channels are completed
@@ -1016,6 +1021,9 @@ void DataProcessingDevice::doPrepare(DataProcessorContext& context)
       LOGP(debug, "Flushing channel {} which is in state {} and has {} parts still pending.", channelSpec.name, (int)info.state, info.parts.Size());
       continue;
     }
+    if (info.channel == nullptr) {
+      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
@@ -21,6 +21,7 @@
 #include "Framework/CallbackService.h"
 #include "Framework/ControlService.h"
 #include "Framework/SourceInfoHeader.h"
+#include "Framework/ChannelInfo.h"
 #include "Framework/ConfigParamRegistry.h"
 #include "Framework/RateLimiter.h"
 #include "Framework/TimingInfo.h"
@@ -397,17 +398,27 @@ DataProcessorSpec specifyExternalFairMQDeviceProxy(char const* name,
     // will be multiple channels. At least we throw a more informative exception.
     // fair::mq::Device calls the custom init before the channels have been configured
     // so we do the check before starting in a dedicated callback
-    auto channelConfigurationChecker = [channel, device]() {
+    auto channelConfigurationChecker = [channel, device, &services = ctx.services()]() {
+      auto& deviceState = services.get<DeviceState>();
       if (device->fChannels.count(channel) == 0) {
         throw std::runtime_error("the required out-of-band channel '" + channel + "' has not been configured, please check the name in the channel configuration");
       }
+      LOGP(detail, "Injecting channel '{}' into DPL configuration", channel);
+      // Converter should pump messages
+      deviceState.inputChannelInfos.push_back(InputChannelInfo{
+        .state = InputChannelState::Running,
+        .hasPendingEvents = false,
+        .readPolled = false,
+        .channel = nullptr,
+        .id = {ChannelIndex::INVALID},
+      });
     };
     ctx.services().get<CallbackService>().set(CallbackService::Id::Start, channelConfigurationChecker);
-    // Converter should pump messages
 
     auto dataHandler = [device, converter,
                         outputRoutes = std::move(outputRoutes),
                         control = &ctx.services().get<ControlService>(),
+                        deviceState = &ctx.services().get<DeviceState>(),
                         &timingInfo = ctx.services().get<TimingInfo>(),
                         outputChannels = std::move(outputChannels)](fair::mq::Parts& inputs, int) {
       // pass a copy of the outputRoutes
@@ -436,6 +447,10 @@ DataProcessorSpec specifyExternalFairMQDeviceProxy(char const* name,
       converter(timingInfo, *device, inputs, channelRetriever);
 
       if (doEos) {
+        // Mark all input channels as closed
+        for (auto& info : deviceState->inputChannelInfos) {
+          info.state = InputChannelState::Completed;
+        }
         control->endOfStream();
       }
     };
