Compare commits

...

11 Commits

Author SHA1 Message Date
Pratik Mankawde
eb83e111af Fix coroutine lambda lifetime and add value-returning tests
Store the coroutine callable on the heap in CoroTaskRunner::init()
via a type-erased FuncStore wrapper. 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. This caused stack-use-after-scope (ASAN), assertion
failures, and hangs across multiple compilers.

Also fix expectEarlyExit() to destroy the coroutine frame when
postCoroTask() fails, breaking a potential shared_ptr cycle.

Switch all coroutine test lambda captures from [&] to explicit
pointer-by-value as defense-in-depth against GCC 14 coroutine
frame corruption. Add value-returning CoroTask<T> tests.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-26 14:09:26 +00:00
Pratik Mankawde
464c09efc7 Apply pre-commit formatting fixes
clang-format: collapse single-line initializer lists and function
arguments. prettier: add blank lines in markdown lists.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-26 13:17:33 +00:00
Pratik Mankawde
897c75bc6b Add cspell dictionary words for coroutine migration plan doc
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-26 12:37:34 +00:00
Pratik Mankawde
ca15c0efd7 Add C++20 coroutine primitives: CoroTask, CoroTaskRunner, JobQueueAwaiter
Introduce the core building blocks for migrating from Boost.Coroutine to
C++20 stackless coroutines (Milestone 1):

- CoroTask<T>: RAII coroutine return type with promise_type, symmetric
  transfer via FinalAwaiter, and lazy start (suspend_always)
- CoroTaskRunner: Lifecycle manager (nested in JobQueue) mirroring the
  existing Coro class — handles LocalValues swap, nSuspend_ accounting,
  mutex-guarded resume, and join/post semantics
- JobQueueAwaiter: Convenience awaiter combining suspend + auto-repost,
  with graceful fallback when JobQueue is stopping
