DPL: drop incoming non-calibration data when we process calibration only (AliceO2Group#13544)

Author: ktf

Framework/Core/include/Framework/DeviceState.h

@@ -17,8 +17,6 @@
 
 #include <vector>
 #include <string>
-#include <map>
-#include <utility>
 #include <atomic>
 
 typedef struct uv_loop_s uv_loop_t;

Framework/Core/src/DataProcessingDevice.cxx

@@ -10,7 +10,6 @@
 // or submit itself to any jurisdiction.
 #include "Framework/AsyncQueue.h"
 #include "Framework/DataProcessingDevice.h"
-#include "Framework/ChannelMatching.h"
 #include "Framework/ControlService.h"
 #include "Framework/ComputingQuotaEvaluator.h"
 #include "Framework/DataProcessingHeader.h"
@@ -28,7 +27,6 @@
 #include "ConfigurationOptionsRetriever.h"
 #include "Framework/FairMQDeviceProxy.h"
 #include "Framework/CallbackService.h"
-#include "Framework/TMessageSerializer.h"
 #include "Framework/InputRecord.h"
 #include "Framework/InputSpan.h"
 #if defined(__APPLE__) || defined(NDEBUG)
@@ -37,23 +35,20 @@
 #include "Framework/Signpost.h"
 #include "Framework/TimingHelpers.h"
 #include "Framework/SourceInfoHeader.h"
-#include "Framework/Logger.h"
 #include "Framework/DriverClient.h"
-#include "Framework/Monitoring.h"
 #include "Framework/TimesliceIndex.h"
 #include "Framework/VariableContextHelpers.h"
 #include "Framework/DataProcessingContext.h"
+#include "Framework/DataProcessingHeader.h"
 #include "Framework/DeviceContext.h"
 #include "Framework/RawDeviceService.h"
 #include "Framework/StreamContext.h"
 #include "Framework/DefaultsHelpers.h"
+#include "Framework/ServiceRegistryRef.h"
 
-#include "PropertyTreeHelpers.h"
-#include "DataProcessingStatus.h"
 #include "DecongestionService.h"
 #include "Framework/DataProcessingHelpers.h"
 #include "DataRelayerHelpers.h"
-#include "ProcessingPoliciesHelpers.h"
 #include "Headers/DataHeader.h"
 #include "Headers/DataHeaderHelpers.h"
 
@@ -66,6 +61,7 @@
 #include <fairmq/ProgOptions.h>
 #include <Configuration/ConfigurationInterface.h>
 #include <Configuration/ConfigurationFactory.h>
+#include <Monitoring/Monitoring.h>
 #include <TMessage.h>
 #include <TClonesArray.h>
 
@@ -74,7 +70,6 @@
 #include <vector>
 #include <numeric>
 #include <memory>
-#include <unordered_map>
 #include <uv.h>
 #include <execinfo.h>
 #include <sstream>

Framework/Core/src/DataRelayer.cxx

@@ -402,6 +402,12 @@ void DataRelayer::pruneCache(TimesliceSlot slot, OnDropCallback onDrop)
   pruneCache(slot);
 }
 
