Framework/interface/globalTransitionAsync.h

0001 #ifndef FWCore_Framework_globalTransitionAsync_h
0002 #define FWCore_Framework_globalTransitionAsync_h
0003 // -*- C++ -*-
0004 //
0005 // Package:     FWCore/Framework
0006 // Function:    globalTransitionAsync
0007 //
0008 /**\function globalTransitionAsync globalTransitionAsync.h "globalTransitionAsync.h"
0009
0010  Description: Helper functions for handling asynchronous global transitions
0011
0012  Usage:
0013     <usage>
0014
0015 */
0016 //
0017 // Original Author:  Chris Jones
0018 //         Created:  Tue, 06 Sep 2016 16:04:26 GMT
0019 //
0020
0021 #include "FWCore/Framework/interface/Schedule.h"
0022 #include "FWCore/Framework/interface/SubProcess.h"
0023 #include "FWCore/Framework/interface/TransitionInfoTypes.h"
0024 #include "FWCore/Concurrency/interface/WaitingTask.h"
0025 #include "FWCore/Concurrency/interface/WaitingTaskHolder.h"
0026 #include "FWCore/Concurrency/interface/chain_first.h"
0027
0028 #include <exception>
0029 #include <utility>
0030 #include <vector>
0031
0032 namespace edm {
0033
0034   //This is code in common between beginStreamRun and beginGlobalLuminosityBlock
0035   template <typename T>
0036   inline void subProcessDoGlobalBeginTransitionAsync(WaitingTaskHolder iHolder,
0037                                                      SubProcess& iSubProcess,
0038                                                      LumiTransitionInfo const& iTransitionInfo,
0039                                                      bool) {
0040     iSubProcess.doBeginLuminosityBlockAsync(std::move(iHolder), iTransitionInfo);
0041   }
0042
0043   template <typename T>
0044   inline void subProcessDoGlobalBeginTransitionAsync(WaitingTaskHolder iHolder,
0045                                                      SubProcess& iSubProcess,
0046                                                      RunTransitionInfo const& iTransitionInfo,
0047                                                      bool) {
0048     iSubProcess.doBeginRunAsync(std::move(iHolder), iTransitionInfo);
0049   }
0050
0051   template <typename Traits>
0052   inline void subProcessDoGlobalBeginTransitionAsync(WaitingTaskHolder iHolder,
0053                                                      SubProcess& iSubProcess,
0054                                                      ProcessBlockTransitionInfo const& iTransitionInfo,
0055                                                      bool cleaningUpAfterException) {
0056     iSubProcess.doBeginProcessBlockAsync<Traits>(std::move(iHolder), iTransitionInfo, cleaningUpAfterException);
0057   }
0058
0059   inline void subProcessDoGlobalEndTransitionAsync(WaitingTaskHolder iHolder,
0060                                                    SubProcess& iSubProcess,
0061                                                    LumiTransitionInfo const& iTransitionInfo,
0062                                                    bool cleaningUpAfterException) {
0063     iSubProcess.doEndLuminosityBlockAsync(std::move(iHolder), iTransitionInfo, cleaningUpAfterException);
0064   }
0065
0066   inline void subProcessDoGlobalEndTransitionAsync(WaitingTaskHolder iHolder,
0067                                                    SubProcess& iSubProcess,
0068                                                    RunTransitionInfo const& iTransitionInfo,
0069                                                    bool cleaningUpAfterException) {
0070     iSubProcess.doEndRunAsync(std::move(iHolder), iTransitionInfo, cleaningUpAfterException);
0071   }
0072
0073   inline void subProcessDoGlobalEndTransitionAsync(WaitingTaskHolder iHolder,
0074                                                    SubProcess& iSubProcess,
0075                                                    ProcessBlockTransitionInfo const& iTransitionInfo,
0076                                                    bool cleaningUpAfterException) {
0077     iSubProcess.doEndProcessBlockAsync(std::move(iHolder), iTransitionInfo, cleaningUpAfterException);
0078   }
0079
0080   template <typename Traits>
0081   void beginGlobalTransitionAsync(WaitingTaskHolder iWait,
0082                                   Schedule& iSchedule,
0083                                   typename Traits::TransitionInfoType& transitionInfo,
0084                                   ServiceToken const& token,
0085                                   std::vector<SubProcess>& iSubProcesses,
0086                                   bool cleaningUpAfterException = false) {
0087     // When we are done processing the global for this process,
0088     // we need to run the global for all SubProcesses
0089     using namespace edm::waiting_task;
0090
0091     chain::first([&](auto nextTask) {
0092       iSchedule.processOneGlobalAsync<Traits>(std::move(nextTask), transitionInfo, token, cleaningUpAfterException);
0093     }) |
0094         chain::then([&iSubProcesses, info = transitionInfo, cleaningUpAfterException](std::exception_ptr const* iPtr,
0095                                                                                       auto nextTask) {
0096           if (iPtr) {
0097             //delay handling exception until after subProcesses run
0098             chain::first([&](auto nextTask) {
0099               for (auto& subProcess : iSubProcesses) {
0100                 subProcessDoGlobalBeginTransitionAsync<Traits>(nextTask, subProcess, info, cleaningUpAfterException);
0101               }
0102             }) | chain::then([excpt = *iPtr](std::exception_ptr const*, auto nextTask) {
0103               nextTask.doneWaiting(excpt);
0104             }) | chain::runLast(nextTask);
0105           } else {
0106             for (auto& subProcess : iSubProcesses) {
0107               subProcessDoGlobalBeginTransitionAsync<Traits>(nextTask, subProcess, info, cleaningUpAfterException);
0108             }
0109           }
0110         }) |
0111         chain::runLast(iWait);
0112   }
0113
0114   template <typename Traits>
0115   void endGlobalTransitionAsync(WaitingTaskHolder iWait,
0116                                 Schedule& iSchedule,
0117                                 typename Traits::TransitionInfoType& transitionInfo,
0118                                 ServiceToken const& token,
0119                                 std::vector<SubProcess>& iSubProcesses,
0120                                 bool cleaningUpAfterException) {
0121     using namespace edm::waiting_task;
0122     chain::first([&](auto nextTask) {
0123       iSchedule.processOneGlobalAsync<Traits>(std::move(nextTask), transitionInfo, token, cleaningUpAfterException);
0124     })
0125         // When we are done processing the global for this process,
0126         // we need to run the global for all SubProcesses
0127         | chain::then([&iSubProcesses, info = transitionInfo, cleaningUpAfterException](std::exception_ptr const* iPtr,
0128                                                                                         auto nextTask) {
0129             if (iPtr) {
0130               //still run the sub process but pass this exception to the nextTask
0131               auto excpt = *iPtr;
0132               chain::first([&](auto nextTask) {
0133                 for (auto& subProcess : iSubProcesses) {
0134                   subProcessDoGlobalEndTransitionAsync(nextTask, subProcess, info, cleaningUpAfterException);
0135                 }
0136               }) | chain::then([excpt](std::exception_ptr const*, auto nextTask) { nextTask.doneWaiting(excpt); }) |
0137                   chain::runLast(std::move(nextTask));
0138             } else {
0139               for (auto& subProcess : iSubProcesses) {
0140                 subProcessDoGlobalEndTransitionAsync(nextTask, subProcess, info, cleaningUpAfterException);
0141               }
0142             }
0143           }) |
0144         chain::runLast(iWait);
0145   }
0146
0147 };  // namespace edm
0148
0149 #endif