TLA Line data 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 HIT 1051 : strand_service_impl(execution_context&)
51 1051 : {
52 1051 : }
53 :
54 : std::shared_ptr<strand_impl>
55 12467 : create_implementation() override
56 : {
57 12467 : auto new_impl = std::make_shared<strand_impl>();
58 :
59 12467 : std::lock_guard<std::mutex> lock(mutex_);
60 12467 : if(shutdown_)
61 : {
62 : // Born closed, like the strands shutdown() closed: it
63 : // borrows nothing from this service.
64 2 : new_impl->locked_ = true;
65 2 : return new_impl;
66 : }
67 :
68 12465 : std::size_t s = salt_++;
69 12465 : std::size_t idx = reinterpret_cast<std::size_t>(new_impl.get());
70 12465 : idx += idx >> 3;
71 12465 : idx ^= s + 0x9e3779b9 + (idx << 6) + (idx >> 2);
72 12465 : idx %= num_mutexes;
73 12465 : if(!mutexes_[idx])
74 1684 : mutexes_[idx] = std::make_shared<std::mutex>();
75 12465 : new_impl->mutex_ = mutexes_[idx].get();
76 :
77 12465 : impl_list_.push_back(new_impl.get());
78 12465 : new_impl->service_.store(this, std::memory_order_release);
79 :
80 12465 : return new_impl;
81 12467 : }
82 :
83 : // Returns the service when the caller must post an invoker.
84 : static strand_service_impl*
85 331439 : 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 331439 : if(impl.service_.load(std::memory_order_acquire))
90 : {
91 131373 : std::lock_guard<std::mutex> lock(*impl.mutex_);
92 131373 : if(auto* svc = impl.service_.load(std::memory_order_relaxed))
93 : {
94 131273 : impl.pending_.push(c);
95 131273 : if(!impl.locked_)
96 : {
97 18235 : impl.locked_ = true;
98 18235 : return svc;
99 : }
100 113038 : return nullptr;
101 : }
102 131373 : }
103 : // Shut down: nothing will run it, so abandon it as its holder.
104 200166 : capy::abandon(c);
105 200166 : 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 5 : withdraw(strand_impl& impl, continuation& c) noexcept
119 : {
120 5 : std::lock_guard<std::mutex> lock(*impl.mutex_);
121 5 : if(!impl.service_.load(std::memory_order_relaxed))
122 1 : return withdrawal::taken;
123 : // Nothing takes from the queue while it waits for an invoker,
124 : // so `c` is still first.
125 4 : BOOST_CAPY_ASSERT(impl.pending_.front() == &c);
126 4 : std::ignore = c;
127 4 : impl.pending_.pop_front();
128 : // Posts queued behind `c` returned trusting its invoker, so
129 : // the strand stays locked until one runs them.
130 4 : if(!impl.pending_.empty())
131 3 : return withdrawal::followed;
132 1 : impl.locked_ = false;
133 1 : return withdrawal::alone;
134 5 : }
135 :
136 : // Gives up queued work that no invoker can be posted to run.
137 : static void
138 2 : abandon_pending(strand_impl& impl) noexcept
139 : {
140 2 : strand_queue::taken_batch batch;
141 : {
142 2 : std::lock_guard<std::mutex> lock(*impl.mutex_);
143 2 : if(!impl.service_.load(std::memory_order_relaxed))
144 1 : return;
145 1 : batch = impl.pending_.take_all();
146 1 : impl.locked_ = false;
147 2 : }
148 1 : strand_queue::abandon_batch(batch);
149 : }
150 :
151 : static void
152 18881 : dispatch_pending(strand_impl& impl)
153 : {
154 18881 : strand_queue::taken_batch batch;
155 : {
156 18881 : std::lock_guard<std::mutex> lock(*impl.mutex_);
157 18881 : batch = impl.pending_.take_all();
158 18881 : }
159 18881 : impl.pending_.dispatch_batch(batch);
160 18881 : }
161 :
162 : static bool
163 18881 : try_unlock(strand_impl& impl)
164 : {
165 18881 : std::lock_guard<std::mutex> lock(*impl.mutex_);
166 18881 : if(impl.pending_.empty())
167 : {
168 17379 : impl.locked_ = false;
169 17379 : return true;
170 : }
171 1502 : return false;
172 18881 : }
173 :
174 : static void
175 18881 : set_dispatch_thread(strand_impl& impl) noexcept
176 : {
177 18881 : impl.dispatch_thread_.store(std::this_thread::get_id());
178 18881 : }
179 :
180 : static void
181 17379 : clear_dispatch_thread(strand_impl& impl) noexcept
182 : {
183 17379 : impl.dispatch_thread_.store(std::thread::id{});
184 17379 : }
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 1051 : shutdown() override
196 : {
197 1051 : strand_queue::taken_batch doomed;
198 : {
199 1051 : std::lock_guard<std::mutex> lock(mutex_);
200 1051 : shutdown_ = true;
201 2065 : while(auto* p = impl_list_.pop_front())
202 : {
203 1014 : std::lock_guard<std::mutex> impl_lock(*p->mutex_);
204 1014 : p->locked_ = true;
205 1014 : p->service_.store(nullptr, std::memory_order_release);
206 1014 : auto b = p->pending_.take_all();
207 1014 : if(b.head)
208 : {
209 851 : if(doomed.tail)
210 1 : doomed.tail->reserved = b.head;
211 : else
212 850 : doomed.head = b.head;
213 851 : doomed.tail = b.tail;
214 : }
215 1014 : }
216 1051 : }
217 : // Outside the mutexes: a destroyed frame may drop the last
218 : // strand handle and re-enter the service.
219 1051 : strand_queue::abandon_batch(doomed);
220 :
221 1051 : void* fp = invoker_frame_cache_.exchange(
222 : kCacheClosed, std::memory_order_acq_rel);
223 1051 : if(fp) ::operator delete(fp);
224 1051 : }
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 18238 : operator new(
251 : std::size_t n,
252 : strand_service_impl* svc,
253 : std::shared_ptr<strand_impl> const&)
254 : {
255 18238 : constexpr auto A = alignof(strand_service_impl*);
256 18238 : std::size_t padded = (n + A - 1) & ~(A - 1);
257 18238 : 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 18238 : void* p = svc->invoker_frame_cache_.load(
262 18238 : std::memory_order_acquire);
263 28603 : while(p && p != kCacheClosed &&
264 10365 : !svc->invoker_frame_cache_.compare_exchange_weak(
265 : p, nullptr, std::memory_order_acquire))
266 : {
267 : }
268 18238 : if(!p || p == kCacheClosed)
269 7873 : p = ::operator new(total);
270 :
271 18238 : *reinterpret_cast<strand_service_impl**>(
272 18238 : static_cast<char*>(p) + padded) = svc;
273 18238 : return p;
274 : }
275 :
276 : void
277 18238 : operator delete(void* p, std::size_t n) noexcept
278 : {
279 18238 : constexpr auto A = alignof(strand_service_impl*);
280 18238 : std::size_t padded = (n + A - 1) & ~(A - 1);
281 18238 : auto* svc = *reinterpret_cast<strand_service_impl**>(
282 : static_cast<char*>(p) + padded);
283 :
284 18238 : void* expected = nullptr;
285 18238 : if(!svc->invoker_frame_cache_.compare_exchange_strong(
286 : expected, p, std::memory_order_release))
287 7844 : ::operator delete(p);
288 18238 : }
289 :
290 : strand_invoker
291 18238 : get_return_object() noexcept
292 : {
293 18238 : return {std::coroutine_handle<promise_type>::from_promise(*this)};
294 : }
295 :
296 18238 : std::suspend_always initial_suspend() noexcept { return {}; }
297 17379 : std::suspend_never final_suspend() noexcept { return {}; }
298 17379 : 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 18238 : 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 36476 : }
324 :
325 : void
326 18238 : strand_service_impl::post_invoker(
327 : strand_service_impl* svc,
328 : std::shared_ptr<strand_impl> impl,
329 : executor_ref ex)
330 : {
331 18238 : auto invoker = make_invoker(svc, std::move(impl));
332 18238 : auto& self = invoker.h_.promise().self_;
333 18238 : self.h = invoker.h_;
334 : try
335 : {
336 18238 : ex.post(self);
337 : }
338 7 : catch(...)
339 : {
340 7 : invoker.h_.destroy();
341 7 : throw;
342 7 : }
343 18231 : }
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 18235 : 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 18240 : strand_service_impl::post_invoker(svc, impl, ex);
358 : }
359 5 : catch(...)
360 : {
361 : using withdrawal = strand_service_impl::withdrawal;
362 5 : auto const w = strand_service_impl::withdraw(*impl, c);
363 5 : if(w == withdrawal::taken)
364 1 : return;
365 4 : if(w == withdrawal::followed)
366 : {
367 : try
368 : {
369 5 : strand_service_impl::post_invoker(svc, impl, ex);
370 : }
371 2 : catch(...)
372 : {
373 2 : strand_service_impl::abandon_pending(*impl);
374 2 : }
375 : }
376 4 : throw;
377 5 : }
378 : }
379 :
380 12467 : strand_impl::~strand_impl()
381 : {
382 12467 : if(auto* svc = service_.load(std::memory_order_acquire))
383 : {
384 11451 : std::lock_guard<std::mutex> lock(svc->mutex_);
385 11451 : svc->impl_list_.remove(this);
386 11451 : }
387 : // Nothing can run these any more: give them up.
388 12467 : auto batch = pending_.take_all();
389 12467 : strand_queue::abandon_batch(batch);
390 12467 : }
391 :
392 1051 : strand_service::
393 1051 : strand_service()
394 1051 : : service()
395 : {
396 1051 : }
397 :
398 1051 : strand_service::
399 : ~strand_service() = default;
400 :
401 : bool
402 100017 : strand_service::
403 : running_in_this_thread(strand_impl& impl) noexcept
404 : {
405 100017 : return impl.dispatch_thread_.load() == std::this_thread::get_id();
406 : }
407 :
408 : std::coroutine_handle<>
409 100013 : strand_service::
410 : dispatch(
411 : std::shared_ptr<strand_impl> const& impl,
412 : executor_ref ex,
413 : continuation& c)
414 : {
415 100013 : if(running_in_this_thread(*impl))
416 3 : return c.h;
417 :
418 100010 : if(auto* svc = strand_service_impl::enqueue(*impl, c))
419 11 : post_or_withdraw(svc, impl, ex, c);
420 100010 : return std::noop_coroutine();
421 : }
422 :
423 : void
424 231429 : strand_service::
425 : post(
426 : std::shared_ptr<strand_impl> const& impl,
427 : executor_ref ex,
428 : continuation& c)
429 : {
430 231429 : if(auto* svc = strand_service_impl::enqueue(*impl, c))
431 18224 : post_or_withdraw(svc, impl, ex, c);
432 231425 : }
433 :
434 : strand_service&
435 12467 : get_strand_service(execution_context& ctx)
436 : {
437 12467 : return ctx.use_service<strand_service_impl>();
438 : }
439 :
440 : } // namespace detail
441 : } // namespace capy
442 : } // namespace boost
|
