#include "EndPathExecutor.h"

Public Member Functions
	EndPathExecutor (PathManager &pm, ActionTable &actions, ActivityRegistry &areg, MasterProductRegistry &mpr)

template<typename T >
void	process (typename T::MyPrincipal &principal)

void	beginJob ()

void	endJob ()

void	writeEvent (EventPrincipal &ep)

void	writeSubRun (SubRunPrincipal &srp)

void	writeRun (RunPrincipal &rp)

void	seedRunRangeSet (std::unique_ptr< RangeSetHandler >)

void	seedSubRunRangeSet (std::unique_ptr< RangeSetHandler >)

void	setAuxiliaryRangeSetID (SubRunPrincipal &srp)

void	setAuxiliaryRangeSetID (RunPrincipal &rp)

void	closeAllOutputFiles ()

void	openAllOutputFiles (FileBlock &fb)

void	closeSomeOutputFiles ()

void	openSomeOutputFiles (FileBlock const &fb)

void	setOutputFileStatus (OutputFileStatus)

void	respondToOpenInputFile (FileBlock const &fb)

void	respondToCloseInputFile (FileBlock const &fb)

void	respondToOpenOutputFiles (FileBlock const &fb)

void	respondToCloseOutputFiles (FileBlock const &fb)

void	recordOutputClosureRequests (Granularity)

bool	outputsToOpen () const

bool	outputsToClose () const

bool	someOutputsOpen () const

void	incrementInputFileNumber ()

bool	terminate () const

void	selectProducts (ProductList const &)

Private Types
using	OutputWorkers = std::vector< OutputWorker * >

using	OutputWorkerSet = std::set< OutputWorker * >

Private Member Functions
void	resetAll ()

template<typename T >
void	runEndPaths (typename T::MyPrincipal &)

template<class F >
void	doForAllEnabledWorkers_ (F f)

template<class F >
void	doForAllEnabledOutputWorkers_ (F f)

Private Attributes
PathsInfo &	endPathInfo_

ActionTable *	act_table_

ActivityRegistry &	actReg_

OutputWorkers	outputWorkers_

OutputWorkerSet	outputWorkersToOpen_

OutputWorkerSet	outputWorkersToClose_ {}

std::vector< unsigned char >	workersEnabled_

std::vector< unsigned char >	outputWorkersEnabled_

OutputFileStatus	fileStatus_ {OutputFileStatus::Closed}

std::unique_ptr< RangeSetHandler >	runRangeSetHandler_ {nullptr}

std::unique_ptr< RangeSetHandler >	subRunRangeSetHandler_ {nullptr}

Detailed Description

Definition at line 41 of file EndPathExecutor.h.

Member Typedef Documentation

using art::EndPathExecutor::OutputWorkers = std::vector<OutputWorker*>

private

Definition at line 90 of file EndPathExecutor.h.

using art::EndPathExecutor::OutputWorkerSet = std::set<OutputWorker*>

private

Definition at line 91 of file EndPathExecutor.h.

Constructor & Destructor Documentation

art::EndPathExecutor::EndPathExecutor	(	PathManager &	pm,
		ActionTable &	actions,
		ActivityRegistry &	areg,
		MasterProductRegistry &	mpr
	)

Definition at line 26 of file EndPathExecutor.cc.

References act_table_, actReg_, evd::details::begin(), evd::details::end(), art::PathManager::endPathInfo(), endPathInfo_, outputWorkers_, outputWorkersEnabled_, outputWorkersToOpen_, selectProducts(), art::PathsInfo::workers(), and workersEnabled_.

   : endPathInfo_{pm.endPathInfo()}
   , act_table_{&actions}
   , actReg_{areg}
   , outputWorkers_{getOutputWorkers(endPathInfo_.workers())}
   , outputWorkersToOpen_(
       std::begin(outputWorkers_),
       std::end(outputWorkers_)) // seed with all output workers
   , workersEnabled_(endPathInfo_.workers().size(), true)
   , outputWorkersEnabled_(outputWorkers_.size(), true)
 {
   mpr.registerProductListUpdatedCallback(
     [this](auto const& productList) { this->selectProducts(productList); });
 }

Member Function Documentation

