Livecache.h

#pragma once
 
#include <xrpld/peerfinder/PeerfinderManager.h>
#include <xrpld/peerfinder/detail/Tuning.h>
#include <xrpld/peerfinder/detail/iosformat.h>
 
#include <xrpl/basics/Log.h>
#include <xrpl/basics/random.h>
#include <xrpl/beast/container/aged_map.h>
#include <xrpl/beast/utility/maybe_const.h>
 
#include <boost/intrusive/list.hpp>
#include <boost/iterator/transform_iterator.hpp>
 
#include <algorithm>
 
namespace xrpl {
namespace PeerFinder {
 
template <class>
class Livecache;
 
namespace detail {
 
class LivecacheBase
{
public:
    explicit LivecacheBase() = default;
 
protected:
    struct Element : boost::intrusive::list_base_hook<>
    {
        Element(Endpoint const& endpoint_) : endpoint(endpoint_)
        {
        }
 
        Endpoint endpoint;
    };
 
    using list_type =
        boost::intrusive::make_list<Element, boost::intrusive::constant_time_size<false>>::type;
 
public:
    template <bool IsConst>
    class Hop
    {
    public:
        // Iterator transformation to extract the endpoint from Element
        struct Transform
        {
            using first_argument = Element;
            using result_type = Endpoint;
 
            explicit Transform() = default;
 
            Endpoint const&
            operator()(Element const& e) const
            {
                return e.endpoint;
            }
        };
 
    public:
        using iterator = boost::transform_iterator<Transform, typename list_type::const_iterator>;
 
        using const_iterator = iterator;
 
        using reverse_iterator =
            boost::transform_iterator<Transform, typename list_type::const_reverse_iterator>;
 
        using const_reverse_iterator = reverse_iterator;
 
        iterator
        begin() const
        {
            return iterator(m_list.get().cbegin(), Transform());
        }
 
        iterator
        cbegin() const
        {
            return iterator(m_list.get().cbegin(), Transform());
        }
 
        iterator
        end() const
        {
            return iterator(m_list.get().cend(), Transform());
        }
 
        iterator
        cend() const
        {
            return iterator(m_list.get().cend(), Transform());
        }
 
        reverse_iterator
        rbegin() const
        {
            return reverse_iterator(m_list.get().crbegin(), Transform());
        }
 
        reverse_iterator
        crbegin() const
        {
            return reverse_iterator(m_list.get().crbegin(), Transform());
        }
 
        reverse_iterator
        rend() const
        {
            return reverse_iterator(m_list.get().crend(), Transform());
        }
 
        reverse_iterator
        crend() const
        {
            return reverse_iterator(m_list.get().crend(), Transform());
        }
 
        // move the element to the end of the container
        void
        move_back(const_iterator pos)
        {
            auto& e(const_cast<Element&>(*pos.base()));
            m_list.get().erase(m_list.get().iterator_to(e));
            m_list.get().push_back(e);
        }
 
    private:
        explicit Hop(typename beast::maybe_const<IsConst, list_type>::type& list) : m_list(list)
        {
        }
 
        friend class LivecacheBase;
 
        std::reference_wrapper<typename beast::maybe_const<IsConst, list_type>::type> m_list;
    };
 
protected:
    // Work-around to call Hop's private constructor from Livecache
    template <bool IsConst>
    static Hop<IsConst>
    make_hop(typename beast::maybe_const<IsConst, list_type>::type& list)
    {
        return Hop<IsConst>(list);
    }
};
 
}  // namespace detail
 
//------------------------------------------------------------------------------
 
template <class Allocator = std::allocator<char>>
class Livecache : protected detail::LivecacheBase
{
private:
    using cache_type = beast::aged_map<
        beast::IP::Endpoint,
        Element,
        std::chrono::steady_clock,
        std::less<beast::IP::Endpoint>,
        Allocator>;
 
    beast::Journal m_journal;
    cache_type m_cache;
 
public:
    using allocator_type = Allocator;
 
    Livecache(clock_type& clock, beast::Journal journal, Allocator alloc = Allocator());
 
    //
    // Iteration by hops
    //
    // The range [begin, end) provides a sequence of list_type
    // where each list contains endpoints at a given hops.
    //
 
