DPL: do own management of shared memory

We do not spawn a shm-monitor because that will create zombies when launched
inside a docker container. Instead by default we cleanup the shared memory
session on exit. Default behavior can still be obtained by passing
--shm-monitor true.

Author: ktf

Framework/Core/src/DeviceSpecHelpers.cxx

@@ -801,6 +801,7 @@ void DeviceSpecHelpers::prepareArguments(bool defaultQuiet, bool defaultStopped,
     std::vector<std::string> tmpArgs = {argv[0],
                                         "--id", spec.id.c_str(),
                                         "--control", "static",
+                                        "--shm-monitor", "false",
                                         "--log-color", "false",
                                         "--color", "false"};
     if (defaultStopped) {
@@ -850,6 +851,7 @@ void DeviceSpecHelpers::prepareArguments(bool defaultQuiet, bool defaultStopped,
         realOdesc.add_options()("child-driver", bpo::value<std::string>());
         realOdesc.add_options()("rate", bpo::value<std::string>());
         realOdesc.add_options()("shm-segment-size", bpo::value<std::string>());
+        realOdesc.add_options()("shm-monitor", bpo::value<std::string>());
         realOdesc.add_options()("session", bpo::value<std::string>());
         filterArgsFct(expansions.we_wordc, expansions.we_wordv, realOdesc);
         wordfree(&expansions);
@@ -949,6 +951,7 @@ boost::program_options::options_description DeviceSpecHelpers::getForwardedDevic
     ("plugin-search-path,S", bpo::value<std::string>(), "FairMQ plugins search path")                           //
     ("control-port", bpo::value<std::string>(), "Utility port to be used by O2 Control")                        //
     ("rate", bpo::value<std::string>(), "rate for a data source device (Hz)")                                   //
+    ("shm-monitor", bpo::value<std::string>(), "whether to use the shared memory monitor")                      //
     ("shm-segment-size", bpo::value<std::string>(), "size of the shared memory segment in bytes")               //
     ("session", bpo::value<std::string>(), "unique label for the shared memory session")                        //
     ("monitoring-backend", bpo::value<std::string>(), "monitoring connection string")                           //

Framework/Core/src/runDataProcessing.cxx

@@ -273,6 +273,18 @@ static void handle_sigint(int)
   }
 }
 
+/// Helper to invoke shared memory cleanup
+void cleanupSHM(std::string const& uniqueWorkflowId)
+{
+  auto shmCleanup = fmt::format("fairmq-shmmonitor --cleanup -s dpl_{} 2>&1 >/dev/null", uniqueWorkflowId);
+  LOG(debug)
+    << "Cleaning up shm memory session with " << shmCleanup;
+  auto result = system(shmCleanup.c_str());
+  if (result != 0) {
+    LOG(error) << "Unable to cleanup shared memory, run " << shmCleanup << "by hand to fix";
+  }
+}
+
 static void handle_sigchld(int) { sigchld_requested = true; }
 
 void spawnRemoteDevice(std::string const& forwardedStdin,
@@ -848,6 +860,9 @@ int runStateMachine(DataProcessorSpecs const& workflow,
         }
         FD_ZERO(&(driverInfo.childFdset));
 
+        /// Cleanup the shared memory for the uniqueWorkflowId, in
+        /// case we are unlucky and an old one is already present.
+        cleanupSHM(driverInfo.uniqueWorkflowId);
         /// After INIT we go into RUNNING and eventually to SCHEDULE from
         /// there and back into running. This is because the general case
         /// would be that we start an application and then we wait for
@@ -1057,8 +1072,10 @@ int runStateMachine(DataProcessorSpecs const& workflow,
           driverInfo.states.push_back(DriverState::GUI);
         }
         break;
-      case DriverState::EXIT:
+      case DriverState::EXIT: {
+        cleanupSHM(driverInfo.uniqueWorkflowId);
         return calculateExitCode(infos);
+      }
       case DriverState::PERFORM_CALLBACKS:
         for (auto& callback : driverControl.callbacks) {
           callback(workflow, deviceSpecs, deviceExecutions, dataProcessorInfos);

Framework/Core/test/test_FrameworkDataFlowToDDS.cxx

@@ -97,16 +97,16 @@ BOOST_AUTO_TEST_CASE(TestDDS)
   dumpDeviceSpec2DDS(ss, devices, executions);
   BOOST_CHECK_EQUAL(ss.str(), R"EXPECTED(<topology id="o2-dataflow">
    <decltask id="A">
-       <exe reachable="true">foo --id A --control static --log-color false --color false --jobs 4 --session dpl_workflow-id --plugin-search-path $FAIRMQ_ROOT/lib --plugin dds</exe>
+       <exe reachable="true">foo --id A --control static --shm-monitor false --log-color false --color false --jobs 4 --session dpl_workflow-id --plugin-search-path $FAIRMQ_ROOT/lib --plugin dds</exe>
    </decltask>
    <decltask id="B">
-       <exe reachable="true">foo --id B --control static --log-color false --color false --jobs 4 --session dpl_workflow-id --plugin-search-path $FAIRMQ_ROOT/lib --plugin dds</exe>
+       <exe reachable="true">foo --id B --control static --shm-monitor false --log-color false --color false --jobs 4 --session dpl_workflow-id --plugin-search-path $FAIRMQ_ROOT/lib --plugin dds</exe>
    </decltask>
    <decltask id="C">
-       <exe reachable="true">foo --id C --control static --log-color false --color false --jobs 4 --session dpl_workflow-id --plugin-search-path $FAIRMQ_ROOT/lib --plugin dds</exe>
+       <exe reachable="true">foo --id C --control static --shm-monitor false --log-color false --color false --jobs 4 --session dpl_workflow-id --plugin-search-path $FAIRMQ_ROOT/lib --plugin dds</exe>
    </decltask>
    <decltask id="D">
-       <exe reachable="true">foo --id D --control static --log-color false --color false --jobs 4 --session dpl_workflow-id --plugin-search-path $FAIRMQ_ROOT/lib --plugin dds</exe>
+       <exe reachable="true">foo --id D --control static --shm-monitor false --log-color false --color false --jobs 4 --session dpl_workflow-id --plugin-search-path $FAIRMQ_ROOT/lib --plugin dds</exe>
    </decltask>
    <declcollection name="DPL">
        <tasks>