void art::EndPathExecutor::beginJob ( )

Definition at line 334 of file EndPathExecutor.cc.

References doForAllEnabledWorkers_(), and w.

 {
   doForAllEnabledWorkers_([](auto w) { w->beginJob(); });
 }

void art::EndPathExecutor::closeAllOutputFiles ( )

Definition at line 90 of file EndPathExecutor.cc.

References actReg_, art::OutputWorker::closeFile(), doForAllEnabledOutputWorkers_(), art::Worker::label(), art::OutputWorker::lastClosedFileName(), art::ActivityRegistry::sPostCloseOutputFile, and art::ActivityRegistry::sPreCloseOutputFile.

 {
   doForAllEnabledOutputWorkers_([this](OutputWorker* ow) {
     actReg_.sPreCloseOutputFile.invoke(ow->label());
     ow->closeFile();
     actReg_.sPostCloseOutputFile.invoke(
       OutputFileInfo(ow->label(), ow->lastClosedFileName()));
   });
 }

void art::EndPathExecutor::closeSomeOutputFiles ( )

Definition at line 269 of file EndPathExecutor.cc.

References actReg_, outputWorkersToClose_, outputWorkersToOpen_, setOutputFileStatus(), art::ActivityRegistry::sPostCloseOutputFile, art::ActivityRegistry::sPreCloseOutputFile, and art::Switching.

 {
   auto invoke_sPreCloseOutputFile = [this](auto ow) {
     actReg_.sPreCloseOutputFile.invoke(ow->label());
   };
   auto closeFile = [](auto ow) { ow->closeFile(); };
   auto invoke_sPostCloseOutputFile = [this](auto ow) {
     actReg_.sPostCloseOutputFile.invoke(
       OutputFileInfo{ow->label(), ow->lastClosedFileName()});
   };
 
   setOutputFileStatus(OutputFileStatus::Switching);
   cet::for_all(outputWorkersToClose_, invoke_sPreCloseOutputFile);
   cet::for_all(outputWorkersToClose_, closeFile);
   cet::for_all(outputWorkersToClose_, invoke_sPostCloseOutputFile);
   outputWorkersToOpen_ = std::move(outputWorkersToClose_);
 }

template<class F >

void art::EndPathExecutor::doForAllEnabledOutputWorkers_ ( F f )

private

Definition at line 170 of file EndPathExecutor.h.

References outputWorkers_, and outputWorkersEnabled_.

Referenced by closeAllOutputFiles(), incrementInputFileNumber(), openAllOutputFiles(), recordOutputClosureRequests(), selectProducts(), setAuxiliaryRangeSetID(), setOutputFileStatus(), writeEvent(), writeRun(), and writeSubRun().

 {
   size_t index{0};
   for (auto ow : outputWorkers_) {
     if (outputWorkersEnabled_[index++]) {
       fcn(ow);
     }
   }
 }

template<class F >

void art::EndPathExecutor::doForAllEnabledWorkers_ ( F f )

private

Definition at line 158 of file EndPathExecutor.h.

References endPathInfo_, art::PathsInfo::workers(), and workersEnabled_.

Referenced by beginJob(), endJob(), resetAll(), respondToCloseInputFile(), respondToCloseOutputFiles(), respondToOpenInputFile(), and respondToOpenOutputFiles().

 {
   size_t index{0};
   for (auto const& val : endPathInfo_.workers()) {
     if (workersEnabled_[index++]) {
       fcn(val.second.get());
     }
   }
 }

void art::EndPathExecutor::endJob ( )

Definition at line 61 of file EndPathExecutor.cc.

References doForAllEnabledWorkers_(), e, art::Worker::endJob(), art::errors::EndJobFailure, and w.

 {
   bool failure{false};
   Exception error{errors::EndJobFailure};
   doForAllEnabledWorkers_([&failure, &error](Worker* w) {
     try {
       w->endJob();
     }
     catch (cet::exception& e) {
       error << "cet::exception caught in Schedule::endJob\n"
             << e.explain_self();
       failure = true;
     }
     catch (std::exception& e) {
       error << "Standard library exception caught in Schedule::endJob\n"
             << e.what();
       failure = true;
     }
     catch (...) {
       error << "Unknown exception caught in Schedule::endJob\n";
       failure = true;
     }
   });
   if (failure) {
     throw error;
   }
 }

