Adding customizable hook to be executed on workflow termination (#2196)

matthiasrichter · ktf · commit a341176bb965 · 2019-07-15T12:15:58.000+02:00
This defines a function callback `OnWorkflowTerminationHook` with signature
`void (const char*)` and allows to set the hook using the `customize` mechanism
of DPL workflows. The callback hook is invoked just before processes terminate.
The hook is invoked for all childs and the main driver, the id string is passed
as parameter to customize the callback action. For the main driver, the nullptr
is passed.
diff --git a/Framework/Core/include/Framework/CustomWorkflowTerminationHook.h b/Framework/Core/include/Framework/CustomWorkflowTerminationHook.h
@@ -0,0 +1,46 @@
+// Copyright CERN and copyright holders of ALICE O2. This software is
+// distributed under the terms of the GNU General Public License v3 (GPL
+// Version 3), copied verbatim in the file "COPYING".
+//
+// See http://alice-o2.web.cern.ch/license for full licensing information.
+//
+// In applying this license CERN does not waive the privileges and immunities
+// granted to it by virtue of its status as an Intergovernmental Organization
+// or submit itself to any jurisdiction.
+#ifndef CUSTOMWORKFLOWTERMINATIONHOOK_H
+#define CUSTOMWORKFLOWTERMINATIONHOOK_H
+
+namespace o2
+{
+namespace framework
+{
+
+/// A callback definition for a hook to be invoked when processes terminate
+///
+/// The parameter is the nullptr if the process is the main driver, for all
+/// child processes, the id string is passed. This allows to customize the
+/// hook depending on the process.
+/// Note that the callback hook is invoked for every process, i.e. main driver
+/// and all childs.
+///
+/// \par Usage:
+/// The customize the hook, add a function with the following signature before
+/// including heder file runDataProcessing.h:
+///
+///     void customize(o2::framework::OnWorkflowTerminationHook& hook)
+///     {
+///       hook = [](const char* idstring){
+///         if (idstring == nullptr) {
+///           std::cout << "hook" << std::endl;
+///         } else {
+///           std::cout << "child process " << idstring << " terminating" << std::endl;
+///         }
+///       };
+///     }
+///     #include "Framework/runDataProcessing.h"
+using OnWorkflowTerminationHook = std::function<void(const char*)>;
+
+} // namespace framework
+} // namespace o2
+
+#endif
diff --git a/Framework/Core/include/Framework/runDataProcessing.h b/Framework/Core/include/Framework/runDataProcessing.h
@@ -17,6 +17,7 @@
 #include "Framework/WorkflowSpec.h"
 #include "Framework/ConfigContext.h"
 #include "Framework/BoostOptionsRetriever.h"
+#include "Framework/CustomWorkflowTerminationHook.h"
 
 #include <boost/program_options/options_description.hpp>
 #include <boost/program_options/variables_map.hpp>
@@ -67,6 +68,10 @@ o2::framework::WorkflowSpec defineDataProcessing(o2::framework::ConfigContext co
 void defaultConfiguration(std::vector<o2::framework::ChannelConfigurationPolicy>& channelPolicies) {}
 void defaultConfiguration(std::vector<o2::framework::ConfigParamSpec> &globalWorkflowOptions) {}
 void defaultConfiguration(std::vector<o2::framework::CompletionPolicy> &completionPolicies) {}
+void defaultConfiguration(o2::framework::OnWorkflowTerminationHook& hook)
+{
+  hook = [](const char*) {};
+}
 
 struct UserCustomizationsHelper {
   template <typename T>
@@ -124,6 +129,16 @@ int main(int argc, char** argv)
     LOG(ERROR) << "Unknown error while setting up workflow.";
   }
 
+  char* idstring = nullptr;
+  for (int argi = 0; argi < argc; argi++) {
+    if (strcmp(argv[argi], "--id") == 0 && argi + 1 < argc) {
+      idstring = argv[argi + 1];
+      break;
+    }
+  }
+  o2::framework::OnWorkflowTerminationHook onWorkflowTerminationHook;
+  UserCustomizationsHelper::userDefinedCustomization(onWorkflowTerminationHook, 0);
+  onWorkflowTerminationHook(idstring);
   LOG(INFO) << "Process " << getpid() << " is exiting.";
   return result;
 }
