TestProcessor.cc

Go to the documentation of this file.

// -*- C++ -*-
//
// Package:     Subsystem/Package
// Class  :     TestProcessor
//
// Implementation:
//     [Notes on implementation]
//
// Original Author:  Chris Jones
//         Created:  Mon, 30 Apr 2018 18:51:08 GMT
//

// system include files

// user include files
#include "FWCore/TestProcessor/interface/TestProcessor.h"
#include "FWCore/TestProcessor/interface/EventSetupTestHelper.h"

#include "FWCore/Common/interface/ProcessBlockHelper.h"
#include "FWCore/Framework/interface/ScheduleItems.h"
#include "FWCore/Framework/interface/EventPrincipal.h"
#include "FWCore/Framework/interface/LuminosityBlockPrincipal.h"
#include "FWCore/Framework/interface/ProcessBlockPrincipal.h"
#include "FWCore/Framework/interface/ExceptionActions.h"
#include "FWCore/Framework/interface/HistoryAppender.h"
#include "FWCore/Framework/interface/PathsAndConsumesOfModules.h"
#include "FWCore/Framework/interface/RunPrincipal.h"
#include "FWCore/Framework/interface/ESRecordsToProductResolverIndices.h"
#include "FWCore/Framework/interface/EventSetupsController.h"
#include "FWCore/Framework/interface/globalTransitionAsync.h"
#include "FWCore/Framework/interface/streamTransitionAsync.h"
#include "FWCore/Framework/interface/TransitionInfoTypes.h"
#include "FWCore/Framework/interface/ProductPutterBase.h"
#include "FWCore/Framework/interface/DelayedReader.h"
#include "FWCore/Framework/interface/ensureAvailableAccelerators.h"
#include "FWCore/Framework/interface/makeModuleTypeResolverMaker.h"
#include "FWCore/Framework/interface/FileBlock.h"
#include "FWCore/Framework/interface/MergeableRunProductMetadata.h"

#include "FWCore/ServiceRegistry/interface/ServiceRegistry.h"
#include "FWCore/ServiceRegistry/interface/SystemBounds.h"

#include "FWCore/ParameterSetReader/interface/ProcessDescImpl.h"
#include "FWCore/ParameterSet/interface/ProcessDesc.h"
#include "FWCore/ParameterSet/interface/validateTopLevelParameterSets.h"

#include "FWCore/Utilities/interface/ExceptionCollector.h"

#include "FWCore/Concurrency/interface/FinalWaitingTask.h"

#include "oneTimeInitialization.h"

#include <mutex>

#define xstr(s) str(s)
#define str(s) #s

namespace edm {
  namespace test {

