57class RPageSynchronizingSink :
public RPageSink {
60 RPageSink *fInnerSink;
64 explicit RPageSynchronizingSink(RPageSink &
inner, std::mutex &mutex)
65 : RPageSink(
inner.GetNTupleName(),
inner.GetWriteOptions()), fInnerSink(&
inner), fMutex(&mutex)
71 RPageSynchronizingSink(
const RPageSynchronizingSink &) =
delete;
72 RPageSynchronizingSink &
operator=(
const RPageSynchronizingSink &) =
delete;
74 const RNTupleDescriptor &GetDescriptor() const
final {
return fInnerSink->GetDescriptor(); }
76 ColumnHandle_t AddColumn(DescriptorId_t, RColumn &)
final {
return {}; }
77 void InitImpl(RNTupleModel &)
final {}
78 void UpdateSchema(
const RNTupleModelChangeset &, NTupleSize_t)
final
80 throw RException(
R__FAIL(
"UpdateSchema not supported via RPageSynchronizingSink"));
82 void UpdateExtraTypeInfo(
const RExtraTypeInfoDescriptor &)
final
84 throw RException(
R__FAIL(
"UpdateExtraTypeInfo not supported via RPageSynchronizingSink"));
87 void CommitSuppressedColumn(ColumnHandle_t handle)
final { fInnerSink->CommitSuppressedColumn(handle); }
88 void CommitPage(ColumnHandle_t,
const RPage &)
final
90 throw RException(
R__FAIL(
"should never commit single pages via RPageSynchronizingSink"));
92 void CommitSealedPage(DescriptorId_t,
const RSealedPage &)
final
94 throw RException(
R__FAIL(
"should never commit sealed pages via RPageSynchronizingSink"));
96 void CommitSealedPageV(std::span<RPageStorage::RSealedPageGroup> ranges)
final
98 fInnerSink->CommitSealedPageV(ranges);
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
105 throw RException(
R__FAIL(
"should never commit cluster group via RPageSynchronizingSink"));
107 void CommitDatasetImpl()
final
109 throw RException(
R__FAIL(
"should never commit dataset via RPageSynchronizingSink"));
112 RSinkGuard GetSinkGuard()
final {
return RSinkGuard(fMutex); }
118 std::unique_ptr<Internal::RPageSink>
sink)
119 : fSink(std::
move(
sink)), fModel(std::
move(model)), fMetrics(
"RNTupleParallelWriter")
121 if (
fModel->GetRegisteredSubfields().size() > 0) {
122 throw RException(
R__FAIL(
"cannot create an RNTupleWriter from a model with registered subfields"));
131 for (
const auto &context : fFillContexts) {
132 if (!context.expired()) {
140 fSink->CommitClusterGroup();
141 fSink->CommitDataset();
147std::unique_ptr<ROOT::Experimental::RNTupleParallelWriter>
160std::unique_ptr<ROOT::Experimental::RNTupleParallelWriter>
175 std::lock_guard
g(fMutex);
177 auto model = fModel->Clone();
181 auto sink = std::make_unique<Internal::RPageSinkBuf>(std::make_unique<RPageSynchronizingSink>(*fSink, fSinkMutex));
186 fFillContexts.push_back(context);
#define R__FAIL(msg)
Short-hand to return an RResult<T> in an error state; the RError is implicitly converted into RResult...
#define R__LOG_ERROR(...)
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 ...
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.
std::string GetReport() const
Format a dignostics report, e.g. for an exception message.
Base class for all ROOT issued exceptions.
const RError & GetError() const
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)
static std::unique_ptr< RNTupleParallelWriter > Append(std::unique_ptr< RNTupleModel > model, std::string_view ntupleName, TDirectory &fileOrDirectory, const RNTupleWriteOptions &options=RNTupleWriteOptions())
Append an ntuple to the existing file, which must not be accessed while data is filled into any creat...
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.
Detail::RNTupleMetrics fMetrics
Common user-tunable settings for storing ntuples.
bool GetUseBufferedWrite() const
Describe directory structure in memory.
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