art::Path Class Reference

#include <Path.h>

Classes
class	WorkerDoneTask

Public Member Functions
	Path (ActionTable const &, ActivityRegistry const &, PathContext const &, std::vector< WorkerInPath > &&, HLTGlobalStatus *, GlobalTaskGroup &) noexcept
ScheduleID	scheduleID () const
PathSpec const &	pathSpec () const
PathID	pathID () const
std::string const &	name () const
std::vector< WorkerInPath > const &	workersInPath () const
hlt::HLTState	state () const
std::size_t	timesRun () const
std::size_t	timesPassed () const
std::size_t	timesFailed () const
std::size_t	timesExcept () const
void	clearCounters ()
void	process (Transition, Principal &)
void	process (hep::concurrency::WaitingTaskPtr pathsDoneTask, EventPrincipal &)

Private Member Functions
void	runWorkerTask (size_t idx, size_t max_idx, EventPrincipal &, hep::concurrency::WaitingTaskPtr pathsDone)
void	process_event_idx_asynch (size_t idx, size_t max_idx, EventPrincipal &, hep::concurrency::WaitingTaskPtr pathsDone)
void	process_event_idx (size_t const idx, size_t const max_idx, EventPrincipal &, hep::concurrency::WaitingTaskPtr pathsDone)
void	process_event_workerFinished (size_t const idx, size_t const max_idx, EventPrincipal &ep, bool should_continue, hep::concurrency::WaitingTaskPtr pathsDone)
void	process_event_pathFinished (size_t const idx, bool should_continue, hep::concurrency::WaitingTaskPtr pathsDone)

Private Attributes
ActionTable const &	actionTable_
ActivityRegistry const &	actReg_
PathContext const	pc_
size_t const	pathPosition_
std::vector< WorkerInPath >	workers_
cet::exempt_ptr< HLTGlobalStatus >	trptr_
GlobalTaskGroup &	taskGroup_
hlt::HLTState	state_ {hlt::Ready}
std::size_t	timesRun_ {}
std::size_t	timesPassed_ {}
std::size_t	timesFailed_ {}
std::size_t	timesExcept_ {}

Detailed Description

Definition at line 31 of file Path.h.

Constructor & Destructor Documentation

art::Path::Path ( ActionTable const &  actions, ActivityRegistry const &  actReg, PathContext const &  pc, std::vector< WorkerInPath > &&  workers, HLTGlobalStatus *  pathResults, GlobalTaskGroup &  taskGroup  )

noexcept

Definition at line 35 of file Path.cc.

    : actionTable_{actions}
    , actReg_{actReg}
    , pc_{pc}
    , pathPosition_{ServiceHandle<TriggerNamesService>()->index_for(
        pc_.pathID())}
    , workers_{move(workers)}
    , trptr_{pathResults}
    , taskGroup_{taskGroup}
  {
    TDEBUG_FUNC_SI(4, pc_.scheduleID()) << hex << this << dec;
  }

Member Function Documentation

void art::Path::clearCounters

(

)

string const & art::Path::name

(

)

const

Definition at line 72 of file Path.cc.

  {
    return pc_.pathName();
  }

PathID art::Path::pathID

(

)

const

Definition at line 66 of file Path.cc.

  {
    return pc_.pathID();
  }

PathSpec const & art::Path::pathSpec

(

)

const

Definition at line 60 of file Path.cc.

  {
    return pc_.pathSpec();
  }

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

