Merge 5ea5aae into sapling-pr-archive-ktf

Author: ktf

Detectors/Upgrades/ALICE3/TRK/reconstruction/CMakeLists.txt

@@ -18,7 +18,6 @@ o2_add_library(TRKReconstruction
                SOURCES src/TimeFrame.cxx
                        src/Clusterer.cxx
                        $<$<BOOL:${Acts_FOUND}>:src/ClustererACTS.cxx>
-                       $<$<BOOL:${Acts_FOUND}>:src/TrackerACTS.cxx>
                PUBLIC_LINK_LIBRARIES
                        O2::ITStracking
                        O2::GPUCommon
@@ -46,8 +45,7 @@ set(dictHeaders include/TRKReconstruction/TimeFrame.h
                 include/TRKReconstruction/Clusterer.h)
 
 if(Acts_FOUND)
-  list(APPEND dictHeaders include/TRKReconstruction/ClustererACTS.h
-                          include/TRKReconstruction/TrackerACTS.h)
+  list(APPEND dictHeaders include/TRKReconstruction/ClustererACTS.h)
 endif()
 
 o2_target_root_dictionary(TRKReconstruction

Framework/AODMerger/src/aodMerger.cxx

@@ -38,6 +38,7 @@ int main(int argc, char* argv[])
   long maxDirSize = 100000000;
   bool skipNonExistingFiles = false;
   bool skipParentFilesList = false;
+  bool mergeByName = false;
   int verbosity = 2;
   int exitCode = 0; // 0: success, >0: failure
   int compression = 505;
@@ -50,6 +51,7 @@ int main(int argc, char* argv[])
     {"skip-non-existing-files", no_argument, nullptr, 3},
     {"skip-parent-files-list", no_argument, nullptr, 4},
     {"compression", required_argument, nullptr, 5},
+    {"merge-by-name", no_argument, nullptr, 6},
     {"verbosity", required_argument, nullptr, 'v'},
     {"help", no_argument, nullptr, 'h'},
     {nullptr, 0, nullptr, 0}};
@@ -70,6 +72,8 @@ int main(int argc, char* argv[])
       skipParentFilesList = true;
     } else if (c == 5) {
       compression = atoi(optarg);
+    } else if (c == 6) {
+      mergeByName = true;
     } else if (c == 'v') {
       verbosity = atoi(optarg);
     } else if (c == 'h') {
@@ -80,6 +84,7 @@ int main(int argc, char* argv[])
       printf("  --skip-non-existing-files    Flag to allow skipping of non-existing files in the input list.\n");
       printf("  --skip-parent-files-list     Flag to allow skipping the merging of the parent files list.\n");
       printf("  --compression <root compression id>  Compression algorithm / level to use (default: %d)\n", compression);
+      printf("  --merge-by-name              Only merge TTrees from folders with the same name.\n");
       printf("  --verbosity <flag>           Verbosity of output (default: %d).\n", verbosity);
       return -1;
     } else {
@@ -94,6 +99,9 @@ int main(int argc, char* argv[])
   if (skipNonExistingFiles) {
     printf("  WARNING: Skipping non-existing files.\n");
   }
+  if (mergeByName) {
+    printf("  Merging only folders with the same name.\n");
+  }
 
   std::map<std::string, TTree*> trees;
   std::map<std::string, uint64_t> sizeCompressed;
@@ -112,6 +120,28 @@ int main(int argc, char* argv[])
   TMap* parentFiles = nullptr;
   int totalMergedDFs = 0;
   int mergedDFs = 0;
+
+  // Write all accumulated trees to outputDir, update stats, and clean up state.
+  auto flushTrees = [&](bool resetState) {
+    if (!outputDir) {
+      return;
+    }
+    for (auto const& tree : trees) {
+      outputDir->cd();
+      tree.second->Write();
+      sizeCompressed[tree.first] += tree.second->GetZipBytes();
+      sizeUncompressed[tree.first] += tree.second->GetTotBytes();
+      delete tree.second;
+    }
+    if (resetState) {
+      outputDir = nullptr;
+      trees.clear();
+      offsets.clear();
+      mergedDFs = 0;
+      currentDirSize = 0;
+    }
+  };
+
   while (in.good() && exitCode == 0) {
     in >> line;
 
@@ -182,6 +212,14 @@ int main(int argc, char* argv[])
 
       auto dfName = ((TObjString*)key1)->GetString().Data();
 
+      // If merge-by-name is active, flush accumulated trees when the folder name changes
+      if (mergeByName && outputDir && std::string(outputDir->GetName()) != std::string(dfName)) {
+        if (verbosity > 0) {
+          printf("Folder name changed: closing folder %s.\n", outputDir->GetName());
+        }
+        flushTrees(true);
+      }
+
       if (verbosity > 0) {
         printf("  Processing folder %s\n", dfName);
       }
@@ -396,21 +434,7 @@ int main(int argc, char* argv[])
         if (verbosity > 0) {
           printf("Maximum size reached: %ld. Closing folder %s.\n", currentDirSize, dfName);
         }
-        for (auto const& tree : trees) {
-          // printf("Writing %s\n", tree.first.c_str());
-          outputDir->cd();
-          tree.second->Write();
-
-          // stats
-          sizeCompressed[tree.first] += tree.second->GetZipBytes();
-          sizeUncompressed[tree.first] += tree.second->GetTotBytes();
-
-          delete tree.second;
-        }
-        outputDir = nullptr;
-        trees.clear();
-        offsets.clear();
-        mergedDFs = 0;
+        flushTrees(true);
       }
     }
     inputFile->Close();
