Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Added an early termination signal for Services #5165

Merged
merged 1 commit into from Sep 4, 2014
Merged

Added an early termination signal for Services #5165

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
130 changes: 114 additions & 16 deletions FWCore/Framework/src/EventProcessor.cc
View file
Open in desktop
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
Open in desktop
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
Open in desktop
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
Open in desktop
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>