src/ex/detail/strand_service.cpp

100.0% Lines (172 / 172, 1 excl) 100.0% Functions (26 / 26, 1 excl)

[ ] prev/next file

strand_service.cpp

f(x) Functions (27)

Function Calls Lines Blocks

Line	Hits	Source Code
1		//
2		// Copyright (c) 2025 Vinnie Falco (vinnie.falco@gmail.com)
3		//
4		// Distributed under the Boost Software License, Version 1.0. (See accompanying
5		// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
6		//
7		// Official repository: https://github.com/cppalliance/capy
8		//
9
10		#include "src/ex/detail/strand_impl.hpp"
11		#include <boost/capy/ex/detail/strand_service.hpp>
12		#include <boost/capy/continuation.hpp>
13		#include <boost/capy/detail/config.hpp>
14		#include <coroutine>
15		#include <memory>
16		#include <tuple>
17		#include <utility>
18
19		namespace boost {
20		namespace capy {
21		namespace detail {
22
23		// Sentinel stored in invoker_frame_cache_ after shutdown to prevent
24		// in-flight invokers from repopulating a freed cache slot.
25		inline void* const kCacheClosed = reinterpret_cast<void*>(1);
26
27		/** Concrete strand_service.
28
29		Holds a shared mutex pool (193 entries), a linked list of live
30		impls (for shutdown traversal), and a single-slot invoker
31		coroutine frame cache shared across all strands of this service.
32
33		The dispatch helpers (`enqueue`, `dispatch_pending`, etc.) are
34		public so the namespace-scope `make_invoker` coroutine and the
35		`strand_service` static methods can call them without friendship.
36		*/
37		class strand_service_impl : public strand_service
38		{
39		public:
40		static constexpr std::size_t num_mutexes = 193;
41
42		std::mutex mutex_;
43		std::size_t salt_ = 0;
44		std::shared_ptr<std::mutex> mutexes_[num_mutexes];
45		intrusive_list<strand_impl> impl_list_;
46		std::atomic<void*> invoker_frame_cache_{nullptr};
47		bool shutdown_ = false;
48
49		explicit
50	1051x	strand_service_impl(execution_context&)
51	1051x	{
52	1051x	}
53
54		std::shared_ptr<strand_impl>
55	12467x	create_implementation() override
56		{
57	12467x	auto new_impl = std::make_shared<strand_impl>();
58
59	12467x	std::lock_guard<std::mutex> lock(mutex_);
60	12467x	if(shutdown_)
61		{
62		// Born closed, like the strands shutdown() closed: it
63		// borrows nothing from this service.
64	2x	new_impl->locked_ = true;
65	2x	return new_impl;
66		}
67
68	12465x	std::size_t s = salt_++;
69	12465x	std::size_t idx = reinterpret_cast<std::size_t>(new_impl.get());
70	12465x	idx += idx >> 3;
71	12465x	idx ^= s + 0x9e3779b9 + (idx << 6) + (idx >> 2);
72	12465x	idx %= num_mutexes;
73	12465x	if(!mutexes_[idx])
74	1684x	mutexes_[idx] = std::make_shared<std::mutex>();
75	12465x	new_impl->mutex_ = mutexes_[idx].get();
76
77	12465x	impl_list_.push_back(new_impl.get());
78	12465x	new_impl->service_.store(this, std::memory_order_release);
79
80	12465x	return new_impl;
81	12467x	}
82
83		// Returns the service when the caller must post an invoker.
84		static strand_service_impl*
85	331439x	enqueue(strand_impl& impl, continuation& c)
86		{
87		// A closed strand never locks: the mutex it borrowed may have
88		// gone with its service.
89	331439x	if(impl.service_.load(std::memory_order_acquire))
90		{
91	131373x	std::lock_guard<std::mutex> lock(*impl.mutex_);
92	131373x	if(auto* svc = impl.service_.load(std::memory_order_relaxed))
93		{
94	131273x	impl.pending_.push(c);
95	131273x	if(!impl.locked_)
96		{
97	18235x	impl.locked_ = true;
98	18235x	return svc;
99		}
100	113038x	return nullptr;
101		}
102	131373x	}
103		// Shut down: nothing will run it, so abandon it as its holder.
104	200166x	capy::abandon(c);
105	200166x	return nullptr;
106		}
107
108		enum class withdrawal
109		{
110		alone,
111		followed,
112		taken
113		};
114
115		// Takes back a continuation whose invoker could not be posted.
116		// `taken` means a shutdown took it first and abandoned it.
117		static withdrawal
118	5x	withdraw(strand_impl& impl, continuation& c) noexcept
119		{
120	5x	std::lock_guard<std::mutex> lock(*impl.mutex_);
121	5x	if(!impl.service_.load(std::memory_order_relaxed))
122	1x	return withdrawal::taken;
123		// Nothing takes from the queue while it waits for an invoker,
124		// so `c` is still first.
125	4x	BOOST_CAPY_ASSERT(impl.pending_.front() == &c);
126	4x	std::ignore = c;
127	4x	impl.pending_.pop_front();
128		// Posts queued behind `c` returned trusting its invoker, so
129		// the strand stays locked until one runs them.
130	4x	if(!impl.pending_.empty())
131	3x	return withdrawal::followed;
132	1x	impl.locked_ = false;
133	1x	return withdrawal::alone;
134	5x	}
135
136		// Gives up queued work that no invoker can be posted to run.
137		static void
138	2x	abandon_pending(strand_impl& impl) noexcept
139		{
140	2x	strand_queue::taken_batch batch;
141		{
142	2x	std::lock_guard<std::mutex> lock(*impl.mutex_);
143	2x	if(!impl.service_.load(std::memory_order_relaxed))
144	1x	return;
145	1x	batch = impl.pending_.take_all();
146	1x	impl.locked_ = false;
147	2x	}
148	1x	strand_queue::abandon_batch(batch);
149		}
150
151		static void
152	18881x	dispatch_pending(strand_impl& impl)
153		{
154	18881x	strand_queue::taken_batch batch;
155		{
156	18881x	std::lock_guard<std::mutex> lock(*impl.mutex_);
157	18881x	batch = impl.pending_.take_all();
158	18881x	}
159	18881x	impl.pending_.dispatch_batch(batch);
160	18881x	}
161
162		static bool
163	18881x	try_unlock(strand_impl& impl)
164		{
165	18881x	std::lock_guard<std::mutex> lock(*impl.mutex_);
166	18881x	if(impl.pending_.empty())
167		{
168	17379x	impl.locked_ = false;
169	17379x	return true;
170		}
171	1502x	return false;
172	18881x	}
173
174		static void
175	18881x	set_dispatch_thread(strand_impl& impl) noexcept
176		{
177	18881x	impl.dispatch_thread_.store(std::this_thread::get_id());
178	18881x	}
179
180		static void
181	17379x	clear_dispatch_thread(strand_impl& impl) noexcept
182		{
183	17379x	impl.dispatch_thread_.store(std::thread::id{});
184	17379x	}
185
186		// Defined below; needs strand_invoker complete.
187		static void
188		post_invoker(
189		strand_service_impl* svc,
190		std::shared_ptr<strand_impl> impl,
191		executor_ref ex);
192
193		protected:
194		void
195	1051x	shutdown() override
196		{
197	1051x	strand_queue::taken_batch doomed;
198		{
199	1051x	std::lock_guard<std::mutex> lock(mutex_);
200	1051x	shutdown_ = true;
201	2065x	while(auto* p = impl_list_.pop_front())
202		{
203	1014x	std::lock_guard<std::mutex> impl_lock(*p->mutex_);
204	1014x	p->locked_ = true;
205	1014x	p->service_.store(nullptr, std::memory_order_release);
206	1014x	auto b = p->pending_.take_all();
207	1014x	if(b.head)
208		{
209	851x	if(doomed.tail)
210	1x	doomed.tail->reserved = b.head;
211		else
212	850x	doomed.head = b.head;
213	851x	doomed.tail = b.tail;
214		}
215	1014x	}
216	1051x	}
217		// Outside the mutexes: a destroyed frame may drop the last
218		// strand handle and re-enter the service.
219	1051x	strand_queue::abandon_batch(doomed);
220
221	1051x	void* fp = invoker_frame_cache_.exchange(
222		kCacheClosed, std::memory_order_acq_rel);
223	1051x	if(fp) ::operator delete(fp);
224	1051x	}
225		};
226
227		/** Invoker coroutine that drains a strand's pending queue.
228
229		Runs once the strand transitions from unlocked to locked. Holds
230		the impl alive via the coroutine parameter (a shared_ptr in the
231		coroutine frame), so user code may drop its strand handle while
232		the invoker is mid-flight.
233
234		The frame's allocator recycles a single per-service slot. The
235		trailer points at the service (lifetime: execution_context),
236		NOT the impl (lifetime: per-strand), so operator delete is
237		safe even after the impl has been destroyed.
238		*/
239		struct strand_invoker
240		{
241		struct promise_type
242		{
243		// Stored in the coroutine frame so its address is stable for
244		// posting to the inner executor.
245		continuation self_;
246
247		// The service is the one enqueue() saw under the strand's lock;
248		// the impl's own pointer may be cleared by a concurrent shutdown.
249		void*
250	18238x	operator new(
251		std::size_t n,
252		strand_service_impl* svc,
253		std::shared_ptr<strand_impl> const&)
254		{
255	18238x	constexpr auto A = alignof(strand_service_impl*);
256	18238x	std::size_t padded = (n + A - 1) & ~(A - 1);
257	18238x	std::size_t total = padded + sizeof(strand_service_impl*);
258
259		// Never takes the closed marker, which would let a later
260		// delete park a frame that nothing frees.
261	18238x	void* p = svc->invoker_frame_cache_.load(
262	18238x	std::memory_order_acquire);
263	28603x	while(p && p != kCacheClosed &&
264	10365x	!svc->invoker_frame_cache_.compare_exchange_weak(
265		p, nullptr, std::memory_order_acquire))
266		{
267		}
268	18238x	if(!p \|\| p == kCacheClosed)
269	7873x	p = ::operator new(total);
270
271	18238x	reinterpret_cast<strand_service_impl*>(
272	18238x	static_cast<char*>(p) + padded) = svc;
273	18238x	return p;
274		}
275
276		void
277	18238x	operator delete(void* p, std::size_t n) noexcept
278		{
279	18238x	constexpr auto A = alignof(strand_service_impl*);
280	18238x	std::size_t padded = (n + A - 1) & ~(A - 1);
281	18238x	auto* svc = reinterpret_cast<strand_service_impl*>(
282		static_cast<char*>(p) + padded);
283
284	18238x	void* expected = nullptr;
285	18238x	if(!svc->invoker_frame_cache_.compare_exchange_strong(
286		expected, p, std::memory_order_release))
287	7844x	::operator delete(p);
288	18238x	}
289
290		strand_invoker
291	18238x	get_return_object() noexcept
292		{
293	18238x	return {std::coroutine_handle<promise_type>::from_promise(*this)};
294		}
295
296	18238x	std::suspend_always initial_suspend() noexcept { return {}; }
297	17379x	std::suspend_never final_suspend() noexcept { return {}; }
298	17379x	void return_void() noexcept {}
299	−	void unhandled_exception() { std::terminate(); } // LCOV_EXCL_LINE terminates the process
300		};
301
302		std::coroutine_handle<promise_type> h_;
303		};
304
305		// The by-value parameter lives in the coroutine frame for the
306		// invoker's lifetime, keeping the impl alive past any user-side
307		// strand drop.
308		static
309		strand_invoker
310	18238x	make_invoker(strand_service_impl*, std::shared_ptr<strand_impl> impl)
311		{
312		auto* p = impl.get();
313		for(;;)
314		{
315		strand_service_impl::set_dispatch_thread(*p);
316		strand_service_impl::dispatch_pending(*p);
317		if(strand_service_impl::try_unlock(*p))
318		{
319		strand_service_impl::clear_dispatch_thread(*p);
320		co_return;
321		}
322		}
323	36476x	}
324
325		void
326	18238x	strand_service_impl::post_invoker(
327		strand_service_impl* svc,
328		std::shared_ptr<strand_impl> impl,
329		executor_ref ex)
330		{
331	18238x	auto invoker = make_invoker(svc, std::move(impl));
332	18238x	auto& self = invoker.h_.promise().self_;
333	18238x	self.h = invoker.h_;
334		try
335		{
336	18238x	ex.post(self);
337		}
338	7x	catch(...)
339		{
340	7x	invoker.h_.destroy();
341	7x	throw;
342	7x	}
343	18231x	}
344
345		// A post that throws leaves the caller holding `c`, unless a
346		// shutdown already took and abandoned it.
347		static
348		void
349	18235x	post_or_withdraw(
350		strand_service_impl* svc,
351		std::shared_ptr<strand_impl> const& impl,
352		executor_ref ex,
353		continuation& c)
354		{
355		try
356		{
357	18240x	strand_service_impl::post_invoker(svc, impl, ex);
358		}
359	5x	catch(...)
360		{
361		using withdrawal = strand_service_impl::withdrawal;
362	5x	auto const w = strand_service_impl::withdraw(*impl, c);
363	5x	if(w == withdrawal::taken)
364	1x	return;
365	4x	if(w == withdrawal::followed)
366		{
367		try
368		{
369	5x	strand_service_impl::post_invoker(svc, impl, ex);
370		}
371	2x	catch(...)
372		{
373	2x	strand_service_impl::abandon_pending(*impl);
374	2x	}
375		}
376	4x	throw;
377	5x	}
378		}
379
380	12467x	strand_impl::~strand_impl()
381		{
382	12467x	if(auto* svc = service_.load(std::memory_order_acquire))
383		{
384	11451x	std::lock_guard<std::mutex> lock(svc->mutex_);
385	11451x	svc->impl_list_.remove(this);
386	11451x	}
387		// Nothing can run these any more: give them up.
388	12467x	auto batch = pending_.take_all();
389	12467x	strand_queue::abandon_batch(batch);
390	12467x	}
391
392	1051x	strand_service::
393	1051x	strand_service()
394	1051x	: service()
395		{
396	1051x	}
397
398	1051x	strand_service::
399		~strand_service() = default;
400
401		bool
402	100017x	strand_service::
403		running_in_this_thread(strand_impl& impl) noexcept
404		{
405	100017x	return impl.dispatch_thread_.load() == std::this_thread::get_id();
406		}
407
408		std::coroutine_handle<>
409	100013x	strand_service::
410		dispatch(
411		std::shared_ptr<strand_impl> const& impl,
412		executor_ref ex,
413		continuation& c)
414		{
415	100013x	if(running_in_this_thread(*impl))
416	3x	return c.h;
417
418	100010x	if(auto* svc = strand_service_impl::enqueue(*impl, c))
419	11x	post_or_withdraw(svc, impl, ex, c);
420	100010x	return std::noop_coroutine();
421		}
422
423		void
424	231429x	strand_service::
425		post(
426		std::shared_ptr<strand_impl> const& impl,
427		executor_ref ex,
428		continuation& c)
429		{
430	231429x	if(auto* svc = strand_service_impl::enqueue(*impl, c))
431	18224x	post_or_withdraw(svc, impl, ex, c);
432	231425x	}
433
434		strand_service&
435	12467x	get_strand_service(execution_context& ctx)
436		{
437	12467x	return ctx.use_service<strand_service_impl>();
438		}
439
440		} // namespace detail
441		} // namespace capy
442		} // namespace boost
443