IOPool/Streamer/src/StreamerInputSource.cc - master - githubmirror/cmssw

IOPool/Streamer/src/StreamerInputSource.cc
364 строки16 KB
cmsbuild
Merge pull request #50290 from Dr15Jones/removeThinning
17 мар 2026, 12:33
17 мар 2026, 12:336adc8d1
#include "IOPool/Streamer/interface/StreamerInputSource.h"

#include "IOPool/Streamer/interface/EventMessage.h"
#include "IOPool/Streamer/interface/InitMessage.h"
#include "IOPool/Streamer/interface/ClassFiller.h"
#include "IOPool/Streamer/interface/uncompress.h"

#include "FWCore/Framework/interface/EventPrincipal.h"
#include "FWCore/Framework/interface/FileBlock.h"
#include "DataFormats/Provenance/interface/ProductDescription.h"
#include "DataFormats/Provenance/interface/ProductProvenance.h"
#include "DataFormats/Provenance/interface/EventAuxiliary.h"
#include "DataFormats/Provenance/interface/LuminosityBlockAuxiliary.h"
#include "DataFormats/Provenance/interface/RunAuxiliary.h"
#include "DataFormats/Provenance/interface/EventSelectionID.h"
#include "DataFormats/Provenance/interface/BranchIDListHelper.h"
#include "DataFormats/Provenance/interface/BranchListIndex.h"

#include "DataFormats/Common/interface/RefCoreStreamer.h"
#include "FWCore/Utilities/interface/WrappedClassName.h"
#include "FWCore/Utilities/interface/Exception.h"
#include "FWCore/ParameterSet/interface/ParameterSet.h"
#include "FWCore/ParameterSet/interface/Registry.h"
#include "FWCore/Utilities/interface/EDMException.h"
#include "FWCore/Utilities/interface/Adler32Calculator.h"
#include "FWCore/Reflection/interface/DictionaryTools.h"

#include "DataFormats/Provenance/interface/ProductRegistry.h"
#include "DataFormats/Provenance/interface/ProcessHistoryRegistry.h"

#include <string>
#include <iostream>
#include <set>

namespace edm::streamer {
  namespace {
    int const init_size = 1024 * 1024;
  }

  StreamerInputSource::StreamerInputSource(ParameterSet const& pset, InputSourceDescription const& desc)
      : RawInputSource(pset, desc),
        tc_(getTClass(typeid(SendEvent))),
        dest_(init_size),
        xbuf_(TBuffer::kRead, init_size),
        sendEvent_(),
        eventPrincipalHolder_(),
        processName_(),
        protocolVersion_(0U) {}

  StreamerInputSource::~StreamerInputSource() {}

  // ---------------------------------------
  void StreamerInputSource::mergeIntoRegistry(SendJobHeader const& header, ProductRegistry& reg, bool subsequent) {
    SendDescs const& descs = header.descs();

    std::set<std::string> processNames;
    for (auto const& item : descs) {
      processNames.insert(item.processName());
    }
    std::vector<std::string> orderedProcessNames;
    orderedProcessNames.reserve(processNames.size());
    if (processNames.size() == 1) {
      orderedProcessNames.push_back(*processNames.begin());
    } else {
      if (processNames.size() == 2) {
        //The LHC name is injected by the DAQProvenanceHelper
        auto it = processNames.find("LHC");
        if (it != processNames.end()) {
          processNames.erase(it);
          orderedProcessNames.push_back(*processNames.begin());
          orderedProcessNames.push_back("LHC");
        }
      }
    }
    if (orderedProcessNames.empty()) {
      cms::Exception toThrow("MismatchedInput", "StreamerInputSource::mergeIntoRegistry");
      toThrow << "Could not determine process name order from input file(s). Found process names: ";
      for (auto const& pn : processNames) {
        toThrow << "\n  " << pn;
      }
      throw toThrow;
    }
    if (subsequent) {
      ProductRegistry pReg;
      pReg.updateFromInput(descs, orderedProcessNames);
      std::string mergeInfo = reg.merge(pReg, std::string(), ProductDescription::Permissive);
      if (!mergeInfo.empty()) {
        throw cms::Exception("MismatchedInput", "StreamerInputSource::mergeIntoRegistry") << mergeInfo;
      }
    } else {
      declareStreamers(descs);
      buildClassCache(descs);
      loadExtraClasses();
      if (!reg.frozen()) {
        reg.updateFromInput(descs, orderedProcessNames);
      }
    }
  }

