overlay/Slot.h

#pragma once
 
#include <xrpld/core/Config.h>
#include <xrpld/overlay/Peer.h>
#include <xrpld/overlay/ReduceRelayCommon.h>
 
#include <xrpl/basics/Log.h>
#include <xrpl/basics/chrono.h>
#include <xrpl/basics/random.h>
#include <xrpl/beast/container/aged_unordered_map.h>
#include <xrpl/beast/utility/Journal.h>
#include <xrpl/protocol/PublicKey.h>
#include <xrpl/protocol/messages.h>
 
#include <algorithm>
#include <optional>
#include <set>
#include <tuple>
#include <unordered_map>
#include <unordered_set>
 
namespace xrpl {
 
namespace reduce_relay {
 
template <typename clock_type>
class Slots;
 
enum class PeerState : uint8_t {
    Counting,   // counting messages
    Selected,   // selected to relay, counting if Slot in Counting
    Squelched,  // squelched, doesn't relay
};
enum class SlotState : uint8_t {
    Counting,  // counting messages
    Selected,  // peers selected, stop counting
};
 
template <typename Unit, typename TP>
Unit
epoch(TP const& t)
{
    return std::chrono::duration_cast<Unit>(t.time_since_epoch());
}
 
class SquelchHandler
{
public:
    virtual ~SquelchHandler()
    {
    }
    virtual void
    squelch(PublicKey const& validator, Peer::id_t id, std::uint32_t duration) const = 0;
    virtual void
    unsquelch(PublicKey const& validator, Peer::id_t id) const = 0;
};
 
template <typename clock_type>
class Slot final
{
private:
    friend class Slots<clock_type>;
    using id_t = Peer::id_t;
    using time_point = typename clock_type::time_point;
 
    // a callback to report ignored squelches
    using ignored_squelch_callback = std::function<void()>;
 
    Slot(SquelchHandler const& handler, beast::Journal journal, uint16_t maxSelectedPeers)
        : reachedThreshold_(0)
        , lastSelected_(clock_type::now())
        , state_(SlotState::Counting)
        , handler_(handler)
        , journal_(journal)
        , maxSelectedPeers_(maxSelectedPeers)
    {
    }
 
    void
    update(PublicKey const& validator, id_t id, protocol::MessageType type, ignored_squelch_callback callback);
 
    void
    deletePeer(PublicKey const& validator, id_t id, bool erase);
 
    time_point const&
    getLastSelected() const
    {
        return lastSelected_;
    }
 
    std::uint16_t
    inState(PeerState state) const;
 
    std::uint16_t
    notInState(PeerState state) const;
 
    SlotState
    getState() const
    {
        return state_;
    }
 
    std::set<id_t>
    getSelected() const;
 
    std::unordered_map<id_t, std::tuple<PeerState, uint16_t, uint32_t, uint32_t>>
    getPeers() const;
 
    void
    deleteIdlePeer(PublicKey const& validator);
 
    std::chrono::seconds
    getSquelchDuration(std::size_t npeers);
 
private:
    void
    resetCounts();
 
    void
    initCounting();
 
    struct PeerInfo
    {
        PeerState state;         // peer's state
        std::size_t count;       // message count
        time_point expire;       // squelch expiration time
        time_point lastMessage;  // time last message received
    };
 
    std::unordered_map<id_t, PeerInfo> peers_;  // peer's data
 
    // pool of peers considered as the source of messages
    // from validator - peers that reached MIN_MESSAGE_THRESHOLD
    std::unordered_set<id_t> considered_;
 
    // number of peers that reached MAX_MESSAGE_THRESHOLD
    std::uint16_t reachedThreshold_;
 
    // last time peers were selected, used to age the slot
    typename clock_type::time_point lastSelected_;
 
    SlotState state_;                // slot's state
    SquelchHandler const& handler_;  // squelch/unsquelch handler
    beast::Journal const journal_;   // logging
 
