art::RootDelayedReader Class Reference

#include <RootDelayedReader.h>

Inheritance diagram for art::RootDelayedReader:

Public Member Functions
	~RootDelayedReader ()
	RootDelayedReader (RootDelayedReader const &)=delete
RootDelayedReader &	operator= (RootDelayedReader const &)=delete
	RootDelayedReader (RootDelayedReader &&)=delete
RootDelayedReader &	operator= (RootDelayedReader &&)=delete
	RootDelayedReader (FileFormatVersion, sqlite3 *db, std::vector< input::EntryNumber > const &entrySet, cet::exempt_ptr< input::BranchMap const >, TBranch *provenanceBranch, int64_t saveMemoryObjectThreshold, cet::exempt_ptr< RootInputFile > primaryFile, cet::exempt_ptr< BranchIDLists const > branchIDLists, BranchType branchType, EventID, bool compactSubRunRanges)
Public Member Functions inherited from art::DelayedReader
virtual	~DelayedReader () noexcept
	DelayedReader ()
std::unique_ptr< EDProduct >	getProduct (Group const *, ProductID, RangeSet &) const
void	setPrincipal (cet::exempt_ptr< Principal >)
std::vector< ProductProvenance >	readProvenance () const
bool	isAvailableAfterCombine (ProductID) const
int	openNextSecondaryFile (int idx)

Private Member Functions
std::unique_ptr< EDProduct >	getProduct_ (Group const *, ProductID, RangeSet &) const override
void	setPrincipal_ (cet::exempt_ptr< Principal >) override
std::vector< ProductProvenance >	readProvenance_ () const override
bool	isAvailableAfterCombine_ (ProductID) const override
int	openNextSecondaryFile_ (int idx) override

Private Attributes
FileFormatVersion	fileFormatVersion_
sqlite3 *	db_
std::vector< input::EntryNumber > const	entrySet_
cet::exempt_ptr< input::BranchMap const >	branches_
TBranch *	provenanceBranch_
int64_t	saveMemoryObjectThreshold_
cet::exempt_ptr< Principal >	principal_
cet::exempt_ptr< RootInputFile >	primaryFile_
cet::exempt_ptr< BranchIDLists const >	branchIDLists_
BranchType	branchType_
EventID	eventID_
bool const	compactSubRunRanges_

Detailed Description

Definition at line 32 of file RootDelayedReader.h.

Constructor & Destructor Documentation

art::RootDelayedReader::~RootDelayedReader ( )

default

art::RootDelayedReader::RootDelayedReader ( RootDelayedReader const &  )

delete

art::RootDelayedReader::RootDelayedReader ( RootDelayedReader &&  )

delete

art::RootDelayedReader::RootDelayedReader	(	FileFormatVersion	version,
		sqlite3 *	db,
		std::vector< input::EntryNumber > const &	entrySet,
		cet::exempt_ptr< input::BranchMap const >	branches,
		TBranch *	provenanceBranch,
		int64_t	saveMemoryObjectThreshold,
		cet::exempt_ptr< RootInputFile >	primaryFile,
		cet::exempt_ptr< BranchIDLists const >	branchIDLists,
		BranchType	branchType,
		EventID	eID,
		bool	compactSubRunRanges
	)

Definition at line 34 of file RootDelayedReader.cc.

    : DelayedReader()
    , fileFormatVersion_{version}
    , db_{db}
    , entrySet_{entrySet}
    , branches_{branches}
    , provenanceBranch_{provenanceBranch}
    , saveMemoryObjectThreshold_{saveMemoryObjectThreshold}
    , primaryFile_{primaryFile}
    , branchIDLists_{bidLists}
    , branchType_{branchType}
    , eventID_{eID}
    , compactSubRunRanges_{compactSubRunRanges}
  {}

Member Function Documentation

unique_ptr< EDProduct > art::RootDelayedReader::getProduct_ ( Group const *  grp, ProductID  pid, RangeSet &  rs  ) const

overrideprivatevirtual

Implements art::DelayedReader.