@@ -421,16 +445,7 @@ int main(int argc, char* argv[])
     parentFiles->Write("parentFiles", TObject::kSingleKey);
   }
 
-  for (auto const& tree : trees) {
-    outputDir->cd();
-    tree.second->Write();
-
-    // stats
-    sizeCompressed[tree.first] += tree.second->GetZipBytes();
-    sizeUncompressed[tree.first] += tree.second->GetTotBytes();
-
-    delete tree.second;
-  }
+  flushTrees(false);
 
   outputFile->Write();
   outputFile->Close();
@@ -462,4 +477,4 @@ int main(int argc, char* argv[])
   printf("\n");
 
   return exitCode;
-}
+}

Framework/Core/include/Framework/DataModelViews.h

@@ -16,6 +16,7 @@
 #include "DomainInfoHeader.h"
 #include "SourceInfoHeader.h"
 #include "Headers/DataHeader.h"
+#include "Framework/DataRef.h"
 #include "Framework/TimesliceSlot.h"
 #include <ranges>
 #include <span>
@@ -80,10 +81,7 @@ struct count_parts {
   }
 };
 
-struct DataRefIndices {
-  size_t headerIdx;
-  size_t payloadIdx;
-};
+// DataRefIndices is defined in Framework/DataRef.h
 
 struct get_pair {
   size_t pairId;
@@ -127,6 +125,43 @@ struct get_pair {
   }
 };
 
+// Advance from a DataRefIndices to the next one in O(1), reading only the
+// current header.  Intended for use in iterators so that ++ is O(1) rather
+// than the O(n) while-loop that get_pair requires.
+//
+// New-style block  (splitPayloadIndex == splitPayloadParts > 1):
+//   layout: [header, payload_0, payload_1, ..., payload_{N-1}]
+//   advance within block while payloads remain, then jump to the next block.
+//
+// Old-style block  (splitPayloadIndex != splitPayloadParts, splitPayloadParts > 1)
+// or single pair   (splitPayloadParts == 0):
+//   layout: [header, payload]  – always advance by two messages.
+struct get_next_pair {
+  DataRefIndices current;
+  template <typename R>
+    requires std::ranges::random_access_range<R> && std::ranges::sized_range<R>
+  friend DataRefIndices operator|(R&& r, get_next_pair self)
+  {
+    size_t hIdx = self.current.headerIdx;
+    auto* header = o2::header::get<o2::header::DataHeader*>(r[hIdx]->GetData());
+    if (!header) {
+      throw std::runtime_error("Not a DataHeader");
+    }
+    if (header->splitPayloadParts > 1 && header->splitPayloadIndex == header->splitPayloadParts) {
+      // New-style block: one header followed by splitPayloadParts contiguous payloads.
+      if (self.current.payloadIdx < hIdx + header->splitPayloadParts) {
+        // More sub-payloads remain in this block.
+        return {hIdx, self.current.payloadIdx + 1};
+      }
+      // Last sub-payload consumed; move to the first pair of the next block.
+      size_t nextHIdx = hIdx + header->splitPayloadParts + 1;
+      return {nextHIdx, nextHIdx + 1};
+    }
+    // Old-style [header, payload] pairs or a single pair: advance by two messages.
+    return {hIdx + 2, hIdx + 3};
+  }
+};
+
 struct get_dataref_indices {
   size_t part;
   size_t subPart;

Framework/Core/include/Framework/DataRef.h

@@ -12,6 +12,7 @@
 #define FRAMEWORK_DATAREF_H
 
 #include <cstddef> // for size_t
+#include <compare>
 
 namespace o2
 {
@@ -29,6 +30,15 @@ struct DataRef {
   size_t payloadSize = 0;
 };
 
+/// Raw indices into the message vector for one (header, payload) pair.
+/// Kept in a lightweight header so InputSpan can use it without pulling in FairMQ.
+struct DataRefIndices {
+  size_t headerIdx;
+  size_t payloadIdx;
+  bool operator==(const DataRefIndices&) const = default;
+  auto operator<=>(const DataRefIndices&) const = default;
+};
+
 } // namespace framework
 } // namespace o2
 

