Skip to content

Commit

Permalink
Added an early termination signal for Services
Browse files Browse the repository at this point in the history 
Services can now listen for a signal which is sent whenever a job is ending because of an early termination. The early termination can be caused by either an external signal or a fatal exception. The signal gives the context, if known, under which the termination was causes. The context includes such information as the run, lumi and event number for the transition which either saw the termination or which is being terminated because another transition caused the termination.
  • Loading branch information
@Dr15Jones
Dr15Jones committed Sep 3, 2014
1 parent a414435 commit 0478509
Show file tree
Hide file tree
Showing 10 changed files with 302 additions and 16 deletions.
130 changes: 114 additions & 16 deletions FWCore/Framework/src/EventProcessor.cc
View file
Expand Up @@ -89,6 +89,30 @@
//Needed for introspection
#include "Cintex/Cintex.h"


namespace {
//Sentry class to only send a signal if an
// exception occurs. An exception is identified
// by the destructor being called without first
// calling completedSuccessfully().
class SendSourceTerminationSignalIfException {
public:
SendSourceTerminationSignalIfException(edm::ActivityRegistry* iReg):
reg_(iReg) {}
~SendSourceTerminationSignalIfException() {
if(reg_) {
reg_->preSourceEarlyTerminationSignal_(edm::TerminationOrigin::ExceptionFromThisContext);
}
}
void completedSuccessfully() {
reg_ = nullptr;
}
private:
edm::ActivityRegistry* reg_;

};
}

