Spawn the o2-sim hit merger before the sim workers

The merger is now listening when the workers send their first hits, so fast
workers (e.g. with --noGeant) no longer wait for it at exit.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Author: sawenzel

run/o2sim_parallel.cxx

@@ -634,43 +634,8 @@ int main(int argc, char* argv[])
     sleep(1); // otherwise wait until server is initialized
   }
 
-  auto internalfork = getenv("ALICE_SIMFORKINTERNAL");
-  if (internalfork) {
-    // forking will be done internally to profit from copy-on-write
-    nworkers = 1;
-  }
-  for (int id = 0; id < nworkers; ++id) {
-    // the workers
-    std::stringstream workerlogss;
-    workerlogss << getWorkerLogName() << id;
-
-    // the workers
-    std::stringstream workerss;
-    workerss << "worker" << id;
-
-    pid = fork();
-    if (pid == 0) {
-      int fd = open(workerlogss.str().c_str(), O_RDWR | O_CREAT, S_IRUSR | S_IWUSR);
-      dup2(fd, 1); // make stdout go to file
-      dup2(fd, 2); // make stderr go to file - you may choose to not do this
-                   // or perhaps send stderr to another file
-      close(fd);   // fd no longer needed - the dup'ed handles are sufficient
-
-      const std::string name("o2-sim-device-runner");
-      const std::string path = installpath + "/" + name;
-
-      execl(path.c_str(), name.c_str(), "--control", "static", "--id", workerss.str().c_str(), "--config-key",
-            "worker", "--mq-config", localconfig.c_str(), "--severity", "info", (char*)nullptr);
-      return 0;
-    } else {
-      gChildProcesses.push_back(pid);
-      setpgid(pid, pid); // the worker processes will form their own group
-      std::cout << "Spawning sim worker " << id << " on PID " << pid
-                << "; Redirect output to " << workerlogss.str() << "\n";
-    }
-  }
-
-  // the hit merger
+  int mergerpid = -1;
+  // the hit merger, spawned before the workers so that it listens when they send hits
   int pipe_mergerdriver_fd[2];
   if (pipe(pipe_mergerdriver_fd) != 0) {
     perror("problem in creating pipe");
@@ -679,6 +644,7 @@ int main(int argc, char* argv[])
   pid = fork();
 
   std::atomic<bool> shutdown_initiated = false;
+  std::function<bool(std::vector<int> const&)> finishCallback;
   if (pid == 0) {
     int fd = open(getMergerLogName().c_str(), O_RDWR | O_CREAT, S_IRUSR | S_IWUSR);
     dup2(fd, 1); // make stdout go to file
@@ -696,12 +662,13 @@ int main(int argc, char* argv[])
     std::cout << "Spawning hit merger on PID " << pid << "; Redirect output to " << getMergerLogName() << "\n";
     setpgid(pid, pid);
     gChildProcesses.push_back(pid);
+    mergerpid = pid;
     close(pipe_mergerdriver_fd[1]);
 
     // A simple callback that determines if the simulation is complete and triggers
     // a shutdown of all child processes. This appears to be more robust than leaving
     // that decision upon the children (sometimes there are problems with that).
-    auto finishCallback = [&shutdown_initiated, &conf, &externalpublishchannel](std::vector<int> const& v) {
+    finishCallback = [&shutdown_initiated, &conf, &externalpublishchannel](std::vector<int> const& v) {
       std::stringstream str;
       str << "EVENT " << v.back() << " FINISHED " << gAskedEvents << " " << v.size();
       o2::simpubsub::publishMessage(externalpublishchannel, o2::simpubsub::simStatusString("O2SIM", "INFO", str.str()));
@@ -723,13 +690,48 @@ int main(int argc, char* argv[])
       }
       return true;
     };
+  }
 
-    launchThreadMonitoringEvents(pipe_mergerdriver_fd[0], "EVENT FINISHED : ", gFinishedEvents, finishCallback);
+  auto internalfork = getenv("ALICE_SIMFORKINTERNAL");
+  if (internalfork) {
+    // forking will be done internally to profit from copy-on-write
+    nworkers = 1;
   }
+  for (int id = 0; id < nworkers; ++id) {
+    // the workers
+    std::stringstream workerlogss;
+    workerlogss << getWorkerLogName() << id;
 
-  // wait on merger (which when exiting completes the workflow)
-  auto mergerpid = gChildProcesses.back();
+    // the workers
+    std::stringstream workerss;
+    workerss << "worker" << id;
+
+    pid = fork();
+    if (pid == 0) {
+      int fd = open(workerlogss.str().c_str(), O_RDWR | O_CREAT, S_IRUSR | S_IWUSR);
+      dup2(fd, 1); // make stdout go to file
+      dup2(fd, 2); // make stderr go to file - you may choose to not do this
+                   // or perhaps send stderr to another file
+      close(fd);   // fd no longer needed - the dup'ed handles are sufficient
+
+      const std::string name("o2-sim-device-runner");
+      const std::string path = installpath + "/" + name;
+
+      execl(path.c_str(), name.c_str(), "--control", "static", "--id", workerss.str().c_str(), "--config-key",
+            "worker", "--mq-config", localconfig.c_str(), "--severity", "info", (char*)nullptr);
+      return 0;
+    } else {
+      gChildProcesses.push_back(pid);
+      setpgid(pid, pid); // the worker processes will form their own group
+      std::cout << "Spawning sim worker " << id << " on PID " << pid
+                << "; Redirect output to " << workerlogss.str() << "\n";
+    }
+  }
 
+  // monitor the merger only once all child pids are known to the shutdown callback
+  launchThreadMonitoringEvents(pipe_mergerdriver_fd[0], "EVENT FINISHED : ", gFinishedEvents, finishCallback);
+
+  // wait on merger (which when exiting completes the workflow)
   int status, cpid;
   // wait just blocks and waits until any child returns; but we make sure to wait until merger is here
   bool errored = false;