Merge branch 'el_parallel' into 'master'

allow running multiple parallel jobs in LocalDriver

See merge request atlas/athena!48132

PhysicsAnalysis/D3PDTools/EventLoop/EventLoop/Job.h

@@ -390,6 +390,16 @@ namespace EL
   public:
     static const std::string optLocalNoUnsetup;
 
+    /// \brief the option to specify the number of parallel jobs in
+    /// \ref LocalDriver (0 = number of hardware cores) (default=1)
+    ///
+    /// As an intermediate between running a single job locally and
+    /// running in batch, this allows to run multiple processes in
+    /// parallel locally.  It is still recommended to just run in
+    /// batch instead, but sometimes this is more practical.
+  public:
+    static const std::string optNumParallelProcs;
+
 
     /// \brief the option to do processing in a background process in PROOF
   public:

PhysicsAnalysis/D3PDTools/EventLoop/Root/Job.cxx

@@ -61,6 +61,7 @@ namespace EL
   const std::string Job::optDisableMetrics = "nc_disable_metrics";
   const std::string Job::optResetShell = "nc_reset_shell";
   const std::string Job::optLocalNoUnsetup = "nc_local_no_unsetup";
+  const std::string Job::optNumParallelProcs = "nc_num_parallel_processes";
   const std::string Job::optBackgroundProcess = "nc_background_process";
   const std::string Job::optOutputSampleName = "nc_outputSampleName";
   const std::string Job::optGridDestSE = "nc_destSE";

PhysicsAnalysis/D3PDTools/EventLoop/Root/LocalDriver.cxx

@@ -20,6 +20,8 @@
 #include <RootCoreUtils/Assert.h>
 #include <RootCoreUtils/ShellExec.h>
 #include <RootCoreUtils/ThrowMsg.h>
+#include <mutex>
+#include <thread>
 
 //
 // method implementations
@@ -68,6 +70,13 @@ namespace EL
           data.options.castString(Job::optDockerImage)};
         const std::string dockerOptions {
           data.options.castString(Job::optDockerOptions)};
+        int numParallelProcs
+          = data.options.castDouble (Job::optNumParallelProcs, 1);
+        if (numParallelProcs < 0)
+        {
+          ANA_MSG_ERROR ("invalid number of parallel processes: " << numParallelProcs);
+          return StatusCode::FAILURE;
+        }
 
         std::ostringstream basedirName;
         basedirName << data.submitDir << "/tmp";
@@ -76,19 +85,71 @@ namespace EL
           if (gSystem->MakeDirectory (basedirName.str().c_str()) != 0)
             RCU_THROW_MSG ("failed to create directory " + basedirName.str());
         }
-        for (std::size_t index : data.batchJobIndices)
+        auto submitSingle = [&, this] (std::size_t index) noexcept -> StatusCode
+        {
+          try
+          {
+            std::ostringstream dirName;
+            dirName << basedirName.str() << "/" << index;
+            if (gSystem->MakeDirectory (dirName.str().c_str()) != 0)
+            {
+              ANA_MSG_ERROR ("failed to create directory " + dirName.str());
+              return StatusCode::FAILURE;
+            }
+
+            std::ostringstream cmd;
+            cmd << "cd " << dirName.str() << " && ";
+            if (!dockerImage.empty())
+              cmd << "docker run --rm -v " << RCU::Shell::quote (data.submitDir) << ":" << RCU::Shell::quote (data.submitDir) << " " << dockerOptions << " " << dockerImage << " ";
+            cmd << RCU::Shell::quote (data.submitDir) << "/submit/run " << index;
+            RCU::Shell::exec (cmd.str());
+          } catch (std::exception& e)
+          {
+            ANA_MSG_ERROR ("exception in job " << index << ": " << e.what());
+            return StatusCode::FAILURE;
+          }
+          return StatusCode::SUCCESS;
+        };
+        if (numParallelProcs == 1)
+        {
+          for (std::size_t index : data.batchJobIndices)
+          {
+            if (submitSingle (index).isFailure())
+              return StatusCode::FAILURE;
+          }
+        } else
         {
-          std::ostringstream dirName;
-          dirName << basedirName.str() << "/" << index;
-          if (gSystem->MakeDirectory (dirName.str().c_str()) != 0)
-            RCU_THROW_MSG ("failed to create directory " + dirName.str());
-
-          std::ostringstream cmd;
-          cmd << "cd " << dirName.str() << " && ";
-          if (!dockerImage.empty())
-            cmd << "docker run --rm -v " << RCU::Shell::quote (data.submitDir) << ":" << RCU::Shell::quote (data.submitDir) << " " << dockerOptions << " " << dockerImage << " ";
-          cmd << RCU::Shell::quote (data.submitDir) << "/submit/run " << index;
-          RCU::Shell::exec (cmd.str());
+          if (numParallelProcs == 0)
+            numParallelProcs = std::thread::hardware_concurrency();
+          if (numParallelProcs > int (data.batchJobIndices.size()))
+            numParallelProcs = data.batchJobIndices.size();
+          std::vector<std::thread> threads;
+          std::mutex mutex;
+          auto indexIter = data.batchJobIndices.begin();
+          bool abort = false;
+          while (threads.size() < unsigned (numParallelProcs))
+          {
+            threads.emplace_back ([&,this] () noexcept
+            {
+              std::unique_lock<std::mutex> lock (mutex);
+              while (indexIter != data.batchJobIndices.end() && !abort)
+              {
+                auto myindex = *indexIter;
+                ++ indexIter;
+                lock.unlock();
+                if (submitSingle (myindex).isFailure())
+                {
+                  abort = true;
+                  return;
+                }
+                lock.lock ();
+              }
+            });
+          }
+          for (auto& thread : threads)
+            thread.join();
+          if (abort)
+            return StatusCode::FAILURE;
         }
         data.submitted = true;
       }

PhysicsAnalysis/D3PDTools/EventLoopTest/test/ut_driver_local_parallel.cxx

+/*
+  Copyright (C) 2002-2021 CERN for the benefit of the ATLAS collaboration
+*/
+
+
+//
+// includes
+//
+
+#include <EventLoop/Global.h>
+
+#include <EventLoop/LocalDriver.h>
+#include <EventLoop/Job.h>
+#include <EventLoopTest/UnitTest.h>
+
+//
+// main program
+//
+
+using namespace EL;
+
+int main ()
+{
+  LocalDriver driver;
+  driver.options()->setDouble (Job::optNumParallelProcs, 0u);
+  UnitTest ut ("local");
+  ut.cleanup = false;
+  return ut.run (driver);
+}