. /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/frame_allocator.hpp> 17   #include <boost/capy/ex/frame_allocator.hpp>
17   18  
  19 + #include <utility>
  20 +
18   namespace boost { 21   namespace boost {
19   namespace capy { 22   namespace capy {
20   namespace detail { 23   namespace detail {
21   24  
22   /** Single-threaded intrusive FIFO of pending continuations. 25   /** Single-threaded intrusive FIFO of pending continuations.
23   26  
24   Links continuations directly through `continuation::reserved` (a 27   Links continuations directly through `continuation::reserved` (a
25   typed `continuation*` round-tripped through the `void*` slot), so 28   typed `continuation*` round-tripped through the `void*` slot), so
26   push() carries no per-item allocation. 29   push() carries no per-item allocation.
27   30  
28   @par Thread Safety 31   @par Thread Safety
29   Not thread-safe. Caller must externally synchronize push() and 32   Not thread-safe. Caller must externally synchronize push() and
30   take_all(). dispatch_batch() does not touch queue state and may 33   take_all(). dispatch_batch() does not touch queue state and may
31   run unlocked once the batch has been taken. 34   run unlocked once the batch has been taken.
32   */ 35   */
33   class strand_queue 36   class strand_queue
34   { 37   {
35   continuation* head_ = nullptr; 38   continuation* head_ = nullptr;
36   continuation* tail_ = nullptr; 39   continuation* tail_ = nullptr;
37   40  
38   public: 41   public:
HITCBC 39   11446 strand_queue() = default; 42   12467 strand_queue() = default;
40   strand_queue(strand_queue const&) = delete; 43   strand_queue(strand_queue const&) = delete;
41   strand_queue& operator=(strand_queue const&) = delete; 44   strand_queue& operator=(strand_queue const&) = delete;
42   45  
43   /** Returns true if there are no pending continuations. */ 46   /** Returns true if there are no pending continuations. */
44   bool 47   bool
HITCBC 45   19987 empty() const noexcept 48   18174 empty() const noexcept
46   { 49   {
HITCBC 47   19987 return head_ == nullptr; 50   18174 return head_ == nullptr;
48   } 51   }
49   52  
50   /** Push a continuation to the queue. 53   /** Push a continuation to the queue.
51   54  
52   @param c The continuation to enqueue; see `continuation` 55   @param c The continuation to enqueue; see `continuation`
53   for lifetime and aliasing requirements. 56   for lifetime and aliasing requirements.
54   */ 57   */
55   void 58   void
HITCBC 56   30342 push(continuation& c) noexcept 59   131157 push(continuation& c) noexcept
57   { 60   {
HITCBC 58   30342 c.reserved = nullptr; 61   131157 c.reserved = nullptr;
HITCBC 59   30342 if(tail_) 62   131157 if(tail_)
HITCBC 60   10355 tail_->reserved = &c; 63   112247 tail_->reserved = &c;
61   else 64   else
HITCBC 62   19987 head_ = &c; 65   18910 head_ = &c;
HITCBC 63   30342 tail_ = &c; 66   131157 tail_ = &c;
HITCBC 64   30342 } 67   131157 }
65   68  
  69 + /// Return the first pending continuation, or null if empty.
  70 + continuation*
HITGNC   71 + 4 front() const noexcept
  72 + {
HITGNC   73 + 4 return head_;
  74 + }
  75 +
  76 + /** Remove the first pending continuation.
  77 +
  78 + @pre The queue is not empty.
  79 + */
  80 + void
HITGNC   81 + 4 pop_front() noexcept
  82 + {
HITGNC   83 + 4 head_ = static_cast<continuation*>(head_->reserved);
HITGNC   84 + 4 if(!head_)
HITGNC   85 + 1 tail_ = nullptr;
HITGNC   86 + 4 }
  87 +
66   /** Batch of taken items for thread-safe dispatch. */ 88   /** Batch of taken items for thread-safe dispatch. */
67   struct taken_batch 89   struct taken_batch
68   { 90   {
69   continuation* head = nullptr; 91   continuation* head = nullptr;
70   continuation* tail = nullptr; 92   continuation* tail = nullptr;
71   }; 93   };
72   94  
73   /** Take all pending items atomically. 95   /** Take all pending items atomically.
74   96  
75   Removes all items from the queue and returns them as a 97   Removes all items from the queue and returns them as a
76   batch. The queue is left empty. 98   batch. The queue is left empty.
77   99  
78   @return The batch of taken items. 100   @return The batch of taken items.
79   */ 101   */
80   taken_batch 102   taken_batch
HITCBC 81   19987 take_all() noexcept 103   31652 take_all() noexcept
82   { 104   {
HITCBC 83   19987 taken_batch batch{head_, tail_}; 105   31652 taken_batch batch{head_, tail_};
HITCBC 84   19987 head_ = tail_ = nullptr; 106   31652 head_ = tail_ = nullptr;
HITCBC 85   19987 return batch; 107   31652 return batch;
86   } 108   }
87   109  
88   /** Resume each continuation in a taken batch. 110   /** Resume each continuation in a taken batch.
89   111  
90   Advances past each node before resuming, since the 112   Advances past each node before resuming, since the
91   resumed coroutine may destroy the awaitable (and thus 113   resumed coroutine may destroy the awaitable (and thus
92   the continuation) before control returns here. 114   the continuation) before control returns here.
93   115  
94   @param batch The batch to dispatch. 116   @param batch The batch to dispatch.
95   117  
96   @note Thread-safe with respect to push() because the queue 118   @note Thread-safe with respect to push() because the queue
97   itself is not touched. 119   itself is not touched.
98   */ 120   */
99   static 121   static
100   void 122   void
HITCBC 101   19987 dispatch_batch(taken_batch& batch) 123   18170 dispatch_batch(taken_batch& batch)
102   { 124   {
HITCBC 103   50329 while(batch.head) 125   148518 while(batch.head)
104   { 126   {
HITCBC 105   30342 continuation* c = batch.head; 127   130348 continuation* c = batch.head;
HITCBC 106   30342 batch.head = static_cast<continuation*>(c->reserved); 128   130348 batch.head = static_cast<continuation*>(c->reserved);
HITCBC 107   30342 safe_resume(c->h); 129   130348 safe_resume(c->h);
  130 + }
HITGNC   131 + 18170 batch.tail = nullptr;
HITGNC   132 + 18170 }
  133 +
  134 + /** Abandon each coroutine in a taken batch without resuming it.
  135 +
  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
  138 + abandon destroys.
  139 +
  140 + @param batch The batch to abandon.
  141 + */
  142 + static
  143 + void
HITGNC   144 + 13519 abandon_batch(taken_batch& batch) noexcept
  145 + {
HITGNC   146 + 27843 while(batch.head)
  147 + {
HITGNC   148 + 805 continuation* c = batch.head;
HITGNC   149 + 805 batch.head = static_cast<continuation*>(c->reserved);
HITGNC   150 + 805 if(c->h)
HITGNC   151 + 805 capy::abandon(*c);
108   } 152   }
HITCBC 109   19987 batch.tail = nullptr; 153   13519 batch.tail = nullptr;
HITCBC 110   19987 } 154   13519 }
111   }; 155   };
112   156  
113   } // namespace detail 157   } // namespace detail
114   } // namespace capy 158   } // namespace capy
115   } // namespace boost 159   } // namespace boost
116   160  
117   #endif 161   #endif