    // the maximum number of peers that should be selected as a validator
    // message source
    uint16_t const maxSelectedPeers_;
};
 
template <typename clock_type>
void
Slot<clock_type>::deleteIdlePeer(PublicKey const& validator)
{
    using namespace std::chrono;
    auto now = clock_type::now();
    for (auto it = peers_.begin(); it != peers_.end();)
    {
        auto& peer = it->second;
        auto id = it->first;
        ++it;
        if (now - peer.lastMessage > IDLED)
        {
            JLOG(journal_.trace()) << "deleteIdlePeer: " << Slice(validator) << " " << id << " idled "
                                   << duration_cast<seconds>(now - peer.lastMessage).count() << " selected "
                                   << (peer.state == PeerState::Selected);
            deletePeer(validator, id, false);
        }
    }
}
 
template <typename clock_type>
void
Slot<clock_type>::update(
    PublicKey const& validator,
    id_t id,
    protocol::MessageType type,
    ignored_squelch_callback callback)
{
    using namespace std::chrono;
    auto now = clock_type::now();
    auto it = peers_.find(id);
    // First message from this peer
    if (it == peers_.end())
    {
        JLOG(journal_.trace()) << "update: adding peer " << Slice(validator) << " " << id;
        peers_.emplace(std::make_pair(id, PeerInfo{PeerState::Counting, 0, now, now}));
        initCounting();
        return;
    }
    // Message from a peer with expired squelch
    if (it->second.state == PeerState::Squelched && now > it->second.expire)
    {
        JLOG(journal_.trace()) << "update: squelch expired " << Slice(validator) << " " << id;
        it->second.state = PeerState::Counting;
        it->second.lastMessage = now;
        initCounting();
        return;
    }
 
    auto& peer = it->second;
 
    JLOG(journal_.trace()) << "update: existing peer " << Slice(validator) << " " << id << " slot state "
                           << static_cast<int>(state_) << " peer state " << static_cast<int>(peer.state) << " count "
                           << peer.count << " last " << duration_cast<milliseconds>(now - peer.lastMessage).count()
                           << " pool " << considered_.size() << " threshold " << reachedThreshold_ << " "
                           << (type == protocol::mtVALIDATION ? "validation" : "proposal");
 
    peer.lastMessage = now;
 
    // report if we received a message from a squelched peer
    if (peer.state == PeerState::Squelched)
        callback();
 
    if (state_ != SlotState::Counting || peer.state == PeerState::Squelched)
        return;
 
    if (++peer.count > MIN_MESSAGE_THRESHOLD)
        considered_.insert(id);
    if (peer.count == (MAX_MESSAGE_THRESHOLD + 1))
        ++reachedThreshold_;
 
    if (now - lastSelected_ > 2 * MAX_UNSQUELCH_EXPIRE_DEFAULT)
    {
        JLOG(journal_.trace()) << "update: resetting due to inactivity " << Slice(validator) << " " << id << " "
                               << duration_cast<seconds>(now - lastSelected_).count();
        initCounting();
        return;
    }
 
    if (reachedThreshold_ == maxSelectedPeers_)
    {
        // Randomly select maxSelectedPeers_ peers from considered.
        // Exclude peers that have been idling > IDLED -
        // it's possible that deleteIdlePeer() has not been called yet.
        // If number of remaining peers != maxSelectedPeers_
        // then reset the Counting state and let deleteIdlePeer() handle
        // idled peers.
        std::unordered_set<id_t> selected;
        auto const consideredPoolSize = considered_.size();
        while (selected.size() != maxSelectedPeers_ && considered_.size() != 0)
        {
            auto i = considered_.size() == 1 ? 0 : rand_int(considered_.size() - 1);
            auto it = std::next(considered_.begin(), i);
            auto id = *it;
            considered_.erase(it);
            auto const& itPeers = peers_.find(id);
            if (itPeers == peers_.end())
            {
                JLOG(journal_.error()) << "update: peer not found " << Slice(validator) << " " << id;
                continue;
            }
            if (now - itPeers->second.lastMessage < IDLED)
                selected.insert(id);
        }
 
        if (selected.size() != maxSelectedPeers_)
        {
            JLOG(journal_.trace()) << "update: selection failed " << Slice(validator) << " " << id;
            initCounting();
            return;
        }
 
        lastSelected_ = now;
 
        auto s = selected.begin();
        JLOG(journal_.trace()) << "update: " << Slice(validator) << " " << id << " pool size " << consideredPoolSize
                               << " selected " << *s << " " << *std::next(s, 1) << " " << *std::next(s, 2);
 
        XRPL_ASSERT(peers_.size() >= maxSelectedPeers_, "xrpl::reduce_relay::Slot::update : minimum peers");
 
        // squelch peers which are not selected and
        // not already squelched
        std::stringstream str;
        for (auto& [k, v] : peers_)
        {
            v.count = 0;
 
            if (selected.find(k) != selected.end())
                v.state = PeerState::Selected;
            else if (v.state != PeerState::Squelched)
            {
                if (journal_.trace())
                    str << k << " ";
                v.state = PeerState::Squelched;
                std::chrono::seconds duration = getSquelchDuration(peers_.size() - maxSelectedPeers_);
                v.expire = now + duration;
                handler_.squelch(validator, k, duration.count());
            }
        }
        JLOG(journal_.trace()) << "update: squelching " << Slice(validator) << " " << id << " " << str.str();
        considered_.clear();
        reachedThreshold_ = 0;
        state_ = SlotState::Selected;
    }
}
 
template <typename clock_type>
std::chrono::seconds
Slot<clock_type>::getSquelchDuration(std::size_t npeers)
{
    using namespace std::chrono;
    auto m = std::max(MAX_UNSQUELCH_EXPIRE_DEFAULT, seconds{SQUELCH_PER_PEER * npeers});
    if (m > MAX_UNSQUELCH_EXPIRE_PEERS)
    {
        m = MAX_UNSQUELCH_EXPIRE_PEERS;
        JLOG(journal_.warn()) << "getSquelchDuration: unexpected squelch duration " << npeers;
    }
    return seconds{xrpl::rand_int(MIN_UNSQUELCH_EXPIRE / 1s, m / 1s)};
}
 
template <typename clock_type>
void
Slot<clock_type>::deletePeer(PublicKey const& validator, id_t id, bool erase)
{
    auto it = peers_.find(id);
    if (it != peers_.end())
    {
        std::vector<Peer::id_t> toUnsquelch;
 
        JLOG(journal_.trace()) << "deletePeer: " << Slice(validator) << " " << id << " selected "
                               << (it->second.state == PeerState::Selected) << " considered "
                               << (considered_.find(id) != considered_.end()) << " erase " << erase;
        auto now = clock_type::now();
        if (it->second.state == PeerState::Selected)
        {
            for (auto& [k, v] : peers_)
            {
                if (v.state == PeerState::Squelched)
                    toUnsquelch.push_back(k);
                v.state = PeerState::Counting;
                v.count = 0;
                v.expire = now;
            }
 
            considered_.clear();
            reachedThreshold_ = 0;
            state_ = SlotState::Counting;
        }
        else if (considered_.find(id) != considered_.end())
        {
            if (it->second.count > MAX_MESSAGE_THRESHOLD)
                --reachedThreshold_;
            considered_.erase(id);
        }
 
        it->second.lastMessage = now;
        it->second.count = 0;
 
        if (erase)
            peers_.erase(it);
 
        // Must be after peers_.erase(it)
        for (auto const& k : toUnsquelch)
            handler_.unsquelch(validator, k);
    }
}
 
template <typename clock_type>
void
Slot<clock_type>::resetCounts()
{
    for (auto& [_, peer] : peers_)
    {
        (void)_;
        peer.count = 0;
    }
}
 
template <typename clock_type>
void
Slot<clock_type>::initCounting()
{
    state_ = SlotState::Counting;
    considered_.clear();
    reachedThreshold_ = 0;
    resetCounts();
}
 
template <typename clock_type>
std::uint16_t
Slot<clock_type>::inState(PeerState state) const
{
    return std::count_if(peers_.begin(), peers_.end(), [&](auto const& it) { return (it.second.state == state); });
}
 
template <typename clock_type>
std::uint16_t
Slot<clock_type>::notInState(PeerState state) const
{
    return std::count_if(peers_.begin(), peers_.end(), [&](auto const& it) { return (it.second.state != state); });
}
 
template <typename clock_type>
std::set<typename Peer::id_t>
Slot<clock_type>::getSelected() const
{
    std::set<id_t> r;
    for (auto const& [id, info] : peers_)
        if (info.state == PeerState::Selected)
            r.insert(id);
    return r;
}
 
template <typename clock_type>
std::unordered_map<typename Peer::id_t, std::tuple<PeerState, uint16_t, uint32_t, uint32_t>>
Slot<clock_type>::getPeers() const
{
    using namespace std::chrono;
    auto r = std::unordered_map<id_t, std::tuple<PeerState, std::uint16_t, std::uint32_t, std::uint32_t>>();
 
    for (auto const& [id, info] : peers_)
        r.emplace(std::make_pair(
            id,
            std::move(std::make_tuple(
                info.state,
                info.count,
                epoch<milliseconds>(info.expire).count(),
                epoch<milliseconds>(info.lastMessage).count()))));
 
    return r;
}
 
template <typename clock_type>
class Slots final
{
    using time_point = typename clock_type::time_point;
    using id_t = typename Peer::id_t;
    using messages =
        beast::aged_unordered_map<uint256, std::unordered_set<Peer::id_t>, clock_type, hardened_hash<strong_hash>>;
 
public:
    Slots(Logs& logs, SquelchHandler const& handler, Config const& config)
        : handler_(handler)
        , logs_(logs)
        , journal_(logs.journal("Slots"))
        , baseSquelchEnabled_(config.VP_REDUCE_RELAY_BASE_SQUELCH_ENABLE)
        , maxSelectedPeers_(config.VP_REDUCE_RELAY_SQUELCH_MAX_SELECTED_PEERS)
    {
    }
    ~Slots() = default;
 
