CacheLoader.hpp

#pragma once
 
#include "data/BackendInterface.hpp"
#include "etl/ETLHelpers.hpp"
#include "etl/impl/BaseCursorProvider.hpp"
#include "util/async/AnyExecutionContext.hpp"
#include "util/async/AnyOperation.hpp"
#include "util/log/Logger.hpp"
 
#include <boost/algorithm/string/predicate.hpp>
#include <boost/context/detail/config.hpp>
#include <fmt/format.h>
#include <xrpl/basics/Blob.h>
#include <xrpl/basics/base_uint.h>
#include <xrpl/basics/strHex.h>
 
#include <algorithm>
#include <atomic>
#include <chrono>
#include <cstddef>
#include <cstdint>
#include <exception>
#include <functional>
#include <memory>
#include <optional>
#include <ranges>
#include <string>
#include <utility>
#include <vector>
 
struct CacheLoaderImplTests;
 
namespace etl::impl {
 
template <typename CacheType>
class CacheLoaderImpl {
    util::Logger log_{"ETL"};
 
    util::async::AnyExecutionContext ctx_;
    std::shared_ptr<BackendInterface> backend_;
    std::reference_wrapper<CacheType> cache_;
 
    ThreadSafeQueue<CursorPair> queue_;
    std::atomic_int16_t remaining_;
 
    std::chrono::steady_clock::time_point startTime_ = std::chrono::steady_clock::now();
    std::vector<util::async::AnyOperation<void>> tasks_;
 
public:
    template <typename CtxType>
    CacheLoaderImpl(
        CtxType& ctx,
        std::shared_ptr<BackendInterface> backend,
        CacheType& cache,
        uint32_t const seq,
        std::size_t const numCacheMarkers,
        std::size_t const cachePageFetchSize,
        std::vector<CursorPair> const& cursors
    )
        : ctx_{ctx}
        , backend_{std::move(backend)}
        , cache_{std::ref(cache)}
        , queue_{cursors.size()}
        , remaining_{cursors.size()}
    {
        std::ranges::for_each(cursors, [this](auto const& cursor) { queue_.push(cursor); });
        load(seq, numCacheMarkers, cachePageFetchSize);
    }
 
    ~CacheLoaderImpl()
    {
        stop();
        wait();
    }
 
    void
    stop() noexcept
    {
        for (auto& t : tasks_)
            t.abort();
    }
 
    void
    wait() noexcept
    {
        for (auto& t : tasks_)
            t.wait();
    }
 
private:
    void
    load(uint32_t const seq, size_t numCacheMarkers, size_t cachePageFetchSize)
    {
        namespace vs = std::views;
 
        LOG(log_.info()) << "Loading cache. Num cursors = " << queue_.size();
        tasks_.reserve(numCacheMarkers);
 
        for ([[maybe_unused]] auto taskId : vs::iota(0u, numCacheMarkers))
            tasks_.push_back(spawnWorker(seq, cachePageFetchSize));
    }
 
    [[nodiscard]] auto
    spawnWorker(uint32_t const seq, size_t cachePageFetchSize)
    {
        return ctx_.execute([this, seq, cachePageFetchSize](auto token) {
            runGuarded([this, seq, cachePageFetchSize, token] {
                loadCacheFromCursors(token, seq, cachePageFetchSize);
            });
        });
    }
 
    template <typename Work>
    void
    runGuarded(Work&& work)
    {
        std::optional<std::string> failure;
        try {
            std::forward<Work>(work)();
        } catch (std::exception const& e) {
            failure = fmt::format("Cache loading failed: {}", e.what());
        } catch (...) {
            failure = "Cache loading failed with an unknown (non-std) error";
        }
 
        if (failure.has_value()) {
            LOG(log_.error()) << *failure
                              << "; disabling cache and continuing without it (reads will be "
                                 "served from the database).";
            cache_.get().setDisabled();
        }
    }
 
    template <typename TokenType>
    void
    loadCacheFromCursors(TokenType token, uint32_t const seq, size_t cachePageFetchSize)
    {
        while (not token.isStopRequested() and not cache_.get().isDisabled()) {
            auto cursor = queue_.tryPop();
            if (not cursor.has_value()) {
                return;  // queue is empty
            }
 
            auto [start, end] = *cursor;
            LOG(log_.debug()) << "Starting a cursor: " << xrpl::strHex(start);
 
            while (not token.isStopRequested() and not cache_.get().isDisabled()) {
                auto res = data::retryOnTimeout([this, seq, cachePageFetchSize, &start, token]() {
                    return backend_->fetchLedgerPage(start, seq, cachePageFetchSize, false, token);
                });
 
                cache_.get().update(res.objects, seq, true);
 
                if (not res.cursor or res.cursor > end) {
                    if (--remaining_ <= 0) {
                        auto endTime = std::chrono::steady_clock::now();
                        auto duration =
                            std::chrono::duration_cast<std::chrono::seconds>(endTime - startTime_);
 
                        LOG(log_.info())
                            << "Finished loading cache. Cache size = " << cache_.get().size()
                            << ". Took " << duration.count() << " seconds";
 
                        cache_.get().setFull();
                    } else {
                        LOG(log_.debug()) << "Finished a cursor. Remaining = " << remaining_;
                    }
 
                    break;  // pick up the next cursor if available
                }
 
                start = *std::move(res.cursor);
            }
        }
    }
 
    // Grants tests access to the private guard/loading members above.
    friend struct ::CacheLoaderImplTests;
};
 
}  // namespace etl::impl