DPL: reduce chances of hanging at the end of processing

ktf · ktf · commit a1bcdbf557d5 · 2019-03-25T12:51:16.000+01:00
Apparently sleep(1) does its own signal handling and that somehow
interferes with the signal handling done by FairMQ. Changing it
to std::this_thread::sleep_for makes it better behaved.
diff --git a/Framework/Core/src/CommonDataProcessors.cxx b/Framework/Core/src/CommonDataProcessors.cxx
@@ -21,6 +21,7 @@
 #include "Framework/OutputSpec.h"
 #include "Framework/Variant.h"
 
+#include <chrono>
 #include <exception>
 #include <fstream>
 #include <functional>
@@ -63,7 +64,7 @@ DataProcessorSpec CommonDataProcessors::getGlobalFileSink(std::vector<InputSpec>
           LOG(INFO) << "No dangling output to be dumped.";
           once = true;
         }
-        sleep(1);
+        std::this_thread::sleep_for(std::chrono::milliseconds(1000));
       });
     }
     auto output = std::make_shared<std::ofstream>(filename.c_str(), std::ios_base::binary);
diff --git a/Framework/Core/test/test_CCDBFetcher.cxx b/Framework/Core/test/test_CCDBFetcher.cxx
@@ -11,6 +11,8 @@
 #include "Framework/ServiceRegistry.h"
 #include "Framework/ControlService.h"
 
+#include <chrono>
+
 using namespace o2::framework;
 using namespace o2::header;
 
diff --git a/Framework/Core/test/test_CallbackService.cxx b/Framework/Core/test/test_CallbackService.cxx
@@ -7,7 +7,6 @@
 // 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.
-#include <iostream>
 #include <boost/algorithm/string.hpp>
 
 #include "Framework/InputSpec.h"
@@ -19,6 +18,9 @@
 #include "Framework/DebugGUI.h"
 #include "DebugGUI/imgui.h"
 
+#include <chrono>
+#include <iostream>
+
 using namespace o2::framework;
 using DataHeader = o2::header::DataHeader;
 
@@ -32,7 +34,7 @@ WorkflowSpec defineDataProcessing(ConfigContext const&)
       },
       AlgorithmSpec{
         [](ProcessingContext& ctx) {
-          sleep(1);
+          std::this_thread::sleep_for(std::chrono::milliseconds(1000));
           auto out = ctx.outputs().make<int>(OutputRef{ "test", 0 });
         } } },
     { "dest",
diff --git a/Framework/Core/test/test_CustomGUIGL.cxx b/Framework/Core/test/test_CustomGUIGL.cxx
@@ -7,9 +7,6 @@
 // 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.
-#include <iostream>
-#include <boost/algorithm/string.hpp>
-
 #include "Framework/InputSpec.h"
 #include "Framework/CallbackService.h"
 #include "Framework/ControlService.h"
@@ -21,6 +18,9 @@
 #include "DebugGUI/Sokol3DUtils.h"
 #include "DebugGUI/imgui.h"
 
+#include <chrono>
+#include <iostream>
+
 using namespace o2::framework;
 using DataHeader = o2::header::DataHeader;
 
@@ -34,7 +34,7 @@ WorkflowSpec defineDataProcessing(ConfigContext const&)
       },
       AlgorithmSpec{
         adaptStateless([](DataAllocator& outputs) {
-          sleep(1);
+          std::this_thread::sleep_for(std::chrono::milliseconds(1000));
           auto out = outputs.make<int>(OutputRef{ "test", 0 });
         }) } },
     { "dest",
diff --git a/Framework/Core/test/test_CustomGUISokol.cxx b/Framework/Core/test/test_CustomGUISokol.cxx
@@ -7,9 +7,6 @@
 // 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.
-#include <iostream>
-#include <boost/algorithm/string.hpp>
-
 #include "Framework/InputSpec.h"
 #include "Framework/CallbackService.h"
 #include "Framework/ControlService.h"
@@ -21,6 +18,11 @@
 #include "DebugGUI/Sokol3DUtils.h"
 #include "DebugGUI/imgui.h"
 
+#include <boost/algorithm/string.hpp>
+
+#include <chrono>
+#include <iostream>
+
 using namespace o2::framework;
 using DataHeader = o2::header::DataHeader;
 
@@ -34,7 +36,7 @@ WorkflowSpec defineDataProcessing(ConfigContext const&)
       },
       AlgorithmSpec{
         adaptStateless([](DataAllocator& outputs) {
-          sleep(1);
+          std::this_thread::sleep_for(std::chrono::milliseconds(1000));
           auto out = outputs.make<int>(OutputRef{ "test", 0 });
         }) } },
     { "dest",
diff --git a/Framework/Core/test/test_DanglingInputs.cxx b/Framework/Core/test/test_DanglingInputs.cxx
@@ -36,7 +36,7 @@ WorkflowSpec defineDataProcessing(ConfigContext const& specs)
         OutputSpec{ { "a2" }, "TST", "A2" } },
       AlgorithmSpec{
         [](ProcessingContext& ctx) {
-          sleep(rand() % 2 + 1);
+          std::this_thread::sleep_for(std::chrono::milliseconds((rand() % 2 + 1) * 1000));
           auto aData = ctx.outputs().make<int>(OutputRef{ "a1" }, 1);
           auto bData = ctx.outputs().make<int>(OutputRef{ "a2" }, 1);
         } } },
