Fix CPU accounting for external generator subprocesses

GeneratorFileOrCmd forks an external generator but the child was only
reaped from ~GeneratorHepMC(), which is not reached on the DPL device
teardown path. Its CPU therefore never showed up in RUSAGE_CHILDREN.

Generators: add Generator::stop(), called from GeneratorTask::run() once
the stream ends, to reap the child deterministically.

Framework: add a Monitoring .stop callback to take the final CPU
measurement on the RUNNING -> READY transition. This also covers devices
that end their own stream via readyToQuit(), for which postEOS is never
invoked.

Details:

* Reaping alone is not enough. terminateCmd() SIGKILLs the process group,
  and a generator started through a wrapper script - epos.sh, and any
  command line that /bin/sh does not exec in place - is a grandchild whose
  CPU reaches us only through the intermediate shell's own reap. Killing
  that shell first loses the accounting completely, which made the fix a
  race that usually lost. terminateCmd() now waits a bounded time for the
  command to exit by itself before escalating to SIGKILL, and
  GeneratorHepMC::stop() closes the read end of the pipe first, so a
  generator still blocked writing to it sees EPIPE and exits promptly
  instead of sitting out the grace period.

  Measured with a Pythia8 -> HepMC3 external command (same sh -> binary
  -> fifo topology as EPOS4, 3.5s of child CPU, ground truth from the
  child's own getrusage): cpuTimeConsumedByProcess was correct in 3 of 10
  runs before and in 10 of 10 after, with no change in wall time. Same
  result when the generator produces more events than the device consumes
  (0.18s reported before, 3.62s after).

* GeneratorTParticle spawns a command the same way but never called
  terminateCmd() at all, leaking the process as well as its CPU. It now
  overrides stop() too.

* The .stop callback stops the sampling thread, while process monitoring
  is armed only once, at device instantiation. Without a counterpart a
  device would report nothing at all from its second run onwards
  (reproduced: no periodic samples and no aggregates after the first
  stop), so .start re-arms it. That call is a no-op while monitoring is
  already running.

* Requires AliceO2Group/Monitoring#378, which introduces
  finalizeProcessMonitoring() and the RUSAGE_CHILDREN sampling. That has
  to be merged, tagged and pinned in alidist before this one, or dev stops
  compiling.

Fixes https://its.cern.ch/jira/browse/O2-7096

Assisted by Claude Opus 5

Author: sawenzel

Framework/Core/src/CommonServices.cxx

@@ -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();

Generators/include/Generators/Generator.h

@@ -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; }
 

Generators/include/Generators/GeneratorFileOrCmd.h

@@ -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

Generators/include/Generators/GeneratorHepMC.h

@@ -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; };

Generators/include/Generators/GeneratorService.h

@@ -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;

Generators/include/Generators/GeneratorTParticle.h

@@ -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

Generators/src/GeneratorFileOrCmd.cxx

@@ -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);

Generators/src/GeneratorHepMC.cxx

@@ -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,

Generators/src/GeneratorService.cxx

@@ -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();
+    }
+  }
+}

Generators/src/GeneratorTParticle.cxx

@@ -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());