void art::EndPathExecutor::incrementInputFileNumber ( )

Definition at line 242 of file EndPathExecutor.cc.

References doForAllEnabledOutputWorkers_().

 {
   doForAllEnabledOutputWorkers_(
     [](auto ow) { ow->incrementInputFileNumber(); });
 }

void art::EndPathExecutor::openAllOutputFiles ( FileBlock & fb )

Definition at line 101 of file EndPathExecutor.cc.

References actReg_, doForAllEnabledOutputWorkers_(), art::Worker::label(), art::OutputWorker::openFile(), and art::ActivityRegistry::sPostOpenOutputFile.

 {
   doForAllEnabledOutputWorkers_([this, &fb](OutputWorker* ow) {
     ow->openFile(fb);
     actReg_.sPostOpenOutputFile.invoke(ow->label());
   });
 }

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

Definition at line 295 of file EndPathExecutor.cc.

References actReg_, fb(), art::Open, outputWorkersToOpen_, setOutputFileStatus(), and art::ActivityRegistry::sPostOpenOutputFile.

 {
   auto openFile = [&fb](auto ow) { ow->openFile(fb); };
   auto invoke_sPostOpenOutputFile = [this](auto ow) {
     actReg_.sPostOpenOutputFile.invoke(ow->label());
   };
 
   cet::for_all(outputWorkersToOpen_, openFile);
   cet::for_all(outputWorkersToOpen_, invoke_sPostOpenOutputFile);
   setOutputFileStatus(OutputFileStatus::Open);
 
   outputWorkersToOpen_.clear();
 }

bool art::EndPathExecutor::outputsToClose ( ) const

Definition at line 255 of file EndPathExecutor.cc.

References outputWorkersToClose_.

 {
   return !outputWorkersToClose_.empty();
 }

bool art::EndPathExecutor::outputsToOpen ( ) const

Definition at line 249 of file EndPathExecutor.cc.

References outputWorkersToOpen_.

 {
   return !outputWorkersToOpen_.empty();
 }

template<typename T >

void art::EndPathExecutor::process ( typename T::MyPrincipal & principal )

Definition at line 118 of file EndPathExecutor.h.

