LedgerReplayer.h

#pragma once
 
#include <xrpld/app/ledger/InboundLedger.h>
#include <xrpld/app/ledger/LedgerReplayTask.h>
#include <xrpld/app/main/Application.h>
#include <xrpld/overlay/PeerSet.h>
 
#include <xrpl/basics/base_uint.h>
#include <xrpl/beast/utility/Journal.h>
#include <xrpl/protocol/LedgerHeader.h>
#include <xrpl/protocol/STTx.h>
#include <xrpl/shamap/SHAMapItem.h>
 
#include <cstddef>
#include <cstdint>
#include <map>
#include <memory>
#include <mutex>
#include <vector>
 
namespace xrpl {
 
namespace test {
class LedgerReplayClient;
}  // namespace test
 
namespace ledger_replay_parameters {
// timeout value for LedgerReplayTask
constexpr auto kTaskTimeout = std::chrono::milliseconds{500};
 
// for LedgerReplayTask to calculate max allowed timeouts
// = max( kTaskMaxTimeoutsMinimum,
//        (# of ledger to replay) * kTaskMaxTimeoutsMultiplier)
constexpr std::uint32_t kTaskMaxTimeoutsMultiplier = 2;
constexpr std::uint32_t kTaskMaxTimeoutsMinimum = 10;
 
// timeout value for subtasks: LedgerDeltaAcquire and SkipListAcquire
constexpr auto kSubTaskTimeout = std::chrono::milliseconds{250};
// max of allowed subtask timeouts
constexpr std::uint32_t kSubTaskMaxTimeouts = 10;
 
// max number of peers that do not support the ledger replay feature
// returned by the PeerSet before switch to fallback
constexpr auto kMaxNoFeaturePeerCount = 2;
// subtask timeout value after fallback
constexpr auto kSubTaskFallbackTimeout = std::chrono::milliseconds{1000};
 
// for LedgerReplayer to limit the number of LedgerReplayTask
constexpr std::uint32_t kMaxTasks = 10;
 
// for LedgerReplayer to limit the number of ledgers to replay in one task
constexpr std::uint32_t kMaxTaskSize = 256;
 
// to limit the number of LedgerReplay related jobs in JobQueue
constexpr std::uint32_t kMaxQueuedTasks = 100;
}  // namespace ledger_replay_parameters

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::scoped_lock const lock(mtx_);
        return tasks_.size();
    }
 
    std::size_t
    deltasSize() const
    {
        std::scoped_lock const lock(mtx_);
        return deltas_.size();
    }
 
    std::size_t
    skipListsSize() const
    {
        std::scoped_lock const lock(mtx_);
        return skipLists_.size();
    }
 
private:
    mutable std::mutex mtx_;
    std::vector<std::shared_ptr<LedgerReplayTask>> tasks_;
    HashMap<UInt256, std::weak_ptr<LedgerDeltaAcquire>> deltas_;
    HashMap<UInt256, std::weak_ptr<SkipListAcquire>> skipLists_;
 
    Application& app_;
    InboundLedgers& inboundLedgers_;
    std::unique_ptr<PeerSetBuilder> peerSetBuilder_;
    beast::Journal j_;
 
    friend class test::LedgerReplayClient;
};
 
}  // namespace xrpl