AliceO2Group · sawenzel · Jul 25, 2026
@@ -54,12 +54,14 @@
 #include "DecongestionService.h"
 #include "ArrowSupport.h"
 #include "DPLMonitoringBackend.h"
+#include "ResourcesMonitoringHelper.h"
 #include "Headers/STFHeader.h"
 #include "Headers/DataHeader.h"
 
 #include <Configuration/ConfigurationInterface.h>
 #include <Configuration/ConfigurationFactory.h>
 #include <Monitoring/MonitoringFactory.h>
+#include <Monitoring/ProcessMonitor.h>
 #include "Framework/Signpost.h"
 
 #include <fairmq/Device.h>
@@ -119,6 +121,15 @@ o2::framework::ServiceSpec CommonServices::monitoringSpec()
     .start = [](ServiceRegistryRef services, void* service) {
       auto* monitoring = (o2::monitoring::Monitoring*)service;
 
+      // Re-arm process monitoring: .stop takes the final measurement and stops
+      // the sampling thread, so without this a device would report nothing at
+      // all from its second run onwards. A no-op while already running.
+      auto interval = services.get<DeviceSpec const>().resourceMonitoringInterval;
+      if (ResourcesMonitoringHelper::isResourcesMonitoringEnabled(interval)) {
+        using o2::monitoring::PmMeasurement;
+        monitoring->enableProcessMonitoring(interval, {PmMeasurement::Cpu, PmMeasurement::Mem, PmMeasurement::Smaps});
+      }
+
       auto extRunNumber = services.get<RawDeviceService>().device()->fConfig->GetProperty<std::string>("runNumber", "unspecified");
       if (extRunNumber == "unspecified") {
         return;
@@ -127,6 +138,12 @@ o2::framework::ServiceSpec CommonServices::monitoringSpec()
         monitoring->setRunNumber(std::stoul(extRunNumber));
       } catch (...) {
       } },
