ARCHIVE/rippled
1
0
Fork 0
mirror of https://github.com/XRPLF/rippled.git synced 2026-07-27 17:10:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'pratik/std-coro/cleanup-boost-coroutine' into pratik/std-coro/tsan-fixes

Browse Source 
# Conflicts:
#	include/xrpl/core/CoroTaskRunner.ipp
This commit is contained in:
Pratik Mankawde
2026-07-27 10:31:08 +01:00
parent de4d93b3c5 598bda0ebc
commit 9857842179
11 changed files with 171 additions and 2047 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
.cspell.config.yaml
View File

@@ -88,7 +88,6 @@ words:
- coros
- cowid
- cpack
- cppcoro
- cryptocondition
- cryptoconditional
- cryptoconditions
@@ -127,7 +126,6 @@ words:
- fsanitize
- funclets
- Gamal
- gantt
- gcov
- gcovr
- ghead
@@ -171,7 +169,6 @@ words:
- lseq
- lsmf
- ltype
- Mankawde
- mathbunnyru
- mcmodel
- MEMORYSTATUSEX
@@ -233,7 +230,6 @@ words:
- permissioned
- pointee
- populator
- Pratik
- preauth
- preauthorization
- preauthorize

1841
BoostToStdCoroutineSwitchPlan.md
View File

File diff suppressed because it is too large Load Diff

142
BoostToStdCoroutineTaskList.md
View File

