DPL GUI: report correct status when done (#5883)

Now a device will send a "DONE" metric associated to a given
input in a slot which was processed. This way we can show a green
square for completed computations, rather than a RUNNING one.

Notice that due to the way the data relaying works, a DONE slot is
really already empty and ready to accept new data, so we only show
it as green for a single iteration and then we switch it to EMPTY.

Author: ktf

Framework/Core/include/Framework/DataRelayer.h

@@ -41,6 +41,13 @@ struct DataRelayerStats {
   uint64_t relayedMessages = 0;         /// How many messages have been successfully relayed
 };
 
+enum struct CacheEntryStatus : int {
+  EMPTY,
+  PENDING,
+  RUNNING,
+  DONE
+};
+
 class DataRelayer
 {
  public:
@@ -107,6 +114,9 @@ class DataRelayer
   /// so that we can mutex on it.
   TimesliceId getTimesliceForSlot(TimesliceSlot slot);
 
+  /// Mark a given slot as done so that the GUI
+  /// can reflect that.
+  void updateCacheStatus(TimesliceSlot slot, CacheEntryStatus oldStatus, CacheEntryStatus newStatus);
   /// Get the firstTFOrbit associate to a given slot.
   uint32_t getFirstTFOrbitForSlot(TimesliceSlot slot);
   /// Get the firstTFCounter associate to a given slot.
@@ -131,7 +141,7 @@ class DataRelayer
   std::vector<size_t> mDistinctRoutesIndex;
   std::vector<data_matcher::DataDescriptorMatcher> mInputMatchers;
   std::vector<data_matcher::VariableContext> mVariableContextes;
-  std::vector<int> mCachedStateMetrics;
+  std::vector<CacheEntryStatus> mCachedStateMetrics;
 
   static std::vector<std::string> sMetricsNames;
   static std::vector<std::string> sVariablesMetricsNames;

Framework/Core/src/DataProcessingDevice.cxx

@@ -838,6 +838,10 @@ bool DataProcessingDevice::tryDispatchComputation(DataProcessorContext& context,
     return InputRecord{spec->inputs, std::move(span)};
   };
 
+  auto markInputsAsDone = [&relayer = context.relayer](TimesliceSlot slot) -> void {
+    relayer->updateCacheStatus(slot, CacheEntryStatus::RUNNING, CacheEntryStatus::DONE);
+  };
+
   // I need a preparation step which gets the current timeslice id and
   // propagates it to the various contextes (i.e. the actual entities which
   // create messages) because the messages need to have the timeslice id into
@@ -1025,6 +1029,7 @@ bool DataProcessingDevice::tryDispatchComputation(DataProcessorContext& context,
         continue;
       }
     }
+    markInputsAsDone(action.slot);
 
     uint64_t tStart = uv_hrtime();
     preUpdateStats(action, record, tStart);

Framework/Core/src/DataRelayer.cxx

@@ -255,7 +255,7 @@ DataRelayer::RelayChoice
     assert(numInputTypes * slot.index < cache.size());
     for (size_t ai = slot.index * numInputTypes, ae = ai + numInputTypes; ai != ae; ++ai) {
       cache[ai].clear();
-      cachedStateMetrics[ai] = 0;
+      cachedStateMetrics[ai] = CacheEntryStatus::EMPTY;
     }
   };
 
@@ -268,7 +268,7 @@ DataRelayer::RelayChoice
                      &metrics](TimesliceId timeslice, int input, TimesliceSlot slot) {
     auto cacheIdx = numInputTypes * slot.index + input;
     std::vector<PartRef>& parts = cache[cacheIdx].parts;
-    cachedStateMetrics[cacheIdx] = 1;
+    cachedStateMetrics[cacheIdx] = CacheEntryStatus::PENDING;
     // TODO: make sure that multiple parts can only be added within the same call of
     // DataRelayer::relay
     PartRef entry{std::move(header), std::move(payload)};
@@ -488,6 +488,25 @@ void DataRelayer::getReadyToProcess(std::vector<DataRelayer::RecordAction>& comp
   }
 }
 
+void DataRelayer::updateCacheStatus(TimesliceSlot slot, CacheEntryStatus oldStatus, CacheEntryStatus newStatus)
+{
+  std::scoped_lock<LockableBase(std::recursive_mutex)> lock(mMutex);
+  const auto numInputTypes = mDistinctRoutesIndex.size();
+  auto& index = mTimesliceIndex;
+
+  auto markInputDone = [&cachedStateMetrics = mCachedStateMetrics,
+                        &index, &numInputTypes](TimesliceSlot s, size_t arg, CacheEntryStatus oldStatus, CacheEntryStatus newStatus) {
+    auto cacheId = s.index * numInputTypes + arg;
+    if (cachedStateMetrics[cacheId] == oldStatus) {
+      cachedStateMetrics[cacheId] = newStatus;
+    }
+  };
+
+  for (size_t ai = 0, ae = numInputTypes; ai != ae; ++ai) {
+    markInputDone(slot, ai, oldStatus, newStatus);
+  }
+}
+
 std::vector<o2::framework::MessageSet> DataRelayer::getInputsForTimeslice(TimesliceSlot slot)
 {
   std::scoped_lock<LockableBase(std::recursive_mutex)> lock(mMutex);
@@ -512,7 +531,7 @@ std::vector<o2::framework::MessageSet> DataRelayer::getInputsForTimeslice(Timesl
                                     &cachedStateMetrics = mCachedStateMetrics,
                                     &cache, &index, &numInputTypes, &metrics](TimesliceSlot s, size_t arg) {
     auto cacheId = s.index * numInputTypes + arg;
-    cachedStateMetrics[cacheId] = 2;
+    cachedStateMetrics[cacheId] = CacheEntryStatus::RUNNING;
     // TODO: in the original implementation of the cache, there have been only two messages per entry,
     // check if the 2 above corresponds to the number of messages.
     if (cache[cacheId].size() > 0) {
@@ -634,7 +653,12 @@ void DataRelayer::sendContextState()
                                mMetrics, sVariablesMetricsNames);
   }
   for (size_t si = 0; si < mCachedStateMetrics.size(); ++si) {
-    mMetrics.send({mCachedStateMetrics[si], sMetricsNames[si]});
+    mMetrics.send({static_cast<int>(mCachedStateMetrics[si]), sMetricsNames[si]});
+    // Anything which is done is actually already empty,
+    // so after we report it we mark it as such.
+    if (mCachedStateMetrics[si] == CacheEntryStatus::DONE) {
+      mCachedStateMetrics[si] = CacheEntryStatus::EMPTY;
+    }
   }
 }