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