SHAMapStoreImp.h

//------------------------------------------------------------------------------
/*
    This file is part of rippled: https://github.com/ripple/rippled
    Copyright (c) 2012, 2013 Ripple Labs Inc.
 
    Permission to use, copy, modify, and/or distribute this software for any
    purpose  with  or without fee is hereby granted, provided that the above
    copyright notice and this permission notice appear in all copies.
 
    THE  SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
    WITH  REGARD  TO  THIS  SOFTWARE  INCLUDING  ALL  IMPLIED  WARRANTIES  OF
    MERCHANTABILITY  AND  FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
    ANY  SPECIAL ,  DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
    WHATSOEVER  RESULTING  FROM  LOSS  OF USE, DATA OR PROFITS, WHETHER IN AN
    ACTION  OF  CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
    OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
//==============================================================================
 
#ifndef RIPPLE_APP_MISC_SHAMAPSTOREIMP_H_INCLUDED
#define RIPPLE_APP_MISC_SHAMAPSTOREIMP_H_INCLUDED
 
#include <ripple/app/ledger/LedgerMaster.h>
#include <ripple/app/misc/SHAMapStore.h>
#include <ripple/core/DatabaseCon.h>
#include <ripple/nodestore/DatabaseRotating.h>
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <thread>
 
namespace ripple {
 
class NetworkOPs;
 
class SHAMapStoreImp : public SHAMapStore
{
private:
    struct SavedState
    {
        std::string writableDb;
        std::string archiveDb;
        LedgerIndex lastRotated;
    };
 
    enum Health : std::uint8_t { ok = 0, stopping, unhealthy };
 
    class SavedStateDB
    {
    public:
        soci::session session_;
        std::mutex mutex_;
        beast::Journal const journal_;
 
        // Just instantiate without any logic in case online delete is not
        // configured
        explicit SavedStateDB() : journal_{beast::Journal::getNullSink()}
        {
        }
 
        // opens database and, if necessary, creates & initializes its tables.
        void
        init(BasicConfig const& config, std::string const& dbName);
        // get/set the ledger index that we can delete up to and including
        LedgerIndex
        getCanDelete();
        LedgerIndex
        setCanDelete(LedgerIndex canDelete);
        SavedState
        getState();
        void
        setState(SavedState const& state);
        void
        setLastRotated(LedgerIndex seq);
    };
 
    Application& app_;
 
    // name of state database
    std::string const dbName_ = "state";
    // prefix of on-disk nodestore backend instances
    std::string const dbPrefix_ = "rippledb";
    // check health/stop status as records are copied
    std::uint64_t const checkHealthInterval_ = 1000;
    // minimum # of ledgers to maintain for health of network
    static std::uint32_t const minimumDeletionInterval_ = 256;
    // minimum # of ledgers required for standalone mode.
    static std::uint32_t const minimumDeletionIntervalSA_ = 8;
    // minimum ledger to maintain online.
    std::atomic<LedgerIndex> minimumOnline_{};
 
    NodeStore::Scheduler& scheduler_;
    beast::Journal const journal_;
    NodeStore::DatabaseRotating* dbRotating_ = nullptr;
    SavedStateDB state_db_;
    std::thread thread_;
    bool stop_ = false;
    bool healthy_ = true;
    mutable std::condition_variable cond_;
    mutable std::condition_variable rendezvous_;
    mutable std::mutex mutex_;
    std::shared_ptr<Ledger const> newLedger_;
    std::atomic<bool> working_;
    std::atomic<LedgerIndex> canDelete_;
    int fdRequired_ = 0;
 
    std::uint32_t deleteInterval_ = 0;
    bool advisoryDelete_ = false;
    std::uint32_t deleteBatch_ = 100;
    std::chrono::milliseconds backOff_{100};
    std::chrono::seconds ageThreshold_{60};
    boost::optional<std::chrono::seconds> recoveryWaitTime_;
 
    // these do not exist upon SHAMapStore creation, but do exist
    // as of onPrepare() or before
    NetworkOPs* netOPs_ = nullptr;
    LedgerMaster* ledgerMaster_ = nullptr;
    FullBelowCache* fullBelowCache_ = nullptr;
    TreeNodeCache* treeNodeCache_ = nullptr;
    DatabaseCon* transactionDb_ = nullptr;
    DatabaseCon* ledgerDb_ = nullptr;
 
    static constexpr auto nodeStoreName_ = "NodeStore";
 
public:
    SHAMapStoreImp(
        Application& app,
        Stoppable& parent,
        NodeStore::Scheduler& scheduler,
        beast::Journal journal);
 
    ~SHAMapStoreImp()
    {
        if (thread_.joinable())
            thread_.join();
    }
 
    std::uint32_t
    clampFetchDepth(std::uint32_t fetch_depth) const override
    {
        return deleteInterval_ ? std::min(fetch_depth, deleteInterval_)
                               : fetch_depth;
    }
 
    std::unique_ptr<NodeStore::Database>
    makeNodeStore(std::string const& name, std::int32_t readThreads) override;
 
    LedgerIndex
    setCanDelete(LedgerIndex seq) override
    {
        if (advisoryDelete_)
            canDelete_ = seq;
        return state_db_.setCanDelete(seq);
    }
 
    bool
    advisoryDelete() const override
    {
        return advisoryDelete_;
    }
 
    // All ledgers prior to this one are eligible
    // for deletion in the next rotation
    LedgerIndex
    getLastRotated() override
    {
        return state_db_.getState().lastRotated;
    }
 
    // All ledgers before and including this are unprotected
    // and online delete may delete them if appropriate
    LedgerIndex
    getCanDelete() override
    {
        return canDelete_;
    }
 
    void
    onLedgerClosed(std::shared_ptr<Ledger const> const& ledger) override;
 
    void
    rendezvous() const override;
    int
    fdRequired() const override;
 
    boost::optional<LedgerIndex>
    minimumOnline() const override;
 
private:
    // callback for visitNodes
    bool
    copyNode(std::uint64_t& nodeCount, SHAMapAbstractNode const& node);
    void
    run();
    void
    dbPaths();
 
    std::unique_ptr<NodeStore::Backend>
    makeBackendRotating(std::string path = std::string());
 
    template <class CacheInstance>
    bool
    freshenCache(CacheInstance& cache)
    {
        std::uint64_t check = 0;
 
        for (auto const& key : cache.getKeys())
        {
            dbRotating_->fetch(key, 0);
            if (!(++check % checkHealthInterval_) && health())
                return true;
        }
 
        return false;
    }
 
    void
    clearSql(
        DatabaseCon& database,
        LedgerIndex lastRotated,
        std::string const& minQuery,
        std::string const& deleteQuery);
    void
    clearCaches(LedgerIndex validatedSeq);
    void
    freshenCaches();
    void
    clearPrior(LedgerIndex lastRotated);
 
    // If rippled is not healthy, defer rotate-delete.
    // If already unhealthy, do not change state on further check.
    // Assume that, once unhealthy, a necessary step has been
    // aborted, so the online-delete process needs to restart
    // at next ledger.
    // If recoveryWaitTime_ is set, this may sleep to give rippled
    // time to recover, so never call it from any thread other than
    // the main "run()".
    Health
    health();
    //
    // Stoppable
    //
    void
    onPrepare() override
    {
    }
 
    void
    onStart() override
    {
        if (deleteInterval_)
            thread_ = std::thread(&SHAMapStoreImp::run, this);
    }
 
    // Called when the application begins shutdown
    void
    onStop() override;
    // Called when all child Stoppable objects have stoped
    void
    onChildrenStopped() override;
};
 
}  // namespace ripple
 
#endif