Reduce relaying of validation and proposal messages.

A server may receive multiple copies of a message from
its directly connected peers. We select three peers as
the source of validation and proposal messages from
the validator and "squelch" other peers.
This change reduces CPU and bandwidth costs and improves
link latency.

Author: gregtatcam

Builds/CMake/RippledCore.cmake

@@ -863,6 +863,7 @@ target_sources (rippled PRIVATE
   src/test/overlay/cluster_test.cpp
   src/test/overlay/short_read_test.cpp
   src/test/overlay/compression_test.cpp
+  src/test/overlay/reduce_relay_test.cpp
   #[===============================[
      test sources:
        subdir: peerfinder

src/ripple/app/consensus/RCLConsensus.cpp

@@ -157,7 +157,7 @@ RCLConsensus::Adaptor::share(RCLCxPeerPos const& peerPos)
     auto const sig = peerPos.signature();
     prop.set_signature(sig.data(), sig.size());
 
-    app_.overlay().relay(prop, peerPos.suppressionID());
+    app_.overlay().relay(prop, peerPos.suppressionID(), peerPos.publicKey());
 }
 
 void

src/ripple/app/misc/HashRouter.cpp

@@ -49,12 +49,19 @@ HashRouter::addSuppression(uint256 const& key)
 
 bool
 HashRouter::addSuppressionPeer(uint256 const& key, PeerShortID peer)
+{
+    auto [added, _] = addSuppressionPeerWithStatus(key, peer);
+    return added;
+}
+
+std::pair<bool, std::optional<Stopwatch::time_point>>
+HashRouter::addSuppressionPeerWithStatus(const uint256& key, PeerShortID peer)
 {
     std::lock_guard lock(mutex_);
 
     auto result = emplace(key);
     result.first.addPeer(peer);
-    return result.second;
+    return {result.second, result.first.relayed()};
 }
 
 bool
@@ -110,14 +117,14 @@ HashRouter::setFlags(uint256 const& key, int flags)
 
 auto
 HashRouter::shouldRelay(uint256 const& key)
-    -> boost::optional<std::set<PeerShortID>>
+    -> std::optional<std::set<PeerShortID>>
 {
     std::lock_guard lock(mutex_);
 
     auto& s = emplace(key).first;
 
     if (!s.shouldRelay(suppressionMap_.clock().now(), holdTime_))
-        return boost::none;
+        return {};
 
     return s.releasePeerSet();
 }

src/ripple/app/misc/HashRouter.h

@@ -97,6 +97,13 @@ class HashRouter
             return std::move(peers_);
         }
 
+        /** Return seated relay time point if the message has been relayed */
+        std::optional<Stopwatch::time_point>
+        relayed() const
+        {
+            return relayed_;
+        }
+
         /** Determines if this item should be relayed.
 
             Checks whether the item has been recently relayed.
@@ -142,8 +149,8 @@ class HashRouter
         std::set<PeerShortID> peers_;
         // This could be generalized to a map, if more
         // than one flag needs to expire independently.
-        boost::optional<Stopwatch::time_point> relayed_;
-        boost::optional<Stopwatch::time_point> processed_;
+        std::optional<Stopwatch::time_point> relayed_;
+        std::optional<Stopwatch::time_point> processed_;
         std::uint32_t recoveries_ = 0;
     };
 
@@ -185,6 +192,14 @@ class HashRouter
     bool
     addSuppressionPeer(uint256 const& key, PeerShortID peer);
 
+    /** Add a suppression peer and get message's relay status.
+     * Return pair:
+     * element 1: true if the peer is added.
+     * element 2: optional is seated to the relay time point or
+     * is unseated if has not relayed yet. */
+    std::pair<bool, std::optional<Stopwatch::time_point>>
+    addSuppressionPeerWithStatus(uint256 const& key, PeerShortID peer);
+
     bool
     addSuppressionPeer(uint256 const& key, PeerShortID peer, int& flags);
 
@@ -214,11 +229,11 @@ class HashRouter
             return `true` again until the hold time has expired.
             The internal set of peers will also be reset.
 
-        @return A `boost::optional` set of peers which do not need to be
+        @return A `std::optional` set of peers which do not need to be
             relayed to. If the result is uninitialized, the item should
             _not_ be relayed.
     */