    //
    // constructors and destructor
    //
    TestProcessor::TestProcessor(Config const& iConfig, ServiceToken iToken)
        : globalControl_(oneapi::tbb::global_control::max_allowed_parallelism, 1),
          arena_(1),
          historyAppender_(std::make_unique<HistoryAppender>()),
          moduleRegistry_(std::make_shared<ModuleRegistry>()) {
      //Setup various singletons
      (void)testprocessor::oneTimeInitialization();

      ProcessDescImpl desc(iConfig.pythonConfiguration(), false);

      auto psetPtr = desc.parameterSet();
      moduleTypeResolverMaker_ = makeModuleTypeResolverMaker(*psetPtr);
      espController_ = std::make_unique<eventsetup::EventSetupsController>(moduleTypeResolverMaker_.get());

      validateTopLevelParameterSets(psetPtr.get());

      ensureAvailableAccelerators(*psetPtr);

      labelOfTestModule_ = psetPtr->getParameter<std::string>("@moduleToTest");

      auto procDesc = desc.processDesc();
      // Now do general initialization
      ScheduleItems items;

      //initialize the services
      auto& serviceSets = procDesc->getServicesPSets();
      ServiceToken token = items.initServices(serviceSets, *psetPtr, iToken, serviceregistry::kOverlapIsError, true);
      serviceToken_ = items.addCPRandTNS(*psetPtr, token);

      //make the services available
      ServiceRegistry::Operate operate(serviceToken_);

      // intialize miscellaneous items
      std::shared_ptr<CommonParams> common(items.initMisc(*psetPtr));

      // intialize the event setup provider
      esp_ = espController_->makeProvider(*psetPtr, items.actReg_.get());

      auto nThreads = 1U;
      auto nStreams = 1U;
      auto nConcurrentLumis = 1U;
      auto nConcurrentRuns = 1U;
      preallocations_ = PreallocationConfiguration{nThreads, nStreams, nConcurrentLumis, nConcurrentRuns};

      if (not iConfig.esProduceEntries().empty()) {
        esHelper_ = std::make_unique<EventSetupTestHelper>(iConfig.esProduceEntries());
        esp_->add(std::dynamic_pointer_cast<eventsetup::ESProductResolverProvider>(esHelper_));
        esp_->add(std::dynamic_pointer_cast<EventSetupRecordIntervalFinder>(esHelper_));
      }

      preg_ = items.preg();
      processConfiguration_ = items.processConfiguration();

      edm::ParameterSet emptyPSet;
      emptyPSet.registerIt();
      auto psetid = emptyPSet.id();

      for (auto const& p : iConfig.extraProcesses()) {
        processHistory_.emplace_back(p, psetid, xstr(PROJECT_VERSION), "0");
        processHistoryRegistry_.registerProcessHistory(processHistory_);
      }

      //setup the products we will be adding to the event
      for (auto const& produce : iConfig.produceEntries()) {
        auto processName = produce.processName_;
        if (processName.empty()) {
          processName = processConfiguration_->processName();
        }
        edm::TypeWithDict twd(produce.type_.typeInfo());
        edm::BranchDescription product(edm::InEvent,
                                       produce.moduleLabel_,
                                       processName,
                                       twd.userClassName(),
                                       twd.friendlyClassName(),
                                       produce.instanceLabel_,
                                       "",
                                       psetid,
                                       twd,
                                       true  //force this to come from 'source'
        );
        product.init();
        dataProducts_.emplace_back(product, std::unique_ptr<WrapperBase>());
        preg_->addProduct(product);
      }

      processBlockHelper_ = std::make_shared<ProcessBlockHelper>();

      schedule_ = items.initSchedule(
          *psetPtr, false, preallocations_, &processContext_, moduleTypeResolverMaker_.get(), *processBlockHelper_);
      // set the data members
      act_table_ = std::move(items.act_table_);
      actReg_ = items.actReg_;
      branchIDListHelper_ = items.branchIDListHelper();
      thinnedAssociationsHelper_ = items.thinnedAssociationsHelper();
      processContext_.setProcessConfiguration(processConfiguration_.get());

      principalCache_.setNumberOfConcurrentPrincipals(preallocations_);

      preg_->setFrozen();
      mergeableRunProductProcesses_.setProcessesWithMergeableRunProducts(*preg_);

      for (unsigned int index = 0; index < preallocations_.numberOfStreams(); ++index) {
        // Reusable event principal
        auto ep = std::make_shared<EventPrincipal>(preg_,
                                                   branchIDListHelper_,
                                                   thinnedAssociationsHelper_,
                                                   *processConfiguration_,
                                                   historyAppender_.get(),
                                                   index);
        principalCache_.insert(std::move(ep));
      }
      for (unsigned int index = 0; index < preallocations_.numberOfRuns(); ++index) {
        auto rp = std::make_unique<RunPrincipal>(
            preg_, *processConfiguration_, historyAppender_.get(), index, true, &mergeableRunProductProcesses_);
        principalCache_.insert(std::move(rp));
      }
      for (unsigned int index = 0; index < preallocations_.numberOfLuminosityBlocks(); ++index) {
        auto lp =
            std::make_unique<LuminosityBlockPrincipal>(preg_, *processConfiguration_, historyAppender_.get(), index);
        principalCache_.insert(std::move(lp));
      }
      {
        auto pb = std::make_unique<ProcessBlockPrincipal>(preg_, *processConfiguration_);
        principalCache_.insert(std::move(pb));
      }
    }

    TestProcessor::~TestProcessor() noexcept(false) { teardownProcessing(); }
    //
    // member functions
    //

    void TestProcessor::put(unsigned int index, std::unique_ptr<WrapperBase> iWrapper) {
      if (index >= dataProducts_.size()) {
        throw cms::Exception("LogicError") << "Products must be declared to the TestProcessor::Config object\n"
                                              "with a call to the function \'produces\' BEFORE passing the\n"
                                              "TestProcessor::Config object to the TestProcessor constructor";
      }
      dataProducts_[index].second = std::move(iWrapper);
    }

