Compare commits

..

16 Commits

Author SHA1 Message Date
Ed Hennis
613a94645d Merge branch 'develop' into ximinez/lending-sendmulti 2026-03-06 13:55:14 -04:00
Ed Hennis
c0323540f9 Merge branch 'develop' into ximinez/lending-sendmulti 2026-03-04 17:12:17 -04:00
Ed Hennis
e0fd480ae7 Merge branch 'develop' into ximinez/lending-sendmulti 2026-03-03 18:38:33 -04:00
Ed Hennis
fe80f0e895 Merge branch 'develop' into ximinez/lending-sendmulti 2026-02-24 17:43:52 -04:00
Ed Hennis
9988e596e9 Merge remote-tracking branch 'XRPLF/develop' into ximinez/lending-sendmulti
* XRPLF/develop:
  ci: [DEPENDABOT] bump actions/upload-artifact from 4.6.2 to 6.0.0 (6396)
  ci: [DEPENDABOT] bump actions/checkout from 4.3.0 to 6.0.2 (6397)
  ci: [DEPENDABOT] bump actions/setup-python from 5.6.0 to 6.2.0 (6395)
  ci: [DEPENDABOT] bump tj-actions/changed-files from 46.0.5 to 47.0.4 (6394)
  ci: [DEPENDABOT] bump codecov/codecov-action from 5.4.3 to 5.5.2 (6398)
  ci: Build docs in PRs and in private repos (6400)
  ci: Add dependabot config (6379)
  Fix tautological assertion (6393)
2026-02-20 17:56:41 -05:00
Ed Hennis
3523c437a8 Merge commit '2c1fad1023' into ximinez/lending-sendmulti
* commit '2c1fad1023':
  chore: Apply clang-format width 100 (6387)
2026-02-20 17:56:19 -05:00
Ed Hennis
0f38b4b541 Update formatting 2026-02-20 17:52:11 -05:00
Ed Hennis
f84350c61c Merge commit '25cca465538a56cce501477f9e5e2c1c7ea2d84c' into ximinez/lending-sendmulti
* commit '25cca465538a56cce501477f9e5e2c1c7ea2d84c':
  chore: Set clang-format width to 100 in config file (6387)
2026-02-20 17:51:51 -05:00
Ed Hennis
7a118245f7 Merge branch 'develop' into ximinez/lending-sendmulti 2026-02-19 16:25:29 -05:00
Ed Hennis
47ddc34fda Merge branch 'develop' into ximinez/lending-sendmulti 2026-02-18 21:15:30 -04:00
Ed Hennis
1f579efc2f Merge branch 'develop' into ximinez/lending-sendmulti 2026-02-04 17:18:43 -04:00
Ed Hennis
e464e101be Fix formatting 2026-01-28 19:23:26 -05:00
Ed Hennis
4dfa6db32a Merge branch 'develop' into ximinez/lending-sendmulti 2026-01-28 19:33:14 -04:00
Ed Hennis
766124ed6d Merge commit '5f638f55536def0d88b970d1018a465a238e55f4' into ximinez/lending-sendmulti
* commit '5f638f55536def0d88b970d1018a465a238e55f4':
  chore: Set ColumnLimit to 120 in clang-format (6288)
2026-01-28 18:32:07 -05:00
Ed Hennis
5c34a7b8fb Merge commit '92046785d1fea5f9efe5a770d636792ea6cab78b' into ximinez/lending-sendmulti
* commit '92046785d1fea5f9efe5a770d636792ea6cab78b':
  test: Fix the `xrpl.net` unit test using async read (6241)
  ci: Upload Conan recipes for develop, release candidates, and releases (6286)
  fix: Stop embedded tests from hanging on ARM by using `atomic_flag` (6248)
  fix:  Remove DEFAULT fields that change to the default in associateAsset (6259) (6273)
  refactor: Update Boost to 1.90 (6280)
  refactor: clean up uses of `std::source_location` (6272)
  ci: Pass missing sanitizers input to actions (6266)
  ci: Properly propagate Conan credentials (6265)
  ci: Explicitly set version when exporting the Conan recipe (6264)
  ci: Use plus instead of hyphen for Conan recipe version suffix (6261)
  chore: Detect uninitialized variables in CMake files (6247)
  ci: Run on-trigger and on-pr when generate-version is modified (6257)
  refactor: Enforce 15-char limit and simplify labels for thread naming (6212)
  docs: Update Ripple Bug Bounty public key (6258)
  ci: Add missing commit hash to Conan recipe version (6256)
  fix: Include `<functional>` header in `Number.h` (6254)
  ci: Upload Conan recipe for merges into develop and commits to release (6235)
  Limit reply size on `TMGetObjectByHash` queries (6110)
  ci: remove 'master' branch as a trigger (6234)
  Improve ledger_entry lookups for fee, amendments, NUNL, and hashes (5644)