@@ -1,142 +0,0 @@
# Boost.Coroutine to C++20 Migration — Task List
> Parent document: [BoostToStdCoroutineSwitchPlan.md](BoostToStdCoroutineSwitchPlan.md)
---
## Milestone 1: New Coroutine Primitives
- [ ] **1.1** Design `CoroTask<T>` class with `promise_type`
- Define `promise_type` with `initial_suspend`, `final_suspend`, `unhandled_exception`, `return_value`/`return_void`
- Implement `FinalAwaiter` for continuation support
- Implement move-only RAII handle wrapper
- Support both `CoroTask<T>` and `CoroTask<void>`
- [ ] **1.2** Design and implement `JobQueueAwaiter`
- `await_suspend()` calls `jq_.addJob(type, name, [h]{ h.resume(); })`
- Handle `addJob()` failure (shutdown) — resume with error flag or throw
- Integrate `nSuspend_` counter increment/decrement
- [ ] **1.3** Implement `LocalValues` swap in new coroutine resume path
- Before `handle.resume()`: save thread-local, install coroutine-local
- After `handle.resume()` returns: restore thread-local
- Ensure this works when coroutine migrates between threads
- [ ] **1.4** Add `postCoroTask()` template to `JobQueue`
- Accept callable returning `CoroTask<void>`
- Schedule initial execution on JobQueue (mirror `postCoro()` behavior)
- Return a handle/shared_ptr for join/cancel
- [ ] **1.5** Write unit tests (`src/test/core/CoroTask_test.cpp`)
- Test `CoroTask<void>` runs to completion
- Test `CoroTask<int>` returns value
- Test exception propagation across co_await
- Test coroutine destruction before completion
- Test `JobQueueAwaiter` schedules on correct thread
- Test `LocalValue` isolation across 4+ coroutines
- Test shutdown rejection (addJob returns false)
- Test `correct_order` equivalent (yield → join → post → complete)
- Test `incorrect_order` equivalent (post → yield → complete)
- Test multiple sequential co_await points
- [ ] **1.6** Verify build on GCC 12+, Clang 16+
- [ ] **1.7** Run ASAN + TSAN on new tests
- [ ] **1.8** Run full `--unittest` suite (no regressions)
- [ ] **1.9** Self-review and create PR #1
---
## Milestone 2: Entry Point Migration
- [ ] **2.1** Migrate `ServerHandler::onRequest()` (`ServerHandler.cpp:287`)
- Replace `m_jobQueue.postCoro(jtCLIENT_RPC, ...)` with `postCoroTask()`
- Update lambda to return `CoroTask<void>` (add `co_return`)
- Update `processSession` to accept new coroutine type
- [ ] **2.2** Migrate `ServerHandler::onWSMessage()` (`ServerHandler.cpp:325`)
- Replace `m_jobQueue.postCoro(jtCLIENT_WEBSOCKET, ...)` with `postCoroTask()`
- Update lambda signature
- [ ] **2.3** Migrate `GRPCServer::CallData::process()` (`GRPCServer.cpp:102`)
- Replace `app_.getJobQueue().postCoro(JobType::jtRPC, ...)` with `postCoroTask()`
- Update `process(shared_ptr<Coro> coro)` overload signature
- [ ] **2.4** Update `RPC::Context` (`Context.h:27`)
- Replace `std::shared_ptr<JobQueue::Coro> coro{}` with new coroutine wrapper type
- Ensure all code that accesses `context.coro` compiles
- [ ] **2.5** Update `ServerHandler.h` signatures
- `processSession()` and `processRequest()` parameter types
- [ ] **2.6** Update `GRPCServer.h` signatures
- `process()` method parameter types
- [ ] **2.7** Run full `--unittest` suite
- [ ] **2.8** Manual smoke test: HTTP + WS + gRPC RPC requests
- [ ] **2.9** Run ASAN + TSAN
- [ ] **2.10** Self-review and create PR #2
---
## Milestone 3: Handler Migration
- [ ] **3.1** Migrate `doRipplePathFind()` (`RipplePathFind.cpp`)
- Replace `context.coro->yield()` with `co_await PathFindAwaiter{...}`
- Replace continuation lambda's `coro->post()` / `coro->resume()` with awaiter scheduling
- Handle shutdown case (post failure) in awaiter
- [ ] **3.2** Create `PathFindAwaiter` (or use generic `JobQueueAwaiter`)
- Encapsulate the continuation + yield pattern from `RipplePathFind.cpp` lines 108-132
- [ ] **3.3** Update `Path_test.cpp`
- Replace `postCoro` usage with `postCoroTask`
- Ensure `context.coro` usage matches new type
- [ ] **3.4** Update `AMMTest.cpp`
- Replace `postCoro` usage with `postCoroTask`
- [ ] **3.5** Rewrite `Coroutine_test.cpp` for new API
- `correct_order`: postCoroTask → co_await → join → resume → complete
- `incorrect_order`: post before yield equivalent
- `thread_specific_storage`: 4 coroutines with LocalValue isolation
- [ ] **3.6** Update `JobQueue_test.cpp` `testPostCoro`
- Migrate to `postCoroTask` API
- [ ] **3.7** Verify `ripple_path_find` works end-to-end with new coroutines
- [ ] **3.8** Test shutdown-during-pathfind scenario
- [ ] **3.9** Run full `--unittest` suite
- [ ] **3.10** Run ASAN + TSAN
- [ ] **3.11** Self-review and create PR #3
---
## Milestone 4: Cleanup & Validation
- [ ] **4.1** Delete `include/xrpl/core/Coro.ipp`
- [ ] **4.2** Remove from `JobQueue.h`:
- `#include <boost/coroutine2/all.hpp>`
- `struct Coro_create_t`
- `class Coro` (entire class)
- `postCoro()` template
- Comment block (lines 322-377) describing old race condition
- [ ] **4.3** Update `cmake/deps/Boost.cmake`:
- Remove `coroutine` from `find_package(Boost REQUIRED COMPONENTS ...)`
- Remove `Boost::coroutine` from `target_link_libraries`
- [ ] **4.4** Update `cmake/XrplInterface.cmake`:
- Remove `BOOST_COROUTINES2_NO_DEPRECATION_WARNING`
- [ ] **4.5** Run memory benchmark
- Create N=1000 coroutines, compare RSS: before vs after
- Document results
- [ ] **4.6** Run context switch benchmark
- 100K yield/resume cycles, compare latency: before vs after
- Document results
- [ ] **4.7** Run RPC throughput benchmark
- Concurrent `ripple_path_find` requests, compare throughput
- Document results
- [ ] **4.8** Run full `--unittest` suite
- [ ] **4.9** Run ASAN, TSAN, UBSan
- Confirm `__asan_handle_no_return` warnings are gone
- [ ] **4.10** Verify build on all supported compilers
- [ ] **4.11** Self-review and create PR #4
- [ ] **4.12** Document final benchmark results in PR description

