Add free SHM based TF rate throttling to raw proxy

davidrohr · davidrohr · commit 8568b609eee0 · 2021-11-08T19:49:21.000+01:00
diff --git a/Framework/Core/include/Framework/ExternalFairMQDeviceProxy.h b/Framework/Core/include/Framework/ExternalFairMQDeviceProxy.h
@@ -83,7 +83,8 @@ std::string defaultOutputProxyChannelSelector(InputSpec const& input, const std:
 DataProcessorSpec specifyExternalFairMQDeviceProxy(char const* label,
                                                    std::vector<OutputSpec> const& outputs,
                                                    const char* defaultChannelConfig,
-                                                   InjectorFunction converter);
+                                                   InjectorFunction converter,
+                                                   uint64_t minSHM = 0);
 
 DataProcessorSpec specifyFairMQDeviceOutputProxy(char const* label,
                                                  Inputs const& inputSpecs,
diff --git a/Framework/Core/src/ExternalFairMQDeviceProxy.cxx b/Framework/Core/src/ExternalFairMQDeviceProxy.cxx
@@ -22,6 +22,7 @@
 #include "Framework/ControlService.h"
 #include "Framework/SourceInfoHeader.h"
 #include "Framework/ConfigParamRegistry.h"
+#include "Framework/RunningWorkflowInfo.h"
 #include "Headers/DataHeader.h"
 #include "Headers/Stack.h"
 
@@ -30,6 +31,7 @@
 
 #include <fairmq/FairMQParts.h>
 #include <fairmq/FairMQDevice.h>
+#include <fairmq/shmem/Monitor.h>
 #include <cstring>
 #include <cassert>
 #include <memory>
@@ -347,7 +349,8 @@ DataProcessorSpec specifyExternalFairMQDeviceProxy(char const* name,
                                                    std::function<void(FairMQDevice&,
                                                                       FairMQParts&,
                                                                       ChannelRetriever)>
-                                                     converter)
+                                                     converter,
+                                                   uint64_t minSHM)
 {
   DataProcessorSpec spec;
   spec.name = strdup(name);
@@ -356,7 +359,7 @@ DataProcessorSpec specifyExternalFairMQDeviceProxy(char const* name,
   // The Init method will register a new "Out of band" channel and
   // attach an OnData to it which is responsible for converting incoming
   // messages into DPL messages.
-  spec.algorithm = AlgorithmSpec{[converter, channel = spec.name](InitContext& ctx) {
+  spec.algorithm = AlgorithmSpec{[converter, channel = spec.name, minSHM](InitContext& ctx) {
     auto device = ctx.services().get<RawDeviceService>().device();
     // make a copy of the output routes and pass to the lambda by move
     auto outputRoutes = ctx.services().get<RawDeviceService>().spec().outputs;
@@ -411,7 +414,7 @@ DataProcessorSpec specifyExternalFairMQDeviceProxy(char const* name,
       }
     };
 
-    auto runHandler = [dataHandler, device, channel](ProcessingContext& ctx) {
+    auto runHandler = [dataHandler, device, channel, minSHM](ProcessingContext& ctx) {
       static int64_t consumedTimeframes = 0;
       static int64_t sentTimeframes = 0;
       auto device = ctx.services().get<RawDeviceService>().device();
@@ -439,6 +442,23 @@ DataProcessorSpec specifyExternalFairMQDeviceProxy(char const* name,
           LOG(important) << (sentTimeframes - consumedTimeframes) << " / " << maxTFInFlight << " TF in flight, continuing to publish";
         }
       }
+      if (minSHM) {
+        int waitMessage = 0;
+        auto& runningWorkflow = ctx.services().get<RunningWorkflowInfo const>();
+        while (true) {
+          uint64_t freeSHM = fair::mq::shmem::Monitor::GetFreeMemory(fair::mq::shmem::SessionId{device->fConfig->GetProperty<std::string>("session")}, runningWorkflow.shmSegmentId);
+          if (freeSHM > minSHM) {
+            if (waitMessage) {
+              LOG(important) << "Sufficient SHM memory free (" << freeSHM << " >= " << minSHM << "), continuing to publish";
+            }
+            break;
+          }
+          if (waitMessage == 0) {
+            LOG(alarm) << "Free SHM memory too low: " << freeSHM << " < " << minSHM << ", waiting";
+            waitMessage = 1;
+          }
+        }
+      }
 
       FairMQParts parts;
       device->Receive(parts, channel, 0);
diff --git a/Framework/Utils/src/raw-proxy.cxx b/Framework/Utils/src/raw-proxy.cxx
@@ -34,6 +34,10 @@ void customize(std::vector<ConfigParamSpec>& workflowOptions)
   workflowOptions.push_back(
     ConfigParamSpec{
       "throwOnUnmatched", VariantType::Bool, false, {"throw if unmatched input data is found"}});
+
+  workflowOptions.push_back(
+    ConfigParamSpec{
+      "timeframes-shm-limit", VariantType::String, "0", {"Minimum amount of SHM required in order to publish data"}});
 }
 
 #include "Framework/runDataProcessing.h"
@@ -43,6 +47,7 @@ WorkflowSpec defineDataProcessing(ConfigContext const& config)
   std::string processorName = config.options().get<std::string>("proxy-name");
   std::string outputconfig = config.options().get<std::string>("dataspec");
   bool throwOnUnmatched = config.options().get<bool>("throwOnUnmatched");
+  uint64_t minSHM = std::stoul(config.options().get<std::string>("timeframes-shm-limit"));
   std::vector<InputSpec> matchers = select(outputconfig.c_str());
   Outputs readoutProxyOutput;
   for (auto const& matcher : matchers) {
@@ -55,7 +60,7 @@ WorkflowSpec defineDataProcessing(ConfigContext const& config)
     processorName.c_str(),
     std::move(readoutProxyOutput),
     "type=pair,method=connect,address=ipc:///tmp/readout-pipe-0,rateLogging=1,transport=shmem",
-    dplModelAdaptor(filterSpecs, throwOnUnmatched));
+    dplModelAdaptor(filterSpecs, throwOnUnmatched), minSHM);
 
   WorkflowSpec workflow;
   workflow.emplace_back(readoutProxy);