  void StreamerInputSource::declareStreamers(SendDescs const& descs) {
    std::vector<std::string> missingDictionaries;
    std::vector<std::string> branchNamesForMissing;
    std::vector<std::string> producedTypes;
    for (auto const& item : descs) {
      //pi->init();
      std::string const real_name = wrappedClassName(item.className());
      if (!loadCap(real_name, missingDictionaries)) {
        branchNamesForMissing.emplace_back(item.branchName());
        producedTypes.emplace_back(item.className() + std::string(" (read from input)"));
      }
    }
    if (!missingDictionaries.empty()) {
      std::string context("Calling StreamerInputSource::declareStreamers, checking dictionaries for input types");
      throwMissingDictionariesException(missingDictionaries, context, producedTypes, branchNamesForMissing, true);
    }
  }

  void StreamerInputSource::buildClassCache(SendDescs const& descs) {
    for (auto const& item : descs) {
      //pi->init();
      std::string const real_name = wrappedClassName(item.className());
      doBuildRealData(real_name);
    }
  }

  /**
   * Deserializes the specified init message into a SendJobHeader object
   * (which is related to the product registry).
   */
  std::unique_ptr<SendJobHeader> StreamerInputSource::deserializeRegistry(InitMsgView const& initView) {
    if (initView.code() != Header::INIT)
      throw cms::Exception("StreamTranslation", "Registry deserialization error")
          << "received wrong message type: expected INIT, got " << initView.code() << "\n";

    //Get the process name and store if for Protocol version 4 and above.
    if (initView.protocolVersion() > 3) {
      processName_ = initView.processName();
      protocolVersion_ = initView.protocolVersion();
    }

    // calculate the adler32 checksum
    uint32_t adler32_chksum = cms::Adler32((char const*)initView.descData(), initView.descLength());
    //std::cout << "Adler32 checksum of init message = " << adler32_chksum << std::endl;
    //std::cout << "Adler32 checksum of init messsage from header = " << initView.adler32_chksum() << " "
    //          << "host name = " << initView.hostName() << " len = " << initView.hostName_len() << std::endl;
    if ((uint32)adler32_chksum != initView.adler32_chksum()) {
      // skip event (based on option?) or throw exception?
      throw cms::Exception("StreamDeserialization", "Checksum error")
          << " chksum from registry data = " << adler32_chksum << " from header = " << initView.adler32_chksum()
          << " host name = " << initView.hostName() << std::endl;
    }

    TClass* desc = getTClass(typeid(SendJobHeader));

    TBufferFile xbuf(
        TBuffer::kRead, initView.descLength(), const_cast<char*>((char const*)initView.descData()), kFALSE);
    std::unique_ptr<SendJobHeader> sd((SendJobHeader*)xbuf.ReadObjectAny(desc));

    if (sd.get() == nullptr) {
      throw cms::Exception("StreamTranslation", "Registry deserialization error")
          << "Could not read the initial product registry list\n";
    }

    sd->initializeTransients();
    return sd;
  }

  /**
   * Deserializes the specified init message into a SendJobHeader object
   * and merges registries.
   */
  void StreamerInputSource::deserializeAndMergeWithRegistry(InitMsgView const& initView, bool subsequent) {
    std::unique_ptr<SendJobHeader> sd = deserializeRegistry(initView);
    mergeIntoRegistry(*sd, productRegistryUpdate(), subsequent);
    SendJobHeader::ParameterSetMap const& psetMap = sd->processParameterSet();
    pset::Registry& psetRegistry = *pset::Registry::instance();
    for (auto const& item : psetMap) {
      ParameterSet pset(item.second.pset());
      pset.setID(item.first);
      psetRegistry.insertMapped(pset);
    }
  }

  void StreamerInputSource::updateEventMetaData() {
    branchIDListHelper()->updateFromInput(sendEvent_->branchIDLists());
  }

  uint32_t StreamerInputSource::eventMetaDataChecksum(EventMsgView const& eventView) const {
    return eventView.adler32_chksum();
  }

  void StreamerInputSource::deserializeEventMetaData(EventMsgView const& eventView) {
    deserializeEventCommon(eventView, true);
  }
  /**
   * Deserializes the specified event message.
   */
  void StreamerInputSource::deserializeEvent(EventMsgView const& eventView) {
    deserializeEventCommon(eventView, false);
  }