2026-01-28 18:31:27 -05:00
Ed Hennis
f6f3542b7e Refactor the sendMulti functions into a common method with callbacks 2026-01-15 12:01:23 -05:00
16 changed files with 298 additions and 544 deletions

View File

@@ -1,139 +0,0 @@
//------------------------------------------------------------------------------
/*
This file is part of rippled: https://github.com/ripple/rippled
Copyright (c) 2024 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.
*/
//==============================================================================
#ifndef RIPPLE_BASICS_CANPROCESS_H_INCLUDED
#define RIPPLE_BASICS_CANPROCESS_H_INCLUDED
#include <functional>
#include <mutex>
#include <set>
/** RAII class to check if an Item is already being processed on another thread,
* as indicated by it's presence in a Collection.
*
* If the Item is not in the Collection, it will be added under lock in the
* ctor, and removed under lock in the dtor. The object will be considered
* "usable" and evaluate to `true`.
*
* If the Item is in the Collection, no changes will be made to the collection,
* and the CanProcess object will be considered "unusable".
*
* It's up to the caller to decide what "usable" and "unusable" mean. (e.g.
* Process or skip a block of code, or set a flag.)
*
* The current use is to avoid lock contention that would be involved in
* processing something associated with the Item.
*
* Examples:
*
* void IncomingLedgers::acquireAsync(LedgerHash const& hash, ...)
* {
* if (CanProcess check{acquiresMutex_, pendingAcquires_, hash})
* {
* acquire(hash, ...);
* }
* }
*
* bool
* NetworkOPsImp::recvValidation(
* std::shared_ptr<STValidation> const& val,
* std::string const& source)
* {
* CanProcess check(
* validationsMutex_, pendingValidations_, val->getLedgerHash());
* BypassAccept bypassAccept =
* check ? BypassAccept::no : BypassAccept::yes;
* handleNewValidation(app_, val, source, bypassAccept, m_journal);
* }
*
*/
class CanProcess
{
public:
template <class Mutex, class Collection, class Item>
CanProcess(Mutex& mtx, Collection& collection, Item const& item)
: cleanup_(insert(mtx, collection, item))
{
}
~CanProcess()
{
if (cleanup_)
cleanup_();
}
CanProcess(CanProcess const&) = delete;
CanProcess&
operator=(CanProcess const&) = delete;
explicit
operator bool() const
{
return static_cast<bool>(cleanup_);
}
private:
template <bool useIterator, class Mutex, class Collection, class Item>
std::function<void()>
doInsert(Mutex& mtx, Collection& collection, Item const& item)
{
std::unique_lock<Mutex> lock(mtx);
// TODO: Use structured binding once LLVM 16 is the minimum supported
// version. See also: https://github.com/llvm/llvm-project/issues/48582
// https://github.com/llvm/llvm-project/commit/127bf44385424891eb04cff8e52d3f157fc2cb7c
auto const insertResult = collection.insert(item);
auto const it = insertResult.first;
if (!insertResult.second)
return {};
if constexpr (useIterator)
return [&, it]() {
std::unique_lock<Mutex> lock(mtx);
collection.erase(it);
};
else
return [&]() {
std::unique_lock<Mutex> lock(mtx);
collection.erase(item);
};
}
// Generic insert() function doesn't use iterators because they may get
// invalidated
template <class Mutex, class Collection, class Item>
std::function<void()>
insert(Mutex& mtx, Collection& collection, Item const& item)
{
return doInsert<false>(mtx, collection, item);
}
// Specialize insert() for std::set, which does not invalidate iterators for
// insert and erase
template <class Mutex, class Item>
std::function<void()>
insert(Mutex& mtx, std::set<Item>& collection, Item const& item)
{
return doInsert<true>(mtx, collection, item);
}
// If set, then the item is "usable"
std::function<void()> cleanup_;
};
#endif

View File

