art::EndPathExecutor Class Reference

#include <EndPathExecutor.h>

Classes
class	PathsDoneTask

Public Member Functions
	EndPathExecutor (ScheduleID sid, PathManager &pm, ActionTable const &actions, ActivityRegistry const &areg, UpdateOutputCallbacks &callbacks, GlobalTaskGroup &task_group)
	EndPathExecutor (EndPathExecutor &&)=delete
EndPathExecutor &	operator= (EndPathExecutor &&)=delete
	EndPathExecutor (EndPathExecutor const &)=delete
EndPathExecutor &	operator= (EndPathExecutor const &)=delete
void	beginJob (detail::SharedResources const &resources)
void	endJob ()
void	selectProducts (ProductTables const &)
void	respondToOpenInputFile (FileBlock const &fb)
void	respondToCloseInputFile (FileBlock const &fb)
void	respondToOpenOutputFiles (FileBlock const &fb)
void	respondToCloseOutputFiles (FileBlock const &fb)
bool	someOutputsOpen () const
void	closeAllOutputFiles ()
void	seedRunRangeSet (RangeSetHandler const &)
void	setRunAuxiliaryRangeSetID (RangeSet const &rs)
void	writeRun (RunPrincipal &rp)
void	seedSubRunRangeSet (RangeSetHandler const &)
void	setSubRunAuxiliaryRangeSetID (RangeSet const &rs)
void	writeSubRun (SubRunPrincipal &srp)
void	process (Transition, Principal &)
void	process_event (hep::concurrency::WaitingTaskPtr finalizeEventTask, EventPrincipal &)
void	writeEvent (EventPrincipal &)
bool	outputsToClose () const
void	closeSomeOutputFiles ()
bool	outputsToOpen () const
void	openSomeOutputFiles (FileBlock const &fb)
void	setOutputFileStatus (OutputFileStatus)
void	recordOutputClosureRequests (Granularity)
void	incrementInputFileNumber ()
bool	allAtLimit () const

Private Attributes
ScheduleContext const	sc_
ActionTable const &	actionTable_
ActivityRegistry const &	actReg_
PathsInfo &	endPathInfo_
GlobalTaskGroup &	taskGroup_
std::vector< OutputWorker * >	outputWorkers_ {}
std::unique_ptr< RangeSetHandler >	runRangeSetHandler_ {nullptr}
std::unique_ptr< RangeSetHandler >	subRunRangeSetHandler_ {nullptr}
std::atomic< OutputFileStatus >	fileStatus_ {OutputFileStatus::Closed}
std::set< OutputWorker * >	outputWorkersToOpen_ {}
std::set< OutputWorker * >	outputWorkersToClose_ {}

Friends
class	Schedule

Detailed Description

Definition at line 42 of file EndPathExecutor.h.

Constructor & Destructor Documentation

art::EndPathExecutor::EndPathExecutor	(	ScheduleID	sid,
		PathManager &	pm,
		ActionTable const &	actions,
		ActivityRegistry const &	areg,
		UpdateOutputCallbacks &	callbacks,
		GlobalTaskGroup &	task_group
	)

Definition at line 37 of file EndPathExecutor.cc.

    : sc_{sid}
    , actionTable_{actionTable}
    , actReg_{areg}
    , endPathInfo_{pm.endPathInfo(sid)}
    , taskGroup_{group}
  {
    for (auto const& val : endPathInfo_.workers()) {
      auto w = val.second.get();
      assert(sid == w->scheduleID());
      auto owp = dynamic_cast<OutputWorker*>(w);
      if (owp != nullptr) {
        outputWorkers_.emplace_back(owp);
      }
    }
    outputWorkersToOpen_.insert(outputWorkers_.cbegin(), outputWorkers_.cend());
    outputCallbacks.registerCallback(
      [this](auto const& tables) { this->selectProducts(tables); });
  }

art::EndPathExecutor::EndPathExecutor ( EndPathExecutor &&  )

delete

art::EndPathExecutor::EndPathExecutor ( EndPathExecutor const &  )

delete

Member Function Documentation

bool art::EndPathExecutor::allAtLimit

(

)

const

Definition at line 455 of file EndPathExecutor.cc.

  {
    if (outputWorkers_.empty()) {
      return false;
    }
    bool all_at_limit = true;
    for (auto w : outputWorkers_) {
      if (!w->limitReached()) {
        all_at_limit = false;
        break;
      }
    }
    if (all_at_limit) {
      mf::LogInfo("SuccessfulTermination")
        << "The job is terminating successfully because each output module\n"
        << "has reached its configured limit.\n";
    }
    return all_at_limit;
  }