  void StreamerInputSource::deserializeEventCommon(EventMsgView const& eventView, bool isMetaData) {
    if (eventView.code() != Header::EVENT)
      throw cms::Exception("StreamTranslation", "Event deserialization error")
          << "received wrong message type: expected EVENT, got " << eventView.code() << "\n";
    // uncompress if we need to
    // 78 was a dummy value (for no uncompressed) - should be 0 for uncompressed
    // need to get rid of this when 090 MTCC streamers are gotten rid of
    unsigned long origsize = eventView.origDataSize();
    unsigned long dest_size;  //(should be >= eventView.origDataSize())

    uint32_t adler32_chksum = cms::Adler32((char const*)eventView.eventData(), eventView.eventLength());
    //std::cout << "Adler32 checksum of event = " << adler32_chksum << std::endl;
    //std::cout << "Adler32 checksum from header = " << eventView.adler32_chksum() << " "
    //          << "host name = " << eventView.hostName() << " len = " << eventView.hostName_len() << std::endl;
    if (static_cast<uint32>(adler32_chksum) != eventView.adler32_chksum()) {
      // skip event (based on option?) or throw exception?
      throw cms::Exception("StreamDeserialization", "Checksum error")
          << " chksum from event = " << adler32_chksum << " from header = " << eventView.adler32_chksum()
          << " host name = " << eventView.hostName() << std::endl;
    }
    if (origsize != 78 && origsize != 0) {
      // compressed
      dest_size = edm::streamer::uncompress::uncompressBuffer(
          (unsigned char const*)eventView.eventData(), eventView.eventLength(), dest_, origsize);
    } else {  // not compressed
      // we need to copy anyway the buffer as we are using dest in xbuf
      dest_size = eventView.eventLength();
      dest_.resize(dest_size);
      unsigned char* pos = (unsigned char*)&dest_[0];
      unsigned char const* from = (unsigned char const*)eventView.eventData();
      std::copy(from, from + dest_size, pos);
    }
    //TBuffer xbuf(TBuffer::kRead, dest_size,
    //             (char const*) &dest[0],kFALSE);
    //TBuffer xbuf(TBuffer::kRead, eventView.eventLength(),
    //             (char const*) eventView.eventData(),kFALSE);
    xbuf_.Reset();
    xbuf_.SetBuffer(&dest_[0], dest_size, kFALSE);

    //We do not yet know which EventPrincipal we will use, therefore
    // we are using a new EventPrincipalHolder as a proxy. We need to
    // make a new one instead of reusing the same one becuase when running
    // multi-threaded there will be multiple EventPrincipals being used
    // simultaneously.
    eventPrincipalHolder_ = std::make_unique<EventPrincipalHolder>();  // propagate_const<T> has no reset() function
    {
      RefCoreStreamerGuard guard(eventPrincipalHolder_.get());
      sendEvent_ = std::unique_ptr<SendEvent>(reinterpret_cast<SendEvent*>(xbuf_.ReadObjectAny(tc_)));
    }

    if (sendEvent_.get() == nullptr) {
      throw cms::Exception("StreamTranslation", "Event deserialization error")
          << "got a null event from input stream\n";
    }

    if (isMetaData) {
      eventMetaDataChecksum_ = adler32_chksum;
      return;
    }

    if (sendEvent_->metaDataChecksum() != eventMetaDataChecksum_) {
      throw cms::Exception("StreamTranslation") << " meta data checksum from event " << sendEvent_->metaDataChecksum()
                                                << " does not match last read meta data " << eventMetaDataChecksum_;
    }

    processHistoryRegistryForUpdate().registerProcessHistory(sendEvent_->processHistory());

    if (runAuxiliary().get() == nullptr || runAuxiliary()->run() != sendEvent_->aux().run() ||
        runAuxiliary()->processHistoryID() != sendEvent_->processHistory().id()) {
      RunAuxiliary* runAuxiliary =
          new RunAuxiliary(sendEvent_->aux().run(), sendEvent_->aux().time(), Timestamp::invalidTimestamp());
      runAuxiliary->setProcessHistoryID(sendEvent_->processHistory().id());
      setRunAuxiliary(runAuxiliary);
      resetLuminosityBlockAuxiliary();
    }
    if (!luminosityBlockAuxiliary() || luminosityBlockAuxiliary()->luminosityBlock() != eventView.lumi()) {
      LuminosityBlockAuxiliary* luminosityBlockAuxiliary = new LuminosityBlockAuxiliary(
          runAuxiliary()->run(), eventView.lumi(), sendEvent_->aux().time(), Timestamp::invalidTimestamp());
      luminosityBlockAuxiliary->setProcessHistoryID(sendEvent_->processHistory().id());
      setLuminosityBlockAuxiliary(luminosityBlockAuxiliary);
    }
    setEventCached();
  }

