LedgerReplayer.h

#pragma once
 
#include <xrpld/app/ledger/LedgerMaster.h>
#include <xrpld/app/ledger/LedgerReplayTask.h>
#include <xrpld/app/main/Application.h>
 
#include <xrpl/beast/utility/Journal.h>
 
#include <mutex>
#include <vector>
 
namespace xrpl {
 
namespace test {
class LedgerReplayClient;
}  // namespace test
 
namespace LedgerReplayParameters {
// timeout value for LedgerReplayTask
auto constexpr TASK_TIMEOUT = std::chrono::milliseconds{500};
 
// for LedgerReplayTask to calculate max allowed timeouts
// = max( TASK_MAX_TIMEOUTS_MINIMUM,
//        (# of ledger to replay) * TASK_MAX_TIMEOUTS_MULTIPLIER)
std::uint32_t constexpr TASK_MAX_TIMEOUTS_MULTIPLIER = 2;
std::uint32_t constexpr TASK_MAX_TIMEOUTS_MINIMUM = 10;
 
// timeout value for subtasks: LedgerDeltaAcquire and SkipListAcquire
auto constexpr SUB_TASK_TIMEOUT = std::chrono::milliseconds{250};
// max of allowed subtask timeouts
std::uint32_t constexpr SUB_TASK_MAX_TIMEOUTS = 10;
 
// max number of peers that do not support the ledger replay feature
// returned by the PeerSet before switch to fallback
auto constexpr MAX_NO_FEATURE_PEER_COUNT = 2;
// subtask timeout value after fallback
auto constexpr SUB_TASK_FALLBACK_TIMEOUT = std::chrono::milliseconds{1000};
 
// for LedgerReplayer to limit the number of LedgerReplayTask
std::uint32_t constexpr MAX_TASKS = 10;
 
// for LedgerReplayer to limit the number of ledgers to replay in one task
std::uint32_t constexpr MAX_TASK_SIZE = 256;
 
// to limit the number of LedgerReplay related jobs in JobQueue
std::uint32_t constexpr MAX_QUEUED_TASKS = 100;
}  // namespace LedgerReplayParameters
 
class LedgerReplayer final
{
public:
    LedgerReplayer(Application& app, InboundLedgers& inboundLedgers, std::unique_ptr<PeerSetBuilder> peerSetBuilder);
 
    ~LedgerReplayer();
 
    void
    replay(InboundLedger::Reason r, uint256 const& finishLedgerHash, std::uint32_t totalNumLedgers);
 
    void
    createDeltas(std::shared_ptr<LedgerReplayTask> task);
 
    void
    gotSkipList(LedgerHeader const& info, boost::intrusive_ptr<SHAMapItem const> const& data);
 
    void
    gotReplayDelta(LedgerHeader const& info, std::map<std::uint32_t, std::shared_ptr<STTx const>>&& txns);
 
    void
    sweep();
 
    void
    stop();
 
    std::size_t
    tasksSize() const
    {
        std::lock_guard<std::mutex> lock(mtx_);
        return tasks_.size();
    }
 
    std::size_t
    deltasSize() const
    {
        std::lock_guard<std::mutex> lock(mtx_);
        return deltas_.size();
    }
 
    std::size_t
    skipListsSize() const
    {
        std::lock_guard<std::mutex> lock(mtx_);
        return skipLists_.size();
    }
 
private:
    mutable std::mutex mtx_;
    std::vector<std::shared_ptr<LedgerReplayTask>> tasks_;
    hash_map<uint256, std::weak_ptr<LedgerDeltaAcquire>> deltas_;
    hash_map<uint256, std::weak_ptr<SkipListAcquire>> skipLists_;
 
    Application& app_;
    InboundLedgers& inboundLedgers_;
    std::unique_ptr<PeerSetBuilder> peerSetBuilder_;
    beast::Journal j_;
 
    friend class test::LedgerReplayClient;
};
 
}  // namespace xrpl