. /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   12472 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   18537 empty() const noexcept 48   18151 empty() const noexcept
49   { 49   {
HITCBC 50   18537 return head_ == nullptr; 50   18151 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   131180 push(continuation& c) noexcept 59   131193 push(continuation& c) noexcept
60   { 60   {
HITCBC 61   131180 c.reserved = nullptr; 61   131193 c.reserved = nullptr;
HITCBC 62   131180 if(tail_) 62   131193 if(tail_)
HITCBC 63   111885 tail_->reserved = &c; 63   112283 tail_->reserved = &c;
64   else 64   else
HITCBC 65   19295 head_ = &c; 65   18910 head_ = &c;
HITCBC 66   131180 tail_ = &c; 66   131193 tail_ = &c;
HITCBC 67   131180 } 67   131193 }
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   32017 take_all() noexcept 103   31635 take_all() noexcept
104   { 104   {
HITCBC 105   32017 taken_batch batch{head_, tail_}; 105   31635 taken_batch batch{head_, tail_};
HITCBC 106   32017 head_ = tail_ = nullptr; 106   31635 head_ = tail_ = nullptr;
HITCBC 107   32017 return batch; 107   31635 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   18533 dispatch_batch(taken_batch& batch) 123   18147 dispatch_batch(taken_batch& batch)
124   { 124   {
HITCBC 125   148881 while(batch.head) 125   148507 while(batch.head)
126   { 126   {
HITCBC 127   130348 continuation* c = batch.head; 127   130360 continuation* c = batch.head;
HITCBC 128   130348 batch.head = static_cast<continuation*>(c->reserved); 128   130360 batch.head = static_cast<continuation*>(c->reserved);
HITCBC 129   130348 safe_resume(c->h); 129   130360 safe_resume(c->h);
130   } 130   }
HITCBC 131   18533 batch.tail = nullptr; 131   18147 batch.tail = nullptr;
HITCBC 132   18533 } 132   18147 }
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   13526 abandon_batch(taken_batch& batch) noexcept
145   { 145   {
HITCBC 146   27870 while(batch.head) 146   27881 while(batch.head)
147   { 147   {
HITCBC 148   828 continuation* c = batch.head; 148   829 continuation* c = batch.head;
HITCBC 149   828 batch.head = static_cast<continuation*>(c->reserved); 149   829 batch.head = static_cast<continuation*>(c->reserved);
HITCBC 150   828 if(c->h) 150   829 if(c->h)
HITCBC 151   828 capy::abandon(*c); 151   829 capy::abandon(*c);
152   } 152   }
HITCBC 153   13521 batch.tail = nullptr; 153   13526 batch.tail = nullptr;
HITCBC 154   13521 } 154   13526 }
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