Concurrently run beginStreams

cms-sw · Sep 10, 2021 · cfa296c · cfa296c
1 parent 2decae3
commit cfa296c
Showing 1 changed file with 19 additions and 3 deletions.
diff --git a/FWCore/Framework/src/EventProcessor.cc b/FWCore/Framework/src/EventProcessor.cc
@@ -684,9 +684,25 @@ namespace edm {
     for_all(subProcesses_, [](auto& subProcess) { subProcess.doBeginJob(); });
     actReg_->postBeginJobSignal_();
 
-    for (unsigned int i = 0; i < preallocations_.numberOfStreams(); ++i) {
-      schedule_->beginStream(i);
-      for_all(subProcesses_, [i](auto& subProcess) { subProcess.doBeginStream(i); });
+    FinalWaitingTask last;
+    tbb::task_group group;
+    using namespace edm::waiting_task::chain;
+    first([this](auto nextTask){
+      for (unsigned int i = 0; i < preallocations_.numberOfStreams(); ++i) {
+        first([i, this](auto nextTask) {
+          ServiceRegistry::Operate operate(serviceToken_);
+          schedule_->beginStream(i);
+        })
+        | ifThen(not subProcesses_.empty(), [this,i](auto nextTask){
+          ServiceRegistry::Operate operate(serviceToken_);
+          for_all(subProcesses_, [i](auto& subProcess) { subProcess.doBeginStream(i); });
+        })
+        |lastTask(nextTask);
+      }
+    }) | runLast(WaitingTaskHolder(group, &last));
+    group.wait();
+    if(last.exceptionPtr()) {
+      std::rethrow_exception(*last.exceptionPtr());
     }
   }