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