+bool isCalibrationData(std::unique_ptr<fair::mq::Message>& first)
+{
+  auto* dh = o2::header::get<DataHeader*>(first->GetData());
+  return dh->flagsDerivedHeader & DataProcessingHeader::KEEP_AT_EOS_FLAG;
+}
+
 DataRelayer::RelayChoice
   DataRelayer::relay(void const* rawHeader,
                      std::unique_ptr<fair::mq::Message>* messages,
@@ -456,7 +462,7 @@ DataRelayer::RelayChoice
                      &nPayloads,
                      &cache = mCache,
                      &services = mContext,
-                     numInputTypes = mDistinctRoutesIndex.size()](TimesliceId timeslice, int input, TimesliceSlot slot, InputInfo const& info) {
+                     numInputTypes = mDistinctRoutesIndex.size()](TimesliceId timeslice, int input, TimesliceSlot slot, InputInfo const& info) -> size_t {
     O2_SIGNPOST_ID_GENERATE(aid, data_relayer);
     O2_SIGNPOST_EVENT_EMIT(data_relayer, aid, "saveInSlot", "saving %{public}s@%zu in slot %zu from %{public}s",
                            fmt::format("{:x}", *o2::header::get<DataHeader*>(messages[0]->GetData())).c_str(),
@@ -468,11 +474,20 @@ DataRelayer::RelayChoice
     // TODO: make sure that multiple parts can only be added within the same call of
     // DataRelayer::relay
     assert(nPayloads > 0);
+    size_t saved = 0;
     for (size_t mi = 0; mi < nMessages; ++mi) {
       assert(mi + nPayloads < nMessages);
+      // We are in calibration mode and the data does not have the calibration bit set.
+      // We do not store it.
+      if (services.get<DeviceState>().allowedProcessing == DeviceState::ProcessingType::CalibrationOnly && !isCalibrationData(messages[mi])) {
+        mi += nPayloads;
+        continue;
+      }
       target.add([&messages, &mi](size_t i) -> fair::mq::MessagePtr& { return messages[mi + i]; }, nPayloads + 1);
       mi += nPayloads;
+      saved += nPayloads;
     }
+    return saved;
   };
 
   auto updateStatistics = [ref = mContext](TimesliceIndex::ActionTaken action) {
@@ -551,7 +566,10 @@ DataRelayer::RelayChoice
       this->pruneCache(slot, onDrop);
       mPruneOps.erase(std::remove_if(mPruneOps.begin(), mPruneOps.end(), [slot](const auto& x) { return x.slot == slot; }), mPruneOps.end());
     }
-    saveInSlot(timeslice, input, slot, info);
+    size_t saved = saveInSlot(timeslice, input, slot, info);
+    if (saved == 0) {
+      return RelayChoice{.type = RelayChoice::Type::Dropped, .timeslice = timeslice};
+    }
     index.publishSlot(slot);
     index.markAsDirty(slot, true);
     stats.updateStats({static_cast<short>(ProcessingStatsId::RELAYED_MESSAGES), DataProcessingStats::Op::Add, (int)1});
@@ -633,7 +651,10 @@ DataRelayer::RelayChoice
       // cache still holds the old data, so we prune it.
       this->pruneCache(slot, onDrop);
       mPruneOps.erase(std::remove_if(mPruneOps.begin(), mPruneOps.end(), [slot](const auto& x) { return x.slot == slot; }), mPruneOps.end());
-      saveInSlot(timeslice, input, slot, info);
+      size_t saved = saveInSlot(timeslice, input, slot, info);
+      if (saved == 0) {
+        return RelayChoice{.type = RelayChoice::Type::Dropped, .timeslice = timeslice};
+      }
       index.publishSlot(slot);
       index.markAsDirty(slot, true);
       return RelayChoice{.type = RelayChoice::Type::WillRelay};

Framework/Core/test/test_DataRelayer.cxx

@@ -10,6 +10,7 @@
 // or submit itself to any jurisdiction.
 
 #include <catch_amalgamated.hpp>
+#include "Framework/DeviceState.h"
 #include "Headers/DataHeader.h"
 #include "Headers/Stack.h"
 #include "MemoryResources/MemoryResources.h"
@@ -60,10 +61,12 @@ TEST_CASE("DataRelayer")
     stats.registerMetric(spec);
   }
 
+  DeviceState state;
   ref.registerService(ServiceRegistryHelpers::handleForService<Monitoring>(&monitoring));
   ref.registerService(ServiceRegistryHelpers::handleForService<DataProcessingStats>(&stats));
   ref.registerService(ServiceRegistryHelpers::handleForService<DataProcessingStates>(&states));
   ref.registerService(ServiceRegistryHelpers::handleForService<DriverConfig const>(&driverConfig));
+  ref.registerService(ServiceRegistryHelpers::handleForService<DeviceState>(&state));
   // A simple test where an input is provided
   // and the subsequent InputRecord is immediately requested.
   SECTION("TestNoWait")