VLink  2.0.0
A high-performance communication middleware
coroutine.h
浏览该文件的文档.
1 /*
2  * Copyright (C) 2026 by Thun Lu. All rights reserved.
3  * Author: Thun Lu <thun.lu@zohomail.cn>
4  * Repo: https://github.com/thun-res/vlink
5  * _ __ __ _ __
6  * | | / / / / (_) ____ / /__
7  * | | / / / / / / / __ \ / //_/
8  * | |/ / / /___ / / / / / / / ,<
9  * |___/ /_____/ /_/ /_/ /_/ /_/|_|
10  *
11  * Licensed under the Apache License, Version 2.0 (the "License");
12  * you may not use this file except in compliance with the License.
13  * You may obtain a copy of the License at
14  *
15  * http://www.apache.org/licenses/LICENSE-2.0
16  *
17  * Unless required by applicable law or agreed to in writing, software
18  * distributed under the License is distributed on an "AS IS" BASIS,
19  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
20  * See the License for the specific language governing permissions and
21  * limitations under the License.
22  */
23 
24 /**
25  * @file coroutine.h
26  * @brief Stackless C++20 coroutine layer that resumes work onto @c MessageLoop threads.
27  *
28  * @details
29  * Adds an additive coroutine binding atop @c MessageLoop, @c GraphTask and @c Schedule. Every
30  * awaiter eventually re-enters the target loop through @c MessageLoop::post_task or
31  * @c MessageLoop::exec_task; the loop itself is unchanged. The layer compiles in only when
32  * @c VLINK_ENABLE_COROUTINE is defined (auto-enabled when @c VLINK_ENABLE_CXX_STD_20 is set and
33  * the toolchain advertises @c __cpp_impl_coroutine and @c __cpp_lib_coroutine); otherwise the
34  * file is empty. The canonical namespace is @c vlink::Coroutine with @c vlink::Co as a short
35  * alias.
36  *
37  * @par State machine of a coroutine resume
38  *
39  * @verbatim
40  * +----------+ co_spawn / +-----------+ awaiter +-------------+
41  * | created | -- co_await ---> | suspended | -- ready -> | resumed on |
42  * +----------+ +-----------+ | target loop |
43  * | ^ +-------------+
44  * | | |
45  * | queue full (kRetry) | | co_return
46  * | +--------------------------+ v
47  * | | +-------------+
48  * | | helper thread retries every ~1ms | finished |
49  * v v +-------------+
50  * +-----------+ loop closed (kClosed)
51  * | failed | --> await_resume throws runtime_error / OperationCancelled
52  * +-----------+
53  * @endverbatim
54  *
55  * @par Awaitable surface
56  *
57  * | Category | Symbol | Purpose |
58  * | ------------- | --------------------------------- | ------------------------------------ |
59  * | Core type | @c Task<TypeT> | Lazily started awaitable |
60  * | Awaiter | @c schedule(loop, prio) | Hop onto @p loop 's thread |
61  * | Awaiter | @c yield(loop, prio) | Re-post to the back of the queue |
62  * | Awaiter | @c delay_ms(loop, ms, prio) | Non-blocking sleep |
63  * | Awaiter | @c await_future(loop, fut) | Wait on @c std::future |
64  * | Entry point | @c co_spawn(loop, task) | Fire-and-forget @c Task<void> |
65  * | Entry point | @c co_spawn(loop, task, on_done) | Spawn with completion callback |
66  * | Entry point | @c co_spawn_with_priority(...) | Same overloads with priority |
67  * | Bridge | @c exec(loop, config, fn) | Wrap @c MessageLoop::exec_task |
68  * | Bridge | @c await_graph(loop, graph) | Wait on @c GraphTask DAG |
69  * | Orchestration | @c when_all(loop, tasks) | Join all, collect results |
70  * | Orchestration | @c when_any(loop, tasks) | First success and its index |
71  * | Orchestration | @c sequence(loop, tasks) | Run in order |
72  *
73  * @par Frame allocation
74  * Coroutine frames flow through @c vlink::MemoryPool::global_instance() via the
75  * @c operator @c new / @c operator @c delete overloads on @c TaskPromise and
76  * @c DetachedTask::promise_type. The allocator throws @c std::bad_alloc on pool exhaustion;
77  * there is no custom-allocator template parameter.
78  *
79  * @par Lock order
80  * @c MessageLoop::AliveState::mtx -> @c MessageLoop::Impl::mtx -> @c TaskHandle::State::mtx ->
81  * awaiter shared-state mutex. Each awaiter owns its own state object that holds the suspended
82  * coroutine handle. When the target loop closes mid-resume the handle is released on the
83  * @c FutureWaitLoop helper thread via a queued retry; callers must therefore not assume
84  * @c await_resume runs on the original loop.
85  *
86  * @par Cancellation contract
87  * - @c await_future rethrows the promise's stored exception via @c future.get(). When the
88  * target loop closes before the ready future can be posted back, @c await_resume throws
89  * @c vlink::Exception::OperationCancelled.
90  * - @c await_graph normally returns @c void; on the late-closure path it throws
91  * @c vlink::Exception::OperationCancelled.
92  * - @c schedule / @c yield / @c delay_ms retry posting through the @c FutureWaitLoop helper on
93  * @c kRetry; once @c detail::kMaxResumePostRetry (@c 30000 ticks of approximately one
94  * millisecond each) is exhausted, or the loop reports @c kClosed, @c await_resume throws
95  * @c std::runtime_error with a descriptive message.
96  *
97  * @par FutureWaitLoop
98  * A process-wide helper thread polls every registered awaiter at a roughly one-millisecond
99  * cadence. Concurrent use of @c await_future shares this thread; latency is bounded by the
100  * cadence plus the target loop's task dispatch delay.
101  *
102  * @par Priority remap on full-queue loops
103  * On @c kPriorityType loops the post helper remaps @c MessageLoop::kNoPriority to
104  * @c MessageLoop::kNormalPriority so default-priority resumes do not sink to the queue tail.
105  *
106  * @par Alive-state mutex hold time
107  * @c post_callback_if_alive acquires @c MessageLoop::AliveState::mtx only briefly; the inner
108  * @c post_callback path uses @c TaskOverflowPolicy::kReject so the alive-state mutex is never
109  * held across a sleep or backoff.
110  *
111  * @par Example
112  * @code
113  * vlink::MessageLoop loop;
114  * loop.async_run();
115  *
116  * vlink::Co::Task<> my_routine(vlink::MessageLoop& loop, int* counter) {
117  * co_await vlink::Co::delay_ms(loop, 100);
118  * ++(*counter);
119  * co_await vlink::Co::yield(loop);
120  * ++(*counter);
121  * }
122  *
123  * int counter = 0;
124  * vlink::Co::co_spawn(loop, my_routine(loop, &counter));
125  * @endcode
126  *
127  * @warning Passing a coroutine lambda as a temporary, for example
128  * @code
129  * co_spawn(loop, []() -> Task<> { co_await ... ; }());
130  * @endcode
131  * destroys the lambda object at the end of the full-expression -- long before the
132  * coroutine body runs. Any captured state (including @c [&] and @c [=]) becomes a
133  * dangling reference. Always thread captures through function parameters so the
134  * compiler copies them into the coroutine frame.
135  */
136 
137 #pragma once
138 
139 #include "../version.h"
140 
141 #ifdef VLINK_ENABLE_CXX_STD_20
142 #if __has_include(<coroutine>)
143 #include <coroutine>
144 #endif
145 #if defined(__cpp_lib_coroutine)
146 #ifndef VLINK_ENABLE_COROUTINE
147 #define VLINK_ENABLE_COROUTINE
148 #endif
149 #endif
150 #endif
151 
152 #ifdef VLINK_ENABLE_COROUTINE
153 
154 #include <atomic>
155 #include <chrono>
156 #include <coroutine>
157 #include <cstdint>
158 #include <exception>
159 #include <future>
160 #include <memory>
161 #include <mutex>
162 #include <optional>
163 #include <stdexcept>
164 #include <type_traits>
165 #include <utility>
166 #include <vector>
167 
168 #include "./exception.h"
169 #include "./functional.h"
170 #include "./graph_task.h"
171 #include "./macros.h"
172 #include "./memory_resource.h"
173 #include "./message_loop.h"
174 #include "./schedule.h"
175 
176 namespace vlink {
177 
178 /**
179  * @namespace vlink::Coroutine
180  * @brief Container namespace for the VLink coroutine integration layer.
181  *
182  * @details
183  * Holds the @c Task template, every awaiter, the @c co_spawn entry points and the orchestrators
184  * (@c when_all / @c when_any / @c sequence). Aliased as @c vlink::Co.
185  */
186 namespace Coroutine { // NOLINT(readability-identifier-naming)
187 
188 template <typename TypeT = void>
189 class Task;
190 
191 /**
192  * @namespace vlink::Coroutine::detail
193  * @brief Internal helpers for the coroutine layer.
194  *
195  * @details
196  * Hosts promise types, the detached-task adapter, the @c ResumePostResult enum, the awaiter
197  * shared state and the helpers that bridge coroutine handles back onto @c MessageLoop. Not part
198  * of the public API surface.
199  */
200 namespace detail {
201 
202 template <typename TypeT>
203 struct TaskPromise;
204 
205 /**
206  * @brief Allocates a coroutine frame through the process-wide @c MemoryPool.
207  *
208  * @param size Frame size requested by the compiler.
209  * @return Pointer to the freshly allocated frame.
210  * @throws std::bad_alloc when the underlying pool is exhausted.
211  */
212 VLINK_EXPORT void* allocate_frame(size_t size);
213 
214 /**
215  * @brief Returns a coroutine frame to the process-wide @c MemoryPool.
216  *
217  * @param ptr Frame previously obtained from @c allocate_frame. @c nullptr is a no-op.
218  * @param size Original frame size; must match the @c allocate_frame call.
219  */
220 VLINK_EXPORT void deallocate_frame(void* ptr, size_t size) noexcept;
221 
222 /**
223  * @struct FinalAwaiter
224  * @brief Final-suspend awaiter for @c Task promises; performs symmetric transfer.
225  *
226  * @details
227  * On final suspension the promise hands its stored continuation back to the dispatcher so the
228  * outer coroutine resumes immediately. When no continuation is set (the @c Task was never
229  * awaited) the awaiter returns @c std::noop_coroutine and the dispatcher unwinds normally.
230  */
231 struct FinalAwaiter final {
232  /**
233  * @brief Always returns @c false; the final suspension is unconditional.
234  *
235  * @return @c false.
236  */
237  // NOLINTNEXTLINE(readability-convert-member-functions-to-static)
238  bool await_ready() const noexcept;
239 
240  /**
241  * @brief Returns the parent coroutine handle for symmetric transfer.
242  *
243  * @tparam PromiseT Promise type of the finishing coroutine.
244  * @param handle Handle to the finishing coroutine; its promise carries
245  * the @c continuation to resume.
246  * @return The stored continuation, or @c std::noop_coroutine() if none.
247  */
248  template <typename PromiseT>
249  std::coroutine_handle<> await_suspend(std::coroutine_handle<PromiseT> handle) noexcept;
250 
251  /**
252  * @brief No-op resume; the final awaiter never produces a value.
253  */
254  // NOLINTNEXTLINE(readability-convert-member-functions-to-static)
255  void await_resume() const noexcept;
256 };
257 
258 /**
259  * @struct TaskPromiseBase
260  * @brief Shared storage and suspension hooks reused by every @c Task promise specialisation.
261  *
262  * @details
263  * Holds the parent continuation handle resumed at final suspension and an
264  * @c std::exception_ptr capturing any unhandled exception thrown by the coroutine body.
265  *
266  * @tparam TypeT Result type of the owning @c Task.
267  */
268 template <typename TypeT>
269 struct TaskPromiseBase {
270  std::coroutine_handle<> continuation; ///< Awaiter coroutine to resume on final suspension.
271  std::exception_ptr exception; ///< Captured unhandled exception, if any.
272 
273  /**
274  * @brief Initial-suspend hook; always suspends so the task starts lazily.
275  *
276  * @return @c std::suspend_always.
277  */
278  std::suspend_always initial_suspend() noexcept;
279 
280  /**
281  * @brief Final-suspend hook; returns a @c FinalAwaiter for symmetric transfer.
282  *
283  * @return @c FinalAwaiter instance bound to this promise.
284  */
285  FinalAwaiter final_suspend() noexcept;
286 
287  /**
288  * @brief Captures the currently active exception into the @c exception slot.
289  */
290  void unhandled_exception() noexcept;
291 };
292 
293 /**
294  * @struct TaskPromise
295  * @brief Promise type for value-returning @c Task<TypeT>.
296  *
297  * @details
298  * Allocates frames from the global @c MemoryPool and stores the @c co_return value in
299  * @c result so @c Task::await_resume can consume it.
300  *
301  * @tparam TypeT Result type produced by the coroutine body.
302  */
303 template <typename TypeT>
304 struct TaskPromise final : TaskPromiseBase<TypeT> {
305  std::optional<TypeT> result; ///< Storage for the @c co_return value.
306 
307  /**
308  * @brief Allocates the coroutine frame from the global memory pool.
309  *
310  * @param size Frame size requested by the compiler.
311  * @return Pointer to the allocated frame.
312  */
313  static void* operator new(size_t size);
314 
315  /**
316  * @brief Returns the coroutine frame to the global memory pool.
317  *
318  * @param ptr Frame previously allocated by @c operator @c new.
319  * @param size Original frame size.
320  */
321  static void operator delete(void* ptr, size_t size) noexcept;
322 
323  /**
324  * @brief Builds the @c Task object that owns this coroutine.
325  *
326  * @return @c Task<TypeT> wrapping the handle for this promise.
327  */
328  Task<TypeT> get_return_object() noexcept;
329 
330  /**
331  * @brief Captures the @c co_return value into @c result.
332  *
333  * @tparam UpT Forwarded value type.
334  * @param v Value produced by @c co_return.
335  */
336  template <typename UpT>
337  void return_value(UpT&& v) noexcept(std::is_nothrow_constructible_v<TypeT, UpT&&>);
338 };
339 
340 /**
341  * @struct TaskPromise<void>
342  * @brief Specialisation of @c TaskPromise for @c Task<void>.
343  *
344  * @details
345  * Carries no result slot; @c return_void is a no-op aside from satisfying the
346  * coroutine promise contract.
347  */
348 template <>
349 struct TaskPromise<void> final : TaskPromiseBase<void> {
350  /**
351  * @brief Allocates the coroutine frame from the global memory pool.
352  *
353  * @param size Frame size requested by the compiler.
354  * @return Pointer to the allocated frame.
355  */
356  static void* operator new(size_t size);
357 
358  /**
359  * @brief Returns the coroutine frame to the global memory pool.
360  *
361  * @param ptr Frame previously allocated by @c operator @c new.
362  * @param size Original frame size.
363  */
364  static void operator delete(void* ptr, size_t size) noexcept;
365 
366  /**
367  * @brief Builds the @c Task<void> that owns this coroutine.
368  *
369  * @return @c Task<void> wrapping the handle for this promise.
370  */
371  Task<void> get_return_object() noexcept;
372 
373  /**
374  * @brief No-op @c co_return for void tasks.
375  */
376  void return_void() noexcept;
377 };
378 
379 /**
380  * @struct DetachedTask
381  * @brief Fire-and-forget coroutine wrapper used internally by @c co_spawn.
382  *
383  * @details
384  * Holds a @c std::coroutine_handle for a coroutine whose frame is owned by the
385  * task itself: the promise type does not suspend at the final point and the
386  * frame is destroyed automatically when the body completes. Exceptions thrown
387  * by the inner coroutine are caught and logged at error level so that a stray
388  * throw cannot tear down the owning @c MessageLoop.
389  */
390 struct VLINK_EXPORT DetachedTask final {
391  /**
392  * @struct promise_type
393  * @brief Promise type that powers @c DetachedTask coroutines.
394  *
395  * @details
396  * Frames are allocated through the global @c MemoryPool. The coroutine
397  * suspends initially (so the caller can decide where to resume it) and
398  * never suspends at the final point (the frame self-destructs).
399  */
400  struct VLINK_EXPORT promise_type final { // NOLINT(readability-identifier-naming)
401  /**
402  * @brief Allocates the coroutine frame from the global memory pool.
403  *
404  * @param size Frame size requested by the compiler.
405  * @return Pointer to the allocated frame.
406  */
407  static void* operator new(size_t size);
408 
409  /**
410  * @brief Returns the coroutine frame to the global memory pool.
411  *
412  * @param ptr Frame previously allocated by @c operator @c new.
413  * @param size Original frame size.
414  */
415  static void operator delete(void* ptr, size_t size) noexcept;
416 
417  /**
418  * @brief Builds the @c DetachedTask wrapper that holds this promise's handle.
419  *
420  * @return @c DetachedTask wrapping the new coroutine handle.
421  */
422  DetachedTask get_return_object() noexcept;
423 
424  /**
425  * @brief Initial suspension; the caller decides when to resume.
426  *
427  * @return @c std::suspend_always.
428  */
429  std::suspend_always initial_suspend() noexcept;
430 
431  /**
432  * @brief Final suspension; never suspends so the frame self-destructs.
433  *
434  * @return @c std::suspend_never.
435  */
436  std::suspend_never final_suspend() noexcept;
437 
438  /**
439  * @brief No-op @c co_return for detached coroutines.
440  */
441  void return_void() noexcept;
442 
443  /**
444  * @brief Catches and logs unhandled exceptions thrown by the body.
445  *
446  * @details
447  * Exceptions are intentionally swallowed; the detached frame must not
448  * propagate failures back into the owning @c MessageLoop.
449  */
450  void unhandled_exception() noexcept;
451  };
452 
453  using Handle = std::coroutine_handle<promise_type>;
454 
455  /**
456  * @brief Constructs an empty detached task with no associated frame.
457  */
458  DetachedTask() noexcept;
459 
460  /**
461  * @brief Destroys the detached task. Does not own the frame after release.
462  */
463  ~DetachedTask();
464 
465  /**
466  * @brief Constructs a detached task that owns @p h.
467  *
468  * @param h Coroutine handle to take ownership of.
469  */
470  explicit DetachedTask(Handle h) noexcept;
471 
472  /**
473  * @brief Move-constructs a detached task, transferring frame ownership.
474  *
475  * @param other Source task; left empty after the move.
476  */
477  DetachedTask(DetachedTask&& other) noexcept;
478 
479  /**
480  * @brief Move-assigns a detached task, transferring frame ownership.
481  *
482  * @param other Source task; left empty after the move.
483  * @return Reference to @c *this.
484  */
485  DetachedTask& operator=(DetachedTask&& other) noexcept;
486 
487  Handle handle; ///< Coroutine handle; @c {} when the task has been released.
488 
489  VLINK_DISALLOW_COPY_AND_ASSIGN(DetachedTask)
490 };
491 
492 /**
493  * @brief Bridges a @c Task<void> into a @c DetachedTask coroutine.
494  *
495  * @param task Task to await inside the detached coroutine.
496  * @return @c DetachedTask owning the new frame.
497  */
498 VLINK_EXPORT DetachedTask co_spawn_void_impl(Task<void> task);
499 
500 /**
501  * @brief Bridges a value-returning @c Task<TypeT> with a completion callback.
502  *
503  * @tparam TypeT Result type produced by the inner task.
504  * @tparam CallbackT Callable invoked with the task's result on success.
505  * @param task Inner task to await.
506  * @param on_complete Callback invoked on the loop thread on successful completion.
507  * @return @c DetachedTask owning the new frame.
508  */
509 template <typename TypeT, typename CallbackT>
510 DetachedTask co_spawn_value_impl(Task<TypeT> task, CallbackT on_complete);
511 
512 /**
513  * @brief Bridges a @c Task<void> with a completion callback.
514  *
515  * @tparam CallbackT Callable invoked after the inner task completes.
516  * @param task Inner task to await.
517  * @param on_complete Callback invoked on the loop thread on successful completion.
518  * @return @c DetachedTask owning the new frame.
519  */
520 template <typename CallbackT>
521 DetachedTask co_spawn_void_with_cb_impl(Task<void> task, CallbackT on_complete);
522 
523 /**
524  * @enum ResumePostResult
525  * @brief Outcome of attempting to post a coroutine resume back onto a loop.
526  *
527  * @details
528  * Returned by @c post_callback_if_alive (and by the awaiter helpers built on
529  * it). Drives the retry logic in @c ScheduleAwaiter, @c YieldAwaiter,
530  * @c DelayAwaiter, @c FutureAwaiter and @c GraphAwaiter.
531  */
532 enum class ResumePostResult : uint8_t {
533  kPosted = 0, ///< Resume callback was accepted by the target loop's queue.
534  kRetry, ///< Target loop is alive but currently full; caller should retry later.
535  kClosed, ///< Target loop has been (or is about to be) quit; resume cannot be delivered.
536 };
537 
538 struct AwaiterResumeState;
539 
540 /**
541  * @brief Maximum number of @c kRetry passes before a queued resume is reported as failed.
542  *
543  * @details
544  * Each retry tick is approximately 1 ms (the @c FutureWaitLoop poll cadence),
545  * giving a total bound of roughly 30 seconds before the awaiter converts a
546  * stuck retry sequence into either an @c OperationCancelled or an
547  * @c std::runtime_error, depending on the awaiter type.
548  */
549 inline constexpr uint32_t kMaxResumePostRetry = 30000U;
550 
551 /**
552  * @brief Posts a callback onto @p loop while honouring the alive-state gate.
553  *
554  * @details
555  * If @p loop or @p alive_state is null, or if the loop has been signalled to
556  * quit, @p drop_callback is invoked and @c ResumePostResult::kClosed is
557  * returned. Otherwise the function takes @p alive_state's mutex briefly and
558  * attempts to post @p resume_callback under @c TaskOverflowPolicy::kReject; on
559  * success it returns @c kPosted, on a full queue it returns @c kRetry without
560  * having held the mutex across any sleep. On @c kPriorityType loops the
561  * @c MessageLoop::kNoPriority value is remapped to
562  * @c MessageLoop::kNormalPriority so default-priority coroutines do not sink to
563  * the bottom of the priority queue.
564  *
565  * @param loop Target loop to post onto.
566  * @param alive_state Loop's alive-state gate; obtained via @c MessageLoop::get_alive_state.
567  * @param resume_callback Callable invoked on the loop thread on success.
568  * @param drop_callback Callable invoked when the loop is closed or the task is dropped.
569  * @param priority Schedule priority; honoured only on @c kPriorityType loops.
570  * @return One of @c kPosted, @c kRetry, @c kClosed.
571  */
572 VLINK_EXPORT ResumePostResult post_callback_if_alive(MessageLoop* loop,
573  const std::shared_ptr<MessageLoop::AliveState>& alive_state,
574  MoveFunction<void()>&& resume_callback,
575  MoveFunction<void()>&& drop_callback, uint16_t priority = 0);
576 
577 /**
578  * @brief Posts the initial resume of a detached coroutine onto @p loop.
579  *
580  * @details
581  * Used by the top-level @c co_spawn entry points. If the loop is already
582  * closed or its queue cannot accept the resume after @c kMaxResumePostRetry
583  * attempts, the handle is destroyed instead of being resumed.
584  *
585  * @param loop Target loop on which the coroutine should start running.
586  * @param handle Detached coroutine handle taking ownership of the frame.
587  * @param priority Schedule priority; honoured only on @c kPriorityType loops.
588  */
589 VLINK_EXPORT void co_spawn_detached_handle(MessageLoop& loop, DetachedTask::Handle handle, uint16_t priority = 0);
590 
591 /**
592  * @brief Enqueues a poll closure on the shared @c FutureWaitLoop helper thread.
593  *
594  * @details
595  * The closure is invoked repeatedly at the helper's poll cadence (~1 ms). It
596  * returns @c true when it has finished and may be removed from the queue, or
597  * @c false to be re-polled on the next pass.
598  *
599  * @param poll Poll closure registered with the helper thread.
600  */
601 VLINK_EXPORT void register_future_wait(MoveFunction<bool()>&& poll);
602 
603 } // namespace detail
604 
605 /**
606  * @class Task
607  * @brief Coroutine return type; lazily started, awaitable from another coroutine.
608  *
609  * @details
610  * The coroutine body does not run until the @c Task is either awaited from
611  * another coroutine or passed to @c co_spawn(). Moving the @c Task transfers
612  * ownership; destroying a still-suspended @c Task destroys its coroutine frame.
613  *
614  * @tparam TypeT Result type produced by @c co_return. Use @c void for tasks
615  * that produce no value.
616  */
617 template <typename TypeT>
618 class Task final {
619  public:
620  using promise_type = detail::TaskPromise<TypeT>;
621  using Handle = std::coroutine_handle<promise_type>;
622 
623  /**
624  * @brief Constructs an empty @c Task that does not own a coroutine frame.
625  */
626  Task() noexcept = default;
627 
628  /**
629  * @brief Constructs a @c Task that takes ownership of @p handle.
630  *
631  * @param handle Coroutine handle produced by @c TaskPromise::get_return_object.
632  */
633  explicit Task(Handle handle) noexcept;
634 
635  /**
636  * @brief Move-constructs a @c Task, transferring frame ownership.
637  *
638  * @param other Source task; left empty after the move.
639  */
640  Task(Task&& other) noexcept;
641 
642  /**
643  * @brief Move-assigns a @c Task, destroying any previously owned frame.
644  *
645  * @param other Source task; left empty after the move.
646  * @return Reference to @c *this.
647  */
648  Task& operator=(Task&& other) noexcept;
649 
650  /**
651  * @brief Destroys the owned coroutine frame, if any.
652  */
653  ~Task();
654 
655  /**
656  * @brief Tests whether this task owns a coroutine frame.
657  *
658  * @return @c true if the task is non-empty.
659  */
660  [[nodiscard]] bool valid() const noexcept;
661 
662  /**
663  * @brief Tests whether the owned frame has finished executing.
664  *
665  * @return @c true if the task is valid and the coroutine has run to completion.
666  */
667  [[nodiscard]] bool done() const noexcept;
668 
669  /**
670  * @brief Coroutine awaiter readiness hook.
671  *
672  * @return @c true if the task is empty or already done; otherwise @c false.
673  */
674  bool await_ready() const noexcept;
675 
676  /**
677  * @brief Coroutine awaiter suspension hook; performs symmetric transfer.
678  *
679  * @param awaiter Coroutine handle awaiting this task's completion.
680  * @return The handle of the inner coroutine to resume.
681  */
682  std::coroutine_handle<> await_suspend(std::coroutine_handle<> awaiter) noexcept;
683 
684  /**
685  * @brief Coroutine awaiter resume hook; extracts the result or rethrows.
686  *
687  * @details
688  * Throws @c std::logic_error when called on an invalid task. If the inner
689  * coroutine captured an unhandled exception, it is rethrown here.
690  *
691  * @return The value stored by @c co_return for non-void tasks.
692  */
693  TypeT await_resume();
694 
695  /**
696  * @brief Releases ownership of the underlying coroutine handle.
697  *
698  * @return Handle owned by this task; the task is empty after the call.
699  */
700  Handle release() noexcept;
701 
702  /**
703  * @brief Returns the underlying coroutine handle without releasing ownership.
704  *
705  * @return Current coroutine handle (may be @c {} for an empty task).
706  */
707  [[nodiscard]] Handle native_handle() const noexcept;
708 
709  private:
710  Handle handle_;
711 
713 };
714 
715 /**
716  * @struct ScheduleAwaiter
717  * @brief Awaiter that resumes the coroutine on @p loop's worker thread.
718  *
719  * @details
720  * Built on top of @c detail::post_callback_if_alive. After @c co_await
721  * @c schedule(loop), the remainder of the coroutine runs on @p loop's thread.
722  * If the target loop's queue is currently full, the awaiter retries via the
723  * shared @c FutureWaitLoop helper at a ~1 ms cadence, up to
724  * @c detail::kMaxResumePostRetry attempts.
725  *
726  * @warning If @p loop is already quitting or retries are exhausted,
727  * @c await_resume() throws @c std::runtime_error. The failure resume
728  * is delivered on the @c FutureWaitLoop helper thread, not on the
729  * target loop thread.
730  */
731 struct VLINK_EXPORT ScheduleAwaiter final {
732  /**
733  * @brief Constructs an empty awaiter with no target loop.
734  */
735  ScheduleAwaiter() noexcept;
736 
737  /**
738  * @brief Constructs an awaiter bound to a target loop and resume state.
739  *
740  * @param loop Target loop on which to resume.
741  * @param alive_state Target loop's alive-state gate.
742  * @param state Shared awaiter resume state.
743  * @param priority Schedule priority for the resume.
744  */
745  ScheduleAwaiter(MessageLoop* loop, std::shared_ptr<MessageLoop::AliveState> alive_state,
746  std::shared_ptr<detail::AwaiterResumeState> state, uint16_t priority) noexcept;
747 
748  /**
749  * @brief Move-constructs the awaiter.
750  */
751  ScheduleAwaiter(ScheduleAwaiter&&) noexcept;
752 
753  /**
754  * @brief Move-assigns the awaiter.
755  */
756  ScheduleAwaiter& operator=(ScheduleAwaiter&&) noexcept;
757 
758  MessageLoop* loop{nullptr}; ///< Target loop on which to resume.
759  std::shared_ptr<MessageLoop::AliveState> alive_state; ///< Target loop's alive-state gate.
760  std::shared_ptr<detail::AwaiterResumeState> state; ///< Shared resume state holding the handle.
761  uint16_t priority{0}; ///< Schedule priority for the resume.
762  bool failed{false}; ///< Set to @c true if posting ultimately failed.
763 
764  /**
765  * @brief Awaiter readiness hook.
766  *
767  * @return Always @c false; the hop must be posted asynchronously.
768  */
769  static bool await_ready() noexcept;
770 
771  /**
772  * @brief Destructor; abandons any pending retry registered on the helper thread.
773  */
774  ~ScheduleAwaiter();
775 
776  /**
777  * @brief Posts the coroutine resume onto the target loop.
778  *
779  * @details
780  * On @c kRetry the awaiter registers a retry closure on the shared
781  * @c FutureWaitLoop helper; on @c kClosed the @c failed flag is raised so
782  * that @c await_resume can throw.
783  *
784  * @param handle Coroutine handle to resume on success.
785  * @return @c true; the awaiter always stays suspended until a resume fires.
786  */
787  bool await_suspend(std::coroutine_handle<> handle);
788 
789  /**
790  * @brief Verifies that the hop succeeded.
791  *
792  * @throws std::runtime_error if the target loop closed or retries were exhausted.
793  */
794  void await_resume();
795 
796  VLINK_DISALLOW_COPY_AND_ASSIGN(ScheduleAwaiter)
797 };
798 
799 /**
800  * @struct YieldAwaiter
801  * @brief Awaiter that re-posts the coroutine to the back of @p loop's queue.
802  *
803  * @details
804  * Useful for cooperative yielding inside a long-running coroutine running on
805  * the loop thread, allowing other queued tasks a chance to execute. Built on
806  * top of the same @c post_callback_if_alive machinery as @c ScheduleAwaiter.
807  *
808  * @warning If @p loop is already quitting or retries are exhausted,
809  * @c await_resume() throws @c std::runtime_error. The failure resume
810  * is delivered on the @c FutureWaitLoop helper thread.
811  */
812 struct VLINK_EXPORT YieldAwaiter final {
813  /**
814  * @brief Constructs an empty awaiter with no target loop.
815  */
816  YieldAwaiter() noexcept;
817 
818  /**
819  * @brief Constructs an awaiter bound to a target loop and resume state.
820  *
821  * @param loop Target loop on which to re-post.
822  * @param alive_state Target loop's alive-state gate.
823  * @param state Shared awaiter resume state.
824  * @param priority Schedule priority for the re-post.
825  */
826  YieldAwaiter(MessageLoop* loop, std::shared_ptr<MessageLoop::AliveState> alive_state,
827  std::shared_ptr<detail::AwaiterResumeState> state, uint16_t priority) noexcept;
828 
829  /**
830  * @brief Move-constructs the awaiter.
831  */
832  YieldAwaiter(YieldAwaiter&&) noexcept;
833 
834  /**
835  * @brief Move-assigns the awaiter.
836  */
837  YieldAwaiter& operator=(YieldAwaiter&&) noexcept;
838 
839  MessageLoop* loop{nullptr}; ///< Target loop to re-post onto.
840  std::shared_ptr<MessageLoop::AliveState> alive_state; ///< Target loop's alive-state gate.
841  std::shared_ptr<detail::AwaiterResumeState> state; ///< Shared resume state holding the handle.
842  uint16_t priority{0}; ///< Schedule priority for the re-post.
843  bool failed{false}; ///< Set to @c true if posting ultimately failed.
844 
845  /**
846  * @brief Awaiter readiness hook.
847  *
848  * @return Always @c false; the yield must complete a queue round-trip.
849  */
850  static bool await_ready() noexcept;
851 
852  /**
853  * @brief Destructor; abandons any pending retry registered on the helper thread.
854  */
855  ~YieldAwaiter();
856 
857  /**
858  * @brief Re-posts the coroutine resume onto the target loop.
859  *
860  * @param handle Coroutine handle to resume on success.
861  * @return @c true; the awaiter always stays suspended until a resume fires.
862  */
863  bool await_suspend(std::coroutine_handle<> handle);
864 
865  /**
866  * @brief Verifies that the yield round-trip succeeded.
867  *
868  * @throws std::runtime_error if the target loop closed or retries were exhausted.
869  */
870  void await_resume();
871 
872  VLINK_DISALLOW_COPY_AND_ASSIGN(YieldAwaiter)
873 };
874 
875 /**
876  * @struct DelayAwaiter
877  * @brief Awaiter that suspends the coroutine for @p ms milliseconds.
878  *
879  * @details
880  * On success the coroutine resumes on @p loop's thread, even for @p ms == 0.
881  * Delay completion is observed by the shared @c FutureWaitLoop helper thread
882  * at a ~1 ms polling cadence. When the deadline is reached the awaiter posts
883  * its resume through @c post_callback_if_alive; full-queue conditions retry up
884  * to @c detail::kMaxResumePostRetry times before failing.
885  *
886  * @warning If @p loop is already quitting or retries are exhausted,
887  * @c await_resume() throws @c std::runtime_error. The failure resume
888  * is delivered on the @c FutureWaitLoop helper thread.
889  */
890 struct VLINK_EXPORT DelayAwaiter final {
891  /**
892  * @brief Constructs an empty awaiter with no target loop.
893  */
894  DelayAwaiter() noexcept;
895 
896  /**
897  * @brief Constructs an awaiter bound to a target loop and delay.
898  *
899  * @param loop Target loop on which to resume.
900  * @param alive_state Target loop's alive-state gate.
901  * @param state Shared awaiter resume state.
902  * @param ms Delay in milliseconds before the resume is posted.
903  * @param priority Schedule priority for the resume.
904  */
905  DelayAwaiter(MessageLoop* loop, std::shared_ptr<MessageLoop::AliveState> alive_state,
906  std::shared_ptr<detail::AwaiterResumeState> state, uint32_t ms, uint16_t priority) noexcept;
907 
908  /**
909  * @brief Move-constructs the awaiter.
910  */
911  DelayAwaiter(DelayAwaiter&&) noexcept;
912 
913  /**
914  * @brief Move-assigns the awaiter.
915  */
916  DelayAwaiter& operator=(DelayAwaiter&&) noexcept;
917 
918  MessageLoop* loop{nullptr}; ///< Target loop to resume on.
919  std::shared_ptr<MessageLoop::AliveState> alive_state; ///< Target loop's alive-state gate.
920  std::shared_ptr<detail::AwaiterResumeState> state; ///< Shared resume state holding the handle.
921  uint32_t ms{0}; ///< Delay in milliseconds.
922  uint16_t priority{0}; ///< Schedule priority for the resume.
923  bool failed{false}; ///< Set to @c true if scheduling failed.
924 
925  /**
926  * @brief Awaiter readiness hook.
927  *
928  * @return Always @c false; the delay must elapse asynchronously.
929  */
930  static bool await_ready() noexcept;
931 
932  /**
933  * @brief Destructor; abandons any pending retry registered on the helper thread.
934  */
935  ~DelayAwaiter();
936 
937  /**
938  * @brief Registers the delayed resume on the shared @c FutureWaitLoop helper.
939  *
940  * @param handle Coroutine handle to resume once the delay has elapsed.
941  * @return @c true; the awaiter always stays suspended until a resume fires.
942  */
943  bool await_suspend(std::coroutine_handle<> handle);
944 
945  /**
946  * @brief Verifies that the delay completed and the resume was posted.
947  *
948  * @throws std::runtime_error if the target loop closed or retries were exhausted.
949  */
950  void await_resume();
951 
952  VLINK_DISALLOW_COPY_AND_ASSIGN(DelayAwaiter)
953 };
954 
955 /**
956  * @class FutureAwaiter
957  * @brief Awaiter that suspends until a @c std::future becomes ready.
958  *
959  * @details
960  * Backed by the process-wide @c FutureWaitLoop helper thread (see
961  * @c detail::register_future_wait). A single long-lived helper thread services
962  * all pending @c await_future calls: it polls each registered future at a
963  * ~1 ms cadence using a non-blocking @c wait_for(0) and posts the resume back
964  * to @p loop as soon as the future is ready. The future is held in shared
965  * waiter state so a destroyed awaiter can retire the pending poll without
966  * leaving a dangling frame pointer in the helper thread.
967  *
968  * @note No per-await thread is spawned. Resume latency is bounded by the
969  * helper's ~1 ms cadence plus the target loop's task dispatch delay.
970  *
971  * @warning If @p loop is closed before the ready future can be posted back,
972  * the coroutine resumes on the helper thread and @c await_resume()
973  * throws @c OperationCancelled. On the normal path, exceptions
974  * stored in the future propagate via @c future.get().
975  *
976  * @tparam TypeT Result type produced by the awaited future.
977  */
978 template <typename TypeT>
979 class FutureAwaiter final {
980  public:
981  /**
982  * @brief Constructs an awaiter that bridges @p fut into the coroutine layer.
983  *
984  * @param loop Target loop on which to resume once @p fut is ready.
985  * @param fut Future to await; ownership is transferred into the awaiter.
986  */
987  FutureAwaiter(MessageLoop* loop, std::future<TypeT> fut) noexcept;
988 
989  /**
990  * @brief Move-constructs the awaiter.
991  */
992  FutureAwaiter(FutureAwaiter&&) noexcept = default;
993 
994  /**
995  * @brief Move-assigns the awaiter.
996  */
997  FutureAwaiter& operator=(FutureAwaiter&&) noexcept = default;
998 
999  /**
1000  * @brief Destructor; abandons the pending helper poll.
1001  */
1002  ~FutureAwaiter();
1003 
1004  /**
1005  * @brief Short-circuits when the future is already in a ready state.
1006  *
1007  * @return @c true if the future is ready or invalid; otherwise @c false.
1008  */
1009  bool await_ready() const noexcept;
1010 
1011  /**
1012  * @brief Registers a poll closure on the shared @c FutureWaitLoop helper.
1013  *
1014  * @param handle Coroutine handle to resume once the future is ready.
1015  */
1016  void await_suspend(std::coroutine_handle<> handle);
1017 
1018  /**
1019  * @brief Extracts the value from the future or rethrows its exception.
1020  *
1021  * @return Future result for non-void instantiations.
1022  * @throws OperationCancelled if the target loop closed before the resume could be posted.
1023  * @note Exceptions stored in the future are rethrown by @c future.get().
1024  */
1025  TypeT await_resume();
1026 
1027  private:
1028  struct State;
1029 
1030  MessageLoop* loop_{nullptr};
1031  std::shared_ptr<State> state_;
1032 
1033  VLINK_DISALLOW_COPY_AND_ASSIGN(FutureAwaiter)
1034 };
1035 
1036 /**
1037  * @class GraphAwaiter
1038  * @brief Awaiter that suspends until a @c GraphTask reaches @c kStatusDone.
1039  *
1040  * @details
1041  * Installs a single-shot status callback on the graph; on @c kStatusDone the
1042  * coroutine resume is posted back onto @p loop. If the graph is already done
1043  * at the await point, @c await_ready returns @c true and no callback is
1044  * registered. Cancelled or never-started graphs leave the coroutine suspended
1045  * indefinitely; the caller must ensure the graph eventually runs. The status
1046  * subscription is unregistered automatically once the awaiter resumes, so
1047  * multiple coroutines awaiting the same graph are safe.
1048  *
1049  * @warning If @p loop closes after the graph becomes done but before the resume
1050  * can be posted, @c await_resume() throws @c OperationCancelled. The
1051  * failure resume is delivered on the @c FutureWaitLoop helper thread,
1052  * not on the target loop thread.
1053  */
1054 class VLINK_EXPORT GraphAwaiter final {
1055  public:
1056  /**
1057  * @brief Constructs an awaiter that monitors @p graph for completion.
1058  *
1059  * @param loop Loop on which to resume the coroutine.
1060  * @param graph Graph task to monitor.
1061  */
1062  GraphAwaiter(MessageLoop* loop, GraphTaskPtr graph) noexcept;
1063 
1064  /**
1065  * @brief Move-constructs the awaiter.
1066  */
1067  GraphAwaiter(GraphAwaiter&&) noexcept;
1068 
1069  /**
1070  * @brief Move-assigns the awaiter.
1071  */
1072  GraphAwaiter& operator=(GraphAwaiter&&) noexcept;
1073 
1074  /**
1075  * @brief Destructor; unregisters the status callback and abandons any pending resume.
1076  */
1077  ~GraphAwaiter();
1078 
1079  /**
1080  * @brief Short-circuits when the graph is empty or already complete.
1081  *
1082  * @return @c true if @c graph_ is null or already @c kStatusDone.
1083  */
1084  [[nodiscard]] bool await_ready() const noexcept;
1085 
1086  /**
1087  * @brief Registers a single-shot status callback waiting for @c kStatusDone.
1088  *
1089  * @param handle Coroutine handle to resume on completion.
1090  */
1091  void await_suspend(std::coroutine_handle<> handle);
1092 
1093  /**
1094  * @brief Verifies that the resume reached the target loop.
1095  *
1096  * @throws OperationCancelled if the target loop closed before the resume could be posted.
1097  */
1098  void await_resume() const;
1099 
1100  private:
1101  struct State;
1102 
1103  MessageLoop* loop_{nullptr};
1104  GraphTaskPtr graph_;
1105  std::shared_ptr<State> state_;
1106 
1107  VLINK_DISALLOW_COPY_AND_ASSIGN(GraphAwaiter)
1108 };
1109 
1110 /**
1111  * @brief Returns an awaiter that hops the coroutine onto @p loop's thread.
1112  *
1113  * @details
1114  * Use @c co_await @c schedule(loop) to migrate the remainder of a coroutine to
1115  * @p loop's thread. On @c kPriorityType loops the @c MessageLoop::kNoPriority
1116  * sentinel is internally remapped to @c MessageLoop::kNormalPriority so default
1117  * coroutines do not sink to the bottom of the queue.
1118  *
1119  * @param loop Target loop to resume on.
1120  * @param priority Schedule priority. Values come from @c MessageLoop::Priority
1121  * (@c kNoPriority is FIFO, @c kNormalPriority is 100, etc.).
1122  * Honoured only on @c kPriorityType loops.
1123  * @return @c ScheduleAwaiter ready to be @c co_await-ed.
1124  */
1125 VLINK_EXPORT ScheduleAwaiter schedule(MessageLoop& loop, uint16_t priority = MessageLoop::kNoPriority) noexcept;
1126 
1127 /**
1128  * @brief Returns an awaiter that yields once back to @p loop's queue.
1129  *
1130  * @param loop Loop owning the coroutine.
1131  * @param priority Re-post priority from @c MessageLoop::Priority; honoured only
1132  * on @c kPriorityType loops.
1133  * @return @c YieldAwaiter ready to be @c co_await-ed.
1134  */
1135 VLINK_EXPORT YieldAwaiter yield(MessageLoop& loop, uint16_t priority = MessageLoop::kNoPriority) noexcept;
1136 
1137 /**
1138  * @brief Returns an awaiter that suspends for @p ms milliseconds.
1139  *
1140  * @param loop Loop on which the coroutine resumes after the delay.
1141  * @param ms Delay in milliseconds. Zero is allowed and still produces a
1142  * resume hop through the helper thread.
1143  * @param priority Resume priority from @c MessageLoop::Priority; honoured only
1144  * on @c kPriorityType loops.
1145  * @return @c DelayAwaiter ready to be @c co_await-ed.
1146  */
1147 VLINK_EXPORT DelayAwaiter delay_ms(MessageLoop& loop, uint32_t ms,
1148  uint16_t priority = MessageLoop::kNoPriority) noexcept;
1149 
1150 /**
1151  * @brief Returns an awaiter that suspends until @p fut is ready, resuming on @p loop.
1152  *
1153  * @tparam TypeT Result type stored in the future.
1154  * @param loop Loop on which the coroutine resumes once @p fut is ready.
1155  * @param fut Future to await; ownership is transferred into the awaiter.
1156  * @return @c FutureAwaiter ready to be @c co_await-ed.
1157  */
1158 template <typename TypeT>
1159 FutureAwaiter<TypeT> await_future(MessageLoop& loop, std::future<TypeT> fut) noexcept;
1160 
1161 /**
1162  * @brief Schedules a top-level @c Task<void> on @p loop (fire-and-forget).
1163  *
1164  * @details
1165  * The coroutine begins executing on @p loop's thread. Ownership of the task
1166  * frame is transferred to an internal @c DetachedTask which destroys the frame
1167  * once the body completes. Exceptions thrown from the task are caught by
1168  * @c DetachedTask::unhandled_exception, logged at error level and swallowed:
1169  * the loop continues running.
1170  *
1171  * @par Joining from synchronous code
1172  * Use the three-argument @c co_spawn overload with a callback to bridge to a
1173  * @c std::promise / @c std::future pair, then wait on the future:
1174  * @code
1175  * std::promise<void> p;
1176  * auto fut = p.get_future();
1177  * Co::co_spawn(loop, my_void_task(), [&p]() { p.set_value(); });
1178  * fut.wait();
1179  * @endcode
1180  *
1181  * @note If @c my_void_task throws, the callback is not invoked and @c fut
1182  * never becomes ready. Either catch inside @c my_void_task or use a
1183  * value-returning task that converts errors into a sentinel result.
1184  *
1185  * @param loop Loop on which the coroutine starts running.
1186  * @param task Task to spawn; ownership is transferred into the call.
1187  */
1188 VLINK_EXPORT void co_spawn(MessageLoop& loop, Task<void>&& task);
1189 
1190 /**
1191  * @brief Same as @c co_spawn, with an explicit dispatch priority.
1192  *
1193  * @param loop Loop on which the coroutine starts running.
1194  * @param task Task to spawn; ownership is transferred into the call.
1195  * @param priority Schedule priority; honoured only on @c kPriorityType loops.
1196  */
1197 VLINK_EXPORT void co_spawn_with_priority(MessageLoop& loop, Task<void>&& task, uint16_t priority);
1198 
1199 /**
1200  * @brief Schedules a top-level @c Task<TypeT> with a completion callback.
1201  *
1202  * @details
1203  * The coroutine begins executing on @p loop's thread. When the task completes
1204  * successfully, @p on_complete is invoked on the loop thread with the result.
1205  *
1206  * @warning Exceptions thrown by the task are swallowed: they are logged via
1207  * @c DetachedTask::unhandled_exception and @p on_complete is NOT
1208  * invoked. This API is unsuitable for code that must observe task
1209  * failure. To observe failure either:
1210  * - Catch inside the task body and convert to a sentinel result, or
1211  * - Use @c when_all / @c when_any / @c sequence, which capture the
1212  * original exception via the runner+guard machinery, or
1213  * - @c co_await the @c Task directly from a parent coroutine.
1214  *
1215  * @par Example
1216  * @code
1217  * auto pp = std::make_shared<std::promise<TypeT>>();
1218  * auto fut = pp->get_future();
1219  * Co::co_spawn(loop, task_that_may_throw(),
1220  * [pp](TypeT v) { pp->set_value(std::move(v)); });
1221  * try {
1222  * TypeT result = co_await Co::await_future(loop, std::move(fut));
1223  * } catch (...) {
1224  * // The future still does not see the original exception.
1225  * }
1226  * @endcode
1227  *
1228  * @tparam TypeT Result type produced by the task.
1229  * @tparam CallbackT Callable invoked with the task's result on success.
1230  * @param loop Loop on which the coroutine starts running.
1231  * @param task Task to spawn; ownership is transferred into the call.
1232  * @param on_complete Completion callback invoked on the loop thread on success.
1233  */
1234 template <typename TypeT, typename CallbackT,
1235  // NOLINTNEXTLINE(modernize-use-constraints)
1236  typename = std::enable_if_t<!std::is_integral_v<std::remove_reference_t<CallbackT>>>>
1237 void co_spawn(MessageLoop& loop, Task<TypeT>&& task, CallbackT&& on_complete);
1238 
1239 /**
1240  * @brief Same as the value-returning @c co_spawn, with an explicit dispatch priority.
1241  *
1242  * @tparam TypeT Result type produced by the task.
1243  * @tparam CallbackT Callable invoked with the task's result on success.
1244  * @param loop Loop on which the coroutine starts running.
1245  * @param task Task to spawn; ownership is transferred into the call.
1246  * @param on_complete Completion callback invoked on the loop thread on success.
1247  * @param priority Schedule priority; honoured only on @c kPriorityType loops.
1248  */
1249 template <typename TypeT, typename CallbackT,
1250  // NOLINTNEXTLINE(modernize-use-constraints)
1251  typename = std::enable_if_t<!std::is_integral_v<std::remove_reference_t<CallbackT>>>>
1252 void co_spawn_with_priority(MessageLoop& loop, Task<TypeT>&& task, CallbackT&& on_complete, uint16_t priority);
1253 
1254 /**
1255  * @brief Returns an awaiter that suspends until @p graph reaches @c kStatusDone.
1256  *
1257  * @param loop Loop on which to resume the coroutine.
1258  * @param graph Graph task to monitor; must be non-null.
1259  * @return @c GraphAwaiter ready to be @c co_await-ed.
1260  * @throws std::invalid_argument if @p graph is null.
1261  */
1262 VLINK_EXPORT GraphAwaiter await_graph(MessageLoop& loop, GraphTaskPtr graph);
1263 
1264 /**
1265  * @brief Coroutine wrapper around @c MessageLoop::exec_task.
1266  *
1267  * @details
1268  * Posts @p callback under the given @c Schedule::Config envelope and suspends
1269  * until the callback has run. Exceptions thrown by @p callback are stored in
1270  * an internal @c std::promise and rethrown from the awaiting @c co_await. If
1271  * the underlying @c exec_task post fails (queue full or loop closed), the
1272  * coroutine resumes with an @c std::runtime_error carrying a descriptive
1273  * message.
1274  *
1275  * @tparam CallbackT Callable returning @c void.
1276  * @param loop Loop on which the callback is scheduled.
1277  * @param config Schedule envelope (delay, priority, timeouts).
1278  * @param callback Callable to execute on the loop thread.
1279  * @return @c Task<void> that completes after @p callback returns.
1280  */
1281 template <typename CallbackT>
1282 Task<void> exec(MessageLoop& loop, const Schedule::Config& config, CallbackT&& callback);
1283 
1284 /**
1285  * @brief Awaits every @c Task<TypeT> in @p tasks and returns their results.
1286  *
1287  * @details
1288  * Each task is dispatched via @c co_spawn on @p loop. Results are stored in
1289  * the order of the input vector. @c when_all waits for every sub-task to
1290  * finish (success or failure) before completing. If any sub-task throws, the
1291  * first exception observed is rethrown from the awaiting @c co_await with its
1292  * original type preserved; subsequent exceptions from sibling tasks are
1293  * dropped.
1294  *
1295  * @note @p TypeT must be default-constructible; the results vector is
1296  * pre-sized via @c std::vector<TypeT>(count).
1297  *
1298  * @tparam TypeT Result type produced by every sub-task.
1299  * @param loop Loop on which sub-tasks are spawned and the caller resumes.
1300  * @param tasks Sub-tasks to await; ownership is transferred into the call.
1301  * @return @c Task that resolves to a vector of results in input order.
1302  */
1303 template <typename TypeT>
1304 Task<std::vector<TypeT>> when_all(MessageLoop& loop, std::vector<Task<TypeT>> tasks);
1305 
1306 /**
1307  * @brief Awaits every @c Task<void> in @p tasks to complete.
1308  *
1309  * @details
1310  * Waits for every sub-task to finish. If any sub-task throws, the first
1311  * exception observed is rethrown from the awaiting @c co_await with its
1312  * original type preserved; subsequent exceptions are dropped.
1313  *
1314  * @param loop Loop on which sub-tasks are spawned and the caller resumes.
1315  * @param tasks Sub-tasks to await; ownership is transferred into the call.
1316  * @return @c Task<void> that completes once every sub-task has finished.
1317  */
1318 VLINK_EXPORT Task<void> when_all(MessageLoop& loop, std::vector<Task<void>> tasks);
1319 
1320 /**
1321  * @brief Records the index and result of the first @c Task<TypeT> to complete successfully.
1322  *
1323  * @details
1324  * The winner (first successful completion) is selected by an atomic
1325  * compare-exchange; losing tasks continue running until completion and their
1326  * results are dropped. @c when_any does not return until every sub-task has
1327  * finished — this is required to avoid leaking orphaned sub-task frames, since
1328  * this layer offers no cross-coroutine cancellation. If "abandon losers"
1329  * semantics with bounded latency are required, each sub-task itself must
1330  * respect a deadline. If every sub-task throws, the first exception observed
1331  * is rethrown from the awaiting @c co_await.
1332  *
1333  * @tparam TypeT Result type produced by every sub-task.
1334  * @param loop Loop on which sub-tasks are spawned and the caller resumes.
1335  * @param tasks Sub-tasks to race; ownership is transferred into the call.
1336  * @return @c Task resolving to @c {winning_index, winning_result}.
1337  * @throws std::invalid_argument if @p tasks is empty; observed at the awaiting
1338  * @c co_await site (since @c when_any is itself a coroutine).
1339  */
1340 template <typename TypeT>
1341 Task<std::pair<size_t, TypeT>> when_any(MessageLoop& loop, std::vector<Task<TypeT>> tasks);
1342 
1343 /**
1344  * @brief Records the index of the first @c Task<void> to complete successfully.
1345  *
1346  * @details
1347  * The winner (first successful completion) is selected by an atomic
1348  * compare-exchange; losing tasks continue running until completion. @c when_any
1349  * does not return until every sub-task has finished — this is required to
1350  * avoid leaking orphaned sub-task frames (no cross-coroutine cancellation is
1351  * provided). If every sub-task throws, the first exception observed is
1352  * rethrown from the awaiting @c co_await.
1353  *
1354  * @param loop Loop on which sub-tasks are spawned and the caller resumes.
1355  * @param tasks Sub-tasks to race; ownership is transferred into the call.
1356  * @return @c Task resolving to the winning sub-task's index.
1357  * @throws std::invalid_argument if @p tasks is empty; observed at the awaiting
1358  * @c co_await site (since @c when_any is itself a coroutine).
1359  */
1360 VLINK_EXPORT Task<size_t> when_any(MessageLoop& loop, std::vector<Task<void>> tasks);
1361 
1362 /**
1363  * @brief Runs @p tasks sequentially, awaiting each one before starting the next.
1364  *
1365  * @details
1366  * Convenience for composing a linear pipeline of @c Task<void>. Each task is
1367  * spawned on @p loop via the same runner+guard machinery used by @c when_all
1368  * and joined through @c await_future, so both the success path and any
1369  * exception thrown by a sub-task resume the caller on @p loop's thread. If a
1370  * task throws, the exception is rethrown from the awaiting @c co_await with
1371  * its original type preserved and the remaining tasks are skipped.
1372  *
1373  * @param loop Loop on which sub-tasks are spawned and the caller resumes.
1374  * @param tasks Sub-tasks to run in order; ownership is transferred into the call.
1375  * @return @c Task<void> that completes once every sub-task has finished or one threw.
1376  */
1377 VLINK_EXPORT Task<void> sequence(MessageLoop& loop, std::vector<Task<void>> tasks);
1378 
1379 ////////////////////////////////////////////////////////////////
1380 /// Details
1381 ////////////////////////////////////////////////////////////////
1382 
1383 namespace detail {
1384 
1385 // NOLINTNEXTLINE(readability-convert-member-functions-to-static)
1386 inline bool FinalAwaiter::await_ready() const noexcept { return false; }
1387 
1388 template <typename PromiseT>
1389 inline std::coroutine_handle<> FinalAwaiter::await_suspend(std::coroutine_handle<PromiseT> handle) noexcept {
1390  auto next = handle.promise().continuation;
1391  return next ? next : std::noop_coroutine();
1392 }
1393 
1394 // NOLINTNEXTLINE(readability-convert-member-functions-to-static)
1395 inline void FinalAwaiter::await_resume() const noexcept {} // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1396 
1397 template <typename TypeT>
1398 inline std::suspend_always TaskPromiseBase<TypeT>::initial_suspend() noexcept {
1399  return {};
1400 }
1401 
1402 template <typename TypeT>
1403 inline FinalAwaiter TaskPromiseBase<TypeT>::final_suspend() noexcept {
1404  return {};
1405 }
1406 
1407 template <typename TypeT>
1408 inline void TaskPromiseBase<TypeT>::unhandled_exception() noexcept {
1409  exception = std::current_exception();
1410 }
1411 
1412 template <typename TypeT>
1413 inline void* TaskPromise<TypeT>::operator new(size_t size) {
1414  return detail::allocate_frame(size);
1415 }
1416 
1417 template <typename TypeT>
1418 inline void TaskPromise<TypeT>::operator delete(void* ptr, size_t size) noexcept {
1419  detail::deallocate_frame(ptr, size);
1420 }
1421 
1422 template <typename TypeT>
1423 inline Task<TypeT> TaskPromise<TypeT>::get_return_object() noexcept {
1424  return Task<TypeT>{std::coroutine_handle<TaskPromise<TypeT>>::from_promise(*this)};
1425 }
1426 
1427 template <typename TypeT>
1428 template <typename UpT>
1429 inline void TaskPromise<TypeT>::return_value(UpT&& v) noexcept(std::is_nothrow_constructible_v<TypeT, UpT&&>) {
1430  result.emplace(std::forward<UpT>(v));
1431 }
1432 
1433 inline void* TaskPromise<void>::operator new(size_t size) { return detail::allocate_frame(size); }
1434 
1435 inline void TaskPromise<void>::operator delete(void* ptr, size_t size) noexcept { detail::deallocate_frame(ptr, size); }
1436 
1437 inline Task<void> TaskPromise<void>::get_return_object() noexcept {
1438  return Task<void>{std::coroutine_handle<TaskPromise<void>>::from_promise(*this)};
1439 }
1440 
1441 inline void TaskPromise<void>::return_void() noexcept {}
1442 
1443 template <typename TypeT, typename CallbackT>
1444 inline DetachedTask co_spawn_value_impl(Task<TypeT> task, CallbackT on_complete) {
1445  on_complete(co_await std::move(task));
1446 }
1447 
1448 template <typename CallbackT>
1449 inline DetachedTask co_spawn_void_with_cb_impl(Task<void> task, CallbackT on_complete) {
1450  co_await std::move(task);
1451  on_complete();
1452 }
1453 
1454 } // namespace detail
1455 
1456 template <typename TypeT>
1457 inline Task<TypeT>::Task(Handle handle) noexcept : handle_(handle) {}
1458 
1459 template <typename TypeT>
1460 inline Task<TypeT>::Task(Task&& other) noexcept : handle_(std::exchange(other.handle_, {})) {}
1461 
1462 template <typename TypeT>
1463 inline Task<TypeT>& Task<TypeT>::operator=(Task&& other) noexcept {
1464  if VLIKELY (this != &other) {
1465  if (handle_) {
1466  handle_.destroy(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1467  }
1468  handle_ = std::exchange(other.handle_, {});
1469  }
1470 
1471  return *this;
1472 }
1473 
1474 template <typename TypeT>
1475 inline Task<TypeT>::~Task() {
1476  if (handle_) {
1477  handle_.destroy();
1478  }
1479 }
1480 
1481 template <typename TypeT>
1482 inline bool Task<TypeT>::valid() const noexcept {
1483  return handle_ != nullptr;
1484 }
1485 
1486 template <typename TypeT>
1487 inline bool Task<TypeT>::done() const noexcept {
1488  return handle_ && handle_.done();
1489 }
1490 
1491 template <typename TypeT>
1492 inline bool Task<TypeT>::await_ready() const noexcept {
1493  if VUNLIKELY (!handle_) {
1494  return true;
1495  }
1496 
1497  return handle_.done();
1498 }
1499 
1500 template <typename TypeT>
1501 inline std::coroutine_handle<> Task<TypeT>::await_suspend(std::coroutine_handle<> awaiter) noexcept {
1502  handle_.promise().continuation = awaiter;
1503  return handle_;
1504 }
1505 
1506 template <typename TypeT>
1507 inline TypeT Task<TypeT>::await_resume() {
1508  if VUNLIKELY (!handle_) {
1509  throw std::logic_error("vlink::Coroutine::Task::await_resume on invalid Task (moved-from or default-constructed)");
1510  }
1511 
1512  if VUNLIKELY (handle_.promise().exception) {
1513  std::rethrow_exception(handle_.promise().exception);
1514  }
1515 
1516  if constexpr (!std::is_void_v<TypeT>) {
1517  return std::move(*handle_.promise().result);
1518  }
1519 }
1520 
1521 template <typename TypeT>
1522 inline typename Task<TypeT>::Handle Task<TypeT>::release() noexcept {
1523  return std::exchange(handle_, {});
1524 }
1525 
1526 template <typename TypeT>
1527 inline typename Task<TypeT>::Handle Task<TypeT>::native_handle() const noexcept {
1528  return handle_;
1529 }
1530 
1531 template <typename TypeT>
1532 struct FutureAwaiter<TypeT>::State final {
1533  explicit State(std::future<TypeT> future) noexcept : fut(std::move(future)) {}
1534 
1535  void set_handle(std::coroutine_handle<> h) {
1536  std::lock_guard lock(mtx);
1537 
1538  if (!abandoned.load(std::memory_order_acquire)) {
1539  handle = h;
1540  }
1541  }
1542 
1543  std::coroutine_handle<> take_handle() {
1544  std::lock_guard lock(mtx);
1545  return std::exchange(handle, {});
1546  }
1547 
1548  void clear_handle() {
1549  std::lock_guard lock(mtx);
1550  handle = {};
1551  }
1552 
1553  void resume_ready() {
1554  auto h = take_handle();
1555 
1556  if (h) {
1557  h.resume();
1558  }
1559  }
1560 
1561  void cancel_and_resume() {
1562  target_closed.store(true, std::memory_order_release);
1563  auto h = take_handle();
1564 
1565  if (h) {
1566  h.resume();
1567  }
1568  }
1569 
1570  std::future<TypeT> fut;
1571  std::mutex mtx;
1572  std::coroutine_handle<> handle;
1573  std::atomic_bool abandoned{false};
1574  std::atomic_bool target_closed{false};
1575 };
1576 
1577 template <typename TypeT>
1578 inline FutureAwaiter<TypeT>::FutureAwaiter(MessageLoop* loop, std::future<TypeT> fut) noexcept
1579  : loop_(loop), state_(MemoryResource::make_shared<State>(std::move(fut))) {}
1580 
1581 template <typename TypeT>
1582 inline FutureAwaiter<TypeT>::~FutureAwaiter() {
1583  if (state_) {
1584  state_->abandoned.store(true, std::memory_order_release);
1585  state_->clear_handle();
1586  }
1587 }
1588 
1589 template <typename TypeT>
1590 inline bool FutureAwaiter<TypeT>::await_ready() const noexcept {
1591  return !state_->fut.valid() || state_->fut.wait_for(std::chrono::nanoseconds::zero()) == std::future_status::ready;
1592 }
1593 
1594 template <typename TypeT>
1595 inline void FutureAwaiter<TypeT>::await_suspend(std::coroutine_handle<> handle) {
1596  auto* loop_ptr = loop_;
1597  auto loop_alive = loop_ptr->get_alive_state();
1598  auto state = state_;
1599  state->set_handle(handle);
1600 
1601  detail::register_future_wait(
1602  MoveFunction<bool()>([loop_ptr, loop_alive, state = std::move(state),
1603  retry_count = 0U]() mutable -> bool { // LCOV_EXCL_BR_LINE GCOVR_EXCL_BR_LINE
1604  if VUNLIKELY (state->abandoned.load(std::memory_order_acquire)) {
1605  return true; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1606  }
1607 
1608  if (state->fut.valid() && state->fut.wait_for(std::chrono::nanoseconds::zero()) != std::future_status::ready) {
1609  return false;
1610  }
1611 
1612  const auto result = detail::post_callback_if_alive(
1613  loop_ptr, loop_alive, MoveFunction<void()>([state]() { state->resume_ready(); }),
1614  MoveFunction<void()>([state]() {
1615  detail::register_future_wait(MoveFunction<bool()>([state]() { // LCOV_EXCL_BR_LINE GCOVR_EXCL_BR_LINE
1616  if (!state->abandoned.load(std::memory_order_acquire)) { // LCOV_EXCL_BR_LINE GCOVR_EXCL_BR_LINE
1617  state->cancel_and_resume();
1618  }
1619 
1620  return true;
1621  }));
1622  }));
1623 
1624  if (result == detail::ResumePostResult::kPosted) { // LCOV_EXCL_BR_LINE GCOVR_EXCL_BR_LINE
1625  return true;
1626  }
1627 
1628  if (result == detail::ResumePostResult::kRetry) { // LCOV_EXCL_BR_LINE GCOVR_EXCL_BR_LINE
1629  if (++retry_count >= detail::kMaxResumePostRetry) { // LCOV_EXCL_BR_LINE GCOVR_EXCL_BR_LINE
1630  detail::register_future_wait(MoveFunction<bool()>([state]() { // LCOV_EXCL_BR_LINE GCOVR_EXCL_BR_LINE
1631  if (!state->abandoned.load(std::memory_order_acquire)) { // LCOV_EXCL_BR_LINE GCOVR_EXCL_BR_LINE
1632  state->cancel_and_resume(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1633  }
1634  return true; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1635  }));
1636  return true; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1637  }
1638 
1639  return false;
1640  }
1641 
1642  return true;
1643  }));
1644 }
1645 
1646 template <typename TypeT>
1647 inline TypeT FutureAwaiter<TypeT>::await_resume() {
1648  state_->abandoned.store(true, std::memory_order_release);
1649  state_->clear_handle();
1650 
1651  if VUNLIKELY (state_->target_closed.load(std::memory_order_acquire)) {
1652  throw Exception::OperationCancelled{};
1653  }
1654 
1655  if constexpr (std::is_void_v<TypeT>) {
1656  state_->fut.get();
1657  } else {
1658  return state_->fut.get();
1659  }
1660 }
1661 
1662 template <typename TypeT>
1663 inline FutureAwaiter<TypeT> await_future(MessageLoop& loop, std::future<TypeT> fut) noexcept {
1664  return FutureAwaiter<TypeT>{&loop, std::move(fut)};
1665 }
1666 
1667 template <typename TypeT, typename CallbackT, typename>
1668 inline void co_spawn(MessageLoop& loop, Task<TypeT>&& task, CallbackT&& on_complete) {
1669  co_spawn_with_priority<TypeT, CallbackT>(loop, std::move(task), std::forward<CallbackT>(on_complete), 0);
1670 }
1671 
1672 template <typename TypeT, typename CallbackT, typename>
1673 inline void co_spawn_with_priority(MessageLoop& loop, Task<TypeT>&& task, CallbackT&& on_complete, uint16_t priority) {
1674  if constexpr (std::is_void_v<TypeT>) {
1675  auto detached = detail::co_spawn_void_with_cb_impl<std::decay_t<CallbackT>>(std::move(task),
1676  std::forward<CallbackT>(on_complete));
1677  auto handle = detached.handle;
1678  detached.handle = {};
1679  detail::co_spawn_detached_handle(loop, handle, priority);
1680  } else {
1681  auto detached = detail::co_spawn_value_impl<TypeT, std::decay_t<CallbackT>>(std::move(task),
1682  std::forward<CallbackT>(on_complete));
1683  auto handle = detached.handle;
1684  detached.handle = {};
1685  detail::co_spawn_detached_handle(loop, handle, priority);
1686  }
1687 }
1688 
1689 template <typename CallbackT>
1690 inline Task<void> exec(MessageLoop& loop, const Schedule::Config& config, CallbackT&& callback) {
1691  auto promise_ptr = MemoryResource::make_shared<std::promise<void>>();
1692  auto fut = promise_ptr->get_future();
1693 
1694  auto status = loop.exec_task(config, [cb = std::forward<CallbackT>(callback), promise_ptr]() mutable {
1695  try {
1696  cb();
1697  promise_ptr->set_value();
1698  } catch (...) {
1699  promise_ptr->set_exception(std::current_exception());
1700  }
1701  });
1702 
1703  if VUNLIKELY (!status.dispatch()) {
1704  promise_ptr->set_exception(std::make_exception_ptr(std::runtime_error("Coroutine::exec: exec_task post failed")));
1705  }
1706 
1707  co_await await_future(loop, std::move(fut));
1708 
1709  co_return;
1710 }
1711 
1712 namespace detail {
1713 
1714 template <typename TypeT>
1715 struct WhenAllState final {
1716  std::vector<TypeT> results;
1717  std::atomic<size_t> remaining{0};
1718  std::mutex exc_mtx;
1719  std::exception_ptr first_exc;
1720  std::promise<void> promise;
1721 };
1722 
1723 template <typename TypeT>
1724 struct WhenAllGuard final {
1725  std::shared_ptr<WhenAllState<TypeT>> state;
1726  bool completed_normally{false};
1727 
1728  explicit WhenAllGuard(std::shared_ptr<WhenAllState<TypeT>> s) noexcept : state(std::move(s)) {}
1729 
1730  WhenAllGuard(WhenAllGuard&& other) noexcept
1731  : state(std::move(other.state)), completed_normally(other.completed_normally) {}
1732 
1733  WhenAllGuard(const WhenAllGuard&) = delete;
1734 
1735  WhenAllGuard& operator=(const WhenAllGuard&) = delete;
1736 
1737  WhenAllGuard& operator=(WhenAllGuard&&) = delete;
1738 
1739  ~WhenAllGuard() {
1740  if (!state) {
1741  return;
1742  }
1743 
1744  if (!completed_normally) {
1745  // LCOV_EXCL_START GCOVR_EXCL_START
1746  std::lock_guard lock(state->exc_mtx);
1747 
1748  if (!state->first_exc) {
1749  state->first_exc = std::make_exception_ptr(
1750  std::runtime_error("vlink::Coroutine::when_all: sub-task aborted before completion"));
1751  }
1752  }
1753  // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1754 
1755  if (state->remaining.fetch_sub(1, std::memory_order_acq_rel) == 1) {
1756  if (state->first_exc) {
1757  state->promise.set_exception(state->first_exc);
1758  } else {
1759  state->promise.set_value();
1760  }
1761  }
1762  }
1763 };
1764 
1765 template <typename TypeT>
1766 inline Task<void> when_all_runner(Task<TypeT> task, size_t i, WhenAllGuard<TypeT> guard) {
1767  try {
1768  auto value = co_await std::move(task);
1769  guard.state->results[i] = std::move(value);
1770  } catch (...) {
1771  std::lock_guard lock(guard.state->exc_mtx);
1772 
1773  if (!guard.state->first_exc) {
1774  guard.state->first_exc = std::current_exception();
1775  }
1776  }
1777 
1778  guard.completed_normally = true;
1779 }
1780 
1781 template <typename TypeT>
1782 struct WhenAnyState final {
1783  std::atomic_bool fired{false};
1784  std::atomic<size_t> remaining{0};
1785  std::mutex exc_mtx;
1786  std::exception_ptr first_exc;
1787  std::optional<std::pair<size_t, TypeT>> winner;
1788  std::promise<std::pair<size_t, TypeT>> promise;
1789 };
1790 
1791 template <typename TypeT>
1792 struct WhenAnyGuard final {
1793  std::shared_ptr<WhenAnyState<TypeT>> state;
1794  bool completed_normally{false};
1795 
1796  explicit WhenAnyGuard(std::shared_ptr<WhenAnyState<TypeT>> s) noexcept : state(std::move(s)) {}
1797 
1798  WhenAnyGuard(WhenAnyGuard&& other) noexcept
1799  : state(std::move(other.state)), completed_normally(other.completed_normally) {}
1800 
1801  WhenAnyGuard(const WhenAnyGuard&) = delete;
1802 
1803  WhenAnyGuard& operator=(const WhenAnyGuard&) = delete;
1804 
1805  WhenAnyGuard& operator=(WhenAnyGuard&&) = delete;
1806 
1807  ~WhenAnyGuard() {
1808  if (!state) {
1809  return;
1810  }
1811 
1812  if (!completed_normally) {
1813  // LCOV_EXCL_START GCOVR_EXCL_START
1814  std::lock_guard lock(state->exc_mtx);
1815 
1816  if (!state->first_exc) {
1817  state->first_exc = std::make_exception_ptr(
1818  std::runtime_error("vlink::Coroutine::when_any: sub-task aborted before completion"));
1819  }
1820  }
1821  // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1822 
1823  if (state->remaining.fetch_sub(1, std::memory_order_acq_rel) == 1) {
1824  if (state->winner) {
1825  state->promise.set_value(std::move(*state->winner));
1826  } else {
1827  state->promise.set_exception(state->first_exc);
1828  }
1829  }
1830  }
1831 };
1832 
1833 template <typename TypeT>
1834 inline Task<void> when_any_runner(Task<TypeT> task, size_t i, WhenAnyGuard<TypeT> guard) {
1835  try {
1836  auto value = co_await std::move(task);
1837  bool expected = false;
1838 
1839  if (guard.state->fired.compare_exchange_strong(expected, true, std::memory_order_acq_rel,
1840  std::memory_order_relaxed)) {
1841  guard.state->winner.emplace(i, std::move(value));
1842  }
1843  } catch (...) {
1844  std::lock_guard lock(guard.state->exc_mtx);
1845 
1846  if (!guard.state->first_exc) {
1847  guard.state->first_exc = std::current_exception();
1848  }
1849  }
1850 
1851  guard.completed_normally = true;
1852 }
1853 
1854 } // namespace detail
1855 
1856 template <typename TypeT>
1857 inline Task<std::vector<TypeT>> when_all(MessageLoop& loop, std::vector<Task<TypeT>> tasks) {
1858  static_assert(std::is_default_constructible_v<TypeT>,
1859  "Co::when_all<T>: T must be default-constructible (results vector pre-sized).");
1860 
1861  if VUNLIKELY (tasks.empty()) {
1862  co_return std::vector<TypeT>{};
1863  }
1864 
1865  const size_t count = tasks.size();
1866  auto state = MemoryResource::make_shared<detail::WhenAllState<TypeT>>();
1867  state->results.resize(count);
1868  state->remaining.store(count, std::memory_order_relaxed);
1869  auto fut = state->promise.get_future();
1870 
1871  for (size_t i = 0; i < count; ++i) {
1872  co_spawn(loop, detail::when_all_runner<TypeT>(std::move(tasks[i]), i, detail::WhenAllGuard<TypeT>{state}));
1873  }
1874 
1875  co_await await_future(loop, std::move(fut));
1876 
1877  co_return std::move(state->results);
1878 }
1879 
1880 template <typename TypeT>
1881 inline Task<std::pair<size_t, TypeT>> when_any(MessageLoop& loop, std::vector<Task<TypeT>> tasks) {
1882  if VUNLIKELY (tasks.empty()) {
1883  throw std::invalid_argument("vlink::Coroutine::when_any: tasks must not be empty");
1884  }
1885 
1886  auto state = MemoryResource::make_shared<detail::WhenAnyState<TypeT>>();
1887  state->remaining.store(tasks.size(), std::memory_order_relaxed);
1888  auto fut = state->promise.get_future();
1889 
1890  for (size_t i = 0; i < tasks.size(); ++i) {
1891  co_spawn(loop, detail::when_any_runner<TypeT>(std::move(tasks[i]), i, detail::WhenAnyGuard<TypeT>{state}));
1892  }
1893 
1894  co_return co_await await_future(loop, std::move(fut));
1895 }
1896 
1897 } // namespace Coroutine
1898 
1899 /**
1900  * @namespace vlink::Co
1901  * @brief Short alias for @c vlink::Coroutine.
1902  *
1903  * @details
1904  * Lets user code write @c vlink::Co::Task / @c vlink::Co::yield(loop) etc.
1905  * @c vlink::Co and @c vlink::Coroutine refer to the same namespace.
1906  */
1907 namespace Co = Coroutine; // NOLINT(readability-identifier-naming)
1908 
1909 } // namespace vlink
1910 
1911 #endif // VLINK_ENABLE_COROUTINE
Thin final wrappers around the standard exception hierarchy used by VLink.
Pool-backed type-erased callables: copyable vlink::Function and move-only vlink::MoveFunction.
Directed acyclic task graph with condition branching, cycle guard and DOT export.
Cross-platform macros for visibility, branch hints, copy prevention, singletons and string helpers.
#define VUNLIKELY(...)
Short alias for VLINK_UNLIKELY.
Definition: macros.h:289
#define VLINK_EXPORT
Definition: macros.h:81
#define VLIKELY(...)
Short alias for VLINK_LIKELY.
Definition: macros.h:284
#define VLINK_DISALLOW_COPY_AND_ASSIGN(classname)
Deletes the copy constructor and copy-assignment operator of classname.
Definition: macros.h:174
PMR adapter that lets standard pmr-aware containers allocate through vlink::MemoryPool.
Single-threaded task dispatcher with three queue backends, timers and scheduling envelopes.
Fluent task-scheduling wrapper used by MessageLoop::exec_task() and family.