Logo ROOT  
Reference Guide
 
Loading...
Searching...
No Matches
RClusterPool.hxx
Go to the documentation of this file.
1/// \file ROOT/RClusterPool.hxx
2/// \ingroup NTuple
3/// \author Jakob Blomer <jblomer@cern.ch>
4/// \date 2020-03-11
5/// \warning This is part of the ROOT 7 prototype! It will change without notice. It might trigger earthquakes. Feedback
6/// is welcome!
7
8/*************************************************************************
9 * Copyright (C) 1995-2020, Rene Brun and Fons Rademakers. *
10 * All rights reserved. *
11 * *
12 * For the licensing terms see $ROOTSYS/LICENSE. *
13 * For the list of contributors see $ROOTSYS/README/CREDITS. *
14 *************************************************************************/
15
16#ifndef ROOT_RClusterPool
17#define ROOT_RClusterPool
18
19#include <ROOT/RCluster.hxx>
20#include <ROOT/RNTupleUtil.hxx>
21
22#include <condition_variable>
23#include <deque>
24#include <memory>
25#include <mutex>
26#include <future>
27#include <thread>
28#include <set>
29#include <vector>
30
31namespace ROOT {
32namespace Internal {
33class RPageSource;
34}
35
36namespace Internal {
37
38// clang-format off
39/**
40\class ROOT::Internal::RClusterPool
41\ingroup NTuple
42\brief Managed a set of clusters containing compressed and packed pages
43
44The cluster pool steers the preloading of (partial) clusters. There is a two-step pipeline: in a first step,
45compressed pages are read from clusters into a memory buffer. The second pipeline step decompresses the pages
46and pushes them into the page pool. The actual logic of reading and unzipping is implemented by the page source.
47The cluster pool only orchestrates the work queues for reading and unzipping. It uses one extra I/O thread for
48reading waits for data from storage and generates no CPU load.
49
50The unzipping step of the pipeline therefore behaves differently depending on whether or not implicit multi-threading
51is turned on. If it is turned off, i.e. in a single-threaded environment, the cluster pool will only read the
52compressed pages and the page source has to uncompresses pages at a later point when data from the page is requested.
53*/
54// clang-format on
56private:
57 /// Request to load a subset of the columns of a particular cluster.
58 /// Work items come in groups and are executed by the page source.
59 struct RReadItem {
60 /// Items with different bunch ids are scheduled for different vector reads
61 std::int64_t fBunchId = -1;
62 std::promise<std::unique_ptr<RCluster>> fPromise;
64 };
65
66 /// Clusters that are currently being processed by the pipeline. Every in-flight cluster has a corresponding
67 /// work item, first a read item and then an unzip item.
69 std::future<std::unique_ptr<RCluster>> fFuture;
71
72 bool operator ==(const RInFlightCluster &other) const {
73 return (fClusterKey.fClusterId == other.fClusterKey.fClusterId) &&
74 (fClusterKey.fPhysicalColumnSet == other.fClusterKey.fPhysicalColumnSet);
75 }
76 bool operator !=(const RInFlightCluster &other) const { return !(*this == other); }
77 /// First order by cluster id, then by number of columns, than by the column ids in fColumns
78 bool operator <(const RInFlightCluster &other) const;
79 };
80
81 /// Every cluster pool is responsible for exactly one page source that triggers loading of the clusters
82 /// (GetCluster()) and is used for implementing the I/O and cluster memory allocation (PageSource::LoadClusters()).
84 /// The number of clusters before the currently active cluster that should stay in the pool if present
85 /// Reserved for later use.
86 unsigned int fWindowPre = 0;
87 /// The number of clusters that are being read in a single vector read.
88 unsigned int fClusterBunchSize;
89 /// Used as an ever-growing counter in GetCluster() to separate bunches of clusters from each other
90 std::int64_t fBunchId = 0;
91 /// The cache of clusters around the currently active cluster
92 std::vector<std::unique_ptr<RCluster>> fPool;
93
94 /// Protects the shared state between the main thread and the I/O thread, namely the work queue and the in-flight
95 /// clusters vector
96 std::mutex fLockWorkQueue;
97 /// The clusters that were handed off to the I/O thread
98 std::vector<RInFlightCluster> fInFlightClusters;
99 /// Signals a non-empty I/O work queue
100 std::condition_variable fCvHasReadWork;
101 /// The communication channel to the I/O thread
102 std::deque<RReadItem> fReadQueue;
103
104 /// The I/O thread calls RPageSource::LoadClusters() asynchronously. The thread is mostly waiting for the
105 /// data to arrive (blocked by the kernel) and therefore can safely run in addition to the application
106 /// main threads.
107 std::thread fThreadIo;
108
109 /// Every cluster id has at most one corresponding RCluster pointer in the pool
111 /// Returns an index of an unused element in fPool; callers of this function (GetCluster() and WaitFor())
112 /// make sure that a free slot actually exists
113 size_t FindFreeSlot() const;
114 /// The I/O thread routine, there is exactly one I/O thread in-flight for every cluster pool
115 void ExecReadClusters();
116 /// Returns the given cluster from the pool, which needs to contain at least the columns `physicalColumns`.
117 /// Executed at the end of GetCluster when all missing data pieces have been sent to the load queue.
118 /// Ideally, the function returns without blocking if the cluster is already in the pool.
120
121public:
122 static constexpr unsigned int kDefaultClusterBunchSize = 1;
130
131 /// Returns the requested cluster either from the pool or, in case of a cache miss, lets the I/O thread load
132 /// the cluster in the pool, blocks until done, and then returns it. Triggers along the way the background loading
133 /// of the following fWindowPost number of clusters. The returned cluster has at least all the pages of
134 /// `physicalColumns` and possibly pages of other columns, too. If implicit multi-threading is turned on, the
135 /// uncompressed pages of the returned cluster are already pushed into the page pool associated with the page source
136 /// upon return. The cluster remains valid until the next call to GetCluster().
138
139 /// Used by the unit tests to drain the queue of clusters to be preloaded
141}; // class RClusterPool
142
143} // namespace Internal
144} // namespace ROOT
145
146#endif
ROOT::Detail::TRangeCast< T, true > TRangeDynCast
TRangeDynCast is an adapter class that allows the typed iteration through a TCollection.
Managed a set of clusters containing compressed and packed pages.
RCluster * WaitFor(ROOT::DescriptorId_t clusterId, const RCluster::ColumnSet_t &physicalColumns)
Returns the given cluster from the pool, which needs to contain at least the columns physicalColumns.
std::condition_variable fCvHasReadWork
Signals a non-empty I/O work queue.
RClusterPool(ROOT::Internal::RPageSource &pageSource)
unsigned int fClusterBunchSize
The number of clusters that are being read in a single vector read.
void WaitForInFlightClusters()
Used by the unit tests to drain the queue of clusters to be preloaded.
std::vector< RInFlightCluster > fInFlightClusters
The clusters that were handed off to the I/O thread.
std::vector< std::unique_ptr< RCluster > > fPool
The cache of clusters around the currently active cluster.
RCluster * FindInPool(ROOT::DescriptorId_t clusterId) const
Every cluster id has at most one corresponding RCluster pointer in the pool.
std::deque< RReadItem > fReadQueue
The communication channel to the I/O thread.
void ExecReadClusters()
The I/O thread routine, there is exactly one I/O thread in-flight for every cluster pool.
unsigned int fWindowPre
The number of clusters before the currently active cluster that should stay in the pool if present Re...
RCluster * GetCluster(ROOT::DescriptorId_t clusterId, const RCluster::ColumnSet_t &physicalColumns)
Returns the requested cluster either from the pool or, in case of a cache miss, lets the I/O thread l...
RClusterPool(ROOT::Internal::RPageSource &pageSource, unsigned int clusterBunchSize)
std::int64_t fBunchId
Used as an ever-growing counter in GetCluster() to separate bunches of clusters from each other.
RClusterPool(const RClusterPool &other)=delete
RClusterPool & operator=(const RClusterPool &other)=delete
std::mutex fLockWorkQueue
Protects the shared state between the main thread and the I/O thread, namely the work queue and the i...
size_t FindFreeSlot() const
Returns an index of an unused element in fPool; callers of this function (GetCluster() and WaitFor())...
ROOT::Internal::RPageSource & fPageSource
Every cluster pool is responsible for exactly one page source that triggers loading of the clusters (...
static constexpr unsigned int kDefaultClusterBunchSize
std::thread fThreadIo
The I/O thread calls RPageSource::LoadClusters() asynchronously.
An in-memory subset of the packed and compressed pages of a cluster.
Definition RCluster.hxx:148
std::unordered_set< ROOT::DescriptorId_t > ColumnSet_t
Definition RCluster.hxx:150
Abstract interface to read data from an ntuple.
tbb::task_arena is an alias of tbb::interface7::task_arena, which doesn't allow to forward declare tb...
std::uint64_t DescriptorId_t
Distriniguishes elements of the same type within a descriptor, e.g. different fields.
Clusters that are currently being processed by the pipeline.
bool operator!=(const RInFlightCluster &other) const
bool operator<(const RInFlightCluster &other) const
First order by cluster id, then by number of columns, than by the column ids in fColumns.
bool operator==(const RInFlightCluster &other) const
std::future< std::unique_ptr< RCluster > > fFuture
Request to load a subset of the columns of a particular cluster.
std::int64_t fBunchId
Items with different bunch ids are scheduled for different vector reads.
std::promise< std::unique_ptr< RCluster > > fPromise
The identifiers that specifies the content of a (partial) cluster.
Definition RCluster.hxx:152
ROOT::DescriptorId_t fClusterId
Definition RCluster.hxx:153