    edm::test::Event TestProcessor::testImpl() {
      bool result = arena_.execute([this]() {
        setupProcessing();
        event();

        return schedule_->totalEventsPassed() > 0;
      });
      schedule_->clearCounters();
      if (esHelper_) {
        //We want each test to have its own ES data products
        esHelper_->resetAllProxies();
      }
      return edm::test::Event(
          principalCache_.eventPrincipal(0), labelOfTestModule_, processConfiguration_->processName(), result);
    }

    edm::test::LuminosityBlock TestProcessor::testBeginLuminosityBlockImpl(edm::LuminosityBlockNumber_t iNum) {
      arena_.execute([this, iNum]() {
        if (not beginJobCalled_) {
          beginJob();
        }
        if (not respondToOpenInputFileCalled_) {
          respondToOpenInputFile();
        }
        if (not beginProcessBlockCalled_) {
          beginProcessBlock();
        }
        if (not openOutputFilesCalled_) {
          openOutputFiles();
        }

        if (not beginRunCalled_) {
          beginRun();
        }
        if (beginLumiCalled_) {
          endLuminosityBlock();
          assert(lumiNumber_ != iNum);
        }
        lumiNumber_ = iNum;
        beginLuminosityBlock();
      });

      if (esHelper_) {
        //We want each test to have its own ES data products
        esHelper_->resetAllProxies();
      }
      return edm::test::LuminosityBlock(lumiPrincipal_, labelOfTestModule_, processConfiguration_->processName());
    }

    edm::test::LuminosityBlock TestProcessor::testEndLuminosityBlockImpl() {
      //using a return value from arena_.execute lead to double delete of shared_ptr
      // based on valgrind output when exception occurred. Use lambda capture instead.
      std::shared_ptr<edm::LuminosityBlockPrincipal> lumi;
      arena_.execute([this, &lumi]() {
        if (not beginJobCalled_) {
          beginJob();
        }
        if (not respondToOpenInputFileCalled_) {
          respondToOpenInputFile();
        }
        if (not beginProcessBlockCalled_) {
          beginProcessBlock();
        }
        if (not openOutputFilesCalled_) {
          openOutputFiles();
        }
        if (not beginRunCalled_) {
          beginRun();
        }
        if (not beginLumiCalled_) {
          beginLuminosityBlock();
        }
        lumi = endLuminosityBlock();
      });
      if (esHelper_) {
        //We want each test to have its own ES data products
        esHelper_->resetAllProxies();
      }

      return edm::test::LuminosityBlock(std::move(lumi), labelOfTestModule_, processConfiguration_->processName());
    }

    edm::test::Run TestProcessor::testBeginRunImpl(edm::RunNumber_t iNum) {
      arena_.execute([this, iNum]() {
        if (not beginJobCalled_) {
          beginJob();
        }
        if (not respondToOpenInputFileCalled_) {
          respondToOpenInputFile();
        }
        if (not beginProcessBlockCalled_) {
          beginProcessBlock();
        }
        if (not openOutputFilesCalled_) {
          openOutputFiles();
        }
        if (beginRunCalled_) {
          assert(runNumber_ != iNum);
          endRun();
        }
        runNumber_ = iNum;
        beginRun();
      });
      if (esHelper_) {
        //We want each test to have its own ES data products
        esHelper_->resetAllProxies();
      }
      return edm::test::Run(runPrincipal_, labelOfTestModule_, processConfiguration_->processName());
    }
    edm::test::Run TestProcessor::testEndRunImpl() {
      //using a return value from arena_.execute lead to double delete of shared_ptr
      // based on valgrind output when exception occurred. Use lambda capture instead.
      std::shared_ptr<edm::RunPrincipal> rp;
      arena_.execute([this, &rp]() {
        if (not beginJobCalled_) {
          beginJob();
        }
        if (not respondToOpenInputFileCalled_) {
          respondToOpenInputFile();
        }
        if (not beginProcessBlockCalled_) {
          beginProcessBlock();
        }
        if (not openOutputFilesCalled_) {
          openOutputFiles();
        }
        if (not beginRunCalled_) {
          beginRun();
        }
        rp = endRun();
      });
      if (esHelper_) {
        //We want each test to have its own ES data products
        esHelper_->resetAllProxies();
      }

      return edm::test::Run(rp, labelOfTestModule_, processConfiguration_->processName());
    }