Definition at line 130 of file RootDelayedReader.cc.

  {
    auto iter = branches_->find(pid);
    assert(iter != branches_->end());
    input::BranchInfo const& branchInfo = iter->second;
    TBranch* br = branchInfo.productBranch_;
    assert(br != nullptr);
    auto const& pd = branchInfo.branchDescription_;
    // Note: It is not an error to attempt to delay read a produced
    // run or subrun product because there might be many of them spread
    // across multiple fragments of the same run or subrun which will
    // be combined below.
    if ((branchType_ != InSubRun) && (branchType_ != InRun)) {
      if (pd.produced()) {
        throw Exception{errors::LogicError, "RootDelayedReader::getProduct_"}
          << "Attempt to delay read a produced product!\n";
      }
    }
    // Note: threading: The configure ref core streamer and the related i/o
    // operations must be done with the source lock held!
    InputSourceMutexSentry sentry;
    configureProductIDStreamer(branchIDLists_);
    configureRefCoreStreamer(principal_.get());
    TClass* cl = TClass::GetClass(pd.wrappedName().c_str());
    auto get_product = [this, cl, br](auto entry) {
      unique_ptr<EDProduct> p{static_cast<EDProduct*>(cl->New())};
      EDProduct* pp = p.get();
      br->SetAddress(&pp);
      unsigned long long ticks = 0;
      auto const bytesRead = input::getEntry(br, entry, ticks);
      if ((saveMemoryObjectThreshold_ > -1) &&
          (bytesRead > saveMemoryObjectThreshold_)) {
        br->DropBaskets("all");
      }
      return p;
    };
    // Retrieve first product
    auto result = get_product(entrySet_[0]);
    if ((branchType_ != InSubRun) && (branchType_ != InRun)) {
      // Not a run or subrun product, all done.
      configureProductIDStreamer();
      configureRefCoreStreamer();
      return result;
    }
    //
    //  Retrieve and combine multiple Run/SubRun products as needed (this is
    //  aggregation!).
    //
    // Products from files that did not support RangeSets are
    // assigned RangeSets that correspond to the entire run/subrun.
    if (fileFormatVersion_.value_ < 9) {
      if (branchType_ == InRun) {
        rs = RangeSet::forRun(eventID_.runID());
      } else {
        rs = RangeSet::forSubRun(eventID_.subRunID());
      }
      configureProductIDStreamer();
      configureRefCoreStreamer();
      return result;
    }
    // Fetch the provenance for the first product.
    vector<ProductProvenance> ppv;
    unique_ptr<ProductProvenance const> prov;
    {
      auto p_ppv = &ppv;
      provenanceBranch_->SetAddress(&p_ppv);
      input::getEntry(provenanceBranch_, entrySet_[0]);
      for (auto const& val : ppv) {
        if (val.productID() == pid) {
          prov.reset(new ProductProvenance(val));
          break;
        }
      }
    }
    // Note: Cannot make this assert here because it can be so that the first
    // product
    //       is a dummy wrapper with the present flag false, and no provenance
    //       has been written (How???), but the second product is present and
    //       has provenance.
    // assert((prov.get() != nullptr) && "Could not find provenance for this
    // Run/SubRun product!");
    // Unfortunately, we cannot use detail::resolveRangeSetInfo in
    // this case because products that represent a full (Sub)Run are
    // allowed to be duplicated in an input file.  The behavior in
    // such a case is a NOP.
    RangeSet mergedRangeSet = detail::resolveRangeSet(db_,
                                                      "SomeInput"s,
                                                      branchType_,
                                                      result->getRangeSetID(),
                                                      compactSubRunRanges_);
    // Note: If the mergedRangeSet is invalid here that means the first product
    // was a dummy created
    //       by RootOutputFile to prevent double-counting when combining
    //       products. Or the user is playing games that are going to hurt them
    //       by intentionally created a product with an invalid range set.
    // Note: Also in that case we have a product status of unknown which we may
    // have to replace later.
    for (auto it = entrySet_.cbegin() + 1, e = entrySet_.cend(); it != e;
         ++it) {
      auto p = get_product(*it);
      vector<ProductProvenance> new_ppv;
      unique_ptr<ProductProvenance const> new_prov;
      {
        auto p_ppv = &new_ppv;
        provenanceBranch_->SetAddress(&p_ppv);
        input::getEntry(provenanceBranch_, *it);
        for (auto const& val : new_ppv) {
          if (val.productID() == pid) {
            new_prov.reset(new ProductProvenance(val));
            break;
          }
        }
      }
      // assert((new_prov.get() != nullptr) && "Could not find provenance for
      // this Run/SubRun product!");  auto const id = p->getRangeSetID();
      RangeSet const& newRS = detail::resolveRangeSet(db_,
                                                      "SomeInput"s,
                                                      branchType_,
                                                      p->getRangeSetID(),
                                                      compactSubRunRanges_);
      if (!mergedRangeSet.is_valid() && !newRS.is_valid()) {
        // Both range sets are invalid, do nothing.
        // RootOutputFile creates this situation to prevent double-counting when
        // combining products. Possibly can also happen when a produced product
        // is not produced or when a product is dropped?
      } else if (mergedRangeSet.is_valid() && !newRS.is_valid()) {
        // FIXME: Can a neverCreated or dropped product have a valid range set?
        // assert(prov->productStatus() == productstatus::present());
        // Our current merged range set is valid and the next one is invalid, do
        // nothing. RootOutputFile creates this situation to prevent
        // double-counting when combining products. Possibly can also happen
        // when a produced product is not produced or when a product is dropped?
      } else if (!mergedRangeSet.is_valid() && newRS.is_valid()) {
        // We finally have a valid range set to use.
        // FIXME: Can a neverCreated or dropped product have a valid range set?
        // assert(new_prov->productStatus() == productstatus::present());
        mergedRangeSet = newRS;
        std::swap(result, p);
        // Replace the provenance.
        // Note: We do not worry about productstatus::unknown() here because
        // newRS is valid.
        principal_->insert_pp(const_cast<Group*>(grp),
                              make_unique<ProductProvenance const>(*new_prov));
        prov = move(new_prov);
      } else if (art::disjoint_ranges(mergedRangeSet, newRS)) {
        // Old and new range sets can be combined, do it.
        // FIXME: Can a neverCreated or dropped product have a valid range set?
        // assert(prov->productStatus() == productstatus::present());
        // assert(new_prov->productStatus() == productstatus::present());
        result->combine(p.get());
        mergedRangeSet.merge(newRS);
        // FIXME: What do we do about provenance??? What if one product has
        // status present and the other neverCreated or dropped?
        // FIXME: Possibly cannot happen because in the case that the new
        // provenance is not present it would have
        // FIXME: an invalid range set?
        // Note: We do not worry about productstatus::unknown() here because
        // newRS is valid. assert((prov->productStatus() ==
        // new_prov->productStatus()) && "Unequal product status when merging
        // range sets!");
      } else if (art::same_ranges(mergedRangeSet, newRS)) {
        // The ranges are the same, so the behavior is a NOP.  If
        // the stakeholders decide that products with the same
        // ranges should be checked for equality, the condition
        // will be added here.
        // FIXME: Can a neverCreated or dropped product have a valid range set?
        // assert(prov->productStatus() == productstatus::present());
        // assert(new_prov->productStatus() == productstatus::present());
        // FIXME: What do we do about provenance??? What if one product has
        // status present and the other neverCreated or dropped?
        // FIXME: Possibly cannot happen because in the case that the new
        // provenance is not present it would have
        // FIXME: an invalid range set?
        // Note: We do not worry about productstatus::unknown() here because
        // newRS is valid. assert((prov->productStatus() ==
        // new_prov->productStatus()) && "Unequal product status when products
        // have identical range sets!");
      } else if (art::overlapping_ranges(mergedRangeSet, newRS)) {
        throw Exception{errors::ProductCannotBeAggregated,
                        "RootDelayedReader::getProduct_"}
          << "\nThe following ranges corresponding to the product:\n"
          << "   '" << branchInfo.branchDescription_ << "'"
          << "\ncannot be aggregated\n"
          << mergedRangeSet << " and\n"
          << newRS << "\nPlease contact artists@fnal.gov.\n";
      }
    }
    // Now transfer the calculated mergedRangeSet to the output argument.
    std::swap(rs, mergedRangeSet);
    // And now we are done.
    configureProductIDStreamer();
    configureRefCoreStreamer();
    return result;
  }

