. /src /ex /detail /strand_queue.hpp
100.00% Lines (36/36) 100.00% Functions (8/8)
[ ] prev/next file n p prev/next uncovered line
TLA Baseline Branch
Line Hits Code Line Hits Code
1   // 1   //
2   // Copyright (c) 2025 Vinnie Falco (vinnie.falco@gmail.com) 2   // Copyright (c) 2025 Vinnie Falco (vinnie.falco@gmail.com)
3   // Copyright (c) 2026 Michael Vandeberg 3   // Copyright (c) 2026 Michael Vandeberg
4   // 4   //
5   // Distributed under the Boost Software License, Version 1.0. (See accompanying 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) 6   // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
7   // 7   //
8   // Official repository: https://github.com/cppalliance/capy 8   // Official repository: https://github.com/cppalliance/capy
9   // 9   //
10   10  
11   #ifndef BOOST_CAPY_SRC_EX_DETAIL_STRAND_QUEUE_HPP 11   #ifndef BOOST_CAPY_SRC_EX_DETAIL_STRAND_QUEUE_HPP
12   #define BOOST_CAPY_SRC_EX_DETAIL_STRAND_QUEUE_HPP 12   #define BOOST_CAPY_SRC_EX_DETAIL_STRAND_QUEUE_HPP
13   13  
14   #include <boost/capy/continuation.hpp> 14   #include <boost/capy/continuation.hpp>
15   #include <boost/capy/detail/config.hpp> 15   #include <boost/capy/detail/config.hpp>
16   #include <boost/capy/ex/abandon.hpp> 16   #include <boost/capy/ex/abandon.hpp>
17   #include <boost/capy/ex/frame_allocator.hpp> 17   #include <boost/capy/ex/frame_allocator.hpp>
18   18  
19   #include <utility> 19   #include <utility>
20   20  
21   namespace boost { 21   namespace boost {
22   namespace capy { 22   namespace capy {
23   namespace detail { 23   namespace detail {
24   24  
25   /** Single-threaded intrusive FIFO of pending continuations. 25   /** Single-threaded intrusive FIFO of pending continuations.
26   26  
27   Links continuations directly through `continuation::reserved` (a 27   Links continuations directly through `continuation::reserved` (a
28   typed `continuation*` round-tripped through the `void*` slot), so 28   typed `continuation*` round-tripped through the `void*` slot), so
29   push() carries no per-item allocation. 29   push() carries no per-item allocation.
30   30  
31   @par Thread Safety 31   @par Thread Safety
32   Not thread-safe. Caller must externally synchronize push() and 32   Not thread-safe. Caller must externally synchronize push() and
33   take_all(). dispatch_batch() does not touch queue state and may 33   take_all(). dispatch_batch() does not touch queue state and may
34   run unlocked once the batch has been taken. 34   run unlocked once the batch has been taken.
35   */ 35   */
36   class strand_queue 36   class strand_queue
37   { 37   {
38   continuation* head_ = nullptr; 38   continuation* head_ = nullptr;
39   continuation* tail_ = nullptr; 39   continuation* tail_ = nullptr;
40   40  
41   public: 41   public:
HITCBC 42   12468 strand_queue() = default; 42   12468 strand_queue() = default;
43   strand_queue(strand_queue const&) = delete; 43   strand_queue(strand_queue const&) = delete;
44   strand_queue& operator=(strand_queue const&) = delete; 44   strand_queue& operator=(strand_queue const&) = delete;
45   45  
46   /** Returns true if there are no pending continuations. */ 46   /** Returns true if there are no pending continuations. */
47   bool 47   bool
HITCBC 48   15416 empty() const noexcept 48   16134 empty() const noexcept
49   { 49   {
HITCBC 50   15416 return head_ == nullptr; 50   16134 return head_ == nullptr;
51   } 51   }
52   52  
53   /** Push a continuation to the queue. 53   /** Push a continuation to the queue.
54   54  
55   @param c The continuation to enqueue; see `continuation` 55   @param c The continuation to enqueue; see `continuation`
56   for lifetime and aliasing requirements. 56   for lifetime and aliasing requirements.
57   */ 57   */
58   void 58   void
HITCBC 59   131198 push(continuation& c) noexcept 59   131412 push(continuation& c) noexcept
60   { 60   {
HITCBC 61   131198 c.reserved = nullptr; 61   131412 c.reserved = nullptr;
HITCBC 62   131198 if(tail_) 62   131412 if(tail_)
HITCBC 63   115006 tail_->reserved = &c; 63   114288 tail_->reserved = &c;
64   else 64   else
HITCBC 65   16192 head_ = &c; 65   17124 head_ = &c;
HITCBC 66   131198 tail_ = &c; 66   131412 tail_ = &c;
HITCBC 67   131198 } 67   131412 }
68   68  
69   /// Return the first pending continuation, or null if empty. 69   /// Return the first pending continuation, or null if empty.
70   continuation* 70   continuation*
HITCBC 71   4 front() const noexcept 71   4 front() const noexcept
72   { 72   {
HITCBC 73   4 return head_; 73   4 return head_;
74   } 74   }
75   75  
76   /** Remove the first pending continuation. 76   /** Remove the first pending continuation.
77   77  
78   @pre The queue is not empty. 78   @pre The queue is not empty.
79   */ 79   */
80   void 80   void
HITCBC 81   4 pop_front() noexcept 81   4 pop_front() noexcept
82   { 82   {
HITCBC 83   4 head_ = static_cast<continuation*>(head_->reserved); 83   4 head_ = static_cast<continuation*>(head_->reserved);
HITCBC 84   4 if(!head_) 84   4 if(!head_)
HITCBC 85   1 tail_ = nullptr; 85   1 tail_ = nullptr;
HITCBC 86   4 } 86   4 }
87   87  
88   /** Batch of taken items for thread-safe dispatch. */ 88   /** Batch of taken items for thread-safe dispatch. */
89   struct taken_batch 89   struct taken_batch
90   { 90   {
91   continuation* head = nullptr; 91   continuation* head = nullptr;
92   continuation* tail = nullptr; 92   continuation* tail = nullptr;
93   }; 93   };
94   94  
95   /** Take all pending items atomically. 95   /** Take all pending items atomically.
96   96  
97   Removes all items from the queue and returns them as a 97   Removes all items from the queue and returns them as a
98   batch. The queue is left empty. 98   batch. The queue is left empty.
99   99  
100   @return The batch of taken items. 100   @return The batch of taken items.
101   */ 101   */
102   taken_batch 102   taken_batch
HITCBC 103   28896 take_all() noexcept 103   29614 take_all() noexcept
104   { 104   {
HITCBC 105   28896 taken_batch batch{head_, tail_}; 105   29614 taken_batch batch{head_, tail_};
HITCBC 106   28896 head_ = tail_ = nullptr; 106   29614 head_ = tail_ = nullptr;
HITCBC 107   28896 return batch; 107   29614 return batch;
108   } 108   }
109   109  
110   /** Resume each continuation in a taken batch. 110   /** Resume each continuation in a taken batch.
111   111  
112   Advances past each node before resuming, since the 112   Advances past each node before resuming, since the
113   resumed coroutine may destroy the awaitable (and thus 113   resumed coroutine may destroy the awaitable (and thus
114   the continuation) before control returns here. 114   the continuation) before control returns here.
115   115  
116   @param batch The batch to dispatch. 116   @param batch The batch to dispatch.
117   117  
118   @note Thread-safe with respect to push() because the queue 118   @note Thread-safe with respect to push() because the queue
119   itself is not touched. 119   itself is not touched.
120   */ 120   */
121   static 121   static
122   void 122   void
HITCBC 123   15412 dispatch_batch(taken_batch& batch) 123   16130 dispatch_batch(taken_batch& batch)
124   { 124   {
HITCBC 125   145760 while(batch.head) 125   146478 while(batch.head)
126   { 126   {
HITCBC 127   130348 continuation* c = batch.head; 127   130348 continuation* c = batch.head;
HITCBC 128   130348 batch.head = static_cast<continuation*>(c->reserved); 128   130348 batch.head = static_cast<continuation*>(c->reserved);
HITCBC 129   130348 safe_resume(c->h); 129   130348 safe_resume(c->h);
130   } 130   }
HITCBC 131   15412 batch.tail = nullptr; 131   16130 batch.tail = nullptr;
HITCBC 132   15412 } 132   16130 }
133   133  
134   /** Abandon each coroutine in a taken batch without resuming it. 134   /** Abandon each coroutine in a taken batch without resuming it.
135   135  
136   Used when the strand can no longer run them. Advances past 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 137   each node first, since the node may live in a frame the
138   abandon destroys. 138   abandon destroys.
139   139  
140   @param batch The batch to abandon. 140   @param batch The batch to abandon.
141   */ 141   */
142   static 142   static
143   void 143   void
HITCBC 144   13521 abandon_batch(taken_batch& batch) noexcept 144   13521 abandon_batch(taken_batch& batch) noexcept
145   { 145   {
HITCBC 146   27888 while(batch.head) 146   28102 while(batch.head)
147   { 147   {
HITCBC 148   846 continuation* c = batch.head; 148   1060 continuation* c = batch.head;
HITCBC 149   846 batch.head = static_cast<continuation*>(c->reserved); 149   1060 batch.head = static_cast<continuation*>(c->reserved);
HITCBC 150   846 if(c->h) 150   1060 if(c->h)
HITCBC 151   846 capy::abandon(*c); 151   1060 capy::abandon(*c);
152   } 152   }
HITCBC 153   13521 batch.tail = nullptr; 153   13521 batch.tail = nullptr;
HITCBC 154   13521 } 154   13521 }
155   }; 155   };
156   156  
157   } // namespace detail 157   } // namespace detail
158   } // namespace capy 158   } // namespace capy
159   } // namespace boost 159   } // namespace boost
160   160  
161   #endif 161   #endif