Skip to content

Fix o2-sim hang when the hit merger starts after the workers #16048

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
sawenzel wants to merge 2 commits into
base: dev
Choose a base branch
from
+72 −57
Open

Fix o2-sim hang when the hit merger starts after the workers #16048

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6cf9a20
Flush queued hits before an o2-sim worker exits
sawenzel Nov 5, 2026
033437c
Spawn the o2-sim hit merger before the sim workers
sawenzel Nov 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 29 additions & 16 deletions run/O2SimDeviceRunner.cxx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,9 @@
#include <cmath>
#include <csignal>
#include <unistd.h>
#ifdef __linux__
#include <sys/prctl.h>
#endif
#include "SimPublishChannelHelper.h"

#include "rapidjson/document.h"
Expand All @@ -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) {
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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.
Expand All @@ -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<fair::mq::Channel>(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??
Expand All @@ -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();

Expand All @@ -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
Expand All @@ -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
Expand Down
84 changes: 43 additions & 41 deletions run/o2sim_parallel.cxx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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");
Expand All @@ -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
Expand All @@ -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()));
Expand All @@ -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;
Expand Down
Loading