    class hops_t
    {
    private:
        // An endpoint at hops=0 represents the local node.
        // Endpoints coming in at maxHops are stored at maxHops +1,
        // but not given out (since they would exceed maxHops). They
        // are used for automatic connection attempts.
        //
        using Histogram = std::array<int, 1 + Tuning::maxHops + 1>;
        using lists_type = std::array<list_type, 1 + Tuning::maxHops + 1>;
 
        template <bool IsConst>
        struct Transform
        {
            using first_argument = typename lists_type::value_type;
            using result_type = Hop<IsConst>;
 
            explicit Transform() = default;
 
            Hop<IsConst>
            operator()(typename beast::maybe_const<IsConst, typename lists_type::value_type>::type&
                           list) const
            {
                return make_hop<IsConst>(list);
            }
        };
 
    public:
        using iterator = boost::transform_iterator<Transform<false>, typename lists_type::iterator>;
 
        using const_iterator =
            boost::transform_iterator<Transform<true>, typename lists_type::const_iterator>;
 
        using reverse_iterator =
            boost::transform_iterator<Transform<false>, typename lists_type::reverse_iterator>;
 
        using const_reverse_iterator =
            boost::transform_iterator<Transform<true>, typename lists_type::const_reverse_iterator>;
 
        iterator
        begin()
        {
            return iterator(m_lists.begin(), Transform<false>());
        }
 
        const_iterator
        begin() const
        {
            return const_iterator(m_lists.cbegin(), Transform<true>());
        }
 
        const_iterator
        cbegin() const
        {
            return const_iterator(m_lists.cbegin(), Transform<true>());
        }
 
        iterator
        end()
        {
            return iterator(m_lists.end(), Transform<false>());
        }
 
        const_iterator
        end() const
        {
            return const_iterator(m_lists.cend(), Transform<true>());
        }
 
        const_iterator
        cend() const
        {
            return const_iterator(m_lists.cend(), Transform<true>());
        }
 
        reverse_iterator
        rbegin()
        {
            return reverse_iterator(m_lists.rbegin(), Transform<false>());
        }
 
        const_reverse_iterator
        rbegin() const
        {
            return const_reverse_iterator(m_lists.crbegin(), Transform<true>());
        }
 
        const_reverse_iterator
        crbegin() const
        {
            return const_reverse_iterator(m_lists.crbegin(), Transform<true>());
        }
 
        reverse_iterator
        rend()
        {
            return reverse_iterator(m_lists.rend(), Transform<false>());
        }
 
        const_reverse_iterator
        rend() const
        {
            return const_reverse_iterator(m_lists.crend(), Transform<true>());
        }
 
        const_reverse_iterator
        crend() const
        {
            return const_reverse_iterator(m_lists.crend(), Transform<true>());
        }
 
        void
        shuffle();
 
        std::string
        histogram() const;
 
    private:
        explicit hops_t(Allocator const& alloc);
 
        void
        insert(Element& e);
 
        // Reinsert e at a new hops
        void
        reinsert(Element& e, std::uint32_t hops);
 
        void
        remove(Element& e);
 
        friend class Livecache;
        lists_type m_lists;
        Histogram m_hist{};
    } hops;
 
    bool
    empty() const
    {
        return m_cache.empty();
    }
 
    typename cache_type::size_type
    size() const
    {
        return m_cache.size();
    }
 
    void
    expire();
 
    void
    insert(Endpoint const& ep);
 