    bool
    baseSquelchReady()
    {
        return baseSquelchEnabled_ && reduceRelayReady();
    }
 
    bool
    reduceRelayReady()
    {
        if (!reduceRelayReady_)
            reduceRelayReady_ =
                reduce_relay::epoch<std::chrono::minutes>(clock_type::now()) > reduce_relay::WAIT_ON_BOOTUP;
 
        return reduceRelayReady_;
    }
 
    void
    updateSlotAndSquelch(uint256 const& key, PublicKey const& validator, id_t id, protocol::MessageType type)
    {
        updateSlotAndSquelch(key, validator, id, type, []() {});
    }
 
    void
    updateSlotAndSquelch(
        uint256 const& key,
        PublicKey const& validator,
        id_t id,
        protocol::MessageType type,
        typename Slot<clock_type>::ignored_squelch_callback callback);
 
    void
    deleteIdlePeers();
 
    std::optional<std::uint16_t>
    inState(PublicKey const& validator, PeerState state) const
    {
        auto const& it = slots_.find(validator);
        if (it != slots_.end())
            return it->second.inState(state);
        return {};
    }
 
    std::optional<std::uint16_t>
    notInState(PublicKey const& validator, PeerState state) const
    {
        auto const& it = slots_.find(validator);
        if (it != slots_.end())
            return it->second.notInState(state);
        return {};
    }
 
