ARCHIVE/clio
1
0
Fork 0
mirror of https://github.com/XRPLF/clio.git synced 2025-12-06 17:27:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: Async framework submit on strand/ctx (#2751)

Browse Source
This commit is contained in:
Alex Kremer
2025-11-04 19:14:31 +00:00
committed by GitHub
parent d6ab2cc1e4
commit 6d79dd6b2b
14 changed files with 296 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
src/util/Assert.hpp
View File

@@ -22,6 +22,7 @@
#include "util/SourceLocation.hpp" #include "util/SourceLocation.hpp"
#include <boost/log/core/core.hpp> #include <boost/log/core/core.hpp>
#include <fmt/base.h>
#include <functional> #include <functional>
#include <string_view> #include <string_view>

79
src/util/async/AnyExecutionContext.hpp
View File

@@ -85,15 +85,15 @@ public:
[[nodiscard]] auto [[nodiscard]] auto
execute(SomeHandlerWithoutStopToken auto&& fn) execute(SomeHandlerWithoutStopToken auto&& fn)
{ {
using RetType = std::decay_t<decltype(fn())>; using RetType = std::decay_t<std::invoke_result_t<decltype(fn)>>;
static_assert(not std::is_same_v<RetType, std::any>); static_assert(not std::is_same_v<RetType, std::any>);
return AnyOperation<RetType>(pimpl_->execute([fn = std::forward<decltype(fn)>(fn)]() -> std::any { return AnyOperation<RetType>(pimpl_->execute([fn = std::forward<decltype(fn)>(fn)] mutable -> std::any {
if constexpr (std::is_void_v<RetType>) { if constexpr (std::is_void_v<RetType>) {
fn(); std::invoke(std::forward<decltype(fn)>(fn));
return {}; return {};
} else { } else {
return std::make_any<RetType>(fn()); return std::make_any<RetType>(std::invoke(std::forward<decltype(fn)>(fn)));
} }
})); }));
} }
@@ -109,17 +109,19 @@ public:
[[nodiscard]] auto [[nodiscard]] auto
execute(SomeHandlerWith<AnyStopToken> auto&& fn) execute(SomeHandlerWith<AnyStopToken> auto&& fn)
{ {
using RetType = std::decay_t<decltype(fn(std::declval<AnyStopToken>()))>; using RetType = std::decay_t<std::invoke_result_t<decltype(fn), AnyStopToken>>;
static_assert(not std::is_same_v<RetType, std::any>); static_assert(not std::is_same_v<RetType, std::any>);
return AnyOperation<RetType>(pimpl_->execute([fn = std::forward<decltype(fn)>(fn)](auto stopToken) -> std::any { return AnyOperation<RetType>(
if constexpr (std::is_void_v<RetType>) { pimpl_->execute([fn = std::forward<decltype(fn)>(fn)](auto stopToken) mutable -> std::any {
fn(std::move(stopToken)); if constexpr (std::is_void_v<RetType>) {
return {}; std::invoke(std::forward<decltype(fn)>(fn), std::move(stopToken));
} else { return {};
return std::make_any<RetType>(fn(std::move(stopToken))); } else {
} return std::make_any<RetType>(std::invoke(std::forward<decltype(fn)>(fn), std::move(stopToken)));
})); }
})
);
} }
/** /**
@@ -134,16 +136,16 @@ public:
[[nodiscard]] auto [[nodiscard]] auto
execute(SomeHandlerWith<AnyStopToken> auto&& fn, SomeStdDuration auto timeout) execute(SomeHandlerWith<AnyStopToken> auto&& fn, SomeStdDuration auto timeout)
{ {
using RetType = std::decay_t<decltype(fn(std::declval<AnyStopToken>()))>; using RetType = std::decay_t<std::invoke_result_t<decltype(fn), AnyStopToken>>;
static_assert(not std::is_same_v<RetType, std::any>); static_assert(not std::is_same_v<RetType, std::any>);
return AnyOperation<RetType>(pimpl_->execute( return AnyOperation<RetType>(pimpl_->execute(
[fn = std::forward<decltype(fn)>(fn)](auto stopToken) -> std::any { [fn = std::forward<decltype(fn)>(fn)](auto stopToken) mutable -> std::any {
if constexpr (std::is_void_v<RetType>) { if constexpr (std::is_void_v<RetType>) {
fn(std::move(stopToken)); std::invoke(std::forward<decltype(fn)>(fn), std::move(stopToken));
return {}; return {};
} else { } else {
return std::make_any<RetType>(fn(std::move(stopToken))); return std::make_any<RetType>(std::invoke(std::forward<decltype(fn)>(fn), std::move(stopToken)));
} }
}, },
std::chrono::duration_cast<std::chrono::milliseconds>(timeout) std::chrono::duration_cast<std::chrono::milliseconds>(timeout)
@@ -162,17 +164,17 @@ public:
[[nodiscard]] auto [[nodiscard]] auto
scheduleAfter(SomeStdDuration auto delay, SomeHandlerWith<AnyStopToken> auto&& fn) scheduleAfter(SomeStdDuration auto delay, SomeHandlerWith<AnyStopToken> auto&& fn)
{ {
using RetType = std::decay_t<decltype(fn(std::declval<AnyStopToken>()))>; using RetType = std::decay_t<std::invoke_result_t<decltype(fn), AnyStopToken>>;
static_assert(not std::is_same_v<RetType, std::any>); static_assert(not std::is_same_v<RetType, std::any>);
auto const millis = std::chrono::duration_cast<std::chrono::milliseconds>(delay); auto const millis = std::chrono::duration_cast<std::chrono::milliseconds>(delay);
return AnyOperation<RetType>( return AnyOperation<RetType>(
pimpl_->scheduleAfter(millis, [fn = std::forward<decltype(fn)>(fn)](auto stopToken) -> std::any { pimpl_->scheduleAfter(millis, [fn = std::forward<decltype(fn)>(fn)](auto stopToken) mutable -> std::any {
if constexpr (std::is_void_v<RetType>) { if constexpr (std::is_void_v<RetType>) {
fn(std::move(stopToken)); std::invoke(std::forward<decltype(fn)>(fn), std::move(stopToken));
return {}; return {};
} else { } else {
return std::make_any<RetType>(fn(std::move(stopToken))); return std::make_any<RetType>(std::invoke(std::forward<decltype(fn)>(fn), std::move(stopToken)));
} }
}) })
); );
@@ -191,17 +193,19 @@ public:
[[nodiscard]] auto [[nodiscard]] auto
scheduleAfter(SomeStdDuration auto delay, SomeHandlerWith<AnyStopToken, bool> auto&& fn) scheduleAfter(SomeStdDuration auto delay, SomeHandlerWith<AnyStopToken, bool> auto&& fn)
{ {
using RetType = std::decay_t<decltype(fn(std::declval<AnyStopToken>(), true))>; using RetType = std::decay_t<std::invoke_result_t<decltype(fn), AnyStopToken, bool>>;
static_assert(not std::is_same_v<RetType, std::any>); static_assert(not std::is_same_v<RetType, std::any>);
auto const millis = std::chrono::duration_cast<std::chrono::milliseconds>(delay); auto const millis = std::chrono::duration_cast<std::chrono::milliseconds>(delay);
return AnyOperation<RetType>(pimpl_->scheduleAfter( return AnyOperation<RetType>(pimpl_->scheduleAfter(
millis, [fn = std::forward<decltype(fn)>(fn)](auto stopToken, auto cancelled) -> std::any { millis, [fn = std::forward<decltype(fn)>(fn)](auto stopToken, auto cancelled) mutable -> std::any {
if constexpr (std::is_void_v<RetType>) { if constexpr (std::is_void_v<RetType>) {
fn(std::move(stopToken), cancelled); std::invoke(std::forward<decltype(fn)>(fn), std::move(stopToken), cancelled);
return {}; return {};
} else { } else {
return std::make_any<RetType>(fn(std::move(stopToken), cancelled)); return std::make_any<RetType>(
std::invoke(std::forward<decltype(fn)>(fn), std::move(stopToken), cancelled)
);
} }
} }
)); ));
@@ -217,18 +221,30 @@ public:
[[nodiscard]] auto [[nodiscard]] auto
executeRepeatedly(SomeStdDuration auto interval, SomeHandlerWithoutStopToken auto&& fn) executeRepeatedly(SomeStdDuration auto interval, SomeHandlerWithoutStopToken auto&& fn)
{ {
using RetType = std::decay_t<decltype(fn())>; using RetType = std::decay_t<std::invoke_result_t<decltype(fn)>>;
static_assert(not std::is_same_v<RetType, std::any>); static_assert(not std::is_same_v<RetType, std::any>);
auto const millis = std::chrono::duration_cast<std::chrono::milliseconds>(interval); auto const millis = std::chrono::duration_cast<std::chrono::milliseconds>(interval);
return AnyOperation<RetType>( // return AnyOperation<RetType>( //
pimpl_->executeRepeatedly(millis, [fn = std::forward<decltype(fn)>(fn)] -> std::any { pimpl_->executeRepeatedly(millis, [fn = std::forward<decltype(fn)>(fn)] mutable -> std::any {
fn(); std::invoke(std::forward<decltype(fn)>(fn));
return {}; return {};
}) })
); );
} }
/**
* @brief Schedule an operation on the execution context without expectations of a result
* @note Exceptions are caught internally and `ASSERT`ed on
*
* @param fn The block of code to execute
*/
void
submit(SomeHandlerWithoutStopToken auto&& fn)
{
pimpl_->submit(std::forward<decltype(fn)>(fn));
}
/** /**
* @brief Make a strand for this execution context * @brief Make a strand for this execution context
* *
@@ -276,6 +292,7 @@ private:
virtual impl::ErasedOperation virtual impl::ErasedOperation
scheduleAfter(std::chrono::milliseconds, std::function<std::any(AnyStopToken, bool)>) = 0; scheduleAfter(std::chrono::milliseconds, std::function<std::any(AnyStopToken, bool)>) = 0;
virtual impl::ErasedOperation executeRepeatedly(std::chrono::milliseconds, std::function<std::any()>) = 0; virtual impl::ErasedOperation executeRepeatedly(std::chrono::milliseconds, std::function<std::any()>) = 0;
virtual void submit(std::function<void()>) = 0;
virtual AnyStrand virtual AnyStrand
makeStrand() = 0; makeStrand() = 0;
virtual void virtual void
@@ -323,6 +340,12 @@ private:
return ctx.executeRepeatedly(interval, std::move(fn)); return ctx.executeRepeatedly(interval, std::move(fn));
} }
void
submit(std::function<void()> fn) override
{
return ctx.submit(std::move(fn));
}
AnyStrand AnyStrand
makeStrand() override makeStrand() override
{ {

51
src/util/async/AnyStrand.hpp
View File

@@ -64,16 +64,16 @@ public:
[[nodiscard]] auto [[nodiscard]] auto
execute(SomeHandlerWithoutStopToken auto&& fn) execute(SomeHandlerWithoutStopToken auto&& fn)
{ {
using RetType = std::decay_t<decltype(fn())>; using RetType = std::decay_t<std::invoke_result_t<decltype(fn)>>;
static_assert(not std::is_same_v<RetType, std::any>); static_assert(not std::is_same_v<RetType, std::any>);
return AnyOperation<RetType>( // return AnyOperation<RetType>( //
pimpl_->execute([fn = std::forward<decltype(fn)>(fn)]() -> std::any { pimpl_->execute([fn = std::forward<decltype(fn)>(fn)] mutable -> std::any {
if constexpr (std::is_void_v<RetType>) { if constexpr (std::is_void_v<RetType>) {
fn(); std::invoke(std::forward<decltype(fn)>(fn));
return {}; return {};
} else { } else {
return std::make_any<RetType>(fn()); return std::make_any<RetType>(std::invoke(std::forward<decltype(fn)>(fn)));
} }
}) })
); );
@@ -88,16 +88,16 @@ public:
[[nodiscard]] auto [[nodiscard]] auto
execute(SomeHandlerWith<AnyStopToken> auto&& fn) execute(SomeHandlerWith<AnyStopToken> auto&& fn)
{ {
using RetType = std::decay_t<decltype(fn(std::declval<AnyStopToken>()))>; using RetType = std::decay_t<std::invoke_result_t<decltype(fn), AnyStopToken>>;
static_assert(not std::is_same_v<RetType, std::any>); static_assert(not std::is_same_v<RetType, std::any>);
return AnyOperation<RetType>( // return AnyOperation<RetType>( //
pimpl_->execute([fn = std::forward<decltype(fn)>(fn)](auto stopToken) -> std::any { pimpl_->execute([fn = std::forward<decltype(fn)>(fn)](auto stopToken) mutable -> std::any {
if constexpr (std::is_void_v<RetType>) { if constexpr (std::is_void_v<RetType>) {
fn(std::move(stopToken)); std::invoke(std::forward<decltype(fn)>(fn), std::move(stopToken));
return {}; return {};
} else { } else {
return std::make_any<RetType>(fn(std::move(stopToken))); return std::make_any<RetType>(std::invoke(std::forward<decltype(fn)>(fn), std::move(stopToken)));
} }
}) })
); );
@@ -113,17 +113,19 @@ public:
[[nodiscard]] auto [[nodiscard]] auto
execute(SomeHandlerWith<AnyStopToken> auto&& fn, SomeStdDuration auto timeout) execute(SomeHandlerWith<AnyStopToken> auto&& fn, SomeStdDuration auto timeout)
{ {
using RetType = std::decay_t<decltype(fn(std::declval<AnyStopToken>()))>; using RetType = std::decay_t<std::invoke_result_t<decltype(fn), AnyStopToken>>;
static_assert(not std::is_same_v<RetType, std::any>); static_assert(not std::is_same_v<RetType, std::any>);
return AnyOperation<RetType>( // return AnyOperation<RetType>( //
pimpl_->execute( pimpl_->execute(
[fn = std::forward<decltype(fn)>(fn)](auto stopToken) -> std::any { [fn = std::forward<decltype(fn)>(fn)](auto stopToken) mutable -> std::any {
if constexpr (std::is_void_v<RetType>) { if constexpr (std::is_void_v<RetType>) {
fn(std::move(stopToken)); std::invoke(std::forward<decltype(fn)>(fn), std::move(stopToken));
return {}; return {};
} else { } else {
return std::make_any<RetType>(fn(std::move(stopToken))); return std::make_any<RetType>(
std::invoke(std::forward<decltype(fn)>(fn), std::move(stopToken))
);
} }
}, },
std::chrono::duration_cast<std::chrono::milliseconds>(timeout) std::chrono::duration_cast<std::chrono::milliseconds>(timeout)
@@ -141,18 +143,30 @@ public:
[[nodiscard]] auto [[nodiscard]] auto
executeRepeatedly(SomeStdDuration auto interval, SomeHandlerWithoutStopToken auto&& fn) executeRepeatedly(SomeStdDuration auto interval, SomeHandlerWithoutStopToken auto&& fn)
{ {
using RetType = std::decay_t<decltype(fn())>; using RetType = std::decay_t<std::invoke_result_t<decltype(fn)>>;
static_assert(not std::is_same_v<RetType, std::any>); static_assert(not std::is_same_v<RetType, std::any>);
auto const millis = std::chrono::duration_cast<std::chrono::milliseconds>(interval); auto const millis = std::chrono::duration_cast<std::chrono::milliseconds>(interval);
return AnyOperation<RetType>( // return AnyOperation<RetType>( //
pimpl_->executeRepeatedly(millis, [fn = std::forward<decltype(fn)>(fn)] -> std::any { pimpl_->executeRepeatedly(millis, [fn = std::forward<decltype(fn)>(fn)] mutable -> std::any {
fn(); std::invoke(std::forward<decltype(fn)>(fn));
return {}; return {};
}) })
); );
} }
/**
* @brief Schedule an operation on the execution context without expectations of a result
* @note Exceptions are caught internally and `ASSERT`ed on
*
* @param fn The block of code to execute
*/
void
submit(SomeHandlerWithoutStopToken auto&& fn)
{
pimpl_->submit(std::forward<decltype(fn)>(fn));
}
private: private:
struct Concept { struct Concept {
virtual ~Concept() = default; virtual ~Concept() = default;
@@ -165,6 +179,7 @@ private:
[[nodiscard]] virtual impl::ErasedOperation execute(std::function<std::any()>) = 0; [[nodiscard]] virtual impl::ErasedOperation execute(std::function<std::any()>) = 0;
[[nodiscard]] virtual impl::ErasedOperation [[nodiscard]] virtual impl::ErasedOperation
executeRepeatedly(std::chrono::milliseconds, std::function<std::any()>) = 0; executeRepeatedly(std::chrono::milliseconds, std::function<std::any()>) = 0;
virtual void submit(std::function<void()>) = 0;
}; };
template <typename StrandType> template <typename StrandType>
@@ -194,6 +209,12 @@ private:
{ {
return strand.executeRepeatedly(interval, std::move(fn)); return strand.executeRepeatedly(interval, std::move(fn));
} }
void
submit(std::function<void()> fn) override
{
return strand.submit(std::move(fn));
}
}; };
private: private:

