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 1130 : blocking_context::blocking_context()
43 1130 : : impl_(new impl)
44 : {
45 1130 : }
46 :
47 1130 : 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 2263 : auto drain = [this]
54 : {
55 2263 : bool any = false;
56 : for(;;)
57 : {
58 4270 : continuation c;
59 : {
60 4270 : std::lock_guard<std::mutex> lock(impl_->mtx);
61 4270 : if(impl_->queue.empty())
62 4526 : return any;
63 2007 : c = impl_->queue.front();
64 2007 : impl_->queue.pop();
65 4270 : }
66 2007 : any = true;
67 2007 : capy::abandon(c);
68 2007 : }
69 1130 : };
70 1130 : shutdown();
71 1130 : 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 1133 : destroy();
77 : }
78 1133 : while(drain());
79 1130 : delete impl_;
80 1130 : }
81 :
82 : blocking_executor
83 1131 : blocking_context::get_executor() noexcept
84 : {
85 1131 : return blocking_executor{this};
86 : }
87 :
88 : void
89 783 : blocking_context::signal_done() noexcept
90 : {
91 783 : std::lock_guard<std::mutex> lock(impl_->mtx);
92 783 : impl_->done = true;
93 783 : impl_->cv.notify_one();
94 783 : }
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 1120 : blocking_context::run()
117 : {
118 : struct scoped_running
119 : {
120 : blocking_context const* prev_;
121 :
122 1120 : explicit scoped_running(blocking_context const* ctx) noexcept
123 1120 : : prev_(std::exchange(running_, ctx))
124 : {
125 1120 : }
126 :
127 1120 : ~scoped_running()
128 : {
129 1120 : running_ = prev_;
130 1120 : }
131 1120 : } guard(this);
132 :
133 : for(;;)
134 : {
135 2362 : continuation c;
136 : {
137 2362 : std::unique_lock<std::mutex> lock(impl_->mtx);
138 2362 : impl_->cv.wait(lock, [&] {
139 2363 : return impl_->done || !impl_->queue.empty();
140 : });
141 2362 : if(impl_->done && impl_->queue.empty())
142 1120 : break;
143 1242 : c = impl_->queue.front();
144 1242 : impl_->queue.pop();
145 2362 : }
146 1242 : safe_resume(c.h);
147 1242 : }
148 1120 : if(impl_->ep)
149 672 : std::rethrow_exception(impl_->ep);
150 784 : if(impl_->abandoned)
151 1 : throw std::system_error(error::canceled, "task abandoned");
152 1120 : }
153 :
154 : void
155 3249 : blocking_context::enqueue(
156 : continuation const& c)
157 : {
158 3249 : std::lock_guard<std::mutex> lock(impl_->mtx);
159 3249 : impl_->queue.push(c);
160 3249 : impl_->cv.notify_one();
161 3249 : }
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 1122 : blocking_executor::context() const noexcept
174 : {
175 1122 : return *ctx_;
176 : }
177 :
178 : void
179 1125 : blocking_executor::on_work_started() const noexcept
180 : {
181 1125 : }
182 :
183 : void
184 1125 : blocking_executor::on_work_finished() const noexcept
185 : {
186 1125 : }
187 :
188 : std::coroutine_handle<>
189 1168 : blocking_executor::dispatch(
190 : continuation& c) const
191 : {
192 1168 : if(running_ == ctx_)
193 47 : return c.h;
194 1121 : ctx_->enqueue(c);
195 1121 : 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
|
