. /src /ex /detail /strand_queue.hpp
100.00% Lines (21/21) 100.00% Functions (5/5)
[ ] 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/frame_allocator.hpp> 16   #include <boost/capy/ex/frame_allocator.hpp>
17   17  
18   namespace boost { 18   namespace boost {
19   namespace capy { 19   namespace capy {
20   namespace detail { 20   namespace detail {
21   21  
22   /** Single-threaded intrusive FIFO of pending continuations. 22   /** Single-threaded intrusive FIFO of pending continuations.
23   23  
24   Links continuations directly through `continuation::reserved` (a 24   Links continuations directly through `continuation::reserved` (a
25   typed `continuation*` round-tripped through the `void*` slot), so 25   typed `continuation*` round-tripped through the `void*` slot), so
26   push() carries no per-item allocation. 26   push() carries no per-item allocation.
27   27  
28   @par Thread Safety 28   @par Thread Safety
29   Not thread-safe. Caller must externally synchronize push() and 29   Not thread-safe. Caller must externally synchronize push() and
30   take_all(). dispatch_batch() does not touch queue state and may 30   take_all(). dispatch_batch() does not touch queue state and may
31   run unlocked once the batch has been taken. 31   run unlocked once the batch has been taken.
32   */ 32   */
33   class strand_queue 33   class strand_queue
34   { 34   {
35   continuation* head_ = nullptr; 35   continuation* head_ = nullptr;
36   continuation* tail_ = nullptr; 36   continuation* tail_ = nullptr;
37   37  
38   public: 38   public:
HITCBC 39   11446 strand_queue() = default; 39   11446 strand_queue() = default;
40   strand_queue(strand_queue const&) = delete; 40   strand_queue(strand_queue const&) = delete;
41   strand_queue& operator=(strand_queue const&) = delete; 41   strand_queue& operator=(strand_queue const&) = delete;
42   42  
43   /** Returns true if there are no pending continuations. */ 43   /** Returns true if there are no pending continuations. */
44   bool 44   bool
HITCBC 45   17194 empty() const noexcept 45   17360 empty() const noexcept
46   { 46   {
HITCBC 47   17194 return head_ == nullptr; 47   17360 return head_ == nullptr;
48   } 48   }
49   49  
50   /** Push a continuation to the queue. 50   /** Push a continuation to the queue.
51   51  
52   @param c The continuation to enqueue; see `continuation` 52   @param c The continuation to enqueue; see `continuation`
53   for lifetime and aliasing requirements. 53   for lifetime and aliasing requirements.
54   */ 54   */
55   void 55   void
HITCBC 56   30342 push(continuation& c) noexcept 56   30342 push(continuation& c) noexcept
57   { 57   {
HITCBC 58   30342 c.reserved = nullptr; 58   30342 c.reserved = nullptr;
HITCBC 59   30342 if(tail_) 59   30342 if(tail_)
HITCBC 60   13148 tail_->reserved = &c; 60   12982 tail_->reserved = &c;
61   else 61   else
HITCBC 62   17194 head_ = &c; 62   17360 head_ = &c;
HITCBC 63   30342 tail_ = &c; 63   30342 tail_ = &c;
HITCBC 64   30342 } 64   30342 }
65   65  
66   /** Batch of taken items for thread-safe dispatch. */ 66   /** Batch of taken items for thread-safe dispatch. */
67   struct taken_batch 67   struct taken_batch
68   { 68   {
69   continuation* head = nullptr; 69   continuation* head = nullptr;
70   continuation* tail = nullptr; 70   continuation* tail = nullptr;
71   }; 71   };
72   72  
73   /** Take all pending items atomically. 73   /** Take all pending items atomically.
74   74  
75   Removes all items from the queue and returns them as a 75   Removes all items from the queue and returns them as a
76   batch. The queue is left empty. 76   batch. The queue is left empty.
77   77  
78   @return The batch of taken items. 78   @return The batch of taken items.
79   */ 79   */
80   taken_batch 80   taken_batch
HITCBC 81   17194 take_all() noexcept 81   17360 take_all() noexcept
82   { 82   {
HITCBC 83   17194 taken_batch batch{head_, tail_}; 83   17360 taken_batch batch{head_, tail_};
HITCBC 84   17194 head_ = tail_ = nullptr; 84   17360 head_ = tail_ = nullptr;
HITCBC 85   17194 return batch; 85   17360 return batch;
86   } 86   }
87   87  
88   /** Resume each continuation in a taken batch. 88   /** Resume each continuation in a taken batch.
89   89  
90   Advances past each node before resuming, since the 90   Advances past each node before resuming, since the
91   resumed coroutine may destroy the awaitable (and thus 91   resumed coroutine may destroy the awaitable (and thus
92   the continuation) before control returns here. 92   the continuation) before control returns here.
93   93  
94   @param batch The batch to dispatch. 94   @param batch The batch to dispatch.
95   95  
96   @note Thread-safe with respect to push() because the queue 96   @note Thread-safe with respect to push() because the queue
97   itself is not touched. 97   itself is not touched.
98   */ 98   */
99   static 99   static
100   void 100   void
HITCBC 101   17194 dispatch_batch(taken_batch& batch) 101   17360 dispatch_batch(taken_batch& batch)
102   { 102   {
HITCBC 103   47536 while(batch.head) 103   47702 while(batch.head)
104   { 104   {
HITCBC 105   30342 continuation* c = batch.head; 105   30342 continuation* c = batch.head;
HITCBC 106   30342 batch.head = static_cast<continuation*>(c->reserved); 106   30342 batch.head = static_cast<continuation*>(c->reserved);
HITCBC 107   30342 safe_resume(c->h); 107   30342 safe_resume(c->h);
108   } 108   }
HITCBC 109   17194 batch.tail = nullptr; 109   17360 batch.tail = nullptr;
HITCBC 110   17194 } 110   17360 }
111   }; 111   };
112   112  
113   } // namespace detail 113   } // namespace detail
114   } // namespace capy 114   } // namespace capy
115   } // namespace boost 115   } // namespace boost
116   116  
117   #endif 117   #endif