bool art::RootDelayedReader::isAvailableAfterCombine_ ( ProductID  bid ) const

overrideprivatevirtual

Reimplemented from art::DelayedReader.

Definition at line 85 of file RootDelayedReader.cc.

  {
    if ((branchType_ != InSubRun) && (branchType_ != InRun)) {
      // We only handle run and subrun products here, tell
      // the caller he should proceed.
      return true;
    }
    // Products from files that did not support range sets are
    // worth attempting to read, tell the caller to proceed.
    if (fileFormatVersion_.value_ < 9) {
      return true;
    }
    {
      InputSourceMutexSentry sentry;
      for (auto I = entrySet_.cbegin(), E = entrySet_.cend(); I != E; ++I) {
        vector<ProductProvenance> ppv;
        ProductProvenance const* prov = nullptr;
        {
          auto p_ppv = &ppv;
          provenanceBranch_->SetAddress(&p_ppv);
          input::getEntry(provenanceBranch_, *I);
          for (auto const& val : ppv) {
            if (val.productID() == bid) {
              prov = &val;
              break;
            }
          }
        }
        // Note: If this is a produced product then it might not be in
        // any of the fragments, this is not an error.
        if (prov != nullptr) {
          if (prov->productStatus() == productstatus::present()) {
            // We found a usable product in the set of fragments,
            // tell our caller to proceed.
            return true;
          }
        }
      }
      // None of the run or subrun fragments have a usable product,
      // tell our caller that he can give up now.
      return false;
    }
  }