diff --git a/Framework/Core/test/test_DanglingOutputs.cxx b/Framework/Core/test/test_DanglingOutputs.cxx
@@ -10,10 +10,12 @@
 #include "Framework/ConfigParamSpec.h"
 #include "Framework/DeviceSpec.h"
 #include <InfoLogger/InfoLogger.hxx>
-#include <vector>
 #include "Framework/runDataProcessing.h"
 #include "Framework/ControlService.h"
 
+#include <chrono>
+#include <vector>
+
 using namespace o2::framework;
 
 AlgorithmSpec simplePipe(std::string const& what, int minDelay)
@@ -36,7 +38,7 @@ WorkflowSpec defineDataProcessing(ConfigContext const& specs)
         OutputSpec{ { "a2" }, "TST", "A2" } },
       AlgorithmSpec{
         [](ProcessingContext& ctx) {
-          sleep(rand() % 2 + 1);
+          std::this_thread::sleep_for(std::chrono::milliseconds((rand() % 2 + 1) * 1000));
           auto aData1 = ctx.outputs().make<int>(OutputRef{ "a1" }, 1);
           auto aData2 = ctx.outputs().make<int>(Output{ "TST", "A2" }, 1);
         } } },
diff --git a/Framework/Core/test/test_Forwarding.cxx b/Framework/Core/test/test_Forwarding.cxx
@@ -10,10 +10,12 @@
 #include "Framework/ConfigParamSpec.h"
 #include "Framework/DeviceSpec.h"
 #include <InfoLogger/InfoLogger.hxx>
-#include <vector>
 #include "Framework/runDataProcessing.h"
 #include "Framework/ControlService.h"
 
+#include <chrono>
+#include <vector>
+
 using namespace o2::framework;
 
 AlgorithmSpec simplePipe(std::string const& what, int minDelay)
@@ -35,7 +37,7 @@ WorkflowSpec defineDataProcessing(ConfigContext const& specs)
       { OutputSpec{ { "a1" }, "TST", "A1" } },
       AlgorithmSpec{
         [](ProcessingContext& ctx) {
-          sleep(rand() % 2);
+          std::this_thread::sleep_for(std::chrono::seconds(rand() % 2));
           auto aData = ctx.outputs().make<int>(OutputRef{ "a1" }, 1);
         } } },
     { "B",
diff --git a/Framework/Core/test/test_FrameworkDataFlowToDDS.cxx b/Framework/Core/test/test_FrameworkDataFlowToDDS.cxx
@@ -21,6 +21,7 @@
 #include "Framework/ProcessingContext.h"
 #include "Framework/WorkflowSpec.h"
 
+#include <chrono>
 #include <sstream>
 
 using namespace o2::framework;
@@ -39,7 +40,7 @@ WorkflowSpec defineDataProcessing()
              Outputs{ OutputSpec{ "TST", "A1" },
                       OutputSpec{ "TST", "A2" } },
              AlgorithmSpec{ [](ProcessingContext& ctx) {
-               sleep(1);
+               std::this_thread::sleep_for(std::chrono::seconds(1));
                auto aData = ctx.outputs().make<int>(Output{ "TST", "A1", 0 }, 1);
                auto bData = ctx.outputs().make<int>(Output{ "TST", "A2", 0 }, 1);
              } } },
diff --git a/Framework/Core/test/test_GenericSource.cxx b/Framework/Core/test/test_GenericSource.cxx
@@ -9,6 +9,8 @@
 // or submit itself to any jurisdiction.
 #include "Framework/runDataProcessing.h"
 
+#include <chrono>
+
 using namespace o2::framework;
 
 // This is how you can define your processing in a declarative way
@@ -24,7 +26,7 @@ WorkflowSpec defineDataProcessing(ConfigContext const&specs) {
       [](const InputRecord &inputs,
          ServiceRegistry& services,
          DataAllocator& allocator) {
-       sleep(1);
+       std::this_thread::sleep_for(std::chrono::seconds(1));
        auto aData = allocator.make<int>(Output{"TST", "A1", 0}, 1);
       }
     },
diff --git a/Framework/Core/test/test_ParallelProducer.cxx b/Framework/Core/test/test_ParallelProducer.cxx
@@ -7,7 +7,6 @@
 // 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.
