Framework/src/RunProcessingStatus.h

0001 #ifndef FWCore_Framework_RunProcessingStatus_h
0002 #define FWCore_Framework_RunProcessingStatus_h
0003 //
0004 // Package:     FWCore/Framework
0005 // Class  :     RunProcessingStatus
0006 //
0007 /**\class edm::RunProcessingStatus
0008
0009  Description: Keep status information about one Run transition
0010
0011 */
0012 //
0013 // Original Author: W. David Dagenhart
0014 //          Created: 1 Oct 2021
0015 //
0016
0017 #include "DataFormats/Provenance/interface/Timestamp.h"
0018 #include "FWCore/Concurrency/interface/LimitedTaskQueue.h"
0019 #include "FWCore/Concurrency/interface/WaitingTaskHolder.h"
0020 #include "FWCore/Concurrency/interface/WaitingTaskList.h"
0021 #include "FWCore/Framework/interface/InputSource.h"
0022
0023 #include <atomic>
0024 #include <memory>
0025 #include <vector>
0026
0027 namespace edm {
0028
0029   class EventSetupImpl;
0030   class RunPrincipal;
0031
0032   class RunProcessingStatus {
0033   public:
0034     RunProcessingStatus(unsigned int iNStreams, WaitingTaskHolder const& holder);
0035
0036     RunProcessingStatus(RunProcessingStatus const&) = delete;
0037     RunProcessingStatus const& operator=(RunProcessingStatus const&) = delete;
0038
0039     WaitingTaskHolder& holderOfTaskInProcessRuns() { return holderOfTaskInProcessRuns_; }
0040     void setHolderOfTaskInProcessRuns(WaitingTaskHolder const& holder) { holderOfTaskInProcessRuns_ = holder; }
0041
0042     void setResumer(LimitedTaskQueue::Resumer iResumer) { globalRunQueueResumer_ = std::move(iResumer); }
0043     void resumeGlobalRunQueue() {
0044       //free run for next usage
0045       runPrincipal_.reset();
0046       globalRunQueueResumer_.resume();
0047     }
0048
0049     std::shared_ptr<RunPrincipal>& runPrincipal() { return runPrincipal_; }
0050     void setRunPrincipal(std::shared_ptr<RunPrincipal> val) { runPrincipal_ = std::move(val); }
0051
0052     void resetBeginResources();
0053     void resetEndResources();
0054
0055     EventSetupImpl const& eventSetupImpl(unsigned subProcessIndex) const {
0056       return *eventSetupImpls_.at(subProcessIndex);
0057     }
0058
0059     EventSetupImpl const& eventSetupImplEndRun(unsigned subProcessIndex) const {
0060       return *eventSetupImplsEndRun_.at(subProcessIndex);
0061     }
0062
0063     std::vector<std::shared_ptr<const EventSetupImpl>>& eventSetupImpls() { return eventSetupImpls_; }
0064     std::vector<std::shared_ptr<const EventSetupImpl>> const& eventSetupImpls() const { return eventSetupImpls_; }
0065
0066     WaitingTaskList& endIOVWaitingTasks() { return endIOVWaitingTasks_; }
0067
0068     std::vector<std::shared_ptr<const EventSetupImpl>>& eventSetupImplsEndRun() { return eventSetupImplsEndRun_; }
0069     std::vector<std::shared_ptr<const EventSetupImpl>> const& eventSetupImplsEndRun() const {
0070       return eventSetupImplsEndRun_;
0071     }
0072
0073     WaitingTaskList& endIOVWaitingTasksEndRun() { return endIOVWaitingTasksEndRun_; }
0074
0075     void setGlobalEndRunHolder(WaitingTaskHolder holder) { globalEndRunHolder_ = std::move(holder); }
0076     WaitingTaskHolder& globalEndRunHolder() { return globalEndRunHolder_; }
0077
0078     bool streamFinishedBeginRun() { return 0 == (--nStreamsStillProcessingBeginRun_); }
0079     bool streamFinishedRun() { return 0 == (--nStreamsStillProcessingRun_); }
0080
0081     //These should only be called while in the InputSource's task queue
0082     void updateLastTimestamp(edm::Timestamp const& iTime) {
0083       if (iTime > endTime_) {
0084         endTime_ = iTime;
0085       }
0086     }
0087     edm::Timestamp const& lastTimestamp() const { return endTime_; }
0088
0089     void setEndTime();
0090
0091     bool didGlobalBeginSucceed() const { return globalBeginSucceeded_; }
0092     void globalBeginDidSucceed() { globalBeginSucceeded_ = true; }
0093
0094     bool cleaningUpAfterException() const { return cleaningUpAfterException_; }
0095     void setCleaningUpAfterException(bool val) { cleaningUpAfterException_ = val; }
0096
0097     bool stopBeforeProcessingRun() const { return stopBeforeProcessingRun_; }
0098     void setStopBeforeProcessingRun(bool val) { stopBeforeProcessingRun_ = val; }
0099
0100     bool endingEventSetupSucceeded() const { return endingEventSetupSucceeded_; }
0101     void setEndingEventSetupSucceeded(bool val) { endingEventSetupSucceeded_ = val; }
0102
0103   private:
0104     WaitingTaskHolder holderOfTaskInProcessRuns_;
0105     LimitedTaskQueue::Resumer globalRunQueueResumer_;
0106     std::shared_ptr<RunPrincipal> runPrincipal_;
0107     std::vector<std::shared_ptr<const EventSetupImpl>> eventSetupImpls_;
0108     WaitingTaskList endIOVWaitingTasks_;
0109     std::vector<std::shared_ptr<const EventSetupImpl>> eventSetupImplsEndRun_;
0110     WaitingTaskList endIOVWaitingTasksEndRun_;
0111     WaitingTaskHolder globalEndRunHolder_;
0112     std::atomic<unsigned int> nStreamsStillProcessingBeginRun_;
0113     std::atomic<unsigned int> nStreamsStillProcessingRun_;
0114     edm::Timestamp endTime_{};
0115     std::atomic<char> endTimeSetStatus_{0};
0116     bool globalBeginSucceeded_{false};
0117     bool cleaningUpAfterException_{false};
0118     bool stopBeforeProcessingRun_{false};
0119     bool endingEventSetupSucceeded_{true};
0120   };
0121 }  // namespace edm
0122
0123 #endif