TestProcessor.cc

Go to the documentation of this file.

// -*- C++ -*-
//
// Package:     Subsystem/Package
// Class  :     TestProcessor
//
// Implementation:
//     [Notes on implementation]
//
// Original Author:  root
//         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/Framework/interface/ScheduleItems.h"
#include "FWCore/Framework/interface/EventPrincipal.h"
#include "FWCore/Framework/interface/LuminosityBlockPrincipal.h"
#include "FWCore/Framework/interface/ExceptionActions.h"
#include "FWCore/Framework/interface/HistoryAppender.h"
#include "FWCore/Framework/interface/PathsAndConsumesOfModules.h"
#include "FWCore/Framework/interface/ESRecordsToProxyIndices.h"
#include "FWCore/Framework/src/EventSetupsController.h"
#include "FWCore/Framework/src/streamTransitionAsync.h"
#include "FWCore/Framework/src/globalTransitionAsync.h"
#include "FWCore/Framework/interface/DelayedReader.h"

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

#include "FWCore/PluginManager/interface/PluginManager.h"
#include "FWCore/PluginManager/interface/standard.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 "DataFormats/Provenance/interface/ParentageRegistry.h"

#include "tbb/task_scheduler_init.h"

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

namespace edm {
  namespace test {

    //
    // constants, enums and typedefs
    //

    //
    // static data member definitions
    //
    namespace {

      bool oneTimeInitializationImpl() {
        edmplugin::PluginManager::configure(edmplugin::standard::config());

        static std::unique_ptr<tbb::task_scheduler_init> tsiPtr = std::make_unique<tbb::task_scheduler_init>(1);

        // register the empty parentage vector , once and for all
        ParentageRegistry::instance()->insertMapped(Parentage());

        // register the empty parameter set, once and for all.
        ParameterSet().registerIt();
        return true;
      }

      bool oneTimeInitialization() {
        static const bool s_init{oneTimeInitializationImpl()};
        return s_init;
      }
    }  // namespace

    //
    // constructors and destructor
    //
    TestProcessor::TestProcessor(Config const& iConfig)
        : espController_(std::make_unique<eventsetup::EventSetupsController>()),
          historyAppender_(std::make_unique<HistoryAppender>()),
          moduleRegistry_(std::make_shared<ModuleRegistry>()) {
      //Setup various singletons
      (void)oneTimeInitialization();

      ProcessDescImpl desc(iConfig.pythonConfiguration());

      auto psetPtr = desc.parameterSet();

      validateTopLevelParameterSets(psetPtr.get());

      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, ServiceToken(), 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());

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

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

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

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

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

      //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);
      }

      schedule_ = items.initSchedule(*psetPtr, false, preallocations_, &processContext_);
      // 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_.setProcessHistoryRegistry(processHistoryRegistry_);

      principalCache_.setNumberOfConcurrentPrincipals(preallocations_);