    bool
    inState(PublicKey const& validator, SlotState state) const
    {
        auto const& it = slots_.find(validator);
        if (it != slots_.end())
            return it->second.state_ == state;
        return false;
    }
 
    std::set<id_t>
    getSelected(PublicKey const& validator)
    {
        auto const& it = slots_.find(validator);
        if (it != slots_.end())
            return it->second.getSelected();
        return {};
    }
 
    std::unordered_map<typename Peer::id_t, std::tuple<PeerState, uint16_t, uint32_t, std::uint32_t>>
    getPeers(PublicKey const& validator)
    {
        auto const& it = slots_.find(validator);
        if (it != slots_.end())
            return it->second.getPeers();
        return {};
    }
 
    std::optional<SlotState>
    getState(PublicKey const& validator)
    {
        auto const& it = slots_.find(validator);
        if (it != slots_.end())
            return it->second.getState();
        return {};
    }
 
    void
    deletePeer(id_t id, bool erase);
 
private:
    bool
    addPeerMessage(uint256 const& key, id_t id);
 
    std::atomic_bool reduceRelayReady_{false};
 
    hash_map<PublicKey, Slot<clock_type>> slots_;
    SquelchHandler const& handler_;  // squelch/unsquelch handler
    Logs& logs_;
    beast::Journal const journal_;
 