    edm::test::ProcessBlock TestProcessor::testBeginProcessBlockImpl() {
      arena_.execute([this]() {
        if (not beginJobCalled_) {
          beginJob();
        }
        beginProcessBlock();
      });
      return edm::test::ProcessBlock(
          &principalCache_.processBlockPrincipal(), labelOfTestModule_, processConfiguration_->processName());
    }
    edm::test::ProcessBlock TestProcessor::testEndProcessBlockImpl() {
      auto pbp = arena_.execute([this]() {
        if (not beginJobCalled_) {
          beginJob();
        }
        if (not beginProcessBlockCalled_) {
          beginProcessBlock();
        }
        return endProcessBlock();
      });
      return edm::test::ProcessBlock(pbp, labelOfTestModule_, processConfiguration_->processName());
    }

    void TestProcessor::setupProcessing() {
      if (not beginJobCalled_) {
        beginJob();
      }
      if (not respondToOpenInputFileCalled_) {
        respondToOpenInputFile();
      }
      if (not beginProcessBlockCalled_) {
        beginProcessBlock();
      }
      if (not openOutputFilesCalled_) {
        openOutputFiles();
      }
      if (not beginRunCalled_) {
        beginRun();
      }
      if (not beginLumiCalled_) {
        beginLuminosityBlock();
      }
    }

    void TestProcessor::teardownProcessing() {
      arena_.execute([this]() {
        if (beginLumiCalled_) {
          endLuminosityBlock();
          beginLumiCalled_ = false;
        }
        if (beginRunCalled_) {
          endRun();
          beginRunCalled_ = false;
        }
        if (respondToOpenInputFileCalled_) {
          respondToCloseInputFile();
        }
        if (beginProcessBlockCalled_) {
          endProcessBlock();
          beginProcessBlockCalled_ = false;
        }
        if (openOutputFilesCalled_) {
          closeOutputFiles();
          openOutputFilesCalled_ = false;
        }
        if (beginJobCalled_) {
          endJob();
        }
        edm::FinalWaitingTask task{taskGroup_};
        espController_->endIOVsAsync(edm::WaitingTaskHolder(taskGroup_, &task));
        task.waitNoThrow();
      });
    }

    void TestProcessor::beginJob() {
      ServiceRegistry::Operate operate(serviceToken_);

      service::SystemBounds bounds(preallocations_.numberOfStreams(),
                                   preallocations_.numberOfLuminosityBlocks(),
                                   preallocations_.numberOfRuns(),
                                   preallocations_.numberOfThreads());
      actReg_->preallocateSignal_(bounds);
      schedule_->convertCurrentProcessAlias(processConfiguration_->processName());
      PathsAndConsumesOfModules pathsAndConsumesOfModules;

      //The code assumes only modules make data in the current process
      // Since the test os also allowed to do so, it can lead to problems.
      //pathsAndConsumesOfModules.initialize(schedule_.get(), preg_);

      espController_->finishConfiguration();
      actReg_->eventSetupConfigurationSignal_(esp_->recordsToResolverIndices(), processContext_);
      //NOTE: this may throw
      //checkForModuleDependencyCorrectness(pathsAndConsumesOfModules, false);

      schedule_->beginJob(
          *preg_, esp_->recordsToResolverIndices(), *processBlockHelper_, pathsAndConsumesOfModules, processContext_);

      for (unsigned int i = 0; i < preallocations_.numberOfStreams(); ++i) {
        schedule_->beginStream(i);
      }
      beginJobCalled_ = true;
    }

    void TestProcessor::beginProcessBlock() {
      ProcessBlockPrincipal& processBlockPrincipal = principalCache_.processBlockPrincipal();
      processBlockPrincipal.fillProcessBlockPrincipal(processConfiguration_->processName());

      std::vector<edm::SubProcess> emptyList;
      {
        ProcessBlockTransitionInfo transitionInfo(processBlockPrincipal);
        using Traits = OccurrenceTraits<ProcessBlockPrincipal, BranchActionGlobalBegin>;
        FinalWaitingTask globalWaitTask{taskGroup_};
        beginGlobalTransitionAsync<Traits>(
            WaitingTaskHolder(taskGroup_, &globalWaitTask), *schedule_, transitionInfo, serviceToken_, emptyList);
        globalWaitTask.wait();
      }
      beginProcessBlockCalled_ = true;
    }

    void TestProcessor::openOutputFiles() {
      //make the services available
      ServiceRegistry::Operate operate(serviceToken_);

      edm::FileBlock fb;
      schedule_->openOutputFiles(fb);
      openOutputFilesCalled_ = true;
    }