-
 #include "Framework/ConfigContext.h"
 #include "Framework/ControlService.h"
 #include "Framework/DataProcessorSpec.h"
@@ -16,6 +15,7 @@
 
 #include "FairMQLogger.h"
 
+#include <chrono>
 #include <vector>
 
 using namespace o2::framework;
@@ -42,7 +42,7 @@ DataProcessorSpec templateProducer() {
       return [](ProcessingContext &ctx) {
           // Create a single output. 
           size_t index = ctx.services().get<ParallelContext>().index1D();
-          sleep(1);
+          std::this_thread::sleep_for(std::chrono::seconds(1));
           auto aData = ctx.outputs().make<int>(Output{ "TST", "A", index }, 1);
           ctx.services().get<ControlService>().readyToQuit(true);
         };
diff --git a/Framework/Core/test/test_SimpleTimer.cxx b/Framework/Core/test/test_SimpleTimer.cxx
@@ -12,6 +12,8 @@
 #include "Framework/runDataProcessing.h"
 #include "Framework/ControlService.h"
 
+#include <chrono>
+
 using namespace o2::framework;
 
 // This is how you can define your processing in a declarative way
@@ -25,7 +27,7 @@ std::vector<DataProcessorSpec> defineDataProcessing(ConfigContext const&)
       AlgorithmSpec{
         adaptStateless([](ControlService& control) {
           // This is invoked autonomously by the timer.
-          sleep(1);
+          std::this_thread::sleep_for(std::chrono::seconds(1));
           control.readyToQuit(true);
         }) } },
     DataProcessorSpec{
diff --git a/Framework/Core/test/test_SingleDataSource.cxx b/Framework/Core/test/test_SingleDataSource.cxx
@@ -11,6 +11,8 @@
 #include "Framework/ServiceRegistry.h"
 #include "Framework/ControlService.h"
 
+#include <chrono>
+
 using namespace o2::framework;
 
 // This is how you can define your processing in a declarative way
@@ -24,7 +26,7 @@ WorkflowSpec defineDataProcessing(ConfigContext const&) {
       },
       AlgorithmSpec{
         [](ProcessingContext &ctx) {
-         sleep(1);
+         std::this_thread::sleep_for(std::chrono::seconds(1));
          auto aData = ctx.outputs().make<int>(Output{ "TST", "A1", 0}, 1);
          ctx.services().get<ControlService>().readyToQuit(true);
         }
diff --git a/Framework/Core/test/test_TimePipeline.cxx b/Framework/Core/test/test_TimePipeline.cxx
@@ -7,16 +7,16 @@
 // 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.
-
-
-#include <iostream>
-#include <boost/algorithm/string.hpp>
-
 #include "Framework/InputSpec.h"
 #include "Framework/DataProcessorSpec.h"
 #include "Framework/ParallelContext.h"
 #include "Framework/runDataProcessing.h"
 
+#include <boost/algorithm/string.hpp>
+
+#include <chrono>
+#include <iostream>
+
 using namespace o2::framework;
 
 struct FakeCluster {
@@ -78,7 +78,7 @@ WorkflowSpec defineDataProcessing(ConfigContext const&)
 void someDataProducerAlgorithm(ProcessingContext& ctx)
 {
   size_t index = ctx.services().get<ParallelContext>().index1D();
-  sleep(1);
+  std::this_thread::sleep_for(std::chrono::seconds(1));
   // Creates a new message of size collectionChunkSize which
   // has "TPC" as data origin and "CLUSTERS" as data description.
   auto tpcClusters = ctx.outputs().make<FakeCluster>(Output{ "TPC", "CLUSTERS", index }, collectionChunkSize);
diff --git a/Framework/TestWorkflows/src/dataSamplingParallel.cxx b/Framework/TestWorkflows/src/dataSamplingParallel.cxx
@@ -27,7 +27,9 @@ void customize(std::vector<ChannelConfigurationPolicy>& policies)
 #include "Framework/ParallelContext.h"
 #include "Framework/runDataProcessing.h"
 
+#include <chrono>
 #include <iostream>
+
 #include <boost/algorithm/string.hpp>
 
 using namespace o2::framework;
@@ -138,7 +140,7 @@ WorkflowSpec defineDataProcessing(ConfigContext const&)
 void someDataProducerAlgorithm(ProcessingContext& ctx)
 {
   size_t index = ctx.services().get<ParallelContext>().index1D();
-  sleep(1);
+  std::this_thread::sleep_for(std::chrono::seconds(1));
   // Creates a new message of size collectionChunkSize which
   // has "TPC" as data origin and "CLUSTERS" as data description.
   auto tpcClusters = ctx.outputs().make<FakeCluster>(Output{ "TPC", "CLUSTERS", index }, collectionChunkSize);
diff --git a/Framework/TestWorkflows/src/dataSamplingPodAndRoot.cxx b/Framework/TestWorkflows/src/dataSamplingPodAndRoot.cxx
@@ -20,9 +20,7 @@ void customize(std::vector<ChannelConfigurationPolicy>& policies)
   DataSampling::CustomizeInfrastructure(policies);
 }
 
-#include <iostream>
 
-#include <boost/algorithm/string.hpp>
 #include "Framework/InputSpec.h"
 #include "Framework/DataProcessorSpec.h"
 #include "Framework/runDataProcessing.h"
@@ -32,6 +30,11 @@ void customize(std::vector<ChannelConfigurationPolicy>& policies)
 #include <TH1F.h>
 #include <TString.h>
 
+#include <boost/algorithm/string.hpp>
+
+#include <chrono>
+#include <iostream>
+
 struct FakeCluster {
   float x;
   float y;
@@ -120,7 +123,7 @@ WorkflowSpec defineDataProcessing(ConfigContext const&)
     },
     AlgorithmSpec{
       [](ProcessingContext& ctx) {
-        sleep(1);
+        std::this_thread::sleep_for(std::chrono::seconds(1));
         // Create an histogram
         auto& singleHisto = ctx.outputs().make<TH1F>(Output{ "TST", "HISTOS", 0 }, "h1", "test", 100, -10., 10.);
         auto& aString = ctx.outputs().make<TObjString>(Output{ "TST", "STRING", 0 }, "foo");
@@ -209,7 +212,7 @@ WorkflowSpec defineDataProcessing(ConfigContext const&)
 
 void someDataProducerAlgorithm(ProcessingContext& ctx)
 {
-  sleep(1);
+  std::this_thread::sleep_for(std::chrono::seconds(1));
   // Creates a new message of size collectionChunkSize which
   // has "TPC" as data origin and "CLUSTERS" as data description.
   auto tpcClusters = ctx.outputs().make<FakeCluster>(Output{ "TPC", "CLUSTERS", 0 }, collectionChunkSize);
diff --git a/Framework/TestWorkflows/src/dataSamplingTimePipeline.cxx b/Framework/TestWorkflows/src/dataSamplingTimePipeline.cxx
@@ -20,15 +20,17 @@ void customize(std::vector<ChannelConfigurationPolicy>& policies)
   DataSampling::CustomizeInfrastructure(policies);
 }
 
-#include <iostream>
-#include <boost/algorithm/string.hpp>
-
 #include "Framework/InputSpec.h"
 #include "Framework/DataProcessorSpec.h"
 #include "Framework/DataSampling.h"
 #include "Framework/ParallelContext.h"
 #include "Framework/runDataProcessing.h"
 
+#include <boost/algorithm/string.hpp>
+
+#include <chrono>
+#include <iostream>
+
 using namespace o2::framework;
 
 struct FakeCluster {
@@ -126,7 +128,7 @@ WorkflowSpec defineDataProcessing(ConfigContext const&)
 void someDataProducerAlgorithm(ProcessingContext& ctx)
 {
   size_t index = ctx.services().get<ParallelContext>().index1D();
-  sleep(1);
+  std::this_thread::sleep_for(std::chrono::seconds(1));
   // Creates a new message of size collectionChunkSize which
   // has "TPC" as data origin and "CLUSTERS" as data description.
   auto tpcClusters = ctx.outputs().make<FakeCluster>(Output{ "TPC", "CLUSTERS", index }, collectionChunkSize);
diff --git a/Framework/TestWorkflows/src/o2DataQueryWorkflow.cxx b/Framework/TestWorkflows/src/o2DataQueryWorkflow.cxx
diff --git a/Framework/TestWorkflows/src/o2DiamondWorkflow.cxx b/Framework/TestWorkflows/src/o2DiamondWorkflow.cxx
diff --git a/Framework/TestWorkflows/src/o2DummyWorkflow.cxx b/Framework/TestWorkflows/src/o2DummyWorkflow.cxx
diff --git a/Framework/TestWorkflows/src/o2ParallelWorkflow.cxx b/Framework/TestWorkflows/src/o2ParallelWorkflow.cxx
diff --git a/Framework/TestWorkflows/src/o2SyncReconstructionDummy.cxx b/Framework/TestWorkflows/src/o2SyncReconstructionDummy.cxx
diff --git a/Framework/TestWorkflows/src/o2_sim_tpc.cxx b/Framework/TestWorkflows/src/o2_sim_tpc.cxx
diff --git a/Framework/TestWorkflows/src/test_CompletionPolicies.cxx b/Framework/TestWorkflows/src/test_CompletionPolicies.cxx
diff --git a/Framework/TestWorkflows/src/test_o2RootMessageWorkflow.cxx b/Framework/TestWorkflows/src/test_o2RootMessageWorkflow.cxx
