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