TLA Line data Source code
1 : //
2 : // Copyright (c) 2025 Vinnie Falco (vinnie.falco@gmail.com)
3 : //
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)
6 : //
7 : // Official repository: https://github.com/cppalliance/capy
8 : //
9 :
10 : #include <boost/capy/test/run_blocking.hpp>
11 :
12 : #include <boost/capy/ex/abandon.hpp>
13 : #include <boost/capy/ex/frame_allocator.hpp>
14 : #include <condition_variable>
15 : #include <mutex>
16 : #include <queue>
17 : #include <system_error>
18 : #include <utility>
19 :
20 : namespace boost {
21 : namespace capy {
22 : namespace test {
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 :
32 : struct blocking_context::impl
33 : {
34 : std::mutex mtx;
35 : std::condition_variable cv;
36 : std::queue<continuation> queue;
37 : std::exception_ptr ep;
38 : bool done = false;
39 : bool abandoned = false;
40 : };
41 :
42 HIT 1125 : blocking_context::blocking_context()
43 1125 : : impl_(new impl)
44 : {
45 1125 : }
46 :
47 1125 : blocking_context::~blocking_context()
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.
53 2254 : auto drain = [this]
54 : {
55 2254 : bool any = false;
56 : for(;;)
57 : {
58 4261 : continuation c;
59 : {
60 4261 : std::lock_guard<std::mutex> lock(impl_->mtx);
61 4261 : if(impl_->queue.empty())
62 4508 : return any;
63 2007 : c = impl_->queue.front();
64 2007 : impl_->queue.pop();
65 4261 : }
66 2007 : any = true;
67 2007 : capy::abandon(c);
68 2007 : }
69 1125 : };
70 1125 : shutdown();
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 : {
76 1129 : destroy();
77 : }
78 1129 : while(drain());
79 1125 : delete impl_;
80 1125 : }
81 :
82 : blocking_executor
83 1126 : blocking_context::get_executor() noexcept
84 : {
85 1126 : return blocking_executor{this};
86 : }
87 :
88 : void
89 778 : blocking_context::signal_done() noexcept
90 : {
91 778 : std::lock_guard<std::mutex> lock(impl_->mtx);
92 778 : impl_->done = true;
93 778 : impl_->cv.notify_one();
94 778 : }
95 :
96 : void
97 336 : blocking_context::signal_done(
98 : std::exception_ptr ep) noexcept
99 : {
100 336 : std::lock_guard<std::mutex> lock(impl_->mtx);
101 336 : impl_->ep = ep;
102 336 : impl_->done = true;
103 336 : impl_->cv.notify_one();
104 336 : }
105 :
106 : void
107 1 : blocking_context::signal_abandoned() noexcept
108 : {
109 1 : std::lock_guard<std::mutex> lock(impl_->mtx);
110 1 : impl_->abandoned = true;
111 1 : impl_->done = true;
112 1 : impl_->cv.notify_one();
113 1 : }
114 :
115 : void
116 1115 : blocking_context::run()
117 : {
118 : struct scoped_running
119 : {
120 : blocking_context const* prev_;
121 :
122 1115 : explicit scoped_running(blocking_context const* ctx) noexcept
123 1115 : : prev_(std::exchange(running_, ctx))
124 : {
125 1115 : }
126 :
127 1115 : ~scoped_running()
128 : {
129 1115 : running_ = prev_;
130 1115 : }
131 1115 : } guard(this);
132 :
133 : for(;;)
134 : {
135 2352 : continuation c;
136 : {
137 2352 : std::unique_lock<std::mutex> lock(impl_->mtx);
138 2352 : impl_->cv.wait(lock, [&] {
139 2353 : return impl_->done || !impl_->queue.empty();
140 : });
141 2352 : if(impl_->done && impl_->queue.empty())
142 1115 : break;
143 1237 : c = impl_->queue.front();
144 1237 : impl_->queue.pop();
145 2352 : }
146 1237 : safe_resume(c.h);
147 1237 : }
148 1115 : if(impl_->ep)
149 672 : std::rethrow_exception(impl_->ep);
150 779 : if(impl_->abandoned)
151 1 : throw std::system_error(error::canceled, "task abandoned");
152 1115 : }
153 :
154 : void
155 3244 : blocking_context::enqueue(
156 : continuation const& c)
157 : {
158 3244 : std::lock_guard<std::mutex> lock(impl_->mtx);
159 3244 : impl_->queue.push(c);
160 3244 : impl_->cv.notify_one();
161 3244 : }
162 :
163 : //----------------------------------------------------------
164 :
165 : bool
166 MIS 0 : blocking_executor::operator==(
167 : blocking_executor const& other) const noexcept
168 : {
169 0 : return ctx_ == other.ctx_;
170 : }
171 :
172 : blocking_context&
173 HIT 1117 : blocking_executor::context() const noexcept
174 : {
175 1117 : return *ctx_;
176 : }
177 :
178 : void
179 1120 : blocking_executor::on_work_started() const noexcept
180 : {
181 1120 : }
182 :
183 : void
184 1120 : blocking_executor::on_work_finished() const noexcept
185 : {
186 1120 : }
187 :
188 : std::coroutine_handle<>
189 1163 : blocking_executor::dispatch(
190 : continuation& c) const
191 : {
192 1163 : if(running_ == ctx_)
193 47 : return c.h;
194 1116 : ctx_->enqueue(c);
195 1116 : return std::noop_coroutine();
196 : }
197 :
198 : void
199 2128 : blocking_executor::post(
200 : continuation& c) const
201 : {
202 2128 : ctx_->enqueue(c);
203 2128 : }
204 :
205 : } // namespace test
206 : } // namespace capy
207 : } // namespace boost
|
