rippled/src/xrpld/app/ledger/detail/InboundLedgers.cpp

#include <xrpld/app/ledger/InboundLedgers.h>

#include <xrpld/app/ledger/AcquireStats.h>
#include <xrpld/app/ledger/InboundLedger.h>
#include <xrpld/app/ledger/LedgerMaster.h>
#include <xrpld/app/ledger/LedgerNodeHelpers.h>
#include <xrpld/app/main/Application.h>
#include <xrpld/overlay/PeerSet.h>

#include <xrpl/basics/Blob.h>
#include <xrpl/basics/DecayingSample.h>
#include <xrpl/basics/Log.h>
#include <xrpl/basics/UnorderedContainers.h>
#include <xrpl/basics/base_uint.h>
#include <xrpl/basics/scope.h>
#include <xrpl/beast/container/aged_map.h>
#include <xrpl/beast/container/detail/aged_ordered_container.h>
#include <xrpl/beast/insight/Collector.h>
#include <xrpl/beast/utility/instrumentation.h>
#include <xrpl/core/Job.h>
#include <xrpl/core/JobQueue.h>
#include <xrpl/core/PerfLog.h>
#include <xrpl/json/json_value.h>
#include <xrpl/protocol/RippleLedgerHash.h>
#include <xrpl/protocol/Serializer.h>
#include <xrpl/protocol/jss.h>
#include <xrpl/server/NetworkOPs.h>
#include <xrpl/shamap/SHAMapMissingNode.h>
#include <xrpl/shamap/SHAMapTreeNode.h>

#include <xrpl.pb.h>

#include <algorithm>
#include <chrono>
#include <cstddef>
#include <cstdint>
#include <exception>
#include <functional>
#include <memory>
#include <mutex>
#include <set>
#include <string>
#include <utility>
#include <vector>

namespace xrpl {

class InboundLedgersImp : public InboundLedgers
{
private:
    Application& app_;
    std::mutex fetchRateMutex_;
    // measures ledgers per second, constants are important
    DecayWindow<30, clock_type> fetchRate_;
    beast::Journal const j_;

public:
    // How long before we try again to acquire the same ledger
    static constexpr std::chrono::minutes kReacquireInterval{5};

    InboundLedgersImp(
        Application& app,
        clock_type& clock,
        beast::insight::Collector::ptr const& collector,
        std::unique_ptr<PeerSetBuilder> peerSetBuilder)
        : app_(app)
        , fetchRate_(clock.now())
        , j_(app.getJournal("InboundLedger"))
        , clock_(clock)
        , recentFailures_(clock)
        , counter_(collector->makeCounter("ledger_fetches"))
        , peerSetBuilder_(std::move(peerSetBuilder))
    {
    }

    /**
     * @callgraph
     */
    std::shared_ptr<Ledger const>
    acquire(uint256 const& hash, std::uint32_t seq, InboundLedger::Reason reason) override
    {
        auto doAcquire = [&, seq, reason]() -> std::shared_ptr<Ledger const> {
            XRPL_ASSERT(
                hash.isNonZero(), "xrpl::InboundLedgersImp::acquire::doAcquire : nonzero hash");

            // probably not the right rule
            if (app_.getOPs().isNeedNetworkLedger() && (reason != InboundLedger::Reason::GENERIC) &&
                (reason != InboundLedger::Reason::CONSENSUS))
                return {};

            bool isNew = true;
            std::shared_ptr<InboundLedger> inbound;
            {
                ScopedLockType sl(lock_);
                if (stopping_)
                {
                    return {};
                }

                auto it = ledgers_.find(hash);
                if (it != ledgers_.end())
                {
                    isNew = false;
                    inbound = it->second;
                }
                else
                {
                    inbound = std::make_shared<InboundLedger>(
                        app_, hash, seq, reason, std::ref(clock_), peerSetBuilder_->build());
                    ledgers_.emplace(hash, inbound);
                    inbound->init(sl);
                    ++counter_;
                }
            }

            if (inbound->isFailed())
                return {};

            if (!isNew)
                inbound->update(seq);

            if (!inbound->isComplete())
                return {};

            return inbound->getLedger();
        };
        using namespace std::chrono_literals;
        std::shared_ptr<Ledger const> ledger =
            perf::measureDurationAndLog(doAcquire, "InboundLedgersImp::acquire", 500ms, j_);

        return ledger;
    }

