Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
120 changes: 114 additions & 6 deletions include/boost/corosio/detail/dispatch_coro.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -14,33 +14,141 @@
#include <boost/corosio/io_context.hpp>
#include <boost/capy/continuation.hpp>
#include <boost/capy/ex/executor_ref.hpp>
#include <boost/capy/ex/any_executor.hpp>
#include <boost/capy/ex/frame_alloc_mixin.hpp>
#include <boost/capy/detail/type_id.hpp>
#include <coroutine>
#include <typeinfo>

namespace boost::corosio::detail {

/** Trampoline frame that lends an executor a stable continuation.

An executor's queue holds a dispatched continuation by reference
until dequeue (its stable-address contract) — storage a
completion about to recycle, reuse, or delete its op cannot
provide. The trampoline's frame owns the continuation for exactly
the queue-residency window; when the executor resumes it, it
resumes the real handle and self-destroys.

run_async cannot serve here: its launch wrapper calls
`ex.on_work_finished()` after the task completes, but a
completion's `executor_ref` points into the awaiting coroutine's
environment, which the resume destroys — completions must never
touch the executor after resuming the handle. This trampoline
touches nothing after the resume (same accounting as the plain
`ex.dispatch` it replaces).
*/
struct resume_trampoline
{
struct promise_type : capy::frame_alloc_mixin
{
capy::continuation cont;

resume_trampoline get_return_object()
{
return {
std::coroutine_handle<promise_type>::from_promise(*this)};
}
std::suspend_always initial_suspend() noexcept
{
return {};
}
std::suspend_never final_suspend() noexcept
{
return {};
}
void return_void() noexcept {}
void unhandled_exception()
{
std::terminate();
}
};

std::coroutine_handle<promise_type> h;
};

inline resume_trampoline
make_resume_trampoline(std::coroutine_handle<> target)
{
// A nested resume, not a symmetric transfer that destroys this
// frame inside await_suspend: that idiom is legal but miscompiled
// by cl 19.44 and Apple Clang 16 (access violations on exactly
// this path). The cost is one stack frame per executor hop, the
// only path that reaches here.
target.resume();
co_return;
}

/** Dispatch a handle to an executor without lending it caller storage.

Follows the executor's dispatch semantics: the resume may run
inline or be queued, at the executor's discretion.

Allocates one trampoline frame (capy's recycling frame pool);
intended for the cold cross-executor completion path only. An
allocation failure propagates out of the (noexcept-adjacent)
completion path and terminates, matching the initiation paths'
OOM policy. The caller may dispose its op before calling.

@param ex The executor the resume must run on.
@param h The coroutine to resume there.
*/
inline void
dispatch_resume(capy::executor_ref ex, std::coroutine_handle<> h)
{
auto tramp = make_resume_trampoline(h);
tramp.h.promise().cont.h = tramp.h;
ex.dispatch(tramp.h.promise().cont).resume();
}

/** Returns a handle for symmetric transfer on I/O completion.

If the executor is io_context::executor_type, returns `c.h`
directly (fast path). Otherwise dispatches through the
executor, which returns `c.h` or `noop_coroutine()`.
directly (fast path). Otherwise the handle is dispatched through
the executor via a trampoline frame and `noop_coroutine()` is
returned.

Callers in coroutine machinery should return the result
for symmetric transfer. Callers at the scheduler pump
level should call `.resume()` on the result.

@p c is NOT retained past this call: completion ops embed their
continuation in storage that is recycled, reused, or freed the
moment the completion is consumed, so handing a deferring
executor that storage lets a concurrent operation rewrite a
queued node. The deferring branch therefore lends the executor a
frame-owned continuation instead — see `dispatch_resume` — at the
cost of one pooled frame on that cold path.

@param ex The executor to dispatch through.
@param c The continuation to dispatch. Must remain at a
stable address until dequeued by the executor.
@param c Carries the handle to resume; not retained.

@return A handle for symmetric transfer or `std::noop_coroutine()`.
*/
/// Check whether @p ex is the io_context executor, directly or
/// type-erased through `capy::any_executor` (run_async callers and
/// tcp_server erase it that way). The held TYPE is enough:
/// completions only run on the run thread, where
/// `io_context::executor_type::dispatch` returns `c.h` inline —
/// exactly what the fast path reproduces.
inline bool
is_io_context_executor(capy::executor_ref ex) noexcept
{
if (ex.target<io_context::executor_type>() != nullptr)
return true;
if (auto* any = ex.target<capy::any_executor>())
return any->target_type() == typeid(io_context::executor_type);
return false;
}

inline std::coroutine_handle<>
dispatch_coro(capy::executor_ref ex, capy::continuation& c)
{
if (ex.target<io_context::executor_type>() != nullptr)
if (is_io_context_executor(ex))
return c.h;
return ex.dispatch(c);
dispatch_resume(ex, c.h);
return std::noop_coroutine();
}

} // namespace boost::corosio::detail
Expand Down
38 changes: 36 additions & 2 deletions include/boost/corosio/detail/intrusive.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,18 @@ class intrusive_list
return head_;
}

void push_front(T* w) noexcept
{
auto* n = static_cast<node*>(w);
n->prev_ = nullptr;
n->next_ = head_;
if (head_)
static_cast<node*>(head_)->prev_ = w;
else
tail_ = w;
head_ = w;
}

void push_back(T* w) noexcept
{
auto* n = static_cast<node*>(w);
Expand Down Expand Up @@ -118,12 +130,33 @@ class intrusive_list
return w;
}

void remove(T* w) noexcept
/** Unlink @p w if it is a member of this list.

@pre @p w is either a genuine member of `*this` list, or is
fully detached (not linked in *any* `intrusive_list<T>`,
including a different instance over the same `T`). This is
not a membership test: `T::node` holds one shared `next_`/
`prev_` pair regardless of which list currently owns it, so a
node actually linked in a different list looks "not detached"
here and would be spliced out using *this* list's `head_`/
`tail_` for the boundary cases — corrupting both lists. A
caller that cannot rule that out needs its own bookkeeping
(e.g. a flag for which list currently owns the node) before
calling this.

@return True if @p w was linked in and has now been removed;
false if it was already detached (e.g. a reentrant call from
a self-referential keepalive dropping during the same
teardown that already popped it elsewhere). Callers that gate
a one-time action (recycling, deleting) on actually having
removed something rely on this to make that action idempotent.
*/
bool remove(T* w) noexcept
{
auto* n = static_cast<node*>(w);
// Already detached — nothing to do.
if (!n->next_ && !n->prev_ && head_ != w && tail_ != w)
return;
return false;
if (n->prev_)
static_cast<node*>(n->prev_)->next_ = n->next_;
else
Expand All @@ -134,6 +167,7 @@ class intrusive_list
tail_ = n->prev_;
n->next_ = nullptr;
n->prev_ = nullptr;
return true;
}

/// Invoke @p f for each element in the list.
Expand Down
Loading
Loading