- postCoroTask(): JobQueue entry point for launching C++20 coroutines
- CoroTask_test.cpp: 8 unit tests covering completion, suspend/resume
  ordering, LocalValue isolation, exception propagation, and shutdown

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-25 17:38:28 +00:00
Pratik Mankawde
bb4bc1d167 doc updated with branch names
Signed-off-by: Pratik Mankawde <3397372+pratikmankawde@users.noreply.github.com>
2026-02-25 17:38:08 +00:00
Pratik Mankawde
b9d14fb9e1 document update
Signed-off-by: Pratik Mankawde <3397372+pratikmankawde@users.noreply.github.com>
2026-02-25 17:03:00 +00:00
Pratik Mankawde
af30b71043 Plan doc added
Signed-off-by: Pratik Mankawde <3397372+pratikmankawde@users.noreply.github.com>
2026-02-25 16:36:45 +00:00
Ayaz Salikhov
65e63ebef3 chore: Update cleanup-workspace to delete old .conan2 dir on macOS (#6412) 2026-02-25 01:12:16 +00:00
Valentin Balaschenko
bdd106d992 Explicitly trim the heap after cache sweeps (#6022)
Limited to Linux/glibc builds.
2026-02-24 21:33:13 +00:00
Valentin Balaschenko
24cbaf76a5 ci: Update prepare-runner action to fix macOS build environment (empty)
Updates XRPLF/actions prepare-runner to version 2cbf48101 which fixes
pip upgrade failures on macOS runners with Homebrew-managed Python.

* This commit was cherry-picked from "release-3.1", but ended up empty
  because the changes are already present. It is included only for
  accounting - to indicate that all changes/commits from the previous
  release will be in the next one.
2026-02-24 12:52:32 -05:00
Valentin Balaschenko
3a805cc646 Disable featureBatch and fixBatchInnerSigs amendments (#6402) 2026-02-24 12:49:59 -05:00
14 changed files with 2967 additions and 4 deletions

View File

@@ -101,7 +101,7 @@ jobs:
steps:
- name: Cleanup workspace (macOS and Windows)
if: ${{ runner.os == 'macOS' || runner.os == 'Windows' }}
uses: XRPLF/actions/cleanup-workspace@cf0433aa74563aead044a1e395610c96d65a37cf
uses: XRPLF/actions/cleanup-workspace@c7d9ce5ebb03c752a354889ecd870cadfc2b1cd4
- name: Checkout repository
uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2

View File

@@ -64,7 +64,7 @@ jobs:
steps:
- name: Cleanup workspace (macOS and Windows)
if: ${{ runner.os == 'macOS' || runner.os == 'Windows' }}
uses: XRPLF/actions/cleanup-workspace@cf0433aa74563aead044a1e395610c96d65a37cf
uses: XRPLF/actions/cleanup-workspace@c7d9ce5ebb03c752a354889ecd870cadfc2b1cd4
- name: Checkout repository
uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2

File diff suppressed because it is too large Load Diff

View File

@@ -71,6 +71,7 @@ words:
- coldwallet
- compr
- conanfile
- cppcoro
- conanrun
- confs
- connectability
@@ -101,9 +102,11 @@ words:
- Falco
- finalizers
- firewalled
- fcontext
- fmtdur
- fsanitize
- funclets
- gantt
- gcov
- gcovr
- ghead
@@ -185,6 +188,7 @@ words:
- ostr
- pargs
- partitioner
- pratik
- paychan
- paychans
- permdex
@@ -206,6 +210,7 @@ words:
- queuable
- Raphson
- replayer
- repost
- rerere
- retriable
- RIPD
@@ -236,6 +241,7 @@ words:
- soci
- socidb
- sslws
- stackful
- statsd
- STATSDCOLLECTOR
- stissue

View File

@@ -0,0 +1,73 @@
#pragma once
#include <xrpl/beast/utility/Journal.h>
#include <chrono>
#include <cstdint>
#include <string_view>
namespace xrpl {
// cSpell:ignore ptmalloc
// -----------------------------------------------------------------------------
// Allocator interaction note:
// - This facility invokes glibc's malloc_trim(0) on Linux/glibc to request that
// ptmalloc return free heap pages to the OS.
// - If an alternative allocator (e.g. jemalloc or tcmalloc) is linked or
// preloaded (LD_PRELOAD), calling glibc's malloc_trim typically has no effect
// on the *active* heap. The call is harmless but may not reclaim memory
// because those allocators manage their own arenas.
// - Only glibc sbrk/arena space is eligible for trimming; large mmap-backed
// allocations are usually returned to the OS on free regardless of trimming.
// - Call at known reclamation points (e.g., after cache sweeps / online delete)
// and consider rate limiting to avoid churn.
// -----------------------------------------------------------------------------
struct MallocTrimReport
{
bool supported{false};
int trimResult{-1};
std::int64_t rssBeforeKB{-1};
std::int64_t rssAfterKB{-1};
std::chrono::microseconds durationUs{-1};
std::int64_t minfltDelta{-1};
std::int64_t majfltDelta{-1};
[[nodiscard]] std::int64_t
deltaKB() const noexcept
{
if (rssBeforeKB < 0 || rssAfterKB < 0)
return 0;
return rssAfterKB - rssBeforeKB;
}
};
/**
* @brief Attempt to return freed memory to the operating system.
*
* On Linux with glibc malloc, this issues ::malloc_trim(0), which may release
* free space from ptmalloc arenas back to the kernel. On other platforms, or if
* a different allocator is in use, this function is a no-op and the report will
* indicate that trimming is unsupported or had no effect.
*
* @param tag Identifier for logging/debugging purposes.
* @param journal Journal for diagnostic logging.
* @return Report containing before/after metrics and the trim result.
*
* @note If an alternative allocator (jemalloc/tcmalloc) is linked or preloaded,
* calling glibc's malloc_trim may have no effect on the active heap. The
* call is harmless but typically does not reclaim memory under those
* allocators.
*
* @note Only memory served from glibc's sbrk/arena heaps is eligible for trim.
* Large allocations satisfied via mmap are usually returned on free
* independently of trimming.
*
* @note Intended for use after operations that free significant memory (e.g.,
* cache sweeps, ledger cleanup, online delete). Consider rate limiting.
*/
MallocTrimReport
mallocTrim(std::string_view tag, beast::Journal journal);
} // namespace xrpl

View File

@@ -0,0 +1,289 @@
#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

@@ -0,0 +1,158 @@
#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,6 +2,7 @@
#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>
@@ -9,6 +10,7 @@
#include <boost/coroutine/all.hpp>
#include <coroutine>
#include <set>
namespace xrpl {
@@ -119,6 +121,112 @@ 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(
@@ -165,6 +273,19 @@ 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
@@ -379,6 +500,7 @@ private:
} // namespace xrpl
#include <xrpl/core/Coro.ipp>
#include <xrpl/core/CoroTaskRunner.ipp>
namespace xrpl {
@@ -401,4 +523,26 @@ 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

@@ -0,0 +1,56 @@
#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

@@ -15,9 +15,10 @@
// Add new amendments to the top of this list.
// Keep it sorted in reverse chronological order.
XRPL_FIX (PermissionedDomainInvariant, Supported::yes, VoteBehavior::DefaultNo)
XRPL_FIX (ExpiredNFTokenOfferRemoval, Supported::yes, VoteBehavior::DefaultNo)
XRPL_FIX (BatchInnerSigs, Supported::yes, VoteBehavior::DefaultNo)
XRPL_FIX (BatchInnerSigs, Supported::no, VoteBehavior::DefaultNo)
XRPL_FEATURE(LendingProtocol, Supported::yes, VoteBehavior::DefaultNo)
XRPL_FEATURE(PermissionDelegationV1_1, Supported::no, VoteBehavior::DefaultNo)
XRPL_FIX (DirectoryLimit, Supported::yes, VoteBehavior::DefaultNo)
@@ -31,7 +32,7 @@ XRPL_FEATURE(TokenEscrow, Supported::yes, VoteBehavior::DefaultNo
XRPL_FIX (EnforceNFTokenTrustlineV2, Supported::yes, VoteBehavior::DefaultNo)
XRPL_FIX (AMMv1_3, Supported::yes, VoteBehavior::DefaultNo)
XRPL_FEATURE(PermissionedDEX, Supported::yes, VoteBehavior::DefaultNo)
XRPL_FEATURE(Batch, Supported::yes, VoteBehavior::DefaultNo)
XRPL_FEATURE(Batch, Supported::no, VoteBehavior::DefaultNo)
XRPL_FEATURE(SingleAssetVault, Supported::yes, VoteBehavior::DefaultNo)
XRPL_FIX (PayChanCancelAfter, Supported::yes, VoteBehavior::DefaultNo)
// Check flags in Credential transactions

View File

@@ -0,0 +1,157 @@
#include <xrpl/basics/Log.h>
#include <xrpl/basics/MallocTrim.h>
#include <boost/predef.h>
#include <chrono>
#include <cstdint>
#include <cstdio>
#include <fstream>
#include <sstream>
#if defined(__GLIBC__) && BOOST_OS_LINUX
#include <sys/resource.h>
#include <malloc.h>
#include <unistd.h>
// Require RUSAGE_THREAD for thread-scoped page fault tracking
#ifndef RUSAGE_THREAD
#error "MallocTrim rusage instrumentation requires RUSAGE_THREAD on Linux/glibc"
#endif
namespace {
bool
getRusageThread(struct rusage& ru)
{
return ::getrusage(RUSAGE_THREAD, &ru) == 0; // LCOV_EXCL_LINE
}
} // namespace
#endif
namespace xrpl {
namespace detail {
// cSpell:ignore statm
#if defined(__GLIBC__) && BOOST_OS_LINUX
inline int
mallocTrimWithPad(std::size_t padBytes)
{
return ::malloc_trim(padBytes);
}
long
parseStatmRSSkB(std::string const& statm)
{
// /proc/self/statm format: size resident shared text lib data dt
// We want the second field (resident) which is in pages
std::istringstream iss(statm);
long size, resident;
if (!(iss >> size >> resident))
return -1;
// Convert pages to KB
long const pageSize = ::sysconf(_SC_PAGESIZE);
if (pageSize <= 0)
return -1;
return (resident * pageSize) / 1024;
}
#endif // __GLIBC__ && BOOST_OS_LINUX
} // namespace detail
MallocTrimReport
mallocTrim(std::string_view tag, beast::Journal journal)
{
// LCOV_EXCL_START
MallocTrimReport report;
#if !(defined(__GLIBC__) && BOOST_OS_LINUX)
JLOG(journal.debug()) << "malloc_trim not supported on this platform (tag=" << tag << ")";
#else
// Keep glibc malloc_trim padding at 0 (default): 12h Mainnet tests across 0/256KB/1MB/16MB
// showed no clear, consistent benefit from custom padding—0 provided the best overall balance
// of RSS reduction and trim-latency stability without adding a tuning surface.
constexpr std::size_t TRIM_PAD = 0;
report.supported = true;
if (journal.debug())
{
auto readFile = [](std::string const& path) -> std::string {
std::ifstream ifs(path, std::ios::in | std::ios::binary);
if (!ifs.is_open())
return {};
// /proc files are often not seekable; read as a stream.
std::ostringstream oss;
oss << ifs.rdbuf();
return oss.str();
};
std::string const tagStr{tag};
std::string const statmPath = "/proc/self/statm";
auto const statmBefore = readFile(statmPath);
long const rssBeforeKB = detail::parseStatmRSSkB(statmBefore);
struct rusage ru0{};
bool const have_ru0 = getRusageThread(ru0);
auto const t0 = std::chrono::steady_clock::now();
report.trimResult = detail::mallocTrimWithPad(TRIM_PAD);
auto const t1 = std::chrono::steady_clock::now();
struct rusage ru1{};
bool const have_ru1 = getRusageThread(ru1);
auto const statmAfter = readFile(statmPath);
long const rssAfterKB = detail::parseStatmRSSkB(statmAfter);
// Populate report fields
report.rssBeforeKB = rssBeforeKB;
report.rssAfterKB = rssAfterKB;
report.durationUs = std::chrono::duration_cast<std::chrono::microseconds>(t1 - t0);
if (have_ru0 && have_ru1)
{
report.minfltDelta = ru1.ru_minflt - ru0.ru_minflt;
report.majfltDelta = ru1.ru_majflt - ru0.ru_majflt;
}
std::int64_t const deltaKB = (rssBeforeKB < 0 || rssAfterKB < 0)
? 0
: (static_cast<std::int64_t>(rssAfterKB) - static_cast<std::int64_t>(rssBeforeKB));
JLOG(journal.debug()) << "malloc_trim tag=" << tagStr << " result=" << report.trimResult
<< " pad=" << TRIM_PAD << " bytes"
<< " rss_before=" << rssBeforeKB << "kB"
<< " rss_after=" << rssAfterKB << "kB"
<< " delta=" << deltaKB << "kB"
<< " duration_us=" << report.durationUs.count()
<< " minflt_delta=" << report.minfltDelta
<< " majflt_delta=" << report.majfltDelta;
}
else
{
report.trimResult = detail::mallocTrimWithPad(TRIM_PAD);
}
#endif
return report;
// LCOV_EXCL_STOP
}
} // namespace xrpl

View File

@@ -0,0 +1,460 @@
#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

@@ -0,0 +1,209 @@
#include <xrpl/basics/MallocTrim.h>
#include <boost/predef.h>
#include <gtest/gtest.h>
using namespace xrpl;
// cSpell:ignore statm
#if defined(__GLIBC__) && BOOST_OS_LINUX
namespace xrpl::detail {
long
parseStatmRSSkB(std::string const& statm);
} // namespace xrpl::detail
#endif
TEST(MallocTrimReport, structure)
{
// Test default construction
MallocTrimReport report;
EXPECT_EQ(report.supported, false);
EXPECT_EQ(report.trimResult, -1);
EXPECT_EQ(report.rssBeforeKB, -1);
EXPECT_EQ(report.rssAfterKB, -1);
EXPECT_EQ(report.durationUs, std::chrono::microseconds{-1});
EXPECT_EQ(report.minfltDelta, -1);
EXPECT_EQ(report.majfltDelta, -1);
EXPECT_EQ(report.deltaKB(), 0);
// Test deltaKB calculation - memory freed
report.rssBeforeKB = 1000;
report.rssAfterKB = 800;
EXPECT_EQ(report.deltaKB(), -200);
// Test deltaKB calculation - memory increased
report.rssBeforeKB = 500;
report.rssAfterKB = 600;
EXPECT_EQ(report.deltaKB(), 100);
// Test deltaKB calculation - no change
report.rssBeforeKB = 1234;
report.rssAfterKB = 1234;
EXPECT_EQ(report.deltaKB(), 0);
}
#if defined(__GLIBC__) && BOOST_OS_LINUX
TEST(parseStatmRSSkB, standard_format)
{
using xrpl::detail::parseStatmRSSkB;
// Test standard format: size resident shared text lib data dt
// Assuming 4KB page size: resident=1000 pages = 4000 KB
{
std::string statm = "25365 1000 2377 0 0 5623 0";
long result = parseStatmRSSkB(statm);
// Note: actual result depends on system page size
// On most systems it's 4KB, so 1000 pages = 4000 KB
EXPECT_GT(result, 0);
}
// Test with newline
{
std::string statm = "12345 2000 1234 0 0 3456 0\n";
long result = parseStatmRSSkB(statm);
EXPECT_GT(result, 0);
}
// Test with tabs
{
std::string statm = "12345\t2000\t1234\t0\t0\t3456\t0";
long result = parseStatmRSSkB(statm);
EXPECT_GT(result, 0);
}
// Test zero resident pages
{
std::string statm = "25365 0 2377 0 0 5623 0";
long result = parseStatmRSSkB(statm);
EXPECT_EQ(result, 0);
}
// Test with extra whitespace
{
std::string statm = " 25365 1000 2377 ";
long result = parseStatmRSSkB(statm);
EXPECT_GT(result, 0);
}
// Test empty string
{
std::string statm = "";
long result = parseStatmRSSkB(statm);
EXPECT_EQ(result, -1);
}
// Test malformed data (only one field)
{
std::string statm = "25365";
long result = parseStatmRSSkB(statm);
EXPECT_EQ(result, -1);
}
// Test malformed data (non-numeric)
{
std::string statm = "abc def ghi";
long result = parseStatmRSSkB(statm);
EXPECT_EQ(result, -1);
}
// Test malformed data (second field non-numeric)
{
std::string statm = "25365 abc 2377";
long result = parseStatmRSSkB(statm);
EXPECT_EQ(result, -1);
}
}
#endif
TEST(mallocTrim, without_debug_logging)
{
beast::Journal journal{beast::Journal::getNullSink()};
MallocTrimReport report = mallocTrim("without_debug", journal);
#if defined(__GLIBC__) && BOOST_OS_LINUX
EXPECT_EQ(report.supported, true);
EXPECT_GE(report.trimResult, 0);
EXPECT_EQ(report.durationUs, std::chrono::microseconds{-1});
EXPECT_EQ(report.minfltDelta, -1);
EXPECT_EQ(report.majfltDelta, -1);
#else
EXPECT_EQ(report.supported, false);
EXPECT_EQ(report.trimResult, -1);
EXPECT_EQ(report.rssBeforeKB, -1);
EXPECT_EQ(report.rssAfterKB, -1);
EXPECT_EQ(report.durationUs, std::chrono::microseconds{-1});
EXPECT_EQ(report.minfltDelta, -1);
EXPECT_EQ(report.majfltDelta, -1);
#endif
}
TEST(mallocTrim, empty_tag)
{
beast::Journal journal{beast::Journal::getNullSink()};
MallocTrimReport report = mallocTrim("", journal);
#if defined(__GLIBC__) && BOOST_OS_LINUX
EXPECT_EQ(report.supported, true);
EXPECT_GE(report.trimResult, 0);
#else
EXPECT_EQ(report.supported, false);
#endif
}
TEST(mallocTrim, with_debug_logging)
{
struct DebugSink : public beast::Journal::Sink
{
DebugSink() : Sink(beast::severities::kDebug, false)
{
}
void
write(beast::severities::Severity, std::string const&) override
{
}
void
writeAlways(beast::severities::Severity, std::string const&) override
{
}
};
DebugSink sink;
beast::Journal journal{sink};
MallocTrimReport report = mallocTrim("debug_test", journal);
#if defined(__GLIBC__) && BOOST_OS_LINUX
EXPECT_EQ(report.supported, true);
EXPECT_GE(report.trimResult, 0);
EXPECT_GE(report.durationUs.count(), 0);
EXPECT_GE(report.minfltDelta, 0);
EXPECT_GE(report.majfltDelta, 0);
#else
EXPECT_EQ(report.supported, false);
EXPECT_EQ(report.trimResult, -1);
EXPECT_EQ(report.durationUs, std::chrono::microseconds{-1});
EXPECT_EQ(report.minfltDelta, -1);
EXPECT_EQ(report.majfltDelta, -1);
#endif
}
TEST(mallocTrim, repeated_calls)
{
beast::Journal journal{beast::Journal::getNullSink()};
// Call malloc_trim multiple times to ensure it's safe
for (int i = 0; i < 5; ++i)
{
MallocTrimReport report = mallocTrim("iteration_" + std::to_string(i), journal);
#if defined(__GLIBC__) && BOOST_OS_LINUX
EXPECT_EQ(report.supported, true);
EXPECT_GE(report.trimResult, 0);
#else
EXPECT_EQ(report.supported, false);
#endif
}
}

View File

@@ -31,6 +31,7 @@
#include <xrpld/shamap/NodeFamily.h>
#include <xrpl/basics/ByteUtilities.h>
#include <xrpl/basics/MallocTrim.h>
#include <xrpl/basics/ResolverAsio.h>
#include <xrpl/basics/random.h>
#include <xrpl/beast/asio/io_latency_probe.h>
@@ -1053,6 +1054,8 @@ public:
<< "; size after: " << cachedSLEs_.size();
}
mallocTrim("doSweep", m_journal);
// Set timer to do another sweep later.
setSweepTimer();
}