SkipListAcquire.cpp

#include <xrpld/app/ledger/InboundLedger.h>
#include <xrpld/app/ledger/LedgerReplayer.h>
#include <xrpld/app/ledger/detail/SkipListAcquire.h>
#include <xrpld/app/main/Application.h>
#include <xrpld/overlay/PeerSet.h>
 
namespace xrpl {
 
SkipListAcquire::SkipListAcquire(
    Application& app,
    InboundLedgers& inboundLedgers,
    uint256 const& ledgerHash,
    std::unique_ptr<PeerSet> peerSet)
    : TimeoutCounter(
          app,
          ledgerHash,
          LedgerReplayParameters::SUB_TASK_TIMEOUT,
          {jtREPLAY_TASK, "SkipListAcq", LedgerReplayParameters::MAX_QUEUED_TASKS},
          app.journal("LedgerReplaySkipList"))
    , inboundLedgers_(inboundLedgers)
    , peerSet_(std::move(peerSet))
{
    JLOG(journal_.trace()) << "Create " << hash_;
}
 
SkipListAcquire::~SkipListAcquire()
{
    JLOG(journal_.trace()) << "Destroy " << hash_;
}
 
void
SkipListAcquire::init(int numPeers)
{
    ScopedLockType sl(mtx_);
    if (!isDone())
    {
        trigger(numPeers, sl);
        setTimer(sl);
    }
}
 
void
SkipListAcquire::trigger(std::size_t limit, ScopedLockType& sl)
{
    if (auto const l = app_.getLedgerMaster().getLedgerByHash(hash_); l)
    {
        JLOG(journal_.trace()) << "existing ledger " << hash_;
        retrieveSkipList(l, sl);
        return;
    }
 
    if (!fallBack_)
    {
        peerSet_->addPeers(
            limit,
            [this](auto peer) {
                return peer->supportsFeature(ProtocolFeature::LedgerReplay) && peer->hasLedger(hash_, 0);
            },
            [this](auto peer) {
                if (peer->supportsFeature(ProtocolFeature::LedgerReplay))
                {
                    JLOG(journal_.trace()) << "Add a peer " << peer->id() << " for " << hash_;
                    protocol::TMProofPathRequest request;
                    request.set_ledgerhash(hash_.data(), hash_.size());
                    request.set_key(keylet::skip().key.data(), keylet::skip().key.size());
                    request.set_type(protocol::TMLedgerMapType::lmACCOUNT_STATE);
                    peerSet_->sendRequest(request, peer);
                }
                else
                {
                    JLOG(journal_.trace()) << "Add a no feature peer " << peer->id() << " for " << hash_;
                    if (++noFeaturePeerCount_ >= LedgerReplayParameters::MAX_NO_FEATURE_PEER_COUNT)
                    {
                        JLOG(journal_.debug()) << "Fall back for " << hash_;
                        timerInterval_ = LedgerReplayParameters::SUB_TASK_FALLBACK_TIMEOUT;
                        fallBack_ = true;
                    }
                }
            });
    }
 
    if (fallBack_)
        inboundLedgers_.acquire(hash_, 0, InboundLedger::Reason::GENERIC);
}
 
void
SkipListAcquire::onTimer(bool progress, ScopedLockType& sl)
{
    JLOG(journal_.trace()) << "mTimeouts=" << timeouts_ << " for " << hash_;
    if (timeouts_ > LedgerReplayParameters::SUB_TASK_MAX_TIMEOUTS)
    {
        failed_ = true;
        JLOG(journal_.debug()) << "too many timeouts " << hash_;
        notify(sl);
    }
    else
    {
        trigger(1, sl);
    }
}
 
std::weak_ptr<TimeoutCounter>
SkipListAcquire::pmDowncast()
{
    return shared_from_this();
}
 
void
SkipListAcquire::processData(std::uint32_t ledgerSeq, boost::intrusive_ptr<SHAMapItem const> const& item)
{
    XRPL_ASSERT(ledgerSeq != 0 && item, "xrpl::SkipListAcquire::processData : valid inputs");
    ScopedLockType sl(mtx_);
    if (isDone())
        return;
 
    JLOG(journal_.trace()) << "got data for " << hash_;
    try
    {
        if (auto sle = std::make_shared<SLE>(SerialIter{item->slice()}, item->key()); sle)
        {
            if (auto const& skipList = sle->getFieldV256(sfHashes).value(); !skipList.empty())
                onSkipListAcquired(skipList, ledgerSeq, sl);
            return;
        }
    }
    catch (...)
    {
    }
 
    failed_ = true;
    JLOG(journal_.error()) << "failed to retrieve Skip list from verified data " << hash_;
    notify(sl);
}
 
void
SkipListAcquire::addDataCallback(OnSkipListDataCB&& cb)
{
    ScopedLockType sl(mtx_);
    dataReadyCallbacks_.emplace_back(std::move(cb));
    if (isDone())
    {
        JLOG(journal_.debug()) << "task added to a finished SkipListAcquire " << hash_;
        notify(sl);
    }
}
 
std::shared_ptr<SkipListAcquire::SkipListData const>
SkipListAcquire::getData() const
{
    ScopedLockType sl(mtx_);
    return data_;
}
 
void
SkipListAcquire::retrieveSkipList(std::shared_ptr<Ledger const> const& ledger, ScopedLockType& sl)
{
    if (auto const hashIndex = ledger->read(keylet::skip()); hashIndex && hashIndex->isFieldPresent(sfHashes))
    {
        auto const& slist = hashIndex->getFieldV256(sfHashes).value();
        if (!slist.empty())
        {
            onSkipListAcquired(slist, ledger->seq(), sl);
            return;
        }
    }
 
    failed_ = true;
    JLOG(journal_.error()) << "failed to retrieve Skip list from a ledger " << hash_;
    notify(sl);
}
 
void
SkipListAcquire::onSkipListAcquired(std::vector<uint256> const& skipList, std::uint32_t ledgerSeq, ScopedLockType& sl)
{
    complete_ = true;
    data_ = std::make_shared<SkipListData>(ledgerSeq, skipList);
    JLOG(journal_.debug()) << "Skip list acquired " << hash_;
    notify(sl);
}
 
void
SkipListAcquire::notify(ScopedLockType& sl)
{
    XRPL_ASSERT(isDone(), "xrpl::SkipListAcquire::notify : is done");
    std::vector<OnSkipListDataCB> toCall;
    std::swap(toCall, dataReadyCallbacks_);
    auto const good = !failed_;
    sl.unlock();
 
    for (auto& cb : toCall)
    {
        cb(good, hash_);
    }
 
    sl.lock();
}
 
}  // namespace xrpl