Logo ROOT  
Reference Guide
 
Loading...
Searching...
No Matches
RPageStorage.cxx
Go to the documentation of this file.
1/// \file RPageStorage.cxx
2/// \ingroup NTuple ROOT7
3/// \author Jakob Blomer <jblomer@cern.ch>
4/// \date 2018-10-04
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-2019, 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#include <ROOT/RPageStorage.hxx>
18#include <ROOT/RColumn.hxx>
19#include <ROOT/RField.hxx>
22#include <ROOT/RNTupleModel.hxx>
23#include <ROOT/RPagePool.hxx>
24#include <ROOT/RPageSinkBuf.hxx>
26#include <ROOT/RStringView.hxx>
27#ifdef R__ENABLE_DAOS
29#endif
30
31#include <Compression.h>
32#include <TError.h>
33
34#include <utility>
35
36
38{
39}
40
42{
43}
44
45
46//------------------------------------------------------------------------------
47
48
50 : RPageStorage(name), fMetrics(""), fOptions(options)
51{
52}
53
55{
56}
57
58std::unique_ptr<ROOT::Experimental::Detail::RPageSource> ROOT::Experimental::Detail::RPageSource::Create(
59 std::string_view ntupleName, std::string_view location, const RNTupleReadOptions &options)
60{
61 if (ntupleName.empty()) {
62 throw RException(R__FAIL("empty RNTuple name"));
63 }
64 if (location.empty()) {
65 throw RException(R__FAIL("empty storage location"));
66 }
67 if (location.find("daos://") == 0)
68#ifdef R__ENABLE_DAOS
69 return std::make_unique<RPageSourceDaos>(ntupleName, location, options);
70#else
71 throw RException(R__FAIL("This RNTuple build does not support DAOS."));
72#endif
73
74 return std::make_unique<RPageSourceFile>(ntupleName, location, options);
75}
76
79{
81 auto columnId = fDescriptor.FindColumnId(fieldId, column.GetIndex());
83 fActiveColumns.emplace(columnId);
84 return ColumnHandle_t{columnId, &column};
85}
86
88{
89 fActiveColumns.erase(columnHandle.fId);
90}
91
93{
94 return fDescriptor.GetNEntries();
95}
96
98{
99 return fDescriptor.GetNElements(columnHandle.fId);
100}
101
103{
104 // TODO(jblomer) distinguish trees
105 return columnHandle.fId;
106}
107
109{
110 if (fTaskScheduler)
111 UnzipClusterImpl(cluster);
112}
113
114
116 const RSealedPage &sealedPage, const RColumnElementBase &element)
117{
118 const auto bytesPacked = element.GetPackedSize(sealedPage.fNElements);
119 const auto pageSize = element.GetSize() * sealedPage.fNElements;
120
121 // TODO(jblomer): We might be able to do better memory handling for unsealing pages than a new malloc for every
122 // new page.
123 auto pageBuffer = std::make_unique<unsigned char[]>(bytesPacked);
124 if (sealedPage.fSize != bytesPacked) {
125 fDecompressor->Unzip(sealedPage.fBuffer, sealedPage.fSize, bytesPacked, pageBuffer.get());
126 } else {
127 // We cannot simply map the sealed page as we don't know its life time. Specialized page sources
128 // may decide to implement to not use UnsealPage but to custom mapping / decompression code.
129 // Note that usually pages are compressed.
130 memcpy(pageBuffer.get(), sealedPage.fBuffer, bytesPacked);
131 }
132
133 if (!element.IsMappable()) {
134 auto unpackedBuffer = new unsigned char[pageSize];
135 element.Unpack(unpackedBuffer, pageBuffer.get(), sealedPage.fNElements);
136 pageBuffer = std::unique_ptr<unsigned char []>(unpackedBuffer);
137 }
138
139 return pageBuffer;
140}
141
143{
144 fMetrics = RNTupleMetrics(prefix);
145 fCounters = std::unique_ptr<RCounters>(new RCounters{
146 *fMetrics.MakeCounter<RNTupleAtomicCounter*>("nReadV", "", "number of vector read requests"),
147 *fMetrics.MakeCounter<RNTupleAtomicCounter*>("nRead", "", "number of byte ranges read"),
148 *fMetrics.MakeCounter<RNTupleAtomicCounter*>("szReadPayload", "B", "volume read from storage (required)"),
149 *fMetrics.MakeCounter<RNTupleAtomicCounter*>("szReadOverhead", "B", "volume read from storage (overhead)"),
150 *fMetrics.MakeCounter<RNTupleAtomicCounter*>("szUnzip", "B", "volume after unzipping"),
151 *fMetrics.MakeCounter<RNTupleAtomicCounter*>("nClusterLoaded", "",
152 "number of partial clusters preloaded from storage"),
153 *fMetrics.MakeCounter<RNTupleAtomicCounter*>("nPageLoaded", "", "number of pages loaded from storage"),
154 *fMetrics.MakeCounter<RNTupleAtomicCounter*>("nPagePopulated", "", "number of populated pages"),
155 *fMetrics.MakeCounter<RNTupleAtomicCounter*>("timeWallRead", "ns", "wall clock time spent reading"),
156 *fMetrics.MakeCounter<RNTupleAtomicCounter*>("timeWallUnzip", "ns", "wall clock time spent decompressing"),
157 *fMetrics.MakeCounter<RNTupleTickCounter<RNTupleAtomicCounter>*>("timeCpuRead", "ns", "CPU time spent reading"),
158 *fMetrics.MakeCounter<RNTupleTickCounter<RNTupleAtomicCounter>*> ("timeCpuUnzip", "ns",
159 "CPU time spent decompressing"),
160 *fMetrics.MakeCounter<RNTupleCalcPerf*> ("bwRead", "MB/s", "bandwidth compressed bytes read per second",
161 fMetrics, [](const RNTupleMetrics &metrics) -> std::pair<bool, double> {
162 if (const auto szReadPayload = metrics.GetLocalCounter("szReadPayload")) {
163 if (const auto szReadOverhead = metrics.GetLocalCounter("szReadOverhead")) {
164 if (const auto timeWallRead = metrics.GetLocalCounter("timeWallRead")) {
165 if (auto walltime = timeWallRead->GetValueAsInt()) {
166 double payload = szReadPayload->GetValueAsInt();
167 double overhead = szReadOverhead->GetValueAsInt();
168 // unit: bytes / nanosecond = GB/s
169 return {true, (1000. * (payload + overhead) / walltime)};
170 }
171 }
172 }
173 }
174 return {false, -1.};
175 }
176 ),
177 *fMetrics.MakeCounter<RNTupleCalcPerf*> ("bwReadUnzip", "MB/s", "bandwidth uncompressed bytes read per second",
178 fMetrics, [](const RNTupleMetrics &metrics) -> std::pair<bool, double> {
179 if (const auto szUnzip = metrics.GetLocalCounter("szUnzip")) {
180 if (const auto timeWallRead = metrics.GetLocalCounter("timeWallRead")) {
181 if (auto walltime = timeWallRead->GetValueAsInt()) {
182 double unzip = szUnzip->GetValueAsInt();
183 // unit: bytes / nanosecond = GB/s
184 return {true, 1000. * unzip / walltime};
185 }
186 }
187 }
188 return {false, -1.};
189 }
190 ),
191 *fMetrics.MakeCounter<RNTupleCalcPerf*> ("bwUnzip", "MB/s", "decompression bandwidth of uncompressed bytes per second",
192 fMetrics, [](const RNTupleMetrics &metrics) -> std::pair<bool, double> {
193 if (const auto szUnzip = metrics.GetLocalCounter("szUnzip")) {
194 if (const auto timeWallUnzip = metrics.GetLocalCounter("timeWallUnzip")) {
195 if (auto walltime = timeWallUnzip->GetValueAsInt()) {
196 double unzip = szUnzip->GetValueAsInt();
197 // unit: bytes / nanosecond = GB/s
198 return {true, 1000. * unzip / walltime};
199 }
200 }
201 }
202 return {false, -1.};
203 }
204 ),
205 *fMetrics.MakeCounter<RNTupleCalcPerf*> ("rtReadEfficiency", "", "ratio of payload over all bytes read",
206 fMetrics, [](const RNTupleMetrics &metrics) -> std::pair<bool, double> {
207 if (const auto szReadPayload = metrics.GetLocalCounter("szReadPayload")) {
208 if (const auto szReadOverhead = metrics.GetLocalCounter("szReadOverhead")) {
209 if (auto payload = szReadPayload->GetValueAsInt()) {
210 // r/(r+o) = 1/((r+o)/r) = 1/(1 + o/r)
211 return {true, 1./(1. + (1. * szReadOverhead->GetValueAsInt()) / payload)};
212 }
213 }
214 }
215 return {false, -1.};
216 }
217 ),
218 *fMetrics.MakeCounter<RNTupleCalcPerf*> ("rtCompression", "", "ratio of compressed bytes / uncompressed bytes",
219 fMetrics, [](const RNTupleMetrics &metrics) -> std::pair<bool, double> {
220 if (const auto szReadPayload = metrics.GetLocalCounter("szReadPayload")) {
221 if (const auto szUnzip = metrics.GetLocalCounter("szUnzip")) {
222 if (auto unzip = szUnzip->GetValueAsInt()) {
223 return {true, (1. * szReadPayload->GetValueAsInt()) / unzip};
224 }
225 }
226 }
227 return {false, -1.};
228 }
229 )
230 });
231}
232
233
234//------------------------------------------------------------------------------
235
236
238 : RPageStorage(name), fMetrics(""), fOptions(options.Clone())
239{
240}
241
243{
244}
245
246std::unique_ptr<ROOT::Experimental::Detail::RPageSink> ROOT::Experimental::Detail::RPageSink::Create(
247 std::string_view ntupleName, std::string_view location, const RNTupleWriteOptions &options)
248{
249 if (ntupleName.empty()) {
250 throw RException(R__FAIL("empty RNTuple name"));
251 }
252 if (location.empty()) {
253 throw RException(R__FAIL("empty storage location"));
254 }
255 std::unique_ptr<ROOT::Experimental::Detail::RPageSink> realSink;
256 if (location.find("daos://") == 0) {
257#ifdef R__ENABLE_DAOS
258 realSink = std::make_unique<RPageSinkDaos>(ntupleName, location, options);
259#else
260 throw RException(R__FAIL("This RNTuple build does not support DAOS."));
261#endif
262 } else {
263 realSink = std::make_unique<RPageSinkFile>(ntupleName, location, options);
264 }
265
266 if (options.GetUseBufferedWrite())
267 return std::make_unique<RPageSinkBuf>(std::move(realSink));
268 return realSink;
269}
270
273{
274 auto columnId = fLastColumnId++;
275 fDescriptorBuilder.AddColumn(columnId, fieldId, column.GetVersion(), column.GetModel(), column.GetIndex());
276 return ColumnHandle_t{columnId, &column};
277}
278
279
281{
282 fDescriptorBuilder.SetNTuple(fNTupleName, model.GetDescription(), "undefined author",
283 model.GetVersion(), model.GetUuid());
284
285 auto &fieldZero = *model.GetFieldZero();
286 fDescriptorBuilder.AddField(
288 .FieldId(fLastFieldId)
289 .MakeDescriptor()
290 .Unwrap()
291 );
292 fieldZero.SetOnDiskId(fLastFieldId);
293 for (auto& f : *model.GetFieldZero()) {
294 fLastFieldId++;
295 fDescriptorBuilder.AddField(
297 .FieldId(fLastFieldId)
299 .Unwrap()
300 );
301 fDescriptorBuilder.AddFieldLink(f.GetParent()->GetOnDiskId(), fLastFieldId);
302 f.SetOnDiskId(fLastFieldId);
303 f.ConnectPageSink(*this); // issues in turn one or several calls to AddColumn()
304 }
305
306 auto nColumns = fLastColumnId;
307 for (DescriptorId_t i = 0; i < nColumns; ++i) {
309 columnRange.fColumnId = i;
310 columnRange.fFirstElementIndex = 0;
311 columnRange.fNElements = 0;
312 columnRange.fCompressionSettings = GetWriteOptions().GetCompression();
313 fOpenColumnRanges.emplace_back(columnRange);
315 pageRange.fColumnId = i;
316 fOpenPageRanges.emplace_back(std::move(pageRange));
317 }
318
319 CreateImpl(model);
320}
321
322
324{
325 fOpenColumnRanges.at(columnHandle.fId).fNElements += page.GetNElements();
326
328 pageInfo.fNElements = page.GetNElements();
329 pageInfo.fLocator = CommitPageImpl(columnHandle, page);
330 fOpenPageRanges.at(columnHandle.fId).fPageInfos.emplace_back(pageInfo);
331}
332
333
337{
338 fOpenColumnRanges.at(columnId).fNElements += sealedPage.fNElements;
339
341 pageInfo.fNElements = sealedPage.fNElements;
342 pageInfo.fLocator = CommitSealedPageImpl(columnId, sealedPage);
343 fOpenPageRanges.at(columnId).fPageInfos.emplace_back(pageInfo);
344}
345
346
348{
349 auto nbytes = CommitClusterImpl(nEntries);
350
351 R__ASSERT((nEntries - fPrevClusterNEntries) < ClusterSize_t(-1));
352 fDescriptorBuilder.AddCluster(fLastClusterId, RNTupleVersion(), fPrevClusterNEntries,
353 ClusterSize_t(nEntries - fPrevClusterNEntries));
354 for (auto &range : fOpenColumnRanges) {
355 fDescriptorBuilder.AddClusterColumnRange(fLastClusterId, range);
356 range.fFirstElementIndex += range.fNElements;
357 range.fNElements = 0;
358 }
359 for (auto &range : fOpenPageRanges) {
361 std::swap(fullRange, range);
362 range.fColumnId = fullRange.fColumnId;
363 fDescriptorBuilder.AddClusterPageRange(fLastClusterId, std::move(fullRange));
364 }
365 fPrevClusterNEntries = nEntries;
366 ++fLastClusterId;
367 return nbytes;
368}
369
372 const RColumnElementBase &element, int compressionSetting, void *buf)
373{
374 unsigned char *pageBuf = reinterpret_cast<unsigned char *>(page.GetBuffer());
375 bool isAdoptedBuffer = true;
376 auto packedBytes = page.GetNBytes();
377
378 if (!element.IsMappable()) {
379 packedBytes = element.GetPackedSize(page.GetNElements());
380 pageBuf = new unsigned char[packedBytes];
381 isAdoptedBuffer = false;
382 element.Pack(pageBuf, page.GetBuffer(), page.GetNElements());
383 }
384 auto zippedBytes = packedBytes;
385
386 if ((compressionSetting != 0) || !element.IsMappable()) {
387 zippedBytes = RNTupleCompressor::Zip(pageBuf, packedBytes, compressionSetting, buf);
388 if (!isAdoptedBuffer)
389 delete[] pageBuf;
390 pageBuf = reinterpret_cast<unsigned char *>(buf);
391 isAdoptedBuffer = true;
392 }
393
394 R__ASSERT(isAdoptedBuffer);
395
396 return RSealedPage{pageBuf, zippedBytes, page.GetNElements()};
397}
398
401 const RPage &page, const RColumnElementBase &element, int compressionSetting)
402{
403 R__ASSERT(fCompressor);
404 return SealPage(page, element, compressionSetting, fCompressor->GetZipBuffer());
405}
406
408{
409 fMetrics = RNTupleMetrics(prefix);
410 fCounters = std::unique_ptr<RCounters>(new RCounters{
411 *fMetrics.MakeCounter<RNTupleAtomicCounter*>("nPageCommitted", "", "number of pages committed to storage"),
412 *fMetrics.MakeCounter<RNTupleAtomicCounter*>("szWritePayload", "B", "volume written for committed pages"),
413 *fMetrics.MakeCounter<RNTupleAtomicCounter*>("szZip", "B", "volume before zipping"),
414 *fMetrics.MakeCounter<RNTupleAtomicCounter*>("timeWallWrite", "ns", "wall clock time spent writing"),
415 *fMetrics.MakeCounter<RNTupleAtomicCounter*>("timeWallZip", "ns", "wall clock time spent compressing"),
416 *fMetrics.MakeCounter<RNTupleTickCounter<RNTupleAtomicCounter>*>("timeCpuWrite", "ns", "CPU time spent writing"),
417 *fMetrics.MakeCounter<RNTupleTickCounter<RNTupleAtomicCounter>*> ("timeCpuZip", "ns",
418 "CPU time spent compressing")
419 });
420}
#define R__FAIL(msg)
Short-hand to return an RResult<T> in an error state; the RError is implicitly converted into RResult...
Definition RError.hxx:291
#define f(i)
Definition RSha256.hxx:104
#define R__ASSERT(e)
Definition TError.h:118
char name[80]
Definition TGX11.cxx:110
An in-memory subset of the packed and compressed pages of a cluster.
Definition RCluster.hxx:154
virtual void Pack(void *destination, void *source, std::size_t count) const
If the on-storage layout and the in-memory layout differ, packing creates an on-disk page from an in-...
virtual bool IsMappable() const
Derived, typed classes tell whether the on-storage layout is bitwise identical to the memory layout.
virtual void Unpack(void *destination, void *source, std::size_t count) const
If the on-storage layout and the in-memory layout differ, unpacking creates a memory page from an on-...
std::size_t GetPackedSize(std::size_t nElements) const
const RColumnModel & GetModel() const
Definition RColumn.hxx:309
std::uint32_t GetIndex() const
Definition RColumn.hxx:310
RNTupleVersion GetVersion() const
Definition RColumn.hxx:315
NTupleSize_t GetNElements() const
Definition RColumn.hxx:307
void SetOnDiskId(DescriptorId_t id)
Definition RField.hxx:248
A thread-safe integral performance counter.
A metric element that computes its floating point value from other counters.
std::int64_t GetValueAsInt() const override
size_t Zip(const void *from, size_t nbytes, int compression, Writer_t fnWriter)
Returns the size of the compressed data.
A collection of Counter objects with a name, a unit, and a description.
const RNTuplePerfCounter * GetLocalCounter(std::string_view name) const
Searches counters registered in this object only. Returns nullptr if name is not found.
An either thread-safe or non thread safe counter for CPU ticks.
RSealedPage SealPage(const RPage &page, const RColumnElementBase &element, int compressionSetting)
Helper for streaming a page.
void EnableDefaultMetrics(const std::string &prefix)
Enables the default set of metrics provided by RPageSink.
RPageSink(std::string_view ntupleName, const RNTupleWriteOptions &options)
void CommitPage(ColumnHandle_t columnHandle, const RPage &page)
Write a page to the storage. The column must have been added before.
void CommitSealedPage(DescriptorId_t columnId, const RPageStorage::RSealedPage &sealedPage)
Write a preprocessed page to storage.
static std::unique_ptr< RPageSink > Create(std::string_view ntupleName, std::string_view location, const RNTupleWriteOptions &options=RNTupleWriteOptions())
Guess the concrete derived page source from the file name (location)
std::uint64_t CommitCluster(NTupleSize_t nEntries)
Finalize the current cluster and create a new one for the following data.
ColumnHandle_t AddColumn(DescriptorId_t fieldId, const RColumn &column) final
Register a new column.
void EnableDefaultMetrics(const std::string &prefix)
Enables the default set of metrics provided by RPageSource.
std::unique_ptr< unsigned char[]> UnsealPage(const RSealedPage &sealedPage, const RColumnElementBase &element)
Helper for unstreaming a page.
void DropColumn(ColumnHandle_t columnHandle) override
Unregisters a column.
NTupleSize_t GetNElements(ColumnHandle_t columnHandle)
ColumnHandle_t AddColumn(DescriptorId_t fieldId, const RColumn &column) override
Register a new column.
static std::unique_ptr< RPageSource > Create(std::string_view ntupleName, std::string_view location, const RNTupleReadOptions &options=RNTupleReadOptions())
Guess the concrete derived page source from the file name (location)
RPageSource(std::string_view ntupleName, const RNTupleReadOptions &fOptions)
void UnzipCluster(RCluster *cluster)
Parallel decompression and unpacking of the pages in the given cluster.
ColumnId_t GetColumnId(ColumnHandle_t columnHandle)
Common functionality of an ntuple storage for both reading and writing.
A page is a slice of a column that is mapped into memory.
Definition RPage.hxx:41
ClusterSize_t::ValueType GetNElements() const
Definition RPage.hxx:83
ClusterSize_t::ValueType GetNBytes() const
The space taken by column elements in the buffer.
Definition RPage.hxx:81
Base class for all ROOT issued exceptions.
Definition RError.hxx:114
static RFieldDescriptorBuilder FromField(const Detail::RFieldBase &field)
Make a new RFieldDescriptorBuilder based off a live NTuple field.
RResult< RFieldDescriptor > MakeDescriptor() const
Attempt to make a field descriptor.
RFieldDescriptorBuilder & FieldId(DescriptorId_t fieldId)
The RNTupleModel encapulates the schema of an ntuple.
RNTupleVersion GetVersion() const
RFieldZero * GetFieldZero() const
Common user-tunable settings for reading ntuples.
For forward and backward compatibility, attach version information to the consitituents of the file f...
Common user-tunable settings for storing ntuples.
std::uint64_t NTupleSize_t
Integer type long enough to hold the maximum number of entries in a column.
RClusterSize ClusterSize_t
std::uint64_t DescriptorId_t
Distriniguishes elements of the same type within a descriptor, e.g. different fields.
std::int64_t ColumnId_t
Uniquely identifies a physical column within the scope of the current process, used to tag pages.
constexpr DescriptorId_t kInvalidDescriptorId
Default I/O performance counters that get registered in fMetrics.
Default I/O performance counters that get registered in fMetrics.
A sealed page contains the bytes of a page as written to storage (packed & compressed).
The window of element indexes of a particular column in a particular cluster.
std::int64_t fCompressionSettings
The usual format for ROOT compression settings (see Compression.h).
ClusterSize_t fNElements
A 32bit value for the number of column elements in the cluster.
We do not need to store the element size / uncompressed page size because we know to which column the...
RNTupleLocator fLocator
The meaning of fLocator depends on the storage backend.
ClusterSize_t fNElements
The sum of the elements of all the pages must match the corresponding fNElements field in fColumnRang...
Records the parition of data into pages for a particular column in a particular cluster.