Compare commits

..

8 Commits

Author SHA1 Message Date
Mayukha Vadari
df31ff54e6 docs: Add tx/rpc AGENTS.md files and symlink convention
Documents the amendment workflow (features.macro) under src/libxrpl/tx
and the API-CHANGELOG.md requirement under src/xrpld/rpc. Also makes
it an explicit rule in CONTRIBUTING.md that any new AGENTS.md gets a
CLAUDE.md symlink alongside it.
2026-08-20 13:10:45 -04:00
Mayukha Vadari
cf4f627b33 docs: Correct xrpld unittest invocation in AGENTS.md 2026-08-20 12:57:06 -04:00
Mayukha Vadari
3e2e1d58d6 docs: Reflow paragraphs to single lines 2026-08-20 12:28:46 -04:00
Mayukha Vadari
79dcb83d5e docs: Add AGENTS.md/CLAUDE.md for AI coding agent guidance
Adds shared build/test/lint/architecture guidance for AI coding agents,
with CLAUDE.md symlinked to AGENTS.md for Claude Code. Updates
.gitignore so only personal/local AI-tool config is excluded, and
documents the shared vs. local convention in CONTRIBUTING.md.
2026-08-20 12:21:04 -04:00
Jingchen
959a186a0f build: Suppress MSVC linker warning LNK4099 (#8049) 2026-08-20 13:19:07 +00:00
dependabot[bot]
3b1c9e4320 chore: [DEPENDABOT] Bump cxx from 1.0.198 to 1.0.199 in /crates in the rust-dependencies group across 1 directory (#8050)
Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2026-08-20 13:12:14 +00:00
Ayaz Salikhov
a3a2c85c41 build: Use debian any distribution & signed (hosted) rpm repo (#8053) 2026-08-20 13:05:18 +00:00
Ayaz Salikhov
b1ac891210 ci: Do not cache cargo binaries (#8062) 2026-08-20 11:47:27 +00:00
47 changed files with 419 additions and 2831 deletions

View File

@@ -256,8 +256,6 @@ words:
- replayer
- repodata
- repomd
- repost
- reposts
- rerandomize
- rerandomization
- rerandomized
@@ -305,7 +303,6 @@ words:
- sponsees
- SRPMS
- sslws
- stackful
- statsd
- STATSDCOLLECTOR
- stissue

38
.github/actions/cargo-cache/action.yml vendored Normal file
View File

@@ -0,0 +1,38 @@
name: Use cargo artifacts cache
description: >
Cache the cargo build artifacts with rust-cache. Never caches ~/.cargo/bin:
when saving the cache, rust-cache deletes all binaries that were already
present there, which on persistent self-hosted runners wipes the tools
installed by prepare-runner. Harmless on ephemeral runners, but kept
consistent everywhere.
inputs:
workspaces:
description: "Workspaces to cache, as 'workspace -> target' lines."
required: false
default: crates
key:
description: "Additional part of the cache key."
required: false
default: ""
cache-directories:
description: "Additional non-workspace directories to cache."
required: false
default: ""
save-if:
description: "Condition for saving the cache after the job."
required: false
default: "true"
runs:
using: composite
steps:
- name: Use cargo artifacts cache
uses: Swatinem/rust-cache@6323deb102c322ba6fcbdcafc7e3dddab59af2b6 # v2.9.2
with:
cache-bin: "false"
cache-directories: ${{ inputs.cache-directories }}
key: ${{ inputs.key }}
save-if: ${{ inputs.save-if }}
workspaces: ${{ inputs.workspaces }}

View File

@@ -4,6 +4,7 @@ updates:
directories:
- /
- .github/actions/build-deps/
- .github/actions/cargo-cache/
- .github/actions/release-info/
- .github/actions/set-compiler-env/
- .github/actions/setup-conan/

View File

@@ -58,7 +58,7 @@ jobs:
base_image: debian:bookworm
- name: rhel
base_image: registry.access.redhat.com/ubi9/ubi:latest
uses: XRPLF/actions/.github/workflows/build-multiarch-image.yml@9e7e4e80af9e684c116b38369add8eea64451f32
uses: XRPLF/actions/.github/workflows/build-multiarch-image.yml@65d5a0bd72be4ecea95cff0673a6e0672ab5243a
with:
image_name: xrpld/nix-${{ matrix.distro.name }}
dockerfile: nix/docker/Dockerfile

View File

@@ -39,7 +39,7 @@ jobs:
# AlmaLinux rather than UBI9, which does not ship rpm-sign.
- name: rhel
base_image: almalinux:9
uses: XRPLF/actions/.github/workflows/build-multiarch-image.yml@9e7e4e80af9e684c116b38369add8eea64451f32
uses: XRPLF/actions/.github/workflows/build-multiarch-image.yml@65d5a0bd72be4ecea95cff0673a6e0672ab5243a
with:
image_name: xrpld/packaging-${{ matrix.distro.name }}
dockerfile: package/Dockerfile

View File

@@ -30,7 +30,7 @@ jobs:
permissions:
contents: read
packages: write
uses: XRPLF/actions/.github/workflows/build-multiarch-image.yml@9e7e4e80af9e684c116b38369add8eea64451f32
uses: XRPLF/actions/.github/workflows/build-multiarch-image.yml@65d5a0bd72be4ecea95cff0673a6e0672ab5243a
with:
image_name: xrpld/pre-commit
dockerfile: bin/pre-commit/Dockerfile

View File

@@ -79,7 +79,7 @@ jobs:
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
- name: Prepare runner
uses: XRPLF/actions/prepare-runner@51af40f99ea91a08c3528ddf16d98132dcc7e63c
uses: XRPLF/actions/prepare-runner@7bf7ceca5932114abdd0d43493c3c30c5a654e13
with:
enable_ccache: false

View File

@@ -14,7 +14,7 @@ on:
jobs:
# Call the workflow in the XRPLF/actions repo that runs the pre-commit hooks.
run-hooks:
uses: XRPLF/actions/.github/workflows/pre-commit.yml@3ba08d6ddf114092891d48491fc2e26c3ba15552
uses: XRPLF/actions/.github/workflows/pre-commit.yml@f1952595d212e86169935135efc66294b4574131
with:
runs_on: ubuntu-latest
container: '{ "image": "ghcr.io/xrplf/xrpld/pre-commit:sha-f56b79f" }'

View File

@@ -47,7 +47,7 @@ jobs:
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
- name: Prepare runner
uses: XRPLF/actions/prepare-runner@51af40f99ea91a08c3528ddf16d98132dcc7e63c
uses: XRPLF/actions/prepare-runner@7bf7ceca5932114abdd0d43493c3c30c5a654e13
with:
enable_ccache: false

View File

@@ -129,7 +129,7 @@ jobs:
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
- name: Prepare runner
uses: XRPLF/actions/prepare-runner@51af40f99ea91a08c3528ddf16d98132dcc7e63c
uses: XRPLF/actions/prepare-runner@7bf7ceca5932114abdd0d43493c3c30c5a654e13
with:
enable_ccache: ${{ inputs.ccache_enabled }}
@@ -163,7 +163,7 @@ jobs:
compiler: ${{ inputs.compiler }}
- name: Use cargo artifacts cache
uses: Swatinem/rust-cache@6323deb102c322ba6fcbdcafc7e3dddab59af2b6 # v2.9.2
uses: ./.github/actions/cargo-cache
with:
cache-directories: ${{ env.BUILD_DIR }}/corrosion
key: ${{ inputs.config_name }}

View File

@@ -27,7 +27,7 @@ jobs:
determine-files:
permissions:
contents: read
uses: XRPLF/actions/.github/workflows/determine-tidy-files.yml@d041ac9f1fa9f07a4ba335eb4c1c82233fb3fef6
uses: XRPLF/actions/.github/workflows/determine-tidy-files.yml@70145243b905dc3e040a61d39c00e178cfb96f71
run-clang-tidy:
name: Run clang tidy
@@ -43,7 +43,7 @@ jobs:
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
- name: Prepare runner
uses: XRPLF/actions/prepare-runner@51af40f99ea91a08c3528ddf16d98132dcc7e63c
uses: XRPLF/actions/prepare-runner@7bf7ceca5932114abdd0d43493c3c30c5a654e13
with:
enable_ccache: false
@@ -60,7 +60,7 @@ jobs:
compiler: ${{ env.COMPILER }}
- name: Use cargo artifacts cache
uses: Swatinem/rust-cache@6323deb102c322ba6fcbdcafc7e3dddab59af2b6 # v2.9.2
uses: ./.github/actions/cargo-cache
with:
cache-directories: ${{ env.BUILD_DIR }}/corrosion
save-if: ${{ github.ref == 'refs/heads/develop' || startsWith(github.ref, 'refs/heads/release') }}

View File

@@ -33,9 +33,7 @@ jobs:
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
- name: Use cargo artifacts cache
uses: Swatinem/rust-cache@6323deb102c322ba6fcbdcafc7e3dddab59af2b6 # v2.9.2
with:
workspaces: crates
uses: ./.github/actions/cargo-cache
- name: Run clippy
run: cargo clippy --workspace --all-targets --all-features --locked -- -D warnings
@@ -48,9 +46,7 @@ jobs:
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
- name: Use cargo artifacts cache
uses: Swatinem/rust-cache@6323deb102c322ba6fcbdcafc7e3dddab59af2b6 # v2.9.2
with:
workspaces: crates
uses: ./.github/actions/cargo-cache
- name: Generate coverage report
run: cargo llvm-cov nextest --workspace --all-features --locked --no-tests=warn --lcov --output-path lcov.info
@@ -76,9 +72,7 @@ jobs:
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
- name: Use cargo artifacts cache
uses: Swatinem/rust-cache@6323deb102c322ba6fcbdcafc7e3dddab59af2b6 # v2.9.2
with:
workspaces: crates
uses: ./.github/actions/cargo-cache
- name: Build documentation
env:

View File

@@ -68,7 +68,7 @@ jobs:
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
- name: Prepare runner
uses: XRPLF/actions/prepare-runner@c00c22ada3bd6bcda48fcb0d62fbbab49fec8a0f
uses: XRPLF/actions/prepare-runner@7bf7ceca5932114abdd0d43493c3c30c5a654e13
with:
enable_ccache: false

12
.gitignore vendored
View File

@@ -72,11 +72,19 @@ DerivedData
/.zed/
# AI tools.
# Shared/committable AI agent config (AGENTS.md, CLAUDE.md, GEMINI.md, .claude/settings.json,
# tool-specific rules files, etc.) should be checked in — see CONTRIBUTING.md. Only the
# personal/local variants below are ignored.
/.agent
/.agents
/.augment
/.claude
/CLAUDE.md
/.claude/settings.local.json
AGENTS.local.md
CLAUDE.local.md
GEMINI.local.md
.aider.chat.history.md
.aider.input.history
.aider.tags.cache.v3
# Python
__pycache__

73
AGENTS.md Normal file
View File

@@ -0,0 +1,73 @@
# AGENTS.md
This file provides guidance to AI coding agents (Claude Code, and other AGENTS.md-compatible tools) when working with code in this repository. `CLAUDE.md` is a symlink to this file.
For personal, untracked notes to an AI agent (not meant to be shared with other contributors), use `AGENTS.local.md` / `CLAUDE.local.md` instead — see [CONTRIBUTING.md](./CONTRIBUTING.md).
## Build
Preferred: use the Nix devshell, which sets up the compiler, Conan, ccache, and (optionally) Rust automatically.
```bash
nix develop # default shell (clang on macOS, gcc on Linux); also .#gcc, .#clang, .#gcc-plain, .#clang-plain, .#apple-clang
```
Manual build (also what the devshell does under the hood):
```bash
./conan/init.sh # one-time Conan profile/remote setup (auto-run inside nix develop)
mkdir .build && cd .build
conan install .. --output-folder . --build missing --settings build_type=Release
cmake -DCMAKE_TOOLCHAIN_FILE:FILEPATH=build/generators/conan_toolchain.cmake -DCMAKE_BUILD_TYPE=Release -Dxrpld=ON -Dtests=ON ..
cmake --build . --parallel <N>
```
Key CMake options: `-Dxrpld=ON` (build the server binary, not just `libxrpl`), `-Dtests=ON`, `-Drust=ON` (builds `crates/`, requires cargo/rustc — off by default but always on in CI), `-Dunity=ON`, `-Dcoverage=ON`, `-Dwerr=ON`. `-Dverify_headers` is on by default; `cmake --build . --target verify-headers` compiles every header standalone.
Protocol codegen (from `.macro` files) must be regenerated and committed when changed:
```bash
cmake --build . --target setup_code_gen
cmake --build . --target code_gen
```
Rust crate tests (independent of the CMake build): `cargo test --manifest-path crates/Cargo.toml --workspace` (CI uses `cargo nextest`).
## Testing
Unit tests are a custom framework built into the `xrpld` binary itself (not Boost.Test/GTest/Catch). Build first (e.g. `cmake --build .build -j10`), then run from the build directory:
```bash
.build/xrpld --unittest --unittest-jobs <N> # run all suites; N = ~half of available cores
.build/xrpld --unittest xrpl.app.Credentials # run one suite by name
```
(Multi-config generators produce the binary under e.g. `.build/Release/xrpld`.) A suite's `--unittest` name is built from the arguments to its `BEAST_DEFINE_TESTSUITE`/`BEAST_DEFINE_TESTSUITE_PRIO` macro (usually at the bottom of the test file), in reverse order and joined with `.`: `BEAST_DEFINE_TESTSUITE(Credentials, app, xrpl)``xrpl.app.Credentials`. `--unittest` also matches by prefix unless there's an exact name match, in which case only that exact suite runs. `--unittest-arg` does nothing — don't use it. Tests that run offline in under a minute should be automatic `--unittest` suites under `src/test/`; anything else is a manual/integration test. `tests/` (top-level, separate from `src/test/`) holds integration tests exercised against `libxrpl`/`xrpld`.
## Lint/Format
```bash
pip install pre-commit && pre-commit install
pre-commit run --all-files
pre-commit run clang-format --all-files # single hook
TIDY=1 pre-commit run clang-tidy # clang-tidy is opt-in (needs local clang-tidy + generated headers)
```
Manual clang-tidy: build the `tidy_prerequisites` target first, then `run-clang-tidy -p build -allow-no-checks src tests` (add `-fix -format` to auto-fix).
## Architecture
- `include/xrpl/` + `src/libxrpl/` — the core protocol library: ledger, shamap, consensus, crypto, json, resource, nodestore, rdb, peerfinder, and `tx/` (transaction application: `Transactor.cpp`, `applySteps.cpp`, invariants, payment paths). `tx/transactors/` has one file per transaction type, grouped by subsystem: `escrow/`, `vault/`, `lending/`, `sponsor/`, `nft/`, `token/` (MPT), `payment_channel/`, `permissioned_domain/`, `dex/`, `oracle/`, `did/`, `credentials/`, `bridge/`, `check/`, `delegate/`, `account/`, `system/`. Any change to transaction-processing behavior must be gated behind an Amendment.
- `src/xrpld/` — the server application built on top of `libxrpl`: `app`, `core`, `overlay` (P2P networking), `peerfinder`, `perflog`, `rpc`, `shamap`. `main` builds an `ApplicationImp` implementing `Application`; most components hold a reference to it (`app_`), giving broad cross-component access — expect to trace call chains through `Application&`.
- `src/test/` — unit tests mirroring the subsystems above, plus `jtx/` (the transaction-building test DSL — e.g. `jtx/escrow.h`, `jtx/vault.h`, `jtx/sponsor.h`, `jtx/permissioned_dex.h`) and `unit_test/` (the custom test framework itself, derived from Beast).
- `src/tests/` — a second, separate tree of integration-style tests for `libxrpl`.
- `crates/` — a Rust workspace (only built with `-Dxrpld -Drust=ON`) bridged into C++ via `cxxbridge`/the `cxx` crate; currently just a `hello_world` interop scaffold. Requires the Rust toolchain pinned in `rust-toolchain.toml` (the Nix devshell provides it automatically).
## Code Style (see `docs/CodingStyle.md` and `CONTRIBUTING.md` for full detail)
- New file placement is strict: `libxrpl` headers → `include/xrpl`; `libxrpl` sources → `src/libxrpl`; other non-test server code → `src/xrpld`; tests → `src/test`; benchmarks → `src/benchmarks`.
- Header includes must stay levelized (checked by `.github/scripts/levelization`).
- Allman braces, tabs-as-4-spaces (no literal tabs), 80-char lines, east `const`, no naked `new`/`delete`, `*`/`&` bound to the type not the variable (`SomeObject* myObject`), never declare multiple pointers/refs in one statement.
- Class member order: private members first, then the six special members in order (dtor, default ctor, copy ctor, copy assign, move ctor, move assign).
- Use `XRPL_ASSERT`/`UNREACHABLE` instead of raw `assert`/`assert(false)` outside constexpr functions and unit tests; each needs a unique name of the form `scope::function : short description` (used for Antithesis instrumentation).
- Commits: imperative subject line ≤50 chars (72 hard limit), capitalized, no trailing period; each commit should build and pass tests on its own; prefer squashing to one logical commit per PR.

1
CLAUDE.md Symbolic link
View File

@@ -0,0 +1 @@
AGENTS.md

View File

@@ -59,6 +59,12 @@ to an existing XLS. Neither change will be released (in an amendment's
case, marked as `Supported::yes`) until the corresponding XLS's status
is `Final`.
## AI coding agents
[`AGENTS.md`](./AGENTS.md) (and its `CLAUDE.md` symlink, for Claude Code) holds shared, checked-in guidance for AI coding agents working in this repository — build/test/lint commands and architecture notes. Additional `AGENTS.md` files may exist in subdirectories to give agents context specific to that part of the codebase; whenever you add one, also add a `CLAUDE.md` symlink pointing to it (`ln -s AGENTS.md CLAUDE.md`) so Claude Code picks it up too.
If you want to give an agent personal instructions that shouldn't be shared with other contributors (e.g. your own workflow preferences), put them in `AGENTS.local.md` or `CLAUDE.local.md` instead — those are gitignored. Likewise, `.claude/settings.local.json` is for personal, untracked Claude Code settings, while `.claude/settings.json` is shared.
## Before making a pull request
(Or marking a draft pull request as ready.)

View File

@@ -120,7 +120,10 @@ if(MSVC)
_SILENCE_ALL_CXX17_DEPRECATION_WARNINGS
$<$<AND:$<COMPILE_LANGUAGE:CXX>,$<CONFIG:Debug>>:_CRTDBG_MAP_ALLOC>
)
target_link_libraries(common INTERFACE -errorreport:none -machine:X64)
target_link_libraries(
common
INTERFACE -errorreport:none -machine:X64 -ignore:4099
)
else()
target_compile_options(
common

20
crates/Cargo.lock generated
View File

@@ -57,9 +57,9 @@ dependencies = [
[[package]]
name = "cxx"
version = "1.0.198"
version = "1.0.199"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6fe442a792c7c736eea18b32a7f8a3b63cf8aafabda6760042dc2fdeda456291"
checksum = "824894a4a85dca76d4c95c2b9098c036f5a29f627b30c12780774f6654e60974"
dependencies = [
"cc",
"cxx-build",
@@ -72,9 +72,9 @@ dependencies = [
[[package]]
name = "cxx-build"
version = "1.0.198"
version = "1.0.199"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e3184a94384c663718698311a78a51ac00c484c10b4eeac06fb0a068c5f64fa2"
checksum = "f1ae0b651ea5b0000b19513aef5a03f194d7e3486f2d9258b658da8677fe9036"
dependencies = [
"cc",
"codespan-reporting",
@@ -87,9 +87,9 @@ dependencies = [
[[package]]
name = "cxxbridge-cmd"
version = "1.0.198"
version = "1.0.199"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0148d8fd1199329ddf1d157a5e134e51ceff37c6a7ddd38615c399d81cb05d8d"
checksum = "fb05f91d3fb8435d9bab6ac5ce6ac1868be774325fb7fb2a91be39393b21388e"
dependencies = [
"clap",
"codespan-reporting",
@@ -101,15 +101,15 @@ dependencies = [
[[package]]
name = "cxxbridge-flags"
version = "1.0.198"
version = "1.0.199"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "52850339faed2eaadd24e286dc1d8268cc6f8a7bd9524d713adc9099566b4c89"
checksum = "bf293202e0e3e98495785745389e8d0755b217e66f19194a5c695c25e03282ef"
[[package]]
name = "cxxbridge-macro"
version = "1.0.198"
version = "1.0.199"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2c77c856545d886c9bd5215409ebb63b925e262135248b50c79e5a5f194ee47c"
checksum = "ca001d746947c7249ed9d332a10f7a59daedbafeb0ec68c5c18a7db7a93f6ccc"
dependencies = [
"indexmap",
"proc-macro2",

View File

@@ -65,7 +65,7 @@ wherever it appears in the repository configuration.
4. Add the repository, using the channel you picked in [Release channels](#release-channels):
```bash
echo "deb [signed-by=/etc/apt/keyrings/xrplf.asc] https://packages.xrplf.org/repository/deb-stable focal main" | \
echo "deb [signed-by=/etc/apt/keyrings/xrplf.asc] https://packages.xrplf.org/repository/deb-stable any main" | \
sudo tee /etc/apt/sources.list.d/xrplf.list
```
@@ -98,13 +98,13 @@ wherever it appears in the repository configuration.
enabled=1
baseurl=https://packages.xrplf.org/repository/rpm-stable/
gpgcheck=1
repo_gpgcheck=0
repo_gpgcheck=1
gpgkey=https://packages.xrplf.org/xrplf.asc
REPOFILE
```
`gpgcheck=1` verifies each package against the key above.
`repo_gpgcheck` is off because the repository metadata is generated by the server and is not signed.
`repo_gpgcheck=1` verifies the repository metadata, which the server signs with the same key.
3. Install the `xrpld` package:

View File

@@ -1,713 +0,0 @@
#pragma once
#include <xrpl/beast/utility/instrumentation.h>
#include <coroutine>
#include <exception>
#include <type_traits>
#include <utility>
#include <variant>
namespace xrpl {
template <typename T = void>
class CoroTask;
/**
* CoroTask<void> -- coroutine return type for void-returning coroutines.
*
* Class / Dependency Diagram
* ==========================
*
* CoroTask<void>
* +-----------------------------------------------+
* | - handle_ : Handle (coroutine_handle<promise>) |
* +-----------------------------------------------+
* | + handle(), done() |
* | + await_ready/suspend/resume (Awaiter iface) |
* +-----------------------------------------------+
* | owns
* v
* promise_type
* +-----------------------------------------------+
* | - exception_ : std::exception_ptr |
* | - continuation_ : std::coroutine_handle<> |
* +-----------------------------------------------+
* | + get_return_object() -> CoroTask |
* | + initial_suspend() -> suspend_always (lazy) |
* | + final_suspend() -> FinalAwaiter |
* | + return_void() |
* | + unhandled_exception() |
* +-----------------------------------------------+
* | returns at final_suspend
* v
* FinalAwaiter
* +-----------------------------------------------+
* | await_suspend(h): |
* | if continuation_ set -> symmetric transfer |
* | else -> noop_coroutine |
* +-----------------------------------------------+
*
* Design Notes
* ------------
* - Lazy start: initial_suspend returns suspend_always, so the coroutine
* body does not execute until the handle is explicitly resumed.
* - Symmetric transfer: await_suspend returns a coroutine_handle instead
* of void/bool, allowing the scheduler to jump directly to the next
* coroutine without growing the call stack.
* - Continuation chaining: when one CoroTask is co_await-ed inside
* another, the caller's handle is stored as continuation_ so
* FinalAwaiter can resume it when this task finishes.
* - Move-only: the handle is exclusively owned; copy is deleted.
*
* Usage Examples
* ==============
*
* 1. Basic void coroutine (the most common case in xrpld):
*
* CoroTask<void> doWork(std::shared_ptr<CoroTaskRunner> runner) {
* // do something
* co_await runner->suspend(); // yield control
* // resumed later via runner->post() or runner->resume()
* co_return;
* }
*
* 2. co_await-ing one CoroTask<void> from another (chaining):
*
* CoroTask<void> inner() {
* // ...
* co_return;
* }
* CoroTask<void> outer() {
* co_await inner(); // continuation_ links outer -> inner
* co_return; // FinalAwaiter resumes outer
* }
*
* 3. Exceptions propagate through co_await:
*
* CoroTask<void> failing() {
* throw std::runtime_error("oops");
* co_return;
* }
* CoroTask<void> caller() {
* try { co_await failing(); }
* catch (std::runtime_error const&) { // caught here }
* }
*
* Caveats / Pitfalls
* ==================
*
* BUG-RISK: Dangling references in coroutine parameters.
* Coroutine parameters are copied into the frame, but references
* are NOT -- they are stored as-is. If the referent goes out of scope
* before the coroutine finishes, you get use-after-free.
*
* // BROKEN -- local dies before coroutine runs:
* CoroTask<void> bad(int& ref) { co_return; }
* void launch() {
* int local = 42;
* auto task = bad(local); // frame stores &local
* } // local destroyed; frame holds dangling ref
*
* // FIX -- pass by value, or ensure lifetime via shared_ptr.
*
* BUG-RISK: GCC 14 corrupts reference captures in coroutine lambdas.
* When a lambda that returns CoroTask captures by reference ([&]),
* GCC 14 may generate a corrupted coroutine frame. Always capture
* by explicit pointer-to-value instead:
*
* // BROKEN on GCC 14:
* jq.postCoroTask(t, n, [&](auto) -> CoroTask<void> { ... });
*
* // FIX -- capture pointers explicitly:
* jq.postCoroTask(t, n, [ptr = &val](auto) -> CoroTask<void> { ... });
*
* BUG-RISK: Resuming a destroyed or completed CoroTask.
* Calling handle().resume() after the coroutine has already run to
* completion (done() == true) is undefined behavior. The CoroTaskRunner
* guards against this with an XRPL_ASSERT, but standalone usage of
* CoroTask must check done() before resuming.
*
* BUG-RISK: Moving a CoroTask that is being awaited.
* If task A is co_await-ed by task B (so A.continuation_ == B), moving
* or destroying A will invalidate the continuation link. Never move
* or reassign a CoroTask while it is mid-execution or being awaited.
*
* LIMITATION: CoroTask is fire-and-forget for the top-level owner.
* There is no built-in notification when the coroutine finishes.
* The caller must use external synchronization (e.g. CoroTaskRunner::join
* or a gate/condition_variable) to know when it is done.
*
* LIMITATION: No cancellation token.
* There is no way to cancel a suspended CoroTask from outside. The
* coroutine body must cooperatively check a flag (e.g. jq_.isStopping())
* after each co_await and co_return early if needed.
*
* 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 xrpld because all yield() sites are shallow.
*/
template <>
class CoroTask<void>
{
public:
// The C++ coroutine protocol mandates these names (promise_type,
// initial_suspend, await_ready, ...) and instance-callable awaiter
// methods, which conflict with the project naming/static conventions.
// NOLINTBEGIN(readability-identifier-naming, readability-convert-member-functions-to-static)
struct promise_type;
using Handle = std::coroutine_handle<promise_type>;
/**
* Coroutine promise. Compiler uses this to manage coroutine state.
* Stores the exception (if any) and the continuation handle for
* symmetric transfer back to the awaiting coroutine.
*/
struct promise_type
{
// Captured exception from the coroutine body, rethrown in
// await_resume() when this task is co_await-ed by a caller.
std::exception_ptr exception_;
// Handle to the coroutine that is co_await-ing this task.
// Set by await_suspend(). FinalAwaiter uses it for symmetric
// transfer back to the caller. Null if this is a top-level task.
std::coroutine_handle<> continuation_;
/**
* Create the CoroTask return object.
* Called by the compiler at coroutine creation.
*/
CoroTask
get_return_object()
{
return CoroTask{Handle::from_promise(*this)};
}
/**
* Lazy start. The coroutine body does not execute until the
* handle is explicitly resumed (e.g. by CoroTaskRunner::resume).
*/
std::suspend_always
initial_suspend() noexcept
{
return {};
}
/**
* Awaiter returned by final_suspend(). Uses symmetric transfer:
* if a continuation exists, transfers control directly to it
* (tail-call, no stack growth). Otherwise returns noop_coroutine
* so the coroutine frame stays alive for the owner to destroy.
*/
struct FinalAwaiter
{
/**
* Always false. We need await_suspend to run for
* symmetric transfer.
*/
bool
await_ready() noexcept
{
return false;
}
/**
* Symmetric transfer: returns the continuation handle so
* the compiler emits a tail-call instead of a nested resume.
* If no continuation is set, returns noop_coroutine to
* suspend at final_suspend without destroying the frame.
*
* @param h Handle to this completing coroutine
*
* @return Continuation handle, or noop_coroutine
*/
std::coroutine_handle<>
await_suspend(Handle h) noexcept
{
if (auto cont = h.promise().continuation_)
return cont;
return std::noop_coroutine();
}
void
await_resume() noexcept
{
}
};
/**
* Returns FinalAwaiter for symmetric transfer at coroutine end.
*/
FinalAwaiter
final_suspend() noexcept
{
return {};
}
/**
* Called by the compiler for `co_return;` (void coroutine).
*/
void
return_void()
{
}
/**
* Called by the compiler when an exception escapes the coroutine
* body. Captures it for later rethrowing in await_resume().
*/
void
unhandled_exception()
{
exception_ = std::current_exception();
}
};
// NOLINTEND(readability-identifier-naming, readability-convert-member-functions-to-static)
/**
* Default constructor. Creates an empty (null handle) task.
*/
CoroTask() = default;
/**
* Takes ownership of a compiler-generated coroutine handle.
*
* @param h Coroutine handle to own
*/
explicit CoroTask(Handle h) : handle_(h)
{
}
/**
* Destroys the coroutine frame if this task owns one.
*/
~CoroTask()
{
if (handle_)
handle_.destroy();
}
/**
* Move constructor. Transfers handle ownership, leaves other empty.
*/
CoroTask(CoroTask&& other) noexcept : handle_(std::exchange(other.handle_, {}))
{
}
/**
* Move assignment. Destroys current frame (if any), takes other's.
*/
CoroTask&
operator=(CoroTask&& other) noexcept
{
if (this != &other)
{
if (handle_)
handle_.destroy();
handle_ = std::exchange(other.handle_, {});
}
return *this;
}
CoroTask(CoroTask const&) = delete;
CoroTask&
operator=(CoroTask const&) = delete;
/**
* @return The underlying coroutine_handle
*/
[[nodiscard]] Handle
handle() const
{
return handle_;
}
/**
* @return true if the coroutine has run to completion (or thrown)
*/
[[nodiscard]] bool
done() const
{
return handle_ && handle_.done();
}
// -- Awaiter interface: allows `co_await someCoroTask;` --
/**
* Always false. This task is lazy, so co_await always suspends
* the caller to set up the continuation link.
*/
// NOLINTBEGIN(readability-identifier-naming, readability-convert-member-functions-to-static)
[[nodiscard]] bool
await_ready() const noexcept
{
return false;
}
/**
* Stores the caller's handle as our continuation, then returns
* our handle for symmetric transfer (caller suspends, we resume).
*
* @param caller Handle of the coroutine doing co_await on us
*
* @return Our handle for symmetric transfer
*/
std::coroutine_handle<>
await_suspend(std::coroutine_handle<> caller) noexcept
{
XRPL_ASSERT(handle_, "xrpl::CoroTask<void>::await_suspend : handle is valid");
handle_.promise().continuation_ = caller;
return handle_; // Symmetric transfer
}
/**
* Called in the awaiting coroutine's context after this task
* completes. Rethrows any exception captured by
* unhandled_exception().
*/
void
await_resume()
{
XRPL_ASSERT(handle_, "xrpl::CoroTask<void>::await_resume : handle is valid");
if (auto& ep = handle_.promise().exception_)
std::rethrow_exception(ep);
}
// NOLINTEND(readability-identifier-naming, readability-convert-member-functions-to-static)
private:
// Exclusively-owned coroutine handle. Null after move or default
// construction. Destroyed in the destructor.
Handle handle_;
};
/**
* CoroTask<T> -- coroutine return type for value-returning coroutines.
*
* Class / Dependency Diagram
* ==========================
*
* CoroTask<T>
* +-----------------------------------------------+
* | - handle_ : Handle (coroutine_handle<promise>) |
* +-----------------------------------------------+
* | + handle(), done() |
* | + await_ready/suspend/resume (Awaiter iface) |
* +-----------------------------------------------+
* | owns
* v
* promise_type
* +-----------------------------------------------+
* | - result_ : variant<monostate, T, |
* | exception_ptr> |
* | - continuation_ : std::coroutine_handle<> |
* +-----------------------------------------------+
* | + get_return_object() -> CoroTask |
* | + initial_suspend() -> suspend_always (lazy) |
* | + final_suspend() -> FinalAwaiter |
* | + return_value(T) -> stores in result_[1] |
* | + unhandled_exception -> stores in result_[2] |
* +-----------------------------------------------+
* | returns at final_suspend
* v
* FinalAwaiter (same symmetric-transfer pattern as CoroTask<void>)
*
* Value Extraction
* ----------------
* await_resume() inspects the variant:
* - index 2 (exception_ptr) -> rethrow
* - index 1 (T) -> return value via move
*
* Usage Examples
* ==============
*
* 1. Simple value return:
*
* CoroTask<int> computeAnswer() { co_return 42; }
*
* CoroTask<void> caller() {
* int v = co_await computeAnswer(); // v == 42
* }
*
* 2. Chaining value-returning coroutines:
*
* CoroTask<int> add(int a, int b) { co_return a + b; }
* CoroTask<int> doubleSum(int a, int b) {
* int s = co_await add(a, b);
* co_return s * 2;
* }
*
* 3. Exception propagation from inner to outer:
*
* CoroTask<int> failing() {
* throw std::runtime_error("bad");
* co_return 0; // never reached
* }
* CoroTask<void> caller() {
* try {
* int v = co_await failing(); // throws here
* } catch (std::runtime_error const& e) {
* // e.what() == "bad"
* }
* }
*
* Caveats / Pitfalls (in addition to CoroTask<void> caveats above)
* ================================================================
*
* BUG-RISK: await_resume() moves the value out of the variant.
* Calling co_await on the same CoroTask<T> instance twice is undefined
* behavior -- the second call will see a moved-from T. CoroTask is
* single-shot: one co_return, one co_await.
*
* BUG-RISK: T must be move-constructible.
* return_value(T) takes by value and moves into the variant.
* Types that are not movable cannot be used as T.
*
* LIMITATION: No co_yield support.
* CoroTask<T> only supports a single co_return. It does not implement
* yield_value(), so using co_yield inside a CoroTask<T> coroutine is a
* compile error. For streaming values, a different return type
* (e.g. Generator<T>) would be needed.
*
* LIMITATION: Result is only accessible via co_await.
* There is no .get() or .result() method. The value can only be
* extracted by co_await-ing the CoroTask<T> from inside another
* coroutine. For extracting results in non-coroutine code, pass a
* pointer to the caller and write through it (as the tests do).
*/
template <typename T>
class CoroTask
{
static_assert(
std::is_move_constructible_v<T>,
"CoroTask<T> requires T to be move-constructible");
public:
// The C++ coroutine protocol mandates these names (promise_type,
// initial_suspend, await_ready, ...) and instance-callable awaiter
// methods, which conflict with the project naming/static conventions.
// NOLINTBEGIN(readability-identifier-naming, readability-convert-member-functions-to-static)
struct promise_type;
using Handle = std::coroutine_handle<promise_type>;
/**
* Coroutine promise for value-returning coroutines.
* Stores the result as a variant: monostate (not yet set),
* T (co_return value), or exception_ptr (unhandled exception).
*/
struct promise_type
{
// Tri-state result:
// index 0 (monostate) -- coroutine has not yet completed
// index 1 (T) -- co_return value stored here
// index 2 (exception) -- unhandled exception captured here
std::variant<std::monostate, T, std::exception_ptr> result_;
// Handle to the coroutine co_await-ing this task. Used by
// FinalAwaiter for symmetric transfer. Null for top-level tasks.
std::coroutine_handle<> continuation_;
/**
* Create the CoroTask return object.
* Called by the compiler at coroutine creation.
*/
CoroTask
get_return_object()
{
return CoroTask{Handle::from_promise(*this)};
}
/**
* Lazy start. Coroutine body does not run until explicitly resumed.
*/
std::suspend_always
initial_suspend() noexcept
{
return {};
}
/**
* Symmetric-transfer awaiter at coroutine completion.
* Same pattern as CoroTask<void>::FinalAwaiter.
*/
struct FinalAwaiter
{
bool
await_ready() noexcept
{
return false;
}
/**
* Returns continuation for symmetric transfer, or
* noop_coroutine if this is a top-level task.
*
* @param h Handle to this completing coroutine
*
* @return Continuation handle, or noop_coroutine
*/
std::coroutine_handle<>
await_suspend(Handle h) noexcept
{
if (auto cont = h.promise().continuation_)
return cont;
return std::noop_coroutine();
}
void
await_resume() noexcept
{
}
};
FinalAwaiter
final_suspend() noexcept
{
return {};
}
/**
* Called by the compiler for `co_return value;`.
* Moves the value into result_ at index 1.
*
* @param value The value to store
*/
void
return_value(T value)
{
result_.template emplace<1>(std::move(value));
}
/**
* Captures unhandled exceptions at index 2 of result_.
* Rethrown later in await_resume().
*/
void
unhandled_exception()
{
result_.template emplace<2>(std::current_exception());
}
};
// NOLINTEND(readability-identifier-naming, readability-convert-member-functions-to-static)
/**
* Default constructor. Creates an empty (null handle) task.
*/
CoroTask() = default;
/**
* Takes ownership of a compiler-generated coroutine handle.
*
* @param h Coroutine handle to own
*/
explicit CoroTask(Handle h) : handle_(h)
{
}
/**
* Destroys the coroutine frame if this task owns one.
*/
~CoroTask()
{
if (handle_)
handle_.destroy();
}
/**
* Move constructor. Transfers handle ownership, leaves other empty.
*/
CoroTask(CoroTask&& other) noexcept : handle_(std::exchange(other.handle_, {}))
{
}
/**
* Move assignment. Destroys current frame (if any), takes other's.
*/
CoroTask&
operator=(CoroTask&& other) noexcept
{
if (this != &other)
{
if (handle_)
handle_.destroy();
handle_ = std::exchange(other.handle_, {});
}
return *this;
}
CoroTask(CoroTask const&) = delete;
CoroTask&
operator=(CoroTask const&) = delete;
/**
* @return The underlying coroutine_handle
*/
[[nodiscard]] Handle
handle() const
{
return handle_;
}
/**
* @return true if the coroutine has run to completion (or thrown)
*/
[[nodiscard]] bool
done() const
{
return handle_ && handle_.done();
}
// -- Awaiter interface: allows `T val = co_await someCoroTask;` --
/**
* Always false. co_await always suspends to set up continuation.
*/
// NOLINTBEGIN(readability-identifier-naming, readability-convert-member-functions-to-static)
[[nodiscard]] bool
await_ready() const noexcept
{
return false;
}
/**
* Stores caller as continuation, returns our handle for
* symmetric transfer.
*
* @param caller Handle of the coroutine doing co_await on us
*
* @return Our handle for symmetric transfer
*/
std::coroutine_handle<>
await_suspend(std::coroutine_handle<> caller) noexcept
{
XRPL_ASSERT(handle_, "xrpl::CoroTask<T>::await_suspend : handle is valid");
handle_.promise().continuation_ = caller;
return handle_;
}
/**
* Extracts the result: rethrows if exception, otherwise moves
* the T value out of the variant. Single-shot: calling twice
* on the same task is undefined (moved-from T).
*
* @return The co_return-ed value
*/
T
await_resume()
{
XRPL_ASSERT(handle_, "xrpl::CoroTask<T>::await_resume : handle is valid");
auto& result = handle_.promise().result_;
if (auto* ep = std::get_if<2>(&result))
std::rethrow_exception(*ep);
return std::get<1>(std::move(result));
}
// NOLINTEND(readability-identifier-naming, readability-convert-member-functions-to-static)
private:
// Exclusively-owned coroutine handle. Null after move or default
// construction. Destroyed in the destructor.
Handle handle_;
};
} // namespace xrpl

View File

@@ -1,448 +0,0 @@
#pragma once
/**
* @file CoroTaskRunner.ipp
*
* CoroTaskRunner inline implementation.
*
* This file contains the business logic for managing C++20 coroutines
* on the JobQueue. It is included at the bottom of JobQueue.h.
*
* Data Flow: suspend / post / resume cycle
* =========================================
*
* coroutine body CoroTaskRunner JobQueue
* -------------- -------------- --------
* |
* co_await runner->suspend()
* |
* +--- await_suspend ------> onSuspend()
* | ++nSuspend_ ------------> nSuspend_
* | [coroutine is now suspended]
* |
* . (externally or by yieldAndPost())
* .
* +--- (caller calls) -----> post()
* | ++runCount_
* | addJob(resume) ----------> job enqueued
* | |
* | [worker picks up]
* | |
* +--- <----- resume() <-----------------------------------+
* | --nSuspend_ ------> nSuspend_
* | swap in LocalValues (lvs_)
* | task_.handle().resume()
* | |
* | [coroutine body continues here]
* | |
* | swap out LocalValues
* | --runCount_
* | cv_.notify_all()
* v
*
* Thread Safety
* =============
* - mutex_ : guards task_.handle().resume() so that post()-before-suspend
* races cannot resume the coroutine while it is still running.
* (See the race condition discussion in JobQueue.h)
* - mutexRun_ : guards runCount_ counter; used by join() to wait until
* all in-flight resume operations complete.
* - jq_.mutex_: guards nSuspend_ increments/decrements.
*
* Common Mistakes When Modifying This File
* =========================================
*
* 1. Changing lock ordering.
* resume() acquires locks sequentially (never held simultaneously):
* jq_.mutex_ (released immediately), then mutex_ (held across resume),
* then mutexRun_ (released after decrement). post() acquires only
* mutexRun_. Any new code path must follow the same order.
*
* 2. Removing the shared_from_this() capture in post().
* The lambda passed to addJob captures [this, sp = shared_from_this()].
* If you remove sp, 'this' can be destroyed before the job runs,
* causing use-after-free. The sp capture is load-bearing.
*
* 3. Forgetting to decrement nSuspend_ on a new code path.
* Every ++nSuspend_ must have a matching --nSuspend_. If you add a new
* suspension path (e.g. a new awaiter) and forget to decrement on resume
* or on failure, JobQueue::stop() will hang.
*
* 4. Calling task_.handle().resume() without holding mutex_.
* This allows a race where the coroutine runs on two threads
* simultaneously. Always hold mutex_ around resume().
*
* 5. Swapping LocalValues outside of the mutex_ critical section.
* The swap-in and swap-out of LocalValues must bracket the resume()
* call. If you move the swap-out before the lock_guard(mutex_) is
* released, you break LocalValue isolation for any code that runs
* after the coroutine suspends but before the lock is dropped.
*/
namespace xrpl {
/**
* Construct a CoroTaskRunner. Sets runCount_ to 0; does not
* create the coroutine. Call init() afterwards.
*
* @param jq The JobQueue this coroutine will run on
* @param type Job type for scheduling priority
* @param name Human-readable name for logging
*/
inline JobQueue::CoroTaskRunner::CoroTaskRunner(
CreateT,
JobQueue& jq,
JobType type,
std::string name)
: jq_(jq), type_(type), name_(std::move(name))
{
}
/**
* Initialize with a coroutine-returning callable.
* Stores the callable on the heap (FuncStore) so it outlives the
* coroutine frame. Coroutine frames store a reference to the
* callable's implicit object parameter (the lambda). If the callable
* is a temporary, that reference dangles after the caller returns.
* Keeping the callable alive here ensures the coroutine's captures
* remain valid.
*
* @param f Callable: CoroTask<void>(shared_ptr<CoroTaskRunner>)
*/
template <class F>
void
JobQueue::CoroTaskRunner::init(F&& f)
{
using Fn = std::decay_t<F>;
auto store = std::make_unique<FuncStore<Fn>>(std::forward<F>(f));
task_ = store->func(shared_from_this());
storedFunc_ = std::move(store);
}
/**
* Destructor. Waits for any in-flight resume() to complete, then
* asserts (debug) that the coroutine has finished or
* expectEarlyExit() was called.
*
* The join() call is necessary because with async dispatch the
* coroutine runs on a worker thread. The gate signal (which wakes
* the test thread) can arrive before resume() has set finished_.
* join() synchronizes via mutexRun_, establishing a happens-before
* edge: finished_ = true -> unlock(mutexRun_) in resume() ->
* lock(mutexRun_) in join() -> read finished_.
*/
inline JobQueue::CoroTaskRunner::~CoroTaskRunner()
{
#ifndef NDEBUG
join();
XRPL_ASSERT(finished_, "xrpl::JobQueue::CoroTaskRunner::~CoroTaskRunner : is finished");
#endif
}
/**
* Increment the JobQueue's suspended-coroutine count (nSuspend_).
*/
inline void
JobQueue::CoroTaskRunner::onSuspend()
{
std::scoped_lock const lock(jq_.mutex_);
++jq_.nSuspend_;
}
/**
* Decrement nSuspend_ without resuming.
*/
inline void
JobQueue::CoroTaskRunner::onUndoSuspend()
{
std::scoped_lock const lock(jq_.mutex_);
--jq_.nSuspend_;
}
/**
* Return a SuspendAwaiter whose await_suspend() increments nSuspend_
* before the coroutine actually suspends. The caller must later call
* post() or resume() to continue execution.
*
* @return Awaiter for use with `co_await runner->suspend()`
*/
inline auto
JobQueue::CoroTaskRunner::suspend()
{
/**
* Custom awaiter for suspend(). Always suspends (await_ready
* returns false) and increments nSuspend_ in await_suspend().
*/
// The C++ coroutine protocol mandates these awaiter names and
// instance-callable methods, which conflict with the project
// naming/static conventions.
// NOLINTBEGIN(readability-identifier-naming, readability-convert-member-functions-to-static)
struct SuspendAwaiter
{
CoroTaskRunner& runner_; // The runner that owns this coroutine.
/**
* Always returns false so the coroutine suspends.
*/
[[nodiscard]] bool
await_ready() const noexcept
{
return false;
}
/**
* Called when the coroutine suspends. Increments nSuspend_
* so the JobQueue knows a coroutine is waiting.
*/
void
await_suspend(std::coroutine_handle<>) const
{
runner_.onSuspend();
}
void
await_resume() const noexcept
{
}
};
// NOLINTEND(readability-identifier-naming, readability-convert-member-functions-to-static)
return SuspendAwaiter{*this};
}
/**
* Suspend and immediately repost on the JobQueue. Equivalent to
* `co_await JobQueueAwaiter{runner}` but uses an inline struct
* to work around a GCC-12 codegen bug (see declaration in JobQueue.h).
*
* If the JobQueue is stopping (post fails), the suspend count is
* undone and the coroutine continues immediately via symmetric
* transfer back to its own handle.
*
* @return An inline YieldPostAwaiter
*/
inline auto
JobQueue::CoroTaskRunner::yieldAndPost()
{
// The C++ coroutine protocol mandates these awaiter names and
// instance-callable methods, which conflict with the project
// naming/static conventions.
// NOLINTBEGIN(readability-identifier-naming, readability-convert-member-functions-to-static)
struct YieldPostAwaiter
{
CoroTaskRunner& runner_;
[[nodiscard]] bool
await_ready() const noexcept
{
return false;
}
/**
* Returns a coroutine_handle<> (symmetric transfer) rather than
* void + h.resume(). Two reasons:
*
* 1. h.resume() runs the coroutine nested inside this frame. A
* coroutine that yields in a loop against a stopping JobQueue
* fails post() every iteration, so the stack grows without
* bound. Symmetric transfer is a tail call and does not nest.
*
* 2. After h.resume() returns, the coroutine may have completed
* and destroyed its frame -- the frame this awaiter lives in.
* Returning from await_suspend would then touch freed memory.
*
* A bool return would also avoid nesting, but GCC-12 miscompiles
* bool-returning await_suspend (see JobQueueAwaiter.h).
*
* @return noop_coroutine() to stay suspended (job posted);
* the caller's handle to continue now (JQ stopping)
*/
std::coroutine_handle<>
await_suspend(std::coroutine_handle<> h)
{
runner_.onSuspend();
if (!runner_.post())
{
runner_.onUndoSuspend();
return h;
}
return std::noop_coroutine();
}
void
await_resume() const noexcept
{
}
};
// NOLINTEND(readability-identifier-naming, readability-convert-member-functions-to-static)
return YieldPostAwaiter{*this};
}
/**
* Schedule coroutine resumption as a job on the JobQueue.
* A shared_ptr capture (sp) prevents this CoroTaskRunner from being
* destroyed while the job is queued but not yet executed.
*
* @return false if the JobQueue rejected the job (shutting down)
*/
inline bool
JobQueue::CoroTaskRunner::post()
{
{
std::scoped_lock const lk(mutexRun_);
++runCount_;
}
// sp prevents 'this' from being destroyed while the job is pending
if (jq_.addJob(type_, name_, [this, sp = shared_from_this()]() { resume(); }))
{
return true;
}
// The coroutine will not run. Undo the runCount_ increment.
std::scoped_lock const lk(mutexRun_);
--runCount_;
cv_.notify_all();
return false;
}
/**
* Resume the coroutine on the current thread.
*
* Steps:
* 1. Decrement nSuspend_ (under jq_.mutex_)
* 2. Swap in this coroutine's LocalValues for thread-local isolation
* 3. Resume the coroutine handle (under mutex_)
* 4. Swap out LocalValues, restoring the thread's previous state
* 5. Decrement runCount_ and notify join() waiters
*
* @pre post() must have been called before resume(). Direct calls
* without a prior post() will corrupt runCount_ and break join().
* Note: runCount_ is NOT incremented here — post() already did that.
* This ensures join() stays blocked for the entire post->resume lifetime.
*/
inline void
JobQueue::CoroTaskRunner::resume()
{
{
std::scoped_lock const lock(jq_.mutex_);
--jq_.nSuspend_;
}
auto saved = detail::getLocalValues().release();
detail::getLocalValues().reset(&lvs_);
std::scoped_lock const lock(mutex_);
XRPL_ASSERT(
task_.handle() && !task_.done(),
"xrpl::JobQueue::CoroTaskRunner::resume : task handle is valid and not done");
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_
// still holds the old value -- a re-entrancy hazard on GCC-12 if
// frame destruction triggers runner cleanup.
[[maybe_unused]] auto completed = std::move(task_);
}
std::scoped_lock const lk(mutexRun_);
--runCount_;
cv_.notify_all();
}
/**
* @return true if the coroutine has not yet run to completion
*/
inline bool
JobQueue::CoroTaskRunner::runnable() const
{
// After normal completion, task_ is reset to break the shared_ptr cycle
// (handle_ becomes null). A null handle means the coroutine is done.
return task_.handle() && !task_.done();
}
/**
* Handle early termination when the coroutine never ran (e.g. JobQueue
* is stopping). Decrements nSuspend_ and destroys the coroutine frame
* to break the shared_ptr cycle: frame -> lambda -> runner -> frame.
*/
inline void
JobQueue::CoroTaskRunner::expectEarlyExit()
{
if (!finished_)
{
std::scoped_lock const lock(jq_.mutex_);
--jq_.nSuspend_;
finished_ = true;
}
// Break the shared_ptr cycle: frame -> shared_ptr<runner> -> this.
// The coroutine is at initial_suspend and never ran user code, so
// destroying it is safe. Use std::move (not task_ = {}) so
// task_.handle_ is null before the frame is destroyed.
{
[[maybe_unused]] auto completed = std::move(task_);
}
storedFunc_.reset();
}
/**
* Block until all pending/active resume operations complete.
* Uses cv_ + mutexRun_ to wait until runCount_ reaches 0 or
* finished_ becomes true. The finished_ check handles the case
* where resume() is called directly (without post()), which
* 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()
{
std::unique_lock<std::mutex> lk(mutexRun_);
cv_.wait(lk, [this]() { return runCount_ == 0 || finished_; });
}
} // namespace xrpl

View File

@@ -2,7 +2,6 @@
#include <xrpl/basics/LocalValue.h>
#include <xrpl/core/ClosureCounter.h>
#include <xrpl/core/CoroTask.h>
#include <xrpl/core/JobTypeData.h>
#include <xrpl/core/detail/Workers.h>
#include <xrpl/json/json_value.h>
@@ -153,420 +152,6 @@ public:
join();
};
/**
* C++20 coroutine lifecycle manager. Replaces Coro for new code.
*
* Class / Inheritance / Dependency Diagram
* =========================================
*
* std::enable_shared_from_this<CoroTaskRunner>
* ^
* | (public inheritance)
* |
* CoroTaskRunner
* +---------------------------------------------------+
* | - lvs_ : detail::LocalValues |
* | - jq_ : JobQueue& |
* | - type_ : JobType |
* | - name_ : std::string |
* | - runCount_ : int (in-flight resumes) |
* | - mutex_ : std::mutex (coroutine guard) |
* | - mutexRun_ : std::mutex (join guard) |
* | - cv_ : condition_variable |
* | - task_ : CoroTask<void> |
* | - storedFunc_ : unique_ptr<FuncBase> (type-erased)|
* +---------------------------------------------------+
* | + init(F&&) : set up coroutine callable |
* | + onSuspend() : ++jq_.nSuspend_ |
* | + onUndoSuspend() : --jq_.nSuspend_ |
* | + suspend() : returns SuspendAwaiter |
* | + post() : schedule resume on JobQueue |
* | + resume() : resume coroutine on caller |
* | + runnable() : !task_.done() |
* | + expectEarlyExit() : teardown for failed post |
* | + join() : block until not running |
* +---------------------------------------------------+
* | |
* | owns | references
* v v
* CoroTask<void> JobQueue
* (coroutine frame) (thread pool + nSuspend_)
*
* FuncBase / FuncStore<F> (type-erased heap storage
* for the coroutine lambda)
*
* Coroutine Lifecycle (Control Flow)
* ===================================
*
* Caller thread JobQueue worker thread
* ------------- ----------------------
* postCoroTask(f)
* |
* +-- reserve a jobCounter_ slot (reject if JQ shutting down)
* +-- ++nSuspend_ (lazy start counts as suspended)
* +-- make_shared<CoroTaskRunner>
* +-- init(f)
* | +-- store lambda on heap (FuncStore)
* | +-- task_ = f(shared_from_this())
* | [coroutine created, suspended at initial_suspend]
* +-- post()
* | +-- ++runCount_
* | +-- addJob(type_, [resume]{})
* | resume()
* | |
* | +-- --nSuspend_
* | +-- swap in LocalValues
* | +-- task_.handle().resume()
* | | [coroutine body runs]
* | | ...
* | | co_await suspend()
* | | +-- ++nSuspend_
* | | [coroutine suspends]
* | +-- swap out LocalValues
* | +-- --runCount_
* | +-- cv_.notify_all()
* |
* post() <-- called externally or by yieldAndPost()
* +-- ++runCount_
* +-- addJob(type_, [resume]{})
* resume()
* |
* +-- [coroutine body continues]
* +-- co_return
* +-- --runCount_
* +-- cv_.notify_all()
* join()
* +-- cv_.wait([]{runCount_ == 0})
* +-- [done]
*
* Usage Examples
* ==============
*
* 1. Fire-and-forget coroutine (most common pattern):
*
* jq.postCoroTask(JtClient, "MyWork",
* [](auto runner) -> CoroTask<void> {
* doSomeWork();
* co_await runner->suspend(); // yield to other jobs
* doMoreWork();
* co_return;
* });
*
* 2. Manually controlling suspend / resume (external trigger):
*
* auto runner = jq.postCoroTask(JtClient, "ExtTrigger",
* [&result](auto runner) -> CoroTask<void> {
* startAsyncOperation(callback);
* co_await runner->suspend();
* // callback called runner->post() to get here
* result = collectResult();
* co_return;
* });
* // ... later, from the callback:
* runner->post(); // reschedule the coroutine on the JobQueue
*
* 3. Using yieldAndPost() for automatic suspend + repost:
*
* jq.postCoroTask(JtClient, "AutoRepost",
* [](auto runner) -> CoroTask<void> {
* step1();
* co_await runner->yieldAndPost(); // yield + auto-repost
* step2();
* co_await runner->yieldAndPost();
* step3();
* co_return;
* });
*
* 4. Checking shutdown after co_await (cooperative cancellation):
*
* jq.postCoroTask(JtClient, "Cancellable",
* [&jq](auto runner) -> CoroTask<void> {
* while (moreWork()) {
* co_await runner->yieldAndPost();
* if (jq.isStopping())
* co_return; // bail out cleanly
* processNextItem();
* }
* co_return;
* });
*
* Caveats / Pitfalls
* ==================
*
* BUG-RISK: Calling suspend() without a matching post()/resume().
* After co_await runner->suspend(), the coroutine is parked and
* nSuspend_ is incremented. If nothing ever calls post() or
* resume(), the coroutine is leaked and JobQueue::stop() will
* hang forever waiting for nSuspend_ to reach zero.
*
* BUG-RISK: Calling post() on an already-running coroutine.
* post() schedules a resume() job. If the coroutine has not
* actually suspended yet (no co_await executed), the resume job
* will try to call handle().resume() while the coroutine is still
* running on another thread. This is UB. The mutex_ prevents
* data corruption but the logic is wrong — always co_await
* suspend() before calling post(). (The test incorrect_order()
* shows this works only because mutex_ serializes the calls.)
*
* BUG-RISK: Dropping the shared_ptr<CoroTaskRunner> before join().
* The CoroTaskRunner destructor asserts that finished_ is true
* (the coroutine completed). If you let the last shared_ptr die
* while the coroutine is still running or suspended, you get an
* assertion failure in debug and UB in release. Always call
* join() or expectEarlyExit() first.
*
* BUG-RISK: Lambda captures outliving the coroutine frame.
* The lambda passed to postCoroTask is heap-allocated (FuncStore)
* to prevent dangling. But objects captured by pointer still need
* their own lifetime management. If you capture a raw pointer to
* a stack variable, and the stack frame exits before the coroutine
* finishes, the pointer dangles. Use shared_ptr or ensure the
* pointed-to object outlives the coroutine.
*
* BUG-RISK: Forgetting co_return in a void coroutine.
* If the coroutine body falls off the end without co_return,
* the compiler may silently treat it as co_return (per standard),
* but some compilers warn. Always write explicit co_return.
*
* LIMITATION: CoroTaskRunner only supports CoroTask<void>.
* The task_ member is CoroTask<void>. To return values from
* the top-level coroutine, write through a captured pointer
* (as the tests demonstrate), or co_await inner CoroTask<T>
* coroutines that return values.
*
* LIMITATION: One coroutine per CoroTaskRunner.
* init() must be called exactly once. You cannot reuse a
* CoroTaskRunner to run a second coroutine. Create a new one
* via postCoroTask() instead.
*
* LIMITATION: No timeout on join().
* join() blocks indefinitely. If the coroutine is suspended
* and never posted, join() will deadlock. Use timed waits
* on the gate pattern (condition_variable + wait_for) in tests.
*/
class CoroTaskRunner : public std::enable_shared_from_this<CoroTaskRunner>
{
private:
// Per-coroutine thread-local storage. Swapped in before resume()
// and swapped out after, so each coroutine sees its own LocalValue
// state regardless of which worker thread executes it.
detail::LocalValues lvs_;
// Back-reference to the owning JobQueue. Used to post jobs,
// increment/decrement nSuspend_, and acquire jq_.mutex_.
JobQueue& jq_;
// Job type passed to addJob() when posting this coroutine.
JobType type_;
// Human-readable name for this coroutine job (for logging).
std::string name_;
// Number of in-flight resume operations (pending + active).
// Incremented by post(), decremented when resume() finishes.
// Guarded by mutexRun_. join() blocks until this reaches 0.
//
// A counter (not a bool) is needed because post() can be called
// from within the coroutine body (e.g. via yieldAndPost()),
// enqueuing a second resume while the first is still running.
// A bool would be clobbered: R2.post() sets true, then R1's
// cleanup sets false — losing the fact that R2 is still pending.
int runCount_ = 0;
// Serializes all coroutine resume() calls, preventing concurrent
// execution of the coroutine body on multiple threads. Handles the
// race where post() enqueues a resume before the coroutine has
// actually suspended (post-before-suspend pattern).
std::mutex mutex_;
// Guards runCount_. Used with cv_ for join() to wait
// until all pending/active resume operations complete.
std::mutex mutexRun_;
// Notified when runCount_ reaches zero, allowing
// join() waiters to wake up.
std::condition_variable cv_;
// The coroutine handle wrapper. Owns the coroutine frame.
// Set by init(). Reset to empty in resume() upon coroutine
// completion (to break the shared_ptr cycle) or in
// expectEarlyExit() on early termination.
CoroTask<void> task_;
/**
* Type-erased base for heap-stored callables.
* Prevents the coroutine lambda from being destroyed before
* the coroutine frame is done with it.
*
* @see FuncStore
*/
struct FuncBase
{
virtual ~FuncBase() = default;
};
/**
* Concrete type-erased storage for a callable of type F.
* The coroutine frame stores a reference to the lambda's implicit
* object parameter. If the lambda is a temporary, that reference
* dangles after the call returns. FuncStore keeps it alive on
* the heap for the lifetime of the CoroTaskRunner.
*/
template <class F>
struct FuncStore : FuncBase
{
F func; // The stored callable (coroutine lambda).
explicit FuncStore(F&& f) : func(std::move(f))
{
}
};
// Heap-allocated callable storage. Set by init(), ensures the
// lambda outlives the coroutine frame that references it.
std::unique_ptr<FuncBase> storedFunc_;
// True once the coroutine has completed or expectEarlyExit() was
// called. Asserted in the destructor (debug) to catch leaked
// runners. Available in all builds to guard expectEarlyExit()
// against double-decrementing nSuspend_.
bool finished_ = false;
public:
/**
* Tag type for private construction. Prevents external code
* from constructing CoroTaskRunner directly. Use postCoroTask().
*/
struct CreateT
{
explicit CreateT() = default;
};
/**
* Construct a CoroTaskRunner. Private by convention (CreateT tag).
*
* @param jq The JobQueue this coroutine will run on
* @param type Job type for scheduling priority
* @param name Human-readable name for logging
*/
CoroTaskRunner(CreateT, JobQueue&, JobType, std::string);
CoroTaskRunner(CoroTaskRunner const&) = delete;
CoroTaskRunner&
operator=(CoroTaskRunner const&) = delete;
/**
* Destructor. Asserts (debug) that the coroutine has finished
* or expectEarlyExit() was called.
*/
~CoroTaskRunner();
/**
* Initialize with a coroutine-returning callable.
* Must be called exactly once, after the object is managed by
* shared_ptr (because init uses shared_from_this internally).
* This is handled automatically by postCoroTask().
*
* @param f Callable: CoroTask<void>(shared_ptr<CoroTaskRunner>)
*/
template <class F>
void
init(F&& f);
/**
* Increment the JobQueue's suspended-coroutine count (nSuspend_).
* Called when the coroutine is about to suspend. Every call
* must be balanced by a corresponding decrement (via resume()
* or onUndoSuspend()), or JobQueue::stop() will hang.
*/
void
onSuspend();
/**
* Decrement nSuspend_ without resuming.
* Used to undo onSuspend() when a scheduled post() fails
* (e.g. JobQueue is stopping).
*/
void
onUndoSuspend();
/**
* Suspend the coroutine.
* The awaiter's await_suspend() increments nSuspend_ before the
* coroutine actually suspends. The caller must later call post()
* or resume() to continue execution.
*
* @return An awaiter for use with `co_await runner->suspend()`
*/
auto
suspend();
/**
* Suspend the coroutine and immediately repost it on the
* JobQueue. Combines suspend() + post() atomically inside
* await_suspend, so there is no window where an external
* event could race between the two.
*
* Equivalent to JobQueueAwaiter but defined as an inline
* awaiter returned from a member function. This avoids a
* GCC-12 coroutine codegen bug where an external awaiter
* struct (JobQueueAwaiter) used at multiple co_await points
* corrupts the coroutine state machine's resume index,
* causing the coroutine to hang on the third resumption.
*
* @return An awaiter for use with `co_await runner->yieldAndPost()`
*/
auto
yieldAndPost();
/**
* Schedule coroutine resumption as a job on the JobQueue.
* Captures shared_from_this() to prevent this runner from being
* destroyed while the job is queued.
*
* @return true if the job was accepted; false if the JobQueue
* is stopping (caller must handle cleanup)
*/
bool
post();
/**
* Resume the coroutine on the current thread.
* Decrements nSuspend_, swaps in LocalValues, resumes the
* coroutine handle, swaps out LocalValues, and notifies join()
* waiters. Lock ordering (sequential, non-overlapping):
* jq_.mutex_ -> mutex_ -> mutexRun_.
*
* @pre post() must have been called before resume(). Direct
* calls without a prior post() will corrupt runCount_
* and break join().
*/
void
resume();
/**
* @return true if the coroutine has not yet run to completion
*/
bool
runnable() const;
/**
* Handle early termination when the coroutine never ran.
* Decrements nSuspend_ and destroys the coroutine frame to
* break the shared_ptr cycle (frame -> lambda -> runner -> frame).
* Called by postCoroTask() when post() fails.
*/
void
expectEarlyExit();
/**
* Block until all pending/active resume operations complete.
* Uses cv_ + mutexRun_ to wait until runCount_ reaches 0.
* Warning: deadlocks if the coroutine is suspended and never posted.
*/
void
join();
};
using JobFunction = std::function<void()>;
JobQueue(
@@ -612,20 +197,6 @@ public:
std::shared_ptr<Coro>
postCoro(JobType t, std::string const& name, F&& f);
/**
* Creates a C++20 coroutine and adds a job to the queue to run it.
*
* @param t The type of job.
* @param name Name of the job.
* @param f Callable with signature
* CoroTask<void>(std::shared_ptr<CoroTaskRunner>).
*
* @return shared_ptr to posted CoroTaskRunner. nullptr if not successful.
*/
template <class F>
std::shared_ptr<CoroTaskRunner>
postCoroTask(JobType t, std::string const& name, F&& f);
/**
* Jobs waiting at this priority.
*/
@@ -848,8 +419,7 @@ private:
} // namespace xrpl
#include <xrpl/core/Coro.ipp> // IWYU pragma: keep
#include <xrpl/core/CoroTaskRunner.ipp> // IWYU pragma: keep
#include <xrpl/core/Coro.ipp> // IWYU pragma: keep
namespace xrpl {
@@ -872,82 +442,4 @@ JobQueue::postCoro(JobType t, std::string const& name, F&& f)
return coro;
}
// postCoroTask — entry point for launching a C++20 coroutine on the JobQueue.
//
// Control Flow
// ============
//
// postCoroTask(t, name, f)
// |
// +-- 1. Reserve a jobCounter_ slot — reject if JQ shutting down
// |
// +-- 2. ++nSuspend_ (mirrors Boost Coro ctor's implicit yield)
// | The coroutine is "suspended" from the JobQueue's perspective
// | even though it hasn't run yet — this keeps the JQ shutdown
// | logic correct (it waits for nSuspend_ to reach 0).
// |
// +-- 3. Create CoroTaskRunner (shared_ptr, ref-counted)
// |
// +-- 4. runner->init(f)
// | +-- Heap-allocate the lambda (FuncStore) to prevent
// | | dangling captures in the coroutine frame
// | +-- task_ = f(shared_from_this())
// | [coroutine created but NOT started — lazy initial_suspend]
// |
// +-- 5. runner->post()
// | +-- addJob(type_, [resume]{}) → resume on worker thread
// | +-- failure (JQ stopping):
// | +-- runner->expectEarlyExit()
// | | --nSuspend_, destroy coroutine frame
// | +-- return nullptr
// |
// +-- 6. Release the jobCounter_ slot (on return)
//
// Why reserve a jobCounter_ slot?
// ===============================
// JobQueue::stop() joins jobCounter_ before it acquires mutex_ and asserts
// nSuspend_ == 0. Without a reservation, stop() could run to completion in
// the window between the ++nSuspend_ in step 2 and the balancing post() or
// expectEarlyExit() in step 5, tripping that assert. Holding a slot blocks
// stop()'s join() for the whole function, closing the window. wrap() also
// returns nullopt once the counter is joined, so it doubles as the shutdown
// check — a plain stopping_ read cannot do this, because the read and the
// ++nSuspend_ are not a single atomic step.
//
// Why async post() instead of synchronous resume()?
// ==================================================
// The initial dispatch MUST use async post() so the coroutine body runs on
// a JobQueue worker thread, not the caller's thread. resume() swaps the
// caller's thread-local LocalValues with the coroutine's private copy.
// If the coroutine mutates LocalValues (e.g. thread_specific_storage test),
// those mutations bleed back into the caller's thread-local state after the
// swap-out, corrupting subsequent tests that share the same thread pool.
// Async post() avoids this by running the coroutine on a worker thread whose
// LocalValues are managed by the thread pool, not by the caller.
//
template <class F>
std::shared_ptr<JobQueue::CoroTaskRunner>
JobQueue::postCoroTask(JobType t, std::string const& name, F&& f)
{
// Held until this function returns. Null once jobCounter_ is joined,
// which is how a shutting-down JobQueue rejects new coroutines.
auto const shutdownGuard = jobCounter_.wrap([]() {});
if (!shutdownGuard)
return nullptr;
{
std::scoped_lock const lock(mutex_);
++nSuspend_;
}
auto runner = std::make_shared<CoroTaskRunner>(CoroTaskRunner::CreateT{}, *this, t, name);
runner->init(std::forward<F>(f));
if (!runner->post())
{
runner->expectEarlyExit();
runner.reset();
}
return runner;
}
} // namespace xrpl

View File

@@ -1,212 +0,0 @@
#pragma once
#include <xrpl/beast/utility/instrumentation.h>
#include <xrpl/core/JobQueue.h>
#include <coroutine>
#include <memory>
namespace xrpl {
/**
* Awaiter that suspends and immediately reschedules on the JobQueue.
* Equivalent to calling yield() followed by post() in the old Coro API.
*
* Usage:
* co_await JobQueueAwaiter{runner};
*
* What it waits for: The coroutine is re-queued as a job and resumes
* when a worker thread picks it up.
*
* Which thread resumes: A JobQueue worker thread.
*
* What await_resume() returns: void.
*
* Dependency Diagram
* ==================
*
* JobQueueAwaiter
* +----------------------------------------------+
* | + runner : shared_ptr<CoroTaskRunner> |
* +----------------------------------------------+
* | + await_ready() -> false (always suspend) |
* | + await_suspend() -> bool (suspend or cancel) |
* | + await_resume() -> void |
* +----------------------------------------------+
* | |
* | uses | uses
* v v
* CoroTaskRunner JobQueue
* .onSuspend() (via runner->post() -> addJob)
* .onUndoSuspend()
* .post()
*
* Control Flow (await_suspend)
* ============================
*
* co_await JobQueueAwaiter{runner}
* |
* +-- await_ready() -> false
* +-- await_suspend(handle)
* |
* +-- runner->onSuspend() // ++nSuspend_
* +-- runner->post() // addJob to JobQueue
* | |
* | +-- success? return noop_coroutine()
* | | // coroutine stays suspended;
* | | // worker thread will call resume()
* | +-- failure? (JQ stopping)
* | +-- runner->onUndoSuspend() // --nSuspend_
* | +-- return handle // symmetric transfer back
* | // coroutine continues immediately
* | // so it can clean up and co_return
*
* DEPRECATED — prefer `co_await runner->yieldAndPost()`
* =====================================================
*
* GCC-12 has a coroutine codegen bug where using this external awaiter
* struct at multiple co_await points in the same coroutine corrupts the
* state machine's resume index. After the second co_await, the third
* resumption enters handle().resume() but never reaches await_resume()
* or any subsequent user code — the coroutine hangs indefinitely.
*
* The fix is `co_await runner->yieldAndPost()`, which defines the
* awaiter as an inline struct inside a CoroTaskRunner member function.
* GCC-12 handles inline awaiters correctly at multiple co_await points.
*
* This struct is retained for single-use scenarios and documentation
* purposes. For any code that may use co_await in a loop or at
* multiple points, always use `runner->yieldAndPost()`.
*
* Usage Examples
* ==============
*
* 1. Yield and auto-repost (preferred — works on all compilers):
*
* CoroTask<void> handler(auto runner) {
* doPartA();
* co_await runner->yieldAndPost(); // yield + repost
* doPartB(); // runs on a worker thread
* co_return;
* }
*
* 2. Multiple yield points in a loop:
*
* CoroTask<void> batchProcessor(auto runner) {
* for (auto& item : items) {
* process(item);
* co_await runner->yieldAndPost(); // let other jobs run
* }
* co_return;
* }
*
* 3. Graceful shutdown — checking after resume:
*
* CoroTask<void> longTask(auto runner, JobQueue& jq) {
* while (hasWork()) {
* co_await runner->yieldAndPost();
* // If JQ is stopping, await_suspend resumes the coroutine
* // immediately without re-queuing. Always check
* // isStopping() to decide whether to proceed:
* if (jq.isStopping())
* co_return;
* doNextChunk();
* }
* co_return;
* }
*
* Caveats / Pitfalls
* ==================
*
* BUG-RISK: Using a stale or null runner.
* The runner shared_ptr must be valid and point to the CoroTaskRunner
* that owns the coroutine currently executing. Passing a runner from
* a different coroutine, or a default-constructed shared_ptr, is UB.
*
* BUG-RISK: Assuming resume happens on the same thread.
* After co_await, the coroutine resumes on whatever worker thread
* picks up the job. Do not rely on thread-local state unless it is
* managed through LocalValue (which CoroTaskRunner automatically
* swaps in/out).
*
* BUG-RISK: Ignoring the shutdown path.
* When the JobQueue is stopping, post() fails and await_suspend()
* resumes the coroutine immediately (symmetric transfer back to h).
* The coroutine body continues on the same thread. If your code
* after co_await assumes it was re-queued and is running on a worker
* thread, that assumption breaks during shutdown. Always handle the
* "JQ is stopping" case, either by checking jq.isStopping() or by
* letting the coroutine fall through to co_return naturally.
*
* DIFFERENCE from runner->suspend() + runner->post():
* Both JobQueueAwaiter and yieldAndPost() combine suspend + post
* in one atomic operation. With the manual suspend()/post() pattern,
* there is a window between the two calls where an external event
* could race. The atomic awaiters remove that window — onSuspend()
* and post() happen within the same await_suspend() call while the
* coroutine is guaranteed to be suspended. Use yieldAndPost() unless
* you need an external party to decide *when* to call post().
*/
struct JobQueueAwaiter
{
// The CoroTaskRunner that owns the currently executing coroutine.
std::shared_ptr<JobQueue::CoroTaskRunner> runner;
/**
* Always returns false so the coroutine suspends.
*/
// The C++ coroutine protocol mandates these awaiter names and
// instance-callable methods, which conflict with the project
// naming/static conventions.
// NOLINTBEGIN(readability-identifier-naming, readability-convert-member-functions-to-static)
[[nodiscard]] bool
await_ready() const noexcept
{
return false;
}
/**
* Increment nSuspend (equivalent to yield()) and schedule resume
* on the JobQueue (equivalent to post()). If the JobQueue is
* stopping, undoes the suspend count and transfers back to the
* coroutine so it can clean up and co_return.
*
* Returns a coroutine_handle<> (symmetric transfer) instead of
* bool to work around a GCC-12 codegen bug where bool-returning
* await_suspend leaves the coroutine in an invalid state —
* neither properly suspended nor resumed — causing a hang.
*
* WARNING: GCC-12 has an additional codegen bug where using this
* external awaiter struct at multiple co_await points in the same
* coroutine corrupts the state machine's resume index, causing the
* coroutine to hang on the third resumption. Prefer
* `co_await runner->yieldAndPost()` which uses an inline awaiter
* that GCC-12 handles correctly.
*
* @return noop_coroutine() to stay suspended (job posted);
* the caller's handle to resume immediately (JQ stopping)
*/
std::coroutine_handle<>
await_suspend(std::coroutine_handle<> h)
{
XRPL_ASSERT(runner, "xrpl::JobQueueAwaiter::await_suspend : runner is valid");
runner->onSuspend();
if (!runner->post())
{
// JobQueue is stopping. Undo the suspend count and
// transfer back to the coroutine so it can clean up
// and co_return.
runner->onUndoSuspend();
return h;
}
return std::noop_coroutine();
}
void
await_resume() const noexcept
{
}
// NOLINTEND(readability-identifier-naming, readability-convert-member-functions-to-static)
};
} // namespace xrpl

View File

@@ -126,15 +126,15 @@ release defaults to 1 and is overridable with `-Dpkg_release=N`.
Packages are published to the XRPLF repositories on Sonatype Nexus at
`https://packages.xrplf.org`. The `release-info` action decides the channel from
the event, and `publish_pkg.sh` maps that channel to a repository pair:
the event, and `publish_pkg.sh` maps that channel to its repositories:
| Event | Version | Channel | DEB repository | RPM repository |
| ------------------------ | ----------------- | -------------- | ------------------ | ------------------ |
| tag | `X.Y.Z` | `stable` | `deb-stable` | `rpm-stable` |
| tag | `X.Y.Z-rcN` | `unstable` | `deb-unstable` | `rpm-unstable` |
| tag | `X.Y.Z-bN` | `experimental` | `deb-experimental` | `rpm-experimental` |
| push to `develop` | `xrpld --version` | `develop` | `deb-develop` | `rpm-develop` |
| tag, non-public codebase | _any_ | `private` | `deb-private` | `rpm-private` |
| Event | Version | Channel | DEB repository | RPM upload repository |
| ------------------------ | ----------------- | -------------- | ------------------ | ------------------------- |
| tag | `X.Y.Z` | `stable` | `deb-stable` | `rpm-stable-hosted` |
| tag | `X.Y.Z-rcN` | `unstable` | `deb-unstable` | `rpm-unstable-hosted` |
| tag | `X.Y.Z-bN` | `experimental` | `deb-experimental` | `rpm-experimental-hosted` |
| push to `develop` | `xrpld --version` | `develop` | `deb-develop` | `rpm-develop-hosted` |
| tag, non-public codebase | _any_ | `private` | `deb-private` | `rpm-private-hosted` |
Only a tag names a channel — do not extend that to `develop`, where
`BuildInfo.cpp`'s `versionString` moves through `-bN`, `-rcN` and even the final
@@ -155,12 +155,15 @@ Conan remote.
Nexus owns the repository metadata; nothing here indexes anything. Worth knowing:
- Each apt-hosted repository needs a distribution and a PGP signing keypair
configured in Nexus, which rejects one created without a keypair. Nexus signs
the apt metadata with it, never the packages.
- Hosted yum repositories cannot be signed by Nexus at all, so `sign_rpm.sh`
signs the RPMs before they are uploaded, and rpm clients verify with
`gpgcheck=1` rather than `repo_gpgcheck=1`.
- Each apt-hosted repository needs a distribution (ours use `any`) and a PGP
signing keypair configured in Nexus, which rejects one created without a
keypair. Nexus signs the apt metadata with it, never the packages.
- Hosted yum repositories cannot be signed by Nexus, so each `rpm-<channel>-hosted`
repository sits behind a `rpm-<channel>` yum group repository whose metadata
Nexus signs. Uploads go to the hosted repository; clients point at the group
and verify the metadata with `repo_gpgcheck=1`. Nexus never signs the RPMs
themselves, so `sign_rpm.sh` signs them before they are uploaded, and clients
verify them with `gpgcheck=1`.
- yum metadata is rebuilt asynchronously, so a successful publish is not
immediately installable.
- Each job uploads only what it built, and uploads are not transactional, so a

View File

@@ -7,10 +7,13 @@ set -euo pipefail
# Usage: publish_pkg.sh <channel> [package-dir]
#
# channel release channel, selecting the 'deb-<channel>' and
# 'rpm-<channel>' repository pair
# 'rpm-<channel>-hosted' repositories
# package-dir searched recursively for *.deb, *.ddeb and *.rpm ('build' by
# default)
#
# RPMs are uploaded to the hosted repository, but yum clients install from the
# 'rpm-<channel>' group repository in front of it, which serves signed metadata.
#
# NEXUS_USERNAME and NEXUS_PASSWORD are required. NEXUS_URL overrides the target
# instance, and DRY_RUN=1 lists the uploads without performing them.
@@ -24,7 +27,7 @@ if [[ -z "${channel}" ]]; then
fi
deb_repo="deb-${channel}"
rpm_repo="rpm-${channel}"
rpm_repo="rpm-${channel}-hosted"
if [[ -z "${DRY_RUN:-}" ]]; then
: "${NEXUS_USERNAME:?is required}" "${NEXUS_PASSWORD:?is required}"

View File

@@ -1,9 +1,10 @@
#!/usr/bin/env bash
set -euo pipefail
# Sign the RPMs built by build_pkg.sh. Nexus cannot sign hosted yum metadata, so
# the packages carry the signature themselves and rpm clients verify them with
# gpgcheck=1.
# Sign the RPMs built by build_pkg.sh. Nexus signs the yum repository metadata
# (via the 'rpm-<channel>' group repository), but never the packages themselves,
# so they carry their own signature. Clients verify the packages with gpgcheck=1
# and the metadata with repo_gpgcheck=1.
#
# Usage: sign_rpm.sh [package-dir]
#
@@ -12,8 +13,9 @@ set -euo pipefail
# PKG_SIGNING_KEY must hold an armoured PGP private key. It has no flag, to keep
# the key out of the process list.
#
# There is no DEB equivalent: apt trusts the repository metadata, which Nexus
# signs, rather than the packages themselves.
# The DEBs are deliberately not signed: embedded DEB signatures exist (debsigs),
# but apt does not verify them by default and trusts the repository metadata,
# which Nexus signs, instead.
pkg_dir="${1:-build}"

5
src/libxrpl/tx/AGENTS.md Normal file
View File

@@ -0,0 +1,5 @@
# AGENTS.md — tx
See the repo-level [AGENTS.md](../../../AGENTS.md) for general build/test/style guidance. `CLAUDE.md` in this directory is a symlink to this file.
Any change to transaction-processing behavior must be gated behind an amendment. New amendments (and fixes, i.e. `fix*` amendments) are added to [`include/xrpl/protocol/detail/features.macro`](../../../include/xrpl/protocol/detail/features.macro), as an `XRPL_FEATURE(...)` or `XRPL_FIX(...)` entry added to the top of the list (the list is kept in reverse chronological order).

1
src/libxrpl/tx/CLAUDE.md Symbolic link
View File

@@ -0,0 +1 @@
AGENTS.md

View File

@@ -19,7 +19,6 @@
#include <xrpl/basics/base_uint.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/json/json_value.h>
@@ -124,6 +123,7 @@ public:
.ledgerMaster = app.getLedgerMaster(),
.consumer = c,
.role = Role::USER,
.coro = {},
.infoSub = {},
.apiVersion = rpc::kApiVersionIfUnspecified},
{},
@@ -135,28 +135,22 @@ public:
numSrc.reserve(rpc::tuning::kMaxSrcCur);
for (std::uint8_t i = 0; i < rpc::tuning::kMaxSrcCur; ++i)
numSrc.push_back(makeMptID(i, bob));
// Safe capture: the test blocks on g.waitFor() until the coroutine
// completes, so the captured locals outlive the coroutine.
// NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
app.getJobQueue().postCoroTask(JtClient, "RPC-Client", [&](auto) -> CoroTask<void> {
app.getJobQueue().postCoro(JtClient, "RPC-Client", [&](auto const& coro) {
context.params = xrpl::test::detail::rpf(alice, bob, usd, numSrc);
context.coro = coro;
rpc::doCommand(context, result);
g.signal();
co_return;
});
BEAST_EXPECT(g.waitFor(5s));
BEAST_EXPECT(!result.isMember(jss::error));
// Test more than rpc::tuning::max_src_cur source currencies.
numSrc.push_back(makeMptID(rpc::tuning::kMaxSrcCur, bob));
// Safe capture: the test blocks on g.waitFor() until the coroutine
// completes, so the captured locals outlive the coroutine.
// NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
app.getJobQueue().postCoroTask(JtClient, "RPC-Client", [&](auto) -> CoroTask<void> {
app.getJobQueue().postCoro(JtClient, "RPC-Client", [&](auto const& coro) {
context.params = xrpl::test::detail::rpf(alice, bob, usd, numSrc);
context.coro = coro;
rpc::doCommand(context, result);
g.signal();
co_return;
});
BEAST_EXPECT(g.waitFor(5s));
BEAST_EXPECT(result.isMember(jss::error));
@@ -168,28 +162,22 @@ public:
auto curm = MPTTester({.env = env, .issuer = alice, .holders = {bob}});
numSrc.push_back(curm.issuanceID());
}
// Safe capture: the test blocks on g.waitFor() until the coroutine
// completes, so the captured locals outlive the coroutine.
// NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
app.getJobQueue().postCoroTask(JtClient, "RPC-Client", [&](auto) -> CoroTask<void> {
app.getJobQueue().postCoro(JtClient, "RPC-Client", [&](auto const& coro) {
context.params = xrpl::test::detail::rpf(alice, bob, usd, {});
context.coro = coro;
rpc::doCommand(context, result);
g.signal();
co_return;
});
BEAST_EXPECT(g.waitFor(5s));
BEAST_EXPECT(!result.isMember(jss::error));
// Test more than rpc::tuning::max_auto_src_cur source currencies.
auto curm = MPTTester({.env = env, .issuer = alice, .holders = {bob}});
// Safe capture: the test blocks on g.waitFor() until the coroutine
// completes, so the captured locals outlive the coroutine.
// NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
app.getJobQueue().postCoroTask(JtClient, "RPC-Client", [&](auto) -> CoroTask<void> {
app.getJobQueue().postCoro(JtClient, "RPC-Client", [&](auto const& coro) {
context.params = xrpl::test::detail::rpf(alice, bob, usd, {});
context.coro = coro;
rpc::doCommand(context, result);
g.signal();
co_return;
});
BEAST_EXPECT(g.waitFor(5s));
BEAST_EXPECT(result.isMember(jss::error));

View File

@@ -26,7 +26,6 @@
#include <xrpl/basics/base_uint.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/json/json_reader.h>
@@ -164,6 +163,7 @@ public:
.ledgerMaster = app.getLedgerMaster(),
.consumer = c,
.role = Role::USER,
.coro = {},
.infoSub = {},
.apiVersion = rpc::kApiVersionIfUnspecified},
{},
@@ -188,14 +188,11 @@ public:
json::Value result;
Gate g;
// Safe capture: the test blocks on g.waitFor() until the coroutine
// completes, so the captured locals outlive the coroutine.
// NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
app.getJobQueue().postCoroTask(JtClient, "RPC-Client", [&](auto) -> CoroTask<void> {
app.getJobQueue().postCoro(JtClient, "RPC-Client", [&](auto const& coro) {
context.params = std::move(params);
context.coro = coro;
rpc::doCommand(context, result);
g.signal();
co_return;
});
using namespace std::chrono_literals;
@@ -277,6 +274,7 @@ public:
.ledgerMaster = app.getLedgerMaster(),
.consumer = c,
.role = Role::USER,
.coro = {},
.infoSub = {},
.apiVersion = rpc::kApiVersionIfUnspecified},
{},
@@ -284,27 +282,21 @@ public:
json::Value result;
Gate g;
// Test rpc::tuning::max_src_cur source currencies.
// Safe capture: the test blocks on g.waitFor() until the coroutine
// completes, so the captured locals outlive the coroutine.
// NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
app.getJobQueue().postCoroTask(JtClient, "RPC-Client", [&](auto) -> CoroTask<void> {
app.getJobQueue().postCoro(JtClient, "RPC-Client", [&](auto const& coro) {
context.params = rpf(Account("alice"), Account("bob"), rpc::tuning::kMaxSrcCur);
context.coro = coro;
rpc::doCommand(context, result);
g.signal();
co_return;
});
BEAST_EXPECT(g.waitFor(5s));
BEAST_EXPECT(!result.isMember(jss::error));
// Test more than rpc::tuning::max_src_cur source currencies.
// Safe capture: the test blocks on g.waitFor() until the coroutine
// completes, so the captured locals outlive the coroutine.
// NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
app.getJobQueue().postCoroTask(JtClient, "RPC-Client", [&](auto) -> CoroTask<void> {
app.getJobQueue().postCoro(JtClient, "RPC-Client", [&](auto const& coro) {
context.params = rpf(Account("alice"), Account("bob"), rpc::tuning::kMaxSrcCur + 1);
context.coro = coro;
rpc::doCommand(context, result);
g.signal();
co_return;
});
BEAST_EXPECT(g.waitFor(5s));
BEAST_EXPECT(result.isMember(jss::error));
@@ -312,28 +304,22 @@ public:
// Test rpc::tuning::max_auto_src_cur source currencies.
for (auto i = 0; i < (rpc::tuning::kMaxAutoSrcCur - 1); ++i)
env.trust(Account("alice")[std::to_string(i + 100)](100), "bob");
// Safe capture: the test blocks on g.waitFor() until the coroutine
// completes, so the captured locals outlive the coroutine.
// NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
app.getJobQueue().postCoroTask(JtClient, "RPC-Client", [&](auto) -> CoroTask<void> {
app.getJobQueue().postCoro(JtClient, "RPC-Client", [&](auto const& coro) {
context.params = rpf(Account("alice"), Account("bob"), 0);
context.coro = coro;
rpc::doCommand(context, result);
g.signal();
co_return;
});
BEAST_EXPECT(g.waitFor(5s));
BEAST_EXPECT(!result.isMember(jss::error));
// Test more than rpc::tuning::max_auto_src_cur source currencies.
env.trust(Account("alice")["AUD"](100), "bob");
// Safe capture: the test blocks on g.waitFor() until the coroutine
// completes, so the captured locals outlive the coroutine.
// NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
app.getJobQueue().postCoroTask(JtClient, "RPC-Client", [&](auto) -> CoroTask<void> {
app.getJobQueue().postCoro(JtClient, "RPC-Client", [&](auto const& coro) {
context.params = rpf(Account("alice"), Account("bob"), 0);
context.coro = coro;
rpc::doCommand(context, result);
g.signal();
co_return;
});
BEAST_EXPECT(g.waitFor(5s));
BEAST_EXPECT(result.isMember(jss::error));

View File

@@ -24,9 +24,7 @@
#include <xrpl/basics/strHex.h>
#include <xrpl/beast/unit_test/suite.h>
#include <xrpl/beast/utility/Zero.h>
#include <xrpl/core/CoroTask.h>
#include <xrpl/core/Job.h>
#include <xrpl/core/JobQueue.h>
#include <xrpl/core/ServiceRegistry.h>
#include <xrpl/json/json_value.h>
#include <xrpl/json/to_string.h>
@@ -1640,6 +1638,7 @@ struct PayChan_test : public beast::unit_test::Suite
.ledgerMaster = app.getLedgerMaster(),
.consumer = c,
.role = Role::USER,
.coro = {},
.infoSub = {},
.apiVersion = rpc::kApiVersionIfUnspecified},
{},
@@ -1655,13 +1654,10 @@ struct PayChan_test : public beast::unit_test::Suite
BEAST_EXPECT(context.loadType == resource::kFeeReferenceRpc);
json::Value result;
Gate g;
// Safe capture: the test blocks on g.waitFor() until the coroutine
// completes, so the captured locals outlive the coroutine.
// NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
app.getJobQueue().postCoroTask(JtClient, "RPC-Client", [&](auto) -> CoroTask<void> {
app.getJobQueue().postCoro(JtClient, "RPC-Client", [&](auto const& coro) {
context.coro = coro;
result = doChannelVerify(context);
g.signal();
co_return;
});
using namespace std::chrono_literals;
@@ -1709,6 +1705,7 @@ struct PayChan_test : public beast::unit_test::Suite
.ledgerMaster = app.getLedgerMaster(),
.consumer = c,
.role = Role::ADMIN, // channel_authorize requires ADMIN or canSign()
.coro = {},
.infoSub = {},
.apiVersion = rpc::kApiVersionIfUnspecified},
{},
@@ -1723,13 +1720,10 @@ struct PayChan_test : public beast::unit_test::Suite
BEAST_EXPECT(context.loadType == resource::kFeeReferenceRpc);
json::Value result;
Gate g;
// Safe capture: the test blocks on g.waitFor() until the coroutine
// completes, so the captured locals outlive the coroutine.
// NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
app.getJobQueue().postCoroTask(JtClient, "RPC-Client", [&](auto) -> CoroTask<void> {
app.getJobQueue().postCoro(JtClient, "RPC-Client", [&](auto const& coro) {
context.coro = coro;
result = doChannelAuthorize(context);
g.signal();
co_return;
});
using namespace std::chrono_literals;

View File

@@ -1,581 +0,0 @@
#include <test/jtx/Env.h>
#include <test/jtx/envconfig.h>
#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>
// Tests intentionally capture state in coroutine lambdas; lifetimes are
// controlled by Gate synchronization and join() before scope exit.
// NOLINTBEGIN(cppcoreguidelines-avoid-capturing-lambda-coroutines)
namespace xrpl::test {
/**
* Test suite for the C++20 coroutine primitives: CoroTask, CoroTaskRunner,
* and JobQueueAwaiter.
*
* Dependency Diagram
* ==================
*
* CoroTask_test
* +-------------------------------------------------+
* | + Gate (inner class) : condition_variable helper |
* +-------------------------------------------------+
* | uses
* v
* jtx::Env --> JobQueue::postCoroTask()
* |
* +-- CoroTaskRunner (suspend / post / resume)
* +-- CoroTask<void> / CoroTask<T>
* +-- JobQueueAwaiter
*
* Test Coverage Matrix
* ====================
*
* Test | Primitives exercised
* --------------------------+----------------------------------------------
* testVoidCompletion | CoroTask<void> basic lifecycle
* testCorrectOrder | suspend() -> join() -> post() -> complete
* testIncorrectOrder | post() before suspend() (race-safe path)
* testJobQueueAwaiter | JobQueueAwaiter + yieldAndPost suspend/repost
* testThreadSpecificStorage | LocalValue isolation across coroutines
* 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
* testValueChaining | nested CoroTask<T> -> CoroTask<T>
* testShutdownRejection | postCoroTask returns nullptr when stopping
*/
class CoroTask_test : public beast::unit_test::Suite
{
public:
/**
* Simple one-shot gate for synchronizing between test thread
* and coroutine worker threads. signal() sets the flag;
* waitFor() blocks until signaled or timeout.
*/
class Gate
{
private:
std::condition_variable cv_;
std::mutex mutex_;
bool signaled_ = false;
public:
/**
* Block until signaled or timeout expires.
*
* @param relTime Maximum duration to wait
*
* @return true if signaled before timeout
*/
template <class Rep, class Period>
bool
waitFor(std::chrono::duration<Rep, Period> const& relTime)
{
std::unique_lock<std::mutex> lk(mutex_);
auto b = cv_.wait_for(lk, relTime, [this] { return signaled_; });
signaled_ = false;
return b;
}
/**
* Signal the gate, waking any waiting thread.
*/
void
signal()
{
std::scoped_lock const lk(mutex_);
signaled_ = true;
cv_.notify_all();
}
};
// NOTE: All coroutine lambdas passed to postCoroTask use explicit
// pointer-by-value captures instead of [&] to work around a GCC 14
// bug where reference captures in coroutine lambdas are corrupted
// in the coroutine frame.
/**
* CoroTask<void> runs to completion and runner becomes non-runnable.
*/
void
testVoidCompletion()
{
using namespace std::chrono_literals;
using namespace jtx;
testcase("void completion");
Env env(*this, envconfig([](std::unique_ptr<Config> cfg) {
cfg->forceMultiThread = true;
return cfg;
}));
Gate g;
auto runner = env.app().getJobQueue().postCoroTask(
JtClient, "CoroTaskTest", [gp = &g](auto) -> CoroTask<void> {
gp->signal();
co_return;
});
BEAST_EXPECT(runner);
if (!BEAST_EXPECT(g.waitFor(5s)))
return;
runner->join();
BEAST_EXPECT(!runner->runnable());
}
/**
* Correct order: suspend, join, post, complete.
* Mirrors existing Coroutine_test::correct_order.
*/
void
testCorrectOrder()
{
using namespace std::chrono_literals;
using namespace jtx;
testcase("correct order");
Env env(*this, envconfig([](std::unique_ptr<Config> cfg) {
cfg->forceMultiThread = true;
return cfg;
}));
Gate g1, g2;
auto runner = env.app().getJobQueue().postCoroTask(
JtClient, "CoroTaskTest", [g1p = &g1, g2p = &g2](auto runner) -> CoroTask<void> {
g1p->signal();
co_await runner->suspend();
g2p->signal();
co_return;
});
BEAST_EXPECT(runner);
if (!BEAST_EXPECT(g1.waitFor(5s)))
return;
runner->join();
runner->post();
if (!BEAST_EXPECT(g2.waitFor(5s)))
return;
runner->join();
}
/**
* Incorrect order: post() before suspend(). Verifies the
* race-safe path. Mirrors Coroutine_test::incorrect_order.
*/
void
testIncorrectOrder()
{
using namespace std::chrono_literals;
using namespace jtx;
testcase("incorrect order");
Env env(*this, envconfig([](std::unique_ptr<Config> cfg) {
cfg->forceMultiThread = true;
return cfg;
}));
Gate g;
env.app().getJobQueue().postCoroTask(
JtClient, "CoroTaskTest", [gp = &g](auto runner) -> CoroTask<void> {
runner->post();
co_await runner->suspend();
gp->signal();
co_return;
});
BEAST_EXPECT(g.waitFor(5s));
}
/**
* 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()
{
using namespace std::chrono_literals;
using namespace jtx;
testcase("JobQueueAwaiter");
Env env(*this, envconfig([](std::unique_ptr<Config> cfg) {
cfg->forceMultiThread = true;
return cfg;
}));
Gate g;
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->push_back(3);
gp->signal();
co_return;
});
BEAST_EXPECT(runner);
if (!BEAST_EXPECT(g.waitFor(5s)))
return;
runner->join();
BEAST_EXPECT(steps == std::vector<int>({1, 2, 3}));
}
/**
* Per-coroutine LocalValue isolation. Each coroutine sees its own
* copy of thread-local state. Mirrors Coroutine_test::thread_specific_storage.
*/
void
testThreadSpecificStorage()
{
using namespace std::chrono_literals;
using namespace jtx;
testcase("thread specific storage");
Env env(*this);
auto& jq = env.app().getJobQueue();
static constexpr int kN = 4;
std::array<std::shared_ptr<JobQueue::CoroTaskRunner>, kN> a;
LocalValue<int> lv(-1);
BEAST_EXPECT(*lv == -1);
Gate g;
jq.addJob(JtClient, "LocalValTest", [&]() {
this->BEAST_EXPECT(*lv == -1);
*lv = -2;
this->BEAST_EXPECT(*lv == -2);
g.signal();
});
if (!BEAST_EXPECT(g.waitFor(5s)))
return;
BEAST_EXPECT(*lv == -1);
for (int i = 0; i < kN; ++i)
{
jq.postCoroTask(
JtClient,
"CoroTaskTest",
[this, ap = &a, gp = &g, lvp = &lv, id = i](auto runner) -> CoroTask<void> {
(*ap)[id] = runner;
gp->signal();
co_await runner->suspend();
this->BEAST_EXPECT(**lvp == -1);
**lvp = id;
this->BEAST_EXPECT(**lvp == id);
gp->signal();
co_await runner->suspend();
this->BEAST_EXPECT(**lvp == id);
co_return;
});
if (!BEAST_EXPECT(g.waitFor(5s)))
return;
a[i]->join();
}
for (auto const& r : a)
{
r->post();
if (!BEAST_EXPECT(g.waitFor(5s)))
return;
r->join();
}
for (auto const& r : a)
{
r->post();
r->join();
}
jq.addJob(JtClient, "LocalValTest", [&]() {
this->BEAST_EXPECT(*lv == -2);
g.signal();
});
if (!BEAST_EXPECT(g.waitFor(5s)))
return;
BEAST_EXPECT(*lv == -1);
}
/**
* 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()
{
using namespace std::chrono_literals;
using namespace jtx;
testcase("exception propagation");
Env env(*this, envconfig([](std::unique_ptr<Config> cfg) {
cfg->forceMultiThread = true;
return cfg;
}));
Gate g;
std::string what;
auto runner = env.app().getJobQueue().postCoroTask(
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();
co_return;
});
BEAST_EXPECT(runner);
if (!BEAST_EXPECT(g.waitFor(5s)))
return;
runner->join();
BEAST_EXPECT(what == "test exception");
BEAST_EXPECT(!runner->runnable());
}
/**
* Multiple sequential suspend/resume cycles via co_await.
*/
void
testMultipleYields()
{
using namespace std::chrono_literals;
using namespace jtx;
testcase("multiple yields");
Env env(*this, envconfig([](std::unique_ptr<Config> cfg) {
cfg->forceMultiThread = true;
return cfg;
}));
Gate g;
int counter = 0;
auto runner = env.app().getJobQueue().postCoroTask(
JtClient, "CoroTaskTest", [cp = &counter, gp = &g](auto runner) -> CoroTask<void> {
++(*cp);
gp->signal();
co_await runner->suspend();
++(*cp);
gp->signal();
co_await runner->suspend();
++(*cp);
gp->signal();
co_return;
});
BEAST_EXPECT(runner);
if (!BEAST_EXPECT(g.waitFor(5s)))
return;
BEAST_EXPECT(counter == 1);
runner->join();
runner->post();
if (!BEAST_EXPECT(g.waitFor(5s)))
return;
BEAST_EXPECT(counter == 2);
runner->join();
runner->post();
if (!BEAST_EXPECT(g.waitFor(5s)))
return;
BEAST_EXPECT(counter == 3);
runner->join();
BEAST_EXPECT(!runner->runnable());
}
/**
* CoroTask<T> returns a value via co_return. Outer coroutine
* extracts it with co_await.
*/
void
testValueReturn()
{
using namespace std::chrono_literals;
using namespace jtx;
testcase("value return");
Env env(*this, envconfig([](std::unique_ptr<Config> cfg) {
cfg->forceMultiThread = true;
return cfg;
}));
Gate g;
int result = 0;
auto runner = env.app().getJobQueue().postCoroTask(
JtClient, "CoroTaskTest", [rp = &result, gp = &g](auto) -> CoroTask<void> {
auto inner = []() -> CoroTask<int> { co_return 42; };
*rp = co_await inner();
gp->signal();
co_return;
});
BEAST_EXPECT(runner);
if (!BEAST_EXPECT(g.waitFor(5s)))
return;
runner->join();
BEAST_EXPECT(result == 42);
BEAST_EXPECT(!runner->runnable());
}
/**
* CoroTask<T> propagates exceptions from inner coroutines.
* Outer coroutine catches via try/catch around co_await.
*/
void
testValueException()
{
using namespace std::chrono_literals;
using namespace jtx;
testcase("value exception");
Env env(*this, envconfig([](std::unique_ptr<Config> cfg) {
cfg->forceMultiThread = true;
return cfg;
}));
Gate g;
std::string what;
auto runner = env.app().getJobQueue().postCoroTask(
JtClient, "CoroTaskTest", [wp = &what, gp = &g](auto) -> CoroTask<void> {
auto inner = []() -> CoroTask<int> {
throw std::runtime_error("inner error");
co_return 0;
};
try
{
co_await inner();
}
catch (std::runtime_error const& e)
{
*wp = e.what();
}
gp->signal();
co_return;
});
BEAST_EXPECT(runner);
if (!BEAST_EXPECT(g.waitFor(5s)))
return;
runner->join();
BEAST_EXPECT(what == "inner error");
BEAST_EXPECT(!runner->runnable());
}
/**
* CoroTask<T> chaining. Nested value-returning coroutines
* compose via co_await.
*/
void
testValueChaining()
{
using namespace std::chrono_literals;
using namespace jtx;
testcase("value chaining");
Env env(*this, envconfig([](std::unique_ptr<Config> cfg) {
cfg->forceMultiThread = true;
return cfg;
}));
Gate g;
int result = 0;
auto runner = env.app().getJobQueue().postCoroTask(
JtClient, "CoroTaskTest", [rp = &result, gp = &g](auto) -> CoroTask<void> {
auto add = [](int a, int b) -> CoroTask<int> { co_return a + b; };
auto mul = [add](int a, int b) -> CoroTask<int> {
int const sum = co_await add(a, b);
co_return sum * 2;
};
*rp = co_await mul(3, 4);
gp->signal();
co_return;
});
BEAST_EXPECT(runner);
if (!BEAST_EXPECT(g.waitFor(5s)))
return;
runner->join();
BEAST_EXPECT(result == 14); // (3 + 4) * 2
BEAST_EXPECT(!runner->runnable());
}
/**
* postCoroTask returns nullptr when JobQueue is stopping.
*/
void
testShutdownRejection()
{
using namespace std::chrono_literals;
using namespace jtx;
testcase("shutdown rejection");
Env env(*this, envconfig([](std::unique_ptr<Config> cfg) {
cfg->forceMultiThread = true;
return cfg;
}));
// Stop the JobQueue
env.app().getJobQueue().stop();
auto runner = env.app().getJobQueue().postCoroTask(
JtClient, "CoroTaskTest", [](auto) -> CoroTask<void> { co_return; });
BEAST_EXPECT(!runner);
}
void
run() override
{
testVoidCompletion();
testCorrectOrder();
testIncorrectOrder();
testJobQueueAwaiter();
testThreadSpecificStorage();
testExceptionPropagation();
testMultipleYields();
testValueReturn();
testValueException();
testValueChaining();
testShutdownRejection();
}
};
BEAST_DEFINE_TESTSUITE(CoroTask, core, xrpl);
} // namespace xrpl::test
// NOLINTEND(cppcoreguidelines-avoid-capturing-lambda-coroutines)

View File

@@ -6,7 +6,6 @@
#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>
@@ -50,11 +49,6 @@ public:
}
};
// NOTE: All coroutine lambdas passed to postCoroTask use explicit
// pointer-by-value captures instead of [&] to work around a GCC 14
// bug where reference captures in coroutine lambdas are corrupted
// in the coroutine frame.
void
correctOrder()
{
@@ -69,22 +63,14 @@ public:
}));
Gate g1, g2;
std::shared_ptr<JobQueue::CoroTaskRunner> c;
env.app().getJobQueue().postCoroTask(
JtClient,
"CoroTest",
// Safe capture: the test blocks on the Gates until the coroutine
// completes, so the captured pointers outlive the coroutine.
// NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
[cp = &c, g1p = &g1, g2p = &g2](auto runner) -> CoroTask<void> {
*cp = runner;
g1p->signal();
co_await runner->suspend();
g2p->signal();
co_return;
});
if (!BEAST_EXPECT(g1.waitFor(5s)))
return;
std::shared_ptr<JobQueue::Coro> c;
env.app().getJobQueue().postCoro(JtClient, "CoroTest", [&](auto const& cr) {
c = cr;
g1.signal();
c->yield();
g2.signal();
});
BEAST_EXPECT(g1.waitFor(5s));
c->join();
c->post();
BEAST_EXPECT(g2.waitFor(5s));
@@ -104,22 +90,11 @@ public:
}));
Gate g;
env.app().getJobQueue().postCoroTask(
JtClient,
"CoroTest",
// Safe capture: the test blocks on the Gate until the coroutine
// completes, so the captured pointer outlives the coroutine.
// NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
[gp = &g](auto runner) -> CoroTask<void> {
// Schedule a resume before suspending. The posted job
// cannot actually call resume() until the current resume()
// releases CoroTaskRunner::mutex_, which only happens after
// the coroutine suspends at co_await.
runner->post();
co_await runner->suspend();
gp->signal();
co_return;
});
env.app().getJobQueue().postCoro(JtClient, "CoroTest", [&](auto const& c) {
c->post();
c->yield();
g.signal();
});
BEAST_EXPECT(g.waitFor(5s));
}
@@ -135,7 +110,7 @@ public:
auto& jq = env.app().getJobQueue();
static int const kN = 4;
std::array<std::shared_ptr<JobQueue::CoroTaskRunner>, kN> a;
std::array<std::shared_ptr<JobQueue::Coro>, kN> a;
LocalValue<int> lv(-1);
BEAST_EXPECT(*lv == -1);
@@ -152,36 +127,26 @@ public:
for (int i = 0; i < kN; ++i)
{
jq.postCoroTask(
JtClient,
"CoroTest",
// Safe capture: the test drives every coroutine to completion
// via the Gate/post()/join() sequences below, so the captured
// pointers outlive the coroutines.
// NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
[this, ap = &a, gp = &g, lvp = &lv, id = i](auto runner) -> CoroTask<void> {
(*ap)[id] = runner;
gp->signal();
co_await runner->suspend();
jq.postCoro(JtClient, "CoroTest", [&, id = i](auto const& c) {
a[id] = c;
g.signal();
c->yield();
this->BEAST_EXPECT(**lvp == -1);
**lvp = id;
this->BEAST_EXPECT(**lvp == id);
gp->signal();
co_await runner->suspend();
this->BEAST_EXPECT(*lv == -1);
*lv = id;
this->BEAST_EXPECT(*lv == id);
g.signal();
c->yield();
this->BEAST_EXPECT(**lvp == id);
co_return;
});
if (!BEAST_EXPECT(g.waitFor(5s)))
return;
this->BEAST_EXPECT(*lv == id);
});
BEAST_EXPECT(g.waitFor(5s));
a[i]->join();
}
for (auto const& c : a)
{
c->post();
if (!BEAST_EXPECT(g.waitFor(5s)))
return;
BEAST_EXPECT(g.waitFor(5s));
c->join();
}
for (auto const& c : a)

View File

@@ -1,11 +1,11 @@
#include <test/jtx/Env.h>
#include <xrpl/beast/unit_test/suite.h>
#include <xrpl/core/CoroTask.h>
#include <xrpl/core/Job.h>
#include <xrpl/core/JobQueue.h>
#include <atomic>
#include <memory>
namespace xrpl::test {
@@ -46,114 +46,87 @@ class JobQueue_test : public beast::unit_test::Suite
}
}
// NOTE: All coroutine lambdas passed to postCoroTask use explicit
// pointer-by-value captures instead of [&] to work around a GCC 14
// bug where reference captures in coroutine lambdas are corrupted
// in the coroutine frame.
void
testPostCoroTask()
testPostCoro()
{
jtx::Env env{*this};
JobQueue& jQueue = env.app().getJobQueue();
{
// Test repeated post()s until the coroutine completes.
// Test repeated post()s until the Coro completes.
std::atomic<int> yieldCount{0};
auto const runner = jQueue.postCoroTask(
auto const coro = jQueue.postCoro(
JtClient,
"PostCoroTest1",
// Safe capture: the test blocks below until the coroutine
// completes, so the captured pointer outlives the coroutine.
// NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
[ycp = &yieldCount](auto runner) -> CoroTask<void> {
while (++(*ycp) < 4)
co_await runner->suspend();
co_return;
[&yieldCount](std::shared_ptr<JobQueue::Coro> const& coroCopy) {
while (++yieldCount < 4)
coroCopy->yield();
});
BEAST_EXPECT(runner != nullptr);
BEAST_EXPECT(coro != nullptr);
// Wait for the Job to run and yield.
while (yieldCount == 0)
;
// Now re-post until the CoroTaskRunner says it is done.
// Now re-post until the Coro says it is done.
int old = yieldCount;
while (runner->runnable())
while (coro->runnable())
{
BEAST_EXPECT(runner->post());
BEAST_EXPECT(coro->post());
while (old == yieldCount)
{
}
runner->join();
coro->join();
BEAST_EXPECT(++old == yieldCount);
}
BEAST_EXPECT(yieldCount == 4);
}
{
// Test repeated post()+join()s until the coroutine completes.
// Test repeated resume()s until the Coro completes.
int yieldCount{0};
auto const runner = jQueue.postCoroTask(
auto const coro = jQueue.postCoro(
JtClient,
"PostCoroTest2",
// Safe capture: the test blocks below until the coroutine
// completes, so the captured pointer outlives the coroutine.
// NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
[ycp = &yieldCount](auto runner) -> CoroTask<void> {
while (++(*ycp) < 4)
co_await runner->suspend();
co_return;
[&yieldCount](std::shared_ptr<JobQueue::Coro> const& coroCopy) {
while (++yieldCount < 4)
coroCopy->yield();
});
if (!runner)
if (!coro)
{
// There's no good reason we should not get a runner, but we
// There's no good reason we should not get a Coro, but we
// can't continue without one.
BEAST_EXPECT(false);
return;
}
// Wait for the Job to run and yield.
runner->join();
coro->join();
// 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.
// Now resume until the Coro says it is done.
int old = yieldCount;
while (runner->runnable())
while (coro->runnable())
{
BEAST_EXPECT(runner->post());
runner->join();
coro->resume(); // Resume runs synchronously on this thread.
BEAST_EXPECT(++old == yieldCount);
}
BEAST_EXPECT(yieldCount == 4);
}
{
// If the JobQueue is stopped, we should no
// longer be able to post a coroutine (and calling postCoroTask()
// should return nullptr).
// longer be able to add a Coro (and calling postCoro() should
// return false).
using namespace std::chrono_literals;
jQueue.stop();
// The coroutine should never run, so having it access this
// The Coro should never run, so having the Coro access this
// unprotected variable on the stack should be completely safe.
// Not recommended for the faint of heart...
bool unprotected = false;
auto const runner = jQueue.postCoroTask(
JtClient,
"PostCoroTest3",
// Safe capture: the JobQueue is stopped, so the coroutine is
// never started and the captured pointer is never dereferenced.
// NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
[up = &unprotected](auto) -> CoroTask<void> {
*up = true;
co_return;
auto const coro = jQueue.postCoro(
JtClient, "PostCoroTest3", [&unprotected](std::shared_ptr<JobQueue::Coro> const&) {
unprotected = false;
});
BEAST_EXPECT(runner == nullptr);
BEAST_EXPECT(unprotected == false);
BEAST_EXPECT(coro == nullptr);
}
}
@@ -162,7 +135,7 @@ public:
run() override
{
testAddJob();
testPostCoroTask();
testPostCoro();
}
};

View File

@@ -23,9 +23,7 @@
#include <xrpl/basics/strHex.h>
#include <xrpl/beast/unit_test/suite.h>
#include <xrpl/beast/utility/instrumentation.h>
#include <xrpl/core/CoroTask.h>
#include <xrpl/core/Job.h>
#include <xrpl/core/JobQueue.h>
#include <xrpl/core/ServiceRegistry.h>
#include <xrpl/json/json_value.h>
#include <xrpl/json/to_string.h>
@@ -225,6 +223,7 @@ findPathsRequest(
.ledgerMaster = app.getLedgerMaster(),
.consumer = c,
.role = Role::USER,
.coro = {},
.infoSub = {},
.apiVersion = rpc::kApiVersionIfUnspecified},
{},
@@ -251,14 +250,11 @@ findPathsRequest(
json::Value result;
Gate g;
// Safe capture: the caller blocks on g.waitFor() until the coroutine
// completes, so the captured locals outlive the coroutine.
// NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
app.getJobQueue().postCoroTask(JtClient, "RPC-Client", [&](auto) -> CoroTask<void> {
app.getJobQueue().postCoro(JtClient, "RPC-Client", [&](auto const& coro) {
context.params = std::move(params);
context.coro = coro;
rpc::doCommand(context, result);
g.signal();
co_return;
});
using namespace std::chrono_literals;

View File

@@ -1483,6 +1483,7 @@ ApplicationImp::setup(boost::program_options::variables_map const& cmdline)
.ledgerMaster = getLedgerMaster(),
.consumer = c,
.role = Role::ADMIN,
.coro = {},
.infoSub = {},
.apiVersion = rpc::kApiMaximumSupportedVersion},
jvCommand};

View File

@@ -17,7 +17,6 @@
#include <xrpl/beast/utility/instrumentation.h>
#include <xrpl/config/BasicConfig.h>
#include <xrpl/config/Constants.h>
#include <xrpl/core/CoroTask.h>
#include <xrpl/core/Job.h>
#include <xrpl/core/JobQueue.h>
#include <xrpl/protocol/ErrorCodes.h>
@@ -148,19 +147,13 @@ GRPCServerImpl::CallData<Request, Response>::process()
// ensures that finished is always true when this CallData object
// is returned as a tag in handleRpcs(), after sending the response
finished_ = true;
auto runner = app_.getJobQueue().postCoroTask(
JobType::JtRpc,
"gRPC-Client",
// Safe capture: postCoroTask heap-allocates the lambda (FuncStore)
// and thisShared keeps the CallData alive until the coroutine ends.
// NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
[thisShared](auto) -> CoroTask<void> {
thisShared->processRequest();
co_return;
auto coro = app_.getJobQueue().postCoro(
JobType::JtRpc, "gRPC-Client", [thisShared](std::shared_ptr<JobQueue::Coro> coro) {
thisShared->process(coro);
});
// If runner is null, then the JobQueue has already been shutdown
if (!runner)
// If coro is null, then the JobQueue has already been shutdown
if (!coro)
{
grpc::Status const status{grpc::StatusCode::INTERNAL, "Job Queue is already stopped"};
responder_.FinishWithError(status, this);
@@ -169,7 +162,7 @@ GRPCServerImpl::CallData<Request, Response>::process()
template <class Request, class Response>
void
GRPCServerImpl::CallData<Request, Response>::processRequest()
GRPCServerImpl::CallData<Request, Response>::process(std::shared_ptr<JobQueue::Coro> coro)
{
try
{
@@ -211,6 +204,7 @@ GRPCServerImpl::CallData<Request, Response>::processRequest()
app_.getLedgerMaster(),
usage,
role,
coro,
InfoSub::pointer(),
kApiVersion},
request_};

View File

@@ -6,6 +6,7 @@
#include <xrpld/rpc/detail/Handler.h>
#include <xrpl/beast/utility/Journal.h>
#include <xrpl/core/JobQueue.h>
#include <xrpl/proto/org/xrpl/rpc/v1/xrp_ledger.grpc.pb.h>
#include <xrpl/resource/Charge.h>
#include <xrpl/resource/Consumer.h>
@@ -227,12 +228,9 @@ private:
clone() override;
private:
/**
* Process the gRPC request. Called inside the CoroTask lambda
* posted to the JobQueue by process().
*/
// process the request. Called inside the coroutine passed to JobQueue
void
processRequest();
process(std::shared_ptr<JobQueue::Coro> coro);
// return load type of this RPC
resource::Charge

5
src/xrpld/rpc/AGENTS.md Normal file
View File

@@ -0,0 +1,5 @@
# AGENTS.md — rpc
See the repo-level [AGENTS.md](../../../AGENTS.md) for general build/test/style guidance. `CLAUDE.md` in this directory is a symlink to this file.
Any change to a public RPC method's behavior (new/changed/removed fields, parameters, or error conditions) needs a corresponding entry in [`API-CHANGELOG.md`](../../../API-CHANGELOG.md), under the `## Unreleased` section (`### Additions`, `### Deprecations`, etc. as appropriate).

1
src/xrpld/rpc/CLAUDE.md Symbolic link
View File

@@ -0,0 +1 @@
AGENTS.md

View File

@@ -3,11 +3,13 @@
#include <xrpld/rpc/Role.h>
#include <xrpl/beast/utility/Journal.h>
#include <xrpl/core/JobQueue.h>
#include <xrpl/json/json_value.h>
#include <xrpl/resource/Charge.h>
#include <xrpl/resource/Consumer.h>
#include <xrpl/server/InfoSub.h>
#include <memory>
#include <string_view>
namespace xrpl {
@@ -30,6 +32,7 @@ struct Context
LedgerMaster& ledgerMaster;
resource::Consumer& consumer;
Role role;
std::shared_ptr<JobQueue::Coro> coro;
InfoSub::pointer infoSub;
unsigned int apiVersion;
};

View File

@@ -184,10 +184,13 @@ public:
private:
json::Value
processSession(std::shared_ptr<WSSession> const& session, json::Value const& jv);
processSession(
std::shared_ptr<WSSession> const& session,
std::shared_ptr<JobQueue::Coro> const& coro,
json::Value const& jv);
void
processSession(std::shared_ptr<Session> const&);
processSession(std::shared_ptr<Session> const&, std::shared_ptr<JobQueue::Coro> coro);
void
processRequest(
@@ -195,6 +198,7 @@ private:
std::string const& request,
beast::ip::Endpoint const& remoteIPAddress,
Output const&,
std::shared_ptr<JobQueue::Coro> coro,
std::string_view forwardedFor,
std::string_view user);

View File

@@ -6,7 +6,6 @@
#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>
@@ -138,12 +137,9 @@ PathRequestManager::updateAll(std::shared_ptr<ReadView const> const& inLedger)
}
else if (request->hasCompletion())
{
// 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(); }};
// One-shot request with completion function
request->doUpdate(cache, false);
request->updateComplete();
++processed;
}
}

View File

@@ -17,7 +17,6 @@
#include <xrpl/beast/rfc2616.h>
#include <xrpl/beast/utility/Journal.h>
#include <xrpl/config/Constants.h>
#include <xrpl/core/CoroTask.h>
#include <xrpl/core/Job.h>
#include <xrpl/core/JobQueue.h>
#include <xrpl/json/Output.h>
@@ -318,23 +317,9 @@ ServerHandler::onRequest(Session& session)
}
std::shared_ptr<Session> const detachedSession = session.detach();
auto const postResult = jobQueue_.postCoroTask(
JtClientRpc,
"RPC-Client",
// Safe capture: postCoroTask heap-allocates the lambda (FuncStore),
// detachedSession is captured by value, and this (the ServerHandler)
// outlives the JobQueue jobs.
// NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
[this, detachedSession](auto) -> CoroTask<void> {
try
{
processSession(detachedSession);
}
catch (std::exception const& e)
{
JLOG(journal_.error()) << "RPC-Client coroutine exception: " << e.what();
}
co_return;
auto const postResult = jobQueue_.postCoro(
JtClientRpc, "RPC-Client", [this, detachedSession](std::shared_ptr<JobQueue::Coro> coro) {
processSession(detachedSession, coro);
});
if (postResult == nullptr)
{
@@ -370,30 +355,17 @@ ServerHandler::onWSMessage(
JLOG(journal_.trace()) << "Websocket received '" << jv << "'";
auto const postResult = jobQueue_.postCoroTask(
auto const postResult = jobQueue_.postCoro(
JtClientWebsocket,
"WS-Client",
// Safe capture: postCoroTask heap-allocates the lambda (FuncStore),
// session and jv are captured by value, and this (the ServerHandler)
// outlives the JobQueue jobs.
// NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
[this, session, jv = std::move(jv)](auto) -> CoroTask<void> {
try
{
auto const jr = this->processSession(session, jv);
auto const s = to_string(jr);
auto const n = s.length();
boost::beast::multi_buffer sb(n);
sb.commit(
boost::asio::buffer_copy(sb.prepare(n), boost::asio::buffer(s.c_str(), n)));
session->send(std::make_shared<StreambufWSMsg<decltype(sb)>>(std::move(sb)));
session->complete();
}
catch (std::exception const& e)
{
JLOG(journal_.error()) << "WS-Client coroutine exception: " << e.what();
}
co_return;
[this, session, jv = std::move(jv)](std::shared_ptr<JobQueue::Coro> const& coro) {
auto const jr = this->processSession(session, coro, jv);
auto const s = to_string(jr);
auto const n = s.length();
boost::beast::multi_buffer sb(n);
sb.commit(boost::asio::buffer_copy(sb.prepare(n), boost::asio::buffer(s.c_str(), n)));
session->send(std::make_shared<StreambufWSMsg<decltype(sb)>>(std::move(sb)));
session->complete();
});
if (postResult == nullptr)
{
@@ -438,7 +410,10 @@ logDuration(json::Value const& request, T const& duration, beast::Journal& journ
}
json::Value
ServerHandler::processSession(std::shared_ptr<WSSession> const& session, json::Value const& jv)
ServerHandler::processSession(
std::shared_ptr<WSSession> const& session,
std::shared_ptr<JobQueue::Coro> const& coro,
json::Value const& jv)
{
auto is = std::static_pointer_cast<WSInfoSub>(session->appDefined);
if (is->getConsumer().disconnect(journal_))
@@ -505,6 +480,7 @@ ServerHandler::processSession(std::shared_ptr<WSSession> const& session, json::V
.ledgerMaster = app_.getLedgerMaster(),
.consumer = is->getConsumer(),
.role = role,
.coro = coro,
.infoSub = is,
.apiVersion = apiVersion},
jv,
@@ -575,14 +551,18 @@ ServerHandler::processSession(std::shared_ptr<WSSession> const& session, json::V
return jr;
}
// Run as a coroutine.
void
ServerHandler::processSession(std::shared_ptr<Session> const& session)
ServerHandler::processSession(
std::shared_ptr<Session> const& session,
std::shared_ptr<JobQueue::Coro> coro)
{
processRequest(
session->port(),
buffersToString(session->request().body().data()),
session->remoteAddress().atPort(0),
makeOutput(*session),
coro,
forwardedFor(session->request()),
[&] -> std::string_view {
auto const iter = session->request().find("X-User");
@@ -623,6 +603,7 @@ ServerHandler::processRequest(
std::string const& request,
beast::ip::Endpoint const& remoteIPAddress,
Output const& output,
std::shared_ptr<JobQueue::Coro> coro,
std::string_view forwardedFor,
std::string_view user)
{
@@ -882,6 +863,7 @@ ServerHandler::processRequest(
.ledgerMaster = app_.getLedgerMaster(),
.consumer = usage,
.role = role,
.coro = coro,
.infoSub = InfoSub::pointer(),
.apiVersion = apiVersion},
params,

View File

@@ -65,12 +65,6 @@ 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;

View File

@@ -7,15 +7,14 @@
#include <xrpld/rpc/detail/RPCLedgerHelpers.h>
#include <xrpld/rpc/detail/Tuning.h>
#include <xrpl/core/JobQueue.h>
#include <xrpl/json/json_value.h>
#include <xrpl/protocol/ErrorCodes.h>
#include <xrpl/protocol/RPCErr.h>
#include <xrpl/protocol/jss.h>
#include <xrpl/resource/Fees.h>
#include <condition_variable>
#include <memory>
#include <mutex>
#include <utility>
namespace xrpl {
@@ -48,62 +47,98 @@ 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.
// If makeLegacyPathRequest cannot schedule the job (e.g. during
// shutdown), it returns an empty request and we skip the wait.
// Replaces the old Coro yield/resume pattern with synchronous
// blocking, eliminating shutdown race conditions.
// It doesn't look like there's much odd happening here, but you should
// be aware this code runs in a JobQueue::Coro, which is a coroutine.
// And we may be flipping around between threads. Here's an overview:
//
// The state is shared with the continuation so that it stays alive
// even if we stop waiting before the continuation runs.
struct PathDone
{
std::mutex mtx;
std::condition_variable cv;
bool done = false;
};
auto const state = std::make_shared<PathDone>();
// 1. We're running doRipplePathFind() due to a call to
// ripple_path_find. doRipplePathFind() is currently running
// inside of a JobQueue::Coro using a JobQueue thread.
//
// 2. doRipplePathFind's call to makeLegacyPathRequest() enqueues the
// path-finding request. That request will (probably) run at some
// indeterminate future time on a (probably different) JobQueue
// thread.
//
// 3. As a continuation from that path-finding JobQueue thread, the
// coroutine we're currently running in (!) is posted to the
// JobQueue. Because it is a continuation, that post won't
// happen until the path-finding request completes.
//
// 4. Once the continuation is enqueued, and we have reason to think
// the path-finding job is likely to run, then the coroutine we're
// running in yield()s. That means it surrenders its thread in
// the JobQueue. The coroutine is suspended, but ready to run,
// because it is kept resident by a shared_ptr in the
// path-finding continuation.
//
// 5. If all goes well then path-finding runs on a JobQueue thread
// and executes its continuation. The continuation posts this
// same coroutine (!) to the JobQueue.
//
// 6. When the JobQueue calls this coroutine, this coroutine resumes
// from the line below the coro->yield() and returns the
// path-finding result.
//
// With so many moving parts, what could go wrong?
//
// Just in terms of the JobQueue refusing to add jobs at shutdown
// there are two specific things that can go wrong.
//
// 1. The path-finding Job queued by makeLegacyPathRequest() might be
// rejected (because we're shutting down).
//
// Fortunately this problem can be addressed by looking at the
// return value of makeLegacyPathRequest(). If
// makeLegacyPathRequest() cannot get a thread to run the path-find
// on, then it returns an empty request.
//
// 2. The path-finding job might run, but the Coro::post() might be
// rejected by the JobQueue (because we're shutting down).
//
// We handle this case by resuming (not posting) the Coro.
// By resuming the Coro, we allow the Coro to run to completion
// on the current thread instead of requiring that it run on a
// new thread from the JobQueue.
//
// Both of these failure modes are hard to recreate in a unit test
// because they are so dependent on inter-thread timing. However
// the failure modes can be observed by synchronously (inside the
// xrpld source code) shutting down the application. The code to
// do so looks like this:
//
// context.app.signalStop();
// while (! context.app.getJobQueue().jobCounter().joined()) { }
//
// The first line starts the process of shutting down the app.
// The second line waits until no more jobs can be added to the
// JobQueue before letting the thread continue.
//
// May 2017
jvResult = context.app.getPathRequestManager().makeLegacyPathRequest(
request,
[state]() {
[&context]() {
// Copying the shared_ptr keeps the coroutine alive up
// through the return. Otherwise the storage under the
// captured reference could evaporate when we return from
// coroCopy->resume(). This is not strictly necessary, but
// will make maintenance easier.
std::shared_ptr<JobQueue::Coro> const coroCopy{context.coro};
if (!coroCopy->post())
{
std::scoped_lock const lk(state->mtx);
state->done = true;
// The post() failed, so we won't get a thread to let
// the Coro finish. We'll call Coro::resume() so the
// Coro can finish on our thread. Otherwise the
// application will hang on shutdown.
coroCopy->resume();
}
state->cv.notify_one();
},
context.consumer,
lpLedger,
context.params);
if (request)
{
std::unique_lock lk(state->mtx);
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
// rather than blocking the RPC thread indefinitely.
return rpcError(RpcInternal);
}
lk.unlock();
context.coro->yield();
jvResult = request->doStatus(context.params);
}