    void TestProcessor::closeOutputFiles() {
      if (openOutputFilesCalled_) {
        //make the services available
        ServiceRegistry::Operate operate(serviceToken_);
        schedule_->closeOutputFiles();

        openOutputFilesCalled_ = false;
      }
    }

    void TestProcessor::respondToOpenInputFile() {
      respondToOpenInputFileCalled_ = true;
      edm::FileBlock fb;
      //make the services available
      ServiceRegistry::Operate operate(serviceToken_);
      schedule_->respondToOpenInputFile(fb);
    }

    void TestProcessor::respondToCloseInputFile() {
      if (respondToOpenInputFileCalled_) {
        edm::FileBlock fb;
        //make the services available
        ServiceRegistry::Operate operate(serviceToken_);

        schedule_->respondToCloseInputFile(fb);
        respondToOpenInputFileCalled_ = false;
      }
    }

    void TestProcessor::beginRun() {
      runPrincipal_ = principalCache_.getAvailableRunPrincipalPtr();
      runPrincipal_->clearPrincipal();
      assert(runPrincipal_);
      edm::RunAuxiliary aux(runNumber_, Timestamp(), Timestamp());
      aux.setProcessHistoryID(processHistory_.id());
      runPrincipal_->setAux(aux);

      runPrincipal_->fillRunPrincipal(processHistoryRegistry_);

      IOVSyncValue ts(EventID(runPrincipal_->run(), 0, 0), runPrincipal_->beginTime());
      eventsetup::synchronousEventSetupForInstance(ts, taskGroup_, *espController_);

      auto const& es = esp_->eventSetupImpl();

      RunTransitionInfo transitionInfo(*runPrincipal_, es, nullptr);

      std::vector<edm::SubProcess> emptyList;
      {
        using Traits = OccurrenceTraits<RunPrincipal, BranchActionGlobalBegin>;
        FinalWaitingTask globalWaitTask{taskGroup_};
        beginGlobalTransitionAsync<Traits>(
            WaitingTaskHolder(taskGroup_, &globalWaitTask), *schedule_, transitionInfo, serviceToken_, emptyList);
        globalWaitTask.wait();
      }
      {
        //To wait, the ref count has to be 1+#streams
        FinalWaitingTask streamLoopWaitTask{taskGroup_};

        using Traits = OccurrenceTraits<RunPrincipal, BranchActionStreamBegin>;
        beginStreamsTransitionAsync<Traits>(WaitingTaskHolder(taskGroup_, &streamLoopWaitTask),
                                            *schedule_,
                                            preallocations_.numberOfStreams(),
                                            transitionInfo,
                                            serviceToken_,
                                            emptyList);
        streamLoopWaitTask.wait();
      }
      beginRunCalled_ = true;
    }

    void TestProcessor::beginLuminosityBlock() {
      LuminosityBlockAuxiliary aux(runNumber_, lumiNumber_, Timestamp(), Timestamp());
      aux.setProcessHistoryID(processHistory_.id());
      lumiPrincipal_ = principalCache_.getAvailableLumiPrincipalPtr();
      lumiPrincipal_->clearPrincipal();
      assert(lumiPrincipal_);
      lumiPrincipal_->setAux(aux);

      lumiPrincipal_->setRunPrincipal(runPrincipal_);
      lumiPrincipal_->fillLuminosityBlockPrincipal(&processHistory_);

      IOVSyncValue ts(EventID(runNumber_, lumiNumber_, eventNumber_), lumiPrincipal_->beginTime());
      eventsetup::synchronousEventSetupForInstance(ts, taskGroup_, *espController_);

      auto const& es = esp_->eventSetupImpl();

      LumiTransitionInfo transitionInfo(*lumiPrincipal_, es, nullptr);

      std::vector<edm::SubProcess> emptyList;
      {
        using Traits = OccurrenceTraits<LuminosityBlockPrincipal, BranchActionGlobalBegin>;
        FinalWaitingTask globalWaitTask{taskGroup_};
        beginGlobalTransitionAsync<Traits>(
            WaitingTaskHolder(taskGroup_, &globalWaitTask), *schedule_, transitionInfo, serviceToken_, emptyList);
        globalWaitTask.wait();
      }
      {
        //To wait, the ref count has to be 1+#streams
        FinalWaitingTask streamLoopWaitTask{taskGroup_};

        using Traits = OccurrenceTraits<LuminosityBlockPrincipal, BranchActionStreamBegin>;

        beginStreamsTransitionAsync<Traits>(WaitingTaskHolder(taskGroup_, &streamLoopWaitTask),
                                            *schedule_,
                                            preallocations_.numberOfStreams(),
                                            transitionInfo,
                                            serviceToken_,
                                            emptyList);

        streamLoopWaitTask.wait();
      }
      beginLumiCalled_ = true;
    }

