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