void art::EndPathExecutor::beginJob

(

detail::SharedResources const &

resources

)

Definition at line 63 of file EndPathExecutor.cc.

  {
    for (auto& label_and_worker : endPathInfo_.workers()) {
      auto& w = *label_and_worker.second;
      if (detail::skip_non_replicated(w)) {
        continue;
      }
      w.beginJob(resources);
    }
  }

void art::EndPathExecutor::closeAllOutputFiles

(

)

Definition at line 170 of file EndPathExecutor.cc.

  {
    for (auto ow : outputWorkers_) {
      actReg_.sPreCloseOutputFile.invoke(ow->label());
      ow->closeFile();
      actReg_.sPostCloseOutputFile.invoke(
        OutputFileInfo(ow->label(), ow->lastClosedFileName()));
    }
  }

void art::EndPathExecutor::closeSomeOutputFiles

(

)

Definition at line 375 of file EndPathExecutor.cc.

  {
    setOutputFileStatus(OutputFileStatus::Switching);
    for (auto ow : outputWorkersToClose_) {
      // Skip files that are already closed due to other end-path
      // executors already closing them.
      if (!ow->fileIsOpen()) {
        continue;
      }
      actReg_.sPreCloseOutputFile.invoke(ow->label());
      ow->closeFile();
      actReg_.sPostCloseOutputFile.invoke(
        OutputFileInfo{ow->label(), ow->lastClosedFileName()});
    }
    outputWorkersToOpen_ = move(outputWorkersToClose_);
  }

void art::EndPathExecutor::endJob

(

)

Definition at line 75 of file EndPathExecutor.cc.

  {
    Exception error{errors::EndJobFailure};
    // FIXME: There seems to be little value-added by the catch and rethrow
    // here.
    for (auto& label_and_worker : endPathInfo_.workers()) {
      auto& w = *label_and_worker.second;
      if (detail::skip_non_replicated(w)) {
        continue;
      }
      try {
        w.endJob();
      }
      catch (cet::exception& e) {
        error << "cet::exception caught in Schedule::endJob\n"
              << e.explain_self();
        throw error;
      }
      catch (exception& e) {
        error << "Standard library exception caught in Schedule::endJob\n"
              << e.what();
        throw error;
      }
      catch (...) {
        error << "Unknown exception caught in Schedule::endJob\n";
        throw error;
      }
    }
  }

void art::EndPathExecutor::incrementInputFileNumber

(

)

Definition at line 447 of file EndPathExecutor.cc.

  {
    for (auto ow : outputWorkers_) {
      ow->incrementInputFileNumber();
    }
  }

void art::EndPathExecutor::openSomeOutputFiles

(

FileBlock const &

fb

)

Definition at line 399 of file EndPathExecutor.cc.

  {
    for (auto ow : outputWorkersToOpen_) {
      if (!ow->openFile(fb)) {
        continue;
      }
      actReg_.sPostOpenOutputFile.invoke(ow->label());
    }
    setOutputFileStatus(OutputFileStatus::Open);
    outputWorkersToOpen_.clear();
  }

EndPathExecutor& art::EndPathExecutor::operator= ( EndPathExecutor &&  )

delete

EndPathExecutor& art::EndPathExecutor::operator= ( EndPathExecutor const &  )

delete

bool art::EndPathExecutor::outputsToClose

(

)

const

Definition at line 361 of file EndPathExecutor.cc.

  {
    return !outputWorkersToClose_.empty();
  }

bool art::EndPathExecutor::outputsToOpen

(

)

const

Definition at line 393 of file EndPathExecutor.cc.

  {
    return !outputWorkersToOpen_.empty();
  }

void art::EndPathExecutor::process	(	Transition	trans,
		Principal &	principal
	)

Definition at line 237 of file EndPathExecutor.cc.

  {
    for (auto& label_and_worker : endPathInfo_.workers()) {
      auto& w = *label_and_worker.second;
      if (detail::skip_non_replicated(w)) {
        continue;
      }
      w.reset();
    }
    try {
      if (!endPathInfo_.paths().empty()) {
        endPathInfo_.paths().front().process(trans, principal);
      }
    }
    catch (cet::exception& ex) {
      throw Exception(errors::EventProcessorFailure, "EndPathExecutor:")
        << "an exception occurred during current event processing\n"
        << ex;
    }
    catch (...) {
      mf::LogError("PassingThrough")
        << "an exception occurred during current event processing\n";
      throw;
    }
    endPathInfo_.incrementPassedEventCount();
  }

