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 ROOT7
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 ROOT7_RClusterPool
17#define ROOT7_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 Experimental {
33
34namespace Internal {
35class RPageSource;
36
37// clang-format off
38/**
39\class ROOT::Experimental::Internal::RClusterPool
40\ingroup NTuple
41\brief Managed a set of clusters containing compressed and packed pages
42
43The cluster pool steers the preloading of (partial) clusters. There is a two-step pipeline: in a first step,
44compressed pages are read from clusters into a memory buffer. The second pipeline step decompresses the pages
45and pushes them into the page pool. The actual logic of reading and unzipping is implemented by the page source.
46The cluster pool only orchestrates the work queues for reading and unzipping. It uses two threads, one for
47each pipeline step. The I/O thread for reading waits for data from storage and generates no CPU load. In contrast,
48the unzip thread is supposed to submit multi-threaded, CPU heavy work to the application's task scheduler.
49
50The unzipping step of the pipeline therefore behaves differently depending on whether or not implicit multi-threadin
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 /// Request to decompress and if necessary unpack compressed pages. The unzipped pages
67 /// are supposed to be pushed into the page pool by the page source.
68 struct RUnzipItem {
69 std::unique_ptr<RCluster> fCluster;
70 std::promise<std::unique_ptr<RCluster>> fPromise;
71 };
72
73 /// Clusters that are currently being processed by the pipeline. Every in-flight cluster has a corresponding
74 /// work item, first a read item and then an unzip item.
76 std::future<std::unique_ptr<RCluster>> fFuture;
78 /// By the time a cluster has been loaded, this cluster might not be necessary anymore. This can happen if
79 /// there are jumps in the access pattern (i.e. the access pattern deviates from linear access).
80 bool fIsExpired = false;
81
82 bool operator ==(const RInFlightCluster &other) const {
83 return (fClusterKey.fClusterId == other.fClusterKey.fClusterId) &&
85 }
86 bool operator !=(const RInFlightCluster &other) const { return !(*this == other); }
87 /// First order by cluster id, then by number of columns, than by the column ids in fColumns
88 bool operator <(const RInFlightCluster &other) const;
89 };
90
91 /// Every cluster pool is responsible for exactly one page source that triggers loading of the clusters
92 /// (GetCluster()) and is used for implementing the I/O and cluster memory allocation (PageSource::LoadClusters()).
94 /// The number of clusters before the currently active cluster that should stay in the pool if present
95 /// Reserved for later use.
96 unsigned int fWindowPre = 0;
97 /// The number of clusters that are being read in a single vector read.
98 unsigned int fClusterBunchSize;
99 /// Used as an ever-growing counter in GetCluster() to separate bunches of clusters from each other
100 std::int64_t fBunchId = 0;
101 /// The cache of clusters around the currently active cluster
102 std::vector<std::unique_ptr<RCluster>> fPool;
103
104 /// Protects the shared state between the main thread and the pipeline threads, namely the read and unzip
105 /// work queues and the in-flight clusters vector
106 std::mutex fLockWorkQueue;
107 /// The clusters that were handed off to the I/O thread
108 std::vector<RInFlightCluster> fInFlightClusters;
109 /// Signals a non-empty I/O work queue
110 std::condition_variable fCvHasReadWork;
111 /// The communication channel to the I/O thread
112 std::deque<RReadItem> fReadQueue;
113 /// The lock associated with the fCvHasUnzipWork conditional variable
114 std::mutex fLockUnzipQueue;
115 /// Signals non-empty unzip work queue
116 std::condition_variable fCvHasUnzipWork;
117 /// The communication channel between the I/O thread and the unzip thread
118 std::deque<RUnzipItem> fUnzipQueue;
119
120 /// The I/O thread calls RPageSource::LoadClusters() asynchronously. The thread is mostly waiting for the
121 /// data to arrive (blocked by the kernel) and therefore can safely run in addition to the application
122 /// main threads.
123 std::thread fThreadIo;
124 /// The unzip thread takes a loaded cluster and passes it to fPageSource->UnzipCluster() on it. If implicit
125 /// multi-threading is turned off, the UnzipCluster() call is a no-op. Otherwise, the UnzipCluster() call
126 /// schedules the unzipping of pages using the application's task scheduler.
127 std::thread fThreadUnzip;
128
129 /// Every cluster id has at most one corresponding RCluster pointer in the pool
130 RCluster *FindInPool(DescriptorId_t clusterId) const;
131 /// Returns an index of an unused element in fPool; callers of this function (GetCluster() and WaitFor())
132 /// make sure that a free slot actually exists
133 size_t FindFreeSlot() const;
134 /// The I/O thread routine, there is exactly one I/O thread in-flight for every cluster pool
135 void ExecReadClusters();
136 /// The unzip thread routine which takes a loaded cluster and passes it to fPageSource.UnzipCluster (which
137 /// might be a no-op if IMT is off). Marks the cluster as ready to be picked up by the main thread.
138 void ExecUnzipClusters();
139 /// Returns the given cluster from the pool, which needs to contain at least the columns `physicalColumns`.
140 /// Executed at the end of GetCluster when all missing data pieces have been sent to the load queue.
141 /// Ideally, the function returns without blocking if the cluster is already in the pool.
142 RCluster *WaitFor(DescriptorId_t clusterId, const RCluster::ColumnSet_t &physicalColumns);
143
144public:
145 static constexpr unsigned int kDefaultClusterBunchSize = 1;
146 RClusterPool(RPageSource &pageSource, unsigned int clusterBunchSize);
147 explicit RClusterPool(RPageSource &pageSource) : RClusterPool(pageSource, kDefaultClusterBunchSize) {}
148 RClusterPool(const RClusterPool &other) = delete;
149 RClusterPool &operator =(const RClusterPool &other) = delete;
151
152 /// Returns the requested cluster either from the pool or, in case of a cache miss, lets the I/O thread load
153 /// the cluster in the pool, blocks until done, and then returns it. Triggers along the way the background loading
154 /// of the following fWindowPost number of clusters. The returned cluster has at least all the pages of
155 /// `physicalColumns` and possibly pages of other columns, too. If implicit multi-threading is turned on, the
156 /// uncompressed pages of the returned cluster are already pushed into the page pool associated with the page source
157 /// upon return. The cluster remains valid until the next call to GetCluster().
158 RCluster *GetCluster(DescriptorId_t clusterId, const RCluster::ColumnSet_t &physicalColumns);
159
160 /// Used by the unit tests to drain the queue of clusters to be preloaded
162}; // class RClusterPool
163
164} // namespace Internal
165} // namespace Experimental
166} // namespace ROOT
167
168#endif
Managed a set of clusters containing compressed and packed pages.
std::deque< RUnzipItem > fUnzipQueue
The communication channel between the I/O thread and the unzip thread.
unsigned int fWindowPre
The number of clusters before the currently active cluster that should stay in the pool if present Re...
void WaitForInFlightClusters()
Used by the unit tests to drain the queue of clusters to be preloaded.
std::condition_variable fCvHasReadWork
Signals a non-empty I/O work queue.
RCluster * GetCluster(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...
RCluster * FindInPool(DescriptorId_t clusterId) const
Every cluster id has at most one corresponding RCluster pointer in the pool.
unsigned int fClusterBunchSize
The number of clusters that are being read in a single vector read.
RClusterPool(const RClusterPool &other)=delete
static constexpr unsigned int kDefaultClusterBunchSize
RCluster * WaitFor(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::thread fThreadUnzip
The unzip thread takes a loaded cluster and passes it to fPageSource->UnzipCluster() on it.
std::condition_variable fCvHasUnzipWork
Signals non-empty unzip work queue.
std::vector< std::unique_ptr< RCluster > > fPool
The cache of clusters around the currently active cluster.
void ExecReadClusters()
The I/O thread routine, there is exactly one I/O thread in-flight for every cluster pool.
std::mutex fLockUnzipQueue
The lock associated with the fCvHasUnzipWork conditional variable.
void ExecUnzipClusters()
The unzip thread routine which takes a loaded cluster and passes it to fPageSource....
std::mutex fLockWorkQueue
Protects the shared state between the main thread and the pipeline threads, namely the read and unzip...
RClusterPool & operator=(const RClusterPool &other)=delete
RPageSource & fPageSource
Every cluster pool is responsible for exactly one page source that triggers loading of the clusters (...
std::deque< RReadItem > fReadQueue
The communication channel to the I/O thread.
std::vector< RInFlightCluster > fInFlightClusters
The clusters that were handed off to the I/O thread.
size_t FindFreeSlot() const
Returns an index of an unused element in fPool; callers of this function (GetCluster() and WaitFor())...
std::int64_t fBunchId
Used as an ever-growing counter in GetCluster() to separate bunches of clusters from each other.
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:152
std::unordered_set< DescriptorId_t > ColumnSet_t
Definition RCluster.hxx:154
Abstract interface to read data from an ntuple.
std::uint64_t DescriptorId_t
Distriniguishes elements of the same type within a descriptor, e.g. different fields.
This file contains a specialised ROOT message handler to test for diagnostic in unit tests.
Clusters that are currently being processed by the pipeline.
std::future< std::unique_ptr< RCluster > > fFuture
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 fIsExpired
By the time a cluster has been loaded, this cluster might not be necessary anymore.
bool operator==(const RInFlightCluster &other) const
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
Request to decompress and if necessary unpack compressed pages.
std::promise< std::unique_ptr< RCluster > > fPromise
The identifiers that specifies the content of a (partial) cluster.
Definition RCluster.hxx:156