13
src/util/async/README.md
View File

@@ -91,11 +91,14 @@ Scheduled operations can be aborted by calling
### Error handling ### Error handling
By default, exceptions that happen during the execution of user-provided code are caught and returned in the error channel of `std::expected` as an instance of the `ExecutionError` struct. The user can then extract the error message by calling `what()` or directly accessing the `message` member. For APIs that return an Operation, by default, exceptions that happen during the execution of user-provided code are caught and returned in the error channel of `std::expected` as an instance of the `ExecutionError` struct. The user can then extract the error message by calling `what()` or directly accessing the `message` member.
In the `submit` API however, exceptions are caught and `ASSERT`ed on.
### Returned value ### Returned value
If the user-provided lambda returns anything but `void`, the type and value will propagate through the operation object and can be received by calling `get` which will block until a value or an error is available. For `submit` API the return type is always `void`.
For other APIs, if the user-provided lambda returns anything but `void`, the type and value will propagate through the operation object and can be received by calling `get` which will block until a value or an error is available.
The `wait` member function can be used when the user just wants to wait for the value to become available but not necessarily getting at the value just yet. The `wait` member function can be used when the user just wants to wait for the value to become available but not necessarily getting at the value just yet.
@@ -122,6 +125,12 @@ This section provides some examples. For more examples take a look at `Execution
### Regular operation ### Regular operation
#### One shot tasks
```cpp
ctx.submit([]() { /* do something */ });
```
#### Awaiting and reading values #### Awaiting and reading values
```cpp ```cpp

43
src/util/async/context/BasicExecutionContext.hpp
View File

@@ -138,7 +138,8 @@ class BasicExecutionContext {
public: public:
/** @brief Whether operations on this execution context are noexcept */ /** @brief Whether operations on this execution context are noexcept */
static constexpr bool kIS_NOEXCEPT = noexcept(ErrorHandlerType::wrap([](auto&) { throw 0; })); static constexpr bool kIS_NOEXCEPT = noexcept(ErrorHandlerType::wrap([](auto&) { throw 0; })) and
noexcept(ErrorHandlerType::catchAndAssert([] { throw 0; }));
using ContextHolderType = ContextType; using ContextHolderType = ContextType;
@@ -209,17 +210,17 @@ public:
delay, std::forward<decltype(fn)>(fn), timeout delay, std::forward<decltype(fn)>(fn), timeout
); );
} else { } else {
using FnRetType = std::decay_t<decltype(fn(std::declval<StopToken>()))>; using FnRetType = std::decay_t<std::invoke_result_t<decltype(fn), StopToken>>;
return ScheduledOperation<FnRetType>( return ScheduledOperation<FnRetType>(
impl::extractAssociatedExecutor(*this), impl::extractAssociatedExecutor(*this),
delay, delay,
[this, timeout, fn = std::forward<decltype(fn)>(fn)](auto) mutable { [this, timeout, fn = std::forward<decltype(fn)>(fn)](auto) mutable {
return this->execute( return this->execute(
[fn = std::forward<decltype(fn)>(fn)](auto stopToken) { [fn = std::forward<decltype(fn)>(fn)](auto stopToken) mutable {
if constexpr (std::is_void_v<FnRetType>) { if constexpr (std::is_void_v<FnRetType>) {
fn(std::move(stopToken)); std::invoke(std::forward<decltype(fn)>(fn), std::move(stopToken));
} else { } else {
return fn(std::move(stopToken)); return std::invoke(std::forward<decltype(fn)>(fn), std::move(stopToken));
} }
}, },
timeout timeout
@@ -249,18 +250,18 @@ public:
delay, std::forward<decltype(fn)>(fn), timeout delay, std::forward<decltype(fn)>(fn), timeout
); );
} else { } else {
using FnRetType = std::decay_t<decltype(fn(std::declval<StopToken>(), true))>; using FnRetType = std::decay_t<std::invoke_result_t<decltype(fn), StopToken, bool>>;
return ScheduledOperation<FnRetType>( return ScheduledOperation<FnRetType>(
impl::extractAssociatedExecutor(*this), impl::extractAssociatedExecutor(*this),
delay, delay,
[this, timeout, fn = std::forward<decltype(fn)>(fn)](auto ec) mutable { [this, timeout, fn = std::forward<decltype(fn)>(fn)](auto ec) mutable {
return this->execute( return this->execute(
[fn = std::forward<decltype(fn)>(fn), [fn = std::forward<decltype(fn)>(fn),
isAborted = (ec == boost::asio::error::operation_aborted)](auto stopToken) { isAborted = (ec == boost::asio::error::operation_aborted)](auto stopToken) mutable {
if constexpr (std::is_void_v<FnRetType>) { if constexpr (std::is_void_v<FnRetType>) {
fn(std::move(stopToken), isAborted); std::invoke(std::forward<decltype(fn)>(fn), std::move(stopToken), isAborted);
} else { } else {
return fn(std::move(stopToken), isAborted); return std::invoke(std::forward<decltype(fn)>(fn), std::move(stopToken), isAborted);
} }
}, },
timeout timeout
@@ -310,12 +311,12 @@ public:
[[maybe_unused]] auto timeoutHandler = [[maybe_unused]] auto timeoutHandler =
impl::getTimeoutHandleIfNeeded(TimerContextProvider::getContext(*this), timeout, stopSource); impl::getTimeoutHandleIfNeeded(TimerContextProvider::getContext(*this), timeout, stopSource);
using FnRetType = std::decay_t<decltype(fn(std::declval<StopToken>()))>; using FnRetType = std::decay_t<std::invoke_result_t<decltype(fn), StopToken>>;
if constexpr (std::is_void_v<FnRetType>) { if constexpr (std::is_void_v<FnRetType>) {
fn(std::move(stopToken)); std::invoke(std::forward<decltype(fn)>(fn), std::move(stopToken));
outcome.setValue(); outcome.setValue();
} else { } else {
outcome.setValue(fn(std::move(stopToken))); outcome.setValue(std::invoke(std::forward<decltype(fn)>(fn), std::move(stopToken)));
} }
}) })
); );
@@ -350,17 +351,29 @@ public:
context_, context_,
impl::outcomeForHandler<StopSourceType>(fn), impl::outcomeForHandler<StopSourceType>(fn),
ErrorHandlerType::wrap([fn = std::forward<decltype(fn)>(fn)](auto& outcome) mutable { ErrorHandlerType::wrap([fn = std::forward<decltype(fn)>(fn)](auto& outcome) mutable {
using FnRetType = std::decay_t<decltype(fn())>; using FnRetType = std::decay_t<std::invoke_result_t<decltype(fn)>>;
if constexpr (std::is_void_v<FnRetType>) { if constexpr (std::is_void_v<FnRetType>) {
fn(); std::invoke(std::forward<decltype(fn)>(fn));
outcome.setValue(); outcome.setValue();
} else { } else {
outcome.setValue(fn()); outcome.setValue(std::invoke(std::forward<decltype(fn)>(fn)));
} }
}) })
); );
} }
/**
* @brief Schedule an operation on the execution context without expectations of a result
* @note Exceptions are caught internally and `ASSERT`ed on
*
* @param fn The block of code to execute
*/
void
submit(SomeHandlerWithoutStopToken auto&& fn) noexcept(kIS_NOEXCEPT)
{
DispatcherType::post(context_, ErrorHandlerType::catchAndAssert(fn));
}
/** /**
* @brief Create a strand for this execution context * @brief Create a strand for this execution context
* *

53
src/util/async/context/impl/Execution.hpp
View File

@@ -30,68 +30,89 @@
namespace util::async::impl { namespace util::async::impl {
struct SpawnDispatchStrategy { struct SpawnDispatchStrategy {
template <typename ContextType, SomeOutcome OutcomeType> template <typename ContextType, SomeOutcome OutcomeType, typename FnType>
[[nodiscard]] static auto [[nodiscard]] static auto
dispatch(ContextType& ctx, OutcomeType&& outcome, auto&& fn) dispatch(ContextType& ctx, OutcomeType&& outcome, FnType&& fn)
{ {
auto op = outcome.getOperation(); auto op = outcome.getOperation();
util::spawn( util::spawn(
ctx.getExecutor(), ctx.getExecutor(),
[outcome = std::forward<decltype(outcome)>(outcome), [outcome = std::forward<OutcomeType>(outcome), fn = std::forward<FnType>(fn)](auto yield) mutable {
fn = std::forward<decltype(fn)>(fn)](auto yield) mutable {
if constexpr (SomeStoppableOutcome<OutcomeType>) { if constexpr (SomeStoppableOutcome<OutcomeType>) {
auto& stopSource = outcome.getStopSource(); auto& stopSource = outcome.getStopSource();
fn(outcome, stopSource, stopSource[yield]); std::invoke(std::forward<decltype(fn)>(fn), outcome, stopSource, stopSource[yield]);
} else { } else {
fn(outcome); std::invoke(std::forward<decltype(fn)>(fn), outcome);
} }
} }
); );
return op; return op;
} }
template <typename ContextType, typename FnType>
static void
post(ContextType& ctx, FnType&& fn)
{
util::spawn(ctx.getExecutor(), [fn = std::forward<FnType>(fn)](auto) mutable {
std::invoke(std::forward<decltype(fn)>(fn));
});
}
}; };
struct PostDispatchStrategy { struct PostDispatchStrategy {
template <typename ContextType, SomeOutcome OutcomeType> template <typename ContextType, SomeOutcome OutcomeType, typename FnType>
[[nodiscard]] static auto [[nodiscard]] static auto
dispatch(ContextType& ctx, OutcomeType&& outcome, auto&& fn) dispatch(ContextType& ctx, OutcomeType&& outcome, FnType&& fn)
{ {
auto op = outcome.getOperation(); auto op = outcome.getOperation();
boost::asio::post( boost::asio::post(
ctx.getExecutor(), ctx.getExecutor(), [outcome = std::forward<OutcomeType>(outcome), fn = std::forward<FnType>(fn)]() mutable {
[outcome = std::forward<decltype(outcome)>(outcome), fn = std::forward<decltype(fn)>(fn)]() mutable {
if constexpr (SomeStoppableOutcome<OutcomeType>) { if constexpr (SomeStoppableOutcome<OutcomeType>) {
auto& stopSource = outcome.getStopSource(); auto& stopSource = outcome.getStopSource();
fn(outcome, stopSource, stopSource.getToken()); std::invoke(std::forward<decltype(fn)>(fn), outcome, stopSource, stopSource.getToken());
} else { } else {
fn(outcome); std::invoke(std::forward<decltype(fn)>(fn), outcome);
} }
} }
); );
return op; return op;
} }
template <typename ContextType, typename FnType>
static void
post(ContextType& ctx, FnType&& fn)
{
boost::asio::post(ctx.getExecutor(), std::forward<FnType>(fn));
}
}; };
struct SyncDispatchStrategy { struct SyncDispatchStrategy {
template <typename ContextType, SomeOutcome OutcomeType> template <typename ContextType, SomeOutcome OutcomeType, typename FnType>
[[nodiscard]] static auto [[nodiscard]] static auto
dispatch([[maybe_unused]] ContextType& ctx, OutcomeType outcome, auto&& fn) dispatch([[maybe_unused]] ContextType& ctx, OutcomeType outcome, FnType&& fn)
{ {
auto op = outcome.getOperation(); auto op = outcome.getOperation();
if constexpr (SomeStoppableOutcome<OutcomeType>) { if constexpr (SomeStoppableOutcome<OutcomeType>) {
auto& stopSource = outcome.getStopSource(); auto& stopSource = outcome.getStopSource();
fn(outcome, stopSource, stopSource.getToken()); std::invoke(std::forward<FnType>(fn), outcome, stopSource, stopSource.getToken());
} else { } else {
fn(outcome); std::invoke(std::forward<FnType>(fn), outcome);
} }
return op; return op;
} }
template <typename ContextType, typename FnType>
static void
post([[maybe_unused]] ContextType& ctx, FnType&& fn)
{
std::invoke(std::forward<FnType>(fn));
}
}; };
} // namespace util::async::impl } // namespace util::async::impl

18
src/util/async/context/impl/Strand.hpp
View File

@@ -81,12 +81,12 @@ public:
TimerContextProvider::getContext(parentContext_.get()), timeout, stopSource TimerContextProvider::getContext(parentContext_.get()), timeout, stopSource
); );
using FnRetType = std::decay_t<decltype(fn(std::declval<StopToken>()))>; using FnRetType = std::decay_t<std::invoke_result_t<decltype(fn), StopToken>>;
if constexpr (std::is_void_v<FnRetType>) { if constexpr (std::is_void_v<FnRetType>) {
fn(std::move(stopToken)); std::invoke(std::forward<decltype(fn)>(fn), std::move(stopToken));
outcome.setValue(); outcome.setValue();
} else { } else {
outcome.setValue(fn(std::move(stopToken))); outcome.setValue(std::invoke(std::forward<decltype(fn)>(fn), std::move(stopToken)));
} }
}) })
); );
@@ -108,12 +108,12 @@ public:
context_, context_,
impl::outcomeForHandler<StopSourceType>(fn), impl::outcomeForHandler<StopSourceType>(fn),
ErrorHandlerType::wrap([fn = std::forward<decltype(fn)>(fn)](auto& outcome) mutable { ErrorHandlerType::wrap([fn = std::forward<decltype(fn)>(fn)](auto& outcome) mutable {
using FnRetType = std::decay_t<decltype(fn())>; using FnRetType = std::decay_t<std::invoke_result_t<decltype(fn)>>;
if constexpr (std::is_void_v<FnRetType>) { if constexpr (std::is_void_v<FnRetType>) {
fn(); std::invoke(std::forward<decltype(fn)>(fn));
outcome.setValue(); outcome.setValue();
} else { } else {
outcome.setValue(fn()); outcome.setValue(std::invoke(std::forward<decltype(fn)>(fn)));
} }
}) })
); );
@@ -128,6 +128,12 @@ public:
return RepeatedOperation(impl::extractAssociatedExecutor(*this), interval, std::forward<decltype(fn)>(fn)); return RepeatedOperation(impl::extractAssociatedExecutor(*this), interval, std::forward<decltype(fn)>(fn));
} }
} }
void
submit(SomeHandlerWithoutStopToken auto&& fn) noexcept(kIS_NOEXCEPT)
{
DispatcherType::post(context_, ErrorHandlerType::catchAndAssert(fn));
}
}; };
} // namespace util::async::impl } // namespace util::async::impl

5
src/util/async/context/impl/Utils.hpp
View File

@@ -29,6 +29,7 @@
#include <expected> #include <expected>
#include <optional> #include <optional>
#include <type_traits>
namespace util::async::impl { namespace util::async::impl {
@@ -61,12 +62,12 @@ template <SomeStopSource StopSourceType>
outcomeForHandler(auto&& fn) outcomeForHandler(auto&& fn)
{ {
if constexpr (SomeHandlerWith<decltype(fn), typename StopSourceType::Token>) { if constexpr (SomeHandlerWith<decltype(fn), typename StopSourceType::Token>) {
using FnRetType = decltype(fn(std::declval<typename StopSourceType::Token>())); using FnRetType = std::decay_t<std::invoke_result_t<decltype(fn), typename StopSourceType::Token>>;
using RetType = std::expected<FnRetType, ExecutionError>; using RetType = std::expected<FnRetType, ExecutionError>;
return StoppableOutcome<RetType, StopSourceType>(); return StoppableOutcome<RetType, StopSourceType>();
} else { } else {
using FnRetType = decltype(fn()); using FnRetType = std::decay_t<std::invoke_result_t<decltype(fn)>>;
using RetType = std::expected<FnRetType, ExecutionError>; using RetType = std::expected<FnRetType, ExecutionError>;
return Outcome<RetType>(); return Outcome<RetType>();

23
src/util/async/impl/ErrorHandling.hpp
View File

@@ -19,6 +19,7 @@
#pragma once #pragma once
#include "util/Assert.hpp"
#include "util/async/Concepts.hpp" #include "util/async/Concepts.hpp"
#include "util/async/Error.hpp" #include "util/async/Error.hpp"
@@ -38,7 +39,7 @@ struct DefaultErrorHandler {
return return
[fn = std::forward<decltype(fn)>(fn)]<typename... Args>(SomeOutcome auto& outcome, Args&&... args) mutable { [fn = std::forward<decltype(fn)>(fn)]<typename... Args>(SomeOutcome auto& outcome, Args&&... args) mutable {
try { try {
fn(outcome, std::forward<Args>(args)...); std::invoke(std::forward<decltype(fn)>(fn), outcome, std::forward<Args>(args)...);
} catch (std::exception const& e) { } catch (std::exception const& e) {
outcome.setValue( outcome.setValue(
std::unexpected(ExecutionError{fmt::format("{}", std::this_thread::get_id()), e.what()}) std::unexpected(ExecutionError{fmt::format("{}", std::this_thread::get_id()), e.what()})
@@ -50,6 +51,20 @@ struct DefaultErrorHandler {
} }
}; };
} }
[[nodiscard]] static auto
catchAndAssert(auto&& fn) noexcept // note this is a lie when used with MockAssert (use MockAssertNoThrow)
{
return [fn = std::forward<decltype(fn)>(fn)] mutable {
try {
std::invoke(std::forward<decltype(fn)>(fn));
} catch (std::exception const& e) {
ASSERT(false, "Exception caught: {}", e.what());
} catch (...) {
ASSERT(false, "Unknown exception caught");
}
};
}
}; };
struct NoErrorHandler { struct NoErrorHandler {
@@ -58,6 +73,12 @@ struct NoErrorHandler {
{ {
return std::forward<decltype(fn)>(fn); return std::forward<decltype(fn)>(fn);
} }
[[nodiscard]] static constexpr auto
catchAndAssert(auto&& fn)
{
return std::forward<decltype(fn)>(fn);
}
}; };
} // namespace util::async::impl } // namespace util::async::impl

2
tests/common/data/cassandra/FakesAndMocks.hpp
View File

@@ -124,6 +124,6 @@ struct FakeRetryPolicy {
void void
retry(Fn&& fn) retry(Fn&& fn)
{ {
fn(); std::invoke(std::forward<decltype(fn)>(fn));
} }
}; };

57
tests/common/util/MockAssert.hpp
View File

@@ -28,7 +28,6 @@
#include <string_view> #include <string_view>
namespace common::util { namespace common::util {
class WithMockAssert : virtual public testing::Test { class WithMockAssert : virtual public testing::Test {
public: public:
struct MockAssertException { struct MockAssertException {
@@ -48,30 +47,42 @@ public:
~WithMockAssertNoThrow() override; ~WithMockAssertNoThrow() override;
}; };
namespace impl {
template <typename T>
struct MockGuard {
T mock;
~MockGuard()
{
::util::impl::OnAssert::resetAction();
}
};
} // namespace impl
} // namespace common::util } // namespace common::util
#define EXPECT_CLIO_ASSERT_FAIL_WITH_MESSAGE(statement, message_regex) \ #define EXPECT_CLIO_ASSERT_FAIL_WITH_MESSAGE(statement, message_regex) \
if (dynamic_cast<common::util::WithMockAssert*>(this) != nullptr) { \ if (dynamic_cast<common::util::WithMockAssert*>(this) != nullptr) { \
EXPECT_THROW( \ EXPECT_THROW( \
{ \ { \
try { \ try { \
statement; \ statement; \
} catch (common::util::WithMockAssert::MockAssertException const& e) { \ } catch (common::util::WithMockAssert::MockAssertException const& e) { \
EXPECT_THAT(e.message, testing::ContainsRegex(message_regex)); \ EXPECT_THAT(e.message, testing::ContainsRegex(message_regex)); \
throw; \ throw; \
} \ } \
}, \ }, \
common::util::WithMockAssert::MockAssertException \ common::util::WithMockAssert::MockAssertException \
); \ ); \
} else if (dynamic_cast<common::util::WithMockAssertNoThrow*>(this) != nullptr) { \ } else if (dynamic_cast<common::util::WithMockAssertNoThrow*>(this) != nullptr) { \
testing::StrictMock<testing::MockFunction<void(std::string_view)>> callMock; \ using MockGuardType = \
::util::impl::OnAssert::setAction([&callMock](std::string_view m) { callMock.Call(m); }); \ common::util::impl::MockGuard<testing::StrictMock<testing::MockFunction<void(std::string_view)>>>; \
EXPECT_CALL(callMock, Call(testing::ContainsRegex(message_regex))); \ auto mockGuard = std::make_shared<MockGuardType>(); \
statement; \ ::util::impl::OnAssert::setAction([mockGuard](std::string_view m) { mockGuard->mock.Call(m); }); \
::util::impl::OnAssert::resetAction(); \ EXPECT_CALL(mockGuard->mock, Call(testing::ContainsRegex(message_regex))); \
} else { \ statement; \
std::cerr << "EXPECT_CLIO_ASSERT_FAIL_WITH_MESSAGE() can be used only inside test body" << std::endl; \ } else { \
std::terminate(); \ std::cerr << "EXPECT_CLIO_ASSERT_FAIL_WITH_MESSAGE() can be used only inside test body" << std::endl; \
std::terminate(); \
} }
#define EXPECT_CLIO_ASSERT_FAIL(statement) EXPECT_CLIO_ASSERT_FAIL_WITH_MESSAGE(statement, ".*") #define EXPECT_CLIO_ASSERT_FAIL(statement) EXPECT_CLIO_ASSERT_FAIL_WITH_MESSAGE(statement, ".*")

1
tests/common/util/MockExecutionContext.hpp
View File

@@ -84,6 +84,7 @@ struct MockExecutionContext {
(std::chrono::milliseconds, std::function<std::any()>), (std::chrono::milliseconds, std::function<std::any()>),
() ()
); );
MOCK_METHOD(void, submit, (std::function<void()>), ());
MOCK_METHOD(MockStrand const&, makeStrand, (), ()); MOCK_METHOD(MockStrand const&, makeStrand, (), ());
MOCK_METHOD(void, stop, (), (const)); MOCK_METHOD(void, stop, (), (const));

1
tests/common/util/MockStrand.hpp
View File

@@ -69,4 +69,5 @@ struct MockStrand {
(std::chrono::milliseconds, std::function<std::any()>), (std::chrono::milliseconds, std::function<std::any()>),
(const) (const)
); );
MOCK_METHOD(void, submit, (std::function<void()>), (const));
}; };

59
tests/unit/util/async/AsyncExecutionContextTests.cpp
View File

@@ -17,6 +17,7 @@
*/ */
//============================================================================== //==============================================================================
#include "util/MockAssert.hpp"
#include "util/Profiler.hpp" #include "util/Profiler.hpp"
#include "util/async/Operation.hpp" #include "util/async/Operation.hpp"
#include "util/async/context/BasicExecutionContext.hpp" #include "util/async/context/BasicExecutionContext.hpp"
@@ -32,12 +33,13 @@
#include <stdexcept> #include <stdexcept>
#include <string> #include <string>
#include <thread> #include <thread>
#include <vector>
using namespace util::async; using namespace util::async;
using ::testing::Types; using ::testing::Types;
template <typename T> template <typename T>
struct ExecutionContextTests : public ::testing::Test { struct ExecutionContextTests : common::util::WithMockAssertNoThrow {
using ExecutionContextType = T; using ExecutionContextType = T;
ExecutionContextType ctx{2}; ExecutionContextType ctx{2};
@@ -238,6 +240,32 @@ TYPED_TEST(ExecutionContextTests, repeatingOperationForceInvoke)
EXPECT_EQ(callCount, 0uz); EXPECT_EQ(callCount, 0uz);
} }
TYPED_TEST(ExecutionContextTests, submit)
{
EXPECT_CLIO_ASSERT_FAIL(this->ctx.submit([] -> void { throw 0; }));
std::atomic_uint32_t count = 0;
std::binary_semaphore sem{0};
static constexpr auto kNUM_SUBMISSIONS = 1024;
for (auto i = 1; i <= kNUM_SUBMISSIONS; ++i) {
if (i == kNUM_SUBMISSIONS) {
this->ctx.submit([&count, &sem] {
++count;
sem.release();
});
} else {
this->ctx.submit([&count] { ++count; });
}
}
sem.acquire();
// order is not guaranteed (see `strandSubmit` below)
ASSERT_EQ(count, static_cast<size_t>(kNUM_SUBMISSIONS));
}
TYPED_TEST(ExecutionContextTests, strandMove) TYPED_TEST(ExecutionContextTests, strandMove)
{ {
auto strand = this->ctx.makeStrand(); auto strand = this->ctx.makeStrand();
@@ -328,6 +356,35 @@ TYPED_TEST(ExecutionContextTests, strandedRepeatingOperationForceInvoke)
EXPECT_EQ(callCount, 0uz); EXPECT_EQ(callCount, 0uz);
} }
TYPED_TEST(ExecutionContextTests, strandSubmit)
{
auto strand = this->ctx.makeStrand();
EXPECT_CLIO_ASSERT_FAIL(strand.submit([] -> void { throw 0; }));
std::vector<int> results;
std::binary_semaphore sem{0};
static constexpr auto kNUM_SUBMISSIONS = 1024;
for (auto i = 1; i <= kNUM_SUBMISSIONS; ++i) {
if (i == kNUM_SUBMISSIONS) {
strand.submit([&results, &sem, i] {
results.push_back(i);
sem.release();
});
} else {
strand.submit([&results, i] { results.push_back(i); });
}
}
sem.acquire();
ASSERT_EQ(results.size(), static_cast<size_t>(kNUM_SUBMISSIONS));
for (int i = 0; i < kNUM_SUBMISSIONS; ++i) {
EXPECT_EQ(results[i], i + 1);
}
}
TYPED_TEST(AsyncExecutionContextTests, executeAutoAborts) TYPED_TEST(AsyncExecutionContextTests, executeAutoAborts)
{ {
auto value = 0; auto value = 0;