root/ROOT/RPageSinkBuf.hxx

0001 /// \file ROOT/RPageSinkBuf.hxx
0002 /// \ingroup NTuple
0003 /// \author Jakob Blomer <jblomer@cern.ch>
0004 /// \author Max Orok <maxwellorok@gmail.com>
0005 /// \author Javier Lopez-Gomez <javier.lopez.gomez@cern.ch>
0006 /// \date 2021-03-17
0007
0008 /*************************************************************************
0009  * Copyright (C) 1995-2021, Rene Brun and Fons Rademakers.               *
0010  * All rights reserved.                                                  *
0011  *                                                                       *
0012  * For the licensing terms see $ROOTSYS/LICENSE.                         *
0013  * For the list of contributors see $ROOTSYS/README/CREDITS.             *
0014  *************************************************************************/
0015
0016 #ifndef ROOT_RPageSinkBuf
0017 #define ROOT_RPageSinkBuf
0018
0019 #include <ROOT/RNTupleMetrics.hxx>
0020 #include <ROOT/RPageStorage.hxx>
0021
0022 #include <atomic>
0023 #include <cstddef>
0024 #include <deque>
0025 #include <functional>
0026 #include <iterator>
0027 #include <memory>
0028 #include <tuple>
0029
0030 namespace ROOT {
0031 namespace Internal {
0032
0033 // clang-format off
0034 /**
0035 \class ROOT::Internal::RPageSinkBuf
0036 \ingroup NTuple
0037 \brief Wrapper sink that coalesces cluster column page writes
0038 */
0039 // clang-format on
0040 class RPageSinkBuf : public RPageSink {
0041 private:
0042    /// A buffered column. The column is not responsible for RPage memory management (i.e. ReservePage),
0043    /// which is handled by the enclosing RPageSinkBuf.
0044    class RColumnBuf {
0045    public:
0046       struct RPageZipItem {
0047          RPage fPage;
0048          // Compression scratch buffer for fSealedPage.
0049          std::unique_ptr<unsigned char[]> fBuf;
0050          RPageStorage::RSealedPage *fSealedPage = nullptr;
0051          bool IsSealed() const { return fSealedPage != nullptr; }
0052       };
0053    public:
0054       RColumnBuf() = default;
0055       RColumnBuf(const RColumnBuf&) = delete;
0056       RColumnBuf& operator=(const RColumnBuf&) = delete;
0057       RColumnBuf(RColumnBuf&&) = default;
0058       RColumnBuf& operator=(RColumnBuf&&) = default;
0059       ~RColumnBuf() { DropBufferedPages(); }
0060
0061       /// Returns a reference to the newly buffered page. The reference remains
0062       /// valid until DropBufferedPages().
0063       RPageZipItem &BufferPage(RPageStorage::ColumnHandle_t columnHandle)
0064       {
0065          if (!fCol) {
0066             fCol = columnHandle;
0067          }
0068          // Safety: Insertion at the end of a deque never invalidates references
0069          // to existing elements.
0070          return fBufferedPages.emplace_back();
0071       }
0072       const RPageStorage::ColumnHandle_t &GetHandle() const { return fCol; }
0073       bool IsEmpty() const { return fBufferedPages.empty(); }
0074       bool HasSealedPagesOnly() const { return fBufferedPages.size() == fSealedPages.size(); }
0075       const RPageStorage::SealedPageSequence_t &GetSealedPages() const { return fSealedPages; }
0076
0077       void DropBufferedPages();
0078
0079       // The returned reference points to a default-constructed RSealedPage. It can be used
0080       // to fill in data after sealing.
0081       RSealedPage &RegisterSealedPage()
0082       {
0083          return fSealedPages.emplace_back();
0084       }
0085
0086    private:
0087       RPageStorage::ColumnHandle_t fCol;
0088       /// Using a deque guarantees that element iterators are never invalidated
0089       /// by appends to the end of the iterator by BufferPage.
0090       std::deque<RPageZipItem> fBufferedPages;
0091       /// Pages that have been already sealed by a concurrent task. A vector commit can be issued if all
0092       /// buffered pages have been sealed.
0093       /// Note that each RSealedPage refers to the same buffer as `fBufferedPages[i].fBuf` for some value of `i`, and
0094       /// thus owned by RPageZipItem
0095       RPageStorage::SealedPageSequence_t fSealedPages;
0096    };
0097
0098 private:
0099    /// I/O performance counters that get registered in fMetrics
0100    struct RCounters {
0101       ROOT::Experimental::Detail::RNTuplePlainCounter &fParallelZip;
0102       ROOT::Experimental::Detail::RNTupleAtomicCounter &fTimeWallZip;
0103       ROOT::Experimental::Detail::RNTuplePlainCounter &fTimeWallCriticalSection;
0104       ROOT::Experimental::Detail::RNTupleTickCounter<ROOT::Experimental::Detail::RNTupleAtomicCounter> &fTimeCpuZip;
0105       ROOT::Experimental::Detail::RNTupleTickCounter<ROOT::Experimental::Detail::RNTuplePlainCounter>
0106          &fTimeCpuCriticalSection;
0107    };
0108    std::unique_ptr<RCounters> fCounters;
0109    /// The inner sink, responsible for actually performing I/O.
0110    std::unique_ptr<RPageSink> fInnerSink;
0111    /// The buffered page sink maintains a copy of the RNTupleModel for the inner sink.
0112    /// For the unbuffered case, the RNTupleModel is instead managed by a RNTupleWriter.
0113    std::unique_ptr<ROOT::RNTupleModel> fInnerModel;
0114    /// The sum of uncompressed bytes in buffered pages. Used to heuristically reduce the memory usage.
0115    std::atomic<std::size_t> fBufferedUncompressed = 0;
0116    /// Vector of buffered column pages. Indexed by column id.
0117    std::vector<RColumnBuf> fBufferedColumns;
0118    /// Columns committed as suppressed are stored and passed to the inner sink at cluster commit
0119    std::vector<ColumnHandle_t> fSuppressedColumns;
0120    ROOT::DescriptorId_t fNFields = 0;
0121    ROOT::DescriptorId_t fNColumns = 0;
0122
0123    void ConnectFields(const std::vector<ROOT::RFieldBase *> &fields, ROOT::NTupleSize_t firstEntry);
0124    void FlushClusterImpl(std::function<void(void)> FlushClusterFn);
0125
0126 public:
0127    explicit RPageSinkBuf(std::unique_ptr<RPageSink> inner);
0128    RPageSinkBuf(const RPageSinkBuf&) = delete;
0129    RPageSinkBuf& operator=(const RPageSinkBuf&) = delete;
0130    RPageSinkBuf(RPageSinkBuf &&) = delete;
0131    RPageSinkBuf &operator=(RPageSinkBuf &&) = delete;
0132    ~RPageSinkBuf() override;
0133
0134    ColumnHandle_t AddColumn(ROOT::DescriptorId_t fieldId, RColumn &column) final;
0135
0136    const ROOT::RNTupleDescriptor &GetDescriptor() const final;
0137
0138    ROOT::NTupleSize_t GetNEntries() const final { return fInnerSink->GetNEntries(); }
0139
0140    void InitImpl(ROOT::RNTupleModel &model) final;
0141    void UpdateSchema(const RNTupleModelChangeset &changeset, ROOT::NTupleSize_t firstEntry) final;
0142    void UpdateExtraTypeInfo(const ROOT::RExtraTypeInfoDescriptor &extraTypeInfo) final;
0143
0144    void CommitSuppressedColumn(ColumnHandle_t columnHandle) final;
0145    void CommitPage(ColumnHandle_t columnHandle, const RPage &page) final;
0146    void CommitSealedPage(ROOT::DescriptorId_t physicalColumnId, const RSealedPage &sealedPage) final;
0147    void CommitSealedPageV(std::span<RPageStorage::RSealedPageGroup> ranges) final;
0148    std::uint64_t CommitCluster(ROOT::NTupleSize_t nNewEntries) final;
0149    RStagedCluster StageCluster(ROOT::NTupleSize_t nNewEntries) final;
0150    void CommitStagedClusters(std::span<RStagedCluster> clusters) final;
0151    void CommitClusterGroup() final;
0152    void CommitDatasetImpl() final;
0153
0154    RPage ReservePage(ColumnHandle_t columnHandle, std::size_t nElements) final;
0155 }; // RPageSinkBuf
0156
0157 } // namespace Internal
0158 } // namespace ROOT
0159
0160 #endif