From 6cf9a20aba75de3d0d718b9fae36a65a80a1c1a3 Mon Sep 17 00:00:00 2001 From: Sandro Wenzel Date: Thu, 5 Nov 2026 20:44:14 +0100 Subject: [PATCH 1/2] Flush queued hits before an o2-sim worker exits This is fixing a bug where o2-sim hangs when the hit merger starts after the sim workers have finished. A worker called _exit with hits still queued in its simdata socket, so they were lost. The data channel now has an infinite linger and the worker closes it before exiting, which blocks until the merger has taken the hits. A parent-death signal makes sure a blocked worker does not outlive the o2-sim process tree. Co-Authored-By: Claude Opus 5.5 --- run/O2SimDeviceRunner.cxx | 45 +++++++++++++++++++++++++-------------- 1 file changed, 29 insertions(+), 16 deletions(-) diff --git a/run/O2SimDeviceRunner.cxx b/run/O2SimDeviceRunner.cxx index 87b6ae99df5f4..11da3792d0ac5 100644 --- a/run/O2SimDeviceRunner.cxx +++ b/run/O2SimDeviceRunner.cxx @@ -31,6 +31,9 @@ #include #include #include +#ifdef __linux__ +#include +#endif #include "SimPublishChannelHelper.h" #include "rapidjson/document.h" @@ -48,9 +51,9 @@ void sigaction_handler(int signal, siginfo_t* signal_info, void*) auto pid = getpid(); LOG(info) << pid << " caught signal " << signal << " from source " << signal_info->si_pid; auto groupid = getpgrp(); - if (pid == gMasterProcess && signal_info->si_pid != gDriverProcess) { + if (pid == gMasterProcess && (signal_info->si_pid != gDriverProcess || signal == SIGTERM)) { // master worker forwards signal to whole worker process group - // (do this only if not coming from gDriverProcess since this uses killpg and already affected all children) + // (always for SIGTERM, since a parent-death signal also carries the driver pid) killpg(pid, signal); } else { if (signal_info->si_pid != gDriverProcess) { @@ -168,6 +171,7 @@ KernelSetup initSim(std::string transport, std::string primaddress, std::string auto datachannel = new fair::mq::Channel{"simdata", "push", factory}; datachannel->Connect(mergeraddress); datachannel->Validate(); + datachannel->GetSocket().SetLinger(-1); // keep queued hits until the merger has taken them // the channels are setup // init the sim object @@ -393,6 +397,13 @@ int main(int argc, char* argv[]) // we use the current process as one of the workers as it has nothing else to do auto pid = (i == nworkers - 1) ? 0 : fork(); if (pid == 0) { +#ifdef __linux__ + // terminate if our parent dies, so a worker never outlives the o2-sim process tree + prctl(PR_SET_PDEATHSIG, SIGTERM); + if (getppid() != ((i == nworkers - 1) ? driverPID : gMasterProcess)) { + _exit(1); + } +#endif // Each worker can publish its progress/state on a ZMQ channel. // We actually use a push/pull mechanism to collect all messages in the // master worker which can then publish using PUB/SUB. @@ -418,7 +429,7 @@ int main(int argc, char* argv[]) } // everyone else is getting a push socket for notifications - auto pushchannel = o2::simpubsub::createPUBChannel(o2::simpubsub::getPublishAddress("o2sim-workerinternal", driverPID), "push"); + auto pushchannel = std::make_unique(o2::simpubsub::createPUBChannel(o2::simpubsub::getPublishAddress("o2sim-workerinternal", driverPID), "push")); // we will try to pin each worker to a particular CPU // this can be made configurable via environment variables?? @@ -428,7 +439,7 @@ int main(int argc, char* argv[]) std::stringstream worker; worker << "WORKER" << i; - o2::simpubsub::publishMessage(pushchannel, o2::simpubsub::simStatusString(worker.str(), "STATUS", "SETUP COMPLETED")); + o2::simpubsub::publishMessage(*pushchannel, o2::simpubsub::simStatusString(worker.str(), "STATUS", "SETUP COMPLETED")); auto& conf = o2::conf::SimConfig::Instance(); @@ -438,18 +449,28 @@ int main(int argc, char* argv[]) if (conf.asService()) { LOG(info) << "IN SERVICE MODE WAITING"; - o2::simpubsub::publishMessage(pushchannel, o2::simpubsub::simStatusString(worker.str(), "STATUS", "AWAITING INPUT")); + o2::simpubsub::publishMessage(*pushchannel, o2::simpubsub::simStatusString(worker.str(), "STATUS", "AWAITING INPUT")); more = waitForControlInput(kernelSetup.workerID); usleep(100); // --> why? (probably to give the server some chance to come to a "serving" state) } else { - o2::simpubsub::publishMessage(pushchannel, o2::simpubsub::simStatusString(worker.str(), "STATUS", "TERMINATING")); + o2::simpubsub::publishMessage(*pushchannel, o2::simpubsub::simStatusString(worker.str(), "STATUS", "TERMINATING")); LOG(info) << "FINISHING"; more = false; } } - sleep(10); // ---> give some time for message to be delivered to merger (destructing too early might affect the ZQM buffers) - // The process will in any case be terminated by the main o2-sim driver. + // Close the worker transport; this blocks until the merger has taken all queued hits + pushchannel.reset(); + delete kernelSetup.primchannel; + delete kernelSetup.primstatuschannel; + delete kernelSetup.datachannel; + + if (i == nworkers - 1) { + // the master reaps its forked workers + int status, cpid; + while ((cpid = wait(&status)) != -1) { + } + } // destruct setup (using _exit due to problems in ROOT shutdown (segmentation violations) // Clearly at some moment, a more robust solution would be appreciated @@ -458,14 +479,6 @@ int main(int argc, char* argv[]) gChildProcesses.push_back(pid); } } - int status, cpid; - while ((cpid = wait(&status))) { - // LOG(info) << "normal wait " << cpid << " returned "; - if (cpid == -1) { - break; - } - } - _exit(0); } else { // This the solution where we setup an ordinary fair::mq::Device // (each if which will setup its own simulation). Parallelism From 033437cd7df3ec626fac1af4415af23c729a4112 Mon Sep 17 00:00:00 2001 From: Sandro Wenzel Date: Thu, 5 Nov 2026 20:02:17 +0100 Subject: [PATCH 2/2] 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 --- run/o2sim_parallel.cxx | 84 +++++++++++++++++++++--------------------- 1 file changed, 43 insertions(+), 41 deletions(-) diff --git a/run/o2sim_parallel.cxx b/run/o2sim_parallel.cxx index 0d5ff361141a8..08e3e24935d0c 100644 --- a/run/o2sim_parallel.cxx +++ b/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 shutdown_initiated = false; + std::function 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 const& v) { + finishCallback = [&shutdown_initiated, &conf, &externalpublishchannel](std::vector 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;