TLA Line data 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/abandon.hpp>
17 : #include <boost/capy/ex/frame_allocator.hpp>
18 :
19 : #include <utility>
20 :
21 : namespace boost {
22 : namespace capy {
23 : namespace detail {
24 :
25 : /** Single-threaded intrusive FIFO of pending continuations.
26 :
27 : Links continuations directly through `continuation::reserved` (a
28 : typed `continuation*` round-tripped through the `void*` slot), so
29 : push() carries no per-item allocation.
30 :
31 : @par Thread Safety
32 : Not thread-safe. Caller must externally synchronize push() and
33 : take_all(). dispatch_batch() does not touch queue state and may
34 : run unlocked once the batch has been taken.
35 : */
36 : class strand_queue
37 : {
38 : continuation* head_ = nullptr;
39 : continuation* tail_ = nullptr;
40 :
41 : public:
42 HIT 12467 : strand_queue() = default;
43 : strand_queue(strand_queue const&) = delete;
44 : strand_queue& operator=(strand_queue const&) = delete;
45 :
46 : /** Returns true if there are no pending continuations. */
47 : bool
48 19662 : empty() const noexcept
49 : {
50 19662 : return head_ == nullptr;
51 : }
52 :
53 : /** Push a continuation to the queue.
54 :
55 : @param c The continuation to enqueue; see `continuation`
56 : for lifetime and aliasing requirements.
57 : */
58 : void
59 131271 : push(continuation& c) noexcept
60 : {
61 131271 : c.reserved = nullptr;
62 131271 : if(tail_)
63 110759 : tail_->reserved = &c;
64 : else
65 20512 : head_ = &c;
66 131271 : tail_ = &c;
67 131271 : }
68 :
69 : /// Return the first pending continuation, or null if empty.
70 : continuation*
71 4 : front() const noexcept
72 : {
73 4 : return head_;
74 : }
75 :
76 : /** Remove the first pending continuation.
77 :
78 : @pre The queue is not empty.
79 : */
80 : void
81 4 : pop_front() noexcept
82 : {
83 4 : head_ = static_cast<continuation*>(head_->reserved);
84 4 : if(!head_)
85 1 : tail_ = nullptr;
86 4 : }
87 :
88 : /** Batch of taken items for thread-safe dispatch. */
89 : struct taken_batch
90 : {
91 : continuation* head = nullptr;
92 : continuation* tail = nullptr;
93 : };
94 :
95 : /** Take all pending items atomically.
96 :
97 : Removes all items from the queue and returns them as a
98 : batch. The queue is left empty.
99 :
100 : @return The batch of taken items.
101 : */
102 : taken_batch
103 33140 : take_all() noexcept
104 : {
105 33140 : taken_batch batch{head_, tail_};
106 33140 : head_ = tail_ = nullptr;
107 33140 : return batch;
108 : }
109 :
110 : /** Resume each continuation in a taken batch.
111 :
112 : Advances past each node before resuming, since the
113 : resumed coroutine may destroy the awaitable (and thus
114 : the continuation) before control returns here.
115 :
116 : @param batch The batch to dispatch.
117 :
118 : @note Thread-safe with respect to push() because the queue
119 : itself is not touched.
120 : */
121 : static
122 : void
123 19658 : dispatch_batch(taken_batch& batch)
124 : {
125 150006 : while(batch.head)
126 : {
127 130348 : continuation* c = batch.head;
128 130348 : batch.head = static_cast<continuation*>(c->reserved);
129 130348 : safe_resume(c->h);
130 : }
131 19658 : batch.tail = nullptr;
132 19658 : }
133 :
134 : /** Abandon each coroutine in a taken batch without resuming it.
135 :
136 : Used when the strand can no longer run them. Advances past
137 : each node first, since the node may live in a frame the
138 : abandon destroys.
139 :
140 : @param batch The batch to abandon.
141 : */
142 : static
143 : void
144 13519 : abandon_batch(taken_batch& batch) noexcept
145 : {
146 27957 : while(batch.head)
147 : {
148 919 : continuation* c = batch.head;
149 919 : batch.head = static_cast<continuation*>(c->reserved);
150 919 : if(c->h)
151 919 : capy::abandon(*c);
152 : }
153 13519 : batch.tail = nullptr;
154 13519 : }
155 : };
156 :
157 : } // namespace detail
158 : } // namespace capy
159 : } // namespace boost
160 :
161 : #endif
|
