[ntuple] add tutorial to show efficient multi-stream reading

Author: jblomer

tutorials/io/ntuple/ntpl017_shared_reader.C

@@ -0,0 +1,256 @@
+/// \file
+/// \ingroup tutorial_ntuple
+/// \notebook
+/// Example of efficient multi-threaded reading when multiple threads share a single reader.
+/// In this example, two threads share the work as follows: the first thread processes all the even entries numbers
+/// and the second thread all the odd entry numbers. The second thread works twice as slow as the first one.
+///
+/// As a result, the threads need the same clusters and pages but at different points in time.
+/// With the naive way of using the reader, this read pattern will result in cache thrashing.
+///
+/// Using the "active entry token" API, on the other hand, the reader will be informed about which entries
+/// need to be kept in the caches and cache thrashing is prevented.
+///
+/// \macro_image
+/// \macro_code
+///
+/// \date October 2025
+/// \author The ROOT Team
+
+#include <ROOT/RNTupleModel.hxx>
+#include <ROOT/RNTupleReader.hxx>
+#include <ROOT/RNTupleWriter.hxx>
+
+#include <TAxis.h>
+#include <TCanvas.h>
+#include <TGraph.h>
+#include <TLatex.h>
+#include <TLegend.h>
+#include <TRandom3.h>
+#include <TROOT.h>
+#include <TStyle.h>
+
+#include <array>
+#include <atomic>
+#include <chrono>
+#include <exception>
+#include <functional>
+#include <iostream>
+#include <mutex>
+#include <thread>
+#include <vector>
+
+using namespace std::chrono_literals;
+
+// Where to store the ntuple of this example
+constexpr char const *kNTupleFileName = "ntpl017_shared_reader.root";
+
+/// The sample class that is stored in the RNTuple
+struct Point2D {
+   float fX;
+   float fY;
+};
+
+/// Whether we read with setting active entry tokens (informed) or not (naive)
+enum class EReadMode {
+   kNaive,
+   kInformed
+};
+
+/// Nicify output of EReadMode values
+std::ostream &operator<<(std::ostream &os, const EReadMode &e)
+{
+   switch (e) {
+   case EReadMode::kNaive: os << "naive"; break;
+   case EReadMode::kInformed: os << "informed"; break;
+   default: os << "???";
+   }
+   return os;
+}
+
+// To be reset between ProcessEntries calls to Read()
+static std::atomic<int> gNEntriesDone;
+static std::atomic<int> gThreadId;
+
+/// The read thread's main function
+void ProcessEntries(ROOT::RNTupleReader &reader, EReadMode readMode, const std::chrono::microseconds &usPerEvent,
+                    std::vector<int> &sumLoadedClusters, std::vector<int> &sumUnsealedPages,
+                    std::vector<int> &nClusters, std::vector<int> &nPages)
+{
+   static std::mutex gLock;
+
+   const int threadId = gThreadId++;
+
+   auto token = reader.CreateActiveEntryToken();
+   for (unsigned int i = threadId; i < reader.GetNEntries(); i += 2) {
+      {
+         std::lock_guard<std::mutex> guard(gLock);
+
+         // The only difference between naive and informed reading: in informed reading, we indicate which
+         // entry we are going to use before loading it.
+         switch (readMode) {
+         case EReadMode::kInformed: token.SetEntryNumber(i); break;
+         case EReadMode::kNaive: break;
+         default: std::terminate(); // never here
+         }
+
+         reader.LoadEntry(i);
+      }
+
+      std::this_thread::sleep_for(usPerEvent);
+
+      const int entriesDone = gNEntriesDone++;
+      sumLoadedClusters.at(entriesDone) =
+         reader.GetMetrics().GetCounter("RNTupleReader.RPageSourceFile.nClusterLoaded")->GetValueAsInt();
+      sumUnsealedPages[entriesDone] =
+         reader.GetMetrics().GetCounter("RNTupleReader.RPageSourceFile.nPageUnsealed")->GetValueAsInt();
+      nClusters[entriesDone] =
+         reader.GetMetrics().GetCounter("RNTupleReader.RPageSourceFile.RClusterPool.nCluster")->GetValueAsInt();
+      nPages[entriesDone] =
+         reader.GetMetrics().GetCounter("RNTupleReader.RPageSourceFile.RPagePool.nPage")->GetValueAsInt();
+   }
+}
+
+void Read(EReadMode readMode, std::vector<int> &sumLoadedClusters, std::vector<int> &sumUnsealedPages,
+          std::vector<int> &nClusters, std::vector<int> &nPages)
+{
+   auto reader = ROOT::RNTupleReader::Open("ntpl", kNTupleFileName);
+   reader->EnableMetrics();
+
+   gNEntriesDone = 0;
+   gThreadId = 0;
+
+   const auto N = reader->GetNEntries();
+   sumLoadedClusters.resize(N);
+   sumUnsealedPages.resize(N);
+   nClusters.resize(N);
+   nPages.resize(N);
+
+   std::array<std::thread, 2> threads;
+   threads[0] = std::thread(ProcessEntries, std::ref(*reader), readMode, 100us, std::ref(sumLoadedClusters),
+                            std::ref(sumUnsealedPages), std::ref(nClusters), std::ref(nPages));
+   threads[1] = std::thread(ProcessEntries, std::ref(*reader), readMode, 200us, std::ref(sumLoadedClusters),
+                            std::ref(sumUnsealedPages), std::ref(nClusters), std::ref(nPages));
+   for (auto &t : threads) {
+      t.join();
+   }
+
+   std::cout << "Reading in mode '" << readMode << "':" << std::endl;
+   std::cout << "===========================" << std::endl;
+   reader->GetMetrics().Print(std::cout);
+   std::cout << "===========================" << std::endl << std::endl;
+}
+
+void Write()
+{
+   auto model = ROOT::RNTupleModel::Create();
+   auto ptrPoint = model->MakeField<Point2D>("point");
+
+   auto writer = ROOT::RNTupleWriter::Recreate(std::move(model), "ntpl", kNTupleFileName);
+
+   for (int i = 0; i < 1000; ++i) {
+      if (i % 100 == 0)
+         writer->CommitCluster();
+
+      auto prng = std::make_unique<TRandom3>();
+      prng->SetSeed();
+
+      ptrPoint->fX = prng->Rndm(1);
+      ptrPoint->fY = prng->Rndm(1);
+      writer->Fill();
+   }
+}
+
+TGraph *GetGraph(const std::vector<int> &counts)
+{
+   auto graph = new TGraph();
+   for (unsigned int i = 0; i < counts.size(); ++i) {
+      graph->SetPoint(i, i, counts[i]);
+   }
+   graph->GetXaxis()->SetTitle("Number of processed entries");
+   return graph;
+}
+
+void ntpl017_shared_reader()
+{
+   ROOT::EnableImplicitMT();
+
+   Write();
+   ROOT::RNTupleReader::Open("ntpl", kNTupleFileName)->PrintInfo(ROOT::ENTupleInfo::kStorageDetails);
+
+   std::vector<int> sumLoadedClusters;
+   std::vector<int> sumUnsealedPages;
+   std::vector<int> nClusters;
+   std::vector<int> nPages;
+   TLatex latex;
+
+   gStyle->SetOptStat(0);
+   gStyle->SetLineWidth(2);
+   gStyle->SetMarkerStyle(8);
+   TCanvas *canvas = new TCanvas("", "Shared Reader Example", 200, 10, 1500, 1000);
+
+   canvas->Divide(2, 2);
+
+   Read(EReadMode::kNaive, sumLoadedClusters, sumUnsealedPages, nClusters, nPages);
+
+   canvas->cd(1);
+   auto graph1 = GetGraph(sumUnsealedPages);
+   graph1->SetLineColor(kRed);
+   graph1->Draw("AL");
+   auto graph2 = GetGraph(sumLoadedClusters);
+   graph2->SetLineColor(kBlue);
+   graph2->Draw("SAME L");
+
+   auto legend = new TLegend(0.125, 0.725, 0.625, 0.875);
+   legend->AddEntry(graph1, "Number of decompressed pages", "l");
+   legend->AddEntry(graph2, "Number of loaded clusters", "l");
+   legend->Draw();
+
+   latex.SetTextAlign(22);
+   latex.DrawLatexNDC(0.5, 0.95, "Naive Reading");
+
+   canvas->cd(3);
+
+   auto graph3 = GetGraph(nPages);
+   graph3->SetMarkerColor(kRed);
+   graph3->GetYaxis()->SetNdivisions(8);
+   graph3->GetYaxis()->SetRangeUser(-0.5, 14);
+   graph3->Draw("AP");
+
+   auto graph4 = GetGraph(nClusters);
+   graph4->SetMarkerColor(kBlue);
+   graph4->Draw("SAME P");
+
+   legend = new TLegend(0.35, 0.725, 0.85, 0.875);
+   legend->AddEntry(graph3, "Number of currently cached pages", "p");
+   legend->AddEntry(graph4, "Number of currently cached clusters", "p");
+   legend->Draw();
+
+   // ===============================
+
+   Read(EReadMode::kInformed, sumLoadedClusters, sumUnsealedPages, nClusters, nPages);
+
+   canvas->cd(2);
+   auto graph5 = GetGraph(sumUnsealedPages);
+   graph5->SetLineColor(kRed);
+   graph5->Draw("AL");
+
+   auto graph6 = GetGraph(sumLoadedClusters);
+   graph6->SetLineColor(kBlue);
+   graph6->Draw("SAME L");
+
+   latex.SetTextAlign(22);
+   latex.DrawLatexNDC(0.5, 0.95, "Informed Reading");
+
+   canvas->cd(4);
+
+   auto graph7 = GetGraph(nPages);
+   graph7->SetMarkerColor(kRed);
+   graph7->GetYaxis()->SetNdivisions(8);
+   graph7->GetYaxis()->SetRangeUser(-0.5, 14);
+   graph7->Draw("AP");
+
+   auto graph8 = GetGraph(nClusters);
+   graph8->SetMarkerColor(kBlue);
+   graph8->Draw("SAME P");
+}