4
include/xrpl/core/CoroTask.h
View File

@@ -63,7 +63,7 @@ class CoroTask;
* Usage Examples
* ==============
*
* 1. Basic void coroutine (the most common case in rippled):
* 1. Basic void coroutine (the most common case in xrpld):
*
* CoroTask<void> doWork(std::shared_ptr<CoroTaskRunner> runner) {
* // do something
@@ -146,7 +146,7 @@ class CoroTask;
* LIMITATION: Stackless -- cannot suspend from nested non-coroutine calls.
* If a coroutine calls a regular function that wants to "yield", it
* cannot. Only the immediate coroutine body can use co_await.
* This is acceptable for rippled because all yield() sites are shallow.
* This is acceptable for xrpld because all yield() sites are shallow.
*/
template <>
class CoroTask<void>

45
include/xrpl/core/CoroTaskRunner.ipp
View File

@@ -325,12 +325,48 @@ JobQueue::CoroTaskRunner::resume()
XRPL_ASSERT(
task_.handle() && !task_.done(),
"xrpl::JobQueue::CoroTaskRunner::resume : task handle is valid and not done");
task_.handle().resume();
if (task_.handle() && !task_.done())
{
task_.handle().resume();
}
else
{
// A resume() with no coroutine to run (e.g. a duplicate external
// post() after completion). Resuming a null or finished handle is
// undefined behavior, so skip it -- this matches the old
// Coro::resume() `if (coro_)` guard. The bookkeeping below still
// runs to balance the ++runCount_ done by the post() that
// scheduled this call.
JLOG(jq_.journal_.warn())
<< "CoroTaskRunner::resume called for coroutine '" << name_
<< "' with no runnable coroutine (duplicate post or already completed)";
}
detail::getLocalValues().release();
detail::getLocalValues().reset(saved);
if (task_.done())
{
finished_.store(true, std::memory_order_release);
// An exception that escapes a top-level coroutine body is captured
// by promise_type::unhandled_exception() but has no awaiter to
// rethrow it, so it would vanish with the frame. Surface it in the
// log. (The old Boost path propagated it out of resume() instead.)
if (auto const& ep = task_.handle().promise().exception_)
{
try
{
std::rethrow_exception(ep);
}
catch (std::exception const& e)
{
JLOG(jq_.journal_.error())
<< "Unhandled exception in coroutine '" << name_ << "': " << e.what();
}
catch (...)
{
JLOG(jq_.journal_.error())
<< "Unhandled non-standard exception in coroutine '" << name_ << "'";
}
}
// Break the shared_ptr cycle: frame -> shared_ptr<runner> -> this.
// Use std::move (not task_ = {}) so task_.handle_ is null BEFORE the
// frame is destroyed. operator= would destroy the frame while handle_
@@ -388,6 +424,13 @@ JobQueue::CoroTaskRunner::expectEarlyExit()
* decrements runCount_ below zero. In that scenario runCount_
* never returns to 0, but finished_ becoming true guarantees
* the coroutine is done and no more resumes will occur.
*
* Note: when join() returns via the finished_ disjunct, the final
* resume() call may still be executing its post-completion
* bookkeeping (the --runCount_ / notify after finished_ is set).
* That is safe -- the coroutine body has fully completed and the
* runner is kept alive by the resume job's shared_ptr -- but
* callers must not assume resume() itself has returned.
*/
inline void
JobQueue::CoroTaskRunner::join()

125
src/test/core/CoroTask_test.cpp
View File

@@ -3,15 +3,21 @@
#include <xrpld/core/Config.h>
#include <xrpl/basics/LocalValue.h>
#include <xrpl/beast/unit_test/suite.h>
#include <xrpl/core/CoroTask.h>
#include <xrpl/core/Job.h>
#include <xrpl/core/JobQueue.h>
#include <xrpl/core/JobQueueAwaiter.h>
#include <array>
#include <chrono>
#include <condition_variable>
#include <memory>
#include <mutex>
#include <stdexcept>
#include <string>
#include <vector>
namespace xrpl::test {
@@ -42,9 +48,9 @@ namespace xrpl::test {
* testVoidCompletion | CoroTask<void> basic lifecycle
* testCorrectOrder | suspend() -> join() -> post() -> complete
* testIncorrectOrder | post() before suspend() (race-safe path)
* testJobQueueAwaiter | JobQueueAwaiter suspend + auto-repost
* testJobQueueAwaiter | JobQueueAwaiter + yieldAndPost suspend/repost
* testThreadSpecificStorage | LocalValue isolation across coroutines
* testExceptionPropagation | unhandled_exception() in promise_type
* testExceptionPropagation | CoroTask<void> exception via co_await
* testMultipleYields | N sequential suspend/resume cycles
* testValueReturn | CoroTask<T> co_return value
* testValueException | CoroTask<T> exception via co_await
@@ -125,7 +131,8 @@ public:
co_return;
});
BEAST_EXPECT(runner);
BEAST_EXPECT(g.waitFor(5s));
if (!BEAST_EXPECT(g.waitFor(5s)))
return;
runner->join();
BEAST_EXPECT(!runner->runnable());
}
@@ -148,22 +155,20 @@ public:
}));
Gate g1, g2;
std::shared_ptr<JobQueue::CoroTaskRunner> r;
auto runner = env.app().getJobQueue().postCoroTask(
JtClient,
"CoroTaskTest",
[rp = &r, g1p = &g1, g2p = &g2](auto runner) -> CoroTask<void> {
*rp = runner;
JtClient, "CoroTaskTest", [g1p = &g1, g2p = &g2](auto runner) -> CoroTask<void> {
g1p->signal();
co_await runner->suspend();
g2p->signal();
co_return;
});
BEAST_EXPECT(runner);
BEAST_EXPECT(g1.waitFor(5s));
if (!BEAST_EXPECT(g1.waitFor(5s)))
return;
runner->join();
runner->post();
BEAST_EXPECT(g2.waitFor(5s));
if (!BEAST_EXPECT(g2.waitFor(5s)))
return;
runner->join();
}
@@ -196,7 +201,11 @@ public:
}
/**
* JobQueueAwaiter suspend + auto-repost across multiple yield points.
* Suspend + auto-repost across multiple yield points, using the
* external JobQueueAwaiter struct for the first suspension and the
* inline yieldAndPost() awaiter for the second. JobQueueAwaiter is
* used at only one co_await point per coroutine (see the GCC-12
* multi-use warning in JobQueueAwaiter.h).
*/
void
testJobQueueAwaiter()
@@ -212,19 +221,22 @@ public:
}));
Gate g;
int step = 0;
env.app().getJobQueue().postCoroTask(
JtClient, "CoroTaskTest", [sp = &step, gp = &g](auto runner) -> CoroTask<void> {
*sp = 1;
std::vector<int> steps;
auto runner = env.app().getJobQueue().postCoroTask(
JtClient, "CoroTaskTest", [sp = &steps, gp = &g](auto runner) -> CoroTask<void> {
sp->push_back(1);
co_await JobQueueAwaiter{runner};
sp->push_back(2);
co_await runner->yieldAndPost();
*sp = 2;
co_await runner->yieldAndPost();
*sp = 3;
sp->push_back(3);
gp->signal();
co_return;
});
BEAST_EXPECT(g.waitFor(5s));
BEAST_EXPECT(step == 3);
BEAST_EXPECT(runner);
if (!BEAST_EXPECT(g.waitFor(5s)))
return;
runner->join();
BEAST_EXPECT(steps == std::vector<int>({1, 2, 3}));
}
/**
@@ -255,7 +267,8 @@ public:
this->BEAST_EXPECT(*lv == -2);
g.signal();
});
BEAST_EXPECT(g.waitFor(5s));
if (!BEAST_EXPECT(g.waitFor(5s)))
return;
BEAST_EXPECT(*lv == -1);
for (int i = 0; i < N; ++i)
@@ -277,13 +290,15 @@ public:
this->BEAST_EXPECT(**lvp == id);
co_return;
});
BEAST_EXPECT(g.waitFor(5s));
if (!BEAST_EXPECT(g.waitFor(5s)))
return;
a[i]->join();
}
for (auto const& r : a)
{
r->post();
BEAST_EXPECT(g.waitFor(5s));
if (!BEAST_EXPECT(g.waitFor(5s)))
return;
r->join();
}
for (auto const& r : a)
@@ -296,13 +311,17 @@ public:
this->BEAST_EXPECT(*lv == -2);
g.signal();
});
BEAST_EXPECT(g.waitFor(5s));
if (!BEAST_EXPECT(g.waitFor(5s)))
return;
BEAST_EXPECT(*lv == -1);
}
/**
* Exception thrown in coroutine body is caught by
* promise_type::unhandled_exception(). Coroutine completes.
* An exception thrown in an awaited CoroTask<void> is rethrown into
* the awaiting coroutine by await_resume(), with the original
* message intact. (An exception escaping the top-level body has no
* awaiter to rethrow it; it is captured by unhandled_exception()
* and logged by CoroTaskRunner::resume().)
*/
void
testExceptionPropagation()
@@ -318,17 +337,29 @@ public:
}));
Gate g;
std::string what;
auto runner = env.app().getJobQueue().postCoroTask(
JtClient, "CoroTaskTest", [gp = &g](auto) -> CoroTask<void> {
JtClient, "CoroTaskTest", [wp = &what, gp = &g](auto) -> CoroTask<void> {
auto inner = []() -> CoroTask<void> {
throw std::runtime_error("test exception");
co_return;
};
try
{
co_await inner();
}
catch (std::runtime_error const& e)
{
*wp = e.what();
}
gp->signal();
throw std::runtime_error("test exception");
co_return;
});
BEAST_EXPECT(runner);
BEAST_EXPECT(g.waitFor(5s));
if (!BEAST_EXPECT(g.waitFor(5s)))
return;
runner->join();
// The exception is caught by promise_type::unhandled_exception()
// and the coroutine is considered done
BEAST_EXPECT(what == "test exception");
BEAST_EXPECT(!runner->runnable());
}
@@ -350,12 +381,8 @@ public:
Gate g;
int counter = 0;
std::shared_ptr<JobQueue::CoroTaskRunner> r;
auto runner = env.app().getJobQueue().postCoroTask(
JtClient,
"CoroTaskTest",
[rp = &r, cp = &counter, gp = &g](auto runner) -> CoroTask<void> {
*rp = runner;
JtClient, "CoroTaskTest", [cp = &counter, gp = &g](auto runner) -> CoroTask<void> {
++(*cp);
gp->signal();
co_await runner->suspend();
@@ -368,17 +395,20 @@ public:
});
BEAST_EXPECT(runner);
BEAST_EXPECT(g.waitFor(5s));
if (!BEAST_EXPECT(g.waitFor(5s)))
return;
BEAST_EXPECT(counter == 1);
runner->join();
runner->post();
BEAST_EXPECT(g.waitFor(5s));
if (!BEAST_EXPECT(g.waitFor(5s)))
return;
BEAST_EXPECT(counter == 2);
runner->join();
runner->post();
BEAST_EXPECT(g.waitFor(5s));
if (!BEAST_EXPECT(g.waitFor(5s)))
return;
BEAST_EXPECT(counter == 3);
runner->join();
BEAST_EXPECT(!runner->runnable());
@@ -411,7 +441,8 @@ public:
co_return;
});
BEAST_EXPECT(runner);
BEAST_EXPECT(g.waitFor(5s));
if (!BEAST_EXPECT(g.waitFor(5s)))
return;
runner->join();
BEAST_EXPECT(result == 42);
BEAST_EXPECT(!runner->runnable());
@@ -435,9 +466,9 @@ public:
}));
Gate g;
bool caught = false;
std::string what;
auto runner = env.app().getJobQueue().postCoroTask(
JtClient, "CoroTaskTest", [cp = &caught, gp = &g](auto) -> CoroTask<void> {
JtClient, "CoroTaskTest", [wp = &what, gp = &g](auto) -> CoroTask<void> {
auto inner = []() -> CoroTask<int> {
throw std::runtime_error("inner error");
co_return 0;
@@ -448,15 +479,16 @@ public:
}
catch (std::runtime_error const& e)
{
*cp = true;
*wp = e.what();
}
gp->signal();
co_return;
});
BEAST_EXPECT(runner);
BEAST_EXPECT(g.waitFor(5s));
if (!BEAST_EXPECT(g.waitFor(5s)))
return;
runner->join();
BEAST_EXPECT(caught);
BEAST_EXPECT(what == "inner error");
BEAST_EXPECT(!runner->runnable());
}
@@ -491,7 +523,8 @@ public:
co_return;
});
BEAST_EXPECT(runner);
BEAST_EXPECT(g.waitFor(5s));
if (!BEAST_EXPECT(g.waitFor(5s)))
return;
runner->join();
BEAST_EXPECT(result == 14); // (3 + 4) * 2
BEAST_EXPECT(!runner->runnable());