    void TestProcessor::event() {
      auto pep = &(principalCache_.eventPrincipal(0));

      //this resets the EventPrincipal (if it had been used before)
      pep->clearEventPrincipal();
      edm::EventAuxiliary aux(EventID(runNumber_, lumiNumber_, eventNumber_), "", Timestamp(), false);
      aux.setProcessHistoryID(processHistory_.id());
      pep->fillEventPrincipal(aux, nullptr, nullptr);
      assert(lumiPrincipal_.get() != nullptr);
      pep->setLuminosityBlockPrincipal(lumiPrincipal_.get());

      for (auto& p : dataProducts_) {
        if (p.second) {
          pep->put(p.first, std::move(p.second), ProductProvenance(p.first.branchID()));
        } else {
          //The data product was not set so we need to
          // tell the ProductResolver not to wait
          auto r = pep->getProductResolver(p.first.branchID());
          dynamic_cast<ProductPutterBase const*>(r)->putProduct(std::unique_ptr<WrapperBase>());
        }
      }

      ServiceRegistry::Operate operate(serviceToken_);

      FinalWaitingTask waitTask{taskGroup_};

      EventTransitionInfo info(*pep, esp_->eventSetupImpl());
      schedule_->processOneEventAsync(edm::WaitingTaskHolder(taskGroup_, &waitTask), 0, info, serviceToken_);

      waitTask.wait();
      ++eventNumber_;
    }

    std::shared_ptr<LuminosityBlockPrincipal> TestProcessor::endLuminosityBlock() {
      auto lumiPrincipal = lumiPrincipal_;
      if (beginLumiCalled_) {
        //make the services available
        ServiceRegistry::Operate operate(serviceToken_);

        beginLumiCalled_ = false;
        lumiPrincipal_.reset();

        IOVSyncValue ts(EventID(runNumber_, lumiNumber_, eventNumber_), lumiPrincipal->endTime());
        eventsetup::synchronousEventSetupForInstance(ts, taskGroup_, *espController_);

        auto const& es = esp_->eventSetupImpl();

        LumiTransitionInfo transitionInfo(*lumiPrincipal, es, nullptr);

        std::vector<edm::SubProcess> emptyList;

        //To wait, the ref count has to be 1+#streams
        {
          FinalWaitingTask streamLoopWaitTask{taskGroup_};

          using Traits = OccurrenceTraits<LuminosityBlockPrincipal, BranchActionStreamEnd>;

          endStreamsTransitionAsync<Traits>(WaitingTaskHolder(taskGroup_, &streamLoopWaitTask),
                                            *schedule_,
                                            preallocations_.numberOfStreams(),
                                            transitionInfo,
                                            serviceToken_,
                                            emptyList,
                                            false);

          streamLoopWaitTask.wait();
        }
        {
          FinalWaitingTask globalWaitTask{taskGroup_};

          using Traits = OccurrenceTraits<LuminosityBlockPrincipal, BranchActionGlobalEnd>;
          endGlobalTransitionAsync<Traits>(WaitingTaskHolder(taskGroup_, &globalWaitTask),
                                           *schedule_,
                                           transitionInfo,
                                           serviceToken_,
                                           emptyList,
                                           false);
          globalWaitTask.wait();
        }
        {
          FinalWaitingTask globalWaitTask{taskGroup_};
          schedule_->writeLumiAsync(
              WaitingTaskHolder(taskGroup_, &globalWaitTask), *lumiPrincipal, &processContext_, actReg_.get());
          globalWaitTask.wait();
        }
      }
      lumiPrincipal->setRunPrincipal(std::shared_ptr<RunPrincipal>());
      return lumiPrincipal;
    }

