mirror of
https://github.com/XRPLF/rippled.git
synced 2026-07-11 09:10:21 +00:00
Compare commits
38 Commits
ximinez/fi
...
ximinez/ac
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
be0baf1b53 | ||
|
|
b394730bd0 | ||
|
|
8290eb3024 | ||
|
|
d537684df5 | ||
|
|
4085361251 | ||
|
|
1de7c130b0 | ||
|
|
2a678a1157 | ||
|
|
5c254aadf2 | ||
|
|
ad2db1962e | ||
|
|
4dae405762 | ||
|
|
b21b3596f8 | ||
|
|
d648539b62 | ||
|
|
50644c6d30 | ||
|
|
88c248077e | ||
|
|
bad40d7e10 | ||
|
|
74d0c375ec | ||
|
|
200ab3f694 | ||
|
|
a7a4357022 | ||
|
|
e0230a1e22 | ||
|
|
5f1a98397c | ||
|
|
c8ca624974 | ||
|
|
4489cda841 | ||
|
|
784774005e | ||
|
|
904b39a3d3 | ||
|
|
eadfabfe4f | ||
|
|
d991f81e87 | ||
|
|
a34a96e233 | ||
|
|
a3a277518b | ||
|
|
8438698c54 | ||
|
|
0597047d79 | ||
|
|
992c6f525d | ||
|
|
9a870c36e2 | ||
|
|
a4462e65bc | ||
|
|
f0596c0a5d | ||
|
|
dd99bc3ce8 | ||
|
|
25cd7730e5 | ||
|
|
8dcbcb22c2 | ||
|
|
f44dfc89cc |
@@ -58,6 +58,11 @@ public:
|
||||
cleanup_();
|
||||
}
|
||||
|
||||
CanProcess(CanProcess const&) = delete;
|
||||
|
||||
CanProcess&
|
||||
operator=(CanProcess const&) = delete;
|
||||
|
||||
explicit
|
||||
operator bool() const
|
||||
{
|
||||
|
||||
@@ -44,7 +44,7 @@ template <class T>
|
||||
inline void
|
||||
maybeReverseBytes(T& t, std::true_type)
|
||||
{
|
||||
reverseBytes(t);
|
||||
reverse_bytes(t);
|
||||
}
|
||||
|
||||
template <class T, class Hasher>
|
||||
|
||||
@@ -145,13 +145,6 @@ private:
|
||||
return std::move(peers_);
|
||||
}
|
||||
|
||||
/** Return set of peers waiting for reply. Leaves list unchanged. */
|
||||
std::set<PeerShortID> const&
|
||||
peekPeerSet()
|
||||
{
|
||||
return peers_;
|
||||
}
|
||||
|
||||
/** Return seated relay time point if the message has been relayed */
|
||||
[[nodiscard]] std::optional<Stopwatch::time_point>
|
||||
relayed() const
|
||||
@@ -183,20 +176,6 @@ private:
|
||||
return true;
|
||||
}
|
||||
|
||||
bool
|
||||
shouldProcessForPeer(
|
||||
PeerShortID peer,
|
||||
Stopwatch::time_point now,
|
||||
std::chrono::seconds interval)
|
||||
{
|
||||
if (peerProcessed_.contains(peer) && ((peerProcessed_[peer] + interval) > now))
|
||||
return false;
|
||||
// Peer may already be in the list, but adding it again doesn't hurt
|
||||
addPeer(peer);
|
||||
peerProcessed_[peer] = now;
|
||||
return true;
|
||||
}
|
||||
|
||||
private:
|
||||
HashRouterFlags flags_ = HashRouterFlags::UNDEFINED;
|
||||
std::set<PeerShortID> peers_;
|
||||
@@ -204,7 +183,6 @@ private:
|
||||
// than one flag needs to expire independently.
|
||||
std::optional<Stopwatch::time_point> relayed_;
|
||||
std::optional<Stopwatch::time_point> processed_;
|
||||
std::map<PeerShortID, Stopwatch::time_point> peerProcessed_;
|
||||
};
|
||||
|
||||
public:
|
||||
@@ -244,15 +222,6 @@ public:
|
||||
HashRouterFlags& flags,
|
||||
std::chrono::seconds txInterval);
|
||||
|
||||
/** Determines whether the hashed item should be processed for the given
|
||||
peer. Could be an incoming or outgoing message.
|
||||
|
||||
Items filtered with this function should only be processed for the given
|
||||
peer once. Unlike shouldProcess, it can be processed for other peers.
|
||||
*/
|
||||
bool
|
||||
shouldProcessForPeer(uint256 const& key, PeerShortID peer, std::chrono::seconds interval);
|
||||
|
||||
/** Set the flags on a hash.
|
||||
|
||||
@return `true` if the flags were changed. `false` if unchanged.
|
||||
@@ -278,11 +247,6 @@ public:
|
||||
std::optional<std::set<PeerShortID>>
|
||||
shouldRelay(uint256 const& key);
|
||||
|
||||
/** Returns a copy of the set of peers in the Entry for the key
|
||||
*/
|
||||
std::set<PeerShortID>
|
||||
getPeers(uint256 const& key);
|
||||
|
||||
private:
|
||||
// pair.second indicates whether the entry was created
|
||||
std::pair<Entry&, bool>
|
||||
|
||||
@@ -288,18 +288,8 @@ message TMLedgerData {
|
||||
required uint32 ledgerSeq = 2;
|
||||
required TMLedgerInfoType type = 3;
|
||||
repeated TMLedgerNode nodes = 4;
|
||||
// If the peer supports "responseCookies", this field will
|
||||
// never be populated.
|
||||
optional uint32 requestCookie = 5;
|
||||
optional TMReplyError error = 6;
|
||||
// The old field is called "requestCookie", but this is
|
||||
// a response, so this name makes more sense
|
||||
repeated uint32 responseCookies = 7;
|
||||
// If a TMGetLedger request was received without a "requestCookie",
|
||||
// and the peer supports it, this flag will be set to true to
|
||||
// indicate that the receiver should process the result in addition
|
||||
// to forwarding it to its "responseCookies" peers.
|
||||
optional bool directResponse = 8;
|
||||
}
|
||||
|
||||
message TMPing {
|
||||
|
||||
@@ -82,19 +82,6 @@ HashRouter::shouldProcess(
|
||||
return s.shouldProcess(suppressionMap_.clock().now(), txInterval);
|
||||
}
|
||||
|
||||
bool
|
||||
HashRouter::shouldProcessForPeer(
|
||||
uint256 const& key,
|
||||
PeerShortID peer,
|
||||
std::chrono::seconds interval)
|
||||
{
|
||||
std::lock_guard lock(mutex_);
|
||||
|
||||
auto& entry = emplace(key).first;
|
||||
|
||||
return entry.shouldProcessForPeer(peer, suppressionMap_.clock().now(), interval);
|
||||
}
|
||||
|
||||
HashRouterFlags
|
||||
HashRouter::getFlags(uint256 const& key)
|
||||
{
|
||||
@@ -132,13 +119,4 @@ HashRouter::shouldRelay(uint256 const& key) -> std::optional<std::set<PeerShortI
|
||||
return s.releasePeerSet();
|
||||
}
|
||||
|
||||
auto
|
||||
HashRouter::getPeers(uint256 const& key) -> std::set<PeerShortID>
|
||||
{
|
||||
std::lock_guard lock(mutex_);
|
||||
|
||||
auto& s = emplace(key).first;
|
||||
return s.peekPeerSet();
|
||||
}
|
||||
|
||||
} // namespace xrpl
|
||||
|
||||
@@ -395,33 +395,6 @@ class HashRouter_test : public beast::unit_test::Suite
|
||||
BEAST_EXPECT(!any(HF::UNDEFINED));
|
||||
}
|
||||
|
||||
void
|
||||
testProcessPeer()
|
||||
{
|
||||
using namespace std::chrono_literals;
|
||||
TestStopwatch stopwatch;
|
||||
HashRouter router(getSetup(5s, 5s), stopwatch);
|
||||
uint256 const key(1);
|
||||
HashRouter::PeerShortID peer1 = 1;
|
||||
HashRouter::PeerShortID peer2 = 2;
|
||||
auto const timeout = 2s;
|
||||
|
||||
BEAST_EXPECT(router.shouldProcessForPeer(key, peer1, timeout));
|
||||
BEAST_EXPECT(!router.shouldProcessForPeer(key, peer1, timeout));
|
||||
++stopwatch;
|
||||
BEAST_EXPECT(!router.shouldProcessForPeer(key, peer1, timeout));
|
||||
BEAST_EXPECT(router.shouldProcessForPeer(key, peer2, timeout));
|
||||
BEAST_EXPECT(!router.shouldProcessForPeer(key, peer2, timeout));
|
||||
++stopwatch;
|
||||
BEAST_EXPECT(router.shouldProcessForPeer(key, peer1, timeout));
|
||||
BEAST_EXPECT(!router.shouldProcessForPeer(key, peer2, timeout));
|
||||
++stopwatch;
|
||||
BEAST_EXPECT(router.shouldProcessForPeer(key, peer2, timeout));
|
||||
++stopwatch;
|
||||
BEAST_EXPECT(router.shouldProcessForPeer(key, peer1, timeout));
|
||||
BEAST_EXPECT(!router.shouldProcessForPeer(key, peer2, timeout));
|
||||
}
|
||||
|
||||
public:
|
||||
void
|
||||
run() override
|
||||
@@ -434,7 +407,6 @@ public:
|
||||
testProcess();
|
||||
testSetup();
|
||||
testFlagsOps();
|
||||
testProcessPeer();
|
||||
}
|
||||
};
|
||||
|
||||
|
||||
@@ -175,7 +175,12 @@ public:
|
||||
}
|
||||
|
||||
void
|
||||
acquireAsync(uint256 const& hash, std::uint32_t seq, InboundLedger::Reason reason) override
|
||||
acquireAsync(
|
||||
JobType type,
|
||||
std::string const& name,
|
||||
uint256 const& hash,
|
||||
std::uint32_t seq,
|
||||
InboundLedger::Reason reason) override
|
||||
{
|
||||
}
|
||||
|
||||
@@ -384,11 +389,6 @@ public:
|
||||
{
|
||||
return false;
|
||||
}
|
||||
std::set<std::optional<uint64_t>>
|
||||
releaseRequestCookies(uint256 const& requestHash) override
|
||||
{
|
||||
return {};
|
||||
}
|
||||
|
||||
[[nodiscard]] std::string const&
|
||||
fingerprint() const override
|
||||
|
||||
165
src/test/basics/CanProcess_test.cpp
Normal file
165
src/test/basics/CanProcess_test.cpp
Normal file
@@ -0,0 +1,165 @@
|
||||
//------------------------------------------------------------------------------
|
||||
/*
|
||||
This file is part of rippled: https://github.com/ripple/rippled
|
||||
Copyright (c) 2012-2016 Ripple Labs Inc.
|
||||
|
||||
Permission to use, copy, modify, and/or distribute this software for any
|
||||
purpose with or without fee is hereby granted, provided that the above
|
||||
copyright notice and this permission notice appear in all copies.
|
||||
|
||||
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
|
||||
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
|
||||
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
|
||||
ANY SPECIAL , DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
|
||||
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
|
||||
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
|
||||
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
|
||||
*/
|
||||
//==============================================================================
|
||||
|
||||
#include <xrpl/basics/CanProcess.h>
|
||||
#include <xrpl/beast/unit_test.h>
|
||||
|
||||
#include <memory>
|
||||
|
||||
namespace ripple {
|
||||
namespace test {
|
||||
|
||||
struct CanProcess_test : beast::unit_test::Suite
|
||||
{
|
||||
template <class Mutex, class Collection, class Item>
|
||||
void
|
||||
test(
|
||||
std::string const& name,
|
||||
Mutex& mtx,
|
||||
Collection& collection,
|
||||
std::vector<Item> const& items)
|
||||
{
|
||||
testcase(name);
|
||||
|
||||
if (!BEAST_EXPECT(!items.empty()))
|
||||
return;
|
||||
if (!BEAST_EXPECT(collection.empty()))
|
||||
return;
|
||||
|
||||
// CanProcess objects can't be copied or moved. To make that easier,
|
||||
// store shared_ptrs
|
||||
std::vector<std::shared_ptr<CanProcess>> trackers;
|
||||
// Fill up the vector with two CanProcess for each Item. The first
|
||||
// inserts the item into the collection and is "good". The second does
|
||||
// not and is "bad".
|
||||
for (int i = 0; i < items.size(); ++i)
|
||||
{
|
||||
{
|
||||
auto const& good =
|
||||
trackers.emplace_back(std::make_shared<CanProcess>(mtx, collection, items[i]));
|
||||
BEAST_EXPECT(*good);
|
||||
}
|
||||
BEAST_EXPECT(trackers.size() == (2 * i) + 1);
|
||||
BEAST_EXPECT(collection.size() == i + 1);
|
||||
{
|
||||
auto const& bad =
|
||||
trackers.emplace_back(std::make_shared<CanProcess>(mtx, collection, items[i]));
|
||||
BEAST_EXPECT(!*bad);
|
||||
}
|
||||
BEAST_EXPECT(trackers.size() == 2 * (i + 1));
|
||||
BEAST_EXPECT(collection.size() == i + 1);
|
||||
}
|
||||
BEAST_EXPECT(collection.size() == items.size());
|
||||
// Now remove the items from the vector<CanProcess> two at a time, and
|
||||
// try to get another CanProcess for that item.
|
||||
for (int i = 0; i < items.size(); ++i)
|
||||
{
|
||||
// Remove the "bad" one in the second position
|
||||
// This will have no effect on the collection
|
||||
{
|
||||
auto const iter = trackers.begin() + 1;
|
||||
BEAST_EXPECT(!**iter);
|
||||
trackers.erase(iter);
|
||||
}
|
||||
BEAST_EXPECT(trackers.size() == (2 * items.size()) - 1);
|
||||
BEAST_EXPECT(collection.size() == items.size());
|
||||
{
|
||||
// Append a new "bad" one
|
||||
auto const& bad =
|
||||
trackers.emplace_back(std::make_shared<CanProcess>(mtx, collection, items[i]));
|
||||
BEAST_EXPECT(!*bad);
|
||||
}
|
||||
BEAST_EXPECT(trackers.size() == 2 * items.size());
|
||||
BEAST_EXPECT(collection.size() == items.size());
|
||||
|
||||
// Remove the "good" one from the front
|
||||
{
|
||||
auto const iter = trackers.begin();
|
||||
BEAST_EXPECT(**iter);
|
||||
trackers.erase(iter);
|
||||
}
|
||||
BEAST_EXPECT(trackers.size() == (2 * items.size()) - 1);
|
||||
BEAST_EXPECT(collection.size() == items.size() - 1);
|
||||
{
|
||||
// Append a new "good" one
|
||||
auto const& good =
|
||||
trackers.emplace_back(std::make_shared<CanProcess>(mtx, collection, items[i]));
|
||||
BEAST_EXPECT(*good);
|
||||
}
|
||||
BEAST_EXPECT(trackers.size() == 2 * items.size());
|
||||
BEAST_EXPECT(collection.size() == items.size());
|
||||
}
|
||||
// Now remove them all two at a time
|
||||
for (int i = items.size() - 1; i >= 0; --i)
|
||||
{
|
||||
// Remove the "bad" one from the front
|
||||
{
|
||||
auto const iter = trackers.begin();
|
||||
BEAST_EXPECT(!**iter);
|
||||
trackers.erase(iter);
|
||||
}
|
||||
BEAST_EXPECT(trackers.size() == (2 * i) + 1);
|
||||
BEAST_EXPECT(collection.size() == i + 1);
|
||||
// Remove the "good" one now in front
|
||||
{
|
||||
auto const iter = trackers.begin();
|
||||
BEAST_EXPECT(**iter);
|
||||
trackers.erase(iter);
|
||||
}
|
||||
BEAST_EXPECT(trackers.size() == 2 * i);
|
||||
BEAST_EXPECT(collection.size() == i);
|
||||
}
|
||||
BEAST_EXPECT(trackers.empty());
|
||||
BEAST_EXPECT(collection.empty());
|
||||
}
|
||||
|
||||
void
|
||||
run() override
|
||||
{
|
||||
{
|
||||
std::mutex m;
|
||||
std::set<int> collection;
|
||||
std::vector<int> const items{1, 2, 3, 4, 5};
|
||||
test("set of int", m, collection, items);
|
||||
}
|
||||
{
|
||||
std::mutex m;
|
||||
std::set<std::string> collection;
|
||||
std::vector<std::string> const items{"one", "two", "three", "four", "five"};
|
||||
test("set of string", m, collection, items);
|
||||
}
|
||||
{
|
||||
std::mutex m;
|
||||
std::unordered_set<char> collection;
|
||||
std::vector<char> const items{'1', '2', '3', '4', '5'};
|
||||
test("unorderd_set of char", m, collection, items);
|
||||
}
|
||||
{
|
||||
std::mutex m;
|
||||
std::unordered_set<std::uint64_t> collection;
|
||||
std::vector<std::uint64_t> const items{100u, 1000u, 150u, 4u, 0u};
|
||||
test("unordered_set of uint64_t", m, collection, items);
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
BEAST_DEFINE_TESTSUITE(CanProcess, ripple_basics, ripple);
|
||||
|
||||
} // namespace test
|
||||
} // namespace ripple
|
||||
@@ -63,8 +63,8 @@ public:
|
||||
negotiateProtocolVersion("RTXP/1.2, XRPL/2.0, XRPL/2.1") == makeProtocol(2, 1));
|
||||
BEAST_EXPECT(negotiateProtocolVersion("XRPL/2.2") == makeProtocol(2, 2));
|
||||
BEAST_EXPECT(
|
||||
negotiateProtocolVersion("RTXP/1.2, XRPL/2.2, XRPL/2.3, XRPL/2.4, XRPL/999.999") ==
|
||||
makeProtocol(2, 3));
|
||||
negotiateProtocolVersion("RTXP/1.2, XRPL/2.2, XRPL/2.3, XRPL/999.999") ==
|
||||
makeProtocol(2, 2));
|
||||
BEAST_EXPECT(negotiateProtocolVersion("XRPL/999.999, WebSocket/1.0") == std::nullopt);
|
||||
BEAST_EXPECT(negotiateProtocolVersion("") == std::nullopt);
|
||||
}
|
||||
|
||||
@@ -190,11 +190,6 @@ public:
|
||||
removeTxQueue(uint256 const&) override
|
||||
{
|
||||
}
|
||||
std::set<std::optional<uint64_t>>
|
||||
releaseRequestCookies(uint256 const& requestHash) override
|
||||
{
|
||||
return {};
|
||||
}
|
||||
};
|
||||
|
||||
/** Manually advanced clock. */
|
||||
|
||||
@@ -159,10 +159,8 @@ RCLConsensus::Adaptor::acquireLedger(LedgerHash const& hash)
|
||||
// Tell the ledger acquire system that we need the consensus ledger
|
||||
acquiringLedger_ = hash;
|
||||
|
||||
app_.getJobQueue().addJob(JtAdvance, "GetConsL1", [id = hash, &app = app_, this]() {
|
||||
JLOG(j_.debug()) << "JOB advanceLedger getConsensusLedger1 started";
|
||||
app.getInboundLedgers().acquireAsync(id, 0, InboundLedger::Reason::CONSENSUS);
|
||||
});
|
||||
app_.getInboundLedgers().acquireAsync(
|
||||
JtAdvance, "GetConsL1", hash, 0, InboundLedger::Reason::CONSENSUS);
|
||||
}
|
||||
return std::nullopt;
|
||||
}
|
||||
|
||||
@@ -130,12 +130,8 @@ RCLValidationsAdaptor::acquire(LedgerHash const& hash)
|
||||
{
|
||||
JLOG(j_.warn()) << "Need validated ledger for preferred ledger analysis " << hash;
|
||||
|
||||
Application* pApp = &app_;
|
||||
|
||||
app_.getJobQueue().addJob(JtAdvance, "GetConsL2", [pApp, hash, this]() {
|
||||
JLOG(j_.debug()) << "JOB advanceLedger getConsensusLedger2 started";
|
||||
pApp->getInboundLedgers().acquireAsync(hash, 0, InboundLedger::Reason::CONSENSUS);
|
||||
});
|
||||
app_.getInboundLedgers().acquireAsync(
|
||||
JtAdvance, "GetConsL2", hash, 0, InboundLedger::Reason::CONSENSUS);
|
||||
return std::nullopt;
|
||||
}
|
||||
|
||||
|
||||
@@ -189,22 +189,4 @@ private:
|
||||
std::unique_ptr<PeerSet> peerSet_;
|
||||
};
|
||||
|
||||
inline std::string
|
||||
to_string(InboundLedger::Reason reason)
|
||||
{
|
||||
using enum InboundLedger::Reason;
|
||||
switch (reason)
|
||||
{
|
||||
case HISTORY:
|
||||
return "HISTORY";
|
||||
case GENERIC:
|
||||
return "GENERIC";
|
||||
case CONSENSUS:
|
||||
return "CONSENSUS";
|
||||
default:
|
||||
UNREACHABLE("ripple::to_string(InboundLedger::Reason) : unknown value");
|
||||
return "unknown";
|
||||
}
|
||||
}
|
||||
|
||||
} // namespace xrpl
|
||||
|
||||
@@ -40,7 +40,12 @@ public:
|
||||
// Queue. TODO review whether all callers of acquire() can use this
|
||||
// instead. Inbound ledger acquisition is asynchronous anyway.
|
||||
virtual void
|
||||
acquireAsync(uint256 const& hash, std::uint32_t seq, InboundLedger::Reason reason) = 0;
|
||||
acquireAsync(
|
||||
JobType type,
|
||||
std::string const& name,
|
||||
uint256 const& hash,
|
||||
std::uint32_t seq,
|
||||
InboundLedger::Reason reason) = 0;
|
||||
|
||||
virtual std::shared_ptr<InboundLedger>
|
||||
find(LedgerHash const& hash) = 0;
|
||||
|
||||
@@ -79,78 +79,12 @@ public:
|
||||
XRPL_ASSERT(
|
||||
hash.isNonZero(), "xrpl::InboundLedgersImp::acquire::doAcquire : nonzero hash");
|
||||
|
||||
bool const needNetworkLedger = app_.getOPs().isNeedNetworkLedger();
|
||||
bool const shouldAcquire = [&]() {
|
||||
if (!needNetworkLedger)
|
||||
return true;
|
||||
if (reason == InboundLedger::Reason::GENERIC)
|
||||
return true;
|
||||
if (reason == InboundLedger::Reason::CONSENSUS)
|
||||
return true;
|
||||
return false;
|
||||
}();
|
||||
// probably not the right rule
|
||||
if (app_.getOPs().isNeedNetworkLedger() && (reason != InboundLedger::Reason::GENERIC) &&
|
||||
(reason != InboundLedger::Reason::CONSENSUS))
|
||||
return {};
|
||||
|
||||
std::stringstream ss;
|
||||
ss << "InboundLedger::acquire: "
|
||||
<< "Request: " << to_string(hash) << ", " << seq
|
||||
<< " NeedNetworkLedger: " << (needNetworkLedger ? "yes" : "no")
|
||||
<< " Reason: " << to_string(reason)
|
||||
<< " Should acquire: " << (shouldAcquire ? "true." : "false.");
|
||||
|
||||
/* Acquiring ledgers is somewhat expensive. It requires lots of
|
||||
* computation and network communication. Avoid it when it's not
|
||||
* appropriate. Every validation from a peer for a ledger that
|
||||
* we do not have locally results in a call to this function: even
|
||||
* if we are moments away from validating the same ledger.
|
||||
*/
|
||||
bool const shouldBroadcast = [&]() {
|
||||
// If the node is not in "full" state, it needs to sync to
|
||||
// the network, and doesn't have the necessary tx's and
|
||||
// ledger entries to build the ledger.
|
||||
bool const isFull = app_.getOPs().isFull();
|
||||
// If everything else is ok, don't try to acquire the ledger
|
||||
// if the requested seq is in the near future relative to
|
||||
// the validated ledger. If the requested ledger is between
|
||||
// 1 and 19 inclusive ledgers ahead of the valid ledger this
|
||||
// node has not built it yet, but it's possible/likely it
|
||||
// has the tx's necessary to build it and get caught up.
|
||||
// Plus it might not become validated. On the other hand, if
|
||||
// it's more than 20 in the future, this node should request
|
||||
// it so that it can jump ahead and get caught up.
|
||||
LedgerIndex const validSeq = app_.getLedgerMaster().getValidLedgerIndex();
|
||||
constexpr std::size_t lagLeeway = 20;
|
||||
bool const nearFuture = (seq > validSeq) && (seq < validSeq + lagLeeway);
|
||||
// If everything else is ok, don't try to acquire the ledger
|
||||
// if the request is related to consensus. (Note that
|
||||
// consensus calls usually pass a seq of 0, so nearFuture
|
||||
// will be false other than on a brand new network.)
|
||||
bool const consensus = reason == InboundLedger::Reason::CONSENSUS;
|
||||
ss << " Evaluating whether to broadcast requests to peers"
|
||||
<< ". full: " << (isFull ? "true" : "false") << ". ledger sequence " << seq
|
||||
<< ". Valid sequence: " << validSeq << ". Lag leeway: " << lagLeeway
|
||||
<< ". request for near future ledger: " << (nearFuture ? "true" : "false")
|
||||
<< ". Consensus: " << (consensus ? "true" : "false");
|
||||
|
||||
// If the node is not synced, send requests.
|
||||
if (!isFull)
|
||||
return true;
|
||||
// If the ledger is in the near future, do NOT send requests.
|
||||
// This node is probably about to build it.
|
||||
if (nearFuture)
|
||||
return false;
|
||||
// If the request is because of consensus, do NOT send requests.
|
||||
// This node is probably about to build it.
|
||||
if (consensus)
|
||||
return false;
|
||||
return true;
|
||||
}();
|
||||
ss << ". Would broadcast to peers? " << (shouldBroadcast ? "true." : "false.");
|
||||
|
||||
if (!shouldAcquire)
|
||||
{
|
||||
JLOG(j_.debug()) << "Abort(rule): " << ss.str();
|
||||
return {};
|
||||
}
|
||||
|
||||
bool isNew = true;
|
||||
std::shared_ptr<InboundLedger> inbound;
|
||||
@@ -202,25 +136,35 @@ public:
|
||||
}
|
||||
|
||||
void
|
||||
acquireAsync(uint256 const& hash, std::uint32_t seq, InboundLedger::Reason reason) override
|
||||
acquireAsync(
|
||||
JobType type,
|
||||
std::string const& name,
|
||||
uint256 const& hash,
|
||||
std::uint32_t seq,
|
||||
InboundLedger::Reason reason) override
|
||||
{
|
||||
if (CanProcess const check{acquiresMutex_, pendingAcquires_, hash})
|
||||
if (auto check = std::make_shared<CanProcess const>(acquiresMutex_, pendingAcquires_, hash);
|
||||
*check)
|
||||
{
|
||||
try
|
||||
{
|
||||
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;
|
||||
}
|
||||
app_.getJobQueue().addJob(type, name, [check, name, hash, seq, reason, this]() {
|
||||
JLOG(j_.debug()) << "JOB acquireAsync " << name << " started ";
|
||||
try
|
||||
{
|
||||
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;
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -23,7 +23,6 @@ enum class ProtocolFeature {
|
||||
ValidatorListPropagation,
|
||||
ValidatorList2Propagation,
|
||||
LedgerReplay,
|
||||
LedgerDataCookies
|
||||
};
|
||||
|
||||
/** Represents a peer connection in the overlay. */
|
||||
@@ -123,13 +122,6 @@ public:
|
||||
|
||||
[[nodiscard]] virtual bool
|
||||
txReduceRelayEnabled() const = 0;
|
||||
|
||||
//
|
||||
// Messages
|
||||
//
|
||||
|
||||
virtual std::set<std::optional<uint64_t>>
|
||||
releaseRequestCookies(uint256 const& requestHash) = 0;
|
||||
};
|
||||
|
||||
} // namespace xrpl
|
||||
|
||||
@@ -111,9 +111,6 @@ constexpr std::chrono::milliseconds kPeerHighLatency{300};
|
||||
/** How often we PING the peer to check for latency and sendq probe */
|
||||
constexpr std::chrono::seconds kPeerTimerInterval{60};
|
||||
|
||||
/** How often we process duplicate incoming TMGetLedger messages */
|
||||
std::chrono::seconds constexpr kGetLedgerInterval{15};
|
||||
|
||||
} // namespace
|
||||
|
||||
// TODO: Remove this exclusion once unit tests are added after the hotfix
|
||||
@@ -544,8 +541,6 @@ PeerImp::supportsFeature(ProtocolFeature f) const
|
||||
return protocol_ >= makeProtocol(2, 2);
|
||||
case ProtocolFeature::LedgerReplay:
|
||||
return ledgerReplayEnabled_;
|
||||
case ProtocolFeature::LedgerDataCookies:
|
||||
return protocol_ >= makeProtocol(2, 3);
|
||||
}
|
||||
return false;
|
||||
}
|
||||
@@ -1415,9 +1410,8 @@ PeerImp::handleTransaction(
|
||||
void
|
||||
PeerImp::onMessage(std::shared_ptr<protocol::TMGetLedger> const& m)
|
||||
{
|
||||
auto badData = [&](std::string const& msg, bool chargefee = true) {
|
||||
if (chargefee)
|
||||
fee_.update(Resource::kFeeInvalidData, "get_ledger " + msg);
|
||||
auto badData = [&](std::string const& msg) {
|
||||
fee_.update(Resource::kFeeInvalidData, "get_ledger " + msg);
|
||||
JLOG(pJournal_.warn()) << "TMGetLedger: " << msg;
|
||||
};
|
||||
auto const itype{m->itype()};
|
||||
@@ -1515,68 +1509,11 @@ PeerImp::onMessage(std::shared_ptr<protocol::TMGetLedger> const& m)
|
||||
}
|
||||
}
|
||||
|
||||
// Drop duplicate requests from the same peer for at least
|
||||
// `kGetLedgerInterval` seconds.
|
||||
// Append a little junk to prevent the hash of an incoming messsage
|
||||
// from matching the hash of the same outgoing message.
|
||||
// `shouldProcessForPeer` does not distingish between incoming and
|
||||
// outgoing, and some of the message relay logic checks the hash to see
|
||||
// if the message has been relayed already. If the hashes are the same,
|
||||
// a duplicate will be detected when sending the message is attempted,
|
||||
// so it will fail.
|
||||
auto const messageHash = sha512Half(*m, nullptr);
|
||||
// Request cookies are not included in the hash. Track them here.
|
||||
auto const requestCookie = [&m]() -> std::optional<uint64_t> {
|
||||
if (m->has_requestcookie())
|
||||
return m->requestcookie();
|
||||
return std::nullopt;
|
||||
}();
|
||||
auto const [inserted, pending] = [&] {
|
||||
std::lock_guard lock{cookieLock_};
|
||||
auto& cookies = messageRequestCookies_[messageHash];
|
||||
bool const pending = !cookies.empty();
|
||||
return std::pair{cookies.emplace(requestCookie).second, pending};
|
||||
}();
|
||||
// Check if the request has been seen from this peer.
|
||||
if (!app_.getHashRouter().shouldProcessForPeer(messageHash, id_, kGetLedgerInterval))
|
||||
{
|
||||
// This request has already been seen from this peer.
|
||||
// Has it been seen with this request cookie (or lack thereof)?
|
||||
|
||||
if (inserted)
|
||||
{
|
||||
// This is a duplicate request, but with a new cookie. When a
|
||||
// response is ready, one will be sent for each request cookie.
|
||||
JLOG(pJournal_.debug())
|
||||
<< "TMGetLedger: duplicate request with new request cookie: "
|
||||
<< requestCookie.value_or(0) << ". Job pending: " << (pending ? "yes" : "no")
|
||||
<< ": " << messageHash;
|
||||
if (pending)
|
||||
{
|
||||
// Don't bother queueing up a new job if other requests are
|
||||
// already pending. This should limit entries in the job queue
|
||||
// to one per peer per unique request.
|
||||
JLOG(pJournal_.debug()) << "TMGetLedger: Suppressing recvGetLedger job, since one "
|
||||
"is pending: "
|
||||
<< messageHash;
|
||||
return;
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
// Don't punish nodes that don't know any better
|
||||
return badData(
|
||||
"duplicate request: " + to_string(messageHash),
|
||||
supportsFeature(ProtocolFeature::LedgerDataCookies));
|
||||
}
|
||||
}
|
||||
|
||||
// Queue a job to process the request
|
||||
JLOG(pJournal_.debug()) << "TMGetLedger: Adding recvGetLedger job: " << messageHash;
|
||||
std::weak_ptr<PeerImp> const weak = shared_from_this();
|
||||
app_.getJobQueue().addJob(JtLedgerReq, "RcvGetLedger", [weak, m, messageHash]() {
|
||||
app_.getJobQueue().addJob(JtLedgerReq, "RcvGetLedger", [weak, m]() {
|
||||
if (auto peer = weak.lock())
|
||||
peer->processLedgerRequest(m, messageHash);
|
||||
peer->processLedgerRequest(m);
|
||||
});
|
||||
}
|
||||
|
||||
@@ -1683,9 +1620,8 @@ PeerImp::onMessage(std::shared_ptr<protocol::TMReplayDeltaResponse> const& m)
|
||||
void
|
||||
PeerImp::onMessage(std::shared_ptr<protocol::TMLedgerData> const& m)
|
||||
{
|
||||
auto badData = [&](std::string const& msg, bool charge = true) {
|
||||
if (charge)
|
||||
fee_.update(Resource::kFeeInvalidData, msg);
|
||||
auto badData = [&](std::string const& msg) {
|
||||
fee_.update(Resource::kFeeInvalidData, msg);
|
||||
JLOG(pJournal_.warn()) << "TMLedgerData: " << msg;
|
||||
};
|
||||
|
||||
@@ -1742,93 +1678,19 @@ PeerImp::onMessage(std::shared_ptr<protocol::TMLedgerData> const& m)
|
||||
return;
|
||||
}
|
||||
|
||||
auto const messageHash = sha512Half(*m);
|
||||
if (!app_.getHashRouter().addSuppressionPeer(messageHash, id_))
|
||||
// If there is a request cookie, attempt to relay the message
|
||||
if (m->has_requestcookie())
|
||||
{
|
||||
// Don't punish nodes that don't know any better
|
||||
return badData(
|
||||
"Duplicate message: " + to_string(messageHash),
|
||||
supportsFeature(ProtocolFeature::LedgerDataCookies));
|
||||
}
|
||||
|
||||
bool const routed =
|
||||
m->has_directresponse() || m->responsecookies_size() || m->has_requestcookie();
|
||||
|
||||
{
|
||||
// Check if this message needs to be forwarded to one or more peers.
|
||||
// Maximum of one of the relevant fields should be populated.
|
||||
XRPL_ASSERT(
|
||||
!m->has_requestcookie() || !m->responsecookies_size(),
|
||||
"ripple::PeerImp::onMessage(TMLedgerData) : valid cookie fields");
|
||||
|
||||
// Make a copy of the response cookies, then wipe the list so it can be
|
||||
// forwarded cleanly
|
||||
auto const responseCookies = m->responsecookies();
|
||||
m->clear_responsecookies();
|
||||
// Flag indicating if this response should be processed locally,
|
||||
// possibly in addition to being forwarded.
|
||||
bool const directResponse = m->has_directresponse() && m->directresponse();
|
||||
m->clear_directresponse();
|
||||
|
||||
auto const relay = [this, m, &messageHash](auto const cookie) {
|
||||
if (auto peer = overlay_.findPeerByShortID(cookie))
|
||||
{
|
||||
XRPL_ASSERT(
|
||||
!m->has_requestcookie() && !m->responsecookies_size(),
|
||||
"ripple::PeerImp::onMessage(TMLedgerData) relay : no "
|
||||
"cookies");
|
||||
if (peer->supportsFeature(ProtocolFeature::LedgerDataCookies))
|
||||
// Setting this flag is not _strictly_ necessary for peers
|
||||
// that support it if there are no cookies included in the
|
||||
// message, but it is more accurate.
|
||||
m->set_directresponse(true);
|
||||
else
|
||||
m->clear_directresponse();
|
||||
peer->send(std::make_shared<Message>(*m, protocol::mtLEDGER_DATA));
|
||||
}
|
||||
else
|
||||
JLOG(pJournal_.info()) << "Unable to route TX/ledger data reply to peer [" << cookie
|
||||
<< "]: " << messageHash;
|
||||
};
|
||||
// If there is a request cookie, attempt to relay the message
|
||||
if (m->has_requestcookie())
|
||||
if (auto peer = overlay_.findPeerByShortID(m->requestcookie()))
|
||||
{
|
||||
XRPL_ASSERT(
|
||||
responseCookies.empty(),
|
||||
"ripple::PeerImp::onMessage(TMLedgerData) : no response "
|
||||
"cookies");
|
||||
m->clear_requestcookie();
|
||||
relay(m->requestcookie());
|
||||
if (!directResponse && responseCookies.empty())
|
||||
return;
|
||||
peer->send(std::make_shared<Message>(*m, protocol::mtLEDGER_DATA));
|
||||
}
|
||||
// If there's a list of request cookies, attempt to relay the message to
|
||||
// all of them.
|
||||
if (responseCookies.size())
|
||||
else
|
||||
{
|
||||
for (auto const cookie : responseCookies)
|
||||
relay(cookie);
|
||||
if (!directResponse)
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
// Now that any forwarding is done check the base message (data only, no
|
||||
// routing info for duplicates)
|
||||
if (routed)
|
||||
{
|
||||
m->clear_directresponse();
|
||||
XRPL_ASSERT(
|
||||
!m->has_requestcookie() && !m->responsecookies_size(),
|
||||
"ripple::PeerImp::onMessage(TMLedgerData) : no cookies");
|
||||
auto const baseMessageHash = sha512Half(*m);
|
||||
if (!app_.getHashRouter().addSuppressionPeer(baseMessageHash, id_))
|
||||
{
|
||||
// Don't punish nodes that don't know any better
|
||||
return badData(
|
||||
"Duplicate message: " + to_string(baseMessageHash),
|
||||
supportsFeature(ProtocolFeature::LedgerDataCookies));
|
||||
JLOG(pJournal_.info()) << "Unable to route TX/ledger data reply";
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
uint256 const ledgerHash = uint256::fromRaw(m->ledgerhash());
|
||||
@@ -3219,21 +3081,16 @@ PeerImp::checkValidation(
|
||||
// the TX tree with the specified root hash.
|
||||
//
|
||||
static std::shared_ptr<PeerImp>
|
||||
getPeerWithTree(
|
||||
OverlayImpl& ov,
|
||||
uint256 const& rootHash,
|
||||
PeerImp const* skip,
|
||||
std::function<bool(Peer::id_t)> shouldProcessCallback)
|
||||
getPeerWithTree(OverlayImpl& ov, uint256 const& rootHash, PeerImp const* skip)
|
||||
{
|
||||
std::shared_ptr<PeerImp> ret;
|
||||
int retScore = 0;
|
||||
|
||||
XRPL_ASSERT(shouldProcessCallback, "ripple::getPeerWithTree : callback provided");
|
||||
ov.forEach([&](std::shared_ptr<PeerImp>&& p) {
|
||||
if (p->hasTxSet(rootHash) && p.get() != skip)
|
||||
{
|
||||
auto score = p->getScore(true);
|
||||
if (!ret || (score > retScore && shouldProcessCallback(p->id())))
|
||||
if (!ret || (score > retScore))
|
||||
{
|
||||
ret = std::move(p);
|
||||
retScore = score;
|
||||
@@ -3252,18 +3109,16 @@ getPeerWithLedger(
|
||||
OverlayImpl& ov,
|
||||
uint256 const& ledgerHash,
|
||||
LedgerIndex ledger,
|
||||
PeerImp const* skip,
|
||||
std::function<bool(Peer::id_t)> shouldProcessCallback)
|
||||
PeerImp const* skip)
|
||||
{
|
||||
std::shared_ptr<PeerImp> ret;
|
||||
int retScore = 0;
|
||||
|
||||
XRPL_ASSERT(shouldProcessCallback, "ripple::getPeerWithLedger : callback provided");
|
||||
ov.forEach([&](std::shared_ptr<PeerImp>&& p) {
|
||||
if (p->hasLedger(ledgerHash, ledger) && p.get() != skip)
|
||||
{
|
||||
auto score = p->getScore(true);
|
||||
if (!ret || (score > retScore && shouldProcessCallback(p->id())))
|
||||
if (!ret || (score > retScore))
|
||||
{
|
||||
ret = std::move(p);
|
||||
retScore = score;
|
||||
@@ -3277,8 +3132,7 @@ getPeerWithLedger(
|
||||
void
|
||||
PeerImp::sendLedgerBase(
|
||||
std::shared_ptr<Ledger const> const& ledger,
|
||||
protocol::TMLedgerData& ledgerData,
|
||||
PeerCookieMap const& destinations)
|
||||
protocol::TMLedgerData& ledgerData)
|
||||
{
|
||||
JLOG(pJournal_.trace()) << "sendLedgerBase: Base data";
|
||||
|
||||
@@ -3308,92 +3162,14 @@ PeerImp::sendLedgerBase(
|
||||
}
|
||||
}
|
||||
|
||||
sendToMultiple(ledgerData, destinations);
|
||||
}
|
||||
|
||||
void
|
||||
PeerImp::sendToMultiple(protocol::TMLedgerData& ledgerData, PeerCookieMap const& destinations)
|
||||
{
|
||||
bool foundSelf = false;
|
||||
for (auto const& [peer, cookies] : destinations)
|
||||
{
|
||||
if (peer.get() == this)
|
||||
foundSelf = true;
|
||||
bool const multipleCookies = peer->supportsFeature(ProtocolFeature::LedgerDataCookies);
|
||||
std::vector<std::uint64_t> sendCookies;
|
||||
|
||||
bool directResponse = false;
|
||||
if (!multipleCookies)
|
||||
{
|
||||
JLOG(pJournal_.debug()) << "sendToMultiple: Sending " << cookies.size()
|
||||
<< " TMLedgerData messages to peer [" << peer->id()
|
||||
<< "]: " << sha512Half(ledgerData);
|
||||
}
|
||||
for (auto const& cookie : cookies)
|
||||
{
|
||||
// Unfortunately, need a separate Message object for every
|
||||
// combination
|
||||
if (cookie)
|
||||
{
|
||||
if (multipleCookies)
|
||||
{
|
||||
// Save this one for later to send a single message
|
||||
sendCookies.emplace_back(*cookie);
|
||||
continue;
|
||||
}
|
||||
|
||||
// Feature not supported, so send a single message with a
|
||||
// single cookie
|
||||
ledgerData.set_requestcookie(*cookie);
|
||||
}
|
||||
else
|
||||
{
|
||||
if (multipleCookies)
|
||||
{
|
||||
// Set this flag later on the single message
|
||||
directResponse = true;
|
||||
continue;
|
||||
}
|
||||
|
||||
ledgerData.clear_requestcookie();
|
||||
}
|
||||
XRPL_ASSERT(
|
||||
!multipleCookies,
|
||||
"ripple::PeerImp::sendToMultiple : ledger data cookies "
|
||||
"unsupported");
|
||||
auto message{std::make_shared<Message>(ledgerData, protocol::mtLEDGER_DATA)};
|
||||
peer->send(message);
|
||||
}
|
||||
if (multipleCookies)
|
||||
{
|
||||
// Send a single message with all the cookies and/or the direct
|
||||
// response flag, so the receiver can farm out the single message to
|
||||
// multiple peers and/or itself
|
||||
XRPL_ASSERT(
|
||||
sendCookies.size() || directResponse,
|
||||
"ripple::PeerImp::sendToMultiple : valid response options");
|
||||
ledgerData.clear_requestcookie();
|
||||
ledgerData.clear_responsecookies();
|
||||
ledgerData.set_directresponse(directResponse);
|
||||
for (auto const& cookie : sendCookies)
|
||||
ledgerData.add_responsecookies(cookie);
|
||||
auto message{std::make_shared<Message>(ledgerData, protocol::mtLEDGER_DATA)};
|
||||
peer->send(message);
|
||||
|
||||
JLOG(pJournal_.debug())
|
||||
<< "sendToMultiple: Sent 1 TMLedgerData message to peer [" << peer->id()
|
||||
<< "]: including " << (directResponse ? "the direct response flag and " : "")
|
||||
<< sendCookies.size() << " response cookies. "
|
||||
<< ": " << sha512Half(ledgerData);
|
||||
}
|
||||
}
|
||||
XRPL_ASSERT(foundSelf, "ripple::PeerImp::sendToMultiple : current peer included");
|
||||
auto message{std::make_shared<Message>(ledgerData, protocol::mtLEDGER_DATA)};
|
||||
send(message);
|
||||
}
|
||||
|
||||
std::shared_ptr<Ledger const>
|
||||
PeerImp::getLedger(std::shared_ptr<protocol::TMGetLedger> const& m, uint256 const& mHash)
|
||||
PeerImp::getLedger(std::shared_ptr<protocol::TMGetLedger> const& m)
|
||||
{
|
||||
JLOG(pJournal_.trace()) << "getLedger: Ledger " << mHash;
|
||||
JLOG(pJournal_.trace()) << "getLedger: Ledger";
|
||||
|
||||
std::shared_ptr<Ledger const> ledger;
|
||||
|
||||
@@ -3409,30 +3185,16 @@ PeerImp::getLedger(std::shared_ptr<protocol::TMGetLedger> const& m, uint256 cons
|
||||
if (m->has_querytype() && !m->has_requestcookie())
|
||||
{
|
||||
// Attempt to relay the request to a peer
|
||||
// Note repeated messages will not relay to the same peer
|
||||
// before `kGetLedgerInterval` seconds. This prevents one
|
||||
// peer from getting flooded, and distributes the request
|
||||
// load. If a request has been relayed to all eligible
|
||||
// peers, then this message will not be relayed.
|
||||
if (auto const peer = getPeerWithLedger(
|
||||
overlay_,
|
||||
ledgerHash,
|
||||
m->has_ledgerseq() ? m->ledgerseq() : 0,
|
||||
this,
|
||||
[&](Peer::id_t id) {
|
||||
return app_.getHashRouter().shouldProcessForPeer(
|
||||
mHash, id, kGetLedgerInterval);
|
||||
}))
|
||||
overlay_, ledgerHash, m->has_ledgerseq() ? m->ledgerseq() : 0, this))
|
||||
{
|
||||
m->set_requestcookie(id());
|
||||
peer->send(std::make_shared<Message>(*m, protocol::mtGET_LEDGER));
|
||||
JLOG(pJournal_.debug())
|
||||
<< "getLedger: Request relayed to peer [" << peer->id() << "]: " << mHash;
|
||||
JLOG(pJournal_.debug()) << "getLedger: Request relayed to peer";
|
||||
return ledger;
|
||||
}
|
||||
|
||||
JLOG(pJournal_.trace())
|
||||
<< "getLedger: Don't have ledger with hash " << ledgerHash << ": " << mHash;
|
||||
JLOG(pJournal_.trace()) << "getLedger: Failed to find peer to relay request";
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -3441,15 +3203,15 @@ PeerImp::getLedger(std::shared_ptr<protocol::TMGetLedger> const& m, uint256 cons
|
||||
// Attempt to find ledger by sequence
|
||||
if (m->ledgerseq() < app_.getLedgerMaster().getEarliestFetch())
|
||||
{
|
||||
JLOG(pJournal_.debug()) << "getLedger: Early ledger sequence request " << mHash;
|
||||
JLOG(pJournal_.debug()) << "getLedger: Early ledger sequence request";
|
||||
}
|
||||
else
|
||||
{
|
||||
ledger = app_.getLedgerMaster().getLedgerBySeq(m->ledgerseq());
|
||||
if (!ledger)
|
||||
{
|
||||
JLOG(pJournal_.debug()) << "getLedger: Don't have ledger with sequence "
|
||||
<< m->ledgerseq() << ": " << mHash;
|
||||
JLOG(pJournal_.debug())
|
||||
<< "getLedger: Don't have ledger with sequence " << m->ledgerseq();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -3471,29 +3233,27 @@ PeerImp::getLedger(std::shared_ptr<protocol::TMGetLedger> const& m, uint256 cons
|
||||
charge(Resource::kFeeMalformedRequest, "get_ledger ledgerSeq");
|
||||
|
||||
ledger.reset();
|
||||
JLOG(pJournal_.warn())
|
||||
<< "getLedger: Invalid ledger sequence " << ledgerSeq << ": " << mHash;
|
||||
JLOG(pJournal_.warn()) << "getLedger: Invalid ledger sequence " << ledgerSeq;
|
||||
}
|
||||
}
|
||||
else if (ledgerSeq < app_.getLedgerMaster().getEarliestFetch())
|
||||
{
|
||||
ledger.reset();
|
||||
JLOG(pJournal_.debug())
|
||||
<< "getLedger: Early ledger sequence request " << ledgerSeq << ": " << mHash;
|
||||
JLOG(pJournal_.debug()) << "getLedger: Early ledger sequence request " << ledgerSeq;
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
JLOG(pJournal_.debug()) << "getLedger: Unable to find ledger " << mHash;
|
||||
JLOG(pJournal_.debug()) << "getLedger: Unable to find ledger";
|
||||
}
|
||||
|
||||
return ledger;
|
||||
}
|
||||
|
||||
std::shared_ptr<SHAMap const>
|
||||
PeerImp::getTxSet(std::shared_ptr<protocol::TMGetLedger> const& m, uint256 const& mHash) const
|
||||
PeerImp::getTxSet(std::shared_ptr<protocol::TMGetLedger> const& m) const
|
||||
{
|
||||
JLOG(pJournal_.trace()) << "getTxSet: TX set " << mHash;
|
||||
JLOG(pJournal_.trace()) << "getTxSet: TX set";
|
||||
|
||||
uint256 const txSetHash = uint256::fromRaw(m->ledgerhash());
|
||||
std::shared_ptr<SHAMap> shaMap{app_.getInboundTransactions().getSet(txSetHash, false)};
|
||||
@@ -3502,28 +3262,20 @@ PeerImp::getTxSet(std::shared_ptr<protocol::TMGetLedger> const& m, uint256 const
|
||||
if (m->has_querytype() && !m->has_requestcookie())
|
||||
{
|
||||
// Attempt to relay the request to a peer
|
||||
// Note repeated messages will not relay to the same peer
|
||||
// before `kGetLedgerInterval` seconds. This prevents one
|
||||
// peer from getting flooded, and distributes the request
|
||||
// load. If a request has been relayed to all eligible
|
||||
// peers, then this message will not be relayed.
|
||||
if (auto const peer = getPeerWithTree(overlay_, txSetHash, this, [&](Peer::id_t id) {
|
||||
return app_.getHashRouter().shouldProcessForPeer(mHash, id, kGetLedgerInterval);
|
||||
}))
|
||||
if (auto const peer = getPeerWithTree(overlay_, txSetHash, this))
|
||||
{
|
||||
m->set_requestcookie(id());
|
||||
peer->send(std::make_shared<Message>(*m, protocol::mtGET_LEDGER));
|
||||
JLOG(pJournal_.debug())
|
||||
<< "getTxSet: Request relayed to peer [" << peer->id() << "]: " << mHash;
|
||||
JLOG(pJournal_.debug()) << "getTxSet: Request relayed";
|
||||
}
|
||||
else
|
||||
{
|
||||
JLOG(pJournal_.debug()) << "getTxSet: Failed to find relay peer: " << mHash;
|
||||
JLOG(pJournal_.debug()) << "getTxSet: Failed to find relay peer";
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
JLOG(pJournal_.debug()) << "getTxSet: Failed to find TX set " << mHash;
|
||||
JLOG(pJournal_.debug()) << "getTxSet: Failed to find TX set";
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3531,7 +3283,7 @@ PeerImp::getTxSet(std::shared_ptr<protocol::TMGetLedger> const& m, uint256 const
|
||||
}
|
||||
|
||||
void
|
||||
PeerImp::processLedgerRequest(std::shared_ptr<protocol::TMGetLedger> const& m, uint256 const& mHash)
|
||||
PeerImp::processLedgerRequest(std::shared_ptr<protocol::TMGetLedger> const& m)
|
||||
{
|
||||
// Do not resource charge a peer responding to a relay
|
||||
if (!m->has_requestcookie())
|
||||
@@ -3544,72 +3296,9 @@ PeerImp::processLedgerRequest(std::shared_ptr<protocol::TMGetLedger> const& m, u
|
||||
bool fatLeaves{true};
|
||||
auto const itype{m->itype()};
|
||||
|
||||
auto getDestinations = [&] {
|
||||
// If a ledger data message is generated, it's going to be sent to every
|
||||
// peer that is waiting for it.
|
||||
|
||||
PeerCookieMap result;
|
||||
|
||||
std::size_t numCookies = 0;
|
||||
{
|
||||
// Don't do the work under this peer if this peer is not waiting for
|
||||
// any replies
|
||||
auto myCookies = releaseRequestCookies(mHash);
|
||||
if (myCookies.empty())
|
||||
{
|
||||
JLOG(pJournal_.debug()) << "TMGetLedger: peer is no longer "
|
||||
"waiting for response to request: "
|
||||
<< mHash;
|
||||
return result;
|
||||
}
|
||||
numCookies += myCookies.size();
|
||||
result[shared_from_this()] = myCookies;
|
||||
}
|
||||
|
||||
std::set<HashRouter::PeerShortID> const peers = app_.getHashRouter().getPeers(mHash);
|
||||
for (auto const peerID : peers)
|
||||
{
|
||||
// This loop does not need to be done under the HashRouter
|
||||
// lock because findPeerByShortID and releaseRequestCookies
|
||||
// are thread safe, and everything else is local
|
||||
if (auto p = overlay_.findPeerByShortID(peerID))
|
||||
{
|
||||
auto cookies = p->releaseRequestCookies(mHash);
|
||||
numCookies += cookies.size();
|
||||
if (result.contains(p))
|
||||
{
|
||||
// Unlikely, but if a request came in to this peer while
|
||||
// iterating, add the items instead of copying /
|
||||
// overwriting.
|
||||
XRPL_ASSERT(
|
||||
p.get() == this,
|
||||
"ripple::PeerImp::processLedgerRequest : found self in "
|
||||
"map");
|
||||
for (auto const& cookie : cookies)
|
||||
result[p].emplace(cookie);
|
||||
}
|
||||
else if (cookies.size())
|
||||
result[p] = cookies;
|
||||
}
|
||||
}
|
||||
|
||||
JLOG(pJournal_.debug()) << "TMGetLedger: Processing request for " << result.size()
|
||||
<< " peers. Will send " << numCookies
|
||||
<< " messages if successful: " << mHash;
|
||||
|
||||
return result;
|
||||
};
|
||||
// Will only populate this if we're going to do work.
|
||||
PeerCookieMap destinations;
|
||||
|
||||
if (itype == protocol::liTS_CANDIDATE)
|
||||
{
|
||||
destinations = getDestinations();
|
||||
if (destinations.empty())
|
||||
// Nowhere to send the response!
|
||||
return;
|
||||
|
||||
if (sharedMap = getTxSet(m, mHash); !sharedMap)
|
||||
if (sharedMap = getTxSet(m); !sharedMap)
|
||||
return;
|
||||
map = sharedMap.get();
|
||||
|
||||
@@ -3617,6 +3306,8 @@ PeerImp::processLedgerRequest(std::shared_ptr<protocol::TMGetLedger> const& m, u
|
||||
ledgerData.set_ledgerseq(0);
|
||||
ledgerData.set_ledgerhash(m->ledgerhash());
|
||||
ledgerData.set_type(protocol::liTS_CANDIDATE);
|
||||
if (m->has_requestcookie())
|
||||
ledgerData.set_requestcookie(m->requestcookie());
|
||||
|
||||
// We'll already have most transactions
|
||||
fatLeaves = false;
|
||||
@@ -3634,12 +3325,7 @@ PeerImp::processLedgerRequest(std::shared_ptr<protocol::TMGetLedger> const& m, u
|
||||
return;
|
||||
}
|
||||
|
||||
destinations = getDestinations();
|
||||
if (destinations.empty())
|
||||
// Nowhere to send the response!
|
||||
return;
|
||||
|
||||
if (ledger = getLedger(m, mHash); !ledger)
|
||||
if (ledger = getLedger(m); !ledger)
|
||||
return;
|
||||
|
||||
// Fill out the reply
|
||||
@@ -3647,11 +3333,13 @@ PeerImp::processLedgerRequest(std::shared_ptr<protocol::TMGetLedger> const& m, u
|
||||
ledgerData.set_ledgerhash(ledgerHash.begin(), ledgerHash.size());
|
||||
ledgerData.set_ledgerseq(ledger->header().seq);
|
||||
ledgerData.set_type(itype);
|
||||
if (m->has_requestcookie())
|
||||
ledgerData.set_requestcookie(m->requestcookie());
|
||||
|
||||
switch (itype)
|
||||
{
|
||||
case protocol::liBASE:
|
||||
sendLedgerBase(ledger, ledgerData, destinations);
|
||||
sendLedgerBase(ledger, ledgerData);
|
||||
return;
|
||||
|
||||
case protocol::liTX_NODE:
|
||||
@@ -3762,7 +3450,7 @@ PeerImp::processLedgerRequest(std::shared_ptr<protocol::TMGetLedger> const& m, u
|
||||
if (ledgerData.nodes_size() == 0)
|
||||
return;
|
||||
|
||||
sendToMultiple(ledgerData, destinations);
|
||||
send(std::make_shared<Message>(ledgerData, protocol::mtLEDGER_DATA));
|
||||
}
|
||||
|
||||
// Differential pricing helper. Returns only the *dynamic* component
|
||||
@@ -3861,19 +3549,6 @@ PeerImp::isHighLatency() const
|
||||
return latency_ >= kPeerHighLatency;
|
||||
}
|
||||
|
||||
std::set<std::optional<uint64_t>>
|
||||
PeerImp::releaseRequestCookies(uint256 const& requestHash)
|
||||
{
|
||||
std::set<std::optional<uint64_t>> result;
|
||||
std::lock_guard lock(cookieLock_);
|
||||
if (messageRequestCookies_.contains(requestHash))
|
||||
{
|
||||
std::swap(result, messageRequestCookies_[requestHash]);
|
||||
messageRequestCookies_.erase(requestHash);
|
||||
}
|
||||
return result;
|
||||
};
|
||||
|
||||
void
|
||||
PeerImp::Metrics::addMessage(std::uint64_t bytes)
|
||||
{
|
||||
|
||||
@@ -209,13 +209,6 @@ private:
|
||||
bool ledgerReplayEnabled_ = false;
|
||||
LedgerReplayMsgHandler ledgerReplayMsgHandler_;
|
||||
|
||||
// Track message requests and responses
|
||||
// TODO: Use an expiring cache or something
|
||||
using MessageCookieMap = std::map<uint256, std::set<std::optional<uint64_t>>>;
|
||||
using PeerCookieMap = std::map<std::shared_ptr<Peer>, std::set<std::optional<uint64_t>>>;
|
||||
std::mutex mutable cookieLock_;
|
||||
MessageCookieMap messageRequestCookies_;
|
||||
|
||||
friend class OverlayImpl;
|
||||
|
||||
class Metrics
|
||||
@@ -459,13 +452,6 @@ public:
|
||||
return txReduceRelayEnabled_;
|
||||
}
|
||||
|
||||
//
|
||||
// Messages
|
||||
//
|
||||
|
||||
std::set<std::optional<uint64_t>>
|
||||
releaseRequestCookies(uint256 const& requestHash) override;
|
||||
|
||||
private:
|
||||
void
|
||||
close();
|
||||
@@ -664,22 +650,16 @@ private:
|
||||
std::shared_ptr<protocol::TMValidation> const& packet);
|
||||
|
||||
void
|
||||
sendLedgerBase(
|
||||
std::shared_ptr<Ledger const> const& ledger,
|
||||
protocol::TMLedgerData& ledgerData,
|
||||
PeerCookieMap const& destinations);
|
||||
|
||||
void
|
||||
sendToMultiple(protocol::TMLedgerData& ledgerData, PeerCookieMap const& destinations);
|
||||
sendLedgerBase(std::shared_ptr<Ledger const> const& ledger, protocol::TMLedgerData& ledgerData);
|
||||
|
||||
std::shared_ptr<Ledger const>
|
||||
getLedger(std::shared_ptr<protocol::TMGetLedger> const& m, uint256 const& mHash);
|
||||
getLedger(std::shared_ptr<protocol::TMGetLedger> const& m);
|
||||
|
||||
std::shared_ptr<SHAMap const>
|
||||
getTxSet(std::shared_ptr<protocol::TMGetLedger> const& m, uint256 const& mHash) const;
|
||||
getTxSet(std::shared_ptr<protocol::TMGetLedger> const& m) const;
|
||||
|
||||
void
|
||||
processLedgerRequest(std::shared_ptr<protocol::TMGetLedger> const& m, uint256 const& mHash);
|
||||
processLedgerRequest(std::shared_ptr<protocol::TMGetLedger> const& m);
|
||||
|
||||
protected:
|
||||
// Kept `protected` so test subclasses (see
|
||||
|
||||
@@ -7,9 +7,6 @@
|
||||
|
||||
#include <xrpl/basics/Log.h>
|
||||
#include <xrpl/beast/utility/Journal.h>
|
||||
#include <xrpl/core/HashRouter.h>
|
||||
#include <xrpl/core/JobQueue.h>
|
||||
#include <xrpl/protocol/digest.h>
|
||||
|
||||
#include <google/protobuf/message.h>
|
||||
|
||||
@@ -100,45 +97,16 @@ PeerSetImpl::sendRequest(
|
||||
std::shared_ptr<Peer> const& peer)
|
||||
{
|
||||
auto packet = std::make_shared<Message>(message, type);
|
||||
|
||||
auto const messageHash = [&]() {
|
||||
auto const packetBuffer = packet->getBuffer(compression::Compressed::Off);
|
||||
return sha512Half(Slice(packetBuffer.data(), packetBuffer.size()));
|
||||
}();
|
||||
|
||||
// Allow messages to be re-sent to the same peer after a delay
|
||||
using namespace std::chrono_literals;
|
||||
constexpr std::chrono::seconds interval = 30s;
|
||||
|
||||
if (peer)
|
||||
{
|
||||
if (app_.getHashRouter().shouldProcessForPeer(messageHash, peer->id(), interval))
|
||||
{
|
||||
JLOG(journal_.trace()) << "Sending " << protocolMessageName(type) << " message to ["
|
||||
<< peer->id() << "]: " << messageHash;
|
||||
peer->send(packet);
|
||||
}
|
||||
else
|
||||
JLOG(journal_.debug()) << "Suppressing sending duplicate " << protocolMessageName(type)
|
||||
<< " message to [" << peer->id() << "]: " << messageHash;
|
||||
peer->send(packet);
|
||||
return;
|
||||
}
|
||||
|
||||
for (auto id : peers_)
|
||||
{
|
||||
if (auto p = app_.getOverlay().findPeerByShortID(id))
|
||||
{
|
||||
if (app_.getHashRouter().shouldProcessForPeer(messageHash, p->id(), interval))
|
||||
{
|
||||
JLOG(journal_.trace()) << "Sending " << protocolMessageName(type) << " message to ["
|
||||
<< p->id() << "]: " << messageHash;
|
||||
p->send(packet);
|
||||
}
|
||||
else
|
||||
JLOG(journal_.debug())
|
||||
<< "Suppressing sending duplicate " << protocolMessageName(type)
|
||||
<< " message to [" << p->id() << "]: " << messageHash;
|
||||
}
|
||||
p->send(packet);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -30,12 +30,6 @@ protocolMessageType(protocol::TMGetLedger const&)
|
||||
return protocol::mtGET_LEDGER;
|
||||
}
|
||||
|
||||
inline protocol::MessageType
|
||||
protocolMessageType(protocol::TMLedgerData const&)
|
||||
{
|
||||
return protocol::mtLEDGER_DATA;
|
||||
}
|
||||
|
||||
inline protocol::MessageType
|
||||
protocolMessageType(protocol::TMReplayDeltaRequest const&)
|
||||
{
|
||||
@@ -442,63 +436,3 @@ invokeProtocolMessage(Buffers const& buffers, Handler& handler, std::size_t& hin
|
||||
}
|
||||
|
||||
} // namespace xrpl
|
||||
|
||||
namespace protocol {
|
||||
|
||||
template <class Hasher>
|
||||
void
|
||||
hash_append(Hasher& h, TMGetLedger const& msg)
|
||||
{
|
||||
using beast::hash_append;
|
||||
using namespace xrpl;
|
||||
hash_append(h, safeCast<int>(protocolMessageType(msg)));
|
||||
hash_append(h, safeCast<int>(msg.itype()));
|
||||
if (msg.has_ltype())
|
||||
hash_append(h, safeCast<int>(msg.ltype()));
|
||||
|
||||
if (msg.has_ledgerhash())
|
||||
hash_append(h, msg.ledgerhash());
|
||||
|
||||
if (msg.has_ledgerseq())
|
||||
hash_append(h, msg.ledgerseq());
|
||||
|
||||
for (auto const& nodeId : msg.nodeids())
|
||||
hash_append(h, nodeId);
|
||||
hash_append(h, msg.nodeids_size());
|
||||
|
||||
// Do NOT include the request cookie. It does not affect the content of the
|
||||
// request, but only where to route the results.
|
||||
// if (msg.has_requestcookie())
|
||||
// hash_append(h, msg.requestcookie());
|
||||
|
||||
if (msg.has_querytype())
|
||||
hash_append(h, safeCast<int>(msg.querytype()));
|
||||
|
||||
if (msg.has_querydepth())
|
||||
hash_append(h, msg.querydepth());
|
||||
}
|
||||
|
||||
template <class Hasher>
|
||||
void
|
||||
hash_append(Hasher& h, TMLedgerData const& msg)
|
||||
{
|
||||
using beast::hash_append;
|
||||
using namespace xrpl;
|
||||
hash_append(h, safeCast<int>(protocolMessageType(msg)));
|
||||
hash_append(h, msg.ledgerhash());
|
||||
hash_append(h, msg.ledgerseq());
|
||||
hash_append(h, safeCast<int>(msg.type()));
|
||||
for (auto const& node : msg.nodes())
|
||||
{
|
||||
hash_append(h, node.nodedata());
|
||||
if (node.has_nodeid())
|
||||
hash_append(h, node.nodeid());
|
||||
}
|
||||
hash_append(h, msg.nodes_size());
|
||||
if (msg.has_requestcookie())
|
||||
hash_append(h, msg.requestcookie());
|
||||
if (msg.has_error())
|
||||
hash_append(h, safeCast<int>(msg.error()));
|
||||
}
|
||||
|
||||
} // namespace protocol
|
||||
|
||||
@@ -28,8 +28,6 @@ namespace xrpl {
|
||||
constexpr ProtocolVersion const kSupportedProtocolList[]{
|
||||
{2, 1},
|
||||
{2, 2},
|
||||
// Adds TMLedgerData::responseCookies and directResponse
|
||||
{2, 3},
|
||||
};
|
||||
|
||||
// This ugly construct ensures that supportedProtocolList is sorted in strictly
|
||||
|
||||
Reference in New Issue
Block a user