Framework/Core/include/Framework/InputRecord.h

@@ -13,6 +13,7 @@
 
 #include "Framework/DataRef.h"
 #include "Framework/DataRefUtils.h"
+#include "Framework/InputSpan.h"
 #include "Framework/InputRoute.h"
 #include "Framework/TypeTraits.h"
 #include "Framework/TableConsumer.h"
@@ -202,6 +203,15 @@ class InputRecord
 
   [[nodiscard]] size_t getNofParts(int pos) const;
 
+  /// O(1) access to the part described by @a indices in slot @a pos.
+  [[nodiscard]] DataRef getAtIndices(int pos, DataRefIndices indices) const;
+
+  /// O(1) advance from @a current to the next part's indices in slot @a pos.
+  [[nodiscard]] DataRefIndices nextIndices(int pos, DataRefIndices current) const
+  {
+    return mSpan.nextIndices(pos, current);
+  }
+
   // Given a binding by string, return the associated DataRef
   DataRef getDataRefByString(const char* bindingName, int part = 0) const
   {
@@ -568,8 +578,8 @@ class InputRecord
 
     Iterator() = delete;
 
-    Iterator(ParentType const* parent, size_t position = 0, size_t size = 0)
-      : mPosition(position), mSize(size > position ? size : position), mParent(parent), mElement{nullptr, nullptr, nullptr}
+    Iterator(ParentType const* parent, bool isEnd = false)
+      : mPosition(isEnd ? parent->size() : 0), mSize(parent->size()), mParent(parent), mElement{nullptr, nullptr, nullptr}
     {
       if (mPosition < mSize) {
         if (mParent->isValid(mPosition)) {
@@ -678,18 +688,29 @@ class InputRecord
     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>;
+    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)}; }
 
-    InputRecordIterator(InputRecord const* parent, size_t position = 0, size_t size = 0)
-      : BaseType(parent, position, size)
+    /// Get element at the given raw message indices in O(1).
+    [[nodiscard]] ElementType getAtIndices(DataRefIndices indices) const
     {
+      return this->parent()->getAtIndices(this->position(), indices);
     }
 
-    /// Get element at {slotindex, partindex}
-    [[nodiscard]] ElementType getByPos(size_t pos) const
+    /// Advance @a current to the next part's indices in O(1).
+    [[nodiscard]] DataRefIndices nextIndices(DataRefIndices current) const
     {
-      return this->parent()->getByPos(this->position(), pos);
+      return this->parent()->nextIndices(this->position(), current);
     }
 
     /// Check if slot is valid, index of part is not used
@@ -709,12 +730,12 @@ class InputRecord
 
     [[nodiscard]] const_iterator begin() const
     {
-      return const_iterator(this, 0, size());
+      return const_iterator(this, size() == 0);
     }
 
     [[nodiscard]] const_iterator end() const
     {
-      return const_iterator(this, size());
+      return const_iterator(this, true);
     }
   };
 
@@ -723,12 +744,12 @@ class InputRecord
 
   [[nodiscard]] const_iterator begin() const
   {
-    return {this, 0, size()};
+    return {this, false};
   }
 
   [[nodiscard]] const_iterator end() const
   {
-    return {this, size()};
+    return {this, true};
   }
 
   InputSpan& span()