blob: 2f6454d9611dc8ab908eb09697bd426dc631f4ab [file] [log] [blame]
Patrick Williams74187662022-08-26 19:28:24 -05001#include <systemd/sd-bus.h>
2
3#include <sdbusplus/async/context.hpp>
4
Patrick Williams73e278b2022-09-16 08:31:36 -05005#include <chrono>
6
Patrick Williams74187662022-08-26 19:28:24 -05007namespace sdbusplus::async
8{
9
Patrick Williams73e278b2022-09-16 08:31:36 -050010context::context(bus_t&& b) : bus(std::move(b))
11{
12 dbus_source = event_loop.add_io(bus.get_fd(), EPOLLIN, dbus_event_handle,
13 this);
14}
15
Patrick Williams74187662022-08-26 19:28:24 -050016namespace details
17{
18
19/* The sd_bus_wait/process completion event.
20 *
21 * The wait/process handshake is modelled as a Sender with the the worker
22 * task `co_await`ing Senders over and over. This class is the completion
23 * handling for the Sender (to get it back to the Receiver, ie. the worker).
24 */
25struct wait_process_completion : bus::details::bus_friend
26{
27 explicit wait_process_completion(context& ctx) : ctx(ctx) {}
28 virtual ~wait_process_completion() = default;
29
30 // Called by the `caller` to indicate the Sender is completed.
31 virtual void complete() noexcept = 0;
Patrick Williams73e278b2022-09-16 08:31:36 -050032 // Called by the `caller` to indicate the Sender should be stopped.
33 virtual void stop() noexcept = 0;
Patrick Williams74187662022-08-26 19:28:24 -050034
35 // Arm the completion event.
36 void arm() noexcept;
37
38 // Data to share with the worker.
39 context& ctx;
Patrick Williams435eb1b2022-09-16 16:22:07 -050040 event_t::time_resolution timeout{};
Patrick Williams74187662022-08-26 19:28:24 -050041
42 static task<> loop(context& ctx);
43 static void wait_once(context& ctx);
44};
45
46/* The completion template based on receiver type.
47 *
48 * The type of the receivers (typically the co_awaiter) is only known by
49 * a template, so we need a sub-class of completion to hold the receiver.
50 */
51template <execution::receiver R>
52struct wait_process_operation : public wait_process_completion
53{
54 wait_process_operation(context& ctx, R r) :
55 wait_process_completion(ctx), receiver(std::move(r))
56 {}
57
58 wait_process_operation(wait_process_operation&&) = delete;
59
60 void complete() noexcept override final
61 {
62 execution::set_value(std::move(this->receiver));
63 }
64
Patrick Williams73e278b2022-09-16 08:31:36 -050065 void stop() noexcept override final
66 {
67 // Stop can be called when the context is shutting down,
68 // so treat it as if the receiver completed.
69 execution::set_value(std::move(this->receiver));
70 }
71
Patrick Williams74187662022-08-26 19:28:24 -050072 friend void tag_invoke(execution::start_t,
73 wait_process_operation& self) noexcept
74 {
75 self.arm();
76 }
77
78 R receiver;
79};
80
81/* The sender for the wait/process event. */
82struct wait_process_sender
83{
Ed Tanousb8be5992023-01-04 15:56:20 -080084 explicit wait_process_sender(context& ctx) : ctx(ctx) {}
Patrick Williams74187662022-08-26 19:28:24 -050085
86 friend auto tag_invoke(execution::get_completion_signatures_t,
87 const wait_process_sender&, auto)
88 -> execution::completion_signatures<execution::set_value_t()>;
89
90 template <execution::receiver R>
91 friend auto tag_invoke(execution::connect_t, wait_process_sender&& self,
92 R r) -> wait_process_operation<R>
93 {
94 // Create the completion for the wait.
95 return {self.ctx, std::move(r)};
96 }
97
98 private:
99 context& ctx;
100};
101
102task<> wait_process_completion::loop(context& ctx)
103{
Patrick Williams78e436f2022-09-21 10:06:20 -0500104 while (!ctx.final_stop.stop_requested())
Patrick Williams74187662022-08-26 19:28:24 -0500105 {
106 // Handle the next sdbus event.
107 co_await wait_process_sender(ctx);
108
109 // Completion likely happened on the context 'caller' thread, so
110 // we need to switch to the worker thread.
111 co_await execution::schedule(ctx.loop.get_scheduler());
112 }
Patrick Williams10483c92022-09-23 11:33:39 -0500113
114 {
115 std::lock_guard lock{ctx.lock};
116 ctx.wait_process_stopped = true;
117 }
Patrick Williams74187662022-08-26 19:28:24 -0500118}
119
120} // namespace details
121
122context::~context() noexcept(false)
123{
124 if (worker_thread.joinable())
125 {
126 throw std::logic_error(
127 "sdbusplus::async::context destructed without completion.");
128 }
129}
130
Patrick Williams3c242ba2022-09-23 09:51:55 -0500131void context::run()
Patrick Williams74187662022-08-26 19:28:24 -0500132{
Patrick Williams10483c92022-09-23 11:33:39 -0500133 // Run the primary portion of the run-loop.
134 caller_run();
Patrick Williams74187662022-08-26 19:28:24 -0500135
Patrick Williams10483c92022-09-23 11:33:39 -0500136 // Rethrow the pending exception (if it exists).
137 rethrow_pending_exception();
Patrick Williams74187662022-08-26 19:28:24 -0500138
Patrick Williams10483c92022-09-23 11:33:39 -0500139 // Otherwise this should be final_stop...
140
141 // We need to wait for the pending wait process and stop it.
142 wait_for_wait_process_stopped();
Patrick Williams4cfc2842022-09-22 09:53:33 -0500143
Patrick Williams78e436f2022-09-21 10:06:20 -0500144 // Wait for all the internal tasks to complete.
Patrick Williams5e7ef082023-01-05 11:08:53 -0600145 stdexec::this_thread::sync_wait(
Patrick Williams115e9b32022-09-23 16:20:59 -0500146 internal_tasks.empty() | execution::upon_error([&](auto&& e) {
147 pending_exceptions.emplace_back(std::move(e));
148 }));
Patrick Williamsc5b5ff52022-09-21 08:34:22 -0500149
Patrick Williams10483c92022-09-23 11:33:39 -0500150 // Finish up the loop and join the thread.
151 // (There shouldn't be anything going on by this point anyhow.)
Patrick Williams74187662022-08-26 19:28:24 -0500152 loop.finish();
153 if (worker_thread.joinable())
154 {
155 worker_thread.join();
156 }
Patrick Williams4cfc2842022-09-22 09:53:33 -0500157
Patrick Williams10483c92022-09-23 11:33:39 -0500158 // Check for one last exception.
159 rethrow_pending_exception();
Patrick Williams74187662022-08-26 19:28:24 -0500160}
161
Patrick Williams3c242ba2022-09-23 09:51:55 -0500162void context::worker_run()
Patrick Williams74187662022-08-26 19:28:24 -0500163{
Patrick Williams3c242ba2022-09-23 09:51:55 -0500164 // Start the sdbus 'wait/process' loop; treat it as an internal task.
Patrick Williams10483c92022-09-23 11:33:39 -0500165 internal_tasks.spawn(details::wait_process_completion::loop(*this) |
166 execution::upon_stopped([]() {}));
Patrick Williams74187662022-08-26 19:28:24 -0500167
168 // Run the execution::run_loop to handle all the tasks.
169 loop.run();
170}
171
Patrick Williams10483c92022-09-23 11:33:39 -0500172void context::spawn_complete(std::exception_ptr&& e)
173{
174 {
175 std::lock_guard l{lock};
176 spawn_watcher_running = false;
177
178 if (e)
179 {
Patrick Williams115e9b32022-09-23 16:20:59 -0500180 pending_exceptions.emplace_back(std::move(e));
Patrick Williams10483c92022-09-23 11:33:39 -0500181 }
182 }
183
184 if (stop_requested())
185 {
186 final_stop.request_stop();
187 }
188
189 caller_wait.notify_one();
190 event_loop.break_run();
191}
192
193void context::check_stop_requested()
194{
195 if (stop_requested())
196 {
197 throw std::logic_error(
198 "sdbusplus::async::context spawn called while already stopped.");
199 }
200}
201
202void context::spawn_watcher()
203{
204 {
205 std::lock_guard l{lock};
206 if (spawn_watcher_running)
207 {
208 return;
209 }
210
211 spawn_watcher_running = true;
212 }
213
214 // Spawn the watch for completion / exceptions.
215 internal_tasks.spawn(pending_tasks.empty() |
216 execution::then([this]() { spawn_complete(); }) |
217 execution::upon_error([this](auto&& e) {
218 spawn_complete(std::move(e));
219 }));
220}
221
222void context::caller_run()
223{
224 // We are able to run the loop until the context is requested to stop or
225 // we get an exception.
226 auto keep_running = [this]() {
227 std::lock_guard l{lock};
Patrick Williams115e9b32022-09-23 16:20:59 -0500228 return !final_stop.stop_requested() && pending_exceptions.empty();
Patrick Williams10483c92022-09-23 11:33:39 -0500229 };
230
231 // If we are suppose to keep running, start the run loop.
232 if (keep_running())
233 {
234 // Start up the worker thread.
235 if (!worker_thread.joinable())
236 {
237 worker_thread = std::thread{[this]() { worker_run(); }};
238 }
239 else
240 {
241 // We've already been running and there might an exception or
242 // completion pending. Spawn a new watcher that checks for these.
243 spawn_watcher();
244 }
245
246 while (keep_running())
247 {
248 // Handle waiting on all the sd-events.
249 details::wait_process_completion::wait_once(*this);
250 }
251 }
252 else
253 {
254 // There might be pending exceptions still, so spawn a watcher for them.
255 spawn_watcher();
256 }
257}
258
259void context::wait_for_wait_process_stopped()
260{
261 auto worker = std::exchange(pending, nullptr);
262 while (worker == nullptr)
263 {
264 std::lock_guard l{lock};
265 if (wait_process_stopped)
266 {
267 break;
268 }
269
270 worker = std::exchange(staged, nullptr);
271 if (!worker)
272 {
273 std::this_thread::yield();
274 }
275 }
276 if (worker)
277 {
278 worker->stop();
279 wait_process_stopped = true;
280 }
281}
282
283void context::rethrow_pending_exception()
284{
285 {
286 std::lock_guard l{lock};
Patrick Williams115e9b32022-09-23 16:20:59 -0500287 if (!pending_exceptions.empty())
Patrick Williams10483c92022-09-23 11:33:39 -0500288 {
Patrick Williams115e9b32022-09-23 16:20:59 -0500289 auto e = pending_exceptions.front();
290 pending_exceptions.pop_front();
291 std::rethrow_exception(std::move(e));
Patrick Williams10483c92022-09-23 11:33:39 -0500292 }
293 }
294}
295
Patrick Williams74187662022-08-26 19:28:24 -0500296void details::wait_process_completion::arm() noexcept
297{
298 // Call process. True indicates something was handled and we do not
299 // need to `wait`, because there might be yet another pending operation
300 // to process, so immediately signal the operation as complete.
301 if (ctx.get_bus().process_discard())
302 {
303 this->complete();
304 return;
305 }
306
Patrick Williams73e278b2022-09-16 08:31:36 -0500307 // We need to call wait now, get the current timeout and stage ourselves
308 // as the next completion.
Patrick Williams74187662022-08-26 19:28:24 -0500309
310 // Get the bus' timeout.
Patrick Williams73e278b2022-09-16 08:31:36 -0500311 uint64_t to_usec = 0;
312 sd_bus_get_timeout(get_busp(ctx.get_bus()), &to_usec);
Patrick Williams74187662022-08-26 19:28:24 -0500313
Patrick Williams73e278b2022-09-16 08:31:36 -0500314 if (to_usec == UINT64_MAX)
Patrick Williams74187662022-08-26 19:28:24 -0500315 {
316 // sd_bus_get_timeout returns UINT64_MAX to indicate 'wait forever'.
Patrick Williams73e278b2022-09-16 08:31:36 -0500317 // Turn this into -1 for sd-event.
318 timeout = std::chrono::microseconds{-1};
Patrick Williams74187662022-08-26 19:28:24 -0500319 }
320 else
321 {
Patrick Williams73e278b2022-09-16 08:31:36 -0500322 timeout = std::chrono::microseconds(to_usec);
Patrick Williams74187662022-08-26 19:28:24 -0500323 }
324
325 // Assign ourselves as the pending completion and release the caller.
326 std::lock_guard lock{ctx.lock};
Patrick Williams73e278b2022-09-16 08:31:36 -0500327 ctx.staged = this;
Patrick Williams74187662022-08-26 19:28:24 -0500328 ctx.caller_wait.notify_one();
329}
330
331void details::wait_process_completion::wait_once(context& ctx)
332{
Patrick Williams74187662022-08-26 19:28:24 -0500333 // Scope for lock.
334 {
335 std::unique_lock lock{ctx.lock};
336
337 // If there isn't a completion waiting already, wait on the condition
338 // variable for one to show up (we can't call `poll` yet because we
339 // don't have the required parameters).
Patrick Williams73e278b2022-09-16 08:31:36 -0500340 ctx.caller_wait.wait(lock, [&] {
341 return (ctx.pending != nullptr) || (ctx.staged != nullptr) ||
Patrick Williams10483c92022-09-23 11:33:39 -0500342 ctx.final_stop.stop_requested() ||
Patrick Williams115e9b32022-09-23 16:20:59 -0500343 !ctx.pending_exceptions.empty();
Patrick Williams73e278b2022-09-16 08:31:36 -0500344 });
Patrick Williams74187662022-08-26 19:28:24 -0500345
Patrick Williams73e278b2022-09-16 08:31:36 -0500346 // Save the waiter as pending.
347 if (ctx.pending == nullptr)
348 {
349 ctx.pending = std::exchange(ctx.staged, nullptr);
350 }
Patrick Williams74187662022-08-26 19:28:24 -0500351 }
352
Patrick Williams73e278b2022-09-16 08:31:36 -0500353 // Run the event loop to process one request.
Patrick Williams10483c92022-09-23 11:33:39 -0500354 // If the context has been requested to be stopped, skip the event loop.
355 if (!ctx.final_stop.stop_requested() && ctx.pending)
Patrick Williams73e278b2022-09-16 08:31:36 -0500356 {
Patrick Williams10483c92022-09-23 11:33:39 -0500357 ctx.event_loop.run_one(ctx.pending->timeout);
Patrick Williams73e278b2022-09-16 08:31:36 -0500358 }
359}
360
361int context::dbus_event_handle(sd_event_source*, int, uint32_t, void* data)
362{
363 auto self = static_cast<context*>(data);
364
Patrick Williams10483c92022-09-23 11:33:39 -0500365 auto pending = std::exchange(self->pending, nullptr);
Patrick Williams73e278b2022-09-16 08:31:36 -0500366 if (pending != nullptr)
367 {
Patrick Williams10483c92022-09-23 11:33:39 -0500368 pending->complete();
Patrick Williams73e278b2022-09-16 08:31:36 -0500369 }
370
371 return 0;
Patrick Williams74187662022-08-26 19:28:24 -0500372}
373
374} // namespace sdbusplus::async