    void
    onWrite(beast::PropertyStream::Map& map);
};
 
//------------------------------------------------------------------------------
 
template <class Allocator>
Livecache<Allocator>::Livecache(clock_type& clock, beast::Journal journal, Allocator alloc)
    : m_journal(journal), m_cache(clock, alloc), hops(alloc)
{
}
 
template <class Allocator>
void
Livecache<Allocator>::expire()
{
    std::size_t n(0);
    typename cache_type::time_point const expired(
        m_cache.clock().now() - Tuning::liveCacheSecondsToLive);
    for (auto iter(m_cache.chronological.begin());
         iter != m_cache.chronological.end() && iter.when() <= expired;)
    {
        Element& e(iter->second);
        hops.remove(e);
        iter = m_cache.erase(iter);
        ++n;
    }
    if (n > 0)
    {
        JLOG(m_journal.debug()) << beast::leftw(18) << "Livecache expired " << n
                                << ((n > 1) ? " entries" : " entry");
    }
}
 
template <class Allocator>
void
Livecache<Allocator>::insert(Endpoint const& ep)
{
    // The caller already incremented hop, so if we got a
    // message at maxHops we will store it at maxHops + 1.
    // This means we won't give out the address to other peers
    // but we will use it to make connections and hand it out
    // when redirecting.
    //
    XRPL_ASSERT(
        ep.hops <= (Tuning::maxHops + 1),
        "xrpl::PeerFinder::Livecache::insert : maximum input hops");
    auto result = m_cache.emplace(ep.address, ep);
    Element& e(result.first->second);
    if (result.second)
    {
        hops.insert(e);
        JLOG(m_journal.debug()) << beast::leftw(18) << "Livecache insert " << ep.address
                                << " at hops " << ep.hops;
        return;
    }
    else if (!result.second && (ep.hops > e.endpoint.hops))
    {
        // Drop duplicates at higher hops
        std::size_t const excess(ep.hops - e.endpoint.hops);
        JLOG(m_journal.trace()) << beast::leftw(18) << "Livecache drop " << ep.address
                                << " at hops +" << excess;
        return;
    }
 
    m_cache.touch(result.first);
 
    // Address already in the cache so update metadata
    if (ep.hops < e.endpoint.hops)
    {
        hops.reinsert(e, ep.hops);
        JLOG(m_journal.debug()) << beast::leftw(18) << "Livecache update " << ep.address
                                << " at hops " << ep.hops;
    }
    else
    {
        JLOG(m_journal.trace()) << beast::leftw(18) << "Livecache refresh " << ep.address
                                << " at hops " << ep.hops;
    }
}
 
template <class Allocator>
void
Livecache<Allocator>::onWrite(beast::PropertyStream::Map& map)
{
    typename cache_type::time_point const expired(
        m_cache.clock().now() - Tuning::liveCacheSecondsToLive);
    map["size"] = size();
    map["hist"] = hops.histogram();
    beast::PropertyStream::Set set("entries", map);
    for (auto iter(m_cache.cbegin()); iter != m_cache.cend(); ++iter)
    {
        auto const& e(iter->second);
        beast::PropertyStream::Map item(set);
        item["hops"] = e.endpoint.hops;
        item["address"] = e.endpoint.address.to_string();
        std::stringstream ss;
        ss << (iter.when() - expired).count();
        item["expires"] = ss.str();
    }
}
 
//------------------------------------------------------------------------------
 
template <class Allocator>
void
Livecache<Allocator>::hops_t::shuffle()
{
    for (auto& list : m_lists)
    {
        std::vector<std::reference_wrapper<Element>> v;
        v.reserve(list.size());
        std::copy(list.begin(), list.end(), std::back_inserter(v));
        std::shuffle(v.begin(), v.end(), default_prng());
        list.clear();
        for (auto& e : v)
            list.push_back(e);
    }
}
 
template <class Allocator>
std::string
Livecache<Allocator>::hops_t::histogram() const
{
    std::string s;
    for (auto const& h : m_hist)
    {
        if (!s.empty())
            s += ", ";
        s += std::to_string(h);
    }
    return s;
}
 
template <class Allocator>
Livecache<Allocator>::hops_t::hops_t(Allocator const& alloc)
{
    std::fill(m_hist.begin(), m_hist.end(), 0);
}
 
template <class Allocator>
void
Livecache<Allocator>::hops_t::insert(Element& e)
{
    XRPL_ASSERT(
        e.endpoint.hops <= Tuning::maxHops + 1,
        "xrpl::PeerFinder::Livecache::hops_t::insert : maximum input hops");
    // This has security implications without a shuffle
    m_lists[e.endpoint.hops].push_front(e);
    ++m_hist[e.endpoint.hops];
}
 
template <class Allocator>
void
Livecache<Allocator>::hops_t::reinsert(Element& e, std::uint32_t numHops)
{
    XRPL_ASSERT(
        numHops <= Tuning::maxHops + 1,
        "xrpl::PeerFinder::Livecache::hops_t::reinsert : maximum hops input");
 
    auto& list = m_lists[e.endpoint.hops];
    list.erase(list.iterator_to(e));
 
    --m_hist[e.endpoint.hops];
 
    e.endpoint.hops = numHops;
    insert(e);
}
 
template <class Allocator>
void
Livecache<Allocator>::hops_t::remove(Element& e)
{
    --m_hist[e.endpoint.hops];
 
    auto& list = m_lists[e.endpoint.hops];
    list.erase(list.iterator_to(e));
}
 
}  // namespace PeerFinder
}  // namespace xrpl