9
src/test/core/Coroutine_test.cpp
View File

@@ -78,7 +78,8 @@ public:
g2p->signal();
co_return;
});
BEAST_EXPECT(g1.waitFor(5s));
if (!BEAST_EXPECT(g1.waitFor(5s)))
return;
c->join();
c->post();
BEAST_EXPECT(g2.waitFor(5s));
@@ -158,13 +159,15 @@ public:
this->BEAST_EXPECT(**lvp == id);
co_return;
});
BEAST_EXPECT(g.waitFor(5s));
if (!BEAST_EXPECT(g.waitFor(5s)))
return;
a[i]->join();
}
for (auto const& c : a)
{
c->post();
BEAST_EXPECT(g.waitFor(5s));
if (!BEAST_EXPECT(g.waitFor(5s)))
return;
c->join();
}
for (auto const& c : a)

16
src/test/core/JobQueue_test.cpp
View File

@@ -87,7 +87,7 @@ class JobQueue_test : public beast::unit_test::Suite
BEAST_EXPECT(yieldCount == 4);
}
{
// Test repeated resume()s until the coroutine completes.
// Test repeated post()+join()s until the coroutine completes.
int yieldCount{0};
auto const runner = jQueue.postCoroTask(
JtClient, "PostCoroTest2", [ycp = &yieldCount](auto runner) -> CoroTask<void> {
@@ -106,11 +106,18 @@ class JobQueue_test : public beast::unit_test::Suite
// Wait for the Job to run and yield.
runner->join();
// Now resume until the CoroTaskRunner says it is done.
// Now post()+join() until the CoroTaskRunner says it is done.
// resume() requires a prior post() (see the precondition on
// CoroTaskRunner::resume()), so the posted job performs the
// resume and join() blocks until it completes. yieldCount is
// deliberately not atomic: the mutexRun_ handoff inside join()
// must provide the happens-before edge that makes the
// increment visible to this thread.
int old = yieldCount;
while (runner->runnable())
{
runner->resume(); // Resume runs synchronously on this thread.
BEAST_EXPECT(runner->post());
runner->join();
BEAST_EXPECT(++old == yieldCount);
}
BEAST_EXPECT(yieldCount == 4);
@@ -128,10 +135,11 @@ class JobQueue_test : public beast::unit_test::Suite
bool unprotected = false;
auto const runner = jQueue.postCoroTask(
JtClient, "PostCoroTest3", [up = &unprotected](auto) -> CoroTask<void> {
*up = false;
*up = true;
co_return;
});
BEAST_EXPECT(runner == nullptr);
BEAST_EXPECT(unprotected == false);
}
}

8
src/xrpld/rpc/detail/PathRequestManager.cpp
View File

@@ -6,6 +6,7 @@
#include <xrpld/rpc/detail/PathRequest.h>
#include <xrpl/basics/Log.h>
#include <xrpl/basics/scope.h>
#include <xrpl/core/Job.h>
#include <xrpl/core/JobQueue.h>
#include <xrpl/json/json_value.h>
@@ -137,9 +138,12 @@ PathRequestManager::updateAll(std::shared_ptr<ReadView const> const& inLedger)
}
else if (request->hasCompletion())
{
// One-shot request with completion function
// One-shot request with completion function. Fire the
// completion even if doUpdate throws, so the RPC
// handler blocked in doRipplePathFind is released
// immediately instead of waiting out its timeout.
ScopeExit const complete{[&request] { request->updateComplete(); }};
request->doUpdate(cache, false);
request->updateComplete();
++processed;
}
}