    std::shared_ptr<edm::RunPrincipal> TestProcessor::endRun() {
      auto runPrincipal = runPrincipal_;
      runPrincipal_.reset();
      if (beginRunCalled_) {
        beginRunCalled_ = false;

        //make the services available
        ServiceRegistry::Operate operate(serviceToken_);

        IOVSyncValue ts(
            EventID(runPrincipal->run(), LuminosityBlockID::maxLuminosityBlockNumber(), EventID::maxEventNumber()),
            runPrincipal->endTime());
        eventsetup::synchronousEventSetupForInstance(ts, taskGroup_, *espController_);

        auto const& es = esp_->eventSetupImpl();

        RunTransitionInfo transitionInfo(*runPrincipal, es);

        std::vector<edm::SubProcess> emptyList;

        //To wait, the ref count has to be 1+#streams
        {
          FinalWaitingTask streamLoopWaitTask{taskGroup_};

          using Traits = OccurrenceTraits<RunPrincipal, BranchActionStreamEnd>;

          endStreamsTransitionAsync<Traits>(WaitingTaskHolder(taskGroup_, &streamLoopWaitTask),
                                            *schedule_,
                                            preallocations_.numberOfStreams(),
                                            transitionInfo,
                                            serviceToken_,
                                            emptyList,
                                            false);

          streamLoopWaitTask.wait();
        }
        {
          FinalWaitingTask globalWaitTask{taskGroup_};

          using Traits = OccurrenceTraits<RunPrincipal, BranchActionGlobalEnd>;
          endGlobalTransitionAsync<Traits>(WaitingTaskHolder(taskGroup_, &globalWaitTask),
                                           *schedule_,
                                           transitionInfo,
                                           serviceToken_,
                                           emptyList,
                                           false);
          globalWaitTask.wait();
        }
        {
          FinalWaitingTask globalWaitTask{taskGroup_};
          schedule_->writeRunAsync(WaitingTaskHolder(taskGroup_, &globalWaitTask),
                                   *runPrincipal,
                                   &processContext_,
                                   actReg_.get(),
                                   runPrincipal->mergeableRunProductMetadata());
          globalWaitTask.wait();
        }
      }
      return runPrincipal;
    }

    ProcessBlockPrincipal const* TestProcessor::endProcessBlock() {
      ProcessBlockPrincipal& processBlockPrincipal = principalCache_.processBlockPrincipal();
      if (beginProcessBlockCalled_) {
        beginProcessBlockCalled_ = false;

        std::vector<edm::SubProcess> emptyList;
        {
          FinalWaitingTask globalWaitTask{taskGroup_};

          ProcessBlockTransitionInfo transitionInfo(processBlockPrincipal);
          using Traits = OccurrenceTraits<ProcessBlockPrincipal, BranchActionGlobalEnd>;
          endGlobalTransitionAsync<Traits>(WaitingTaskHolder(taskGroup_, &globalWaitTask),
                                           *schedule_,
                                           transitionInfo,
                                           serviceToken_,
                                           emptyList,
                                           false);
          globalWaitTask.wait();
        }
      }
      return &processBlockPrincipal;
    }

    void TestProcessor::endJob() {
      if (!beginJobCalled_) {
        return;
      }
      beginJobCalled_ = false;

      // Collects exceptions, so we don't throw before all operations are performed.
      ExceptionCollector c(
          "Multiple exceptions were thrown while executing endJob. An exception message follows for each.\n");
      std::mutex collectorMutex;

      //make the services available
      ServiceRegistry::Operate operate(serviceToken_);

      //NOTE: this really should go elsewhere in the future
      for (unsigned int i = 0; i < preallocations_.numberOfStreams(); ++i) {
        schedule_->endStream(i, c, collectorMutex);
      }
      auto actReg = actReg_.get();
      c.call([actReg]() { actReg->preEndJobSignal_(); });
      schedule_->endJob(c);
      c.call([actReg]() { actReg->postEndJobSignal_(); });
      if (c.hasThrown()) {
        c.rethrow();
      }
    }

    void TestProcessor::setRunNumber(edm::RunNumber_t iRun) {
      if (beginRunCalled_) {
        endLuminosityBlock();
        endRun();
      }
      runNumber_ = iRun;
    }
    void TestProcessor::setLuminosityBlockNumber(edm::LuminosityBlockNumber_t iLumi) {
      endLuminosityBlock();
      lumiNumber_ = iLumi;
    }

    void TestProcessor::setEventNumber(edm::EventNumber_t iEv) { eventNumber_ = iEv; }

  }  // namespace test
}  // namespace edm