-    boost::optional<std::set<PeerShortID>>
+    std::optional<std::set<PeerShortID>>
     shouldRelay(uint256 const& key);
 
     /** Determines whether the hashed item should be recovered

src/ripple/core/Config.h

@@ -179,6 +179,12 @@ class Config : public BasicConfig
     // Thread pool configuration
     std::size_t WORKERS = 0;
 
+    // Reduce-relay
+    // Enable reduce-relay functionality
+    bool REDUCE_RELAY_ENABLE = false;
+    // Send squelch message to peers
+    bool REDUCE_RELAY_SQUELCH = false;
+
     // These override the command line client settings
     boost::optional<beast::IP::Endpoint> rpc_ip;
 

src/ripple/core/ConfigSections.h

@@ -70,6 +70,7 @@ struct ConfigSection
 #define SECTION_PATH_SEARCH_MAX "path_search_max"
 #define SECTION_PEER_PRIVATE "peer_private"
 #define SECTION_PEERS_MAX "peers_max"
+#define SECTION_REDUCE_RELAY "reduce_relay"
 #define SECTION_RELAY_PROPOSALS "relay_proposals"
 #define SECTION_RELAY_VALIDATIONS "relay_validations"
 #define SECTION_RPC_STARTUP "rpc_startup"

src/ripple/core/impl/Config.cpp

@@ -481,6 +481,13 @@ Config::loadFromString(std::string const& fileContents)
     if (getSingleSection(secConfig, SECTION_COMPRESSION, strTemp, j_))
         COMPRESSION = beast::lexicalCastThrow<bool>(strTemp);
 
+    if (exists(SECTION_REDUCE_RELAY))
+    {
+        auto sec = section(SECTION_REDUCE_RELAY);
+        REDUCE_RELAY_ENABLE = sec.value_or("enable", false);
+        REDUCE_RELAY_SQUELCH = sec.value_or("squelch", false);
+    }
+
     if (getSingleSection(
             secConfig, SECTION_AMENDMENT_MAJORITY_TIME, strTemp, j_))
     {

src/ripple/overlay/Message.h

@@ -21,6 +21,7 @@
 #define RIPPLE_OVERLAY_MESSAGE_H_INCLUDED
 
 #include <ripple/overlay/Compression.h>
+#include <ripple/protocol/PublicKey.h>
 #include <ripple/protocol/messages.h>
 #include <boost/asio/buffer.hpp>
 #include <boost/asio/buffers_iterator.hpp>
@@ -55,8 +56,13 @@ class Message : public std::enable_shared_from_this<Message>
     /** Constructor
      * @param message Protocol message to serialize
      * @param type Protocol message type
+     * @param validator Public Key of the source validator for Validation or
+     * Proposal message. Used to check if the message should be squelched.
      */
-    Message(::google::protobuf::Message const& message, int type);
+    Message(
+        ::google::protobuf::Message const& message,
+        int type,
+        boost::optional<PublicKey> const& validator = {});
 
     /** Retrieve the packed message data. If compressed message is requested but
      * the message is not compressible then the uncompressed buffer is returned.
@@ -74,11 +80,19 @@ class Message : public std::enable_shared_from_this<Message>
         return category_;
     }
 
+    /** Get the validator's key */
+    boost::optional<PublicKey> const&
+    getValidatorKey() const
+    {
+        return validatorKey_;
+    }
+
 private:
     std::vector<uint8_t> buffer_;
     std::vector<uint8_t> bufferCompressed_;
     std::size_t category_;
     std::once_flag once_flag_;
+    boost::optional<PublicKey> validatorKey_;
 
     /** Set the payload header
      * @param in Pointer to the payload

src/ripple/overlay/Overlay.h

@@ -133,7 +133,7 @@ class Overlay : public Stoppable, public beast::PropertyStream::Source
 
     /** Returns the peer with the matching short id, or null. */
     virtual std::shared_ptr<Peer>
-    findPeerByShortID(Peer::id_t const& id) = 0;
+    findPeerByShortID(Peer::id_t const& id) const = 0;
 
     /** Returns the peer with the matching public key, or null. */
     virtual std::shared_ptr<Peer>
@@ -147,13 +147,27 @@ class Overlay : public Stoppable, public beast::PropertyStream::Source
     virtual void
     broadcast(protocol::TMValidation& m) = 0;
 
-    /** Relay a proposal. */
-    virtual void
-    relay(protocol::TMProposeSet& m, uint256 const& uid) = 0;
-
-    /** Relay a validation. */
-    virtual void
-    relay(protocol::TMValidation& m, uint256 const& uid) = 0;
+    /** Relay a proposal. Return set
+     * of peers, which have already seen the
+     * message; i.e. the message has been
+     * received from these peers and added
+     * to the hash router */
+    virtual std::set<Peer::id_t>
+    relay(
+        protocol::TMProposeSet& m,
+        uint256 const& uid,
+        PublicKey const& validator) = 0;
+
+    /** Relay a validation. Return set
+     * of peers, which have already seen the
+     * message; i.e. the message has been
+     * received from these peers and added
+     * to the hash router */
+    virtual std::set<Peer::id_t>
+    relay(
+        protocol::TMValidation& m,
+        uint256 const& uid,
+        PublicKey const& validator) = 0;
 
     /** Visit every active peer.
      *