Compare commits

..

10 Commits

Author SHA1 Message Date
Mayukha Vadari
7ec35ce1a1 Merge remote-tracking branch 'upstream/develop' into copilot/convert-boost-to-std-string-view 2026-02-26 14:50:36 -05:00
Mayukha Vadari
daa032d064 Merge commit '2c1fad1023' into copilot/convert-boost-to-std-string-view
# Conflicts:
#	include/xrpl/server/detail/BaseWSPeer.h
2026-02-26 14:50:01 -05:00
Alex Kremer
2e595b6031 chore: Enable clang-tidy checks without issues (#6414)
This change enables all clang-tidy checks that are already passing. It also modifies the clang-tidy CI job, so it runs against all files if .clang-tidy changed.
2026-02-26 18:26:58 +00:00
Bart
3a8a18c2ca refactor: Use uint256 directly as key instead of void pointer (#6313)
This change replaces `void const*` by `uint256 const&` for database fetches.

Object hashes are expressed using the `uint256` data type, and are converted to `void *` when calling the `fetch` or `fetchBatch` functions. However, in these fetch functions they are converted back to `uint256`, making the conversion process unnecessary. In a few cases the underlying pointer is needed, but that can then be easy obtained via `[hash variable].data()`.
2026-02-25 18:23:34 -05:00
Mayukha Vadari
a2cee3b6ab Merge branch 'develop' into copilot/convert-boost-to-std-string-view 2026-02-06 11:43:57 -05:00
copilot-swe-agent[bot]
ae624ab894 Remove unnecessary boost/beast/core/string.hpp includes
Co-authored-by: mvadari <8029314+mvadari@users.noreply.github.com>
2026-01-30 17:27:50 +00:00
copilot-swe-agent[bot]
21ac390a37 Fix lambda return type inconsistency in ServerHandler
Co-authored-by: mvadari <8029314+mvadari@users.noreply.github.com>
2026-01-30 17:02:18 +00:00
copilot-swe-agent[bot]
86aca89bc4 Fix std::string_view constructor usage in BaseWSPeer
Co-authored-by: mvadari <8029314+mvadari@users.noreply.github.com>
2026-01-30 16:28:46 +00:00
copilot-swe-agent[bot]
b26dd49fb6 Replace all boost::beast::string_view with std::string_view
Co-authored-by: mvadari <8029314+mvadari@users.noreply.github.com>
2026-01-30 16:26:37 +00:00
copilot-swe-agent[bot]
0969281538 Initial plan 2026-01-30 16:23:54 +00:00
35 changed files with 388 additions and 2800 deletions

View File

@@ -1,105 +1,143 @@
---
Checks: "-*,
bugprone-argument-comment
bugprone-argument-comment,
bugprone-assert-side-effect,
bugprone-bad-signal-to-kill-thread,
bugprone-bool-pointer-implicit-conversion,
bugprone-casting-through-void,
bugprone-chained-comparison,
bugprone-compare-pointer-to-member-virtual-function,
bugprone-copy-constructor-init,
bugprone-dangling-handle,
bugprone-dynamic-static-initializers,
bugprone-fold-init-type,
bugprone-forward-declaration-namespace,
bugprone-inaccurate-erase,
bugprone-incorrect-enable-if,
bugprone-incorrect-roundings,
bugprone-infinite-loop,
bugprone-integer-division,
bugprone-lambda-function-name,
bugprone-macro-parentheses,
bugprone-macro-repeated-side-effects,
bugprone-misplaced-operator-in-strlen-in-alloc,
bugprone-misplaced-pointer-arithmetic-in-alloc,
bugprone-misplaced-widening-cast,
bugprone-multi-level-implicit-pointer-conversion,
bugprone-multiple-new-in-one-expression,
bugprone-multiple-statement-macro,
bugprone-no-escape,
bugprone-non-zero-enum-to-bool-conversion,
bugprone-parent-virtual-call,
bugprone-posix-return,
bugprone-redundant-branch-condition,
bugprone-shared-ptr-array-mismatch,
bugprone-signal-handler,
bugprone-signed-char-misuse,
bugprone-sizeof-container,
bugprone-spuriously-wake-up-functions,
bugprone-standalone-empty,
bugprone-string-constructor,
bugprone-string-integer-assignment,
bugprone-string-literal-with-embedded-nul,
bugprone-stringview-nullptr,
bugprone-suspicious-enum-usage,
bugprone-suspicious-include,
bugprone-suspicious-memory-comparison,
bugprone-suspicious-memset-usage,
bugprone-suspicious-realloc-usage,
bugprone-suspicious-semicolon,
bugprone-suspicious-string-compare,
bugprone-swapped-arguments,
bugprone-terminating-continue,
bugprone-throw-keyword-missing,
bugprone-undefined-memory-manipulation,
bugprone-undelegated-constructor,
bugprone-unhandled-exception-at-new,
bugprone-unique-ptr-array-mismatch,
bugprone-unsafe-functions,
bugprone-virtual-near-miss,
cppcoreguidelines-no-suspend-with-lock,
cppcoreguidelines-virtual-class-destructor,
hicpp-ignored-remove-result,
misc-definitions-in-headers,
misc-header-include-cycle,
misc-misplaced-const,
misc-static-assert,
misc-throw-by-value-catch-by-reference,
misc-unused-alias-decls,
misc-unused-using-decls,
readability-duplicate-include,
readability-enum-initial-value,
readability-misleading-indentation,
readability-non-const-parameter,
readability-redundant-declaration,
readability-reference-to-constructed-temporary,
modernize-deprecated-headers,
modernize-make-shared,
modernize-make-unique,
performance-implicit-conversion-in-loop,
performance-move-constructor-init,
performance-trivially-destructible
"
# bugprone-assert-side-effect,
# bugprone-bad-signal-to-kill-thread,
# bugprone-bool-pointer-implicit-conversion,
# bugprone-casting-through-void,
# bugprone-chained-comparison,
# bugprone-compare-pointer-to-member-virtual-function,
# bugprone-copy-constructor-init,
# bugprone-crtp-constructor-accessibility,
# bugprone-dangling-handle,
# bugprone-dynamic-static-initializers,
# ---
# checks that have some issues that need to be resolved:
#
# bugprone-empty-catch,
# bugprone-fold-init-type,
# bugprone-forward-declaration-namespace,
# bugprone-inaccurate-erase,
# bugprone-crtp-constructor-accessibility,
# bugprone-inc-dec-in-conditions,
# bugprone-incorrect-enable-if,
# bugprone-incorrect-roundings,
# bugprone-infinite-loop,
# bugprone-integer-division,
# bugprone-lambda-function-name,
# bugprone-macro-parentheses,
# bugprone-macro-repeated-side-effects,
# bugprone-misplaced-operator-in-strlen-in-alloc,
# bugprone-misplaced-pointer-arithmetic-in-alloc,
# bugprone-misplaced-widening-cast,
# bugprone-move-forwarding-reference,
# bugprone-multi-level-implicit-pointer-conversion,
# bugprone-multiple-new-in-one-expression,
# bugprone-multiple-statement-macro,
# bugprone-no-escape,
# bugprone-non-zero-enum-to-bool-conversion,
# bugprone-optional-value-conversion,
# bugprone-parent-virtual-call,
# bugprone-pointer-arithmetic-on-polymorphic-object,
# bugprone-posix-return,
# bugprone-redundant-branch-condition,
# bugprone-reserved-identifier,
# bugprone-return-const-ref-from-parameter,
# bugprone-shared-ptr-array-mismatch,
# bugprone-signal-handler,
# bugprone-signed-char-misuse,
# bugprone-sizeof-container,
# bugprone-sizeof-expression,
# bugprone-spuriously-wake-up-functions,
# bugprone-standalone-empty,
# bugprone-string-constructor,
# bugprone-string-integer-assignment,
# bugprone-string-literal-with-embedded-nul,
# bugprone-stringview-nullptr,
# bugprone-suspicious-enum-usage,
# bugprone-suspicious-include,
# bugprone-suspicious-memory-comparison,
# bugprone-suspicious-memset-usage,
# bugprone-suspicious-missing-comma,
# bugprone-suspicious-realloc-usage,
# bugprone-suspicious-semicolon,
# bugprone-suspicious-string-compare,
# bugprone-suspicious-stringview-data-usage,
# bugprone-swapped-arguments,
# bugprone-switch-missing-default-case,
# bugprone-terminating-continue,
# bugprone-throw-keyword-missing,
# bugprone-too-small-loop-variable,
# bugprone-undefined-memory-manipulation,
# bugprone-undelegated-constructor,
# bugprone-unhandled-exception-at-new,
# bugprone-unhandled-self-assignment,
# bugprone-unique-ptr-array-mismatch,
# bugprone-unsafe-functions,
# bugprone-move-forwarding-reference,
# bugprone-unused-local-non-trivial-variable,
# bugprone-unused-raii,
# bugprone-return-const-ref-from-parameter,
# bugprone-switch-missing-default-case,
# bugprone-sizeof-expression,
# bugprone-suspicious-stringview-data-usage,
# bugprone-suspicious-missing-comma,
# bugprone-pointer-arithmetic-on-polymorphic-object,
# bugprone-optional-value-conversion,
# bugprone-too-small-loop-variable,
# bugprone-unused-return-value,
# bugprone-use-after-move,
# bugprone-virtual-near-miss,
# cppcoreguidelines-init-variables,
# bugprone-unhandled-self-assignment,
# bugprone-unused-raii,
#
# cppcoreguidelines-misleading-capture-default-by-value,
# cppcoreguidelines-no-suspend-with-lock,
# cppcoreguidelines-init-variables,
# cppcoreguidelines-pro-type-member-init,
# cppcoreguidelines-pro-type-static-cast-downcast,
# cppcoreguidelines-rvalue-reference-param-not-moved,
# cppcoreguidelines-use-default-member-init,
# cppcoreguidelines-virtual-class-destructor,
# hicpp-ignored-remove-result,
# cppcoreguidelines-rvalue-reference-param-not-moved,
#
# llvm-namespace-comment,
# misc-const-correctness,
# misc-definitions-in-headers,
# misc-header-include-cycle,
# misc-include-cleaner,
# misc-misplaced-const,
# misc-redundant-expression,
# misc-static-assert,
# misc-throw-by-value-catch-by-reference,
# misc-unused-alias-decls,
# misc-unused-using-decls,
#
# readability-avoid-nested-conditional-operator,
# readability-avoid-return-with-void-value,
# readability-braces-around-statements,
# readability-container-contains,
# readability-container-size-empty,
# readability-convert-member-functions-to-static,
# readability-const-return-type,
# readability-else-after-return,
# readability-implicit-bool-conversion,
# readability-inconsistent-declaration-parameter-name,
# readability-identifier-naming,
# readability-make-member-function-const,
# readability-math-missing-parentheses,
# readability-redundant-inline-specifier,
# readability-redundant-member-init,
# readability-redundant-casting,
# readability-redundant-string-init,
# readability-simplify-boolean-expr,
# readability-static-definition-in-anonymous-namespace,
# readability-suspicious-call-argument,
# readability-use-std-min-max,
# readability-static-accessed-through-instance,
#
# modernize-concat-nested-namespaces,
# modernize-deprecated-headers,
# modernize-make-shared,
# modernize-make-unique,
# modernize-pass-by-value,
# modernize-type-traits,
# modernize-use-designated-initializers,
@@ -111,79 +149,50 @@ Checks: "-*,
# modernize-use-starts-ends-with,
# modernize-use-std-numbers,
# modernize-use-using,
#
# performance-faster-string-find,
# performance-for-range-copy,
# performance-implicit-conversion-in-loop,
# performance-inefficient-vector-operation,
# performance-move-const-arg,
# performance-move-constructor-init,
# performance-no-automatic-move,
# performance-trivially-destructible,
# readability-avoid-nested-conditional-operator,
# readability-avoid-return-with-void-value,
# readability-braces-around-statements,
# readability-const-return-type,
# readability-container-contains,
# readability-container-size-empty,
# readability-convert-member-functions-to-static,
# readability-duplicate-include,
# readability-else-after-return,
# readability-enum-initial-value,
# readability-implicit-bool-conversion,
# readability-inconsistent-declaration-parameter-name,
# readability-identifier-naming,
# readability-make-member-function-const,
# readability-math-missing-parentheses,
# readability-misleading-indentation,
# readability-non-const-parameter,
# readability-redundant-casting,
# readability-redundant-declaration,
# readability-redundant-inline-specifier,
# readability-redundant-member-init,
# readability-redundant-string-init,
# readability-reference-to-constructed-temporary,
# readability-simplify-boolean-expr,
# readability-static-accessed-through-instance,
# readability-static-definition-in-anonymous-namespace,
# readability-suspicious-call-argument,
# readability-use-std-min-max
# ---
#
# CheckOptions:
# readability-braces-around-statements.ShortStatementLines: 2
# readability-identifier-naming.MacroDefinitionCase: UPPER_CASE
# readability-identifier-naming.ClassCase: CamelCase
# readability-identifier-naming.StructCase: CamelCase
# readability-identifier-naming.UnionCase: CamelCase
# readability-identifier-naming.EnumCase: CamelCase
# readability-identifier-naming.EnumConstantCase: CamelCase
# readability-identifier-naming.ScopedEnumConstantCase: CamelCase
# readability-identifier-naming.GlobalConstantCase: UPPER_CASE
# readability-identifier-naming.GlobalConstantPrefix: "k"
# readability-identifier-naming.GlobalVariableCase: CamelCase
# readability-identifier-naming.GlobalVariablePrefix: "g"
# readability-identifier-naming.ConstexprFunctionCase: camelBack
# readability-identifier-naming.ConstexprMethodCase: camelBack
# readability-identifier-naming.ClassMethodCase: camelBack
# readability-identifier-naming.ClassMemberCase: camelBack
# readability-identifier-naming.ClassConstantCase: UPPER_CASE
# readability-identifier-naming.ClassConstantPrefix: "k"
# readability-identifier-naming.StaticConstantCase: UPPER_CASE
# readability-identifier-naming.StaticConstantPrefix: "k"
# readability-identifier-naming.StaticVariableCase: UPPER_CASE
# readability-identifier-naming.StaticVariablePrefix: "k"
# readability-identifier-naming.ConstexprVariableCase: UPPER_CASE
# readability-identifier-naming.ConstexprVariablePrefix: "k"
# readability-identifier-naming.LocalConstantCase: camelBack
# readability-identifier-naming.LocalVariableCase: camelBack
# readability-identifier-naming.TemplateParameterCase: CamelCase
# readability-identifier-naming.ParameterCase: camelBack
# readability-identifier-naming.FunctionCase: camelBack
# readability-identifier-naming.MemberCase: camelBack
# readability-identifier-naming.PrivateMemberSuffix: _
# readability-identifier-naming.ProtectedMemberSuffix: _
# readability-identifier-naming.PublicMemberSuffix: ""
# readability-identifier-naming.FunctionIgnoredRegexp: ".*tag_invoke.*"
# bugprone-unsafe-functions.ReportMoreUnsafeFunctions: true
CheckOptions:
# readability-braces-around-statements.ShortStatementLines: 2
# readability-identifier-naming.MacroDefinitionCase: UPPER_CASE
# readability-identifier-naming.ClassCase: CamelCase
# readability-identifier-naming.StructCase: CamelCase
# readability-identifier-naming.UnionCase: CamelCase
# readability-identifier-naming.EnumCase: CamelCase
# readability-identifier-naming.EnumConstantCase: CamelCase
# readability-identifier-naming.ScopedEnumConstantCase: CamelCase
# readability-identifier-naming.GlobalConstantCase: UPPER_CASE
# readability-identifier-naming.GlobalConstantPrefix: "k"
# readability-identifier-naming.GlobalVariableCase: CamelCase
# readability-identifier-naming.GlobalVariablePrefix: "g"
# readability-identifier-naming.ConstexprFunctionCase: camelBack
# readability-identifier-naming.ConstexprMethodCase: camelBack
# readability-identifier-naming.ClassMethodCase: camelBack
# readability-identifier-naming.ClassMemberCase: camelBack
# readability-identifier-naming.ClassConstantCase: UPPER_CASE
# readability-identifier-naming.ClassConstantPrefix: "k"
# readability-identifier-naming.StaticConstantCase: UPPER_CASE
# readability-identifier-naming.StaticConstantPrefix: "k"
# readability-identifier-naming.StaticVariableCase: UPPER_CASE
# readability-identifier-naming.StaticVariablePrefix: "k"
# readability-identifier-naming.ConstexprVariableCase: UPPER_CASE
# readability-identifier-naming.ConstexprVariablePrefix: "k"
# readability-identifier-naming.LocalConstantCase: camelBack
# readability-identifier-naming.LocalVariableCase: camelBack
# readability-identifier-naming.TemplateParameterCase: CamelCase
# readability-identifier-naming.ParameterCase: camelBack
# readability-identifier-naming.FunctionCase: camelBack
# readability-identifier-naming.MemberCase: camelBack
# readability-identifier-naming.PrivateMemberSuffix: _
# readability-identifier-naming.ProtectedMemberSuffix: _
# readability-identifier-naming.PublicMemberSuffix: ""
# readability-identifier-naming.FunctionIgnoredRegexp: ".*tag_invoke.*"
bugprone-unsafe-functions.ReportMoreUnsafeFunctions: true
# bugprone-unused-return-value.CheckedReturnTypes: ::std::error_code;::std::error_condition;::std::errc
# misc-include-cleaner.IgnoreHeaders: '.*/(detail|impl)/.*;.*(expected|unexpected).*;.*ranges_lower_bound\.h;time.h;stdlib.h;__chrono/.*;fmt/chrono.h;boost/uuid/uuid_hash.hpp'
#

View File

@@ -78,9 +78,9 @@ jobs:
id: run_clang_tidy
continue-on-error: true
env:
FILES: ${{ inputs.files }}
TARGETS: ${{ inputs.files != '' && inputs.files || 'src tests' }}
run: |
run-clang-tidy -j ${{ steps.nproc.outputs.nproc }} -p "$BUILD_DIR" $FILES 2>&1 | tee clang-tidy-output.txt
run-clang-tidy -j ${{ steps.nproc.outputs.nproc }} -p "${BUILD_DIR}" ${TARGETS} 2>&1 | tee clang-tidy-output.txt
- name: Upload clang-tidy output
if: steps.run_clang_tidy.outcome != 'success'

View File

@@ -22,7 +22,8 @@ jobs:
if: ${{ inputs.check_only_changed }}
runs-on: ubuntu-latest
outputs:
any_changed: ${{ steps.changed_files.outputs.any_changed }}
clang_tidy_config_changed: ${{ steps.changed_clang_tidy.outputs.any_changed }}
any_cpp_changed: ${{ steps.changed_files.outputs.any_changed }}
all_changed_files: ${{ steps.changed_files.outputs.all_changed_files }}
steps:
- name: Checkout repository
@@ -38,10 +39,17 @@ jobs:
**/*.ipp
separator: " "
- name: Get changed clang-tidy configuration
id: changed_clang_tidy
uses: tj-actions/changed-files@7dee1b0c1557f278e5c7dc244927139d78c0e22a # v47.0.4
with:
files: |
.clang-tidy
run-clang-tidy:
needs: [determine-files]
if: ${{ always() && !cancelled() && (!inputs.check_only_changed || needs.determine-files.outputs.any_changed == 'true') }}
if: ${{ always() && !cancelled() && (!inputs.check_only_changed || needs.determine-files.outputs.any_cpp_changed == 'true' || needs.determine-files.outputs.clang_tidy_config_changed == 'true') }}
uses: ./.github/workflows/reusable-clang-tidy-files.yml
with:
files: ${{ inputs.check_only_changed && needs.determine-files.outputs.all_changed_files || '' }}
files: ${{ (needs.determine-files.outputs.clang_tidy_config_changed == 'true' && '') || (inputs.check_only_changed && needs.determine-files.outputs.all_changed_files || '') }}
create_issue_on_failure: ${{ inputs.create_issue_on_failure }}

File diff suppressed because it is too large Load Diff

View File

@@ -251,6 +251,29 @@ pip3 install pre-commit
pre-commit install
```
## Clang-tidy
All code must pass `clang-tidy` checks according to the settings in [`.clang-tidy`](./.clang-tidy).
There is a Continuous Integration job that runs clang-tidy on pull requests. The CI will check:
- All changed C++ files (`.cpp`, `.h`, `.ipp`) when only code files are modified
- **All files in the repository** when the `.clang-tidy` configuration file is changed
This ensures that configuration changes don't introduce new warnings across the codebase.
### Running clang-tidy locally
Before running clang-tidy, you must build the project to generate required files (particularly protobuf headers). Refer to [`BUILD.md`](./BUILD.md) for build instructions.
Then run clang-tidy on your local changes:
```
run-clang-tidy -p build src tests
```
This will check all source files in the `src` and `tests` directories using the compile commands from your `build` directory.
## Contracts and instrumentation
We are using [Antithesis](https://antithesis.com/) for continuous fuzzing,

View File

@@ -71,14 +71,12 @@ words:
- coldwallet
- compr
- conanfile
- cppcoro
- conanrun
- confs
- connectability
- coro
- coros
- cowid
- cppcoro
- cryptocondition
- cryptoconditional
- cryptoconditions
@@ -101,14 +99,11 @@ words:
- endmacro
- exceptioned
- Falco
- fcontext
- finalizers
- firewalled
- fcontext
- fmtdur
- fsanitize
- funclets
- gantt
- gcov
- gcovr
- ghead
@@ -190,7 +185,6 @@ words:
- ostr
- pargs
- partitioner
- pratik
- paychan
- paychans
- permdex
@@ -198,7 +192,6 @@ words:
- permissioned
- pointee
- populator
- pratik
- preauth
- preauthorization
- preauthorize
@@ -213,7 +206,6 @@ words:
- queuable
- Raphson
- replayer
- repost
- rerere
- retriable
- RIPD
@@ -244,7 +236,6 @@ words:
- soci
- socidb
- sslws
- stackful
- statsd
- STATSDCOLLECTOR
- stissue

View File

@@ -10,6 +10,7 @@
#include <cctype>
#include <iterator>
#include <string>
#include <string_view>
#include <vector>
namespace beast {
@@ -181,7 +182,7 @@ split_commas(FwdIt first, FwdIt last)
template <class Result = std::vector<std::string>>
Result
split_commas(boost::beast::string_view const& s)
split_commas(std::string_view const& s)
{
return split_commas(s.begin(), s.end());
}

View File

@@ -1,289 +0,0 @@
#pragma once
#include <coroutine>
#include <exception>
#include <utility>
#include <variant>
namespace xrpl {
template <typename T = void>
class CoroTask;
// --------------------------------------------------------------------------
// CoroTask<void> — coroutine return type for void-returning coroutines
// --------------------------------------------------------------------------
template <>
class CoroTask<void>
{
public:
struct promise_type;
using Handle = std::coroutine_handle<promise_type>;
struct promise_type
{
std::exception_ptr exception_;
std::coroutine_handle<> continuation_;
CoroTask
get_return_object()
{
return CoroTask{Handle::from_promise(*this)};
}
std::suspend_always
initial_suspend() noexcept
{
return {};
}
struct FinalAwaiter
{
bool
await_ready() noexcept
{
return false;
}
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 {};
}
void
return_void()
{
}
void
unhandled_exception()
{
exception_ = std::current_exception();
}
};
CoroTask() = default;
explicit CoroTask(Handle h) : handle_(h)
{
}
~CoroTask()
{
if (handle_)
handle_.destroy();
}
CoroTask(CoroTask&& other) noexcept : handle_(std::exchange(other.handle_, {}))
{
}
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;
Handle
handle() const
{
return handle_;
}
bool
done() const
{
return handle_ && handle_.done();
}
// Awaiter interface — allows co_await on a CoroTask
bool
await_ready() const noexcept
{
return false;
}
std::coroutine_handle<>
await_suspend(std::coroutine_handle<> caller) noexcept
{
handle_.promise().continuation_ = caller;
return handle_; // Symmetric transfer
}
void
await_resume()
{
if (auto& ep = handle_.promise().exception_)
std::rethrow_exception(ep);
}
private:
Handle handle_;
};
// --------------------------------------------------------------------------
// CoroTask<T> — coroutine return type for value-returning coroutines
// --------------------------------------------------------------------------
template <typename T>
class CoroTask
{
public:
struct promise_type;
using Handle = std::coroutine_handle<promise_type>;
struct promise_type
{
std::variant<std::monostate, T, std::exception_ptr> result_;
std::coroutine_handle<> continuation_;
CoroTask
get_return_object()
{
return CoroTask{Handle::from_promise(*this)};
}
std::suspend_always
initial_suspend() noexcept
{
return {};
}
struct FinalAwaiter
{
bool
await_ready() noexcept
{
return false;
}
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 {};
}
void
return_value(T value)
{
result_.template emplace<1>(std::move(value));
}
void
unhandled_exception()
{
result_.template emplace<2>(std::current_exception());
}
};
CoroTask() = default;
explicit CoroTask(Handle h) : handle_(h)
{
}
~CoroTask()
{
if (handle_)
handle_.destroy();
}
CoroTask(CoroTask&& other) noexcept : handle_(std::exchange(other.handle_, {}))
{
}
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;
Handle
handle() const
{
return handle_;
}
bool
done() const
{
return handle_ && handle_.done();
}
bool
await_ready() const noexcept
{
return false;
}
std::coroutine_handle<>
await_suspend(std::coroutine_handle<> caller) noexcept
{
handle_.promise().continuation_ = caller;
return handle_;
}
T
await_resume()
{
auto& result = handle_.promise().result_;
if (auto* ep = std::get_if<2>(&result))
std::rethrow_exception(*ep);
return std::get<1>(std::move(result));
}
private:
Handle handle_;
};
} // namespace xrpl

View File

@@ -1,158 +0,0 @@
#pragma once
namespace xrpl {
inline JobQueue::CoroTaskRunner::CoroTaskRunner(
create_t,
JobQueue& jq,
JobType type,
std::string const& name)
: jq_(jq), type_(type), name_(name), running_(false)
{
}
template <class F>
void
JobQueue::CoroTaskRunner::init(F&& f)
{
// Store the callable on the heap 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.
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);
}
inline JobQueue::CoroTaskRunner::~CoroTaskRunner()
{
#ifndef NDEBUG
XRPL_ASSERT(finished_, "xrpl::JobQueue::CoroTaskRunner::~CoroTaskRunner : is finished");
#endif
}
inline void
JobQueue::CoroTaskRunner::onSuspend()
{
std::lock_guard lock(jq_.m_mutex);
++jq_.nSuspend_;
}
inline void
JobQueue::CoroTaskRunner::onUndoSuspend()
{
std::lock_guard lock(jq_.m_mutex);
--jq_.nSuspend_;
}
inline auto
JobQueue::CoroTaskRunner::suspend()
{
struct SuspendAwaiter
{
CoroTaskRunner& runner_;
bool
await_ready() const noexcept
{
return false;
}
void
await_suspend(std::coroutine_handle<>) const
{
runner_.onSuspend();
}
void
await_resume() const noexcept
{
}
};
return SuspendAwaiter{*this};
}
inline bool
JobQueue::CoroTaskRunner::post()
{
{
std::lock_guard lk(mutex_run_);
running_ = true;
}
// 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. Clean up running_.
std::lock_guard lk(mutex_run_);
running_ = false;
cv_.notify_all();
return false;
}
inline void
JobQueue::CoroTaskRunner::resume()
{
{
std::lock_guard lk(mutex_run_);
running_ = true;
}
{
std::lock_guard lock(jq_.m_mutex);
--jq_.nSuspend_;
}
auto saved = detail::getLocalValues().release();
detail::getLocalValues().reset(&lvs_);
std::lock_guard lock(mutex_);
XRPL_ASSERT(!task_.done(), "xrpl::JobQueue::CoroTaskRunner::resume : task is not done");
task_.handle().resume();
detail::getLocalValues().release();
detail::getLocalValues().reset(saved);
#ifndef NDEBUG
if (task_.done())
finished_ = true;
#endif
std::lock_guard lk(mutex_run_);
running_ = false;
cv_.notify_all();
}
inline bool
JobQueue::CoroTaskRunner::runnable() const
{
return !task_.done();
}
inline void
JobQueue::CoroTaskRunner::expectEarlyExit()
{
#ifndef NDEBUG
if (!finished_)
#endif
{
std::lock_guard lock(jq_.m_mutex);
--jq_.nSuspend_;
#ifndef NDEBUG
finished_ = true;
#endif
}
// Destroy the coroutine frame to break a potential shared_ptr cycle.
// The coroutine is at initial_suspend and never ran user code, so
// destroying it is safe. Without this, the frame holds a shared_ptr
// back to this CoroTaskRunner, creating an unreachable reference cycle.
task_ = {};
}
inline void
JobQueue::CoroTaskRunner::join()
{
std::unique_lock<std::mutex> lk(mutex_run_);
cv_.wait(lk, [this]() { return running_ == false; });
}
} // 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/JobTypes.h>
#include <xrpl/core/detail/Workers.h>
@@ -10,7 +9,6 @@
#include <boost/coroutine/all.hpp>
#include <coroutine>
#include <set>
namespace xrpl {
@@ -121,112 +119,6 @@ public:
join();
};
/** C++20 coroutine lifecycle manager. Replaces Coro for new code. */
class CoroTaskRunner : public std::enable_shared_from_this<CoroTaskRunner>
{
private:
detail::LocalValues lvs_;
JobQueue& jq_;
JobType type_;
std::string name_;
bool running_;
std::mutex mutex_;
std::mutex mutex_run_;
std::condition_variable cv_;
CoroTask<void> task_;
// Type-erased storage to keep the coroutine callable alive.
// 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 call returns. Storing the
// callable on the heap here ensures it outlives the coroutine.
struct FuncBase
{
virtual ~FuncBase() = default;
};
template <class F>
struct FuncStore : FuncBase
{
F func;
explicit FuncStore(F&& f) : func(std::move(f))
{
}
};
std::unique_ptr<FuncBase> storedFunc_;
#ifndef NDEBUG
bool finished_ = false;
#endif
public:
struct create_t
{
explicit create_t() = default;
};
// Private: Used in the implementation of postCoroTask
CoroTaskRunner(create_t, JobQueue&, JobType, std::string const&);
// Not copy-constructible or assignable
CoroTaskRunner(CoroTaskRunner const&) = delete;
CoroTaskRunner&
operator=(CoroTaskRunner const&) = delete;
~CoroTaskRunner();
/** Initialize with a coroutine function.
Must be called exactly once, after the object is managed by
shared_ptr. This is handled automatically by postCoroTask().
*/
template <class F>
void
init(F&& f);
/** Increment the suspended coroutine count.
Called when the coroutine is about to suspend.
*/
void
onSuspend();
/** Decrement the suspended coroutine count without side effects.
Used to undo onSuspend() when a suspend is cancelled.
*/
void
onUndoSuspend();
/** Suspend coroutine execution.
Returns an awaiter for use with co_await.
Effects:
Increments nSuspend_ in the JobQueue.
The coroutine is suspended.
The caller must later call post() or resume() to continue.
*/
auto
suspend();
/** Schedule coroutine execution on the JobQueue.
@return true if the job is added to the JobQueue.
*/
bool
post();
/** Resume coroutine on current thread. */
void
resume();
/** Returns true if coroutine hasn't completed. */
bool
runnable() const;
/** Once called, allows early exit without an assert. */
void
expectEarlyExit();
/** Waits until coroutine completes. */
void
join();
};
using JobFunction = std::function<void()>;
JobQueue(
@@ -273,19 +165,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.
*/
int
@@ -500,7 +379,6 @@ private:
} // namespace xrpl
#include <xrpl/core/Coro.ipp>
#include <xrpl/core/CoroTaskRunner.ipp>
namespace xrpl {
@@ -523,26 +401,4 @@ JobQueue::postCoro(JobType t, std::string const& name, F&& f)
return coro;
}
template <class F>
std::shared_ptr<JobQueue::CoroTaskRunner>
JobQueue::postCoroTask(JobType t, std::string const& name, F&& f)
{
auto runner = std::make_shared<CoroTaskRunner>(CoroTaskRunner::create_t{}, *this, t, name);
runner->init(std::forward<F>(f));
// Account for the initial suspension (lazy start).
// Mirrors the yield() in the Boost Coro constructor.
{
std::lock_guard lock(m_mutex);
++nSuspend_;
}
if (!runner->post())
{
runner->expectEarlyExit();
runner.reset();
}
return runner;
}
} // namespace xrpl

View File

@@ -1,56 +0,0 @@
#pragma once
#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.
*/
struct JobQueueAwaiter
{
std::shared_ptr<JobQueue::CoroTaskRunner> runner;
bool
await_ready() const noexcept
{
return false;
}
bool
await_suspend(std::coroutine_handle<>)
{
// Increment nSuspend (equivalent to yield())
runner->onSuspend();
// Schedule resume on JobQueue (equivalent to post())
if (!runner->post())
{
// JobQueue is stopping. Undo the suspend count and
// don't actually suspend — the coroutine continues
// immediately so it can clean up and co_return.
runner->onUndoSuspend();
return false;
}
return true;
}
void
await_resume() const noexcept
{
}
};
} // namespace xrpl