@@ -199,7 +199,7 @@ public:
/** Add a suppression peer and get message's relay status.
* Return pair:
* element 1: true if the key is added.
* 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>>

View File

@@ -35,8 +35,6 @@ struct LedgerHeader
// If validated is false, it means "not yet validated."
// Once validated is true, it will never be set false at a later time.
// NOTE: If you are accessing this directly, you are probably doing it
// wrong. Use LedgerMaster::isValidated().
// VFALCO TODO Make this not mutable
bool mutable validated = false;
bool accepted = false;

View File

@@ -185,7 +185,7 @@ public:
virtual bool
isFull() = 0;
virtual void
setMode(OperatingMode om, char const* reason) = 0;
setMode(OperatingMode om) = 0;
virtual bool
isBlocked() = 0;
virtual bool

View File

@@ -2020,6 +2020,99 @@ rippleSendIOU(
return terResult;
}
template <class TAsset>
static TER
doSendMulti(
std::string const& name,
ApplyView& view,
AccountID const& senderID,
TAsset const& issue,
MultiplePaymentDestinations const& receivers,
STAmount& actual,
beast::Journal j,
WaiveTransferFee waiveFee,
// Don't pass back parameters that the caller already has
std::function<
TER(AccountID const& senderID,
AccountID const& receiverID,
STAmount const& amount,
bool checkIssuer)> doCredit,
std::function<
TER(AccountID const& issuer, STAmount const& takeFromSender, STAmount const& amount)>
preMint = {})
{
// Use the same pattern for all the SendMulti functions to help avoid
// divergence and copy/paste errors.
auto const& issuer = issue.getIssuer();
// These values may not stay in sync
STAmount takeFromSender{issue};
actual = takeFromSender;
// Failures return immediately.
for (auto const& r : receivers)
{
auto const& receiverID = r.first;
STAmount amount{issue, r.second};
if (amount < beast::zero)
{
return tecINTERNAL; // LCOV_EXCL_LINE
}
/* If we aren't sending anything or if the sender is the same as the
* receiver then we don't need to do anything.
*/
if (!amount || (senderID == receiverID))
continue;
using namespace std::string_literals;
XRPL_ASSERT(!isXRP(receiverID), ("xrpl::"s + name + " : receiver is not XRP").c_str());
if (senderID == issuer || receiverID == issuer || issuer == noAccount())
{
if (preMint)
{
if (auto const ter = preMint(issuer, takeFromSender, amount))
return ter;
}
// Direct send: redeeming IOUs and/or sending own IOUs.
if (auto const ter = doCredit(senderID, receiverID, amount, false))
return ter;
actual += amount;
// Do not add amount to takeFromSender, because doCredit took
// it.
continue;
}
// Sending 3rd party: transit.
// Calculate the amount to transfer accounting
// for any transfer fees if the fee is not waived:
STAmount actualSend = (waiveFee == WaiveTransferFee::Yes || issue.native())
? amount
: multiply(amount, transferRate(view, amount));
actual += actualSend;
takeFromSender += actualSend;
JLOG(j.debug()) << name << "> " << to_string(senderID) << " - > " << to_string(receiverID)
<< " : deliver=" << amount.getFullText()
<< " cost=" << actualSend.getFullText();
if (TER const terResult = doCredit(issuer, receiverID, amount, true))
return terResult;
}
if (senderID != issuer && takeFromSender)
{
if (TER const terResult = doCredit(senderID, issuer, takeFromSender, true))
return terResult;
}
return tesSUCCESS;
}
// Send regardless of limits.
// --> receivers: Amount/currency/issuer to deliver to receivers.
// <-- saActual: Amount actually cost to sender. Sender pays fees.
@@ -2033,65 +2126,18 @@ rippleSendMultiIOU(
beast::Journal j,
WaiveTransferFee waiveFee)
{
auto const& issuer = issue.getIssuer();
XRPL_ASSERT(!isXRP(senderID), "xrpl::rippleSendMultiIOU : sender is not XRP");
// These may diverge
STAmount takeFromSender{issue};
actual = takeFromSender;
auto doCredit = [&view, j](
AccountID const& senderID,
AccountID const& receiverID,
STAmount const& amount,
bool checkIssuer) {
return rippleCreditIOU(view, senderID, receiverID, amount, checkIssuer, j);
};
// Failures return immediately.
for (auto const& r : receivers)
{
auto const& receiverID = r.first;
STAmount amount{issue, r.second};
/* If we aren't sending anything or if the sender is the same as the
* receiver then we don't need to do anything.
*/
if (!amount || (senderID == receiverID))
continue;
XRPL_ASSERT(!isXRP(receiverID), "xrpl::rippleSendMultiIOU : receiver is not XRP");
if (senderID == issuer || receiverID == issuer || issuer == noAccount())
{
// Direct send: redeeming IOUs and/or sending own IOUs.
if (auto const ter = rippleCreditIOU(view, senderID, receiverID, amount, false, j))
return ter;
actual += amount;
// Do not add amount to takeFromSender, because rippleCreditIOU took
// it.
continue;
}
// Sending 3rd party IOUs: transit.
// Calculate the amount to transfer accounting
// for any transfer fees if the fee is not waived:
STAmount actualSend = (waiveFee == WaiveTransferFee::Yes)
? amount
: multiply(amount, transferRate(view, issuer));
actual += actualSend;
takeFromSender += actualSend;
JLOG(j.debug()) << "rippleSendMultiIOU> " << to_string(senderID) << " - > "
<< to_string(receiverID) << " : deliver=" << amount.getFullText()
<< " cost=" << actual.getFullText();
if (TER const terResult = rippleCreditIOU(view, issuer, receiverID, amount, true, j))
return terResult;
}
if (senderID != issuer && takeFromSender)
{
if (TER const terResult = rippleCreditIOU(view, senderID, issuer, takeFromSender, true, j))
return terResult;
}
return tesSUCCESS;
return doSendMulti(
"rippleSendMultiIOU", view, senderID, issue, receivers, actual, j, waiveFee, doCredit);
}
static TER
@@ -2225,9 +2271,9 @@ accountSendMultiIOU(
XRPL_ASSERT_PARTS(
receivers.size() > 1, "xrpl::accountSendMultiIOU", "multiple recipients provided");
STAmount actual;
if (!issue.native())
{
STAmount actual;
JLOG(j.trace()) << "accountSendMultiIOU: " << to_string(senderID) << " sending "
<< receivers.size() << " IOUs";
@@ -2254,6 +2300,85 @@ accountSendMultiIOU(
<< receivers.size() << " receivers.";
}
auto doCredit = [&view, &sender, &receivers, j](
AccountID const& senderID,
AccountID const& receiverID,
STAmount const& amount,
bool /*checkIssuer*/) -> TER {
if (!senderID)
{
SLE::pointer receiver =
receiverID != beast::zero ? view.peek(keylet::account(receiverID)) : SLE::pointer();
if (auto stream = j.trace())
{
std::string receiver_bal("-");
if (receiver)
receiver_bal = receiver->getFieldAmount(sfBalance).getFullText();
stream << "accountSendMultiIOU> " << to_string(senderID) << " -> "
<< to_string(receiverID) << " (" << receiver_bal
<< ") : " << amount.getFullText();
}
if (receiver)
{
// Increment XRP balance.
auto const rcvBal = receiver->getFieldAmount(sfBalance);
receiver->setFieldAmount(sfBalance, rcvBal + amount);
view.creditHook(xrpAccount(), receiverID, amount, -rcvBal);
view.update(receiver);
}
if (auto stream = j.trace())
{
std::string receiver_bal("-");
if (receiver)
receiver_bal = receiver->getFieldAmount(sfBalance).getFullText();
stream << "accountSendMultiIOU< " << to_string(senderID) << " -> "
<< to_string(receiverID) << " (" << receiver_bal
<< ") : " << amount.getFullText();
}
return tesSUCCESS;
}
// Sender
if (sender)
{
if (sender->getFieldAmount(sfBalance) < amount)
{
return TER{tecFAILED_PROCESSING};
}
else
{
auto const sndBal = sender->getFieldAmount(sfBalance);
view.creditHook(senderID, xrpAccount(), amount, sndBal);
// Decrement XRP balance.
sender->setFieldAmount(sfBalance, sndBal - amount);
view.update(sender);
}
}
if (auto stream = j.trace())
{
std::string sender_bal("-");
if (sender)
sender_bal = sender->getFieldAmount(sfBalance).getFullText();
stream << "accountSendMultiIOU< " << to_string(senderID) << " (" << sender_bal
<< ") -> " << receivers.size() << " receivers.";
}
return tesSUCCESS;
};
return doSendMulti(
"accountSendMultiIOU", view, senderID, issue, receivers, actual, j, waiveFee, doCredit);
// Failures return immediately.
STAmount takeFromSender{issue};
for (auto const& r : receivers)
@@ -2472,82 +2597,47 @@ rippleSendMultiMPT(
beast::Journal j,
WaiveTransferFee waiveFee)
{
// Safe to get MPT since rippleSendMultiMPT is only called by
// accountSendMultiMPT
auto const& issuer = mptIssue.getIssuer();
auto const sle = view.read(keylet::mptIssuance(mptIssue.getMptID()));
if (!sle)
return tecOBJECT_NOT_FOUND;
// These may diverge
STAmount takeFromSender{mptIssue};
actual = takeFromSender;
for (auto const& r : receivers)
{
auto const& receiverID = r.first;
STAmount amount{mptIssue, r.second};
if (amount < beast::zero)
auto preMint = [&](AccountID const& issuer,
STAmount const& takeFromSender,
STAmount const& amount) -> TER {
// if sender is issuer, check that the new OutstandingAmount will
// not exceed MaximumAmount
if (senderID == issuer)
{
return tecINTERNAL; // LCOV_EXCL_LINE
XRPL_ASSERT_PARTS(
takeFromSender == beast::zero,
"rippler::rippleSendMultiMPT",
"sender == issuer, takeFromSender == zero");
auto const sendAmount = amount.mpt().value();
auto const maximumAmount = sle->at(~sfMaximumAmount).value_or(maxMPTokenAmount);
if (sendAmount > maximumAmount ||
sle->getFieldU64(sfOutstandingAmount) > maximumAmount - sendAmount)
return tecPATH_DRY;
}
/* If we aren't sending anything or if the sender is the same as the
* receiver then we don't need to do anything.
*/
if (!amount || (senderID == receiverID))
continue;
return tesSUCCESS;
};
auto doCredit =
[&view, j](
AccountID const& senderID, AccountID const& receiverID, STAmount const& amount, bool) {
return rippleCreditMPT(view, senderID, receiverID, amount, j);
};
if (senderID == issuer || receiverID == issuer)
{
// if sender is issuer, check that the new OutstandingAmount will
// not exceed MaximumAmount
if (senderID == issuer)
{
XRPL_ASSERT_PARTS(
takeFromSender == beast::zero,
"rippler::rippleSendMultiMPT",
"sender == issuer, takeFromSender == zero");
auto const sendAmount = amount.mpt().value();
auto const maximumAmount = sle->at(~sfMaximumAmount).value_or(maxMPTokenAmount);
if (sendAmount > maximumAmount ||
sle->getFieldU64(sfOutstandingAmount) > maximumAmount - sendAmount)
return tecPATH_DRY;
}
// Direct send: redeeming MPTs and/or sending own MPTs.
if (auto const ter = rippleCreditMPT(view, senderID, receiverID, amount, j))
return ter;
actual += amount;
// Do not add amount to takeFromSender, because rippleCreditMPT took
// it
continue;
}
// Sending 3rd party MPTs: transit.
STAmount actualSend = (waiveFee == WaiveTransferFee::Yes)
? amount
: multiply(amount, transferRate(view, amount.get<MPTIssue>().getMptID()));
actual += actualSend;
takeFromSender += actualSend;
JLOG(j.debug()) << "rippleSendMultiMPT> " << to_string(senderID) << " - > "
<< to_string(receiverID) << " : deliver=" << amount.getFullText()
<< " cost=" << actualSend.getFullText();
if (auto const terResult = rippleCreditMPT(view, issuer, receiverID, amount, j))
return terResult;
}
if (senderID != issuer && takeFromSender)
{
if (TER const terResult = rippleCreditMPT(view, senderID, issuer, takeFromSender, j))
return terResult;
}
return tesSUCCESS;
return doSendMulti(
"rippleSendMultiMPT",
view,
senderID,
mptIssue,
receivers,
actual,
j,
waiveFee,
doCredit,
preMint);
}
static TER

View File

@@ -85,12 +85,7 @@ public:
}
virtual void
acquireAsync(
JobType type,
std::string const& name,
uint256 const& hash,
std::uint32_t seq,
InboundLedger::Reason reason) override
acquireAsync(uint256 const& hash, std::uint32_t seq, InboundLedger::Reason reason) override
{
}

View File

@@ -1,165 +0,0 @@
//------------------------------------------------------------------------------
/*
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

View File

@@ -107,8 +107,10 @@ RCLConsensus::Adaptor::acquireLedger(LedgerHash const& hash)
// Tell the ledger acquire system that we need the consensus ledger
acquiringLedger_ = hash;
app_.getInboundLedgers().acquireAsync(
jtADVANCE, "GetConsL1", hash, 0, InboundLedger::Reason::CONSENSUS);
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);
});
}
return std::nullopt;
}
@@ -983,7 +985,7 @@ void
RCLConsensus::Adaptor::updateOperatingMode(std::size_t const positions) const
{
if (!positions && app_.getOPs().isFull())
app_.getOPs().setMode(OperatingMode::CONNECTED, "updateOperatingMode: no positions");
app_.getOPs().setMode(OperatingMode::CONNECTED);
}
void

View File

@@ -117,8 +117,12 @@ RCLValidationsAdaptor::acquire(LedgerHash const& hash)
{
JLOG(j_.warn()) << "Need validated ledger for preferred ledger analysis " << hash;
app_.getInboundLedgers().acquireAsync(
jtADVANCE, "GetConsL2", hash, 0, InboundLedger::Reason::CONSENSUS);
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);
});
return std::nullopt;
}

View File

@@ -26,12 +26,7 @@ public:
// Queue. TODO review whether all callers of acquire() can use this
// instead. Inbound ledger acquisition is asynchronous anyway.
virtual void
acquireAsync(
JobType type,
std::string const& name,
uint256 const& hash,
std::uint32_t seq,
InboundLedger::Reason reason) = 0;
acquireAsync(uint256 const& hash, std::uint32_t seq, InboundLedger::Reason reason) = 0;
virtual std::shared_ptr<InboundLedger>
find(LedgerHash const& hash) = 0;

View File

@@ -353,14 +353,7 @@ InboundLedger::onTimer(bool wasProgress, ScopedLockType&)
if (!wasProgress)
{
if (checkLocal())
{
// Done. Something else (probably consensus) built the ledger
// locally while waiting for data (or possibly before requesting)
XRPL_ASSERT(isDone(), "ripple::InboundLedger::onTimer : done");
JLOG(journal_.info()) << "Finished while waiting " << hash_;
return;
}
checkLocal();
mByHash = true;

View File

@@ -2,9 +2,9 @@
#include <xrpld/app/ledger/LedgerMaster.h>
#include <xrpld/app/main/Application.h>
#include <xrpl/basics/CanProcess.h>
#include <xrpl/basics/DecayingSample.h>
#include <xrpl/basics/Log.h>
#include <xrpl/basics/scope.h>
#include <xrpl/beast/container/aged_map.h>
#include <xrpl/core/JobQueue.h>
#include <xrpl/core/PerfLog.h>
@@ -59,15 +59,12 @@ public:
(reason != InboundLedger::Reason::CONSENSUS))
return {};
std::stringstream ss;
bool isNew = true;
std::shared_ptr<InboundLedger> inbound;
{
ScopedLockType sl(mLock);
if (stopping_)
{
JLOG(j_.debug()) << "Abort(stopping): " << ss.str();
return {};
}
@@ -86,61 +83,47 @@ public:
++mCounter;
}
}
ss << " IsNew: " << (isNew ? "true" : "false");
if (inbound->isFailed())
{
JLOG(j_.debug()) << "Abort(failed): " << ss.str();
return {};
}
if (!isNew)
inbound->update(seq);
if (!inbound->isComplete())
{
JLOG(j_.debug()) << "InProgress: " << ss.str();
return {};
}
JLOG(j_.debug()) << "Complete: " << ss.str();
return inbound->getLedger();
};
using namespace std::chrono_literals;
return perf::measureDurationAndLog(doAcquire, "InboundLedgersImp::acquire", 500ms, j_);
std::shared_ptr<Ledger const> ledger =
perf::measureDurationAndLog(doAcquire, "InboundLedgersImp::acquire", 500ms, j_);
return ledger;
}
void
acquireAsync(
JobType type,
std::string const& name,
uint256 const& hash,
std::uint32_t seq,
InboundLedger::Reason reason) override
acquireAsync(uint256 const& hash, std::uint32_t seq, InboundLedger::Reason reason) override
{
if (auto check = std::make_shared<CanProcess const>(acquiresMutex_, pendingAcquires_, hash);
*check)
std::unique_lock lock(acquiresMutex_);
try
{
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;
}
});
if (pendingAcquires_.contains(hash))
return;
pendingAcquires_.insert(hash);
scope_unlock unlock(lock);
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;
}
pendingAcquires_.erase(hash);
}
std::shared_ptr<InboundLedger>

View File

@@ -907,9 +907,8 @@ LedgerMaster::checkAccept(std::shared_ptr<Ledger const> const& ledger)
return;
}
JLOG(m_journal.info()) << "Advancing accepted ledger to " << ledger->header().seq << " ("
<< to_short_string(ledger->header().hash) << ") with >= " << minVal
<< " validations";
JLOG(m_journal.info()) << "Advancing accepted ledger to " << ledger->header().seq
<< " with >= " << minVal << " validations";
ledger->setValidated();
ledger->setFull();

View File

@@ -13,8 +13,7 @@ TimeoutCounter::TimeoutCounter(
QueueJobParameter&& jobParameter,
beast::Journal journal)
: app_(app)
, sink_(journal, to_short_string(hash) + " ")
, journal_(sink_)
, journal_(journal)
, hash_(hash)
, timeouts_(0)
, complete_(false)
@@ -34,7 +33,6 @@ TimeoutCounter::setTimer(ScopedLockType& sl)
{
if (isDone())
return;
JLOG(journal_.debug()) << "Setting timer for " << timerInterval_.count() << "ms";
timer_.expires_after(timerInterval_);
timer_.async_wait([wptr = pmDowncast()](boost::system::error_code const& ec) {
if (ec == boost::asio::error::operation_aborted)
@@ -42,10 +40,6 @@ TimeoutCounter::setTimer(ScopedLockType& sl)
if (auto ptr = wptr.lock())
{
JLOG(ptr->journal_.debug())
<< "timer: ec: " << ec
<< " (operation_aborted: " << boost::asio::error::operation_aborted << " - "
<< (ec == boost::asio::error::operation_aborted ? "aborted" : "other") << ")";
ScopedLockType sl(ptr->mtx_);
ptr->queueJob(sl);
}

View File

@@ -3,7 +3,6 @@
#include <xrpld/app/main/Application.h>
#include <xrpl/beast/utility/Journal.h>
#include <xrpl/beast/utility/WrappedSink.h>
#include <xrpl/core/Job.h>
#include <boost/asio/basic_waitable_timer.hpp>
@@ -104,7 +103,6 @@ protected:
// Used in this class for access to boost::asio::io_context and
// xrpl::Overlay. Used in subtypes for the kitchen sink.
Application& app_;
beast::WrappedSink sink_;
beast::Journal journal_;
mutable std::recursive_mutex mtx_;

View File

@@ -30,10 +30,10 @@
#include <xrpld/rpc/MPTokenIssuanceID.h>
#include <xrpld/rpc/ServerHandler.h>
#include <xrpl/basics/CanProcess.h>
#include <xrpl/basics/UptimeClock.h>
#include <xrpl/basics/mulDiv.h>
#include <xrpl/basics/safe_cast.h>
#include <xrpl/basics/scope.h>
#include <xrpl/beast/utility/rngfill.h>
#include <xrpl/core/HashRouter.h>
#include <xrpl/core/NetworkIDService.h>
@@ -397,7 +397,7 @@ public:
isFull() override;
void
setMode(OperatingMode om, char const* reason) override;
setMode(OperatingMode om) override;
bool
isBlocked() override;
@@ -842,7 +842,7 @@ NetworkOPsImp::strOperatingMode(bool const admin /* = false */) const
inline void
NetworkOPsImp::setStandAlone()
{
setMode(OperatingMode::FULL, "setStandAlone");
setMode(OperatingMode::FULL);
}
inline void
@@ -985,7 +985,7 @@ NetworkOPsImp::processHeartbeatTimer()
{
if (mMode != OperatingMode::DISCONNECTED)
{
setMode(OperatingMode::DISCONNECTED, "Heartbeat: insufficient peers");
setMode(OperatingMode::DISCONNECTED);
std::stringstream ss;
ss << "Node count (" << numPeers << ") has fallen "
<< "below required minimum (" << minPeerCount_ << ").";
@@ -1009,7 +1009,7 @@ NetworkOPsImp::processHeartbeatTimer()
if (mMode == OperatingMode::DISCONNECTED)
{
setMode(OperatingMode::CONNECTED, "Heartbeat: sufficient peers");
setMode(OperatingMode::CONNECTED);
JLOG(m_journal.info()) << "Node count (" << numPeers << ") is sufficient.";
CLOG(clog.ss()) << "setting mode to CONNECTED based on " << numPeers << " peers. ";
}
@@ -1019,9 +1019,9 @@ NetworkOPsImp::processHeartbeatTimer()
auto origMode = mMode.load();
CLOG(clog.ss()) << "mode: " << strOperatingMode(origMode, true);
if (mMode == OperatingMode::SYNCING)
setMode(OperatingMode::SYNCING, "Heartbeat: check syncing");
setMode(OperatingMode::SYNCING);
else if (mMode == OperatingMode::CONNECTED)
setMode(OperatingMode::CONNECTED, "Heartbeat: check connected");
setMode(OperatingMode::CONNECTED);
auto newMode = mMode.load();
if (origMode != newMode)
{
@@ -1711,7 +1711,7 @@ void
NetworkOPsImp::setAmendmentBlocked()
{
amendmentBlocked_ = true;
setMode(OperatingMode::CONNECTED, "setAmendmentBlocked");
setMode(OperatingMode::CONNECTED);
}
inline bool
@@ -1742,7 +1742,7 @@ void
NetworkOPsImp::setUNLBlocked()
{
unlBlocked_ = true;
setMode(OperatingMode::CONNECTED, "setUNLBlocked");
setMode(OperatingMode::CONNECTED);
}
inline void
@@ -1838,7 +1838,7 @@ NetworkOPsImp::checkLastClosedLedger(Overlay::PeerSequence const& peerList, uint
if ((mMode == OperatingMode::TRACKING) || (mMode == OperatingMode::FULL))
{
setMode(OperatingMode::CONNECTED, "check LCL: not on consensus ledger");
setMode(OperatingMode::CONNECTED);
}
if (consensus)
@@ -1923,8 +1923,8 @@ NetworkOPsImp::beginConsensus(
// this shouldn't happen unless we jump ledgers
if (mMode == OperatingMode::FULL)
{
JLOG(m_journal.warn()) << "beginConsensus Don't have LCL, going to tracking";
setMode(OperatingMode::TRACKING, "beginConsensus: No LCL");
JLOG(m_journal.warn()) << "Don't have LCL, going to tracking";
setMode(OperatingMode::TRACKING);
CLOG(clog) << "beginConsensus Don't have LCL, going to tracking. ";
}
@@ -2053,7 +2053,7 @@ NetworkOPsImp::endConsensus(std::unique_ptr<std::stringstream> const& clog)
// validations we have for LCL. If the ledger is good enough, go to
// TRACKING - TODO
if (!needNetworkLedger_)
setMode(OperatingMode::TRACKING, "endConsensus: check tracking");
setMode(OperatingMode::TRACKING);
}
if (((mMode == OperatingMode::CONNECTED) || (mMode == OperatingMode::TRACKING)) &&
@@ -2066,7 +2066,7 @@ NetworkOPsImp::endConsensus(std::unique_ptr<std::stringstream> const& clog)
if (registry_.timeKeeper().now() <
(current->header().parentCloseTime + 2 * current->header().closeTimeResolution))
{
setMode(OperatingMode::FULL, "endConsensus: check full");
setMode(OperatingMode::FULL);
}
}
@@ -2078,7 +2078,7 @@ NetworkOPsImp::consensusViewChange()
{
if ((mMode == OperatingMode::FULL) || (mMode == OperatingMode::TRACKING))
{
setMode(OperatingMode::CONNECTED, "consensusViewChange");
setMode(OperatingMode::CONNECTED);
}
}
@@ -2380,7 +2380,7 @@ NetworkOPsImp::pubPeerStatus(std::function<Json::Value(void)> const& func)
}
void
NetworkOPsImp::setMode(OperatingMode om, char const* reason)
NetworkOPsImp::setMode(OperatingMode om)
{
using namespace std::chrono_literals;
if (om == OperatingMode::CONNECTED)
@@ -2400,12 +2400,11 @@ NetworkOPsImp::setMode(OperatingMode om, char const* reason)
if (mMode == om)
return;
auto const sink = om < mMode ? m_journal.warn() : m_journal.info();
mMode = om;
accounting_.mode(om);
JLOG(sink) << "STATE->" << strOperatingMode() << " - " << reason;
JLOG(m_journal.info()) << "STATE->" << strOperatingMode();
pubServer();
}
@@ -2414,24 +2413,32 @@ NetworkOPsImp::recvValidation(std::shared_ptr<STValidation> const& val, std::str
{
JLOG(m_journal.trace()) << "recvValidation " << val->getLedgerHash() << " from " << source;
std::unique_lock lock(validationsMutex_);
BypassAccept bypassAccept = BypassAccept::no;
try
{
CanProcess const check(validationsMutex_, pendingValidations_, val->getLedgerHash());
try
{
BypassAccept bypassAccept = check ? BypassAccept::no : BypassAccept::yes;
handleNewValidation(registry_.app(), val, source, bypassAccept, m_journal);
}
catch (std::exception const& e)
{
JLOG(m_journal.warn()) << "Exception thrown for handling new validation "
<< val->getLedgerHash() << ": " << e.what();
}
catch (...)
{
JLOG(m_journal.warn())
<< "Unknown exception thrown for handling new validation " << val->getLedgerHash();
}
if (pendingValidations_.contains(val->getLedgerHash()))
bypassAccept = BypassAccept::yes;
else
pendingValidations_.insert(val->getLedgerHash());
scope_unlock unlock(lock);
handleNewValidation(registry_.app(), val, source, bypassAccept, m_journal);
}
catch (std::exception const& e)
{
JLOG(m_journal.warn()) << "Exception thrown for handling new validation "
<< val->getLedgerHash() << ": " << e.what();
}
catch (...)
{
JLOG(m_journal.warn()) << "Unknown exception thrown for handling new validation "
<< val->getLedgerHash();
}
if (bypassAccept == BypassAccept::no)
{
pendingValidations_.erase(val->getLedgerHash());
}
lock.unlock();
pubValidation(val);