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