DPL: Make sure that DPL proxy ready fast channels fast-enough in multi-channel setup

davidrohr · ktf · commit 913f6e72aef5 · 2023-01-25T11:06:45.000+01:00
diff --git a/Framework/Core/src/ExternalFairMQDeviceProxy.cxx b/Framework/Core/src/ExternalFairMQDeviceProxy.cxx
@@ -547,23 +547,32 @@ DataProcessorSpec specifyExternalFairMQDeviceProxy(char const* name,
 
       for (size_t ci = 0; ci < channels.size(); ++ci) {
         std::string const& channel = channels[ci];
-        fair::mq::Parts parts;
-        device->Receive(parts, channel, 0, channels.size() == 1 ? -1 : 1);
-        // Populate TimingInfo from the first message
-        if (parts.Size() != 0) {
-          auto const dh = o2::header::get<DataHeader*>(parts.At(0)->GetData());
-          auto& timingInfo = ctx.services().get<TimingInfo>();
-          if (dh != nullptr) {
-            timingInfo.runNumber = dh->runNumber;
-            timingInfo.firstTForbit = dh->firstTForbit;
-            timingInfo.tfCounter = dh->tfCounter;
+        int waitTime = channels.size() == 1 ? -1 : 1;
+        int maxRead = 1000;
+        while (maxRead-- > 0) {
+          fair::mq::Parts parts;
+          device->Receive(parts, channel, 0, waitTime);
+          // Populate TimingInfo from the first message
+          unsigned int nReceived = parts.Size();
+          if (nReceived != 0) {
+            auto const dh = o2::header::get<DataHeader*>(parts.At(0)->GetData());
+            auto& timingInfo = ctx.services().get<TimingInfo>();
+            if (dh != nullptr) {
+              timingInfo.runNumber = dh->runNumber;
+              timingInfo.firstTForbit = dh->firstTForbit;
+              timingInfo.tfCounter = dh->tfCounter;
+            }
+            auto const dph = o2::header::get<DataProcessingHeader*>(parts.At(0)->GetData());
+            if (dph != nullptr) {
+              timingInfo.timeslice = dph->startTime;
+              timingInfo.creation = dph->creation;
+            }
+            dataHandler(timingInfo, parts, 0, ci);
           }
-          auto const dph = o2::header::get<DataProcessingHeader*>(parts.At(0)->GetData());
-          if (dph != nullptr) {
-            timingInfo.timeslice = dph->startTime;
-            timingInfo.creation = dph->creation;
+          if (nReceived == 0 || channels.size() == 1) {
+            break;
           }
-          dataHandler(timingInfo, parts, 0, ci);
+          waitTime = 0;
         }
       }
     };