References act_table_, art::PathsInfo::addEvent(), art::PathsInfo::addPass(), endPathInfo_, art::Event, art::errors::EventProcessorFailure, art::ActionTable::find(), art::actions::IgnoreCompletely, art::PathsInfo::pathPtrs(), resetAll(), and art::actions::Rethrow.

 {
   this->resetAll();
 
   if (T::level == Level::Event) {
     endPathInfo_.addEvent();
   }
   try {
     if (!endPathInfo_.pathPtrs().empty()) {
       endPathInfo_.pathPtrs().front()->process<T>(ep);
     }
   }
   catch (cet::exception& ex) {
     actions::ActionCodes const action{T::level == Level::Event ?
                                         act_table_->find(ex.root_cause()) :
                                         actions::Rethrow};
     switch (action) {
       case actions::IgnoreCompletely: {
         mf::LogWarning(ex.category())
           << "exception being ignored for current event:\n"
           << cet::trim_right_copy(ex.what(), " \n");
         break;
       }
       default: {
         throw art::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_.addPass();
 }

void art::EndPathExecutor::recordOutputClosureRequests ( Granularity b )

Definition at line 212 of file EndPathExecutor.cc.

References doForAllEnabledOutputWorkers_(), and outputWorkersToClose_.

 {
   doForAllEnabledOutputWorkers_([this, b](auto ow) {
     // We need to support the following case:
     //
     //   fileProperties: {
     //     maxEvents: 10
     //     maxRuns: 1
     //     granularity: Event
     //   }
     //
     // If a request to close is made on a run boundary, but the
     // granularity is still Event, then the file should close.  For
     // that reason, the comparison is 'granularity > b' instead of
     // 'granularity != b'.
 
     auto const granularity = ow->fileGranularity();
     if (granularity > b || !ow->requestsToCloseFile())
       return;
 
     // Technical note: although the outputWorkersToClose_ container
     // is "moved from" in closeSomeOutputFiles, it is safe to call
     // 'insert' vis-a-vis the [lib.types.movedfrom] section of the
     // standard.  There are no preconditions for std::set::insert,
     // so no state-checking is required.
     outputWorkersToClose_.insert(ow);
   });
 }

void art::EndPathExecutor::resetAll ( )

private

Definition at line 340 of file EndPathExecutor.cc.

References doForAllEnabledWorkers_(), and w.

Referenced by process().

 {
   doForAllEnabledWorkers_([](auto w) { w->reset(); });
 }

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

Definition at line 316 of file EndPathExecutor.cc.

References doForAllEnabledWorkers_(), and w.

 {
   doForAllEnabledWorkers_([&fb](auto w) { w->respondToCloseInputFile(fb); });
 }

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

Definition at line 328 of file EndPathExecutor.cc.

References doForAllEnabledWorkers_(), and w.

 {
   doForAllEnabledWorkers_([&fb](auto w) { w->respondToCloseOutputFiles(fb); });
 }

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

Definition at line 310 of file EndPathExecutor.cc.

References doForAllEnabledWorkers_(), and w.

 {
   doForAllEnabledWorkers_([&fb](auto w) { w->respondToOpenInputFile(fb); });
 }

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

Definition at line 322 of file EndPathExecutor.cc.

References doForAllEnabledWorkers_(), and w.

 {
   doForAllEnabledWorkers_([&fb](auto w) { w->respondToOpenOutputFiles(fb); });
 }

template<typename T >

void art::EndPathExecutor::runEndPaths ( typename T::MyPrincipal & )

private

void art::EndPathExecutor::seedRunRangeSet ( std::unique_ptr< RangeSetHandler > rsh )

Definition at line 143 of file EndPathExecutor.cc.

References runRangeSetHandler_.

 {
   runRangeSetHandler_ = std::move(rsh);
 }

void art::EndPathExecutor::seedSubRunRangeSet ( std::unique_ptr< RangeSetHandler > rsh )

Definition at line 149 of file EndPathExecutor.cc.

References subRunRangeSetHandler_.

 {
   subRunRangeSetHandler_ = std::move(rsh);
 }

void art::EndPathExecutor::selectProducts ( ProductList const & productList )

Definition at line 205 of file EndPathExecutor.cc.

References doForAllEnabledOutputWorkers_(), and w.

Referenced by EndPathExecutor().

 {
   doForAllEnabledOutputWorkers_(
     [&productList](auto w) { w->selectProducts(productList); });
 }

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

Definition at line 155 of file EndPathExecutor.cc.

References doForAllEnabledOutputWorkers_(), fileStatus_, runRangeSetHandler_, subRunRangeSetHandler_, art::Switching, art::SubRunPrincipal::updateSeenRanges(), and w.

 {
   assert(subRunRangeSetHandler_);
   assert(runRangeSetHandler_);
   // Ranges are split/flushed only for a RangeSetHandler whose dynamic
   // type is 'ClosedRangeSetHandler'.  The implementations for the
   // 'OpenRangeSetHandler' are nops.
   //
   // Consider the following range-sets
   //  SubRun RangeSet: { Run 1 : SubRun 1 : Events [1,7) }  <-- Current iterator
   //  of handler Run    RangeSet: { Run 1 : SubRun 0 : Events [5,11)
   //                             SubRun 1 : Events [1,7)    <-- Current iterator
   //                             of handler SubRun 1 : Events [9,15) }
   if (fileStatus_ == OutputFileStatus::Switching) {
     // For a range split just before SubRun 1, Event 6, the range sets
     // should become:
     //
     //  SubRun RangeSet: { Run 1 : SubRun 1 : Events [1,6)
     //                             SubRun 1 : Events [6,7) } <-- Updated
     //                             iterator of handler
     //  Run    RangeSet: { Run 1 : SubRun 0 : Events [5,11)
     //                             SubRun 1 : Events [1,6)
     //                             SubRun 1 : Events [6,7)   <-- Updated
     //                             iterator of handler SubRun 1 : Events [9,15)
     //                             }
     subRunRangeSetHandler_->maybeSplitRange();
     runRangeSetHandler_->maybeSplitRange();
   } else {
     subRunRangeSetHandler_->flushRanges();
   }
   auto const& ranges = subRunRangeSetHandler_->seenRanges();
   srp.updateSeenRanges(ranges);
   doForAllEnabledOutputWorkers_(
     [&ranges](auto w) { w->setSubRunAuxiliaryRangeSetID(ranges); });
 }

void art::EndPathExecutor::setAuxiliaryRangeSetID ( RunPrincipal & rp )

Definition at line 192 of file EndPathExecutor.cc.

References doForAllEnabledOutputWorkers_(), fileStatus_, runRangeSetHandler_, art::Switching, art::RunPrincipal::updateSeenRanges(), and w.

 {
   assert(runRangeSetHandler_);
   if (fileStatus_ != OutputFileStatus::Switching) {
     runRangeSetHandler_->flushRanges();
   }
   auto const& ranges = runRangeSetHandler_->seenRanges();
   rp.updateSeenRanges(ranges);
   doForAllEnabledOutputWorkers_(
     [&ranges](auto w) { w->setRunAuxiliaryRangeSetID(ranges); });
 }

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

Definition at line 288 of file EndPathExecutor.cc.

References doForAllEnabledOutputWorkers_(), and fileStatus_.

Referenced by closeSomeOutputFiles(), and openSomeOutputFiles().

 {
   doForAllEnabledOutputWorkers_([ofs](auto ow) { ow->setFileStatus(ofs); });
   fileStatus_ = ofs;
 }

bool art::EndPathExecutor::someOutputsOpen ( ) const

Definition at line 261 of file EndPathExecutor.cc.

References outputWorkers_.

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

bool art::EndPathExecutor::terminate ( ) const

Definition at line 45 of file EndPathExecutor.cc.

References outputWorkers_, and w.

 {
   bool const rc = !outputWorkers_.empty() && // Necessary because std::all_of()
                                              // returns true if range is empty.
                   std::all_of(outputWorkers_.cbegin(),
                               outputWorkers_.cend(),
                               [](auto& w) { return w->limitReached(); });
   if (rc) {
     mf::LogInfo("SuccessfulTermination")
       << "The job is terminating successfully because each output module\n"
       << "has reached its configured limit.\n";
   }
   return rc;
 }

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

Definition at line 128 of file EndPathExecutor.cc.

References actReg_, doForAllEnabledOutputWorkers_(), art::EventPrincipal::id(), art::EventPrincipal::isLastInSubRun(), runRangeSetHandler_, art::ActivityRegistry::sPostWriteEvent, art::ActivityRegistry::sPreWriteEvent, subRunRangeSetHandler_, and w.

 {
   doForAllEnabledOutputWorkers_([this, &ep](auto w) {
     auto const& md = w->description();
     actReg_.sPreWriteEvent.invoke(md);
     w->writeEvent(ep);
     actReg_.sPostWriteEvent.invoke(md);
   });
   auto const& eid = ep.id();
   bool const lastInSubRun{ep.isLastInSubRun()};
   runRangeSetHandler_->update(eid, lastInSubRun);
   subRunRangeSetHandler_->update(eid, lastInSubRun);
 }

void art::EndPathExecutor::writeRun ( RunPrincipal & rp )

Definition at line 110 of file EndPathExecutor.cc.

References doForAllEnabledOutputWorkers_(), fileStatus_, runRangeSetHandler_, art::Switching, and w.

 {
   doForAllEnabledOutputWorkers_([&rp](auto w) { w->writeRun(rp); });
   if (fileStatus_ == OutputFileStatus::Switching) {
     runRangeSetHandler_->rebase();
   }
 }

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

Definition at line 119 of file EndPathExecutor.cc.

References doForAllEnabledOutputWorkers_(), fileStatus_, subRunRangeSetHandler_, art::Switching, and w.

 {
   doForAllEnabledOutputWorkers_([&srp](auto w) { w->writeSubRun(srp); });
   if (fileStatus_ == OutputFileStatus::Switching) {
     subRunRangeSetHandler_->rebase();
   }
 }