FWCore/Framework/src/TransitionProcessors.icc - master - githubmirror/cmssw

FWCore/Framework/src/TransitionProcessors.icc
221 строка6 KB
Chris Jones
Better handling of SourceStatus values
01 май 2026, 22:08
01 май 2026, 22:08930683a
//
//  TransitionProcessors.cpp
//
//  This file is intended to be included into other source files.
//  It was split into its own file to allow easier testing.
//
//  Created by Chris Jones on 6/29/17.
//
//

//Transition processing helpers

#include <cassert>
#include "FWCore/Utilities/interface/thread_safety_macros.h"

struct FileResources {
  FileResources(EventProcessor& iEP) : ep_(iEP) {}

  ~FileResources() {
    CMS_SA_ALLOW try {
      // Don't try to execute the following sequence of functions twice.
      // If the sequence was already attempted and failed, then do nothing.
      if (!closingSequenceAlreadyFailed_) {
        ep_.respondToCloseInputFile();
        ep_.closeInputFile(cleaningUpAfterException_);
        if (needToCloseOutputFiles_) {
          ep_.endProcessBlock(cleaningUpAfterException_, beginProcessBlockSucceeded_);
          ep_.closeOutputFiles();
        }
      }
    } catch (...) {
      if (cleaningUpAfterException_ or not ep_.setDeferredException(std::current_exception())) {
        std::string message(
            "Another exception was caught while trying to clean up files after the primary fatal exception.");
        ep_.setExceptionMessageFiles(message);
      }
    }
  }

  void normalEnd() { cleaningUpAfterException_ = false; }

  EventProcessor& ep_;
  bool cleaningUpAfterException_ = true;
  bool closingSequenceAlreadyFailed_ = false;
  bool beginProcessBlockSucceeded_ = false;
  bool needToCloseOutputFiles_ = false;
};

struct RunResources {
  RunResources(EventProcessor& iEP) noexcept : ep_(iEP) {}

  ~RunResources() noexcept {
    // Usually runs and lumis are closed inside processRuns and then endUnfinishedRun
    // and endUnfinishedLumi do nothing. We try to close runs and lumis even when an
    // exception is thrown in processRuns.  These calls are necessary when we are between
    // calls to processRuns because of a file transition and an exception is thrown or
    // there is an empty last file. In addition, this protects against rare cases where
    // processRuns returns with open runs or lumis.
    // Caught exception is propagated via EventProcessor::setDeferredException()
    CMS_SA_ALLOW try { ep_.endUnfinishedLumi(cleaningUpAfterException_); } catch (...) {
      if (cleaningUpAfterException_ or not ep_.setDeferredException(std::current_exception())) {
        ep_.setExceptionMessageLumis();
      }
    }
    CMS_SA_ALLOW try { ep_.endUnfinishedRun(cleaningUpAfterException_); } catch (...) {
      if (cleaningUpAfterException_ or not ep_.setDeferredException(std::current_exception())) {
        ep_.setExceptionMessageRuns();
      }
    }
  }

  void normalEnd() { cleaningUpAfterException_ = false; }

  EventProcessor& ep_;
  bool cleaningUpAfterException_ = true;
};

class RunsInFileProcessor {
public:
  [[nodiscard]] edm::SourceStatus processRuns(EventProcessor& iEP, edm::SourceStatus const& oSourceStatus) {
    if (!runResources_) {
      runResources_ = std::make_unique<RunResources>(iEP);
    }
    return iEP.processRuns(oSourceStatus);
  }

  void normalEnd() {
    if (runResources_) {
      runResources_->normalEnd();
      runResources_.reset();
    }
  }

private:
  std::unique_ptr<RunResources> runResources_;
};

class FilesProcessor {
public:
  explicit FilesProcessor(bool iDoNotMerge) : doNotMerge_(iDoNotMerge) {}

  [[nodiscard]] edm::SourceStatus processFiles(EventProcessor& iEP) {
    bool finished = false;
    auto nextTransition = iEP.thread_unsafe_peekNextTransitionType();
    if (nextTransition.nextTransitionType() != edm::InputSource::ItemType::IsFile)
      return nextTransition;
    do {
      switch (nextTransition.nextTransitionType()) {
        case edm::InputSource::ItemType::IsFile: {
          processFile(iEP);
          nextTransition = iEP.thread_unsafe_peekNextTransitionType();
          break;
        }
        case edm::InputSource::ItemType::IsRun: {
          nextTransition = runs_.processRuns(iEP, nextTransition);
          break;
        }
        default:
          finished = true;
      }
    } while (not finished);

    return nextTransition;
  }

  void normalEnd() {
    runs_.normalEnd();
    if (filesOpen_) {
      filesOpen_->normalEnd();
      filesOpen_.reset();
    }
  }

private:
  void processFile(EventProcessor& iEP) {
    if (not filesOpen_) {
      readFirstFile(iEP);
    } else {
      if (shouldWeCloseOutput(iEP)) {
        //Need to end this run on the file boundary
        runs_.normalEnd();
        gotoNewInputAndOutputFiles(iEP);
      } else {
        gotoNewInputFile(iEP);
      }
    }
  }

  void readFirstFile(EventProcessor& iEP) {
    iEP.readFile();
    assert(iEP.fileBlockValid());
    iEP.respondToOpenInputFile();

    iEP.openOutputFiles();
    filesOpen_ = std::make_unique<FileResources>(iEP);
    filesOpen_->needToCloseOutputFiles_ = true;

    iEP.beginProcessBlock(filesOpen_->beginProcessBlockSucceeded_);
    iEP.inputProcessBlocks();
  }

  bool shouldWeCloseOutput(EventProcessor& iEP) {
    if (doNotMerge_)
      return true;
    return iEP.shouldWeCloseOutput();
  }

  void gotoNewInputFile(EventProcessor& iEP) {
    iEP.respondToCloseInputFile();
    iEP.closeInputFile(false);

    iEP.readFile();
    if (!iEP.fileBlockValid()) {
      // handle case with last file bad and
      // skipBadFiles true
      return;
    }
    iEP.respondToOpenInputFile();

    iEP.inputProcessBlocks();
  }

  void gotoNewInputAndOutputFiles(EventProcessor& iEP) {
    {
      // If this is still true when we hit the destructor for
      // the filesOpen_ object, then we will know an exception
      // was thrown on one of the following 5 lines.
      filesOpen_->closingSequenceAlreadyFailed_ = true;

      iEP.respondToCloseInputFile();
      bool cleaningUpAfterException = false;
      iEP.closeInputFile(cleaningUpAfterException);
      iEP.endProcessBlock(cleaningUpAfterException, filesOpen_->beginProcessBlockSucceeded_);
      iEP.closeOutputFiles();

      filesOpen_->needToCloseOutputFiles_ = false;
      filesOpen_->closingSequenceAlreadyFailed_ = false;
    }
    {
      filesOpen_->beginProcessBlockSucceeded_ = false;

      iEP.readFile();
      if (!iEP.fileBlockValid()) {
        // handle case with last file bad and
        // skipBadFiles true
        return;
      }
      iEP.respondToOpenInputFile();

      iEP.openOutputFiles();
      filesOpen_->needToCloseOutputFiles_ = true;

      iEP.beginProcessBlock(filesOpen_->beginProcessBlockSucceeded_);
    }
    iEP.inputProcessBlocks();
  }

  std::unique_ptr<FileResources> filesOpen_;
  RunsInFileProcessor runs_;
  bool doNotMerge_;
};