DPL: separate iteration over the slots vs over parts in the slots

Author: ktf

Detectors/TPC/workflow/readers/include/TPCReaderWorkflow/TPCSectorCompletionPolicy.h

@@ -101,9 +101,9 @@ class TPCSectorCompletionPolicy
       size_t nMaxPartsPerRoute = 0;
       int inputType = -1;
       for (auto it = inputs.begin(), end = inputs.end(); it != end; ++it) {
-        nMaxPartsPerRoute = it.size() > nMaxPartsPerRoute ? it.size() : nMaxPartsPerRoute;
+        nMaxPartsPerRoute = it.parts().size() > nMaxPartsPerRoute ? it.parts().size() : nMaxPartsPerRoute;
         bool haveActivePart = false;
-        for (auto const& ref : it) {
+        for (auto const& ref : it.parts()) {
           if (!framework::DataRefUtils::isValid(ref)) {
             continue;
           }

Framework/Core/include/Framework/InputRecord.h

@@ -693,55 +693,20 @@ class InputRecord
     using reference = typename BaseType::reference;
     using pointer = typename BaseType::pointer;
     using ElementType = typename std::remove_const<value_type>::type;
-    using iterator = InputSpan::Iterator<SelfType, T>;
-    using const_iterator = InputSpan::Iterator<SelfType, const T>;
 
     InputRecordIterator(InputRecord const* parent, bool isEnd = false)
       : BaseType(parent, isEnd)
     {
     }
 
-    /// Initial indices for part-level iteration: first part starts at {headerIdx=0, payloadIdx=1}.
-    [[nodiscard]] DataRefIndices initialIndices() const { return {0, 1}; }
-    /// Sentinel used by nextIndicesGetter to signal end-of-slot.
-    [[nodiscard]] DataRefIndices endIndices() const { return {size_t(-1), size_t(-1)}; }
-
-    /// Get element at the given raw message indices in O(1).
-    [[nodiscard]] ElementType getAtIndices(DataRefIndices indices) const
-    {
-      return this->parent()->getAtIndices(this->position(), indices);
-    }
-
-    /// Advance @a current to the next part's indices in O(1).
-    [[nodiscard]] DataRefIndices nextIndices(DataRefIndices current) const
-    {
-      return this->parent()->nextIndices(this->position(), current);
-    }
-
-    /// Check if slot is valid, index of part is not used
+    /// Check if slot is valid
     [[nodiscard]] bool isValid(size_t = 0) const
     {
       if (this->position() < this->parent()->size()) {
         return this->parent()->isValid(this->position());
       }
       return false;
     }
-
-    /// Get number of parts in input slot
-    [[nodiscard]] size_t size() const
-    {
-      return this->parent()->getNofParts(this->position());
-    }
-
-    [[nodiscard]] const_iterator begin() const
-    {
-      return const_iterator(this, size() == 0);
-    }
-
-    [[nodiscard]] const_iterator end() const
-    {
-      return const_iterator(this, true);
-    }
   };
 
   using iterator = InputRecordIterator<DataRef>;

Framework/Core/include/Framework/InputRecordWalker.h

@@ -73,16 +73,19 @@ class InputRecordWalker
     // the iterator over the input routes
     using input_iterator = decltype(std::declval<InputRecord>().begin());
 
+    // the range over parts in one slot — must be stored as a member so the
+    // part_iterator (which holds a pointer into it) does not dangle
+    using part_range = InputRecord::PartRange;
+    using part_iterator = InputSpan::Iterator<part_range, const DataRef>;
+
     Iterator() = delete;
 
-    Iterator(InputRecord& parent, input_iterator it, input_iterator end, std::vector<InputSpec> const& filterSpecs)
-      : mParent(parent), mInputIterator(it), mEnd(end), mCurrent(mInputIterator.begin()), mFilterSpecs(filterSpecs)
+    Iterator(input_iterator it, input_iterator end, std::vector<InputSpec> const& filterSpecs)
+      : mInputIterator(it), mEnd(end), mCurrentRange(it.parts()), mCurrent(mCurrentRange.begin()), mFilterSpecs(filterSpecs)
     {
       next(true);
     }
 
-    ~Iterator() = default;
-
     // prefix increment
     self_type& operator++()
     {
@@ -104,9 +107,7 @@ class InputRecordWalker
     // comparison
     bool operator==(const self_type& other) const
     {
-      bool result = mInputIterator == other.mInputIterator;
-      result = result && mCurrent == other.mCurrent;
-      return result;
+      return mInputIterator == other.mInputIterator && mCurrent == other.mCurrent;
     }
 
     bool operator!=(const self_type& rh) const
@@ -115,19 +116,13 @@ class InputRecordWalker
     }
 
    private:
-    // the iterator over the parts in one channel
-    using part_iterator = typename input_iterator::const_iterator;
-
     bool next(bool isInitialPart = false)
     {
+      if (!isInitialPart) {
+        ++mCurrent;
+      }
       while (mInputIterator != mEnd) {
-        while (mCurrent != mInputIterator.end()) {
-          // increment on the level of one input
-          if (!isInitialPart && (mCurrent == mInputIterator.end() || ++mCurrent == mInputIterator.end())) {
-            // no more parts, go to next input
-            break;
-          }
-          isInitialPart = false;
+        for (; mCurrent != mCurrentRange.end(); ++mCurrent) {
           // check filter rules
           if (mFilterSpecs.size() > 0) {
             bool isSelected = false;
@@ -143,15 +138,15 @@ class InputRecordWalker
           return true;
         }
         ++mInputIterator;
-        mCurrent = mInputIterator.begin();
-        isInitialPart = true;
+        mCurrentRange = mInputIterator.parts();
+        mCurrent = mCurrentRange.begin();
       } // end loop over record
       return false;
     }
 
-    InputRecord& mParent;
     input_iterator mInputIterator;
     input_iterator mEnd;
+    part_range mCurrentRange; // declared before mCurrent — initialized first
     part_iterator mCurrent;
     std::vector<InputSpec> const& mFilterSpecs;
   };
@@ -160,12 +155,12 @@ class InputRecordWalker
 
   const_iterator begin() const
   {
-    return const_iterator(mRecord, mRecord.begin(), mRecord.end(), mFilterSpecs);
+    return const_iterator(mRecord.begin(), mRecord.end(), mFilterSpecs);
   }
 
   const_iterator end() const
   {
-    return const_iterator(mRecord, mRecord.end(), mRecord.end(), mFilterSpecs);
+    return const_iterator(mRecord.end(), mRecord.end(), mFilterSpecs);
   }
 
  private:

Framework/Core/include/Framework/InputSpan.h

@@ -179,69 +179,40 @@ class InputSpan
       return mCurrentIndices.headerIdx;
     }
 
+    // return an iterable range over all parts in the current slot
+    // only available for slot-level iterators whose parent has parts(size_t)
+    [[nodiscard]] auto parts() const
+      requires requires(ParentType const* p, size_t i) { p->parts(i); }
+    {
+      return mParent->parts(mCurrentIndices.headerIdx);
+    }
+
    private:
     ParentType const* mParent;
     DataRefIndices mCurrentIndices;
     ElementType mElement;
   };
 
-  /// @class InputSpanIterator
-  /// An iterator over the input slots.
-  /// It supports an iterator interface to access the parts in the slot.
-  template <typename T>
-  class InputSpanIterator : public Iterator<InputSpan, T>
-  {
-   public:
-    using SelfType = InputSpanIterator;
-    using BaseType = Iterator<InputSpan, T>;
-    using value_type = typename BaseType::value_type;
-    using reference = typename BaseType::reference;
-    using pointer = typename BaseType::pointer;
-    using ElementType = typename std::remove_const<value_type>::type;
-    using iterator = Iterator<SelfType, T>;
-    using const_iterator = Iterator<SelfType, const T>;
-
-    InputSpanIterator(InputSpan const* parent, bool isEnd = false)
-      : BaseType(parent, isEnd)
-    {
-    }
+  /// A range over the parts of a single slot, supporting range-based for.
+  struct PartRange {
+    InputSpan const* span;
+    size_t slot;
 
-    /// Initial indices for part-level iteration: first part starts at {headerIdx=0, payloadIdx=1}.
     [[nodiscard]] DataRefIndices initialIndices() const { return {0, 1}; }
-    /// Sentinel used by nextIndicesGetter to signal end-of-slot.
     [[nodiscard]] DataRefIndices endIndices() const { return {size_t(-1), size_t(-1)}; }
+    [[nodiscard]] DataRef getAtIndices(DataRefIndices idx) const { return span->getAtIndices(slot, idx); }
+    [[nodiscard]] DataRefIndices nextIndices(DataRefIndices idx) const { return span->nextIndices(slot, idx); }
+    [[nodiscard]] size_t size() const { return span->getNofParts(slot); }
 
-    /// Get element at the given raw message indices in O(1).
-    [[nodiscard]] ElementType getAtIndices(DataRefIndices indices) const
-    {
-      return this->parent()->getAtIndices(this->position(), indices);
-    }
-
-    /// Advance @a current to the next part's indices in O(1).
-    [[nodiscard]] DataRefIndices nextIndices(DataRefIndices current) const
-    {
-      return this->parent()->nextIndices(this->position(), current);
-    }
-
-    /// Get number of parts in input slot
-    [[nodiscard]] size_t size() const
-    {
-      return this->parent()->getNofParts(this->position());
-    }
-
-    [[nodiscard]] const_iterator begin() const
-    {
-      return const_iterator(this, size() == 0);
-    }
-
-    [[nodiscard]] const_iterator end() const
-    {
-      return const_iterator(this, true);
-    }
+    [[nodiscard]] Iterator<PartRange, const DataRef> begin() const { return {this, size() == 0}; }
+    [[nodiscard]] Iterator<PartRange, const DataRef> end() const { return {this, true}; }
   };
 
-  using iterator = InputSpanIterator<DataRef>;
-  using const_iterator = InputSpanIterator<const DataRef>;
+  /// Return an iterable range over all parts in slot @a i.
+  [[nodiscard]] PartRange parts(size_t i) const { return {this, i}; }
+
+  using const_iterator = Iterator<InputSpan, const DataRef>;
+  using iterator = const_iterator;
 
   // supporting read-only access and returning const_iterator
   [[nodiscard]] const_iterator begin() const

Framework/Core/src/CompletionPolicyHelpers.cxx

@@ -343,7 +343,7 @@ CompletionPolicy CompletionPolicyHelpers::consumeWhenAnyWithAllConditions(const
     }
     if (canConsume || conditionMissing) {
       for (auto it = inputs.begin(), end = inputs.end(); it != end; ++it) {
-        for (auto const& ref : it) {
+        for (auto const& ref : it.parts()) {
           if (!framework::DataRefUtils::isValid(ref)) {
             continue;
           }

Framework/Core/test/test_InputSpan.cxx

@@ -61,8 +61,8 @@ TEST_CASE("TestInputSpan")
   routeNo = 0;
   for (auto it = span.begin(), end = span.end(); it != end; ++it) {
     size_t partNo = 0;
-    REQUIRE(it.size() * 2 == inputs[routeNo].size());
-    for (auto const& ref : it) {
+    REQUIRE(it.parts().size() * 2 == inputs[routeNo].size());
+    for (auto const& ref : it.parts()) {
       REQUIRE(inputs[routeNo].at(partNo++) == ref.header);
       REQUIRE(inputs[routeNo].at(partNo++) == ref.payload);
       INFO(ref.header << " " << ref.payload);

Framework/Utils/include/DPLUtils/DPLRawParser.h

@@ -105,14 +105,12 @@ class DPLRawParser
     Iterator() = delete;
 
     Iterator(InputRecord& parent, input_iterator it, input_iterator end, std::vector<InputSpec> const& filterSpecs, fair::Severity sev = fair::Severity::alarm, size_t maxErrMsg = -1, size_t* cntErrMsg = nullptr)
-      : mParent(parent), mInputIterator(it), mEnd(end), mPartIterator(mInputIterator.begin()), mParser(std::make_unique<parser_type>(reinterpret_cast<const char*>(&initializer), sizeof(initializer))), mCurrent(mParser->begin()), mFilterSpecs(filterSpecs), mMaxFailureMessages(maxErrMsg), mExtFailureCounter(cntErrMsg), mSeverity(sev)
+      : mParent(parent), mInputIterator(it), mEnd(end), mCurrentRange(it.parts()), mPartIterator(mCurrentRange.begin()), mParser(std::make_unique<parser_type>(reinterpret_cast<const char*>(&initializer), sizeof(initializer))), mCurrent(mParser->begin()), mFilterSpecs(filterSpecs), mMaxFailureMessages(maxErrMsg), mExtFailureCounter(cntErrMsg), mSeverity(sev)
     {
       mParser.reset();
       next();
     }
 
-    ~Iterator() = default;
-
     // prefix increment
     self_type& operator++()
     {
@@ -205,7 +203,7 @@ class DPLRawParser
 
     friend std::ostream& operator<<(std::ostream& os, self_type const& it)
     {
-      if (it.mInputIterator != it.mEnd && it.mPartIterator != it.mInputIterator.end() && it.mParser != nullptr) {
+      if (it.mInputIterator != it.mEnd && it.mPartIterator != it.mCurrentRange.end() && it.mParser != nullptr) {
         os << it.mCurrent;
       }
       return os;
@@ -223,7 +221,7 @@ class DPLRawParser
     friend std::ostream& operator<<(std::ostream& os, Fmt<FmtCtrl> const& fmt)
     {
       auto const& it = fmt.it;
-      if (it.mInputIterator != it.mEnd && it.mPartIterator != it.mInputIterator.end() && it.mParser != nullptr) {
+      if (it.mInputIterator != it.mEnd && it.mPartIterator != it.mCurrentRange.end() && it.mParser != nullptr) {
         if constexpr (FmtCtrl == raw_parser::FormatSpec::Info) {
           // TODO: need to propagate the format spec also on the RawParser object
           // for now this operation prints the RDH version info and the table header
@@ -236,8 +234,10 @@ class DPLRawParser
     }
 
    private:
-    // the iterator over the parts in one channel
-    using part_iterator = typename input_iterator::const_iterator;
+    // the range over parts in one slot — must be stored as a member so the
+    // part_iterator (which holds a pointer into it) does not dangle
+    using part_range = InputRecord::PartRange;
+    using part_iterator = InputSpan::Iterator<part_range, const DataRef>;
     // the iterator over the over the parser pages
     using parser_iterator = typename parser_type::const_iterator;
 
@@ -265,15 +265,15 @@ class DPLRawParser
 
       while (mInputIterator != mEnd) {
         bool isInitial = mParser == nullptr;
-        while (mPartIterator != mInputIterator.end()) {
+        while (mPartIterator != mCurrentRange.end()) {
           // first increment on the parser level
           if (mParser && mCurrent != mParser->end() && ++mCurrent != mParser->end()) {
             // we have an active parser and there is still data at the incremented iterator
             return true;
           }
           // now increment on the level of one input
           mParser.reset();
-          if (!isInitial && (mPartIterator == mInputIterator.end() || ++mPartIterator == mInputIterator.end())) {
+          if (!isInitial && ++mPartIterator == mCurrentRange.end()) {
             // no more parts, go to next input
             break;
           }
@@ -312,14 +312,16 @@ class DPLRawParser
           }
         } // end loop over parts on one input
         ++mInputIterator;
-        mPartIterator = mInputIterator.begin();
+        mCurrentRange = mInputIterator.parts();
+        mPartIterator = mCurrentRange.begin();
       } // end loop over inputs
       return false;
     }
 
     InputRecord& mParent;
     input_iterator mInputIterator;
     input_iterator mEnd;
+    part_range mCurrentRange; // declared before mPartIterator — initialized first
     part_iterator mPartIterator;
     std::unique_ptr<parser_type> mParser;
     parser_iterator mCurrent;
@@ -355,4 +357,4 @@ class DPLRawParser
 
 } // namespace o2::framework
 
-#endif //FRAMEWORK_UTILS_DPLRAWPARSER_H
+#endif // FRAMEWORK_UTILS_DPLRAWPARSER_H