Framework/interface/Path.h

0001 #ifndef FWCore_Framework_Path_h
0002 #define FWCore_Framework_Path_h
0003
0004 /*
0005   Author: Jim Kowalkowski 28-01-06
0006
0007   An object of this type represents one path in a job configuration.
0008   It holds the assigned bit position and the list of workers that are
0009   an event must pass through when this parh is processed.  The workers
0010   are held in WorkerInPath wrappers so that per path execution statistics
0011   can be kept for each worker.
0012 */
0013
0014 #include "FWCore/Framework/interface/WorkerInPath.h"
0015 #include "FWCore/Framework/interface/maker/Worker.h"
0016 #include "DataFormats/Common/interface/HLTenums.h"
0017 #include "DataFormats/Common/interface/TriggerResults.h"
0018 #include "FWCore/ServiceRegistry/interface/PathContext.h"
0019 #include "FWCore/Concurrency/interface/WaitingTaskHolder.h"
0020 #include "FWCore/Utilities/interface/BranchType.h"
0021 #include "FWCore/Utilities/interface/Exception.h"
0022 #include "FWCore/Utilities/interface/ConvertException.h"
0023 #include "FWCore/Utilities/interface/make_sentry.h"
0024
0025 #include <memory>
0026
0027 #include <string>
0028 #include <vector>
0029 #include <map>
0030 #include <exception>
0031 #include <sstream>
0032
0033 namespace edm {
0034   class EventTransitionInfo;
0035   class ModuleDescription;
0036   class PathStatusInserter;
0037   class EarlyDeleteHelper;
0038   class StreamContext;
0039   class StreamID;
0040
0041   class Path {
0042   public:
0043     typedef hlt::HLTState State;
0044
0045     typedef std::vector<WorkerInPath> WorkersInPath;
0046     typedef WorkersInPath::size_type size_type;
0047     typedef std::shared_ptr<HLTGlobalStatus> TrigResPtr;
0048
0049     Path(int bitpos,
0050          std::string const& path_name,
0051          WorkersInPath const& workers,
0052          TrigResPtr trptr,
0053          ExceptionToActionTable const& actions,
0054          std::shared_ptr<ActivityRegistry> reg,
0055          StreamContext const* streamContext,
0056          PathContext::PathType pathType);
0057
0058     Path(Path const&);
0059
0060     Path& operator=(Path const&) = delete;
0061
0062     void processEventUsingPathAsync(
0063         WaitingTaskHolder, EventTransitionInfo const&, ServiceToken const&, StreamID const&, StreamContext const*);
0064
0065     int bitPosition() const { return bitpos_; }
0066     std::string const& name() const { return pathContext_.pathName(); }
0067
0068     void clearCounters();
0069
0070     int timesRun() const { return timesRun_; }
0071     int timesPassed() const { return timesPassed_; }
0072     int timesFailed() const { return timesFailed_; }
0073     int timesExcept() const { return timesExcept_; }
0074     //int abortWorker() const { return abortWorker_; }
0075
0076     size_type size() const { return workers_.size(); }
0077     int timesVisited(size_type i) const { return workers_.at(i).timesVisited(); }
0078     int timesPassed(size_type i) const { return workers_.at(i).timesPassed(); }
0079     int timesFailed(size_type i) const { return workers_.at(i).timesFailed(); }
0080     int timesExcept(size_type i) const { return workers_.at(i).timesExcept(); }
0081     Worker const* getWorker(size_type i) const { return workers_.at(i).getWorker(); }
0082     unsigned int bitPosition(size_type i) const { return workers_.at(i).bitPosition(); }
0083
0084     void setEarlyDeleteHelpers(std::map<const Worker*, EarlyDeleteHelper*> const&);
0085
0086     void setPathStatusInserter(PathStatusInserter* pathStatusInserter, Worker* pathStatusInserterWorker);
0087
0088   private:
0089     int timesRun_;
0090     int timesPassed_;
0091     int timesFailed_;
0092     int timesExcept_;
0093     //int abortWorker_;
0094     std::atomic<bool> printedException_ = false;
0095     //When an exception happens, it is possible for multiple modules in a path to fail
0096     // and then try to change the state concurrently.
0097     std::atomic<bool> stateLock_ = false;
0098     CMS_THREAD_GUARD(stateLock_) int failedModuleIndex_;
0099     CMS_THREAD_GUARD(stateLock_) State state_;
0100
0101     int const bitpos_;
0102     TrigResPtr const trptr_;
0103     // We do not use propagate_const because the registry itself is mutable.
0104     std::shared_ptr<ActivityRegistry> const actReg_;
0105     ExceptionToActionTable const* const act_table_;
0106
0107     WorkersInPath workers_;
0108
0109     PathContext pathContext_;
0110     WaitingTaskList waitingTasks_;
0111     std::atomic<unsigned int> modulesToRun_;
0112
0113     PathStatusInserter* pathStatusInserter_;
0114     Worker* pathStatusInserterWorker_;
0115
0116     // Helper functions
0117     // nwrwue = numWorkersRunWithoutUnhandledException (really!)
0118     void handleWorkerFailure(cms::Exception& e, int nwrwue, ModuleDescription const&, std::string const& id);
0119     static void exceptionContext(cms::Exception& ex,
0120                                  bool isEvent,
0121                                  bool begin,
0122                                  BranchType branchType,
0123                                  ModuleDescription const&,
0124                                  std::string const& id,
0125                                  PathContext const&);
0126     void threadsafe_setFailedModuleInfo(int nwrwue, bool iExceptionHappened);
0127     void recordStatus(int nwrwue, hlt::HLTState state);
0128     void updateCounters(hlt::HLTState state);
0129
0130     void finished(std::exception_ptr, StreamContext const*, EventTransitionInfo const&, StreamID const&);
0131
0132     //Handle asynchronous processing
0133     void workerFinished(std::exception_ptr const*,
0134                         unsigned int iModuleIndex,
0135                         EventTransitionInfo const&,
0136                         ServiceToken const&,
0137                         StreamID const&,
0138                         StreamContext const*,
0139                         oneapi::tbb::task_group& iGroup);
0140     void runNextWorkerAsync(unsigned int iNextModuleIndex,
0141                             EventTransitionInfo const&,
0142                             ServiceToken const&,
0143                             StreamID const&,
0144                             StreamContext const*,
0145                             oneapi::tbb::task_group& iGroup);
0146   };
0147
0148   namespace {
0149     template <typename T>
0150     class PathSignalSentry {
0151     public:
0152       PathSignalSentry(ActivityRegistry* a,
0153                        int const& nwrwue,
0154                        hlt::HLTState const& state,
0155                        PathContext const* pathContext)
0156           : a_(a), nwrwue_(nwrwue), state_(state), pathContext_(pathContext) {
0157         if (a_)
0158           T::prePathSignal(a_, pathContext_);
0159       }
0160       ~PathSignalSentry() {
0161         HLTPathStatus status(state_, nwrwue_);
0162         if (a_)
0163           T::postPathSignal(a_, status, pathContext_);
0164       }
0165
0166     private:
0167       ActivityRegistry* a_;  // We do not use propagate_const because the registry itself is mutable.
0168       int const& nwrwue_;
0169       hlt::HLTState const& state_;
0170       PathContext const* pathContext_;
0171     };
0172   }  // namespace
0173
0174 }  // namespace edm
0175
0176 #endif