Definition at line 114 of file Path.cc.

  {
    // Invoke pre-path signals only for the first schedule.
    if (pc_.scheduleID() == ScheduleID::first()) {
      switch (trans) {
      case Transition::BeginRun:
        actReg_.sPrePathBeginRun.invoke(name());
        break;
      case Transition::EndRun:
        actReg_.sPrePathEndRun.invoke(name());
        break;
      case Transition::BeginSubRun:
        actReg_.sPrePathBeginSubRun.invoke(name());
        break;
      case Transition::EndSubRun:
        actReg_.sPrePathEndSubRun.invoke(name());
        break;
      default: {} // No other pre-path signals supported.
      }
    }
    state_ = hlt::Ready;
    std::size_t idx = 0;
    bool all_passed{false};
    for (WorkerInPath& wip : workers_) {
      // We do not want to call (e.g.) beginRun once per schedule for
      // non-replicated modules.
      if (detail::skip_non_replicated(*wip.getWorker())) {
        continue;
      }
      try {
        all_passed = wip.run(trans, principal);
        if (!all_passed)
          break;
      }
      catch (cet::exception& e) {
        state_ = hlt::Exception;
        throw art::Exception{
          errors::ScheduleExecutionFailure, "Path: ProcessingStopped.", e}
          << "Exception going through path " << name() << "\n";
      }
      catch (...) {
        mf::LogError("PassingThrough")
          << "Exception passing through path " << name() << "\n";
        state_ = hlt::Exception;
        throw;
      }
      ++idx;
    }
    if (all_passed) {
      state_ = hlt::Pass;
    } else {
      state_ = hlt::Fail;
    }
    // Invoke post-path signals only for the last schedule.
    if (pc_.scheduleID().id() == art::Globals::instance()->nschedules() - 1) {
      HLTPathStatus const status(state_, idx);
      switch (trans) {
      case Transition::BeginRun:
        actReg_.sPostPathBeginRun.invoke(name(), status);
        break;
      case Transition::EndRun:
        actReg_.sPostPathEndRun.invoke(name(), status);
        break;
      case Transition::BeginSubRun:
        actReg_.sPostPathBeginSubRun.invoke(name(), status);
        break;
      case Transition::EndSubRun:
        actReg_.sPostPathEndSubRun.invoke(name(), status);
        break;
      default: {} // No other post-path signals supported.
      }
    }
  }

void art::Path::process	(	hep::concurrency::WaitingTaskPtr	pathsDoneTask,
		EventPrincipal &
	)

void art::Path::process_event_idx ( size_t const  idx, size_t const  max_idx, EventPrincipal &  , hep::concurrency::WaitingTaskPtr  pathsDone  )

private

Definition at line 333 of file Path.cc.

  {
    auto const sid = pc_.scheduleID();
    TDEBUG_FUNC_SI(4, sid) << "idx: " << idx << " max_idx: " << max_idx;
    auto workerDoneTask = make_waiting_task<WorkerDoneTask>(
      this, idx, max_idx, ep, pathsDone, taskGroup_);
    auto& workerInPath = workers_[idx];
    workerInPath.run(workerDoneTask, ep);
    TDEBUG_FUNC_SI(4, sid) << "idx: " << idx << " max_idx: " << max_idx;
  }

void art::Path::process_event_idx_asynch ( size_t  idx, size_t  max_idx, EventPrincipal &  , hep::concurrency::WaitingTaskPtr  pathsDone  )

private

Definition at line 233 of file Path.cc.

  {
    auto const sid = pc_.scheduleID();
    TDEBUG_BEGIN_FUNC_SI(4, sid) << "idx: " << idx << " max_idx: " << max_idx;
    taskGroup_.run([this, idx, max_idx, &ep, pathsDone] {
      runWorkerTask(idx, max_idx, ep, pathsDone);
    });
    TDEBUG_END_FUNC_SI(4, sid) << "idx: " << idx << " max_idx: " << max_idx;
  }

void art::Path::process_event_pathFinished ( size_t const  idx, bool  should_continue, hep::concurrency::WaitingTaskPtr  pathsDone  )

private

Definition at line 375 of file Path.cc.

  {
    // We come here as as part of a runWorker task.
    auto const sid = pc_.scheduleID();
    TDEBUG_FUNC_SI(4, sid) << "idx: " << idx
                           << " should_continue: " << should_continue;
    if (should_continue) {
      ++timesPassed_;
      state_ = hlt::Pass;
    } else {
      ++timesFailed_;
      state_ = hlt::Fail;
    }

    auto ex_ptr = std::exception_ptr{};
    try {
      HLTPathStatus const status{state_, idx};
      if (trptr_) {
        // Not the end path.
        trptr_->at(pathPosition_) = status;
      }
      actReg_.sPostProcessPath.invoke(pc_, status);
    }
    catch (...) {
      ex_ptr = std::current_exception();
    }
    taskGroup_.may_run(pathsDone, ex_ptr);
    TDEBUG_FUNC_SI(4, sid) << "idx: " << idx
                           << " should_continue: " << should_continue
                           << (ex_ptr ? " EXCEPTION" : "");
  }

