edm::GlobalSchedule Class Reference

#include <GlobalSchedule.h>

Classes
class	SendTerminationSignalIfException

Public Types
typedef std::vector< Worker * >	AllWorkers
typedef std::vector< std::string >	vstring
typedef std::shared_ptr< Worker >	WorkerPtr
typedef std::vector< Worker * >	Workers

Public Member Functions
AllWorkers const &	allWorkers () const
	returns the collection of pointers to workers More...
void	beginJob (ProductRegistry const &)
void	endJob (ExceptionCollector &collector)
std::vector< ModuleDescription const * >	getAllModuleDescriptions () const
void	getTriggerReport (TriggerReport &rep) const
	GlobalSchedule (std::shared_ptr< TriggerResultInserter > inserter, std::vector< edm::propagate_const< std::shared_ptr< PathStatusInserter >>> &pathStatusInserters, std::vector< edm::propagate_const< std::shared_ptr< EndPathStatusInserter >>> &endPathStatusInserters, std::shared_ptr< ModuleRegistry > modReg, std::vector< std::string > const &modulesToUse, ParameterSet &proc_pset, ProductRegistry &pregistry, PreallocationConfiguration const &prealloc, ExceptionToActionTable const &actions, std::shared_ptr< ActivityRegistry > areg, std::shared_ptr< ProcessConfiguration > processConfiguration, ProcessContext const *processContext)
	GlobalSchedule (GlobalSchedule const &)=delete
template<typename T >
void	processOneGlobalAsync (WaitingTaskHolder holder, typename T::MyPrincipal &principal, EventSetup const &eventSetup, ServiceToken const &token, bool cleaningUpAfterException=false)
void	replaceModule (maker::ModuleHolder *iMod, std::string const &iLabel)
	clone the type of module with label iLabel but configure with iPSet. More...
bool	terminate () const
	Return whether each output module has reached its maximum count. More...

Private Member Functions
ExceptionToActionTable const &	actionTable () const
	returns the action table More...

Private Attributes
std::shared_ptr< ActivityRegistry >	actReg_
std::vector< edm::propagate_const< WorkerPtr > >	extraWorkers_
ProcessContext const *	processContext_
std::vector< WorkerManager >	workerManagers_

Detailed Description

Definition at line 75 of file GlobalSchedule.h.

Member Typedef Documentation

typedef std::vector<Worker*> edm::GlobalSchedule::AllWorkers

Definition at line 78 of file GlobalSchedule.h.

typedef std::vector<std::string> edm::GlobalSchedule::vstring

Definition at line 77 of file GlobalSchedule.h.

typedef std::shared_ptr<Worker> edm::GlobalSchedule::WorkerPtr

Definition at line 79 of file GlobalSchedule.h.

typedef std::vector<Worker*> edm::GlobalSchedule::Workers

Definition at line 80 of file GlobalSchedule.h.

Constructor & Destructor Documentation

edm::GlobalSchedule::GlobalSchedule	(	std::shared_ptr< TriggerResultInserter >	inserter,
		std::vector< edm::propagate_const< std::shared_ptr< PathStatusInserter >>> &	pathStatusInserters,
		std::vector< edm::propagate_const< std::shared_ptr< EndPathStatusInserter >>> &	endPathStatusInserters,
		std::shared_ptr< ModuleRegistry >	modReg,
		std::vector< std::string > const &	modulesToUse,
		ParameterSet &	proc_pset,
		ProductRegistry &	pregistry,
		PreallocationConfiguration const &	prealloc,
		ExceptionToActionTable const &	actions,
		std::shared_ptr< ActivityRegistry >	areg,
		std::shared_ptr< ProcessConfiguration >	processConfiguration,
		ProcessContext const *	processContext
	)

Definition at line 23 of file GlobalSchedule.cc.

