. /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   18750 empty() const noexcept 48   18315 empty() const noexcept
49   { 49   {
HITCBC 50   18750 return head_ == nullptr; 50   18315 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   131142 push(continuation& c) noexcept 59   131099 push(continuation& c) noexcept
60   { 60   {
HITCBC 61   131142 c.reserved = nullptr; 61   131099 c.reserved = nullptr;
HITCBC 62   131142 if(tail_) 62   131099 if(tail_)
HITCBC 63   111672 tail_->reserved = &c; 63   112107 tail_->reserved = &c;
64   else 64   else
HITCBC 65   19470 head_ = &c; 65   18992 head_ = &c;
HITCBC 66   131142 tail_ = &c; 66   131099 tail_ = &c;
HITCBC 67   131142 } 67   131099 }
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   32230 take_all() noexcept 103   31795 take_all() noexcept
104   { 104   {
HITCBC 105   32230 taken_batch batch{head_, tail_}; 105   31795 taken_batch batch{head_, tail_};
HITCBC 106   32230 head_ = tail_ = nullptr; 106   31795 head_ = tail_ = nullptr;
HITCBC 107   32230 return batch; 107   31795 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   18746 dispatch_batch(taken_batch& batch) 123   18311 dispatch_batch(taken_batch& batch)
124   { 124   {
HITCBC 125   149094 while(batch.head) 125   148659 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   18746 batch.tail = nullptr; 131   18311 batch.tail = nullptr;
HITCBC 132   18746 } 132   18311 }
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   27832 while(batch.head) 146   27789 while(batch.head)
147   { 147   {
HITCBC 148   790 continuation* c = batch.head; 148   747 continuation* c = batch.head;
HITCBC 149   790 batch.head = static_cast<continuation*>(c->reserved); 149   747 batch.head = static_cast<continuation*>(c->reserved);
HITCBC 150   790 if(c->h) 150   747 if(c->h)
HITCBC 151   790 capy::abandon(*c); 151   747 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