    bool const baseSquelchEnabled_;
    uint16_t const maxSelectedPeers_;
 
    // Maintain aged container of message/peers. This is required
    // to discard duplicate message from the same peer. A message
    // is aged after IDLED seconds. A message received IDLED seconds
    // after it was relayed is ignored by PeerImp.
    inline static messages peersWithMessage_{beast::get_abstract_clock<clock_type>()};
};
 
template <typename clock_type>
bool
Slots<clock_type>::addPeerMessage(uint256 const& key, id_t id)
{
    beast::expire(peersWithMessage_, reduce_relay::IDLED);
 
    if (key.isNonZero())
    {
        auto it = peersWithMessage_.find(key);
        if (it == peersWithMessage_.end())
        {
            JLOG(journal_.trace()) << "addPeerMessage: new " << to_string(key) << " " << id;
            peersWithMessage_.emplace(key, std::unordered_set<id_t>{id});
            return true;
        }
 
        if (it->second.find(id) != it->second.end())
        {
            JLOG(journal_.trace()) << "addPeerMessage: duplicate message " << to_string(key) << " " << id;
            return false;
        }
 
        JLOG(journal_.trace()) << "addPeerMessage: added " << to_string(key) << " " << id;
 
        it->second.insert(id);
    }
 
    return true;
}
 
template <typename clock_type>
void
Slots<clock_type>::updateSlotAndSquelch(
    uint256 const& key,
    PublicKey const& validator,
    id_t id,
    protocol::MessageType type,
    typename Slot<clock_type>::ignored_squelch_callback callback)
{
    if (!addPeerMessage(key, id))
        return;
 
    auto it = slots_.find(validator);
    if (it == slots_.end())
    {
        JLOG(journal_.trace()) << "updateSlotAndSquelch: new slot " << Slice(validator);
        auto it = slots_
                      .emplace(std::make_pair(
                          validator, Slot<clock_type>(handler_, logs_.journal("Slot"), maxSelectedPeers_)))
                      .first;
        it->second.update(validator, id, type, callback);
    }
    else
        it->second.update(validator, id, type, callback);
}
 
template <typename clock_type>
void
Slots<clock_type>::deletePeer(id_t id, bool erase)
{
    for (auto& [validator, slot] : slots_)
        slot.deletePeer(validator, id, erase);
}
 
template <typename clock_type>
void
Slots<clock_type>::deleteIdlePeers()
{
    auto now = clock_type::now();
 
    for (auto it = slots_.begin(); it != slots_.end();)
    {
        it->second.deleteIdlePeer(it->first);
        if (now - it->second.getLastSelected() > MAX_UNSQUELCH_EXPIRE_DEFAULT)
        {
            JLOG(journal_.trace()) << "deleteIdlePeers: deleting idle slot " << Slice(it->first);
            it = slots_.erase(it);
        }
        else
            ++it;
    }
}
 
}  // namespace reduce_relay
 
}  // namespace xrpl