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