Index/src / ex / detail / strand_queue.hpp

src/ex/detail/strand_queue.hpp

100.0% Lines (21/21) 100.0% List of functions (5/5)
[ ] prev/next file
strand_queue.hpp
f(x) Functions (5)
Function Calls Lines Blocks
boost::capy::detail::strand_queue::strand_queue() :39 11443x 100.0% 100.0% boost::capy::detail::strand_queue::empty() const :45 20107x 100.0% 100.0% boost::capy::detail::strand_queue::push(boost::capy::continuation&) :56 30340x 100.0% 100.0% boost::capy::detail::strand_queue::take_all() :81 20107x 100.0% 100.0% boost::capy::detail::strand_queue::dispatch_batch(boost::capy::detail::strand_queue::taken_batch&) :101 20107x 100.0% 100.0%
Line TLA Hits Source Code
1 //
2 // Copyright (c) 2025 Vinnie Falco (vinnie.falco@gmail.com)
3 // Copyright (c) 2026 Michael Vandeberg
4 //
5 // Distributed under the Boost Software License, Version 1.0. (See accompanying
6 // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
7 //
8 // Official repository: https://github.com/cppalliance/capy
9 //
10
11 #ifndef BOOST_CAPY_SRC_EX_DETAIL_STRAND_QUEUE_HPP
12 #define BOOST_CAPY_SRC_EX_DETAIL_STRAND_QUEUE_HPP
13
14 #include <boost/capy/continuation.hpp>
15 #include <boost/capy/detail/config.hpp>
16 #include <boost/capy/ex/frame_allocator.hpp>
17
18 namespace boost {
19 namespace capy {
20 namespace detail {
21
22 /** Single-threaded intrusive FIFO of pending continuations.
23
24 Links continuations directly through `continuation::reserved` (a
25 typed `continuation*` round-tripped through the `void*` slot), so
26 push() carries no per-item allocation.
27
28 @par Thread Safety
29 Not thread-safe. Caller must externally synchronize push() and
30 take_all(). dispatch_batch() does not touch queue state and may
31 run unlocked once the batch has been taken.
32 */
33 class strand_queue
34 {
35 continuation* head_ = nullptr;
36 continuation* tail_ = nullptr;
37
38 public:
39 11443x strand_queue() = default;
40 strand_queue(strand_queue const&) = delete;
41 strand_queue& operator=(strand_queue const&) = delete;
42
43 /** Returns true if there are no pending continuations. */
44 bool
45 20107x empty() const noexcept
46 {
47 20107x return head_ == nullptr;
48 }
49
50 /** Push a continuation to the queue.
51
52 @param c The continuation to enqueue; see `continuation`
53 for lifetime and aliasing requirements.
54 */
55 void
56 30340x push(continuation& c) noexcept
57 {
58 30340x c.reserved = nullptr;
59 30340x if(tail_)
60 10233x tail_->reserved = &c;
61 else
62 20107x head_ = &c;
63 30340x tail_ = &c;
64 30340x }
65
66 /** Batch of taken items for thread-safe dispatch. */
67 struct taken_batch
68 {
69 continuation* head = nullptr;
70 continuation* tail = nullptr;
71 };
72
73 /** Take all pending items atomically.
74
75 Removes all items from the queue and returns them as a
76 batch. The queue is left empty.
77
78 @return The batch of taken items.
79 */
80 taken_batch
81 20107x take_all() noexcept
82 {
83 20107x taken_batch batch{head_, tail_};
84 20107x head_ = tail_ = nullptr;
85 20107x return batch;
86 }
87
88 /** Resume each continuation in a taken batch.
89
90 Advances past each node before resuming, since the
91 resumed coroutine may destroy the awaitable (and thus
92 the continuation) before control returns here.
93
94 @param batch The batch to dispatch.
95
96 @note Thread-safe with respect to push() because the queue
97 itself is not touched.
98 */
99 static
100 void
101 20107x dispatch_batch(taken_batch& batch)
102 {
103 50447x while(batch.head)
104 {
105 30340x continuation* c = batch.head;
106 30340x batch.head = static_cast<continuation*>(c->reserved);
107 30340x safe_resume(c->h);
108 }
109 20107x batch.tail = nullptr;
110 20107x }
111 };
112
113 } // namespace detail
114 } // namespace capy
115 } // namespace boost
116
117 #endif
118