Database.h

#ifndef XRPL_NODESTORE_DATABASE_H_INCLUDED
#define XRPL_NODESTORE_DATABASE_H_INCLUDED
 
#include <xrpl/basics/BasicConfig.h>
#include <xrpl/basics/Log.h>
#include <xrpl/basics/TaggedCache.ipp>
#include <xrpl/nodestore/Backend.h>
#include <xrpl/nodestore/NodeObject.h>
#include <xrpl/nodestore/Scheduler.h>
#include <xrpl/protocol/SystemParameters.h>
 
#include <condition_variable>
 
namespace ripple {
 
namespace NodeStore {
 
class Database
{
public:
    Database() = delete;
 
    Database(
        Scheduler& scheduler,
        int readThreads,
        Section const& config,
        beast::Journal j);
 
    virtual ~Database();
 
    virtual std::string
    getName() const = 0;
 
    virtual void
    importDatabase(Database& source) = 0;
 
    virtual std::int32_t
    getWriteLoad() const = 0;
 
    virtual void
    store(
        NodeObjectType type,
        Blob&& data,
        uint256 const& hash,
        std::uint32_t ledgerSeq) = 0;
 
    /* Check if two ledgers are in the same database
 
        If these two sequence numbers map to the same database,
        the result of a fetch with either sequence number would
        be identical.
 
        @param s1 The first sequence number
        @param s2 The second sequence number
 
        @return 'true' if both ledgers would be in the same DB
 
    */
    virtual bool
    isSameDB(std::uint32_t s1, std::uint32_t s2) = 0;
 
    virtual void
    sync() = 0;
 
    std::shared_ptr<NodeObject>
    fetchNodeObject(
        uint256 const& hash,
        std::uint32_t ledgerSeq = 0,
        FetchType fetchType = FetchType::synchronous,
        bool duplicate = false);
 
    virtual void
    asyncFetch(
        uint256 const& hash,
        std::uint32_t ledgerSeq,
        std::function<void(std::shared_ptr<NodeObject> const&)>&& callback);
 
    virtual void
    sweep() = 0;
 
    std::uint64_t
    getStoreCount() const
    {
        return storeCount_;
    }
 
    std::uint32_t
    getFetchTotalCount() const
    {
        return fetchTotalCount_;
    }
 
    std::uint32_t
    getFetchHitCount() const
    {
        return fetchHitCount_;
    }
 
    std::uint64_t
    getStoreSize() const
    {
        return storeSz_;
    }
 
    std::uint32_t
    getFetchSize() const
    {
        return fetchSz_;
    }
 
    void
    getCountsJson(Json::Value& obj);
 
    int
    fdRequired() const
    {
        return fdRequired_;
    }
 
    virtual void
    stop();
 
    bool
    isStopping() const;
 
    [[nodiscard]] std::uint32_t
    earliestLedgerSeq() const noexcept
    {
        return earliestLedgerSeq_;
    }
 
protected:
    beast::Journal const j_;
    Scheduler& scheduler_;
    int fdRequired_{0};
 
    std::atomic<std::uint32_t> fetchHitCount_{0};
    std::atomic<std::uint32_t> fetchSz_{0};
 
    // The default is XRP_LEDGER_EARLIEST_SEQ (32570) to match the XRP ledger
    // network's earliest allowed ledger sequence. Can be set through the
    // configuration file using the 'earliest_seq' field under the 'node_db'
    // stanza. If specified, the value must be greater than zero.
    // Only unit tests or alternate
    // networks should change this value.
    std::uint32_t const earliestLedgerSeq_;
 
    // The maximum number of requests a thread extracts from the queue in an
    // attempt to minimize the overhead of mutex acquisition. This is an
    // advanced tunable, via the config file. The default value is 4.
    int const requestBundle_;
 
    void
    storeStats(std::uint64_t count, std::uint64_t sz)
    {
        XRPL_ASSERT(
            count <= sz,
            "ripple::NodeStore::Database::storeStats : valid inputs");
        storeCount_ += count;
        storeSz_ += sz;
    }
 
    // Called by the public import function
    void
    importInternal(Backend& dstBackend, Database& srcDB);
 
    void
    updateFetchMetrics(uint64_t fetches, uint64_t hits, uint64_t duration)
    {
        fetchTotalCount_ += fetches;
        fetchHitCount_ += hits;
        fetchDurationUs_ += duration;
    }
 
private:
    std::atomic<std::uint64_t> storeCount_{0};
    std::atomic<std::uint64_t> storeSz_{0};
    std::atomic<std::uint64_t> fetchTotalCount_{0};
    std::atomic<std::uint64_t> fetchDurationUs_{0};
    std::atomic<std::uint64_t> storeDurationUs_{0};
 
    mutable std::mutex readLock_;
    std::condition_variable readCondVar_;
 
    // reads to do
    std::map<
        uint256,
        std::vector<std::pair<
            std::uint32_t,
            std::function<void(std::shared_ptr<NodeObject> const&)>>>>
        read_;
 
    std::atomic<bool> readStopping_ = false;
    std::atomic<int> readThreads_ = 0;
    std::atomic<int> runningThreads_ = 0;
 
    virtual std::shared_ptr<NodeObject>
    fetchNodeObject(
        uint256 const& hash,
        std::uint32_t ledgerSeq,
        FetchReport& fetchReport,
        bool duplicate) = 0;
 
    virtual void
    for_each(std::function<void(std::shared_ptr<NodeObject>)> f) = 0;
 
    void
    threadEntry();
};
 
}  // namespace NodeStore
}  // namespace ripple
 
#endif