CMSSW Reference Manual

00001 #ifndef FWCore_Framework_Worker_h
00002 #define FWCore_Framework_Worker_h
00003 
00004 /*----------------------------------------------------------------------
00005   
00006 Worker: this is a basic scheduling unit - an abstract base class to
00007 something that is really a producer or filter.
00008 
00009 $Id: Worker.h,v 1.35 2008/10/16 23:06:28 wmtan Exp $
00010 
00011 A worker will not actually call through to the module unless it is
00012 in a Ready state.  After a module is actually run, the state will not
00013 be Ready.  The Ready state can only be reestablished by doing a reset().
00014 
00015 Pre/post module signals are posted only in the Ready state.
00016 
00017 Execution statistics are kept here.
00018 
00019 If a module has thrown an exception during execution, that exception
00020 will be rethrown if the worker is entered again and the state is not Ready.
00021 In other words, execution results (status) are cached and reused until
00022 the worker is reset().
00023 
00024 ----------------------------------------------------------------------*/
00025 
00026 #include "DataFormats/Provenance/interface/ModuleDescription.h"
00027 #include "FWCore/MessageLogger/interface/MessageLogger.h"
00028 #include "FWCore/Framework/src/WorkerParams.h"
00029 #include "FWCore/Framework/interface/Actions.h"
00030 #include "FWCore/Framework/interface/BranchActionType.h"
00031 #include "FWCore/Framework/interface/CurrentProcessingContext.h"
00032 #include "FWCore/Framework/interface/OccurrenceTraits.h"
00033 #include "FWCore/ServiceRegistry/interface/ActivityRegistry.h"
00034 #include "FWCore/Utilities/interface/Exception.h"
00035 
00036 #include "boost/shared_ptr.hpp"
00037 
00038 #include "FWCore/Framework/src/RunStopwatch.h"
00039 #include "FWCore/Framework/interface/Frameworkfwd.h"
00040 
00041 namespace edm {
00042 
00043   class Worker {
00044   public:
00045     enum State { Ready, Pass, Fail, Exception };
00046 
00047     Worker(ModuleDescription const& iMD, WorkerParams const& iWP);
00048     virtual ~Worker();
00049 
00050     template <typename T>
00051     bool doWork(typename T::MyPrincipal&, EventSetup const& c,
00052                 CurrentProcessingContext const* cpc);
00053     void beginJob(EventSetup const&) ;
00054     void endJob();
00055     void respondToOpenInputFile(FileBlock const& fb) {implRespondToOpenInputFile(fb);}
00056     void respondToCloseInputFile(FileBlock const& fb) {implRespondToCloseInputFile(fb);}
00057     void respondToOpenOutputFiles(FileBlock const& fb) {implRespondToOpenOutputFiles(fb);}
00058     void respondToCloseOutputFiles(FileBlock const& fb) {implRespondToCloseOutputFiles(fb);}
00059 
00060     void reset() { state_ = Ready; }
00061     
00062     ModuleDescription const& description() const {return md_;}
00063     ModuleDescription const* descPtr() const {return &md_; }
00066     void setActivityRegistry(boost::shared_ptr<ActivityRegistry> areg);
00067 
00068     std::pair<double,double> timeCpuReal() const {
00069       return std::pair<double,double>(stopwatch_->cpuTime(),stopwatch_->realTime());
00070     }
00071 
00072     void clearCounters() {
00073       timesRun_ = timesVisited_ = timesPassed_ = timesFailed_ = timesExcept_ = 0;
00074     }
00075 
00076     int timesRun() const { return timesRun_; }
00077     int timesVisited() const { return timesVisited_; }
00078     int timesPassed() const { return timesPassed_; }
00079     int timesFailed() const { return timesFailed_; }
00080     int timesExcept() const { return timesExcept_; }
00081     State state() const { return state_; }
00082    
00083     int timesPass() const { return timesPassed(); } // for backward compatibility only - to be removed soon
00084 
00085   protected:
00086     virtual std::string workerType() const = 0;
00087     virtual bool implDoBegin(EventPrincipal&, EventSetup const& c, 
00088                             CurrentProcessingContext const* cpc) = 0;
00089     virtual bool implDoEnd(EventPrincipal&, EventSetup const& c, 
00090                             CurrentProcessingContext const* cpc) = 0;
00091     virtual bool implDoBegin(RunPrincipal& rp, EventSetup const& c,
00092                             CurrentProcessingContext const* cpc) = 0;
00093     virtual bool implDoEnd(RunPrincipal& rp, EventSetup const& c,
00094                             CurrentProcessingContext const* cpc) = 0;
00095     virtual bool implDoBegin(LuminosityBlockPrincipal& lbp, EventSetup const& c,
00096                             CurrentProcessingContext const* cpc) = 0;
00097     virtual bool implDoEnd(LuminosityBlockPrincipal& lbp, EventSetup const& c,
00098                             CurrentProcessingContext const* cpc) = 0;
00099     virtual void implBeginJob(EventSetup const&) = 0;
00100     virtual void implEndJob() = 0;
00101 
00102   private:
00103     virtual void implRespondToOpenInputFile(FileBlock const& fb) = 0;
00104     virtual void implRespondToCloseInputFile(FileBlock const& fb) = 0;
00105     virtual void implRespondToOpenOutputFiles(FileBlock const& fb) = 0;
00106     virtual void implRespondToCloseOutputFiles(FileBlock const& fb) = 0;
00107 
00108     RunStopwatch::StopwatchPointer stopwatch_;
00109 
00110     int timesRun_;
00111     int timesVisited_;
00112     int timesPassed_;
00113     int timesFailed_;
00114     int timesExcept_;
00115     State state_;
00116 
00117     ModuleDescription md_;
00118     ActionTable const* actions_; // memory assumed to be managed elsewhere
00119     boost::shared_ptr<edm::Exception> cached_exception_; // if state is 'exception'
00120 
00121     boost::shared_ptr<ActivityRegistry> actReg_;
00122   };
00123 
00124   namespace {
00125     template <typename T>
00126     class ModuleSignalSentry {
00127     public:
00128       ModuleSignalSentry(ActivityRegistry *a, ModuleDescription& md) : a_(a), md_(&md) {
00129         if(a_) T::preModuleSignal(a_, md_);
00130       }
00131       ~ModuleSignalSentry() {
00132         if(a_) T::postModuleSignal(a_, md_);
00133       }
00134     private:
00135       ActivityRegistry* a_;
00136       ModuleDescription* md_;
00137     };
00138 
00139     template <typename T>
00140     cms::Exception& exceptionContext(ModuleDescription const& iMD,
00141                                      T const& ip,
00142                                      cms::Exception& iEx) {
00143       iEx << iMD.moduleName_ << "/" << iMD.moduleLabel_ 
00144         << " " << ip.id() << "\n";
00145       return iEx;
00146     }
00147   }
00148 
00149    template <typename T>
00150    bool Worker::doWork(typename T::MyPrincipal& ep, EventSetup const& es,
00151                       CurrentProcessingContext const* cpc) {
00152 
00153     // A RunStopwatch, but only if we are processing an event.
00154     std::auto_ptr<RunStopwatch> stopwatch(T::isEvent_ ? new RunStopwatch(stopwatch_) : 0);
00155 
00156     if (T::isEvent_) {
00157       ++timesVisited_;
00158     }
00159     bool rc = false;
00160 
00161     switch(state_) {
00162       case Ready: break;
00163       case Pass: return true;
00164       case Fail: return false;
00165       case Exception: {
00166           // rethrow the cached exception again
00167           // It seems impossible to
00168           // get here a second time until a cms::Exception has been 
00169           // thrown prviously.
00170           LogWarning("repeat") << "A module has been invoked a second "
00171                                << "time even though it caught an "
00172                                << "exception during the previous "
00173                                << "invocation.\n"
00174                                << "This may be an indication of a "
00175                                << "configuration problem.\n";
00176 
00177           throw *cached_exception_;
00178       }
00179     }
00180 
00181     if (T::isEvent_) ++timesRun_;
00182 
00183     try {
00184 
00185         ModuleSignalSentry<T> cpp(actReg_.get(), md_);
00186         if (T::begin_) {
00187           rc = implDoBegin(ep, es, cpc);
00188         } else {
00189           rc = implDoEnd(ep, es, cpc);
00190         }
00191 
00192         if (rc) {
00193           state_ = Pass;
00194           if (T::isEvent_) ++timesPassed_;
00195         } else {
00196           state_ = Fail;
00197           if (T::isEvent_) ++timesFailed_;
00198         }
00199     }
00200 
00201     catch(cms::Exception& e) {
00202       
00203         // NOTE: the warning printed as a result of ignoring or failing
00204         // a module will only be printed during the full true processing
00205         // pass of this module
00206 
00207         // Get the action corresponding to this exception.  However, if processing
00208         // something other than an event (e.g. run, lumi) always rethrow.
00209         actions::ActionCodes action = (T::isEvent_ ? actions_->find(e.rootCause()) : actions::Rethrow);
00210 
00211         // If we are processing an endpath, treat SkipEvent or FailPath
00212         // as FailModule, so any subsequent OutputModules are still run.
00213         if (cpc && cpc->isEndPath()) {
00214           if (action == actions::SkipEvent || action == actions::FailPath) action = actions::FailModule;
00215         }
00216         switch(action) {
00217           case actions::IgnoreCompletely: {
00218               rc=true;
00219               ++timesPassed_;
00220               state_ = Pass;
00221               LogWarning("IgnoreCompletely")
00222                 << "Module ignored an exception\n"
00223                 <<e.what()<<"\n";
00224               break;
00225           }
00226 
00227           case actions::FailModule: {
00228               rc=true;
00229               LogWarning("FailModule")
00230                 << "Module failed due to an exception\n"
00231                 << e.what() << "\n";
00232               ++timesFailed_;
00233               state_ = Fail;
00234               break;
00235           }
00236             
00237           default: {
00238 
00239               // we should not need to include the event/run/module names
00240               // the exception because the error logger will pick this
00241               // up automatically.  I'm leaving it in until this is 
00242               // verified
00243 
00244               // here we simply add a small amount of data to the
00245               // exception to add some context, we could have rethrown
00246               // it as something else and embedded with this exception
00247               // as an argument to the constructor.
00248 
00249               if (T::isEvent_) ++timesExcept_;
00250               state_ = Exception;
00251               e << "cms::Exception going through module ";
00252               exceptionContext(md_, ep, e);
00253               edm::Exception *edmEx = dynamic_cast<edm::Exception *>(&e);
00254               if (edmEx) {
00255                 cached_exception_.reset(new edm::Exception(*edmEx));
00256               } else {
00257                 cached_exception_.reset(new edm::Exception(errors::OtherCMS, std::string(), e));
00258               }
00259               throw;
00260           }
00261         }
00262       }
00263     
00264     catch(std::bad_alloc& bda) {
00265         if (T::isEvent_) ++timesExcept_;
00266         state_ = Exception;
00267         cached_exception_.reset(new edm::Exception(errors::BadAlloc));
00268         *cached_exception_
00269           << "A std::bad_alloc exception occurred during a call to the module ";
00270         exceptionContext(md_, ep, *cached_exception_)
00271           << "The job has probably exhausted the virtual memory available to the process.\n";
00272         throw *cached_exception_;
00273     }
00274     catch(std::exception& e) {
00275         if (T::isEvent_) ++timesExcept_;
00276         state_ = Exception;
00277         cached_exception_.reset(new edm::Exception(errors::StdException));
00278         *cached_exception_
00279           << "A std::exception occurred during a call to the module ";
00280         exceptionContext(md_, ep, *cached_exception_) << "and cannot be repropagated.\n"
00281           << "Previous information:\n" << e.what();
00282         throw *cached_exception_;
00283     }
00284     catch(std::string& s) {
00285         if (T::isEvent_) ++timesExcept_;
00286         state_ = Exception;
00287         cached_exception_.reset(new edm::Exception(errors::BadExceptionType, "std::string"));
00288         *cached_exception_
00289           << "A std::string thrown as an exception occurred during a call to the module ";
00290         exceptionContext(md_, ep, *cached_exception_) << "and cannot be repropagated.\n"
00291           << "Previous information:\n string = " << s;
00292         throw *cached_exception_;
00293     }
00294     catch(char const* c) {
00295         if (T::isEvent_) ++timesExcept_;
00296         state_ = Exception;
00297         cached_exception_.reset(new edm::Exception(errors::BadExceptionType, "const char *"));
00298         *cached_exception_
00299           << "A const char* thrown as an exception occurred during a call to the module ";
00300         exceptionContext(md_, ep, *cached_exception_) << "and cannot be repropagated.\n"
00301           << "Previous information:\n const char* = " << c << "\n";
00302         throw *cached_exception_;
00303     }
00304     catch(...) {
00305         if (T::isEvent_) ++timesExcept_;
00306         state_ = Exception;
00307         cached_exception_.reset(new edm::Exception(errors::Unknown, "repeated"));
00308         *cached_exception_
00309           << "An unknown occurred during a previous call to the module ";
00310         exceptionContext(md_, ep, *cached_exception_) << "and cannot be repropagated.\n";
00311         throw *cached_exception_;
00312     }
00313 
00314     return rc;
00315   }
00316 
00317 }
00318 #endif