6
src/xrpld/rpc/detail/Tuning.h
View File

@@ -65,6 +65,12 @@ static constexpr int kDefaultAutoFillFeeMultiplier = 10;
static constexpr int kDefaultAutoFillFeeDivisor = 1;
static constexpr int kMaxPathfindsInProgress = 2;
static constexpr int kMaxPathfindJobCount = 50;
/**
* How long a blocking ripple_path_find call waits for the path-finding
* continuation before giving up and returning an internal error.
*/
constexpr auto kPathfindCompletionTimeout = std::chrono::seconds{30};
static constexpr int kMaxJobQueueClients = 500;
constexpr auto kMaxValidatedLedgerAge = std::chrono::minutes{2};
static constexpr int kMaxRequestSize = 1000000;

18
src/xrpld/rpc/handlers/orderbook/RipplePathFind.cpp
View File

@@ -49,6 +49,20 @@ doRipplePathFind(RPC::JsonContext& context)
PathRequest::pointer request;
lpLedger = context.ledgerMaster.getClosedLedger();
// The wait below parks this JobQueue worker thread until the
// path-finding continuation fires. The continuation is fired by a
// JtUpdatePf job, which itself needs a free worker to run. Bound
// the number of concurrently parked workers (LegacyPathFind admits
// at most kMaxPathfindsInProgress non-admin requests) so that
// concurrent ripple_path_find calls cannot occupy every worker and
// stall the whole JobQueue. The guard must stay in scope until the
// wait completes. (The old Boost.Coroutine implementation did not
// need this: it suspended and released the worker instead of
// blocking it.)
RPC::LegacyPathFind const lpf(isUnlimited(context.role), context.app);
if (!lpf.isOk())
return rpcError(RpcTooBusy);
// makeLegacyPathRequest enqueues a path-finding job that runs
// asynchronously. We block this thread with a condition_variable
// until the path-finding continuation signals completion.
@@ -81,9 +95,9 @@ doRipplePathFind(RPC::JsonContext& context)
context.params);
if (request)
{
using namespace std::chrono_literals;
std::unique_lock lk(state->mtx);
if (!state->cv.wait_for(lk, 30s, [&state] { return state->done; }))
if (!state->cv.wait_for(
lk, RPC::Tuning::kPathfindCompletionTimeout, [&state] { return state->done; }))
{
// Path-finding continuation never fired (e.g. shutdown
// race or unexpected failure). Return an internal error