void art::EndPathExecutor::process_event	(	hep::concurrency::WaitingTaskPtr	finalizeEventTask,
		EventPrincipal &
	)

Definition at line 317 of file EndPathExecutor.cc.

  {
    auto const sid = sc_.id();
    TDEBUG_BEGIN_FUNC_SI(4, sid);
    endPathInfo_.reset_for_event();
    endPathInfo_.incrementTotalEventCount();
    try {
      auto pathsDoneTask =
        make_waiting_task<PathsDoneTask>(this, finalizeEventTask, taskGroup_);
      if (endPathInfo_.paths().empty()) {
        taskGroup_.may_run(pathsDoneTask);
      } else {
        endPathInfo_.paths().front().process(pathsDoneTask, ep);
      }
    }
    catch (...) {
      taskGroup_.may_run(finalizeEventTask, current_exception());
    }
    TDEBUG_END_FUNC_SI(4, sid);
  }

void art::EndPathExecutor::recordOutputClosureRequests

(

Granularity

atBoundary

)

Definition at line 431 of file EndPathExecutor.cc.

  {
    for (auto ow : outputWorkers_) {
      if (atBoundary < ow->fileGranularity()) {
        // The boundary we are checking at is finer than the checks
        // the output worker needs, nothing to do.
        continue;
      }
      auto wants_to_close = ow->requestsToCloseFile();
      if (wants_to_close) {
        outputWorkersToClose_.insert(ow);
      }
    }
  }

void art::EndPathExecutor::respondToCloseInputFile

(

FileBlock const &

fb

)

Definition at line 126 of file EndPathExecutor.cc.

  {
    for (auto& label_and_worker : endPathInfo_.workers()) {
      auto& w = *label_and_worker.second;
      if (detail::skip_non_replicated(w)) {
        continue;
      }
      w.respondToCloseInputFile(fb);
    }
  }

void art::EndPathExecutor::respondToCloseOutputFiles

(

FileBlock const &

fb

)

Definition at line 150 of file EndPathExecutor.cc.

  {
    for (auto& label_and_worker : endPathInfo_.workers()) {
      auto& w = *label_and_worker.second;
      if (detail::skip_non_replicated(w)) {
        continue;
      }
      w.respondToCloseOutputFiles(fb);
    }
  }

void art::EndPathExecutor::respondToOpenInputFile

(

FileBlock const &

fb

)

Definition at line 114 of file EndPathExecutor.cc.

  {
    for (auto& label_and_worker : endPathInfo_.workers()) {
      auto& w = *label_and_worker.second;
      if (detail::skip_non_replicated(w)) {
        continue;
      }
      w.respondToOpenInputFile(fb);
    }
  }

void art::EndPathExecutor::respondToOpenOutputFiles

(

FileBlock const &

fb

)

Definition at line 138 of file EndPathExecutor.cc.

  {
    for (auto& label_and_worker : endPathInfo_.workers()) {
      auto& w = *label_and_worker.second;
      if (detail::skip_non_replicated(w)) {
        continue;
      }
      w.respondToOpenOutputFiles(fb);
    }
  }

void art::EndPathExecutor::seedRunRangeSet

(

RangeSetHandler const &

rsh

)

Definition at line 181 of file EndPathExecutor.cc.

  {
    runRangeSetHandler_.reset(rsh.clone());
  }

void art::EndPathExecutor::seedSubRunRangeSet

(

RangeSetHandler const &

rsh

)

Definition at line 206 of file EndPathExecutor.cc.

  {
    subRunRangeSetHandler_.reset(rsh.clone());
  }

void art::EndPathExecutor::selectProducts

(

ProductTables const &

tables

)

Definition at line 106 of file EndPathExecutor.cc.

  {
    for (auto ow : outputWorkers_) {
      ow->selectProducts(tables);
    }
  }

void art::EndPathExecutor::setOutputFileStatus

(

OutputFileStatus

ofs

)

Definition at line 422 of file EndPathExecutor.cc.

  {
    for (auto ow : outputWorkers_) {
      ow->setFileStatus(ofs);
    }
    fileStatus_ = ofs;
  }

void art::EndPathExecutor::setRunAuxiliaryRangeSetID

(

RangeSet const &

rs

)

