. /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   1130 blocking_context::blocking_context() 42   1125 blocking_context::blocking_context()
HITCBC 43   1130 : impl_(new impl) 43   1125 : impl_(new impl)
44   { 44   {
HITCBC 45   1130 } 45   1125 }
46   46  
HITCBC 47   1130 blocking_context::~blocking_context() 47   1125 blocking_context::~blocking_context()
48   { 48   {
49   // Nothing will run these now; abandon them as their holder. Each is 49   // Nothing will run these now; abandon them as their holder. Each is
50   // taken under the lock and abandoned outside it, since abandoning 50   // taken under the lock and abandoned outside it, since abandoning
51   // one may post another. 51   // one may post another.
HITCBC 52   2260 auto drain = [this] 52   2250 auto drain = [this]
53   { 53   {
54   for(;;) 54   for(;;)
55   { 55   {
HITCBC 56   2283 continuation c; 56   2273 continuation c;
57   { 57   {
HITCBC 58   2283 std::lock_guard<std::mutex> lock(impl_->mtx); 58   2273 std::lock_guard<std::mutex> lock(impl_->mtx);
HITCBC 59   2283 if(impl_->queue.empty()) 59   2273 if(impl_->queue.empty())
HITCBC 60   4520 return; 60   4500 return;
HITCBC 61   23 c = impl_->queue.front(); 61   23 c = impl_->queue.front();
HITCBC 62   23 impl_->queue.pop(); 62   23 impl_->queue.pop();
HITCBC 63   2283 } 63   2273 }
HITCBC 64   23 capy::abandon(c); 64   23 capy::abandon(c);
HITCBC 65   23 } 65   23 }
HITCBC 66   1130 }; 66   1125 };
HITCBC 67   1130 shutdown(); 67   1125 shutdown();
68   // Takes what the unwinds post too, and the work of any service they 68   // Takes what the unwinds post too, and the work of any service they
69   // create, which is shut down as it is created. 69   // create, which is shut down as it is created.
HITCBC 70   1130 drain(); 70   1125 drain();
HITCBC 71   1130 destroy(); 71   1125 destroy();
72   // Services give up their work in shutdown(), but anything a 72   // Services give up their work in shutdown(), but anything a
73   // destructor posts anyway is still destroyed. 73   // destructor posts anyway is still destroyed.
HITCBC 74   1130 drain(); 74   1125 drain();
HITCBC 75   1130 delete impl_; 75   1125 delete impl_;
HITCBC 76   1130 } 76   1125 }
77   77  
78   blocking_executor 78   blocking_executor
HITCBC 79   1131 blocking_context::get_executor() noexcept 79   1126 blocking_context::get_executor() noexcept
80   { 80   {
HITCBC 81   1131 return blocking_executor{this}; 81   1126 return blocking_executor{this};
82   } 82   }
83   83  
84   void 84   void
HITCBC 85   783 blocking_context::signal_done() noexcept 85   778 blocking_context::signal_done() noexcept
86   { 86   {
HITCBC 87   783 std::lock_guard<std::mutex> lock(impl_->mtx); 87   778 std::lock_guard<std::mutex> lock(impl_->mtx);
HITCBC 88   783 impl_->done = true; 88   778 impl_->done = true;
HITCBC 89   783 impl_->cv.notify_one(); 89   778 impl_->cv.notify_one();
HITCBC 90   783 } 90   778 }
91   91  
92   void 92   void
HITCBC 93   336 blocking_context::signal_done( 93   336 blocking_context::signal_done(
94   std::exception_ptr ep) noexcept 94   std::exception_ptr ep) noexcept
95   { 95   {
HITCBC 96   336 std::lock_guard<std::mutex> lock(impl_->mtx); 96   336 std::lock_guard<std::mutex> lock(impl_->mtx);
HITCBC 97   336 impl_->ep = ep; 97   336 impl_->ep = ep;
HITCBC 98   336 impl_->done = true; 98   336 impl_->done = true;
HITCBC 99   336 impl_->cv.notify_one(); 99   336 impl_->cv.notify_one();
HITCBC 100   336 } 100   336 }
101   101  
102   void 102   void
HITCBC 103   1 blocking_context::signal_abandoned() noexcept 103   1 blocking_context::signal_abandoned() noexcept
104   { 104   {
HITCBC 105   1 std::lock_guard<std::mutex> lock(impl_->mtx); 105   1 std::lock_guard<std::mutex> lock(impl_->mtx);
HITCBC 106   1 impl_->abandoned = true; 106   1 impl_->abandoned = true;
HITCBC 107   1 impl_->done = true; 107   1 impl_->done = true;
HITCBC 108   1 impl_->cv.notify_one(); 108   1 impl_->cv.notify_one();
HITCBC 109   1 } 109   1 }
110   110  
111   void 111   void
HITCBC 112   1120 blocking_context::run() 112   1115 blocking_context::run()
113   { 113   {
114   struct scoped_running 114   struct scoped_running
115   { 115   {
116   blocking_context const* prev_; 116   blocking_context const* prev_;
117   117  
HITCBC 118   1120 explicit scoped_running(blocking_context const* ctx) noexcept 118   1115 explicit scoped_running(blocking_context const* ctx) noexcept
HITCBC 119   1120 : prev_(std::exchange(running_, ctx)) 119   1115 : prev_(std::exchange(running_, ctx))
120   { 120   {
HITCBC 121   1120 } 121   1115 }
122   122  
HITCBC 123   1120 ~scoped_running() 123   1115 ~scoped_running()
124   { 124   {
HITCBC 125   1120 running_ = prev_; 125   1115 running_ = prev_;
HITCBC 126   1120 } 126   1115 }
HITCBC 127   1120 } guard(this); 127   1115 } guard(this);
128   128  
129   for(;;) 129   for(;;)
130   { 130   {
HITCBC 131   2362 continuation c; 131   2352 continuation c;
132   { 132   {
HITCBC 133   2362 std::unique_lock<std::mutex> lock(impl_->mtx); 133   2352 std::unique_lock<std::mutex> lock(impl_->mtx);
HITCBC 134   2362 impl_->cv.wait(lock, [&] { 134   2352 impl_->cv.wait(lock, [&] {
HITCBC 135   2363 return impl_->done || !impl_->queue.empty(); 135   2353 return impl_->done || !impl_->queue.empty();
136   }); 136   });
HITCBC 137   2362 if(impl_->done && impl_->queue.empty()) 137   2352 if(impl_->done && impl_->queue.empty())
HITCBC 138   1120 break; 138   1115 break;
HITCBC 139   1242 c = impl_->queue.front(); 139   1237 c = impl_->queue.front();
HITCBC 140   1242 impl_->queue.pop(); 140   1237 impl_->queue.pop();
HITCBC 141   2362 } 141   2352 }
HITCBC 142   1242 safe_resume(c.h); 142   1237 safe_resume(c.h);
HITCBC 143   1242 } 143   1237 }
HITCBC 144   1120 if(impl_->ep) 144   1115 if(impl_->ep)
HITCBC 145   672 std::rethrow_exception(impl_->ep); 145   672 std::rethrow_exception(impl_->ep);
HITCBC 146   784 if(impl_->abandoned) 146   779 if(impl_->abandoned)
HITCBC 147   1 throw std::system_error(error::canceled, "task abandoned"); 147   1 throw std::system_error(error::canceled, "task abandoned");
HITCBC 148   1120 } 148   1115 }
149   149  
150   void 150   void
HITCBC 151   1265 blocking_context::enqueue( 151   1260 blocking_context::enqueue(
152   continuation const& c) 152   continuation const& c)
153   { 153   {
HITCBC 154   1265 std::lock_guard<std::mutex> lock(impl_->mtx); 154   1260 std::lock_guard<std::mutex> lock(impl_->mtx);
HITCBC 155   1265 impl_->queue.push(c); 155   1260 impl_->queue.push(c);
HITCBC 156   1265 impl_->cv.notify_one(); 156   1260 impl_->cv.notify_one();
HITCBC 157   1265 } 157   1260 }
158   158  
159   //---------------------------------------------------------- 159   //----------------------------------------------------------
160   160  
161   bool 161   bool
MISUBC 162   ✗ blocking_executor::operator==( 162   ✗ blocking_executor::operator==(
163   blocking_executor const& other) const noexcept 163   blocking_executor const& other) const noexcept
164   { 164   {
MISUBC 165   ✗ return ctx_ == other.ctx_; 165   ✗ return ctx_ == other.ctx_;
166   } 166   }
167   167  
168   blocking_context& 168   blocking_context&
HITCBC 169   1122 blocking_executor::context() const noexcept 169   1117 blocking_executor::context() const noexcept
170   { 170   {
HITCBC 171   1122 return *ctx_; 171   1117 return *ctx_;
172   } 172   }
173   173  
174   void 174   void
HITCBC 175   1125 blocking_executor::on_work_started() const noexcept 175   1120 blocking_executor::on_work_started() const noexcept
176   { 176   {
HITCBC 177   1125 } 177   1120 }
178   178  
179   void 179   void
HITCBC 180   1125 blocking_executor::on_work_finished() const noexcept 180   1120 blocking_executor::on_work_finished() const noexcept
181   { 181   {
HITCBC 182   1125 } 182   1120 }
183   183  
184   std::coroutine_handle<> 184   std::coroutine_handle<>
HITCBC 185   1168 blocking_executor::dispatch( 185   1163 blocking_executor::dispatch(
186   continuation& c) const 186   continuation& c) const
187   { 187   {
HITCBC 188   1168 if(running_ == ctx_) 188   1163 if(running_ == ctx_)
HITCBC 189   47 return c.h; 189   47 return c.h;
HITCBC 190   1121 ctx_->enqueue(c); 190   1116 ctx_->enqueue(c);
HITCBC 191   1121 return std::noop_coroutine(); 191   1116 return std::noop_coroutine();
192   } 192   }
193   193  
194   void 194   void
HITCBC 195   144 blocking_executor::post( 195   144 blocking_executor::post(
196   continuation& c) const 196   continuation& c) const
197   { 197   {
HITCBC 198   144 ctx_->enqueue(c); 198   144 ctx_->enqueue(c);
HITCBC 199   144 } 199   144 }
200   200  
201   } // namespace test 201   } // namespace test
202   } // namespace capy 202   } // namespace capy
203   } // namespace boost 203   } // namespace boost