. /src /test /run_blocking.cpp
97.65% Lines (83/85) 94.44% Functions (17/18)
[ ] 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 <boost/capy/test/run_blocking.hpp> 10   #include <boost/capy/test/run_blocking.hpp>
11   11  
12   #include <boost/capy/ex/abandon.hpp> 12   #include <boost/capy/ex/abandon.hpp>
13   #include <boost/capy/ex/frame_allocator.hpp> 13   #include <boost/capy/ex/frame_allocator.hpp>
14   #include <condition_variable> 14   #include <condition_variable>
15   #include <mutex> 15   #include <mutex>
16   #include <queue> 16   #include <queue>
17   #include <system_error> 17   #include <system_error>
18   #include <utility> 18   #include <utility>
19   19  
20   namespace boost { 20   namespace boost {
21   namespace capy { 21   namespace capy {
22   namespace test { 22   namespace test {
23   23  
24   namespace { 24   namespace {
25   25  
26   // The context whose run() is on this thread's stack. Only there is it 26   // The context whose run() is on this thread's stack. Only there is it
27   // safe to resume dispatched work inline. 27   // safe to resume dispatched work inline.
28   thread_local blocking_context const* running_ = nullptr; 28   thread_local blocking_context const* running_ = nullptr;
29   29  
30   } // namespace 30   } // namespace
31   31  
32   struct blocking_context::impl 32   struct blocking_context::impl
33   { 33   {
34   std::mutex mtx; 34   std::mutex mtx;
35   std::condition_variable cv; 35   std::condition_variable cv;
36   std::queue<continuation> queue; 36   std::queue<continuation> queue;
37   std::exception_ptr ep; 37   std::exception_ptr ep;
38   bool done = false; 38   bool done = false;
39   bool abandoned = false; 39   bool abandoned = false;
40   }; 40   };
41   41  
HITCBC 42   1125 blocking_context::blocking_context() 42   1125 blocking_context::blocking_context()
HITCBC 43   1125 : impl_(new impl) 43   1125 : impl_(new impl)
44   { 44   {
HITCBC 45   1125 } 45   1125 }
46   46  
HITCBC 47   1125 blocking_context::~blocking_context() 47   1125 blocking_context::~blocking_context()
48   { 48   {
49 - // Nothing will run these now; abandon them as their holder. The 49 + // Nothing will run these now; abandon them as their holder. Each is
50 - // services may post while they shut down or are destroyed, so the 50 + // taken under the lock and abandoned outside it, since abandoning
51 - // queue outlives both. Each is taken under the lock, as other 51 + // one may post another.
52 - // threads may still post, and abandoned outside it.  
HITCBC 53   2254 auto drain = [this] 52   2250 auto drain = [this]
54 - bool any = false;  
ECB 55   2254 { 53   {
56   for(;;) 54   for(;;)
57   { 55   {
HITCBC 58   4261 continuation c; 56   2273 continuation c;
59   { 57   {
HITCBC 60   4261 std::lock_guard<std::mutex> lock(impl_->mtx); 58   2273 std::lock_guard<std::mutex> lock(impl_->mtx);
HITCBC 61   4261 if(impl_->queue.empty()) 59   2273 if(impl_->queue.empty())
HITCBC 62 - 4508 return any; 60 + 4500 return;
HITCBC 63   2007 c = impl_->queue.front(); 61   23 c = impl_->queue.front();
HITCBC 64   2007 impl_->queue.pop(); 62   23 impl_->queue.pop();
DCB 65 - 4261 any = true;  
HITCBC 66   2007 } 63   2273 }
HITCBC 67   2007 capy::abandon(c); 64   23 capy::abandon(c);
HITCBC 68   2007 } 65   23 }
HITCBC 69   1125 }; 66   1125 };
HITCBC 70   1125 shutdown(); 67   1125 shutdown();
  68 + // Takes what the unwinds post too, and the work of any service they
  69 + // create, which is shut down as it is created.
HITGNC   70 + 1125 drain();
HITGNC   71 + 1125 destroy();
  72 + // Services give up their work in shutdown(), but anything a
  73 + // destructor posts anyway is still destroyed.
DCB 71 - 1125 // A frame the drain unwinds can create a service, whose destructor  
72 - // can post again: repeat until nothing is left of either.  
73 - do  
74 - {  
75 - destroy();  
DCB 76 - 1129 }  
77 - while(drain());  
HITCBC 78   1129 drain(); 74   1125 drain();
HITCBC 79   1125 delete impl_; 75   1125 delete impl_;
HITCBC 80   1125 } 76   1125 }
81   77  
82   blocking_executor 78   blocking_executor
HITCBC 83   1126 blocking_context::get_executor() noexcept 79   1126 blocking_context::get_executor() noexcept
84   { 80   {
HITCBC 85   1126 return blocking_executor{this}; 81   1126 return blocking_executor{this};
86   } 82   }
87   83  
88   void 84   void
HITCBC 89   778 blocking_context::signal_done() noexcept 85   778 blocking_context::signal_done() noexcept
90   { 86   {
HITCBC 91   778 std::lock_guard<std::mutex> lock(impl_->mtx); 87   778 std::lock_guard<std::mutex> lock(impl_->mtx);
HITCBC 92   778 impl_->done = true; 88   778 impl_->done = true;
HITCBC 93   778 impl_->cv.notify_one(); 89   778 impl_->cv.notify_one();
HITCBC 94   778 } 90   778 }
95   91  
96   void 92   void
HITCBC 97   336 blocking_context::signal_done( 93   336 blocking_context::signal_done(
98   std::exception_ptr ep) noexcept 94   std::exception_ptr ep) noexcept
99   { 95   {
HITCBC 100   336 std::lock_guard<std::mutex> lock(impl_->mtx); 96   336 std::lock_guard<std::mutex> lock(impl_->mtx);
HITCBC 101   336 impl_->ep = ep; 97   336 impl_->ep = ep;
HITCBC 102   336 impl_->done = true; 98   336 impl_->done = true;
HITCBC 103   336 impl_->cv.notify_one(); 99   336 impl_->cv.notify_one();
HITCBC 104   336 } 100   336 }
105   101  
106   void 102   void
HITCBC 107   1 blocking_context::signal_abandoned() noexcept 103   1 blocking_context::signal_abandoned() noexcept
108   { 104   {
HITCBC 109   1 std::lock_guard<std::mutex> lock(impl_->mtx); 105   1 std::lock_guard<std::mutex> lock(impl_->mtx);
HITCBC 110   1 impl_->abandoned = true; 106   1 impl_->abandoned = true;
HITCBC 111   1 impl_->done = true; 107   1 impl_->done = true;
HITCBC 112   1 impl_->cv.notify_one(); 108   1 impl_->cv.notify_one();
HITCBC 113   1 } 109   1 }
114   110  
115   void 111   void
HITCBC 116   1115 blocking_context::run() 112   1115 blocking_context::run()
117   { 113   {
118   struct scoped_running 114   struct scoped_running
119   { 115   {
120   blocking_context const* prev_; 116   blocking_context const* prev_;
121   117  
HITCBC 122   1115 explicit scoped_running(blocking_context const* ctx) noexcept 118   1115 explicit scoped_running(blocking_context const* ctx) noexcept
HITCBC 123   1115 : prev_(std::exchange(running_, ctx)) 119   1115 : prev_(std::exchange(running_, ctx))
124   { 120   {
HITCBC 125   1115 } 121   1115 }
126   122  
HITCBC 127   1115 ~scoped_running() 123   1115 ~scoped_running()
128   { 124   {
HITCBC 129   1115 running_ = prev_; 125   1115 running_ = prev_;
HITCBC 130   1115 } 126   1115 }
HITCBC 131   1115 } guard(this); 127   1115 } guard(this);
132   128  
133   for(;;) 129   for(;;)
134   { 130   {
HITCBC 135   2352 continuation c; 131   2352 continuation c;
136   { 132   {
HITCBC 137   2352 std::unique_lock<std::mutex> lock(impl_->mtx); 133   2352 std::unique_lock<std::mutex> lock(impl_->mtx);
HITCBC 138   2352 impl_->cv.wait(lock, [&] { 134   2352 impl_->cv.wait(lock, [&] {
HITCBC 139   2353 return impl_->done || !impl_->queue.empty(); 135   2353 return impl_->done || !impl_->queue.empty();
140   }); 136   });
HITCBC 141   2352 if(impl_->done && impl_->queue.empty()) 137   2352 if(impl_->done && impl_->queue.empty())
HITCBC 142   1115 break; 138   1115 break;
HITCBC 143   1237 c = impl_->queue.front(); 139   1237 c = impl_->queue.front();
HITCBC 144   1237 impl_->queue.pop(); 140   1237 impl_->queue.pop();
HITCBC 145   2352 } 141   2352 }
HITCBC 146   1237 safe_resume(c.h); 142   1237 safe_resume(c.h);
HITCBC 147   1237 } 143   1237 }
HITCBC 148   1115 if(impl_->ep) 144   1115 if(impl_->ep)
HITCBC 149   672 std::rethrow_exception(impl_->ep); 145   672 std::rethrow_exception(impl_->ep);
HITCBC 150   779 if(impl_->abandoned) 146   779 if(impl_->abandoned)
HITCBC 151   1 throw std::system_error(error::canceled, "task abandoned"); 147   1 throw std::system_error(error::canceled, "task abandoned");
HITCBC 152   1115 } 148   1115 }
153   149  
154   void 150   void
HITCBC 155   3244 blocking_context::enqueue( 151   1260 blocking_context::enqueue(
156   continuation const& c) 152   continuation const& c)
157   { 153   {
HITCBC 158   3244 std::lock_guard<std::mutex> lock(impl_->mtx); 154   1260 std::lock_guard<std::mutex> lock(impl_->mtx);
HITCBC 159   3244 impl_->queue.push(c); 155   1260 impl_->queue.push(c);
HITCBC 160   3244 impl_->cv.notify_one(); 156   1260 impl_->cv.notify_one();
HITCBC 161   3244 } 157   1260 }
162   158  
163   //---------------------------------------------------------- 159   //----------------------------------------------------------
164   160  
165   bool 161   bool
MISUBC 166   ✗ blocking_executor::operator==( 162   ✗ blocking_executor::operator==(
167   blocking_executor const& other) const noexcept 163   blocking_executor const& other) const noexcept
168   { 164   {
MISUBC 169   ✗ return ctx_ == other.ctx_; 165   ✗ return ctx_ == other.ctx_;
170   } 166   }
171   167  
172   blocking_context& 168   blocking_context&
HITCBC 173   1117 blocking_executor::context() const noexcept 169   1117 blocking_executor::context() const noexcept
174   { 170   {
HITCBC 175   1117 return *ctx_; 171   1117 return *ctx_;
176   } 172   }
177   173  
178   void 174   void
HITCBC 179   1120 blocking_executor::on_work_started() const noexcept 175   1120 blocking_executor::on_work_started() const noexcept
180   { 176   {
HITCBC 181   1120 } 177   1120 }
182   178  
183   void 179   void
HITCBC 184   1120 blocking_executor::on_work_finished() const noexcept 180   1120 blocking_executor::on_work_finished() const noexcept
185   { 181   {
HITCBC 186   1120 } 182   1120 }
187   183  
188   std::coroutine_handle<> 184   std::coroutine_handle<>
HITCBC 189   1163 blocking_executor::dispatch( 185   1163 blocking_executor::dispatch(
190   continuation& c) const 186   continuation& c) const
191   { 187   {
HITCBC 192   1163 if(running_ == ctx_) 188   1163 if(running_ == ctx_)
HITCBC 193   47 return c.h; 189   47 return c.h;
HITCBC 194   1116 ctx_->enqueue(c); 190   1116 ctx_->enqueue(c);
HITCBC 195   1116 return std::noop_coroutine(); 191   1116 return std::noop_coroutine();
196   } 192   }
197   193  
198   void 194   void
HITCBC 199   2128 blocking_executor::post( 195   144 blocking_executor::post(
200   continuation& c) const 196   continuation& c) const
201   { 197   {
HITCBC 202   2128 ctx_->enqueue(c); 198   144 ctx_->enqueue(c);
HITCBC 203   2128 } 199   144 }
204   200  
205   } // namespace test 201   } // namespace test
206   } // namespace capy 202   } // namespace capy
207   } // namespace boost 203   } // namespace boost