  void StreamerInputSource::read(EventPrincipal& eventPrincipal) {
    EventSelectionIDVector ids(sendEvent_->eventSelectionIDs());
    BranchListIndexes indexes(sendEvent_->branchListIndexes());
    branchIDListHelper()->fixBranchListIndexes(indexes);
    auto history = processHistoryRegistry().getMapped(sendEvent_->aux().processHistoryID());
    eventPrincipal.fillEventPrincipal(sendEvent_->aux(), history, std::move(ids), std::move(indexes));

    //We now know which eventPrincipal to use and we can reuse the slot in
    // streamToEventPrincipalHolders to own the memory
    eventPrincipalHolder_->setEventPrincipal(&eventPrincipal);
    if (streamToEventPrincipalHolders_.size() < eventPrincipal.streamID().value() + 1) {
      streamToEventPrincipalHolders_.resize(eventPrincipal.streamID().value() + 1);
    }
    streamToEventPrincipalHolders_[eventPrincipal.streamID().value()] = std::move(eventPrincipalHolder_);

    // no process name list handling

    SendProds& sps = sendEvent_->products();
    for (auto& spitem : sps) {
      if (spitem.desc() == nullptr)
        throw cms::Exception("StreamTranslation", "Empty Provenance");

      ProductDescription const branchDesc(*spitem.desc());
      // This ProductProvenance constructor inserts into the entry description registry
      if (spitem.parents()) {
        std::optional<ProductProvenance> productProvenance{std::in_place, spitem.branchID(), *spitem.parents()};
        if (spitem.prod() != nullptr) {
          eventPrincipal.putOnRead(
              branchDesc, std::unique_ptr<WrapperBase>(const_cast<WrapperBase*>(spitem.prod())), productProvenance);
        } else {
          eventPrincipal.putOnRead(branchDesc, std::unique_ptr<WrapperBase>(), productProvenance);
        }
      } else {
        std::optional<ProductProvenance> productProvenance;
        if (spitem.prod() != nullptr) {
          eventPrincipal.putOnRead(
              branchDesc, std::unique_ptr<WrapperBase>(const_cast<WrapperBase*>(spitem.prod())), productProvenance);
        } else {
          eventPrincipal.putOnRead(branchDesc, std::unique_ptr<WrapperBase>(), productProvenance);
        }
      }
      spitem.clear();
    }
  }

  void StreamerInputSource::resetAfterEndRun() {
    // called from an online streamer source to reset after a stop command
    // so an enable command will work
    resetLuminosityBlockAuxiliary();
    resetRunAuxiliary();
    assert(!eventCached());
    reset();
  }

  void StreamerInputSource::setRun(RunNumber_t) {
    // Need to define a dummy setRun here or else the InputSource::setRun is called
    // if we have a source inheriting from this and wants to define a setRun method
    throw Exception(errors::LogicError) << "StreamerInputSource::setRun()\n"
                                        << "Run number cannot be modified for this type of Input Source\n"
                                        << "Contact a Storage Manager Developer\n";
  }

  StreamerInputSource::EventPrincipalHolder::EventPrincipalHolder() : eventPrincipal_(nullptr) {}

  StreamerInputSource::EventPrincipalHolder::~EventPrincipalHolder() {}

  WrapperBase const* StreamerInputSource::EventPrincipalHolder::getIt(ProductID const& id) const {
    return eventPrincipal_ ? eventPrincipal_->getIt(id) : nullptr;
  }

  unsigned int StreamerInputSource::EventPrincipalHolder::transitionIndex_() const {
    assert(eventPrincipal_ != nullptr);
    return eventPrincipal_->transitionIndex();
  }

  void StreamerInputSource::EventPrincipalHolder::setEventPrincipal(EventPrincipal* ep) { eventPrincipal_ = ep; }

  void StreamerInputSource::fillDescription(ParameterSetDescription& desc) { RawInputSource::fillDescription(desc); }
}  // namespace edm::streamer