DPL: add metric for shm memory usage

DPL: add metric for shm memory usage

Author: ktf

Framework/Core/include/Framework/DataProcessingStats.h

@@ -37,6 +37,7 @@ struct DataProcessingStats {
   std::atomic<int> totalProcessedSize = 0;
   std::atomic<int> totalSigusr1 = 0;
   std::atomic<int> consumedTimeframes = 0;
+  std::atomic<uint64_t> availableManagedShm = 0; /// Available shared memory in bytes.
 
   std::atomic<uint64_t> lastSlowMetricSentTimestamp = 0; /// The timestamp of the last time we sent slow metrics
   std::atomic<uint64_t> lastMetricFlushedTimestamp = 0;  /// The timestamp of the last time we actually flushed metrics

Framework/Core/include/Framework/RunningWorkflowInfo.h

@@ -23,6 +23,8 @@ struct RunningDeviceRef {
 
 /// Information about the running workflow
 struct RunningWorkflowInfo {
+  std::string uniqueWorkflowId = "";
+  int16_t shmSegmentId;
   std::vector<DeviceSpec> devices;
 };
 

Framework/Core/src/CommonServices.cxx

@@ -27,6 +27,7 @@
 #include "InputRouteHelpers.h"
 #include "Framework/EndOfStreamContext.h"
 #include "Framework/RawDeviceService.h"
+#include "Framework/RunningWorkflowInfo.h"
 #include "Framework/Tracing.h"
 #include "Framework/Monitoring.h"
 #include "TextDriverClient.h"
@@ -43,6 +44,7 @@
 #include <InfoLogger/InfoLogger.hxx>
 
 #include <FairMQDevice.h>
+#include <fairmq/shmem/Monitor.h>
 #include <options/FairMQProgOptions.h>
 
 #include <cstdlib>
@@ -438,6 +440,20 @@ auto sendRelayerMetrics(ServiceRegistry& registry, DataProcessingStats& stats) -
   if (timeSinceLastUpdate < 5000) {
     return;
   }
+  // Derive the amount of shared memory used
+  auto& runningWorkflow = registry.get<RunningWorkflowInfo const>();
+  using namespace fair::mq::shmem;
+  auto& spec = registry.get<DeviceSpec const>();
+
+  // FIXME: Ugly, but we do it only every 5 seconds...
+  if (spec.name == "readout-proxy") {
+    auto device = registry.get<RawDeviceService>().device();
+    try {
+      stats.availableManagedShm.store(Monitor::GetFreeMemory(SessionId{device->fConfig->GetProperty<std::string>("session")}, runningWorkflow.shmSegmentId));
+    } catch (...) {
+    }
+  }
+
   auto performedComputationsSinceLastUpdate = stats.performedComputations - stats.lastReportedPerformedComputations;
 
   ZoneScopedN("send metrics");
@@ -473,6 +489,10 @@ auto sendRelayerMetrics(ServiceRegistry& registry, DataProcessingStats& stats) -
                     .addTag(Key::Subsystem, Value::DPL));
   monitoring.send(Metric{((float)performedComputationsSinceLastUpdate / (float)timeSinceLastUpdate) * 1000, "processing_rate_hz"}.addTag(Key::Subsystem, Value::DPL));
 
+  if (stats.availableManagedShm) {
+    monitoring.send(Metric{(uint64_t)stats.availableManagedShm, fmt::format("available_managed_shm_{}", runningWorkflow.shmSegmentId)}.addTag(Key::Subsystem, Value::DPL));
+  }
+
   if (stats.consumedTimeframes) {
     monitoring.send(Metric{stats.consumedTimeframes, "consumed-timeframes"}.addTag(Key::Subsystem, Value::DPL));
   }

Framework/Core/src/runDataProcessing.cxx

@@ -72,9 +72,7 @@
 
 #include "FairMQDevice.h"
 #include <fairmq/DeviceRunner.h>
-#if __has_include(<fairmq/shmem/Monitor.h>)
 #include <fairmq/shmem/Monitor.h>
-#endif
 #include "options/FairMQProgOptions.h"
 
 #include <boost/program_options.hpp>
@@ -115,6 +113,9 @@
 #include <sys/wait.h>
 #include <unistd.h>
 #include <execinfo.h>
+// This is to allow C++20 aggregate initialisation
+#pragma GCC diagnostic push
+#pragma GCC diagnostic ignored "-Wpedantic"
 #if defined(__linux__) && __has_include(<sched.h>)
 #include <sched.h>
 #elif __has_include(<linux/getcpu.h>)
@@ -319,19 +320,8 @@ static void handle_sigint(int)
 /// Helper to invoke shared memory cleanup
 void cleanupSHM(std::string const& uniqueWorkflowId)
 {
-#if __has_include(<fairmq/shmem/Monitor.h>)
   using namespace fair::mq::shmem;
   Monitor::Cleanup(SessionId{"dpl_" + uniqueWorkflowId}, false);
-#else
-  // Old code, invoking external fairmq-shmmonitor
-  auto shmCleanup = fmt::format("fairmq-shmmonitor --cleanup -s dpl_{} 2>&1 >/dev/null", uniqueWorkflowId);
-  LOG(debug)
-    << "Cleaning up shm memory session with " << shmCleanup;
-  auto result = system(shmCleanup.c_str());
-  if (result != 0) {
-    LOG(error) << "Unable to cleanup shared memory, run " << shmCleanup << "by hand to fix";
-  }
-#endif
 }
 
 static void handle_sigchld(int) { sigchld_requested = true; }
@@ -1162,7 +1152,9 @@ int runStateMachine(DataProcessorSpecs const& workflow,
                     boost::program_options::variables_map& varmap,
                     std::string frameworkId)
 {
-  RunningWorkflowInfo runningWorkflow;
+  RunningWorkflowInfo runningWorkflow{
+    .uniqueWorkflowId = driverInfo.uniqueWorkflowId,
+    .shmSegmentId = (int16_t)atoi(varmap["shm-segment-id"].as<std::string>().c_str())};
   DeviceInfos infos;
   DeviceControls controls;
   DevicesManager* devicesManager = new DevicesManager{controls, infos, runningWorkflow.devices};
@@ -2573,3 +2565,4 @@ void doBoostException(boost::exception& e, char const* processName)
   LOGP(ERROR, "error while setting up workflow in {}: {}",
        processName, boost::current_exception_diagnostic_information(true));
 }
+#pragma GCC diagnostic push