References actions, actReg_, extraWorkers_, edm::get_underlying(), edm::ParameterSet::getPSetForUpdate(), mps_fire::i, eostools::move(), edm::PreallocationConfiguration::numberOfLuminosityBlocks(), and workerManagers_.

                                                                       :
    actReg_(areg),
    processContext_(processContext)
  {
    workerManagers_.reserve(prealloc.numberOfLuminosityBlocks());
    for(unsigned int i = 0; i<prealloc.numberOfLuminosityBlocks(); ++i) {
      workerManagers_.emplace_back(modReg,areg,actions);
    }
    for (auto const& moduleLabel : iModulesToUse) {
      bool isTracked;
      ParameterSet* modpset = proc_pset.getPSetForUpdate(moduleLabel, isTracked);
      if (modpset != nullptr) { // It will be null for PathStatusInserters, it should
                                // be impossible to be null for anything else
        assert(isTracked);

        //side effect keeps this module around
        for(auto& wm: workerManagers_) {
          wm.addToAllWorkers(wm.getWorker(*modpset, pregistry, &prealloc,processConfiguration, moduleLabel));
        }
      }
    }
    if(inserter) {
      inserter->doPreallocate(prealloc);
      for(auto& wm: workerManagers_) {
        auto results_inserter = WorkerPtr(new edm::WorkerT<TriggerResultInserter::ModuleType>(inserter, inserter->moduleDescription(), &actions)); // propagate_const<T> has no reset() function
        results_inserter->setActivityRegistry(actReg_);
        wm.addToAllWorkers(results_inserter.get());
        extraWorkers_.emplace_back(std::move(results_inserter));
      }
    }

    for(auto & pathStatusInserter : pathStatusInserters) {
      std::shared_ptr<PathStatusInserter> inserterPtr = get_underlying(pathStatusInserter);
      inserterPtr->doPreallocate(prealloc);
      
      for(auto& wm: workerManagers_) {
        WorkerPtr workerPtr(new edm::WorkerT<PathStatusInserter::ModuleType>(inserterPtr,
                                                                             inserterPtr->moduleDescription(),
                                                                             &actions));
        workerPtr->setActivityRegistry(actReg_);
        wm.addToAllWorkers(workerPtr.get());
        extraWorkers_.emplace_back(std::move(workerPtr));
      }
    }

    for(auto & endPathStatusInserter : endPathStatusInserters) {
      std::shared_ptr<EndPathStatusInserter> inserterPtr = get_underlying(endPathStatusInserter);
      inserterPtr->doPreallocate(prealloc);
      for(auto& wm: workerManagers_) {
        WorkerPtr workerPtr(new edm::WorkerT<EndPathStatusInserter::ModuleType>(inserterPtr,
                                                                                inserterPtr->moduleDescription(),
                                                                                &actions));
        workerPtr->setActivityRegistry(actReg_);
        wm.addToAllWorkers(workerPtr.get());
        extraWorkers_.emplace_back(std::move(workerPtr));
      }
    }

  } // GlobalSchedule::GlobalSchedule

edm::GlobalSchedule::GlobalSchedule ( GlobalSchedule const &  )

delete

Member Function Documentation

ExceptionToActionTable const& edm::GlobalSchedule::actionTable ( ) const

inlineprivate

returns the action table

Definition at line 153 of file GlobalSchedule.h.

                                                      {
      return workerManagers_[0].actionTable();
    }

AllWorkers const& edm::GlobalSchedule::allWorkers ( ) const

inline

returns the collection of pointers to workers

Definition at line 125 of file GlobalSchedule.h.

Referenced by getAllModuleDescriptions().

                                         {
      return workerManagers_[0].allWorkers();
    }

void edm::GlobalSchedule::beginJob

(

ProductRegistry const &

iRegistry

)

Definition at line 98 of file GlobalSchedule.cc.

References workerManagers_.

                                                                {
    workerManagers_[0].beginJob(iRegistry);
  }

void edm::GlobalSchedule::endJob

(

ExceptionCollector &

collector

)

Definition at line 94 of file GlobalSchedule.cc.

References workerManagers_.

                                                            {
    workerManagers_[0].endJob(collector);
  }

std::vector< ModuleDescription const * > edm::GlobalSchedule::getAllModuleDescriptions

(

)

const

Return a vector allowing const access to all the ModuleDescriptions for this GlobalSchedule. *** N.B. *** Ownership of the ModuleDescriptions is not *** passed to the caller. Do not call delete on these *** pointers!

Definition at line 122 of file GlobalSchedule.cc.

References allWorkers(), AlCaHLTBitMon_ParallelJobs::p, mps_fire::result, and findQualityFiles::size.

                                                 {
    std::vector<ModuleDescription const*> result;
    result.reserve(allWorkers().size());

    for (auto const& worker : allWorkers()) {
      ModuleDescription const* p = worker->descPtr();
      result.push_back(p);
    }
    return result;
  }

void edm::GlobalSchedule::getTriggerReport

(

TriggerReport &

rep

)

const

Return the trigger report information on paths, modules-in-path, modules-in-endpath, and modules.

template<typename T >

void edm::GlobalSchedule::processOneGlobalAsync	(	WaitingTaskHolder	holder,
		typename T::MyPrincipal &	principal,
		EventSetup const &	eventSetup,
		ServiceToken const &	token,
		bool	cleaningUpAfterException = false
	)

Definition at line 166 of file GlobalSchedule.h.