    void
    acquireAsync(uint256 const& hash, std::uint32_t seq, InboundLedger::Reason reason) override
    {
        std::unique_lock lock(acquiresMutex_);
        try
        {
            if (pendingAcquires_.contains(hash))
                return;
            pendingAcquires_.insert(hash);
            ScopeUnlock const unlock(lock);
            acquire(hash, seq, reason);
        }
        catch (std::exception const& e)
        {
            JLOG(j_.warn()) << "Exception thrown for acquiring new inbound ledger " << hash << ": "
                            << e.what();
        }
        catch (...)
        {
            JLOG(j_.warn()) << "Unknown exception thrown for acquiring new inbound ledger " << hash;
        }
        pendingAcquires_.erase(hash);
    }

    std::shared_ptr<InboundLedger>
    find(uint256 const& hash) override
    {
        XRPL_ASSERT(hash.isNonZero(), "xrpl::InboundLedgersImp::find : nonzero input");

        std::shared_ptr<InboundLedger> ret;

        {
            ScopedLockType const sl(lock_);

            auto it = ledgers_.find(hash);
            if (it != ledgers_.end())
            {
                ret = it->second;
            }
        }

        return ret;
    }

    /*
    This gets called when
        "We got some data from an inbound ledger"

    inboundLedgerTrigger:
      "What do we do with this partial data?"
      Figures out what to do with the responses to our requests for information.

    */
    // means "We got some data from an inbound ledger"

    // VFALCO TODO Remove the dependency on the Peer object.
    /**
     * We received a TMLedgerData from a peer.
     */
    bool
    gotLedgerData(
        LedgerHash const& hash,
        std::shared_ptr<Peer> peer,
        std::shared_ptr<protocol::TMLedgerData> packet) override
    {
        if (auto ledger = find(hash))
        {
            JLOG(j_.trace()) << "Got data (" << packet->nodes().size()
                             << ") for acquiring ledger: " << hash;

            // Stash the data for later processing and see if we need to
            // dispatch
            if (ledger->gotData(std::weak_ptr<Peer>(peer), packet))
            {
                app_.getJobQueue().addJob(
                    JtLedgerData, "ProcessLData", [ledger]() { ledger->runData(); });
            }

            return true;
        }

        JLOG(j_.trace()) << "Got data for ledger " << hash << " which we're no longer acquiring";

        // If it's state node data, stash it because it still might be
        // useful.
        if (packet->type() == protocol::liAS_NODE)
        {
            app_.getJobQueue().addJob(
                JtLedgerData, "GotStaleData", [this, packet]() { gotStaleData(packet); });
        }

        return false;
    }

    void
    logFailure(uint256 const& h, std::uint32_t seq) override
    {
        ScopedLockType const sl(lock_);

        recentFailures_.emplace(h, seq);
    }

    bool
    isFailure(uint256 const& h) override
    {
        ScopedLockType const sl(lock_);

        beast::expire(recentFailures_, kReacquireInterval);
        return recentFailures_.find(h) != recentFailures_.end();
    }

    /**
     * We got some data for a ledger we are no longer acquiring Since we paid
     * the price to receive it, we might as well stash it in case we need it.
     *
     * Nodes are received in wire format and must be stashed/hashed in prefix
     * format
     */
    void
    gotStaleData(std::shared_ptr<protocol::TMLedgerData> packetPtr) override
    {
        Serializer s;
        try
        {
            for (auto const& ledgerNode : packetPtr->nodes())
            {
                auto const treeNode = getTreeNode(ledgerNode.nodedata());
                if (!treeNode)
                    return;

                s.erase();
                treeNode->serializeWithPrefix(s);

                app_.getLedgerMaster().addFetchPack(
                    treeNode->getHash().asUInt256(), std::make_shared<Blob>(s.begin(), s.end()));
            }
        }
        catch (std::exception const&)  // NOLINT(bugprone-empty-catch)
        {
        }
    }

    void
    clearFailures() override
    {
        ScopedLockType const sl(lock_);

        recentFailures_.clear();
        ledgers_.clear();
    }

    std::size_t
    fetchRate() override
    {
        std::scoped_lock const lock(fetchRateMutex_);
        return 60 * fetchRate_.value(clock_.now());
    }

    // Should only be called with an inboundledger that has
    // a reason of history
    void
    onLedgerFetched() override
    {
        std::scoped_lock const lock(fetchRateMutex_);
        fetchRate_.add(1, clock_.now());
    }

    json::Value
    getInfo() override
    {
        json::Value ret(json::ValueType::Object);

        std::vector<std::pair<uint256, std::shared_ptr<InboundLedger>>> acqs;

        {
            ScopedLockType const sl(lock_);

            acqs.reserve(ledgers_.size());
            for (auto const& it : ledgers_)
            {
                XRPL_ASSERT(it.second, "xrpl::InboundLedgersImp::getInfo : non-null ledger");
                acqs.emplace_back(it);
            }
            for (auto const& it : recentFailures_)
            {
                if (it.second > 1)
                {
                    ret[std::to_string(it.second)][jss::failed] = true;
                }
                else
                {
                    ret[to_string(it.first)][jss::failed] = true;
                }
            }
        }

        for (auto const& it : acqs)
        {
            // getJson is expensive, so call without the lock
            std::uint32_t const seq = it.second->getSeq();
            if (seq > 1)
            {
                ret[std::to_string(seq)] = it.second->getJson(0);
            }
            else
            {
                ret[to_string(it.first)] = it.second->getJson(0);
            }
        }

        return ret;
    }

