src/ex/detail/strand_queue.hpp

100.0% Lines (36 / 36) 100.0% Functions (8 / 8)

[ ] prev/next file

strand_queue.hpp

f(x) Functions (8)

Function Calls Lines Blocks

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