35 testInvariant ()
const
51 doManagerStep (Detail::ManagerData&
data)
const
54 using namespace msgEventLoop;
58 case Detail::ManagerStep::batchScriptVar:
60 data.batchSkipReleaseSetup =
true;
64 case Detail::ManagerStep::submitJob:
65 case Detail::ManagerStep::doResubmit:
69 const std::string dockerImage {
70 data.options.castString(Job::optDockerImage)};
71 const std::string dockerOptions {
72 data.options.castString(Job::optDockerOptions)};
74 =
data.options.castDouble (Job::optNumParallelProcs, 1);
75 if (numParallelProcs < 0)
77 ANA_MSG_ERROR (
"invalid number of parallel processes: " << numParallelProcs);
78 return StatusCode::FAILURE;
81 std::ostringstream basedirName;
82 basedirName <<
data.submitDir <<
"/tmp";
85 if (gSystem->MakeDirectory (basedirName.str().c_str()) != 0)
86 RCU_THROW_MSG (
"failed to create directory " + basedirName.str());
92 std::ostringstream dirName;
93 dirName << basedirName.str() <<
"/" <<
index;
94 if (gSystem->MakeDirectory (dirName.str().c_str()) != 0)
96 ANA_MSG_ERROR (
"failed to create directory " + dirName.str());
97 return StatusCode::FAILURE;
100 std::ostringstream
cmd;
101 cmd <<
"cd " << dirName.str() <<
" && ";
102 if (!dockerImage.empty())
109 return StatusCode::FAILURE;
111 return StatusCode::SUCCESS;
113 if (numParallelProcs == 1)
115 for (std::size_t
index :
data.batchJobIndices)
117 if (submitSingle (
index).isFailure())
118 return StatusCode::FAILURE;
122 if (numParallelProcs == 0)
123 numParallelProcs = std::thread::hardware_concurrency();
124 if (numParallelProcs >
int (
data.batchJobIndices.size()))
125 numParallelProcs =
data.batchJobIndices.size();
126 std::vector<std::thread>
threads;
128 auto indexIter =
data.batchJobIndices.begin();
132 threads.emplace_back ([&] () noexcept
134 std::unique_lock<std::mutex> lock (
mutex);
135 while (indexIter !=
data.batchJobIndices.end() && !abort)
137 auto myindex = *indexIter;
140 if (submitSingle (myindex).isFailure())
152 return StatusCode::FAILURE;
154 data.submitted =
true;
161 return ::StatusCode::SUCCESS;