    void
    gotFetchPack() override
    {
        std::vector<std::shared_ptr<InboundLedger>> acquires;
        {
            ScopedLockType const sl(lock_);

            acquires.reserve(ledgers_.size());
            for (auto const& it : ledgers_)
            {
                XRPL_ASSERT(
                    it.second,
                    "xrpl::InboundLedgersImp::gotFetchPack : non-null "
                    "ledger");
                acquires.push_back(it.second);
            }
        }

        for (auto const& acquire : acquires)
        {
            acquire->checkLocal();
        }
    }

    void
    sweep() override
    {
        auto const start = clock_.now();

        // Make a list of things to sweep, while holding the lock
        std::vector<MapType::mapped_type> stuffToSweep;
        std::size_t total = 0;

        {
            ScopedLockType const sl(lock_);
            auto it = ledgers_.begin();
            total = ledgers_.size();

            stuffToSweep.reserve(total);

            while (it != ledgers_.end())
            {
                auto const la = it->second->getLastAction();

                if (la > start)
                {
                    it->second->touch();
                    ++it;
                }
                else if ((la + std::chrono::minutes(1)) < start)
                {
                    stuffToSweep.push_back(it->second);
                    // An eviction here discards whatever the acquisition had
                    // built, so the work restarts. Counted to tell that apart
                    // from an acquisition that ended on its own.
                    app_.getAcquireStats().recordSweepEviction();
                    // shouldn't cause the actual final delete
                    // since we are holding a reference in the vector.
                    it = ledgers_.erase(it);
                }
                else
                {
                    ++it;
                }
            }

            beast::expire(recentFailures_, kReacquireInterval);
        }

        JLOG(j_.debug())
            << "Swept " << stuffToSweep.size() << " out of " << total
            << " inbound ledgers. Duration: "
            << std::chrono::duration_cast<std::chrono::milliseconds>(clock_.now() - start).count()
            << "ms";
    }

    void
    stop() override
    {
        ScopedLockType const lock(lock_);
        stopping_ = true;
        ledgers_.clear();
        recentFailures_.clear();
    }

    std::size_t
    cacheSize() override
    {
        ScopedLockType const lock(lock_);
        return ledgers_.size();
    }

    AcquireProgress
    acquireProgress() override
    {
        // Copy the handles under the lock, then read each acquire's atomics
        // without it -- same pattern gotFetchPack() uses, so the ~10 s telemetry
        // poll never contends with the node-receive path.
        std::vector<std::shared_ptr<InboundLedger>> acquires;
        {
            ScopedLockType const sl(lock_);
            acquires.reserve(ledgers_.size());
            for (auto const& it : ledgers_)
            {
                XRPL_ASSERT(
                    it.second, "xrpl::InboundLedgersImp::acquireProgress : non-null ledger");
                acquires.push_back(it.second);
            }
        }

        AcquireProgress out;
        out.inFlight = acquires.size();
        for (auto const& acquire : acquires)
        {
            out.maxMissingStateNodes =
                std::max(out.maxMissingStateNodes, acquire->getMissingNodeCount(SHAMapType::STATE));
            out.maxMissingTxNodes = std::max(
                out.maxMissingTxNodes, acquire->getMissingNodeCount(SHAMapType::TRANSACTION));
            out.receivedDataDepth += acquire->getReceivedDataDepth();
        }
        return out;
    }

private:
    clock_type& clock_;

    using ScopedLockType = std::unique_lock<std::recursive_mutex>;
    std::recursive_mutex lock_;

    bool stopping_ = false;
    using MapType = hash_map<uint256, std::shared_ptr<InboundLedger>>;
    MapType ledgers_;

    beast::aged_map<uint256, std::uint32_t> recentFailures_;

    beast::insight::Counter counter_;

    std::unique_ptr<PeerSetBuilder> peerSetBuilder_;

    std::set<uint256> pendingAcquires_;
    std::mutex acquiresMutex_;
};

//------------------------------------------------------------------------------

std::unique_ptr<InboundLedgers>
makeInboundLedgers(
    Application& app,
    InboundLedgers::clock_type& clock,
    beast::insight::Collector::ptr const& collector)
{
    return std::make_unique<InboundLedgersImp>(app, clock, collector, makePeerSetBuilder(app));
}

}  // namespace xrpl