ATLAS Offline Software
Loading...
Searching...
No Matches
PrunDriver.cxx
Go to the documentation of this file.
1/*
2 Copyright (C) 2002-2024 CERN for the benefit of the ATLAS collaboration
3*/
4
7
8
9
11#include <EventLoop/Algorithm.h>
14#include <EventLoop/Job.h>
19#include <RootCoreUtils/hadd.h>
26
27
28#include <TList.h>
29#include <TPython.h>
30#include <TROOT.h>
31#include <TFile.h>
32#include <TSystem.h>
33
34#include <algorithm>
35#include <cstdlib>
36#include <iostream>
37#include <sstream>
38#include <string>
39#include <vector>
40#include <stdexcept>
41
42#include <ranges>
43
44#include "pool.h"
45#include <mutex>
46
48
49namespace {
50 namespace JobState {
51 static const unsigned int NSTATES = 6;
52 enum Enum { INIT=0, RUN=1, DOWNLOAD=2, MERGE=3, FINISHED=4, FAILED=5 };
53 static const char* name[NSTATES] =
54 { "INIT", "RUNNING", "DOWNLOAD", "MERGE", "FINISHED", "FAILED" };
55 Enum parse(const std::string& what)
56 {
57 for (unsigned int i = 0; i != NSTATES; ++i) {
58 if (what == name[i]) { return static_cast<Enum>(i); }
59 }
60 RCU_ASSERT0("Failed to parse job state string");
61 throw std::runtime_error("PrunDriver.cxx: Failed to parse job state string"); //compiler dummy
62 }
63 }
64
65 // When changing the values in the enum make sure
66 // corresponding values in `data/ELG_jediState.py` script
67 // are changed accordingly
68 namespace Status {
69 //static const int NSTATES = 3;
70 enum Enum { DONE=0, PENDING=1, FAIL=2 };
71 }
72
73 struct TransitionRule {
74 JobState::Enum fromState;
75 Status::Enum status;
76 JobState::Enum toState;
77 TransitionRule(JobState::Enum fromState,
78 Status::Enum status,
79 JobState::Enum toState)
80 : fromState(fromState)
81 , status(status)
82 , toState(toState)
83 {
84 }
85 };
86
87 struct TmpCd {
88 const std::string origDir;
89 TmpCd(const std::string & dir)
90 : origDir(gSystem->pwd())
91 {
92 gSystem->cd(dir.c_str());
93 }
94 ~TmpCd()
95 {
96 gSystem->cd(origDir.c_str());
97 }
98 };
99}
100
101static JobState::Enum sampleState(SH::Sample* sample)
102{
103 RCU_REQUIRE(sample);
104 static const std::string defaultState = JobState::name[JobState::INIT];
105 std::string label = sample->meta()->castString ("nc_ELG_state", defaultState, SH::MetaObject::CAST_NOCAST_DEFAULT);
106 return JobState::parse(label);
107}
108
109static JobState::Enum nextState(JobState::Enum state, Status::Enum status)
110{
111 RCU_REQUIRE(state != JobState::FINISHED);
112 RCU_REQUIRE(state != JobState::FAILED);
113 static const TransitionRule TABLE[] =
114 {
115 TransitionRule(JobState::INIT, Status::DONE, JobState::RUN),
116 TransitionRule(JobState::INIT, Status::PENDING, JobState::INIT),
117 TransitionRule(JobState::INIT, Status::FAIL, JobState::FAILED),
118 TransitionRule(JobState::RUN, Status::DONE, JobState::DOWNLOAD),
119 TransitionRule(JobState::RUN, Status::PENDING, JobState::RUN),
120 TransitionRule(JobState::RUN, Status::FAIL, JobState::FAILED),
121 TransitionRule(JobState::DOWNLOAD, Status::DONE, JobState::MERGE),
122 TransitionRule(JobState::DOWNLOAD, Status::PENDING, JobState::DOWNLOAD),
123 TransitionRule(JobState::DOWNLOAD, Status::FAIL, JobState::FAILED),
124 TransitionRule(JobState::MERGE, Status::DONE, JobState::FINISHED),
125 TransitionRule(JobState::MERGE, Status::PENDING, JobState::MERGE),
126 TransitionRule(JobState::MERGE, Status::FAIL, JobState::DOWNLOAD)
127 };
128 static const unsigned int TABLE_SIZE = sizeof(TABLE) / sizeof(TABLE[0]);
129 for (unsigned int i = 0; i != TABLE_SIZE; ++i) {
130 if (TABLE[i].fromState == state && TABLE[i].status == status) {
131 return TABLE[i].toState;
132 }
133 }
134 RCU_ASSERT0("Missing state transition rule");
135 throw std::logic_error("PrunDriver.cxx: Missing state transition rule");
136}
137
139{
141 o.setString("nc_nGBPerJob", "MAX");
142 o.setString("nc_mergeOutput", "true");
143 o.setString("nc_cmtConfig", gSystem->ExpandPathName("$AnalysisBase_PLATFORM"));
144 o.setString("nc_useAthenaPackages", "true");
145 const std::string mergestr = "elg_merge jobdef.root %OUT %IN";
146 o.setString("nc_mergeScript", mergestr);
147 return o;
148}
149
150static bool downloadContainer(const std::string& name,
151 const std::string& location)
152{
153 RCU_ASSERT(not name.empty());
154 RCU_ASSERT(name[name.size()-1] == '/');
155 RCU_ASSERT(not location.empty());
156
157 try {
158 gSystem->Exec(Form("mkdir -p %s", location.c_str()));
159
160 std::vector<std::string> datasets;
161 for (auto& entry : SH::rucioListDids (name))
162 {
163 if (entry.type == "CONTAINER" || entry.type == "DIDType.CONTAINER")
164 datasets.push_back (entry.name);
165 }
166
167 auto downloadResult = SH::rucioDownloadList (location, datasets);
168 for (const auto& result : downloadResult)
169 {
170 if (result.notDownloaded != 0)
171 return false;
172 }
173 } catch (...) {
174 return false;
175 }
176 return true;
177}
178
179static Status::Enum submit(SH::Sample* const sample, const bool isFirstSample)
180{
181 RCU_REQUIRE(sample);
182 using namespace EL::msgEventLoop;
183
184 ANA_MSG_INFO( "Submitting " << sample->name() << "..." );
185
186 static bool loaded = false;
187 if (not loaded) {
188 // TString path = "$ROOTCOREBIN/python/EventLoopGrid/ELG_prun.py";
189 // gSystem->ExpandPathName(path);
190 // TPython::LoadMacro(path.Data());
191 std::string path = PathResolverFindCalibFile("EventLoopGrid/ELG_prun.py");
192 TPython::LoadMacro(path.c_str());
193 loaded = true;
194 }
195
196 TPython::Bind(dynamic_cast<TObject*>(sample), "ELG_SAMPLE");
197#if ROOT_VERSION_CODE >= ROOT_VERSION(6,33,01)
198 std::any result;
199 TPython::Exec("_anyresult = ROOT.std.make_any['int'](ELG_prun(ELG_SAMPLE))", &result);
200 int ret = std::any_cast<int>(result);
201#else
202 int ret = TPython::Eval("ELG_prun(ELG_SAMPLE)");
203#endif
204 TPython::Bind(0, "ELG_SAMPLE");
205
206 // Tarball is created for the first sample to be submitted
207 // then the tarball is simply reused for the other samples
208 // If the returned value is 1 it implies the tarball creation failed
209 // See EventLoopGrid/data/ELG_prun.py script
210 // Abort any further processing as the tarball was not succesfully created
211 if (isFirstSample && ret == 1){
212 ANA_MSG_ERROR("Failed to create tarball");
213 throw std::runtime_error("PrunDriver.cxx: aborting due to tarball creation issue");
214 }
215
216 if (ret < 100) {
217 sample->meta()->setString("nc_ELG_state_details",
218 "problem submitting");
219 return Status::FAIL;
220 }
221
222 sample->meta()->setDouble("nc_jediTaskID", ret);
223
224 // Let's also tell people about their task ID
225 ANA_MSG_INFO( "Task submitted; jediTaskID=" << ret );
226
227 return Status::DONE;
228}
229
230static Status::Enum checkPandaTask(SH::Sample* const sample)
231{
232 RCU_REQUIRE(sample);
233 RCU_REQUIRE(static_cast<int>(sample->meta()->castDouble("nc_jediTaskID",0, SH::MetaObject::CAST_NOCAST_DEFAULT)) > 100);
234
235 static bool loaded = false;
236 if (not loaded) {
237 std::string path = PathResolverFindCalibFile("EventLoopGrid/ELG_jediState.py");
238 TPython::LoadMacro(path.c_str());
239 loaded = true;
240 }
241
242 TPython::Bind(dynamic_cast<TObject*>(sample), "ELG_SAMPLE");
243#if ROOT_VERSION_CODE >= ROOT_VERSION(6,33,01)
244 std::any result;
245 TPython::Exec("_anyresult = ROOT.std.make_any['int'](ELG_jediState(ELG_SAMPLE))", &result);
246 int ret = std::any_cast<int>(result);
247#else
248 int ret = TPython::Eval("ELG_jediState(ELG_SAMPLE)");
249#endif
250 TPython::Bind(0, "ELG_SAMPLE");
251
252 if (ret == Status::DONE) return Status::DONE;
253 if (ret == Status::FAIL) return Status::FAIL;
254
255 // Value 90 corresponds to `running` state of the job
256 if (ret != 90) { sample->meta()->setString("nc_ELG_state_details", "task status other than done/finished/failed/running"); }
257 // Value 99 is returned if there is error in the script (import, missing ID)
258 if (ret == 99) {
259 sample->meta()->setString("nc_ELG_state_details",
260 "problem checking jedi task status");
261 }
262
263 return Status::PENDING;
264}
265
266static Status::Enum download(SH::Sample* const sample)
267{
268 RCU_REQUIRE(sample);
269
270 {
271 static std::mutex mutex;
272 std::lock_guard<std::mutex> lock(mutex);
273 std::cout << "Downloading output from: "
274 << sample->name() << "..." << std::endl;
275 }
276
277 std::string container = sample->meta()->castString("nc_outDS", "", SH::MetaObject::CAST_NOCAST_DEFAULT);
278 RCU_ASSERT(not container.empty());
279 if (container[container.size()-1] == '/') {
280 container.resize(container.size() - 1);
281 }
282 container += "_hist/";
283
284 bool downloadOk = downloadContainer(container, "elg/download/" + container);
285
286 if (not downloadOk) {
287 std::cerr << "Failed to download one or more files" << std::endl;
288 sample->meta()->setString("nc_ELG_state_details",
289 "error, check log for details");
290 return Status::PENDING;
291 }
292
293 return Status::DONE;
294}
295
296static Status::Enum merge(SH::Sample* const sample)
297{
298 RCU_REQUIRE(sample);
299
300 std::string container = sample->meta()->castString("nc_outDS", "", SH::MetaObject::CAST_NOCAST_DEFAULT);
301 RCU_ASSERT(not container.empty());
302 if (container[container.size()-1] == '/') {
303 container.resize(container.size() - 1);
304 }
305 container += "_hist/";
306 const std::string dir = "elg/download/" + container;
307
308 const std::string fileName = "hist-output.root";
309
310 const std::string target = Form("hist-%s.root", sample->name().c_str());
311
312 const std::string findCmd(Form("find %s -name \"*.%s*\" | tr '\n' ' '",
313 dir.c_str(), fileName.c_str()));
314 std::istringstream input(gSystem->GetFromPipe(findCmd.c_str()).Data());
315 std::vector<std::string> files((std::istream_iterator<std::string>(input)),
316 std::istream_iterator<std::string>());
317
318 std::sort(files.begin(), files.end());
319 RCU_ASSERT(std::unique(files.begin(), files.end()) == files.end());
320
321 if (not files.size()) {
322 std::cerr << "Found no input files for merging! "
323 << "Requeueing sample for download..." << std::endl;
324 sample->meta()->setString("nc_ELG_state_details", "retry, files were lost");
325 return Status::FAIL;
326 }
327
328 try {
329 RCU::hadd(target.c_str(), files);
330 } catch (...) {
331 sample->meta()->setString("nc_ELG_state_details",
332 "error, check log for details");
333 gSystem->Exec(Form("rm -f %s", target.c_str()));
334 return Status::PENDING;
335 }
336
337 for (size_t i = 0; i != files.size(); ++i) {
338 gSystem->Exec(Form("rm %s", files[i].c_str()));
339 }
340 gSystem->Exec(Form("rmdir %s/*", dir.c_str()));
341 gSystem->Exec(Form("rmdir %s", dir.c_str()));
342
343 return Status::DONE;
344}
345
346static void processTask(SH::Sample* const sample, const bool isFirstSample)
347{
348 RCU_REQUIRE(sample);
349
350 JobState::Enum state = sampleState(sample);
351
352 sample->meta()->setString("nc_ELG_state_details", "");
353
354 Status::Enum status = Status::PENDING;
355 switch (state) {
356 case JobState::INIT:
357 status = submit(sample, isFirstSample);
358 break;
359 case JobState::RUN:
360 status = checkPandaTask(sample);
361 break;
362 case JobState::DOWNLOAD:
363 status = download(sample);
364 break;
365 case JobState::MERGE:
366 status = merge(sample);
367 break;
368 case JobState::FINISHED:
369 case JobState::FAILED:
370 break;
371 }
372
373 state = nextState(state, status);
374 sample->meta()->setString("nc_ELG_state", JobState::name[state]);
375}
376
377static void processAllInState(const SH::SampleHandler& sh, JobState::Enum state,
378 const size_t nThreads)
379{
380 RCU_REQUIRE(sh.size());
381
382 WorkList workList;
383
384 bool isFirstSample = true;
385 for (SH::SampleHandler::iterator s = sh.begin(); s != sh.end(); ++s) {
386 if (sampleState(*s) == state) {
387 workList.push_back([s, isFirstSample]()->void{ processTask(*s, isFirstSample); });
388 // Change boolean to false as already processed one sample
389 isFirstSample = false;
390 }
391 }
392 process(workList, nThreads);
393}
394
395static std::string formatOutputName(const SH::MetaObject& sampleMeta,
396 const std::string & pattern)
397{
398 const std::string sampleName = sampleMeta.castString("sample_name");
399 RCU_REQUIRE(not pattern.empty());
400 using namespace EL::msgEventLoop;
401
402 static const std::string nickname =
403 gSystem->GetFromPipe(Form("python -c \"%s\" 2>/dev/null",
404 "from pandatools import PsubUtils;"
405 "print(PsubUtils.getNickname());")).Data();
406
407 TString out = pattern.c_str();
408
409 // Handle case of no proxy; will create a proxy later in the submission
410 if (nickname.length()>20){
411 ANA_MSG_WARNING( "No proxy available - cannot use nickname yet. Will try a late replacement.");
412 } else {
413 out.ReplaceAll("%nickname%", nickname);
414 }
415
416 out.ReplaceAll("%in:name%", sampleName);
417
418 std::stringstream ss(sampleName);
419 std::string item;
420 int field = 0;
421 while(std::getline(ss, item, '.')) {
422 std::stringstream sskey;
423 sskey << "%in:name[" << ++field << "]%";
424 out.ReplaceAll(sskey.str(), item);
425 }
426 while (out.Index("%in:") != -1) {
427 int i1 = out.Index("%in:");
428 int i2 = out.Index("%", i1+1);
429 TString metaName = out(i1+4, i2-i1-4);
430 out.ReplaceAll("%in:"+metaName+"%",
431 sampleMeta.castString(std::string(metaName.Data())));
432 }
433 out.ReplaceAll("/", "");
434 return out.Data();
435}
436
437std::string outputFileNames(const EL::Job& job)
438{
439 TList outputs;
440 for (EL::Job::outputIter out = job.outputBegin(),
441 end = job.outputEnd(); out != end; ++out) {
442 outputs.Add(out->Clone());
443 }
444 std::string out = "hist:hist-output.root";
445 TIter itr(&outputs);
446 TObject *obj = 0;
447 while ((obj = itr())) {
448 EL::OutputStream *os = dynamic_cast<EL::OutputStream*>(obj);
449 const std::string name = os->label() + ".root";
450 const std::string ds =
451 os->options()->castString(EL::OutputStream::optContainerSuffix);
452 out += "," + (ds.empty() ? name : ds + ":" + name);
453 }
454 return out;
455}
456
457// Save algortihms and lists of inputs and outputs to a root file
458static void saveJobDef(const std::string& fileName,
459 const EL::Job& job,
460 const SH::SampleHandler sh)
461{
462 TFile file(fileName.c_str(), "RECREATE");
463 TList outputs;
464 for (EL::Job::outputIter o = job.outputBegin(); o !=job.outputEnd(); ++o)
465 outputs.Add(o->Clone());
466 file.WriteTObject(&job.jobConfig(), "jobConfig", "SingleKey");
467 file.WriteTObject(&outputs, "outputs", "SingleKey");
468 bool haveDefault = false;
469 for (SH::SampleHandler::iterator s = sh.begin(); s != sh.end(); ++s) {
470 const SH::MetaObject& meta = *((*s)->meta());
471 file.WriteObject(&meta, meta.castString("sample_name").c_str());
472 if (!haveDefault)
473 {
474 file.WriteObject (&meta, "defaultMetaObject");
475 haveDefault = true;
476 }
477 }
478}
479
480// Create a sample handler with grid locations of outputs with given label
482 const std::string& outputLabel)
483{
485 const std::string outputFile = "*" + outputLabel + ".root*";
486 const std::string outDSSuffix = '_' + outputLabel + ".root/";
487 for (SH::SampleHandler::iterator s = in.begin(); s != in.end(); ++s) {
488 auto outSample = std::make_unique<SH::SampleGrid>((*s)->name());
489 const std::string outputDS = (*s)->meta()->castString("nc_outDS", "", SH::MetaObject::CAST_NOCAST_DEFAULT) + outDSSuffix;
490 outSample->meta()->setString("nc_grid", outputDS);
491 outSample->meta()->setString("nc_grid_filter", outputFile);
492 out.add(std::move(outSample));
493 }
494 out.fetch(in);
495 return out;
496}
497
499{
500 RCU_INVARIANT(this != 0);
501}
502
507
508::StatusCode EL::PrunDriver ::
509doManagerStep (Detail::ManagerData& data) const
510{
511 using namespace msgEventLoop;
513 switch (data.step)
514 {
516 {
517 const std::string jobELGDir = data.submitDir + "/elg";
518 const std::string runShFile = jobELGDir + "/runjob.sh";
519 //const std::string runShOrig = "$ROOTCOREBIN/data/EventLoopGrid/runjob.sh";
520 const std::string mergeShFile = jobELGDir + "/elg_merge";
521 //const std::string mergeShOrig =
522 // "$ROOTCOREBIN/user_scripts/EventLoopGrid/elg_merge";
523 const std::string runShOrig = PathResolverFindCalibFile("EventLoopGrid/runjob.sh");
524 const std::string mergeShOrig = PathResolverFindCalibFile("EventLoopGrid/elg_merge");
525
526 const std::string jobDefFile = jobELGDir + "/jobdef.root";
527 gSystem->Exec(Form("mkdir -p %s", jobELGDir.c_str()));
528 gSystem->Exec(Form("cp %s %s", runShOrig.c_str(), runShFile.c_str()));
529 gSystem->Exec(Form("chmod +x %s", runShFile.c_str()));
530 gSystem->Exec(Form("cp %s %s", mergeShOrig.c_str(), mergeShFile.c_str()));
531 gSystem->Exec(Form("chmod +x %s", mergeShFile.c_str()));
532
533 // create symbolic links for additionnal files/directories if any to ship to the grid
534 std::string listToShipToGrid = data.options.castString(EL::Job::optGridPrunShipAdditionalFilesOrDirs, "");
535 // parse the list of comma separated files and/or directories to ship to the grid
536 if (listToShipToGrid.size()){
538 "Creating symbolic links for additional files or directories to be sent to grid.\n"
539 "For root or heavy files you should also add their name (not the full path) to EL::Job::optUserFiles.\n"
540 "Otherwise prun ignores those files."
541 );
542
543 std::vector<std::string> vect_filesOrDirToShip;
544 for (auto&& part : std::views::split(listToShipToGrid, ',')) vect_filesOrDirToShip.emplace_back(part.begin(), part.end());
545 // Create symbolic links of files or directories to the submission directory
546 for (const std::string & fileOrDirToShip: vect_filesOrDirToShip){
547 ANA_MSG_INFO (("Creating symbolic link for: " +fileOrDirToShip).c_str());
548 RCU::Shell::exec("ln -sf " + fileOrDirToShip + " " + jobELGDir);
549 }
550 ANA_MSG_INFO ("Finished creation of symbolic links");
551 }
552
553 const SH::SampleHandler& sh = data.job->sampleHandler();
554
555 for (SH::SampleHandler::iterator s = sh.begin(); s != sh.end(); ++s) {
556 SH::MetaObject& meta = *(*s)->meta();
557 meta.fetchDefaults(data.options);
558 meta.fetchDefaults(defaultOpts());
559 meta.setString("nc_outputs", outputFileNames(*data.job));
560 std::string outputSampleName = meta.castString("nc_outputSampleName");
561 if (outputSampleName.empty()) {
562 outputSampleName = "user.%nickname%.%in:name%";
563 }
564 meta.setString("nc_outDS", formatOutputName(meta, outputSampleName));
565 meta.setString("nc_inDS", meta.castString("nc_grid", (*s)->name()));
566 meta.setString("nc_writeInputToTxt", "IN:input.txt");
567 meta.setString("nc_match", meta.castString("nc_grid_filter"));
568 const std::string execstr = "runjob.sh " + (*s)->name();
569 meta.setString("nc_exec", execstr);
570 meta.setString("nc_framework", "EventLoopGrid");
571 }
572
573 saveJobDef(jobDefFile, *data.job, sh);
574
575 for (EL::Job::outputIter out = data.job->outputBegin();
576 out != data.job->outputEnd(); ++out) {
577 SH::SampleHandler shOut = outputSH(sh, out->label());
578 shOut.save(data.submitDir + "/output-" + out->label());
579 }
580 SH::SampleHandler shHist = outputSH(sh, "hist-output");
581 shHist.save(data.submitDir + "/output-hist");
582
583 TmpCd keepDir(jobELGDir);
584
585 processAllInState(sh, JobState::INIT, 0);
586
587 sh.save(data.submitDir + "/input");
588 data.submitted = true;
589 }
590 break;
591
593 {
594 ANA_CHECK (doRetrieve (data));
595 }
596 break;
597
598 default:
599 (void) true; // safe to do nothing
600 }
601 return ::StatusCode::SUCCESS;
602}
603
605{
606 RCU_READ_INVARIANT(this);
607 RCU_REQUIRE(not data.submitDir.empty());
608
609 TmpCd tmpDir(data.submitDir);
610
612 sh.load("input");
613 RCU_ASSERT(sh.size());
614
615 const size_t nRunThreads = options()->castDouble("nc_run_threads", 0);
616 const size_t nDlThreads = options()->castDouble("nc_download_threads", 0);
617 processAllInState(sh, JobState::INIT, 0);
618 processAllInState(sh, JobState::RUN, nRunThreads);
619 processAllInState(sh, JobState::DOWNLOAD, nDlThreads);
620 processAllInState(sh, JobState::MERGE, 0);
621
622 sh.save("input");
623
624 std::cout << std::endl;
625
626 bool allDone = true;
627 for (SH::SampleHandler::iterator s = sh.begin(); s != sh.end(); ++s) {
628 JobState::Enum state = sampleState(*s);
629 std::string details = (*s)->meta()->castString("nc_ELG_state_details", "", SH::MetaObject::CAST_NOCAST_DEFAULT);
630 if (not details.empty()) { details = '(' + details + ')'; }
631
632 std::cout << (*s)->name() << "\t";
633 switch (state) {
634 case JobState::INIT:
635 case JobState::RUN:
636 case JobState::DOWNLOAD:
637 case JobState::MERGE:
638 std::cout << JobState::name[state] << "\t";
639 break;
640 case JobState::FINISHED:
641 std::cout << "\033[1;32m" << JobState::name[state] << "\033[0m\t";
642 break;
643 case JobState::FAILED:
644 std::cout << "\033[1;31m" << JobState::name[state] << "\033[0m\t";
645 break;
646 }
647 std::cout << details << std::endl;
648
649 allDone &= (state == JobState::FINISHED || state == JobState::FAILED);
650 }
651
652 std::cout << std::endl;
653
654 data.retrieved = true;
655 data.completed = allDone;
656 return ::StatusCode::SUCCESS;
657}
658
659void EL::PrunDriver::status(const std::string& location)
660{
661 RCU_REQUIRE(not location.empty());
662 TmpCd tmpDir(location);
664 sh.load("input");
665 RCU_ASSERT(sh.size());
666 processAllInState(sh, JobState::RUN, 0);
667 sh.save("input");
668 for (SH::SampleHandler::iterator s = sh.begin(); s != sh.end(); ++s) {
669 JobState::Enum state = sampleState(*s);
670 std::string details = (*s)->meta()->castString("nc_ELG_state_details", "", SH::MetaObject::CAST_NOCAST_DEFAULT);
671 if (not details.empty()) { details = '(' + details + ')'; }
672 std::cout << (*s)->name() << "\t" << JobState::name[state]
673 << "\t" << details << std::endl;
674 }
675}
676
677void EL::PrunDriver::setState(const std::string& location,
678 const std::string& task,
679 const std::string& state)
680{
681 RCU_REQUIRE(not location.empty());
682 RCU_REQUIRE(not task.empty());
683 RCU_REQUIRE(not state.empty());
684 TmpCd tmpDir(location);
686 sh.load("input");
687 RCU_ASSERT(sh.size());
688 if (not sh.get(task)) {
689 std::cout << "Unknown task: " << task << std::endl;
690 std::cout << "Choose one of: " << std::endl;
691 sh.print();
692 return;
693 }
694 JobState::parse(state);
695 sh.get(task)->meta()->setString("nc_ELG_state", state);
696 sh.save("input");
697}
#define RCU_INVARIANT(x)
Definition Assert.h:189
#define RCU_ASSERT(x)
Definition Assert.h:210
#define RCU_NEW_INVARIANT(x)
Definition Assert.h:221
#define RCU_REQUIRE(x)
Definition Assert.h:196
#define RCU_ASSERT0(y)
Definition Assert.h:214
#define RCU_READ_INVARIANT(x)
Definition Assert.h:217
#define ANA_MSG_INFO(xmsg)
Macro printing info messages.
#define ANA_MSG_ERROR(xmsg)
Macro printing error messages.
#define ANA_MSG_WARNING(xmsg)
Macro printing warning messages.
#define ANA_CHECK(EXP)
check whether the given expression was successful
#define INIT(__TYPE)
virtual void lock()=0
Interface to allow an object to lock itself when made const in SG.
static Double_t ss
const std::string outputLabel
std::string PathResolverFindCalibFile(const std::string &logical_file_name)
std::vector< WorkUnit > WorkList
static SH::MetaObject defaultOpts()
static void processAllInState(const SH::SampleHandler &sh, JobState::Enum state, const size_t nThreads)
ClassImp(EL::PrunDriver) namespace
std::string outputFileNames(const EL::Job &job)
static Status::Enum submit(SH::Sample *const sample, const bool isFirstSample)
static bool downloadContainer(const std::string &name, const std::string &location)
static JobState::Enum nextState(JobState::Enum state, Status::Enum status)
static std::string formatOutputName(const SH::MetaObject &sampleMeta, const std::string &pattern)
static Status::Enum checkPandaTask(SH::Sample *const sample)
static void saveJobDef(const std::string &fileName, const EL::Job &job, const SH::SampleHandler sh)
static Status::Enum download(SH::Sample *const sample)
static JobState::Enum sampleState(SH::Sample *sample)
static SH::SampleHandler outputSH(const SH::SampleHandler &in, const std::string &outputLabel)
static void processTask(SH::Sample *const sample, const bool isFirstSample)
SH::MetaObject * options()
the list of options to jobs with this driver
virtual::StatusCode doManagerStep(Detail::ManagerData &data) const
Definition Job.h:42
const OutputStream * outputIter
Definition Job.h:139
static const std::string optGridPrunShipAdditionalFilesOrDirs
Enables to ship additional files to the tarbal sent to the grid Should be a list of comma separated p...
Definition Job.h:473
static const std::string optContainerSuffix
a Driver to submit jobs via prun
Definition PrunDriver.h:23
static void status(const std::string &location)
::StatusCode doRetrieve(Detail::ManagerData &data) const
static void setState(const std::string &location, const std::string &task, const std::string &state)
void testInvariant() const
A class that manages meta-data to be associated with an object.
Definition MetaObject.h:48
@ CAST_NOCAST_DEFAULT
cast and return the default value if the input has the wrong type
Definition MetaObject.h:70
void setString(const std::string &name, const std::string &value)
set the meta-data string with the given name
std::string castString(const std::string &name, const std::string &def_val="", CastMode mode=CAST_ERROR_THROW) const
the meta-data string with the given name
A class that manages a list of Sample objects.
void save(const std::string &directory) const
save the list of samples to the given directory
iterator begin() const
the begin iterator to use
boost::transform_iterator< SamplePtrToRawSample, std::vector< std::shared_ptr< Sample > >::const_iterator > iterator
the iterator to use
iterator end() const
the end iterator to use
a base class that manages a set of files belonging to a particular data set and the associated meta-d...
Definition Sample.h:49
STL class.
std::map< std::string, std::string, std::less<> > parse(const std::string &list)
const std::string process
std::vector< std::string > files
file names and file pointers
Definition hcg.cxx:52
std::string label(const std::string &format, int i)
Definition label.h:19
@ doRetrieve
call the actual doRetrieve method
@ submitJob
do the actual job submission
Definition ManagerStep.h:92
::StatusCode StatusCode
StatusCode definition for legacy code.
void exec(const std::string &cmd)
effects: execute the given command guarantee: strong failures: out of memory II failures: system fail...
Definition ShellExec.cxx:27
void hadd(const std::string &output_file, const std::vector< std::string > &input_files, unsigned max_files)
effects: perform the hadd functionality guarantee: basic failures: out of memory III failures: i/o er...
Definition hadd.cxx:29
std::vector< RucioDownloadResult > rucioDownloadList(const std::string &location, const std::vector< std::string > &datasets)
run rucio-download with multiple datasets
std::vector< RucioListDidsEntry > rucioListDids(const std::string &dataset)
run rucio-list-dids for the given dataset
Definition merge.py:1
-diff
DataModel_detail::iterator< DVL > unique(typename DataModel_detail::iterator< DVL > beg, typename DataModel_detail::iterator< DVL > end)
Specialization of unique for DataVector/List.
void sort(typename DataModel_detail::iterator< DVL > beg, typename DataModel_detail::iterator< DVL > end)
Specialization of sort for DataVector/List.
an internal data structure for passing data between different manager objects anbd step
Definition ManagerData.h:46
TFile * file