namespace edm {

// ---------------------------------------------------------------
Expand Down Expand Up @@ -1270,19 +1294,28 @@ namespace edm {
bool more = true;
if(numberOfForkedChildren_ > 0) {
size_t size = preg_->size();
more = input_->skipForForking();
{
SendSourceTerminationSignalIfException sentry(actReg_.get());
more = input_->skipForForking();
sentry.completedSuccessfully();
}
if(more) {
if(size < preg_->size()) {
principalCache_.adjustIndexesAfterProductRegistryAddition();
}
principalCache_.adjustEventsToNewProductRegistry(preg_);
}
}
itemType = (more ? input_->nextItemType() : InputSource::IsStop);
}
{
SendSourceTerminationSignalIfException sentry(actReg_.get());
itemType = (more ? input_->nextItemType() : InputSource::IsStop);
sentry.completedSuccessfully();
}

FDEBUG(1) << "itemType = " << itemType << "\n";

if(checkForAsyncStopRequest(returnCode)) {
actReg_->preSourceEarlyTerminationSignal_(TerminationOrigin::ExternalSignal);
forceLooperToEnd_ = true;
machine->process_event(statemachine::Stop());
forceLooperToEnd_ = false;
Expand Down Expand Up @@ -1425,6 +1458,8 @@ namespace edm {
void EventProcessor::readFile() {
FDEBUG(1) << " \treadFile\n";
size_t size = preg_->size();
SendSourceTerminationSignalIfException sentry(actReg_.get());

fb_ = input_->readFile();
if(size < preg_->size()) {
principalCache_.adjustIndexesAfterProductRegistryAddition();
Expand All @@ -1435,11 +1470,14 @@ namespace edm {
preallocations_.numberOfThreads()>1)) {
fb_->setNotFastClonable(FileBlock::ParallelProcesses);
}
sentry.completedSuccessfully();
}

void EventProcessor::closeInputFile(bool cleaningUpAfterException) {
if (fb_.get() != nullptr) {
SendSourceTerminationSignalIfException sentry(actReg_.get());
input_->closeFile(fb_.get(), cleaningUpAfterException);
sentry.completedSuccessfully();
}
FDEBUG(1) << "\tcloseInputFile\n";
}
Expand Down Expand Up @@ -1531,13 +1569,23 @@ namespace edm {

void EventProcessor::beginRun(statemachine::Run const& run) {
RunPrincipal& runPrincipal = principalCache_.runPrincipal(run.processHistoryID(), run.runNumber());
input_->doBeginRun(runPrincipal, &processContext_);
{
SendSourceTerminationSignalIfException sentry(actReg_.get());

input_->doBeginRun(runPrincipal, &processContext_);
sentry.completedSuccessfully();
}

IOVSyncValue ts(EventID(runPrincipal.run(), 0, 0),
runPrincipal.beginTime());
if(forceESCacheClearOnNewRun_){
espController_->forceCacheClear();
}
espController_->eventSetupForInstance(ts);
{
SendSourceTerminationSignalIfException sentry(actReg_.get());
espController_->eventSetupForInstance(ts);
sentry.completedSuccessfully();
}
EventSetup const& es = esp_->eventSetup();
if(looper_ && looperBeginJobRun_== false) {
looper_->copyInfo(ScheduleInfo(schedule_.get()));
Expand Down Expand Up @@ -1573,10 +1621,20 @@ namespace edm {

void EventProcessor::endRun(statemachine::Run const& run, bool cleaningUpAfterException) {
RunPrincipal& runPrincipal = principalCache_.runPrincipal(run.processHistoryID(), run.runNumber());
input_->doEndRun(runPrincipal, cleaningUpAfterException, &processContext_);
{
SendSourceTerminationSignalIfException sentry(actReg_.get());

input_->doEndRun(runPrincipal, cleaningUpAfterException, &processContext_);
sentry.completedSuccessfully();
}

IOVSyncValue ts(EventID(runPrincipal.run(), LuminosityBlockID::maxLuminosityBlockNumber(), EventID::maxEventNumber()),
runPrincipal.endTime());
espController_->eventSetupForInstance(ts);
{
SendSourceTerminationSignalIfException sentry(actReg_.get());
espController_->eventSetupForInstance(ts);
sentry.completedSuccessfully();
}
EventSetup const& es = esp_->eventSetup();
{
for(unsigned int i=0; i<preallocations_.numberOfStreams();++i) {
Expand Down Expand Up @@ -1606,7 +1664,12 @@ namespace edm {

void EventProcessor::beginLumi(ProcessHistoryID const& phid, RunNumber_t run, LuminosityBlockNumber_t lumi) {
LuminosityBlockPrincipal& lumiPrincipal = principalCache_.lumiPrincipal(phid, run, lumi);
input_->doBeginLumi(lumiPrincipal, &processContext_);
{
SendSourceTerminationSignalIfException sentry(actReg_.get());

input_->doBeginLumi(lumiPrincipal, &processContext_);
sentry.completedSuccessfully();
}

Service<RandomNumberGenerator> rng;
if(rng.isAvailable()) {
Expand All @@ -1617,7 +1680,11 @@ namespace edm {
// NOTE: Using 0 as the event number for the begin of a lumi block is a bad idea
// lumi blocks know their start and end times why not also start and end events?
IOVSyncValue ts(EventID(lumiPrincipal.run(), lumiPrincipal.luminosityBlock(), 0), lumiPrincipal.beginTime());
espController_->eventSetupForInstance(ts);
{
SendSourceTerminationSignalIfException sentry(actReg_.get());
espController_->eventSetupForInstance(ts);
sentry.completedSuccessfully();
}
EventSetup const& es = esp_->eventSetup();
{
typedef OccurrenceTraits<LuminosityBlockPrincipal, BranchActionGlobalBegin> Traits;
Expand Down Expand Up @@ -1647,12 +1714,21 @@ namespace edm {

void EventProcessor::endLumi(ProcessHistoryID const& phid, RunNumber_t run, LuminosityBlockNumber_t lumi, bool cleaningUpAfterException) {
LuminosityBlockPrincipal& lumiPrincipal = principalCache_.lumiPrincipal(phid, run, lumi);
input_->doEndLumi(lumiPrincipal, cleaningUpAfterException, &processContext_);
{
SendSourceTerminationSignalIfException sentry(actReg_.get());

input_->doEndLumi(lumiPrincipal, cleaningUpAfterException, &processContext_);
sentry.completedSuccessfully();
}
//NOTE: Using the max event number for the end of a lumi block is a bad idea
// lumi blocks know their start and end times why not also start and end events?
IOVSyncValue ts(EventID(lumiPrincipal.run(), lumiPrincipal.luminosityBlock(), EventID::maxEventNumber()),
lumiPrincipal.endTime());
espController_->eventSetupForInstance(ts);
{
SendSourceTerminationSignalIfException sentry(actReg_.get());
espController_->eventSetupForInstance(ts);
sentry.completedSuccessfully();
}
EventSetup const& es = esp_->eventSetup();
{
for(unsigned int i=0; i<preallocations_.numberOfStreams();++i) {
Expand Down Expand Up @@ -1688,7 +1764,11 @@ namespace edm {
<< "Contact a Framework Developer\n";
}
auto rp = std::make_shared<RunPrincipal>(input_->runAuxiliary(), preg_, *processConfiguration_, historyAppender_.get(), 0);
input_->readRun(*rp, *historyAppender_);
{
SendSourceTerminationSignalIfException sentry(actReg_.get());
input_->readRun(*rp, *historyAppender_);
sentry.completedSuccessfully();
}
assert(input_->reducedProcessHistoryID() == rp->reducedProcessHistoryID());
principalCache_.insert(rp);
return statemachine::Run(rp->reducedProcessHistoryID(), input_->run());
Expand All @@ -1697,7 +1777,11 @@ namespace edm {
statemachine::Run EventProcessor::readAndMergeRun() {
principalCache_.merge(input_->runAuxiliary(), preg_);
auto runPrincipal =principalCache_.runPrincipalPtr();
input_->readAndMergeRun(*runPrincipal);
{
SendSourceTerminationSignalIfException sentry(actReg_.get());
input_->readAndMergeRun(*runPrincipal);
sentry.completedSuccessfully();
}
assert(input_->reducedProcessHistoryID() == runPrincipal->reducedProcessHistoryID());
return statemachine::Run(runPrincipal->reducedProcessHistoryID(), input_->run());
}
Expand All @@ -1717,15 +1801,23 @@ namespace edm {
<< "Contact a Framework Developer\n";
}
auto lbp = std::make_shared<LuminosityBlockPrincipal>(input_->luminosityBlockAuxiliary(), preg_, *processConfiguration_, historyAppender_.get(), 0);
input_->readLuminosityBlock(*lbp, *historyAppender_);
{
SendSourceTerminationSignalIfException sentry(actReg_.get());
input_->readLuminosityBlock(*lbp, *historyAppender_);
sentry.completedSuccessfully();
}
lbp->setRunPrincipal(principalCache_.runPrincipalPtr());
principalCache_.insert(lbp);
return input_->luminosityBlock();
}

int EventProcessor::readAndMergeLumi() {
principalCache_.merge(input_->luminosityBlockAuxiliary(), preg_);
input_->readAndMergeLumi(*principalCache_.lumiPrincipalPtr());
{
SendSourceTerminationSignalIfException sentry(actReg_.get());
input_->readAndMergeLumi(*principalCache_.lumiPrincipalPtr());
sentry.completedSuccessfully();
}
return input_->luminosityBlock();
}

Expand Down Expand Up @@ -1816,7 +1908,6 @@ namespace edm {
if(sr) {
delayedReaderGuard = std::unique_lock<SharedResourcesAcquirer>(*sr);
}

InputSource::ItemType itemType = input_->nextItemType();
if (InputSource::IsEvent !=itemType) {
nextItemTypeFromProcessingEvents_ = itemType;
Expand All @@ -1826,6 +1917,7 @@ namespace edm {
}
if((asyncStopRequestedWhileProcessingEvents_=checkForAsyncStopRequest(asyncStopStatusCodeFromProcessingEvents_))) {
//std::cerr<<"task told to async stop\n";
actReg_->preSourceEarlyTerminationSignal_(TerminationOrigin::ExternalSignal);
break;
}
readEvent(iStreamIndex);
Expand All @@ -1834,6 +1926,8 @@ namespace edm {
if(deferredExceptionPtrIsSet_.load(std::memory_order_acquire)) {
//another thread hit an exception
//std::cerr<<"another thread saw an exception\n";
actReg_->preSourceEarlyTerminationSignal_(TerminationOrigin::ExceptionFromAnotherContext);

break;
}
processEvent(iStreamIndex);
Expand Down Expand Up @@ -1885,7 +1979,11 @@ namespace edm {
//TODO this will have to become per stream
auto& event = principalCache_.eventPrincipal(iStreamIndex);
StreamContext streamContext(event.streamID(), &processContext_);

SendSourceTerminationSignalIfException sentry(actReg_.get());
input_->readEvent(event, streamContext);
sentry.completedSuccessfully();

FDEBUG(1) << "\treadEvent\n";
}
void EventProcessor::processEvent(unsigned int iStreamIndex) {
Expand Down
26 changes: 26 additions & 0 deletions FWCore/Framework/src/GlobalSchedule.h
View file
Expand Up @@ -117,6 +117,28 @@ namespace edm {
}

private:
//Sentry class to only send a signal if an
// exception occurs. An exception is identified
// by the destructor being called without first
// calling completedSuccessfully().
class SendTerminationSignalIfException {
public:
SendTerminationSignalIfException(edm::ActivityRegistry* iReg, edm::GlobalContext const* iContext):
reg_(iReg),
context_(iContext){}
~SendTerminationSignalIfException() {
if(reg_) {
reg_->preGlobalEarlyTerminationSignal_(*context_,TerminationOrigin::ExceptionFromThisContext);
}
}
void completedSuccessfully() {
reg_ = nullptr;
}
private:
edm::ActivityRegistry* reg_;
GlobalContext const* context_;
};


template<typename T>
void runNow(typename T::MyPrincipal& p, EventSetup const& es,
Expand Down Expand Up @@ -146,6 +168,8 @@ namespace edm {
GlobalContext globalContext = T::makeGlobalContext(ep, processContext_);

GlobalScheduleSignalSentry<T> sentry(actReg_.get(), &globalContext);

SendTerminationSignalIfException terminationSentry(actReg_.get(), &globalContext);

// This call takes care of the unscheduled processing.
workerManager_.processOneOccurrence<T>(ep, es, StreamID::invalidStreamID(), &globalContext, &globalContext, cleaningUpAfterException);
Expand All @@ -163,6 +187,8 @@ namespace edm {
}
throw;
}
terminationSentry.completedSuccessfully();

//If we got here no other exception has happened so we can propogate any Service related exceptions
sentry.allowThrow();
}
Expand Down
29 changes: 29 additions & 0 deletions FWCore/Framework/src/StreamSchedule.h
View file
Expand Up @@ -242,6 +242,28 @@ namespace edm {
}

private:
//Sentry class to only send a signal if an
// exception occurs. An exception is identified
// by the destructor being called without first
// calling completedSuccessfully().
class SendTerminationSignalIfException {
public:
SendTerminationSignalIfException(edm::ActivityRegistry* iReg, edm::StreamContext const* iContext):
reg_(iReg),
context_(iContext){}
~SendTerminationSignalIfException() {
if(reg_) {
reg_->preStreamEarlyTerminationSignal_(*context_,TerminationOrigin::ExceptionFromThisContext);
}
}
void completedSuccessfully() {
reg_ = nullptr;
}
private:
edm::ActivityRegistry* reg_;
StreamContext const* context_;
};

/// returns the action table
ExceptionToActionTable const& actionTable() const {
return workerManager_.actionTable();
Expand Down Expand Up @@ -341,6 +363,7 @@ namespace edm {
T::setStreamContext(streamContext_, ep);
StreamScheduleSignalSentry<T> sentry(actReg_.get(), &streamContext_);

SendTerminationSignalIfException terminationSentry(actReg_.get(), &streamContext_);
// This call takes care of the unscheduled processing.
workerManager_.processOneOccurrence<T>(ep, es, streamID_, &streamContext_, &streamContext_, cleaningUpAfterException);

Expand Down Expand Up @@ -389,6 +412,8 @@ namespace edm {
}
throw;
}
terminationSentry.completedSuccessfully();

//If we got here no other exception has happened so we can propogate any Service related exceptions
sentry.allowThrow();
}
Expand All @@ -402,6 +427,8 @@ namespace edm {
T::setStreamContext(streamContext_, ep);
StreamScheduleSignalSentry<T> sentry(actReg_.get(), &streamContext_);

SendTerminationSignalIfException terminationSentry(actReg_.get(), &streamContext_);

// This call takes care of the unscheduled processing.
workerManager_.processOneOccurrence<T>(ep, es, streamID_, &streamContext_, &streamContext_, cleaningUpAfterException);

Expand All @@ -420,6 +447,8 @@ namespace edm {
}
throw;
}
terminationSentry.completedSuccessfully();

//If we got here no other exception has happened so we can propogate any Service related exceptions
sentry.allowThrow();
}
Expand Down
4 changes: 4 additions & 0 deletions FWCore/Framework/test/BuildFile.xml
View file
Expand Up @@ -301,3 +301,7 @@
<flags TEST_RUNNER_ARGS=" /bin/bash FWCore/Framework/test test_deleteEarly.sh"/>
<use name="FWCore/Utilities"/>
</bin>
<bin name="TestFWCoreFrameworkEarlyTerminationSignal" file="TestDriver.cpp">
<flags TEST_RUNNER_ARGS=" /bin/bash FWCore/Framework/test test_earlyTerminationSignal.sh"/>
<use name="FWCore/Utilities"/>
</bin>

0 comments on commit 0478509

Please sign in to comment.