View File

@@ -1,20 +1,19 @@
#pragma once
#include <boost/beast/core/string.hpp>
#include <functional>
#include <string>
#include <string_view>
namespace Json {
class Value;
using Output = std::function<void(boost::beast::string_view const&)>;
using Output = std::function<void(std::string_view const&)>;
inline Output
stringOutput(std::string& s)
{
return [&](boost::beast::string_view const& b) { s.append(b.data(), b.size()); };
return [&](std::string_view const& b) { s.append(b.data(), b.size()); };
}
/** Writes a minimal representation of a Json value to an Output in O(n) time.

View File

@@ -77,16 +77,16 @@ public:
If the object is not found or an error is encountered, the
result will indicate the condition.
@note This will be called concurrently.
@param key A pointer to the key data.
@param hash The hash of the object.
@param pObject [out] The created object if successful.
@return The result of the operation.
*/
virtual Status
fetch(void const* key, std::shared_ptr<NodeObject>* pObject) = 0;
fetch(uint256 const& hash, std::shared_ptr<NodeObject>* pObject) = 0;
/** Fetch a batch synchronously. */
virtual std::pair<std::vector<std::shared_ptr<NodeObject>>, Status>
fetchBatch(std::vector<uint256 const*> const& hashes) = 0;
fetchBatch(std::vector<uint256> const& hashes) = 0;
/** Store a single object.
Depending on the implementation this may happen immediately

View File

@@ -17,6 +17,7 @@
#include <functional>
#include <list>
#include <string_view>
namespace xrpl {
@@ -48,8 +49,7 @@ private:
bool ping_active_ = false;
boost::beast::websocket::ping_data payload_;
error_code ec_;
std::function<void(boost::beast::websocket::frame_type, boost::beast::string_view)>
control_callback_;
std::function<void(boost::beast::websocket::frame_type, std::string_view)> control_callback_;
public:
template <class Body, class Headers>
@@ -137,7 +137,7 @@ protected:
on_ping(error_code const& ec);
void
on_ping_pong(boost::beast::websocket::frame_type kind, boost::beast::string_view payload);
on_ping_pong(boost::beast::websocket::frame_type kind, std::string_view payload);
void
on_timer(error_code ec);
@@ -414,11 +414,11 @@ template <class Handler, class Impl>
void
BaseWSPeer<Handler, Impl>::on_ping_pong(
boost::beast::websocket::frame_type kind,
boost::beast::string_view payload)
std::string_view payload)
{
if (kind == boost::beast::websocket::frame_type::pong)
{
boost::beast::string_view p(payload_.begin());
std::string_view p(payload_.begin(), payload_.size());
if (payload == p)
{
close_on_timer_ = false;

View File

@@ -8,6 +8,7 @@
#include <set>
#include <stack>
#include <string>
#include <string_view>
#include <utility>
#include <vector>
@@ -87,14 +88,14 @@ public:
}
void
output(boost::beast::string_view const& bytes)
output(std::string_view const& bytes)
{
markStarted();
output_(bytes);
}
void
stringOutput(boost::beast::string_view const& bytes)
stringOutput(std::string_view const& bytes)
{
markStarted();
std::size_t position = 0, writtenUntil = 0;

View File

@@ -33,7 +33,7 @@ DatabaseNodeImp::fetchNodeObject(
try
{
status = backend_->fetch(hash.data(), &nodeObject);
status = backend_->fetch(hash, &nodeObject);
}
catch (std::exception const& e)
{
@@ -68,18 +68,10 @@ DatabaseNodeImp::fetchBatch(std::vector<uint256> const& hashes)
using namespace std::chrono;
auto const before = steady_clock::now();
std::vector<uint256 const*> batch{};
batch.reserve(hashes.size());
for (size_t i = 0; i < hashes.size(); ++i)
{
auto const& hash = hashes[i];
batch.push_back(&hash);
}
// Get the node objects that match the hashes from the backend. To protect
// against the backends returning fewer or more results than expected, the
// container is resized to the number of hashes.
auto results = backend_->fetchBatch(batch).first;
auto results = backend_->fetchBatch(hashes).first;
XRPL_ASSERT(
results.size() == hashes.size() || results.empty(),
"number of output objects either matches number of input hashes or is empty");

View File

@@ -105,7 +105,7 @@ DatabaseRotatingImp::fetchNodeObject(
std::shared_ptr<NodeObject> nodeObject;
try
{
status = backend->fetch(hash.data(), &nodeObject);
status = backend->fetch(hash, &nodeObject);
}
catch (std::exception const& e)
{

View File

@@ -116,10 +116,9 @@ public:
//--------------------------------------------------------------------------
Status
fetch(void const* key, std::shared_ptr<NodeObject>* pObject) override
fetch(uint256 const& hash, std::shared_ptr<NodeObject>* pObject) override
{
XRPL_ASSERT(db_, "xrpl::NodeStore::MemoryBackend::fetch : non-null database");
uint256 const hash(uint256::fromVoid(key));
std::lock_guard _(db_->mutex);
@@ -134,14 +133,14 @@ public:
}
std::pair<std::vector<std::shared_ptr<NodeObject>>, Status>
fetchBatch(std::vector<uint256 const*> const& hashes) override
fetchBatch(std::vector<uint256> const& hashes) override
{
std::vector<std::shared_ptr<NodeObject>> results;
results.reserve(hashes.size());
for (auto const& h : hashes)
{
std::shared_ptr<NodeObject> nObj;
Status status = fetch(h->begin(), &nObj);
Status status = fetch(h, &nObj);
if (status != ok)
results.push_back({});
else

View File

@@ -179,17 +179,17 @@ public:
}
Status
fetch(void const* key, std::shared_ptr<NodeObject>* pno) override
fetch(uint256 const& hash, std::shared_ptr<NodeObject>* pno) override
{
Status status;
pno->reset();
nudb::error_code ec;
db_.fetch(
key,
[key, pno, &status](void const* data, std::size_t size) {
hash.data(),
[&hash, pno, &status](void const* data, std::size_t size) {
nudb::detail::buffer bf;
auto const result = nodeobject_decompress(data, size, bf);
DecodedBlob decoded(key, result.first, result.second);
DecodedBlob decoded(hash.data(), result.first, result.second);
if (!decoded.wasOk())
{
status = dataCorrupt;
@@ -207,14 +207,14 @@ public:
}
std::pair<std::vector<std::shared_ptr<NodeObject>>, Status>
fetchBatch(std::vector<uint256 const*> const& hashes) override
fetchBatch(std::vector<uint256> const& hashes) override
{
std::vector<std::shared_ptr<NodeObject>> results;
results.reserve(hashes.size());
for (auto const& h : hashes)
{
std::shared_ptr<NodeObject> nObj;
Status status = fetch(h->begin(), &nObj);
Status status = fetch(h, &nObj);
if (status != ok)
results.push_back({});
else

View File

@@ -36,13 +36,13 @@ public:
}
Status
fetch(void const*, std::shared_ptr<NodeObject>*) override
fetch(uint256 const&, std::shared_ptr<NodeObject>*) override
{
return notFound;
}
std::pair<std::vector<std::shared_ptr<NodeObject>>, Status>
fetchBatch(std::vector<uint256 const*> const& hashes) override
fetchBatch(std::vector<uint256> const& hashes) override
{
return {};
}

View File

@@ -244,7 +244,7 @@ public:
//--------------------------------------------------------------------------
Status
fetch(void const* key, std::shared_ptr<NodeObject>* pObject) override
fetch(uint256 const& hash, std::shared_ptr<NodeObject>* pObject) override
{
XRPL_ASSERT(m_db, "xrpl::NodeStore::RocksDBBackend::fetch : non-null database");
pObject->reset();
@@ -252,7 +252,7 @@ public:
Status status(ok);
rocksdb::ReadOptions const options;
rocksdb::Slice const slice(static_cast<char const*>(key), m_keyBytes);
rocksdb::Slice const slice(std::bit_cast<char const*>(hash.data()), m_keyBytes);
std::string string;
@@ -260,7 +260,7 @@ public:
if (getStatus.ok())
{
DecodedBlob decoded(key, string.data(), string.size());
DecodedBlob decoded(hash.data(), string.data(), string.size());
if (decoded.wasOk())
{
@@ -295,14 +295,14 @@ public:
}
std::pair<std::vector<std::shared_ptr<NodeObject>>, Status>
fetchBatch(std::vector<uint256 const*> const& hashes) override
fetchBatch(std::vector<uint256> const& hashes) override
{
std::vector<std::shared_ptr<NodeObject>> results;
results.reserve(hashes.size());
for (auto const& h : hashes)
{
std::shared_ptr<NodeObject> nObj;
Status status = fetch(h->begin(), &nObj);
Status status = fetch(h, &nObj);
if (status != ok)
results.push_back({});
else
@@ -332,9 +332,8 @@ public:
EncodedBlob encoded(e);
wb.Put(
rocksdb::Slice(reinterpret_cast<char const*>(encoded.getKey()), m_keyBytes),
rocksdb::Slice(
reinterpret_cast<char const*>(encoded.getData()), encoded.getSize()));
rocksdb::Slice(std::bit_cast<char const*>(encoded.getKey()), m_keyBytes),
rocksdb::Slice(std::bit_cast<char const*>(encoded.getData()), encoded.getSize()));
}
rocksdb::WriteOptions const options;

View File

@@ -8,7 +8,6 @@
#include <xrpld/rpc/detail/Tuning.h>
#include <xrpl/beast/unit_test.h>
#include <xrpl/core/CoroTask.h>
#include <xrpl/core/JobQueue.h>
#include <xrpl/json/json_reader.h>
#include <xrpl/protocol/ApiVersion.h>
@@ -132,6 +131,7 @@ public:
c,
Role::USER,
{},
{},
RPC::apiVersionIfUnspecified},
{},
{}};
@@ -155,11 +155,11 @@ public:
Json::Value result;
gate g;
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;
@@ -240,27 +240,28 @@ public:
c,
Role::USER,
{},
{},
RPC::apiVersionIfUnspecified},
{},
{}};
Json::Value result;
gate g;
// Test RPC::Tuning::max_src_cur source currencies.
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::max_src_cur);
context.coro = coro;
RPC::doCommand(context, result);
g.signal();
co_return;
});
BEAST_EXPECT(g.wait_for(5s));
BEAST_EXPECT(!result.isMember(jss::error));
// Test more than RPC::Tuning::max_src_cur source currencies.
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::max_src_cur + 1);
context.coro = coro;
RPC::doCommand(context, result);
g.signal();
co_return;
});
BEAST_EXPECT(g.wait_for(5s));
BEAST_EXPECT(result.isMember(jss::error));
@@ -268,22 +269,22 @@ public:
// Test RPC::Tuning::max_auto_src_cur source currencies.
for (auto i = 0; i < (RPC::Tuning::max_auto_src_cur - 1); ++i)
env.trust(Account("alice")[std::to_string(i + 100)](100), "bob");
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.wait_for(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");
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.wait_for(5s));
BEAST_EXPECT(result.isMember(jss::error));

View File

@@ -1,460 +0,0 @@
#include <test/jtx.h>
#include <xrpl/core/JobQueue.h>
#include <xrpl/core/JobQueueAwaiter.h>
#include <chrono>
#include <mutex>
namespace xrpl {
namespace test {
class CoroTask_test : public beast::unit_test::suite
{
public:
class gate
{
private:
std::condition_variable cv_;
std::mutex mutex_;
bool signaled_ = false;
public:
template <class Rep, class Period>
bool
wait_for(std::chrono::duration<Rep, Period> const& rel_time)
{
std::unique_lock<std::mutex> lk(mutex_);
auto b = cv_.wait_for(lk, rel_time, [this] { return signaled_; });
signaled_ = false;
return b;
}
void
signal()
{
std::lock_guard 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.
// Test: CoroTask<void> runs to completion
void
testVoidCompletion()
{
using namespace std::chrono_literals;
using namespace jtx;
testcase("void completion");
Env env(*this, envconfig([](std::unique_ptr<Config> cfg) {
cfg->FORCE_MULTI_THREAD = 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);
BEAST_EXPECT(g.wait_for(5s));
runner->join();
BEAST_EXPECT(!runner->runnable());
}
// Test: 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->FORCE_MULTI_THREAD = true;
return cfg;
}));
gate g1, g2;
std::shared_ptr<JobQueue::CoroTaskRunner> r;
auto runner = env.app().getJobQueue().postCoroTask(
jtCLIENT,
"CoroTaskTest",
[rp = &r, g1p = &g1, g2p = &g2](auto runner) -> CoroTask<void> {
*rp = runner;
g1p->signal();
co_await runner->suspend();
g2p->signal();
co_return;
});
BEAST_EXPECT(runner);
BEAST_EXPECT(g1.wait_for(5s));
runner->join();
runner->post();
BEAST_EXPECT(g2.wait_for(5s));
runner->join();
}
// Test: incorrect_order — post before suspend
// Mirrors existing 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->FORCE_MULTI_THREAD = 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.wait_for(5s));
}
// Test: JobQueueAwaiter — suspend + auto-repost
void
testJobQueueAwaiter()
{
using namespace std::chrono_literals;
using namespace jtx;
testcase("JobQueueAwaiter");
Env env(*this, envconfig([](std::unique_ptr<Config> cfg) {
cfg->FORCE_MULTI_THREAD = true;
return cfg;
}));
gate g;
int step = 0;
env.app().getJobQueue().postCoroTask(
jtCLIENT, "CoroTaskTest", [sp = &step, gp = &g](auto runner) -> CoroTask<void> {
*sp = 1;
co_await JobQueueAwaiter{runner};
*sp = 2;
co_await JobQueueAwaiter{runner};
*sp = 3;
gp->signal();
co_return;
});
BEAST_EXPECT(g.wait_for(5s));
BEAST_EXPECT(step == 3);
}
// Test: thread_specific_storage — per-coroutine LocalValue isolation
// Mirrors existing 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 int const N = 4;
std::array<std::shared_ptr<JobQueue::CoroTaskRunner>, N> 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();
});
BEAST_EXPECT(g.wait_for(5s));
BEAST_EXPECT(*lv == -1);
for (int i = 0; i < N; ++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;
});
BEAST_EXPECT(g.wait_for(5s));
a[i]->join();
}
for (auto const& r : a)
{
r->post();
BEAST_EXPECT(g.wait_for(5s));
r->join();
}
for (auto const& r : a)
{
r->post();
r->join();
}
jq.addJob(jtCLIENT, "LocalValTest", [&]() {
this->BEAST_EXPECT(*lv == -2);
g.signal();
});
BEAST_EXPECT(g.wait_for(5s));
BEAST_EXPECT(*lv == -1);
}
// Test: exception propagation
void
testExceptionPropagation()
{
using namespace std::chrono_literals;
using namespace jtx;
testcase("exception propagation");
Env env(*this, envconfig([](std::unique_ptr<Config> cfg) {
cfg->FORCE_MULTI_THREAD = true;
return cfg;
}));
gate g;
auto runner = env.app().getJobQueue().postCoroTask(
jtCLIENT, "CoroTaskTest", [gp = &g](auto) -> CoroTask<void> {
gp->signal();
throw std::runtime_error("test exception");
co_return;
});
BEAST_EXPECT(runner);
BEAST_EXPECT(g.wait_for(5s));
runner->join();
// The exception is caught by promise_type::unhandled_exception()
// and the coroutine is considered done
BEAST_EXPECT(!runner->runnable());
}
// Test: multiple sequential co_await points
void
testMultipleYields()
{
using namespace std::chrono_literals;
using namespace jtx;
testcase("multiple yields");
Env env(*this, envconfig([](std::unique_ptr<Config> cfg) {
cfg->FORCE_MULTI_THREAD = true;
return cfg;
}));
gate g;
int counter = 0;
std::shared_ptr<JobQueue::CoroTaskRunner> r;
auto runner = env.app().getJobQueue().postCoroTask(
jtCLIENT,
"CoroTaskTest",
[rp = &r, cp = &counter, gp = &g](auto runner) -> CoroTask<void> {
*rp = runner;
++(*cp);
gp->signal();
co_await runner->suspend();
++(*cp);
gp->signal();
co_await runner->suspend();
++(*cp);
gp->signal();
co_return;
});
BEAST_EXPECT(runner);
BEAST_EXPECT(g.wait_for(5s));
BEAST_EXPECT(counter == 1);
runner->join();
runner->post();
BEAST_EXPECT(g.wait_for(5s));
BEAST_EXPECT(counter == 2);
runner->join();
runner->post();
BEAST_EXPECT(g.wait_for(5s));
BEAST_EXPECT(counter == 3);
runner->join();
BEAST_EXPECT(!runner->runnable());
}
// Test: CoroTask<T> returns a value via co_return
void
testValueReturn()
{
using namespace std::chrono_literals;
using namespace jtx;
testcase("value return");
Env env(*this, envconfig([](std::unique_ptr<Config> cfg) {
cfg->FORCE_MULTI_THREAD = 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);
BEAST_EXPECT(g.wait_for(5s));
runner->join();
BEAST_EXPECT(result == 42);
BEAST_EXPECT(!runner->runnable());
}
// Test: CoroTask<T> propagates exceptions from inner coroutines
void
testValueException()
{
using namespace std::chrono_literals;
using namespace jtx;
testcase("value exception");
Env env(*this, envconfig([](std::unique_ptr<Config> cfg) {
cfg->FORCE_MULTI_THREAD = true;
return cfg;
}));
gate g;
bool caught = false;
auto runner = env.app().getJobQueue().postCoroTask(
jtCLIENT, "CoroTaskTest", [cp = &caught, 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)
{
*cp = true;
}
gp->signal();
co_return;
});
BEAST_EXPECT(runner);
BEAST_EXPECT(g.wait_for(5s));
runner->join();
BEAST_EXPECT(caught);
BEAST_EXPECT(!runner->runnable());
}
// Test: CoroTask<T> chaining — nested value-returning coroutines
void
testValueChaining()
{
using namespace std::chrono_literals;
using namespace jtx;
testcase("value chaining");
Env env(*this, envconfig([](std::unique_ptr<Config> cfg) {
cfg->FORCE_MULTI_THREAD = 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 sum = co_await add(a, b);
co_return sum * 2;
};
*rp = co_await mul(3, 4);
gp->signal();
co_return;
});
BEAST_EXPECT(runner);
BEAST_EXPECT(g.wait_for(5s));
runner->join();
BEAST_EXPECT(result == 14); // (3 + 4) * 2
BEAST_EXPECT(!runner->runnable());
}
// Test: 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->FORCE_MULTI_THREAD = 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 test
} // namespace xrpl

View File

@@ -6,7 +6,6 @@
#include <xrpld/rpc/RPCHandler.h>
#include <xrpl/core/CoroTask.h>
#include <xrpl/protocol/ApiVersion.h>
#include <xrpl/protocol/STParsedJSON.h>
#include <xrpl/resource/Fees.h>
@@ -194,6 +193,7 @@ AMMTest::find_paths_request(
c,
Role::USER,
{},
{},
RPC::apiVersionIfUnspecified},
{},
{}};
@@ -215,11 +215,11 @@ AMMTest::find_paths_request(
Json::Value result;
gate g;
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

@@ -138,7 +138,7 @@ public:
{
std::shared_ptr<NodeObject> object;
Status const status = backend.fetch(batch[i]->getHash().cbegin(), &object);
Status const status = backend.fetch(batch[i]->getHash(), &object);
BEAST_EXPECT(status == ok);
@@ -158,7 +158,7 @@ public:
{
std::shared_ptr<NodeObject> object;
Status const status = backend.fetch(batch[i]->getHash().cbegin(), &object);
Status const status = backend.fetch(batch[i]->getHash(), &object);
BEAST_EXPECT(status == notFound);
}

View File

@@ -314,7 +314,7 @@ public:
std::shared_ptr<NodeObject> obj;
std::shared_ptr<NodeObject> result;
obj = seq1_.obj(dist_(gen_));
backend_.fetch(obj->getHash().data(), &result);
backend_.fetch(obj->getHash(), &result);
suite_.expect(result && isSame(result, obj));
}
catch (std::exception const& e)
@@ -377,9 +377,9 @@ public:
{
try
{
auto const key = seq2_.key(i);
auto const hash = seq2_.key(i);
std::shared_ptr<NodeObject> result;
backend_.fetch(key.data(), &result);
backend_.fetch(hash, &result);
suite_.expect(!result);
}
catch (std::exception const& e)
@@ -449,9 +449,9 @@ public:
{
if (rand_(gen_) < missingNodePercent)
{
auto const key = seq2_.key(dist_(gen_));
auto const hash = seq2_.key(dist_(gen_));
std::shared_ptr<NodeObject> result;
backend_.fetch(key.data(), &result);
backend_.fetch(hash, &result);
suite_.expect(!result);
}
else
@@ -459,7 +459,7 @@ public:
std::shared_ptr<NodeObject> obj;
std::shared_ptr<NodeObject> result;
obj = seq1_.obj(dist_(gen_));
backend_.fetch(obj->getHash().data(), &result);
backend_.fetch(obj->getHash(), &result);
suite_.expect(result && isSame(result, obj));
}
}
@@ -540,8 +540,7 @@ public:
std::shared_ptr<NodeObject> result;
auto const j = older_(gen_);
obj = seq1_.obj(j);
std::shared_ptr<NodeObject> result1;
backend_.fetch(obj->getHash().data(), &result);
backend_.fetch(obj->getHash(), &result);
suite_.expect(result != nullptr);
suite_.expect(isSame(result, obj));
}
@@ -559,7 +558,7 @@ public:
std::shared_ptr<NodeObject> result;
auto const j = recent_(gen_);
obj = seq1_.obj(j);
backend_.fetch(obj->getHash().data(), &result);
backend_.fetch(obj->getHash(), &result);
suite_.expect(!result || isSame(result, obj));
break;
}

View File

@@ -1425,6 +1425,7 @@ ApplicationImp::setup(boost::program_options::variables_map const& cmdline)
c,
Role::ADMIN,
{},
{},
RPC::apiMaximumSupportedVersion},
jvCommand};

View File

@@ -3,7 +3,6 @@
#include <xrpl/beast/core/CurrentThreadName.h>
#include <xrpl/beast/net/IPAddressConversion.h>
#include <xrpl/core/CoroTask.h>
#include <xrpl/resource/Fees.h>
namespace xrpl {
@@ -100,14 +99,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", [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 status{grpc::StatusCode::INTERNAL, "Job Queue is already stopped"};
responder_.FinishWithError(status, this);
@@ -116,7 +114,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
{
@@ -158,6 +156,7 @@ GRPCServerImpl::CallData<Request, Response>::processRequest()
app_.getLedgerMaster(),
usage,
role,
coro,
InfoSub::pointer(),
apiVersion},
request_};

View File

@@ -208,7 +208,7 @@ private:
private:
// 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

View File

@@ -59,7 +59,7 @@ to_string(ProtocolVersion const& p)
}
std::vector<ProtocolVersion>
parseProtocolVersions(boost::beast::string_view const& value)
parseProtocolVersions(std::string_view const& value)
{
static boost::regex re(
"^" // start of line
@@ -130,7 +130,7 @@ negotiateProtocolVersion(std::vector<ProtocolVersion> const& versions)
}
std::optional<ProtocolVersion>
negotiateProtocolVersion(boost::beast::string_view const& versions)
negotiateProtocolVersion(std::string_view const& versions)
{
auto const them = parseProtocolVersions(versions);

View File

@@ -1,10 +1,9 @@
#pragma once
#include <boost/beast/core/string.hpp>
#include <cstdint>
#include <optional>
#include <string>
#include <string_view>
#include <utility>
#include <vector>
@@ -39,7 +38,7 @@ to_string(ProtocolVersion const& p);
no duplicates and will be sorted in ascending protocol order.
*/
std::vector<ProtocolVersion>
parseProtocolVersions(boost::beast::string_view const& s);
parseProtocolVersions(std::string_view const& s);
/** Given a list of supported protocol versions, choose the one we prefer. */
std::optional<ProtocolVersion>
@@ -47,7 +46,7 @@ negotiateProtocolVersion(std::vector<ProtocolVersion> const& versions);
/** Given a list of supported protocol versions, choose the one we prefer. */
std::optional<ProtocolVersion>
negotiateProtocolVersion(boost::beast::string_view const& versions);
negotiateProtocolVersion(std::string_view const& versions);
/** The list of all the protocol versions we support. */
std::string const&

View File

@@ -3,6 +3,7 @@
#include <xrpld/rpc/Role.h>
#include <xrpl/beast/utility/Journal.h>
#include <xrpl/core/JobQueue.h>
#include <xrpl/server/InfoSub.h>
namespace xrpl {
@@ -23,6 +24,7 @@ struct Context
LedgerMaster& ledgerMaster;
Resource::Consumer& consumer;
Role role;
std::shared_ptr<JobQueue::Coro> coro{};
InfoSub::pointer infoSub{};
unsigned int apiVersion;
};

View File

@@ -169,10 +169,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(
@@ -180,6 +183,7 @@ private:
std::string const& request,
beast::IP::Endpoint const& remoteIPAddress,
Output&&,
std::shared_ptr<JobQueue::Coro> coro,
std::string_view forwardedFor,
std::string_view user);

View File

@@ -14,7 +14,6 @@
#include <xrpl/basics/make_SSLContext.h>
#include <xrpl/beast/net/IPAddressConversion.h>
#include <xrpl/beast/rfc2616.h>
#include <xrpl/core/CoroTask.h>
#include <xrpl/core/JobQueue.h>
#include <xrpl/json/json_reader.h>
#include <xrpl/json/to_string.h>
@@ -35,6 +34,7 @@
#include <algorithm>
#include <memory>
#include <stdexcept>
#include <string_view>
namespace xrpl {
@@ -231,7 +231,7 @@ ServerHandler::onHandoff(
static inline Json::Output
makeOutput(Session& session)
{
return [&](boost::beast::string_view const& b) { session.write(b.data(), b.size()); };
return [&](std::string_view const& b) { session.write(b.data(), b.size()); };
}
static std::map<std::string, std::string>
@@ -285,10 +285,9 @@ ServerHandler::onRequest(Session& session)
}
std::shared_ptr<Session> detachedSession = session.detach();
auto const postResult = m_jobQueue.postCoroTask(
jtCLIENT_RPC, "RPC-Client", [this, detachedSession](auto) -> CoroTask<void> {
processSession(detachedSession);
co_return;
auto const postResult = m_jobQueue.postCoro(
jtCLIENT_RPC, "RPC-Client", [this, detachedSession](std::shared_ptr<JobQueue::Coro> coro) {
processSession(detachedSession, coro);
});
if (postResult == nullptr)
{
@@ -324,18 +323,17 @@ ServerHandler::onWSMessage(
JLOG(m_journal.trace()) << "Websocket received '" << jv << "'";
auto const postResult = m_jobQueue.postCoroTask(
auto const postResult = m_jobQueue.postCoro(
jtCLIENT_WEBSOCKET,
"WS-Client",
[this, session, jv = std::move(jv)](auto) -> CoroTask<void> {
auto const jr = this->processSession(session, jv);
[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();
co_return;
});
if (postResult == nullptr)
{
@@ -376,7 +374,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(m_journal))
@@ -443,6 +444,7 @@ ServerHandler::processSession(std::shared_ptr<WSSession> const& session, Json::V
app_.getLedgerMaster(),
is->getConsumer(),
role,
coro,
is,
apiVersion},
jv,
@@ -513,20 +515,27 @@ 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(),
buffers_to_string(session->request().body().data()),
session->remoteAddress().at_port(0),
makeOutput(*session),
coro,
forwardedFor(session->request()),
[&] {
[&]() -> std::string_view {
auto const iter = session->request().find("X-User");
if (iter != session->request().end())
return iter->value();
return boost::beast::string_view{};
{
auto const val = iter->value();
return std::string_view(val.data(), val.size());
}
return std::string_view{};
}());
if (beast::rfc2616::is_keep_alive(session->request()))
@@ -557,6 +566,7 @@ ServerHandler::processRequest(
std::string const& request,
beast::IP::Endpoint const& remoteIPAddress,
Output&& output,
std::shared_ptr<JobQueue::Coro> coro,
std::string_view forwardedFor,
std::string_view user)
{
@@ -813,6 +823,7 @@ ServerHandler::processRequest(
app_.getLedgerMaster(),
usage,
role,
coro,
InfoSub::pointer(),
apiVersion},
params,

View File

@@ -7,9 +7,6 @@
#include <xrpl/protocol/RPCErr.h>
#include <xrpl/resource/Fees.h>
#include <condition_variable>
#include <mutex>
namespace xrpl {
// This interface is deprecated.
@@ -40,31 +37,98 @@ doRipplePathFind(RPC::JsonContext& context)
PathRequest::pointer request;
lpLedger = context.ledgerMaster.getClosedLedger();
// 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.
std::mutex mtx;
std::condition_variable cv;
bool pathDone = false;
// 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:
//
// 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
// rippled 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.getPathRequests().makeLegacyPathRequest(
request,
[&]() {
[&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> coroCopy{context.coro};
if (!coroCopy->post())
{
std::lock_guard lk(mtx);
pathDone = 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();
}
cv.notify_one();
},
context.consumer,
lpLedger,
context.params);
if (request)
{
std::unique_lock lk(mtx);
cv.wait(lk, [&] { return pathDone; });
context.coro->yield();
jvResult = request->doStatus(context.params);
}