References edm::addContextAndPrintException(), lumiQTWidget::aw, cms::Exception::context(), edm::WaitingTaskHolder::doneWaiting(), edm::ExceptionFromThisContext, edm::StreamID::invalidStreamID(), edm::make_waiting_task(), groupFilesInBlocks::reverse, and edm::convertException::wrap().

                                                                       {
    try {
      //need the doneTask to own the memory
      auto globalContext = std::make_shared<GlobalContext>(T::makeGlobalContext(ep, processContext_));
      
      if(actReg_) {
        //Services may depend upon each other
        ServiceRegistry::Operate op(token);
        T::preScheduleSignal(actReg_.get(), globalContext.get());
      }
      
      auto doneTask = make_waiting_task(tbb::task::allocate_root(),
                                        [this,iHolder, cleaningUpAfterException, globalContext, token](std::exception_ptr const* iPtr) mutable
                                        {
                                          std::exception_ptr excpt;
                                          if(iPtr) {
                                            excpt = *iPtr;
                                            //add context information to the exception and print message
                                            try {
                                              convertException::wrap([&]() {
                                                std::rethrow_exception(excpt);
                                              });
                                            } catch(cms::Exception& ex) {
                                              //TODO: should add the transition type info
                                              std::ostringstream ost;
                                              if(ex.context().empty()) {
                                                ost<<"Processing "<<T::transitionName()<<" ";
                                              }
                                              ServiceRegistry::Operate op(token);
                                              addContextAndPrintException(ost.str().c_str(), ex, cleaningUpAfterException);
                                              excpt = std::current_exception();
                                            }
                                            if(actReg_) {
                                              ServiceRegistry::Operate op(token);
                                              actReg_->preGlobalEarlyTerminationSignal_(*globalContext,TerminationOrigin::ExceptionFromThisContext);
                                            }
                                          }
                                          if(actReg_) {
                                            try {
                                              ServiceRegistry::Operate op(token);
                                              T::postScheduleSignal(actReg_.get(), globalContext.get());
                                            } catch(...) {
                                              if(not excpt) {
                                                excpt = std::current_exception();
                                              }
                                            }
                                          }
                                          iHolder.doneWaiting(excpt);
                                          
                                        });
      workerManagers_[ep.index()].resetAll();
      
      ParentContext parentContext(globalContext.get());
      //make sure the ProductResolvers know about their
      // workers to allow proper data dependency handling
      workerManagers_[ep.index()].setupOnDemandSystem(ep,es);
      
      //make sure the task doesn't get run until all workers have beens started
      WaitingTaskHolder holdForLoop(doneTask);
      auto& aw = workerManagers_[ep.index()].allWorkers();
      for(Worker* worker: boost::adaptors::reverse(aw) ) {
        worker->doWorkAsync<T>(doneTask,ep,es,token, StreamID::invalidStreamID(),parentContext,globalContext.get());
      }
    } catch(...) {
      iHolder.doneWaiting(std::current_exception());
    }
  }

void edm::GlobalSchedule::replaceModule	(	maker::ModuleHolder *	iMod,
		std::string const &	iLabel
	)

clone the type of module with label iLabel but configure with iPSet.

Definition at line 102 of file GlobalSchedule.cc.

References edm::Worker::beginJob(), runEdmFileComparison::found, edm::maker::ModuleHolder::replaceModuleFor(), and workerManagers_.

                                                              {
    Worker* found = nullptr;
    for(auto& wm: workerManagers_) {
      for (auto const& worker : wm.allWorkers()) {
        if (worker->description().moduleLabel() == iLabel) {
          found = worker;
          break;
        }
      }
      if (nullptr == found) {
        return;
      }
      
      iMod->replaceModuleFor(found);
      found->beginJob();
    }
  }

bool edm::GlobalSchedule::terminate

(

)

const

Return whether each output module has reached its maximum count.

Member Data Documentation

std::shared_ptr<ActivityRegistry> edm::GlobalSchedule::actReg_

private

Definition at line 158 of file GlobalSchedule.h.

Referenced by GlobalSchedule().

std::vector<edm::propagate_const<WorkerPtr> > edm::GlobalSchedule::extraWorkers_

private

Definition at line 159 of file GlobalSchedule.h.

Referenced by GlobalSchedule().

ProcessContext const* edm::GlobalSchedule::processContext_

private

Definition at line 160 of file GlobalSchedule.h.

std::vector<WorkerManager> edm::GlobalSchedule::workerManagers_

private

Definition at line 157 of file GlobalSchedule.h.

Referenced by beginJob(), endJob(), GlobalSchedule(), and replaceModule().