. /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   12467 strand_queue() = default; 42   12467 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   19359 empty() const noexcept 48   19662 empty() const noexcept
49   { 49   {
HITCBC 50   19359 return head_ == nullptr; 50   19662 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   131285 push(continuation& c) noexcept 59   131271 push(continuation& c) noexcept
60   { 60   {
HITCBC 61   131285 c.reserved = nullptr; 61   131271 c.reserved = nullptr;
HITCBC 62   131285 if(tail_) 62   131271 if(tail_)
HITCBC 63   111062 tail_->reserved = &c; 63   110759 tail_->reserved = &c;
64   else 64   else
HITCBC 65   20223 head_ = &c; 65   20512 head_ = &c;
HITCBC 66   131285 tail_ = &c; 66   131271 tail_ = &c;
HITCBC 67   131285 } 67   131271 }
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   32837 take_all() noexcept 103   33140 take_all() noexcept
104   { 104   {
HITCBC 105   32837 taken_batch batch{head_, tail_}; 105   33140 taken_batch batch{head_, tail_};
HITCBC 106   32837 head_ = tail_ = nullptr; 106   33140 head_ = tail_ = nullptr;
HITCBC 107   32837 return batch; 107   33140 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   19355 dispatch_batch(taken_batch& batch) 123   19658 dispatch_batch(taken_batch& batch)
124   { 124   {
HITCBC 125   149703 while(batch.head) 125   150006 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   19355 batch.tail = nullptr; 131   19658 batch.tail = nullptr;
HITCBC 132   19355 } 132   19658 }
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   13519 abandon_batch(taken_batch& batch) noexcept 144   13519 abandon_batch(taken_batch& batch) noexcept
145   { 145   {
HITCBC 146   27971 while(batch.head) 146   27957 while(batch.head)
147   { 147   {
HITCBC 148   933 continuation* c = batch.head; 148   919 continuation* c = batch.head;
HITCBC 149   933 batch.head = static_cast<continuation*>(c->reserved); 149   919 batch.head = static_cast<continuation*>(c->reserved);
HITCBC 150   933 if(c->h) 150   919 if(c->h)
HITCBC 151   933 capy::abandon(*c); 151   919 capy::abandon(*c);
152   } 152   }
HITCBC 153   13519 batch.tail = nullptr; 153   13519 batch.tail = nullptr;
HITCBC 154   13519 } 154   13519 }
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