#include "EndPathExecutor.h"

Classes
class	PathsDoneTask

Public Member Functions
	EndPathExecutor (ScheduleID sid, PathManager &pm, ActionTable const &actions, 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 ()

Private Attributes
ScheduleContext const	sc_

ActionTable const &	actionTable_

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 41 of file EndPathExecutor.h.

Constructor & Destructor Documentation

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

Definition at line 44 of file EndPathExecutor.cc.

References actionTable_, endPathInfo_, outputWorkers_, outputWorkersToOpen_, selectProducts(), taskGroup_, util::values(), and art::PathsInfo::workers().

     : sc_{sid}
     , actionTable_{actionTable}
     , endPathInfo_{pm.endPathInfo(sid)}
     , taskGroup_{group}
   {
     for (auto const& worker :
          endPathInfo_.workers() | ::ranges::views::values) {
       assert(sid == worker->scheduleID());
       if (auto owp = std::dynamic_pointer_cast<OutputWorker>(worker)) {
         outputWorkers_.emplace_back(owp.get());
       }
     }
     outputWorkersToOpen_.insert(outputWorkers_.cbegin(), outputWorkers_.cend());
     outputCallbacks.registerCallback(
       [this](auto const& tables) { selectProducts(tables); });
   }

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

delete

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

delete

Member Function Documentation

void art::EndPathExecutor::beginJob ( detail::SharedResources const & resources )

Definition at line 67 of file EndPathExecutor.cc.

References endPathInfo_.

Referenced by art::Schedule::beginJob().

   {
     for (auto& worker : unique_workers(endPathInfo_)) {
       worker.beginJob(resources);
     }
   }

void art::EndPathExecutor::closeAllOutputFiles ( )

Definition at line 150 of file EndPathExecutor.cc.

References outputWorkers_.

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

void art::EndPathExecutor::closeSomeOutputFiles ( )

Definition at line 345 of file EndPathExecutor.cc.

References outputWorkersToClose_, outputWorkersToOpen_, setOutputFileStatus(), and art::Switching.

   {
     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;
       }
       ow->closeFile();
     }
     outputWorkersToOpen_ = std::move(outputWorkersToClose_);
   }

void art::EndPathExecutor::endJob ( )

Definition at line 75 of file EndPathExecutor.cc.

References e, art::errors::EndJobFailure, and endPathInfo_.

