added a monitoring queue to push spontaneous measurements

Author: sy-c

CMakeLists.txt

@@ -258,6 +258,7 @@ add_library(
         ${SOURCE_DIR}/MemoryBank.cxx
         ${SOURCE_DIR}/MemoryBankManager.cxx
         ${SOURCE_DIR}/MemoryPagesPool.cxx
+	${SOURCE_DIR}/ReadoutMonitoringQueue.cxx
 	$<$<BOOL:${ZMQ_FOUND}>:${SOURCE_DIR}/ZmqServer.cxx>
 	$<$<BOOL:${ZMQ_FOUND}>:${SOURCE_DIR}/ZmqClient.cxx>
 )

src/ConsumerStats.cxx

@@ -23,6 +23,7 @@
 #include "DataSet.h"
 #include "ReadoutUtils.h"
 #include "ReadoutStats.h"
+#include "ReadoutMonitoringQueue.h"
 
 using namespace o2::monitoring;
 
@@ -182,6 +183,12 @@ class ConsumerStats : public Consumer
 	  sendMetricNoException(Metric{"readout.bufferUsage"}.addValue((int)(r*100), "value").addValue(b, "bytes").addTag(tags::Key::ID, i));
 	}
       }
+
+      // publish measurements stored in monitoring queue
+      auto ff = [&] (const ReadoutMonitoringMetric &m) -> void {
+        sendMetricNoException(Metric{m.value, m.name}.addTag(tags::Key::ID, m.tag));
+      };
+      gReadoutMonitoringQueue.execute(ff);
     }
 
 #ifdef WITH_ZMQ

src/ReadoutMonitoringQueue.cxx

@@ -0,0 +1,36 @@
+#include "ReadoutMonitoringQueue.h"
+
+ReadoutMonitoringQueue::ReadoutMonitoringQueue() {
+}
+
+ReadoutMonitoringQueue::~ReadoutMonitoringQueue() {
+}
+  
+void ReadoutMonitoringQueue::push(ReadoutMonitoringMetric m) {
+  std::unique_lock<std::mutex> lock(qMutex);
+  q.push_front(std::move(m));
+}
+
+void ReadoutMonitoringQueue::execute(std::function<void(const ReadoutMonitoringMetric &)> f) {
+  for (;;) {
+    ReadoutMonitoringMetric m;
+    
+    {
+      std::unique_lock<std::mutex> lock(qMutex);
+      if (q.empty()) {
+	break;
+      }
+      m = std::move(q.back());
+      q.pop_back();
+    }
+    
+    f(m);
+  }
+} 
+  
+void ReadoutMonitoringQueue::clear() {
+  std::unique_lock<std::mutex> lock(qMutex);
+  q.clear();
+}
+
+ReadoutMonitoringQueue gReadoutMonitoringQueue;

src/ReadoutMonitoringQueue.h

@@ -0,0 +1,42 @@
+#include <deque>
+#include <mutex>
+#include <string>
+#include <functional>
+
+// a metric to be stored in queue for later processing
+// fields as for o2::monitoring::metric
+struct ReadoutMonitoringMetric {
+  std::string name;
+  unsigned short int tag;
+  uint64_t value;
+};
+
+
+// producer-consumer queue to define and publish metrics
+// typical use:
+// the module that pushes has no access to o2 Monitoring
+// the module that publishes reads from the queue and publish them to o2 Monitoring
+// the class is not aware of o2::monitoring, it's just a transient thread-safe storage
+
+class ReadoutMonitoringQueue {
+  public:
+  
+  ReadoutMonitoringQueue();
+  ~ReadoutMonitoringQueue();
+  
+  // push an element in the queue
+  void push(ReadoutMonitoringMetric);
+  
+  // execute provided functions on all elements in the queue
+  // (and remove them from the queue)
+  void execute(std::function<void(const ReadoutMonitoringMetric &)>); 
+  
+  // remove all elements in queue
+  void clear();
+  
+  private:
+  std::mutex qMutex;
+  std::deque<ReadoutMonitoringMetric> q;
+};
+
+extern ReadoutMonitoringQueue gReadoutMonitoringQueue;

src/mainReadout.cxx

@@ -79,6 +79,7 @@
 #include "ReadoutVersion.h"
 #include "TtyChecker.h"
 #include "ReadoutConst.h"
+#include "ReadoutMonitoringQueue.h"
 
 #ifdef WITH_NUMA
 #include <numa.h>
@@ -774,6 +775,9 @@ int Readout::_configure(const boost::property_tree::ptree& properties)
   // reset some flags
   gReadoutStats.isFairMQ = 0; // disable FMQ stats
 
+  // reset monitoring queue
+  gReadoutMonitoringQueue.clear();
+
   // load configuration file
   theLog.log(LogInfoSupport, "Reading configuration from %s %s", cfgFileURI, cfgFileEntryPoint);