void art::Path::process_event_workerFinished ( size_t const  idx, size_t const  max_idx, EventPrincipal &  ep, bool  should_continue, hep::concurrency::WaitingTaskPtr  pathsDone  )

private

Definition at line 348 of file Path.cc.

  {
    auto const sid = pc_.scheduleID();
    TDEBUG_BEGIN_FUNC_SI(4, sid) << "idx: " << idx << " max_idx: " << max_idx
                                 << " should_continue: " << should_continue;
    auto new_idx = idx + 1;
    // Move on to the next worker.
    if (should_continue && (new_idx < max_idx)) {
      // Spawn the next worker.
      process_event_idx_asynch(new_idx, max_idx, ep, pathsDone);
      // And end this one.
      TDEBUG_END_FUNC_SI(4, sid)
        << "new_idx: " << new_idx << " max_idx: " << max_idx;
      return;
    }

    // All done, or filter rejected, or error.
    process_event_pathFinished(new_idx, should_continue, pathsDone);
    // And end the path here.
    TDEBUG_END_FUNC_SI(4, sid) << "idx: " << idx << " max_idx: " << max_idx;
  }

void art::Path::runWorkerTask ( size_t  idx, size_t  max_idx, EventPrincipal &  , hep::concurrency::WaitingTaskPtr  pathsDone  )

private

Definition at line 211 of file Path.cc.

  {
    auto const sid = pc_.scheduleID();
    TDEBUG_BEGIN_TASK_SI(4, sid);
    try {
      process_event_idx(idx, max_idx, ep, pathsDone);
      TDEBUG_END_TASK_SI(4, sid);
    }
    catch (...) {
      taskGroup_.may_run(pathsDone, current_exception());
      TDEBUG_END_TASK_SI(4, sid) << "path terminate because of EXCEPTION";
    }
  }

ScheduleID art::Path::scheduleID

(

)

const

Definition at line 54 of file Path.cc.

  {
    return pc_.scheduleID();
  }

hlt::HLTState art::Path::state

(

)

const

Definition at line 102 of file Path.cc.

  {
    return state_;
  }

size_t art::Path::timesExcept

(

)

const

Definition at line 96 of file Path.cc.

  {
    return timesExcept_;
  }

size_t art::Path::timesFailed

(

)

const

Definition at line 90 of file Path.cc.

  {
    return timesFailed_;
  }

size_t art::Path::timesPassed

(

)

const

Definition at line 84 of file Path.cc.

  {
    return timesPassed_;
  }

size_t art::Path::timesRun

(

)

const

Definition at line 78 of file Path.cc.

  {
    return timesRun_;
  }

vector< WorkerInPath > const & art::Path::workersInPath

(

)

const

Definition at line 108 of file Path.cc.

  {
    return workers_;
  }

Member Data Documentation

ActionTable const& art::Path::actionTable_

private

Definition at line 82 of file Path.h.

ActivityRegistry const& art::Path::actReg_

private

Definition at line 83 of file Path.h.

size_t const art::Path::pathPosition_

private

Definition at line 85 of file Path.h.

PathContext const art::Path::pc_

private

Definition at line 84 of file Path.h.

hlt::HLTState art::Path::state_ {hlt::Ready}

private

Definition at line 95 of file Path.h.

GlobalTaskGroup& art::Path::taskGroup_

private

Definition at line 92 of file Path.h.

std::size_t art::Path::timesExcept_ {}

private

Definition at line 99 of file Path.h.

std::size_t art::Path::timesFailed_ {}

private

Definition at line 98 of file Path.h.

std::size_t art::Path::timesPassed_ {}

private

Definition at line 97 of file Path.h.

std::size_t art::Path::timesRun_ {}

private

Definition at line 96 of file Path.h.

cet::exempt_ptr<HLTGlobalStatus> art::Path::trptr_

private

Definition at line 90 of file Path.h.

std::vector<WorkerInPath> art::Path::workers_

private

Definition at line 87 of file Path.h.

---

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

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