Referenced by art::Schedule::endJob().

   {
     Exception error{errors::EndJobFailure};
     // FIXME: There seems to be little value-added by the catch and rethrow
     // here.
     for (auto& worker : unique_workers(endPathInfo_)) {
       try {
         worker.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 410 of file EndPathExecutor.cc.

References outputWorkers_.

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

void art::EndPathExecutor::openSomeOutputFiles ( FileBlock const & fb )

Definition at line 366 of file EndPathExecutor.cc.

References art::Open, outputWorkersToOpen_, and setOutputFileStatus().

   {
     for (auto ow : outputWorkersToOpen_) {
       ow->openFile(fb);
     }
     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 331 of file EndPathExecutor.cc.

References outputWorkersToClose_.

   {
     return !outputWorkersToClose_.empty();
   }

bool art::EndPathExecutor::outputsToOpen ( ) const

Definition at line 360 of file EndPathExecutor.cc.

References outputWorkersToOpen_.

   {
     return !outputWorkersToOpen_.empty();
   }

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

Definition at line 214 of file EndPathExecutor.cc.

References endPathInfo_, art::errors::EventProcessorFailure, art::PathsInfo::incrementPassedEventCount(), and art::PathsInfo::paths().

Referenced by art::Schedule::process().

   {
     for (auto& worker : unique_workers(endPathInfo_)) {
       worker.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 290 of file EndPathExecutor.cc.

References endPathInfo_, art::ScheduleContext::id(), art::PathsInfo::incrementTotalEventCount(), art::GlobalTaskGroup::may_run(), art::PathsInfo::paths(), art::PathsInfo::reset_for_event(), sc_, taskGroup_, TDEBUG_BEGIN_FUNC_SI, and TDEBUG_END_FUNC_SI.

Referenced by art::Schedule::process_event_observers().

   {
     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 395 of file EndPathExecutor.cc.

References outputWorkers_, and outputWorkersToClose_.

   {
     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;
       }
       if (ow->requestsToCloseFile()) {
         outputWorkersToClose_.insert(ow);
       }
     }
   }

void art::EndPathExecutor::respondToCloseInputFile ( FileBlock const & fb )

Definition at line 118 of file EndPathExecutor.cc.

References endPathInfo_.

Referenced by art::Schedule::respondToCloseInputFile().

   {
     for (auto& worker : unique_workers(endPathInfo_)) {
       worker.respondToCloseInputFile(fb);
     }
   }

void art::EndPathExecutor::respondToCloseOutputFiles ( FileBlock const & fb )

Definition at line 134 of file EndPathExecutor.cc.

References endPathInfo_.

Referenced by art::Schedule::respondToCloseOutputFiles().

   {
     for (auto& worker : unique_workers(endPathInfo_)) {
       worker.respondToCloseOutputFiles(fb);
     }
   }

void art::EndPathExecutor::respondToOpenInputFile ( FileBlock const & fb )

Definition at line 110 of file EndPathExecutor.cc.

References endPathInfo_.

Referenced by art::Schedule::respondToOpenInputFile().

   {
     for (auto& worker : unique_workers(endPathInfo_)) {
       worker.respondToOpenInputFile(fb);
     }
   }

void art::EndPathExecutor::respondToOpenOutputFiles ( FileBlock const & fb )

Definition at line 126 of file EndPathExecutor.cc.

References endPathInfo_.

Referenced by art::Schedule::respondToOpenOutputFiles().

   {
     for (auto& worker : unique_workers(endPathInfo_)) {
       worker.respondToOpenOutputFiles(fb);
     }
   }

void art::EndPathExecutor::seedRunRangeSet ( RangeSetHandler const & rsh )

Definition at line 158 of file EndPathExecutor.cc.

References art::RangeSetHandler::clone(), and runRangeSetHandler_.

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

void art::EndPathExecutor::seedSubRunRangeSet ( RangeSetHandler const & rsh )

Definition at line 183 of file EndPathExecutor.cc.

References art::RangeSetHandler::clone(), and subRunRangeSetHandler_.

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

void art::EndPathExecutor::selectProducts ( ProductTables const & tables )

Definition at line 102 of file EndPathExecutor.cc.

References outputWorkers_.

Referenced by EndPathExecutor().

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

void art::EndPathExecutor::setOutputFileStatus ( OutputFileStatus ofs )

Definition at line 386 of file EndPathExecutor.cc.

References fileStatus_, and outputWorkers_.

Referenced by closeSomeOutputFiles(), and openSomeOutputFiles().

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

void art::EndPathExecutor::setRunAuxiliaryRangeSetID ( RangeSet const & rs )

Definition at line 164 of file EndPathExecutor.cc.

References outputWorkers_.

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

void art::EndPathExecutor::setSubRunAuxiliaryRangeSetID ( RangeSet const & rs )

Definition at line 189 of file EndPathExecutor.cc.

References outputWorkers_.

   {
     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 142 of file EndPathExecutor.cc.

References outputWorkers_.

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

void art::EndPathExecutor::writeEvent ( EventPrincipal & ep )

Definition at line 313 of file EndPathExecutor.cc.

References art::PathContext::end_path_spec(), art::EventPrincipal::eventID(), art::ScheduleContext::id(), art::EventPrincipal::isLastInSubRun(), outputWorkers_, runRangeSetHandler_, sc_, subRunRangeSetHandler_, and TDEBUG_FUNC_SI.

   {
     // 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_) {
       ow->writeEvent(ep, pc);
     }
     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 172 of file EndPathExecutor.cc.

References fileStatus_, outputWorkers_, runRangeSetHandler_, and art::Switching.

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

void art::EndPathExecutor::writeSubRun ( SubRunPrincipal & srp )

Definition at line 199 of file EndPathExecutor.cc.

References fileStatus_, outputWorkers_, subRunRangeSetHandler_, and art::Switching.

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