- check throttling also in the output module on each event.

This reduces the chance to overshoot and fill the disk even if check is done periodically in the input source. - updated unit tests to use new global evf output module
cms-sw · Jan 31, 2022 · 6092450 · 6092450
1 parent a0bdd3d
commit 6092450
Show file tree

Hide file tree

Showing 4 changed files with 28 additions and 2 deletions.
diff --git a/EventFilter/Utilities/plugins/GlobalEvFOutputModule.cc b/EventFilter/Utilities/plugins/GlobalEvFOutputModule.cc
@@ -12,6 +12,7 @@
 #include "FWCore/Framework/interface/RunForOutput.h"
 #include "FWCore/Framework/interface/LuminosityBlockForOutput.h"
 #include "FWCore/Framework/interface/LuminosityBlock.h"
+#include "FWCore/Utilities/interface/UnixSignalHandlers.h"
 
 #include "FWCore/Concurrency/interface/SerialTaskQueue.h"
 
@@ -56,6 +57,7 @@ namespace evf {
     }
 
     void doOutputEventAsync(std::unique_ptr<EventMsgBuilder> msg, edm::WaitingTaskHolder iHolder) {
+      throttledCheck();
       auto group = iHolder.group();
       writeQueue_.push(*group, [holder = std::move(iHolder), msg = msg.release(), this]() {
         try {
@@ -68,6 +70,18 @@ namespace evf {
       });
     }
 
+    inline void throttledCheck() {
+      unsigned int counter = 0;
+      while (edm::Service<evf::EvFDaqDirector>()->inputThrottled()) {
+        if (edm::shutdown_flag.load(std::memory_order_relaxed))
+          break;
+        if (!(counter % 100))
+          edm::LogWarning("FedRawDataInputSource") << "Input throttled detected, writing is paused...";
+        usleep(100000);
+        counter++;
+      }
+    }
+
     uint32 get_adler32() const { return stream_writer_events_->adler32(); }
 
     std::string const& getFilePath() const { return filePath_; }

diff --git a/EventFilter/Utilities/src/EvFOutputModule.cc b/EventFilter/Utilities/src/EvFOutputModule.cc
@@ -16,6 +16,7 @@
 
 #include "IOPool/Streamer/interface/InitMsgBuilder.h"
 #include "IOPool/Streamer/interface/EventMsgBuilder.h"
+#include "FWCore/Utilities/interface/UnixSignalHandlers.h"
 
 #include <sys/stat.h>
 #include <filesystem>
@@ -225,6 +226,17 @@ namespace evf {
   }
 
   void EvFOutputModule::write(edm::EventForOutput const& e) {
+
+    unsigned int counter = 0;
+    while (edm::Service<evf::EvFDaqDirector>()->inputThrottled()) {
+      if (edm::shutdown_flag.load(std::memory_order_relaxed))
+        break;
+      if (!(counter % 100))
+        edm::LogWarning("FedRawDataInputSource") << "Input throttled detected, writing is paused...";
+      usleep(100000);
+      counter++;
+    }
+
     edm::Handle<edm::TriggerResults> const& triggerResults = getTriggerResults(trToken_, e);
 
     //auto lumiWriter = const_cast<EvFOutputEventWriter*>(luminosityBlockCache(e.getLuminosityBlock().index() ));

diff --git a/EventFilter/Utilities/test/startFU.py b/EventFilter/Utilities/test/startFU.py
@@ -136,7 +136,7 @@
     SelectEvents = cms.untracked.PSet(SelectEvents = cms.vstring( 'p2' ))
 )
 
-process.streamC = cms.OutputModule("ShmStreamConsumer",
+process.streamC = cms.OutputModule("GlobalEvFOutputModule",
     SelectEvents = cms.untracked.PSet(SelectEvents = cms.vstring( 'p2' ))
 )
 

diff --git a/EventFilter/Utilities/test/unittest_FU.py b/EventFilter/Utilities/test/unittest_FU.py
@@ -145,7 +145,7 @@
     SelectEvents = cms.untracked.PSet(SelectEvents = cms.vstring( 'p2' ))
 )
 
-process.streamC = cms.OutputModule("ShmStreamConsumer",
+process.streamC = cms.OutputModule("GlobalEvFOutputModule",
     SelectEvents = cms.untracked.PSet(SelectEvents = cms.vstring( 'p2' ))
 )