. /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   18754 empty() const noexcept 48   18885 empty() const noexcept
49   { 49   {
HITCBC 50   18754 return head_ == nullptr; 50   18885 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   131383 push(continuation& c) noexcept 59   131273 push(continuation& c) noexcept
60   { 60   {
HITCBC 61   131383 c.reserved = nullptr; 61   131273 c.reserved = nullptr;
HITCBC 62   131383 if(tail_) 62   131273 if(tail_)
HITCBC 63   111667 tail_->reserved = &c; 63   111536 tail_->reserved = &c;
64   else 64   else
HITCBC 65   19716 head_ = &c; 65   19737 head_ = &c;
HITCBC 66   131383 tail_ = &c; 66   131273 tail_ = &c;
HITCBC 67   131383 } 67   131273 }
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   32232 take_all() noexcept 103   32363 take_all() noexcept
104   { 104   {
HITCBC 105   32232 taken_batch batch{head_, tail_}; 105   32363 taken_batch batch{head_, tail_};
HITCBC 106   32232 head_ = tail_ = nullptr; 106   32363 head_ = tail_ = nullptr;
HITCBC 107   32232 return batch; 107   32363 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   18750 dispatch_batch(taken_batch& batch) 123   18881 dispatch_batch(taken_batch& batch)
124   { 124   {
HITCBC 125   149098 while(batch.head) 125   149229 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   18750 batch.tail = nullptr; 131   18881 batch.tail = nullptr;
HITCBC 132   18750 } 132   18881 }
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   28069 while(batch.head) 146   27959 while(batch.head)
147   { 147   {
HITCBC 148   1031 continuation* c = batch.head; 148   921 continuation* c = batch.head;
HITCBC 149   1031 batch.head = static_cast<continuation*>(c->reserved); 149   921 batch.head = static_cast<continuation*>(c->reserved);
HITCBC 150   1031 if(c->h) 150   921 if(c->h)
HITCBC 151   1031 capy::abandon(*c); 151   921 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