+    // Final measurement here rather than in ~Monitoring() at .exit, which is
+    // not reliably reached before the process exits. Unlike postEOS this also
+    // covers devices that quit themselves via readyToQuit().
+    .stop = [](ServiceRegistryRef, void* service) {
+                       auto* monitoring = reinterpret_cast<Monitoring*>(service);
+                       monitoring->finalizeProcessMonitoring(); },
     .exit = [](ServiceRegistryRef registry, void* service) {
                        auto* monitoring = reinterpret_cast<Monitoring*>(service);
                        monitoring->flushBuffer();

@@ -105,6 +105,10 @@ class Generator : public FairGenerator
   /** notification methods **/
   virtual void notifyEmbedding(const o2::dataformats::MCEventHeader* eventHeader){};
 
+  /** Release external resources (forked subprocesses, open files, ...) once
+   * the generator is no longer needed, without relying on destructor timing **/
+  virtual void stop() {}
+
   void setTriggerOkHook(std::function<void(std::vector<TParticle> const& p, int eventCount)> f) { mTriggerOkHook = f; }
   void setTriggerFalseHook(std::function<void(std::vector<TParticle> const& p, int eventCount)> f) { mTriggerFalseHook = f; }
 

@@ -143,9 +143,23 @@ struct GeneratorFileOrCmd {
    * Terminates the background command using PID of the child
    * process generated by fork.
    *
+   * @param graceMillis How long to wait for the command to exit on its
+   *        own before killing its process group.  A command started
+   *        through an intermediate shell (a wrapper script, say) only
+   *        contributes the CPU time of the actual generator to this
+   *        process' RUSAGE_CHILDREN once that shell has reaped it, which
+   *        it can no longer do once we have killed it.  Zero preserves
+   *        the previous kill-immediately behaviour.
+   *
    * @return true if the process was terminated successfully
    */
-  virtual bool terminateCmd();
+  virtual bool terminateCmd(unsigned int graceMillis = 0);
+  /**
+   * Time granted to the command to exit by itself when the generator is
+   * stopped normally. It has produced everything that was asked of it by
+   * then, so this only covers its teardown.
+   */
+  static constexpr unsigned int sStopGraceMillis = 5000;
   /**
    * Create a temporary file (and close it immediately).  On success,
    * the list of file names is cleared and the name of the temporary

@@ -86,6 +86,9 @@ class GeneratorHepMC : public Generator, public GeneratorFileOrCmd
    */
   Bool_t importParticles() override;
 
+  /** Terminate the background command (if any), see Generator::stop(). */
+  void stop() override;
+
   /** setters **/
   void setEventsToSkip(uint64_t val) { mEventsToSkip = val; };
   void setVersion(const int& ver) { mVersion = ver; };

@@ -70,6 +70,9 @@ class GeneratorService
   void generateEvent_MCTracks(o2::pmr::vector<MCTrack>& tracks, o2::dataformats::MCEventHeader& header);
   void generateEvent_TParticles(std::vector<TParticle>& tparts, o2::dataformats::MCEventHeader& header);
 
+  /** Calls Generator::stop() on all registered generators **/
+  void stopGenerators();
+
  private:
   PrimaryGenerator mPrimGen;
   o2::data::Stack mStack;

@@ -73,6 +73,9 @@ class GeneratorTParticle : public Generator, public GeneratorFileOrCmd
    * program */
   Bool_t Init() override;
 
+  /** Terminate the background command (if any), see Generator::stop(). */
+  void stop() override;
+
   /**
    * Configure the generator from parameters and the general
    * simulation configuration.  This is implemented as a member

@@ -142,13 +142,32 @@ bool GeneratorFileOrCmd::executeCmdLine(const std::string& cmd)
   return true;
 }
 // -----------------------------------------------------------------
-bool GeneratorFileOrCmd::terminateCmd()
+bool GeneratorFileOrCmd::terminateCmd(unsigned int graceMillis)
 {
   if (mCmdPid == -1) {
     LOG(info) << "No command is currently running";
     return false;
   }
 
+  // Let the command finish by itself if it is about to: killing the process
+  // group first would deprive an intermediate shell of the chance to reap the
+  // actual generator, and with it this process of the generator's CPU time,
+  // which only reaches RUSAGE_CHILDREN through that reap.
+  constexpr unsigned int pollMillis = 10;
+  for (unsigned int waited = 0; waited < graceMillis; waited += pollMillis) {
+    int status;
+    pid_t reaped = waitpid(mCmdPid, &status, WNOHANG);
+    if (reaped == mCmdPid) {
+      LOG(info) << "Command with process ID " << mCmdPid << " exited by itself";
+      mCmdPid = -1;
+      return true;
+    }
+    if (reaped == -1) {
+      break; // not our child (any more): let the kill path report it
+    }
+    std::this_thread::sleep_for(std::chrono::milliseconds(pollMillis));
+  }
+
   LOG(info) << "Terminating process ID group " << mCmdPid;
   if (kill(-mCmdPid, SIGKILL) == -1) {
     LOG(fatal) << "Failed to kill process: " << std::strerror(errno);

@@ -66,15 +66,29 @@ GeneratorHepMC::~GeneratorHepMC()
   if (mEvent) {
     delete mEvent;
   }
-  if (not mCmd.empty()) {
-    // Must be executed before removing the temporary file
-    // otherwise the current child process might still be writing on it
-    // causing unwanted stdout messages which could slow down the system
-    terminateCmd();
-  }
+  stop();
   removeTemp();
 }
 
+/*****************************************************************/
+
+void GeneratorHepMC::stop()
+{
+  if (mCmd.empty()) {
+    return;
+  }
+  // Close our end of the pipe first: a generator still blocked writing to it
+  // then sees EPIPE and exits promptly, instead of sitting out the whole grace
+  // period and being killed - which would lose its CPU time (see terminateCmd).
+  if (mReader) {
+    mReader->close();
+  }
+  // Must be executed before removing the temporary file
+  // otherwise the current child process might still be writing on it
+  // causing unwanted stdout messages which could slow down the system
+  terminateCmd(sStopGraceMillis);
+}
+
 /*****************************************************************/
 void GeneratorHepMC::setup(const GeneratorFileOrCmdParam& param0,
                            const GeneratorHepMCParam& param,

@@ -14,6 +14,8 @@
 #include "SimConfig/SimConfig.h"
 #include "Generators/Generator.h"
 #include "DataFormatsCalibration/MeanVertexObject.h"
+#include <TCollection.h>
+#include <TObjArray.h>
 
 using namespace o2::eventgen;
 
@@ -90,3 +92,17 @@ void GeneratorService::generateEvent_TParticles(std::vector<TParticle>& tracks,
   tracks.clear();
   tracks = mStack.getPrimaries();
 }
+
+void GeneratorService::stopGenerators()
+{
+  auto* generators = mPrimGen.GetListOfGenerators();
+  if (!generators) {
+    return;
+  }
+  TIter next(generators);
+  while (TObject* obj = next()) {
+    if (auto* gen = dynamic_cast<o2::eventgen::Generator*>(obj)) {
+      gen->stop();
+    }
+  }
+}
@@ -43,10 +43,20 @@ GeneratorTParticle::~GeneratorTParticle()
   if (mCmd.empty()) {
     return;
   }
-
+  // Must be executed before removing the temporary file, otherwise the child
+  // process might still be writing to it
+  stop();
   removeTemp();
 }
 /*****************************************************************/
+void GeneratorTParticle::stop()
+{
+  if (mCmd.empty()) {
+    return;
+  }
+  terminateCmd(sStopGraceMillis);
+}
+/*****************************************************************/
 Bool_t GeneratorTParticle::Init()
 {
   mChain = new TChain(mTreeName.c_str());

@@ -132,6 +132,9 @@ struct GeneratorTask {
       }
     }
     if (eventCounter >= nEvents || time_expired) {
+      // terminate and reap external generator subprocesses here: device
+      // teardown is not guaranteed to run the generators' destructors
+      genservice->stopGenerators();
       pc.services().get<ControlService>().endOfStream();
       pc.services().get<ControlService>().readyToQuit(QuitRequest::Me);