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

Merge branch 'pratik/std-coro/add-coroutine-primitives' into pratik/std-coro/migrate-entry-points

Browse Source
This commit is contained in:
Pratik Mankawde
2026-07-25 12:37:14 +01:00
parent 4c18dd867c 8d4ea00453
commit 5e9ce16de6
6 changed files with 125 additions and 2036 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
@@ -126,7 +125,6 @@ words:
- fsanitize
- funclets
- Gamal
- gantt
- gcov
- gcovr
- ghead
@@ -170,7 +168,6 @@ words:
- lseq
- lsmf
- ltype
- Mankawde
- mathbunnyru
- mcmodel
- MEMORYSTATUSEX
@@ -232,7 +229,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

@@ -323,12 +323,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_ = true;
// 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_
@@ -384,6 +420,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
@@ -124,7 +130,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());
}
@@ -147,22 +154,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();
}
@@ -195,7 +200,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()
@@ -211,19 +220,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}));
}
/**
@@ -254,7 +266,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)
@@ -276,13 +289,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)
@@ -295,13 +310,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()
@@ -317,17 +336,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());
}
@@ -349,12 +380,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();
@@ -367,17 +394,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());
@@ -410,7 +440,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());
@@ -434,9 +465,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;
@@ -447,15 +478,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());
}
@@ -490,7 +522,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());