DPL: do delta publishing of ContextVariables

ktf · ktf · commit a9e0d070d990 · 2022-09-06T19:30:53.000+02:00
This should bring the driver CPU usage close to 0, even when a lot of
data processors are present.
diff --git a/Framework/Core/include/Framework/DataDescriptorMatcher.h b/Framework/Core/include/Framework/DataDescriptorMatcher.h
@@ -14,6 +14,7 @@
 #include "Framework/ConcreteDataMatcher.h"
 #include "Framework/DataProcessingHeader.h"
 #include "Framework/RuntimeError.h"
+#include "Framework/TimesliceSlot.h"
 #include "Headers/DataHeader.h"
 
 #include <array>
@@ -62,6 +63,8 @@ struct ContextElement {
   using Value = std::variant<uint32_t, uint64_t, std::string, None>;
   char label[24];       /// The name of the variable contained in this element.
   Value value = None{}; /// The actual contents of the element.
+  size_t commitVersion = -1;  /// The committed version of the element. Every time we commit something to it, we bump the version.
+  size_t publishVersion = -1; /// The version of the element which has been published to the GUI.
 };
 
 struct ContextUpdate {
@@ -79,6 +82,12 @@ class VariableContext
 
   ContextElement::Value const& get(size_t pos) const;
 
+  /// Publish the context to the GUI / monitoring.
+  /// @a callback is a function which will be called for each element
+  /// @a context is userdata which will be passed to the callback.
+  /// @a name is the name of the metrics to be used.
+  void publish(void (*callback)(ContextElement::Value const&, std::string const& name, void* context), void* context, TimesliceSlot slot, std::vector<std::string> const& names);
+
   inline void put(ContextUpdate&& update);
 
   /// Use this after a query to actually commit the matched fields.  Notice the
diff --git a/Framework/Core/include/Framework/VariableContextHelpers.h b/Framework/Core/include/Framework/VariableContextHelpers.h
@@ -12,6 +12,7 @@
 #define O2_FRAMEWORK_DATADESCRIPTORMATCHER_H_
 
 #include "Framework/DataDescriptorMatcher.h"
+#include "Framework/TimesliceSlot.h"
 #include <variant>
 
 namespace o2::framework
diff --git a/Framework/Core/src/DataDescriptorMatcher.cxx b/Framework/Core/src/DataDescriptorMatcher.cxx
@@ -34,10 +34,27 @@ ContextElement::Value const& VariableContext::get(size_t pos) const
   return mElements.at(pos).value;
 }
 
+void VariableContext::publish(void (*callback)(ContextElement::Value const&, std::string const&, void*), void* context, TimesliceSlot slot, std::vector<std::string> const& names)
+{
+  for (size_t i = 0; i < MAX_MATCHING_VARIABLE; i++) {
+    auto& var = this->get(i);
+    auto& name = names[16 * slot.index + i];
+    auto& element = mElements[i];
+
+    if (element.commitVersion == element.publishVersion) {
+      continue;
+    }
+    callback(var, name, context);
+    element.publishVersion = element.commitVersion;
+  }
+}
+
 void VariableContext::commit()
 {
   for (size_t i = 0; i < mPerformedUpdates; ++i) {
-    mElements[mUpdates[i].position].value = mUpdates[i].newValue;
+    auto& element = mElements[mUpdates[i].position];
+    element.value = mUpdates[i].newValue;
+    element.commitVersion++;
   }
   mPerformedUpdates = 0;
 }
diff --git a/Framework/Core/src/DataRelayer.cxx b/Framework/Core/src/DataRelayer.cxx
@@ -251,21 +251,21 @@ size_t matchToContext(void const* data,
 void sendVariableContextMetrics(VariableContext& context, TimesliceSlot slot,
                                 monitoring::Monitoring& metrics, std::vector<std::string> const& names)
 {
-  const std::string nullstring{"null"};
+  static const std::string nullstring{"null"};
 
-  for (size_t i = 0; i < MAX_MATCHING_VARIABLE; i++) {
-    auto& var = context.get(i);
-    auto& name = names[16 * slot.index + i];
+  context.publish([](ContextElement::Value const& var, std::string const& name, void* context) {
+    monitoring::Monitoring* metrics = reinterpret_cast<monitoring::Monitoring*>(context);
     if (auto pval = std::get_if<uint64_t>(&var)) {
-      metrics.send(monitoring::Metric{std::to_string(*pval), name, Verbosity::Debug});
+      metrics->send(monitoring::Metric{std::to_string(*pval), name, Verbosity::Debug});
     } else if (auto pval = std::get_if<uint32_t>(&var)) {
-      metrics.send(monitoring::Metric{std::to_string(*pval), name, Verbosity::Debug});
+      metrics->send(monitoring::Metric{std::to_string(*pval), name, Verbosity::Debug});
     } else if (auto pval2 = std::get_if<std::string>(&var)) {
-      metrics.send(monitoring::Metric{*pval2, name, Verbosity::Debug});
+      metrics->send(monitoring::Metric{*pval2, name, Verbosity::Debug});
     } else {
-      metrics.send(monitoring::Metric{nullstring, name, Verbosity::Debug});
+      metrics->send(monitoring::Metric{nullstring, name, Verbosity::Debug});
     }
-  }
+  },
+                  &metrics, slot, names);
 }
 
 void DataRelayer::setOldestPossibleInput(TimesliceId proposed, ChannelIndex channel)
