. /src /test /run_blocking.cpp
97.70% Lines (85/87) 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. The
50   // services may post while they shut down or are destroyed, so the 50   // services may post while they shut down or are destroyed, so the
51   // queue outlives both. Each is taken under the lock, as other 51   // queue outlives both. Each is taken under the lock, as other
52   // threads may still post, and abandoned outside it. 52   // threads may still post, and abandoned outside it.
HITCBC 53   2253 auto drain = [this] 53   2254 auto drain = [this]
54   { 54   {
HITCBC 55   2253 bool any = false; 55   2254 bool any = false;
56   for(;;) 56   for(;;)
57   { 57   {
HITCBC 58   4260 continuation c; 58   4261 continuation c;
59   { 59   {
HITCBC 60   4260 std::lock_guard<std::mutex> lock(impl_->mtx); 60   4261 std::lock_guard<std::mutex> lock(impl_->mtx);
HITCBC 61   4260 if(impl_->queue.empty()) 61   4261 if(impl_->queue.empty())
HITCBC 62   4506 return any; 62   4508 return any;
HITCBC 63   2007 c = impl_->queue.front(); 63   2007 c = impl_->queue.front();
HITCBC 64   2007 impl_->queue.pop(); 64   2007 impl_->queue.pop();
HITCBC 65   4260 } 65   4261 }
HITCBC 66   2007 any = true; 66   2007 any = true;
HITCBC 67   2007 capy::abandon(c); 67   2007 capy::abandon(c);
HITCBC 68   2007 } 68   2007 }
HITCBC 69   1125 }; 69   1125 };
HITCBC 70   1125 shutdown(); 70   1125 shutdown();
HITCBC 71   1125 drain(); 71   1125 drain();
72   // A frame the drain unwinds can create a service, whose destructor 72   // A frame the drain unwinds can create a service, whose destructor
73   // can post again: repeat until nothing is left of either. 73   // can post again: repeat until nothing is left of either.
74   do 74   do
75   { 75   {
HITCBC 76   1128 destroy(); 76   1129 destroy();
77   } 77   }
HITCBC 78   1128 while(drain()); 78   1129 while(drain());
HITCBC 79   1125 delete impl_; 79   1125 delete impl_;
HITCBC 80   1125 } 80   1125 }
81   81  
82   blocking_executor 82   blocking_executor
HITCBC 83   1126 blocking_context::get_executor() noexcept 83   1126 blocking_context::get_executor() noexcept
84   { 84   {
HITCBC 85   1126 return blocking_executor{this}; 85   1126 return blocking_executor{this};
86   } 86   }
87   87  
88   void 88   void
HITCBC 89   778 blocking_context::signal_done() noexcept 89   778 blocking_context::signal_done() noexcept
90   { 90   {
HITCBC 91   778 std::lock_guard<std::mutex> lock(impl_->mtx); 91   778 std::lock_guard<std::mutex> lock(impl_->mtx);
HITCBC 92   778 impl_->done = true; 92   778 impl_->done = true;
HITCBC 93   778 impl_->cv.notify_one(); 93   778 impl_->cv.notify_one();
HITCBC 94   778 } 94   778 }
95   95  
96   void 96   void
HITCBC 97   336 blocking_context::signal_done( 97   336 blocking_context::signal_done(
98   std::exception_ptr ep) noexcept 98   std::exception_ptr ep) noexcept
99   { 99   {
HITCBC 100   336 std::lock_guard<std::mutex> lock(impl_->mtx); 100   336 std::lock_guard<std::mutex> lock(impl_->mtx);
HITCBC 101   336 impl_->ep = ep; 101   336 impl_->ep = ep;
HITCBC 102   336 impl_->done = true; 102   336 impl_->done = true;
HITCBC 103   336 impl_->cv.notify_one(); 103   336 impl_->cv.notify_one();
HITCBC 104   336 } 104   336 }
105   105  
106   void 106   void
HITCBC 107   1 blocking_context::signal_abandoned() noexcept 107   1 blocking_context::signal_abandoned() noexcept
108   { 108   {
HITCBC 109   1 std::lock_guard<std::mutex> lock(impl_->mtx); 109   1 std::lock_guard<std::mutex> lock(impl_->mtx);
HITCBC 110   1 impl_->abandoned = true; 110   1 impl_->abandoned = true;
HITCBC 111   1 impl_->done = true; 111   1 impl_->done = true;
HITCBC 112   1 impl_->cv.notify_one(); 112   1 impl_->cv.notify_one();
HITCBC 113   1 } 113   1 }
114   114  
115   void 115   void
HITCBC 116   1115 blocking_context::run() 116   1115 blocking_context::run()
117   { 117   {
118   struct scoped_running 118   struct scoped_running
119   { 119   {
120   blocking_context const* prev_; 120   blocking_context const* prev_;
121   121  
HITCBC 122   1115 explicit scoped_running(blocking_context const* ctx) noexcept 122   1115 explicit scoped_running(blocking_context const* ctx) noexcept
HITCBC 123   1115 : prev_(std::exchange(running_, ctx)) 123   1115 : prev_(std::exchange(running_, ctx))
124   { 124   {
HITCBC 125   1115 } 125   1115 }
126   126  
HITCBC 127   1115 ~scoped_running() 127   1115 ~scoped_running()
128   { 128   {
HITCBC 129   1115 running_ = prev_; 129   1115 running_ = prev_;
HITCBC 130   1115 } 130   1115 }
HITCBC 131   1115 } guard(this); 131   1115 } guard(this);
132   132  
133   for(;;) 133   for(;;)
134   { 134   {
HITCBC 135   2352 continuation c; 135   2352 continuation c;
136   { 136   {
HITCBC 137   2352 std::unique_lock<std::mutex> lock(impl_->mtx); 137   2352 std::unique_lock<std::mutex> lock(impl_->mtx);
HITCBC 138   2352 impl_->cv.wait(lock, [&] { 138   2352 impl_->cv.wait(lock, [&] {
HITCBC 139   2353 return impl_->done || !impl_->queue.empty(); 139   2353 return impl_->done || !impl_->queue.empty();
140   }); 140   });
HITCBC 141   2352 if(impl_->done && impl_->queue.empty()) 141   2352 if(impl_->done && impl_->queue.empty())
HITCBC 142   1115 break; 142   1115 break;
HITCBC 143   1237 c = impl_->queue.front(); 143   1237 c = impl_->queue.front();
HITCBC 144   1237 impl_->queue.pop(); 144   1237 impl_->queue.pop();
HITCBC 145   2352 } 145   2352 }
HITCBC 146   1237 safe_resume(c.h); 146   1237 safe_resume(c.h);
HITCBC 147   1237 } 147   1237 }
HITCBC 148   1115 if(impl_->ep) 148   1115 if(impl_->ep)
HITCBC 149   672 std::rethrow_exception(impl_->ep); 149   672 std::rethrow_exception(impl_->ep);
HITCBC 150   779 if(impl_->abandoned) 150   779 if(impl_->abandoned)
HITCBC 151   1 throw std::system_error(error::canceled, "task abandoned"); 151   1 throw std::system_error(error::canceled, "task abandoned");
HITCBC 152   1115 } 152   1115 }
153   153  
154   void 154   void
HITCBC 155   3244 blocking_context::enqueue( 155   3244 blocking_context::enqueue(
156   continuation const& c) 156   continuation const& c)
157   { 157   {
HITCBC 158   3244 std::lock_guard<std::mutex> lock(impl_->mtx); 158   3244 std::lock_guard<std::mutex> lock(impl_->mtx);
HITCBC 159   3244 impl_->queue.push(c); 159   3244 impl_->queue.push(c);
HITCBC 160   3244 impl_->cv.notify_one(); 160   3244 impl_->cv.notify_one();
HITCBC 161   3244 } 161   3244 }
162   162  
163   //---------------------------------------------------------- 163   //----------------------------------------------------------
164   164  
165   bool 165   bool
MISUBC 166   ✗ blocking_executor::operator==( 166   ✗ blocking_executor::operator==(
167   blocking_executor const& other) const noexcept 167   blocking_executor const& other) const noexcept
168   { 168   {
MISUBC 169   ✗ return ctx_ == other.ctx_; 169   ✗ return ctx_ == other.ctx_;
170   } 170   }
171   171  
172   blocking_context& 172   blocking_context&
HITCBC 173   1117 blocking_executor::context() const noexcept 173   1117 blocking_executor::context() const noexcept
174   { 174   {
HITCBC 175   1117 return *ctx_; 175   1117 return *ctx_;
176   } 176   }
177   177  
178   void 178   void
HITCBC 179   1120 blocking_executor::on_work_started() const noexcept 179   1120 blocking_executor::on_work_started() const noexcept
180   { 180   {
HITCBC 181   1120 } 181   1120 }
182   182  
183   void 183   void
HITCBC 184   1120 blocking_executor::on_work_finished() const noexcept 184   1120 blocking_executor::on_work_finished() const noexcept
185   { 185   {
HITCBC 186   1120 } 186   1120 }
187   187  
188   std::coroutine_handle<> 188   std::coroutine_handle<>
HITCBC 189   1163 blocking_executor::dispatch( 189   1163 blocking_executor::dispatch(
190   continuation& c) const 190   continuation& c) const
191   { 191   {
HITCBC 192   1163 if(running_ == ctx_) 192   1163 if(running_ == ctx_)
HITCBC 193   47 return c.h; 193   47 return c.h;
HITCBC 194   1116 ctx_->enqueue(c); 194   1116 ctx_->enqueue(c);
HITCBC 195   1116 return std::noop_coroutine(); 195   1116 return std::noop_coroutine();
196   } 196   }
197   197  
198   void 198   void
HITCBC 199   2128 blocking_executor::post( 199   2128 blocking_executor::post(
200   continuation& c) const 200   continuation& c) const
201   { 201   {
HITCBC 202   2128 ctx_->enqueue(c); 202   2128 ctx_->enqueue(c);
HITCBC 203   2128 } 203   2128 }
204   204  
205   } // namespace test 205   } // namespace test
206   } // namespace capy 206   } // namespace capy
207   } // namespace boost 207   } // namespace boost