o2-sim: More robust termination initiated by driver

sawenzel · sawenzel · commit 1b5a5f5b1de7 · 2021-03-05T14:12:27.000+01:00
So far, we've been relying on child-procs to return
correctly for termination. This has proven not to be
100% reliable (FairMQ sometimes hangs during shutdown).

The new treatment initiates the termination from the driver
via a callback-extension of the existing event-monitoring.

This actually provides a door-opener for more advanced
communication and triggering between all the parties involved.
diff --git a/run/o2sim_parallel.cxx b/run/o2sim_parallel.cxx
@@ -19,6 +19,7 @@
 #include <SimConfig/SimConfig.h>
 #include <sys/wait.h>
 #include <vector>
+#include <functional>
 #include <thread>
 #include <csignal>
 #include "TStopwatch.h"
@@ -135,11 +136,14 @@ void sighandler(int signal)
   }
 }
 
-// monitores a certain incoming pipe and displays new information
-void launchThreadMonitoringEvents(int pipefd, std::string text)
+// monitores a certain incoming event pipes and displays new information
+// gives possibility to exec a callback at these events
+void launchThreadMonitoringEvents(
+  int pipefd, std::string text, std::vector<int>& eventcontainer,
+  std::function<void(std::vector<int> const&)> callback = [](std::vector<int> const&) {})
 {
   static std::vector<std::thread> threads;
-  auto lambda = [pipefd, text]() {
+  auto lambda = [pipefd, text, callback, &eventcontainer]() {
     int eventcounter;
     while (1) {
       ssize_t count = read(pipefd, &eventcounter, sizeof(eventcounter));
@@ -154,6 +158,8 @@ void launchThreadMonitoringEvents(int pipefd, std::string text)
         break;
       } else {
         LOG(INFO) << text.c_str() << eventcounter;
+        eventcontainer.push_back(eventcounter);
+        callback(eventcontainer);
       }
     };
   };
@@ -302,13 +308,14 @@ int main(int argc, char* argv[])
     o2::utils::ShmManager::Instance().disable();
   }
 
-  std::vector<int> childpids;
-
   int pipe_serverdriver_fd[2];
   if (pipe(pipe_serverdriver_fd) != 0) {
     perror("problem in creating pipe");
   }
 
+  // record distributed events in a container
+  std::vector<int> distributedEvents;
+
   // the server
   int pid = fork();
   if (pid == 0) {
@@ -356,7 +363,7 @@ int main(int argc, char* argv[])
     gChildProcesses.push_back(pid);
     close(pipe_serverdriver_fd[1]);
     std::cout << "Spawning particle server on PID " << pid << "; Redirect output to " << getServerLogName() << "\n";
-    launchThreadMonitoringEvents(pipe_serverdriver_fd[0], "DISTRIBUTING EVENT : ");
+    launchThreadMonitoringEvents(pipe_serverdriver_fd[0], "DISTRIBUTING EVENT : ", distributedEvents);
   }
 
   auto internalfork = getenv("ALICE_SIMFORKINTERNAL");
@@ -399,6 +406,9 @@ int main(int argc, char* argv[])
     perror("problem in creating pipe");
   }
 
+  // record finished events in a container
+  std::vector<int> finishedEvents;
+
   pid = fork();
   if (pid == 0) {
     int fd = open(getMergerLogName().c_str(), O_RDWR | O_CREAT, S_IRUSR | S_IWUSR);
@@ -418,7 +428,20 @@ int main(int argc, char* argv[])
     std::cout << "Spawning hit merger on PID " << pid << "; Redirect output to " << getMergerLogName() << "\n";
     gChildProcesses.push_back(pid);
     close(pipe_mergerdriver_fd[1]);
-    launchThreadMonitoringEvents(pipe_mergerdriver_fd[0], "EVENT FINISHED : ");
+
+    // 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 = [&conf](std::vector<int> const& v) {
+      if (conf.getNEvents() == v.size()) {
+        LOG(INFO) << "SIMULATION IS DONE. INITIATING SHUTDOWN.";
+        for (auto p : gChildProcesses) {
+          kill(p, SIGTERM);
+        }
+      }
+    };
+
+    launchThreadMonitoringEvents(pipe_mergerdriver_fd[0], "EVENT FINISHED : ", finishedEvents, finishCallback);
   }
 
   // wait on merger (which when exiting completes the workflow)