      preg_->setFrozen();
      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_.numberOfLuminosityBlocks(); ++index) {
        auto lp =
            std::make_unique<LuminosityBlockPrincipal>(preg_, *processConfiguration_, historyAppender_.get(), index);
        principalCache_.insert(std::move(lp));
      }
    }

    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() {
      setupProcessing();
      event();

      bool result = 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);
    }

    void TestProcessor::setupProcessing() {
      if (not beginJobCalled_) {
        beginJob();
      }
      if (not beginRunCalled_) {
        beginRun();
      }
      if (not beginLumiCalled_) {
        beginLuminosityBlock();
      }
    }

    void TestProcessor::teardownProcessing() {
      if (beginLumiCalled_) {
        endLuminosityBlock();
        beginLumiCalled_ = false;
      }
      if (beginRunCalled_) {
        endRun();
        beginRunCalled_ = false;
      }
      if (beginJobCalled_) {
        endJob();
      }
    }

    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_);

      //NOTE: this may throw
      //checkForModuleDependencyCorrectness(pathsAndConsumesOfModules, false);
      actReg_->preBeginJobSignal_(pathsAndConsumesOfModules, processContext_);

      schedule_->beginJob(*preg_, esp_->recordsToProxyIndices());
      actReg_->postBeginJobSignal_();

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

    void TestProcessor::beginRun() {
      ProcessHistoryID phid;
      auto aux = std::make_shared<RunAuxiliary>(runNumber_, Timestamp(), Timestamp());
      auto rp = std::make_shared<RunPrincipal>(aux, preg_, *processConfiguration_, historyAppender_.get(), 0);

      principalCache_.insert(rp);
      RunPrincipal& runPrincipal = principalCache_.runPrincipal(phid, runNumber_);

      IOVSyncValue ts(EventID(runPrincipal.run(), 0, 0), runPrincipal.beginTime());
      espController_->eventSetupForInstance(ts);

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

      std::vector<edm::SubProcess> emptyList;
      {
        typedef OccurrenceTraits<RunPrincipal, BranchActionGlobalBegin> Traits;
        auto globalWaitTask = make_empty_waiting_task();
        globalWaitTask->increment_ref_count();
        beginGlobalTransitionAsync<Traits>(
            WaitingTaskHolder(globalWaitTask.get()), *schedule_, runPrincipal, ts, es, serviceToken_, emptyList);
        globalWaitTask->wait_for_all();
        if (globalWaitTask->exceptionPtr() != nullptr) {
          std::rethrow_exception(*(globalWaitTask->exceptionPtr()));
        }
      }
      {
        //To wait, the ref count has to be 1+#streams
        auto streamLoopWaitTask = make_empty_waiting_task();
        streamLoopWaitTask->increment_ref_count();

        typedef OccurrenceTraits<RunPrincipal, BranchActionStreamBegin> Traits;

        beginStreamsTransitionAsync<Traits>(streamLoopWaitTask.get(),
                                            *schedule_,
                                            preallocations_.numberOfStreams(),
                                            runPrincipal,
                                            ts,
                                            es,
                                            serviceToken_,
                                            emptyList);

        streamLoopWaitTask->wait_for_all();
        if (streamLoopWaitTask->exceptionPtr() != nullptr) {
          std::rethrow_exception(*(streamLoopWaitTask->exceptionPtr()));
        }
      }
      beginRunCalled_ = true;
    }

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

      lumiPrincipal_->setRunPrincipal(principalCache_.runPrincipalPtr());

      IOVSyncValue ts(EventID(runNumber_, lumiNumber_, eventNumber_), lumiPrincipal_->beginTime());
      espController_->eventSetupForInstance(ts);

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

      std::vector<edm::SubProcess> emptyList;
      {
        typedef OccurrenceTraits<LuminosityBlockPrincipal, BranchActionGlobalBegin> Traits;
        auto globalWaitTask = make_empty_waiting_task();
        globalWaitTask->increment_ref_count();
        beginGlobalTransitionAsync<Traits>(
            WaitingTaskHolder(globalWaitTask.get()), *schedule_, *lumiPrincipal_, ts, es, serviceToken_, emptyList);
        globalWaitTask->wait_for_all();
        if (globalWaitTask->exceptionPtr() != nullptr) {
          std::rethrow_exception(*(globalWaitTask->exceptionPtr()));
        }
      }
      {
        //To wait, the ref count has to be 1+#streams
        auto streamLoopWaitTask = make_empty_waiting_task();
        streamLoopWaitTask->increment_ref_count();

        typedef OccurrenceTraits<LuminosityBlockPrincipal, BranchActionStreamBegin> Traits;

        beginStreamsTransitionAsync<Traits>(streamLoopWaitTask.get(),
                                            *schedule_,
                                            preallocations_.numberOfStreams(),
                                            *lumiPrincipal_,
                                            ts,
                                            es,
                                            serviceToken_,
                                            emptyList);

        streamLoopWaitTask->wait_for_all();
        if (streamLoopWaitTask->exceptionPtr() != nullptr) {
          std::rethrow_exception(*(streamLoopWaitTask->exceptionPtr()));
        }
      }
      beginLumiCalled_ = true;
    }

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

      //this resets the EventPrincipal (if it had been used before)
      pep->clearEventPrincipal();
      pep->fillEventPrincipal(
          edm::EventAuxiliary(EventID(runNumber_, lumiNumber_, eventNumber_), "", Timestamp(), false),
          processHistoryRegistry_,
          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());
          r->putProduct(std::unique_ptr<WrapperBase>());
        }
      }

      ServiceRegistry::Operate operate(serviceToken_);

      auto waitTask = make_empty_waiting_task();
      waitTask->increment_ref_count();

      schedule_->processOneEventAsync(
          edm::WaitingTaskHolder(waitTask.get()), 0, *pep, esp_->eventSetup(), serviceToken_);

      waitTask->wait_for_all();
      if (waitTask->exceptionPtr() != nullptr) {
        std::rethrow_exception(*(waitTask->exceptionPtr()));
      }
      ++eventNumber_;
    }

    void TestProcessor::endLuminosityBlock() {
      if (beginLumiCalled_) {
        beginLumiCalled_ = false;
        auto lumiPrincipal = lumiPrincipal_;
        lumiPrincipal_.reset();

        IOVSyncValue ts(EventID(runNumber_, lumiNumber_, eventNumber_), lumiPrincipal->endTime());
        espController_->eventSetupForInstance(ts);

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

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

        //To wait, the ref count has to be 1+#streams
        {
          auto streamLoopWaitTask = make_empty_waiting_task();
          streamLoopWaitTask->increment_ref_count();

          using Traits = OccurrenceTraits<LuminosityBlockPrincipal, BranchActionStreamEnd>;

          endStreamsTransitionAsync<Traits>(WaitingTaskHolder(streamLoopWaitTask.get()),
                                            *schedule_,
                                            preallocations_.numberOfStreams(),
                                            *lumiPrincipal,
                                            ts,
                                            es,
                                            serviceToken_,
                                            emptyList,
                                            false);

          streamLoopWaitTask->wait_for_all();
          if (streamLoopWaitTask->exceptionPtr() != nullptr) {
            std::rethrow_exception(*(streamLoopWaitTask->exceptionPtr()));
          }
        }
        {
          auto globalWaitTask = make_empty_waiting_task();
          globalWaitTask->increment_ref_count();

          typedef OccurrenceTraits<LuminosityBlockPrincipal, BranchActionGlobalEnd> Traits;
          endGlobalTransitionAsync<Traits>(WaitingTaskHolder(globalWaitTask.get()),
                                           *schedule_,
                                           *lumiPrincipal,
                                           ts,
                                           es,
                                           serviceToken_,
                                           emptyList,
                                           false);
          globalWaitTask->wait_for_all();
          if (globalWaitTask->exceptionPtr() != nullptr) {
            std::rethrow_exception(*(globalWaitTask->exceptionPtr()));
          }
        }
      }
    }

    void TestProcessor::endRun() {
      if (beginRunCalled_) {
        beginRunCalled_ = false;
        ProcessHistoryID phid;

        RunPrincipal& runPrincipal = principalCache_.runPrincipal(phid, runNumber_);

        IOVSyncValue ts(
            EventID(runPrincipal.run(), LuminosityBlockID::maxLuminosityBlockNumber(), EventID::maxEventNumber()),
            runPrincipal.endTime());
        espController_->eventSetupForInstance(ts);

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

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

        //To wait, the ref count has to be 1+#streams
        {
          auto streamLoopWaitTask = make_empty_waiting_task();
          streamLoopWaitTask->increment_ref_count();

          using Traits = OccurrenceTraits<RunPrincipal, BranchActionStreamEnd>;

          endStreamsTransitionAsync<Traits>(WaitingTaskHolder(streamLoopWaitTask.get()),
                                            *schedule_,
                                            preallocations_.numberOfStreams(),
                                            runPrincipal,
                                            ts,
                                            es,
                                            serviceToken_,
                                            emptyList,
                                            false);

          streamLoopWaitTask->wait_for_all();
          if (streamLoopWaitTask->exceptionPtr() != nullptr) {
            std::rethrow_exception(*(streamLoopWaitTask->exceptionPtr()));
          }
        }
        {
          auto globalWaitTask = make_empty_waiting_task();
          globalWaitTask->increment_ref_count();

          typedef OccurrenceTraits<RunPrincipal, BranchActionGlobalEnd> Traits;
          endGlobalTransitionAsync<Traits>(WaitingTaskHolder(globalWaitTask.get()),
                                           *schedule_,
                                           runPrincipal,
                                           ts,
                                           es,
                                           serviceToken_,
                                           emptyList,
                                           false);
          globalWaitTask->wait_for_all();
          if (globalWaitTask->exceptionPtr() != nullptr) {
            std::rethrow_exception(*(globalWaitTask->exceptionPtr()));
          }
        }

        principalCache_.deleteRun(phid, runNumber_);
      }
    }

    void TestProcessor::endJob() {
      // 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");

      //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) {
        c.call([this, i]() { this->schedule_->endStream(i); });
      }
      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; }

    //
    // const member functions
    //

    //
    // static member functions
    //

  }  // namespace test
}  // namespace edm