Merge remote-tracking branch 'origin/dev' into stable-sync

Author: davidrohr

CCDB/src/CcdbApi.cxx

@@ -1688,7 +1688,7 @@ void CcdbApi::vectoredLoadFileToMemory(std::vector<RequestContext>& requestConte
         saveSnapshot(requestContext);
       }
     } else {
-      LOG(error) << "Did not receive content for " << requestContext.path << "\n";
+      LOG(warning) << "Did not receive content for " << requestContext.path << "\n"; // Temporarily demoted to warning, since it floods the infologger
     }
   }
 }

Detectors/GlobalTracking/src/MatchTPCITS.cxx

@@ -633,9 +633,9 @@ bool MatchTPCITS::prepareITSData()
   for (int irof = 0; irof < nROFs; irof++) {
     const auto& rofRec = mITSTrackROFRec[irof];
     long nBC = rofRec.getBCData().differenceInBC(mStartIR);
-    if (nBC > maxBCs) {
+    if (nBC > maxBCs || nBC < 0) {
       if (++errCount < MaxErrors2Report) {
-        LOGP(alarm, "ITS ROF#{} start is not compatible with TF 1st orbit {} and TF length of {} HBFs",
+        LOGP(alarm, "ITS ROF#{} start {} is not compatible with TF 1st orbit {} or TF length of {} HBFs",
              irof, rofRec.getBCData().asString(), mStartIR.asString(), nHBF);
       }
       break;

Detectors/ITSMFT/common/reconstruction/src/Clusterer.cxx

@@ -36,6 +36,7 @@ void Clusterer::process(int nThreads, PixelReader& reader, CompClusCont* compClu
   }
   auto autoDecode = reader.getDecodeNextAuto();
   int rofcount{0};
+  o2::InteractionRecord lastIR{};
   do {
     if (autoDecode) {
       reader.setDecodeNextAuto(false); // internally do not autodecode
@@ -46,6 +47,15 @@ void Clusterer::process(int nThreads, PixelReader& reader, CompClusCont* compClu
     if (reader.getInteractionRecord().isDummy()) {
       continue; // No IR info was found
     }
+    if (!lastIR.isDummy() && lastIR >= reader.getInteractionRecord()) {
+      const int MaxErrLog = 2;
+      static int errLocCount = 0;
+      if (errLocCount++ < MaxErrLog) {
+        LOGP(warn, "Impossible ROF IR {}, does not exceed previous {}, discarding in clusterization", reader.getInteractionRecord().asString(), lastIR.asString());
+      }
+      continue;
+    }
+    lastIR = reader.getInteractionRecord();
     // pre-fetch all non-empty chips of current ROF
     ChipPixelData* curChipData = nullptr;
     mFiredChipsPtr.clear();

Detectors/ITSMFT/common/workflow/src/STFDecoderSpec.cxx

@@ -155,7 +155,17 @@ void STFDecoder<Mapping>::run(ProcessingContext& pc)
     }
 
     mDecoder->setDecodeNextAuto(false);
+    o2::InteractionRecord lastIR{}, firstIR{0, pc.services().get<o2::framework::TimingInfo>().firstTForbit};
     while (mDecoder->decodeNextTrigger() >= 0) {
+      if ((!lastIR.isDummy() && lastIR >= mDecoder->getInteractionRecord()) || firstIR > mDecoder->getInteractionRecord()) {
+        const int MaxErrLog = 2;
+        static int errLocCount = 0;
+        if (errLocCount++ < MaxErrLog) {
+          LOGP(warn, "Impossible ROF IR {}, previous was {}, TF 1st IR was {}, discarding in decoding", mDecoder->getInteractionRecord().asString(), lastIR.asString(), firstIR.asString());
+        }
+        continue;
+      }
+      lastIR = mDecoder->getInteractionRecord();
       if (mDoDigits || mClusterer->getMaxROFDepthToSquash()) { // call before clusterization, since the latter will hide the digits
         mDecoder->fillDecodedDigits(digVec, digROFVec);        // lot of copying involved
         if (mDoCalibData) {

Framework/Core/include/Framework/DataProcessingDevice.h

@@ -107,19 +107,18 @@ class DataProcessingDevice : public fair::mq::Device
   std::unique_ptr<ConfigParamRegistry> mConfigRegistry;
   ServiceRegistry& mServiceRegistry;
 
-  uint64_t mLastSlowMetricSentTimestamp = 0;         /// The timestamp of the last time we sent slow metrics
-  uint64_t mLastMetricFlushedTimestamp = 0;          /// The timestamp of the last time we actually flushed metrics
-  uint64_t mBeginIterationTimestamp = 0;             /// The timestamp of when the current ConditionalRun was started
+  uint64_t mLastSlowMetricSentTimestamp = 0;             /// The timestamp of the last time we sent slow metrics
+  uint64_t mLastMetricFlushedTimestamp = 0;              /// The timestamp of the last time we actually flushed metrics
+  uint64_t mBeginIterationTimestamp = 0;                 /// The timestamp of when the current ConditionalRun was started
   std::vector<fair::mq::RegionInfo> mPendingRegionInfos; /// A list of the region infos not yet notified.
   std::mutex mRegionInfoMutex;
-  ProcessingPolicies mProcessingPolicies;                        /// User policies related to data processing
-  bool mWasActive = false;                                       /// Whether or not the device was active at last iteration.
-  std::vector<uv_work_t> mHandles;                               /// Handles to use to schedule work.
-  std::vector<TaskStreamInfo> mStreams;                          /// Information about the task running in the associated mHandle.
+  ProcessingPolicies mProcessingPolicies; /// User policies related to data processing
+  bool mWasActive = false;                /// Whether or not the device was active at last iteration.
+  std::vector<uv_work_t> mHandles;        /// Handles to use to schedule work.
+  std::vector<TaskStreamInfo> mStreams;   /// Information about the task running in the associated mHandle.
   /// Handle to wake up the main loop from other threads
   /// e.g. when FairMQ notifies some callback in an asynchronous way
   uv_async_t* mAwakeHandle = nullptr;
-  int64_t mCleanupCount = -1;
 };
 
 } // namespace o2::framework

Framework/Core/include/Framework/DeviceState.h

@@ -19,6 +19,7 @@
 #include <string>
 #include <map>
 #include <utility>
+#include <atomic>
 
 typedef struct uv_loop_s uv_loop_t;
 typedef struct uv_timer_s uv_timer_t;
@@ -59,6 +60,7 @@ struct DeviceState {
   std::vector<InputChannelInfo> inputChannelInfos;
   StreamingState streaming = StreamingState::Streaming;
   bool quitRequested = false;
+  std::atomic<int64_t> cleanupCount = -1;
 
   /// ComputingQuotaOffers which have not yet been
   /// evaluated by the ComputingQuotaEvaluator

Framework/Core/src/DataProcessingDevice.cxx

@@ -136,15 +136,16 @@ DataProcessingDevice::DataProcessingDevice(RunningDeviceRef running, ServiceRegi
     mServiceRegistry{registry},
     mProcessingPolicies{policies}
 {
-  GetConfig()->Subscribe<std::string>("dpl", [&cleanupCount = mCleanupCount, &registry = mServiceRegistry](const std::string& key, std::string value) {
+  GetConfig()->Subscribe<std::string>("dpl", [&registry = mServiceRegistry](const std::string& key, std::string value) {
     if (key == "cleanup") {
+      auto ref = ServiceRegistryRef{registry, ServiceRegistry::globalDeviceSalt()};
+      auto& deviceState = ref.get<DeviceState>();
+      int64_t cleanupCount = deviceState.cleanupCount.load();
       int64_t newCleanupCount = std::stoll(value);
       if (newCleanupCount <= cleanupCount) {
         return;
       }
-      cleanupCount = newCleanupCount;
-      auto ref = ServiceRegistryRef{registry, ServiceRegistry::globalDeviceSalt()};
-      auto& deviceState = ref.get<DeviceState>();
+      deviceState.cleanupCount.store(newCleanupCount);
       for (auto& info : deviceState.inputChannelInfos) {
         fair::mq::Parts parts;
         while (info.channel->Receive(parts, 0)) {
@@ -982,6 +983,10 @@ void DataProcessingDevice::InitTask()
     uv_signal_init(state.loop, deviceContext.sigusr1Handle);
     uv_signal_start(deviceContext.sigusr1Handle, on_signal_callback, SIGUSR1);
   }
+  // If there is any signal, we want to make sure they are active
+  for (auto& handle : state.activeSignals) {
+    handle->data = &state;
+  }
   // When we start, we must make sure that we do listen to the signal
   deviceContext.sigusr1Handle->data = &mServiceRegistry;
 
@@ -1691,6 +1696,11 @@ void DataProcessingDevice::ResetTask()
   if (deviceContext.sigusr1Handle) {
     deviceContext.sigusr1Handle->data = nullptr;
   }
+  // Makes sure we do not have a working context on
+  // shutdown.
+  for (auto& handle : ref.get<DeviceState>().activeSignals) {
+    handle->data = nullptr;
+  }
 }
 
 struct WaitBackpressurePolicy {

Framework/Core/src/DeviceSpecHelpers.cxx

@@ -90,6 +90,9 @@ void signal_callback(uv_signal_t* handle, int)
 {
   // We simply wake up the event loop. Nothing to be done here.
   auto* state = (DeviceState*)handle->data;
+  if (!state) {
+    return;
+  }
   state->loopReason |= DeviceState::SIGNAL_ARRIVED;
   state->loopReason |= DeviceState::DATA_INCOMING;
 }

Framework/Core/src/ExternalFairMQDeviceProxy.cxx

@@ -687,53 +687,25 @@ DataProcessorSpec specifyExternalFairMQDeviceProxy(char const* name,
       if (fair::mq::State{state} != fair::mq::State::Ready) {
         return;
       }
-      // We keep track of whether or not all channels have seen a new state.
-      std::vector<bool> lastNewStatePending(deviceState.inputChannelInfos.size(), false);
       uv_update_time(deviceState.loop);
-      auto start = uv_now(deviceState.loop);
-
-      // Continue iterating until all channels have seen a new state.
-      while (std::all_of(lastNewStatePending.begin(), lastNewStatePending.end(), [](bool b) { return b; }) != true) {
-        if (uv_now(deviceState.loop) - start > 5000) {
-          LOGP(info, "Timeout while draining messages, going to next state anyway.");
-          break;
-        }
+      bool doDrain = true;
+      // Cleanup count is set by the cleanup property of the device.
+      // It is incremented every time the device is cleaned up.
+      // We use it to detect when the device is cleaned up.
+      int64_t cleanupCount = deviceState.cleanupCount.load();
+
+      // Continue iterating we saw the cleanup property being reset or
+      // the device state changing.
+      while (doDrain) {
+        doDrain = device->NewStatePending() || deviceState.cleanupCount == cleanupCount;
         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);
-          // 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;
-          }
-          if (!lastNewStatePending[ci]) {
-            LOGP(warn, "Unexpected {} message on channel {} while in Ready state. Dropping.", parts.Size(), info.channel->GetName());
-          } else if (lastNewStatePending[ci]) {
-            LOGP(detail, "Some {} parts were received on channel {} while switching away from Ready. Keeping.", parts.Size(), info.channel->GetName());
-            for (int pi = 0; pi < parts.Size(); ++pi) {
-              info.parts.fParts.emplace_back(std::move(parts.At(pi)));
-            }
-            info.readPolled = true;
-          }
         }
         // Keep state transitions going also when running with the standalone GUI.
         uv_run(deviceState.loop, UV_RUN_NOWAIT);

Generators/CMakeLists.txt

@@ -50,6 +50,7 @@ o2_add_library(Generators
                        $<$<BOOL:${pythia_FOUND}>:src/DecayerPythia8Param.cxx>
                        $<$<BOOL:${HepMC3_FOUND}>:src/GeneratorHepMC.cxx>
                        $<$<BOOL:${HepMC3_FOUND}>:src/GeneratorHepMCParam.cxx>
+                       $<$<BOOL:${HepMC3_FOUND}>:src/AODToHepMC.cxx>
                PUBLIC_LINK_LIBRARIES FairRoot::Base O2::SimConfig O2::CommonUtils O2::DetectorsBase O2::ZDCBase
                                      O2::SimulationDataFormat ${pythia6Target} ${pythiaTarget} ${hepmcTarget}
                                      FairRoot::Gen
@@ -106,6 +107,7 @@ endif()
 if(HepMC3_FOUND)
   list(APPEND headers include/Generators/GeneratorHepMC.h)
   list(APPEND headers include/Generators/GeneratorHepMCParam.h)
+  list(APPEND headers include/Generators/AODToHepMC.h)
 endif()
 
 o2_target_root_dictionary(Generators HEADERS ${headers})