int art::RootDelayedReader::openNextSecondaryFile_ ( int  idx )

overrideprivatevirtual

Reimplemented from art::DelayedReader.

Definition at line 329 of file RootDelayedReader.cc.

  {
    // idx being a number we can actually use is a precondition of this
    // function.
    assert(!(idx < 0));
    // Note:
    //
    // Return code of -2 means stop, -1 means event-not-found,
    // otherwise 0 for success.
    //
    auto const& sfnm = primaryFile_->secondaryFileNames();
    assert(!(static_cast<decltype(sfnm.size())>(idx) > sfnm.size()));
    if (sfnm.empty()) { // No configured secondary files.
      return -2;
    }
    auto const& sf = primaryFile_->secondaryFiles();
    if (static_cast<decltype(sfnm.size())>(idx) == sfnm.size()) {
      // We're done.
      return -2;
    }
    if (!sf[idx]) {
      primaryFile_->openSecondaryFile(idx);
    }
    switch (branchType_) {
      case InEvent: {
        if (!sf[idx]->readEventForSecondaryFile(eventID_)) {
          return -1;
        }
      } break;
      case InSubRun: {
        if (!sf[idx]->readSubRunForSecondaryFile(eventID_.subRunID())) {
          return -1;
        }
      } break;
      case InRun: {
        if (!sf[idx]->readRunForSecondaryFile(eventID_.runID())) {
          return -1;
        }
      } break;
      default: {
        assert(false && "RootDelayedReader encountered an unknown BranchType!");
        return -2;
      }
    }
    return 0;
  }

RootDelayedReader& art::RootDelayedReader::operator= ( RootDelayedReader const &  )

delete

RootDelayedReader& art::RootDelayedReader::operator= ( RootDelayedReader &&  )

delete

std::vector< ProductProvenance > art::RootDelayedReader::readProvenance_ ( ) const

overrideprivatevirtual

Reimplemented from art::DelayedReader.

Definition at line 67 of file RootDelayedReader.cc.

  {
    InputSourceMutexSentry sentry;
    vector<ProductProvenance> ppv;
    auto p_ppv = &ppv;
    provenanceBranch_->SetAddress(&p_ppv);
    // Note: This provenance may be replaced later for
    //       run and subrun products by the combination
    //       process (agggregation).
    input::getEntry(provenanceBranch_, entrySet_[0]);
    return ppv;
  }

void art::RootDelayedReader::setPrincipal_ ( cet::exempt_ptr< Principal >  principal )

overrideprivatevirtual

Reimplemented from art::DelayedReader.

Definition at line 61 of file RootDelayedReader.cc.

  {
    principal_ = principal;
  }

Member Data Documentation

cet::exempt_ptr<input::BranchMap const> art::RootDelayedReader::branches_

private

Definition at line 64 of file RootDelayedReader.h.

cet::exempt_ptr<BranchIDLists const> art::RootDelayedReader::branchIDLists_

private

Definition at line 69 of file RootDelayedReader.h.

BranchType art::RootDelayedReader::branchType_

private

Definition at line 70 of file RootDelayedReader.h.

bool const art::RootDelayedReader::compactSubRunRanges_

private

Definition at line 72 of file RootDelayedReader.h.

sqlite3* art::RootDelayedReader::db_

private

Definition at line 62 of file RootDelayedReader.h.

std::vector<input::EntryNumber> const art::RootDelayedReader::entrySet_

private

Definition at line 63 of file RootDelayedReader.h.

EventID art::RootDelayedReader::eventID_

private

Definition at line 71 of file RootDelayedReader.h.

FileFormatVersion art::RootDelayedReader::fileFormatVersion_

private

Definition at line 61 of file RootDelayedReader.h.

cet::exempt_ptr<RootInputFile> art::RootDelayedReader::primaryFile_

private

Definition at line 68 of file RootDelayedReader.h.

cet::exempt_ptr<Principal> art::RootDelayedReader::principal_

private

Definition at line 67 of file RootDelayedReader.h.

TBranch* art::RootDelayedReader::provenanceBranch_

private

Definition at line 65 of file RootDelayedReader.h.

int64_t art::RootDelayedReader::saveMemoryObjectThreshold_

private

Definition at line 66 of file RootDelayedReader.h.

---

The documentation for this class was generated from the following files:

• art/art/Framework/IO/Root/RootDelayedReader.h
• art/art/Framework/IO/Root/RootDelayedReader.cc