Logo ROOT  
Reference Guide
 
Loading...
Searching...
No Matches
RNTupleParallelWriter.cxx
Go to the documentation of this file.
1/// \file RNTupleParallelWriter.cxx
2/// \ingroup NTuple ROOT7
3/// \author Jonas Hahnfeld <jonas.hahnfeld@cern.ch>
4/// \date 2024-02-01
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-2024, 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
17
18#include <ROOT/RNTupleModel.hxx>
20#include <ROOT/RPageSinkBuf.hxx>
21#include <ROOT/RPageStorage.hxx>
23
24#include <TError.h>
25
26namespace {
27
38
39/// An internal RPageSink that enables multiple RNTupleFillContext to write into a single common RPageSink.
40///
41/// The setup with two contexts looks as follows:
42///
43/// +------ owned by RNTupleFillContext ------+
44/// | |
45/// RPageSinkBuf --- forwards to ---> RPageSynchronizingSink ---+
46/// (and owns) |
47/// (via raw fInnerSink ptr) +-- RPageSink (usually a persistent sink)
48/// |
49/// RPageSinkBuf --- forwards to ---> RPageSynchronizingSink ---+
50/// | (and owns) |
51/// | |
52/// +------ owned by RNTupleFillContext ------+
53///
54/// The mutex used by the synchronizing sinks is owned by the RNTupleParallelWriter that also owns the original model,
55/// the "final" sink (usually a persistent sink) and keeps weak_ptr's of the contexts (to make sure they are destroyed
56/// before the writer is destructed).
57class RPageSynchronizingSink : public RPageSink {
58private:
59 /// The wrapped inner sink, not owned by this class.
60 RPageSink *fInnerSink;
61 std::mutex *fMutex;
62
63public:
64 explicit RPageSynchronizingSink(RPageSink &inner, std::mutex &mutex)
65 : RPageSink(inner.GetNTupleName(), inner.GetWriteOptions()), fInnerSink(&inner), fMutex(&mutex)
66 {
67 // Do not observe the sink's metrics: It will contain some counters for all threads, which is misleading for the
68 // users.
69 // fMetrics.ObserveMetrics(fSink->GetMetrics());
70 }
71 RPageSynchronizingSink(const RPageSynchronizingSink &) = delete;
72 RPageSynchronizingSink &operator=(const RPageSynchronizingSink &) = delete;
73
74 const RNTupleDescriptor &GetDescriptor() const final { return fInnerSink->GetDescriptor(); }
75
76 ColumnHandle_t AddColumn(DescriptorId_t, RColumn &) final { return {}; }
77 void InitImpl(RNTupleModel &) final {}
78 void UpdateSchema(const RNTupleModelChangeset &, NTupleSize_t) final
79 {
80 throw RException(R__FAIL("UpdateSchema not supported via RPageSynchronizingSink"));
81 }
82 void UpdateExtraTypeInfo(const RExtraTypeInfoDescriptor &) final
83 {
84 throw RException(R__FAIL("UpdateExtraTypeInfo not supported via RPageSynchronizingSink"));
85 }
86
87 void CommitSuppressedColumn(ColumnHandle_t handle) final { fInnerSink->CommitSuppressedColumn(handle); }
88 void CommitPage(ColumnHandle_t, const RPage &) final
89 {
90 throw RException(R__FAIL("should never commit single pages via RPageSynchronizingSink"));
91 }
92 void CommitSealedPage(DescriptorId_t, const RSealedPage &) final
93 {
94 throw RException(R__FAIL("should never commit sealed pages via RPageSynchronizingSink"));
95 }
96 void CommitSealedPageV(std::span<RPageStorage::RSealedPageGroup> ranges) final
97 {
98 fInnerSink->CommitSealedPageV(ranges);
99 }
100 std::uint64_t CommitCluster(NTupleSize_t nNewEntries) final { return fInnerSink->CommitCluster(nNewEntries); }
101 RStagedCluster StageCluster(NTupleSize_t nNewEntries) final { return fInnerSink->StageCluster(nNewEntries); }
102 void CommitStagedClusters(std::span<RStagedCluster> clusters) final { fInnerSink->CommitStagedClusters(clusters); }
103 void CommitClusterGroup() final
104 {
105 throw RException(R__FAIL("should never commit cluster group via RPageSynchronizingSink"));
106 }
107 void CommitDatasetImpl() final
108 {
109 throw RException(R__FAIL("should never commit dataset via RPageSynchronizingSink"));
110 }
111
112 RSinkGuard GetSinkGuard() final { return RSinkGuard(fMutex); }
113};
114
115} // namespace
116
118 std::unique_ptr<Internal::RPageSink> sink)
119 : fSink(std::move(sink)), fModel(std::move(model)), fMetrics("RNTupleParallelWriter")
120{
121 fModel->Freeze();
122 fSink->Init(*fModel.get());
123 fMetrics.ObserveMetrics(fSink->GetMetrics());
124}
125
127{
128 for (const auto &context : fFillContexts) {
129 if (!context.expired()) {
130 R__LOG_ERROR(NTupleLog()) << "RNTupleFillContext has not been destructed";
131 return;
132 }
133 }
134
135 // Now commit all clusters as a cluster group and then the dataset.
136 try {
137 fSink->CommitClusterGroup();
138 fSink->CommitDataset();
139 } catch (const RException &err) {
140 R__LOG_ERROR(NTupleLog()) << "failure committing ntuple: " << err.GetError().GetReport();
141 }
142}
143
144std::unique_ptr<ROOT::Experimental::RNTupleParallelWriter>
145ROOT::Experimental::RNTupleParallelWriter::Recreate(std::unique_ptr<RNTupleModel> model, std::string_view ntupleName,
146 std::string_view storage, const RNTupleWriteOptions &options)
147{
148 if (!options.GetUseBufferedWrite()) {
149 throw RException(R__FAIL("parallel writing requires buffering"));
150 }
151
152 auto sink = Internal::RPagePersistentSink::Create(ntupleName, storage, options);
153 // Cannot use std::make_unique because the constructor of RNTupleParallelWriter is private.
154 return std::unique_ptr<RNTupleParallelWriter>(new RNTupleParallelWriter(std::move(model), std::move(sink)));
155}
156
157std::unique_ptr<ROOT::Experimental::RNTupleParallelWriter>
158ROOT::Experimental::RNTupleParallelWriter::Append(std::unique_ptr<RNTupleModel> model, std::string_view ntupleName,
159 TFile &file, const RNTupleWriteOptions &options)
160{
161 if (!options.GetUseBufferedWrite()) {
162 throw RException(R__FAIL("parallel writing requires buffering"));
163 }
164
165 auto sink = std::make_unique<Internal::RPageSinkFile>(ntupleName, file, options);
166 // Cannot use std::make_unique because the constructor of RNTupleParallelWriter is private.
167 return std::unique_ptr<RNTupleParallelWriter>(new RNTupleParallelWriter(std::move(model), std::move(sink)));
168}
169
170std::shared_ptr<ROOT::Experimental::RNTupleFillContext> ROOT::Experimental::RNTupleParallelWriter::CreateFillContext()
171{
172 std::lock_guard g(fMutex);
173
174 auto model = fModel->Clone();
175
176 // TODO: Think about honoring RNTupleWriteOptions::SetUseBufferedWrite(false); this requires synchronization on every
177 // call to CommitPage() *and* preparing multiple cluster descriptors in parallel!
178 auto sink = std::make_unique<Internal::RPageSinkBuf>(std::make_unique<RPageSynchronizingSink>(*fSink, fSinkMutex));
179
180 // Cannot use std::make_shared because the constructor of RNTupleFillContext is private. Also it would mean that the
181 // (direct) memory of all contexts stays around until the vector of weak_ptr's is cleared.
182 std::shared_ptr<RNTupleFillContext> context(new RNTupleFillContext(std::move(model), std::move(sink)));
183 fFillContexts.push_back(context);
184 return context;
185}
#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:290
#define R__LOG_ERROR(...)
Definition RLogger.hxx:362
#define g(i)
Definition RSha256.hxx:105
Binding & operator=(OUT(*fun)(void))
void ObserveMetrics(RNTupleMetrics &observee)
A column is a storage-backed array of a simple, fixed-size type, from which pages can be mapped into ...
Definition RColumn.hxx:40
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 location.
Abstract interface to write data into an ntuple.
A page is a slice of a column that is mapped into memory.
Definition RPage.hxx:46
std::string GetReport() const
Format a dignostics report, e.g. for an exception message.
Definition RError.cxx:25
Base class for all ROOT issued exceptions.
Definition RError.hxx:78
const RError & GetError() const
Definition RError.hxx:82
Field specific extra type information from the header / extenstion header.
The on-storage meta-data of an ntuple.
A context for filling entries (data) into clusters of an RNTuple.
The RNTupleModel encapulates the schema of an ntuple.
A writer to fill an RNTuple from multiple contexts.
RNTupleParallelWriter(std::unique_ptr< RNTupleModel > model, std::unique_ptr< Internal::RPageSink > sink)
std::unique_ptr< RNTupleModel > fModel
The original RNTupleModel connected to fSink; needs to be destructed before it.
std::unique_ptr< Internal::RPageSink > fSink
The final RPageSink that represents the synchronization point.
static std::unique_ptr< RNTupleParallelWriter > Recreate(std::unique_ptr< RNTupleModel > model, std::string_view ntupleName, std::string_view storage, const RNTupleWriteOptions &options=RNTupleWriteOptions())
Recreate a new file and return a writer to write an ntuple.
std::shared_ptr< RNTupleFillContext > CreateFillContext()
Create a new RNTupleFillContext that can be used to fill entries and prepare clusters in parallel.
static std::unique_ptr< RNTupleParallelWriter > Append(std::unique_ptr< RNTupleModel > model, std::string_view ntupleName, TFile &file, const RNTupleWriteOptions &options=RNTupleWriteOptions())
Append an ntuple to the existing file, which must not be accessed while data is filled into any creat...
Common user-tunable settings for storing ntuples.
A ROOT file is an on-disk file, usually with extension .root, that stores objects in a file-system-li...
Definition TFile.h:53
RLogChannel & NTupleLog()
Log channel for RNTuple diagnostics.
std::uint64_t NTupleSize_t
Integer type long enough to hold the maximum number of entries in a column.
std::uint64_t DescriptorId_t
Distriniguishes elements of the same type within a descriptor, e.g. different fields.
The incremental changes to a RNTupleModel