Definition at line 187 of file EndPathExecutor.cc.

  {
    for (auto ow : outputWorkers_) {
      ow->setRunAuxiliaryRangeSetID(rangeSet);
    }
  }

void art::EndPathExecutor::setSubRunAuxiliaryRangeSetID

(

RangeSet const &

rs

)

Definition at line 212 of file EndPathExecutor.cc.

  {
    for (auto ow : outputWorkers_) {
      // For RootOutput this enters the possibly split range set into
      // the range set db.
      ow->setSubRunAuxiliaryRangeSetID(rs);
    }
  }

bool art::EndPathExecutor::someOutputsOpen

(

)

const

Definition at line 162 of file EndPathExecutor.cc.

  {
    return any_of(outputWorkers_.cbegin(), outputWorkers_.cend(), [](auto ow) {
      return ow->fileIsOpen();
    });
  }

void art::EndPathExecutor::writeEvent

(

EventPrincipal &

ep

)

Definition at line 340 of file EndPathExecutor.cc.

  {
    // We don't worry about providing the sorted list of module names
    // for the end_path right now.  If users decide it is necessary to
    // know what they are, then we can provide them.
    PathContext const pc{sc_, PathContext::end_path_spec(), {}};
    for (auto ow : outputWorkers_) {
      ModuleContext const mc{pc, ow->description()};
      actReg_.sPreWriteEvent.invoke(mc);
      ow->writeEvent(ep, mc);
      actReg_.sPostWriteEvent.invoke(mc);
    }
    auto const& eid = ep.eventID();
    bool const lastInSubRun{ep.isLastInSubRun()};
    TDEBUG_FUNC_SI(5, sc_.id())
      << "eid: " << eid.run() << ", " << eid.subRun() << ", " << eid.event();
    runRangeSetHandler_->update(eid, lastInSubRun);
    subRunRangeSetHandler_->update(eid, lastInSubRun);
  }

void art::EndPathExecutor::writeRun

(

RunPrincipal &

rp

)

Definition at line 195 of file EndPathExecutor.cc.

  {
    for (auto ow : outputWorkers_) {
      ow->writeRun(rp);
    }
    if (fileStatus_.load() == OutputFileStatus::Switching) {
      runRangeSetHandler_->rebase();
    }
  }

void art::EndPathExecutor::writeSubRun

(

SubRunPrincipal &

srp

)

Definition at line 222 of file EndPathExecutor.cc.

  {
    for (auto ow : outputWorkers_) {
      ow->writeSubRun(srp);
    }
    if (fileStatus_.load() == OutputFileStatus::Switching) {
      subRunRangeSetHandler_->rebase();
    }
  }

Friends And Related Function Documentation

friend class Schedule

friend

Definition at line 43 of file EndPathExecutor.h.

Member Data Documentation

ActionTable const& art::EndPathExecutor::actionTable_

private

Definition at line 129 of file EndPathExecutor.h.

ActivityRegistry const& art::EndPathExecutor::actReg_

private

Definition at line 130 of file EndPathExecutor.h.

PathsInfo& art::EndPathExecutor::endPathInfo_

private

Definition at line 131 of file EndPathExecutor.h.

std::atomic<OutputFileStatus> art::EndPathExecutor::fileStatus_ {OutputFileStatus::Closed}

private

Definition at line 141 of file EndPathExecutor.h.

std::vector<OutputWorker*> art::EndPathExecutor::outputWorkers_ {}

private

Definition at line 134 of file EndPathExecutor.h.

std::set<OutputWorker*> art::EndPathExecutor::outputWorkersToClose_ {}

private

Definition at line 150 of file EndPathExecutor.h.

std::set<OutputWorker*> art::EndPathExecutor::outputWorkersToOpen_ {}

private

Definition at line 142 of file EndPathExecutor.h.

std::unique_ptr<RangeSetHandler> art::EndPathExecutor::runRangeSetHandler_ {nullptr}

private

Definition at line 136 of file EndPathExecutor.h.

ScheduleContext const art::EndPathExecutor::sc_

private

Definition at line 125 of file EndPathExecutor.h.

std::unique_ptr<RangeSetHandler> art::EndPathExecutor::subRunRangeSetHandler_ {nullptr}

private

Definition at line 138 of file EndPathExecutor.h.

GlobalTaskGroup& art::EndPathExecutor::taskGroup_

private

Definition at line 132 of file EndPathExecutor.h.

---

The documentation for this class was generated from the following files:

• art/art/Framework/Core/EndPathExecutor.h
• art/art/Framework/Core/EndPathExecutor.cc