blob: 4807a6a0953324ba68bf453f1c0288db5b47e1a8 [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{
84 explicit wait_process_sender(context& ctx) : ctx(ctx){};
85
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 }
113}
114
115} // namespace details
116
117context::~context() noexcept(false)
118{
119 if (worker_thread.joinable())
120 {
121 throw std::logic_error(
122 "sdbusplus::async::context destructed without completion.");
123 }
124}
125
Patrick Williams73e278b2022-09-16 08:31:36 -0500126bool context::request_stop() noexcept
127{
Patrick Williams78e436f2022-09-21 10:06:20 -0500128 auto first_stop = initial_stop.request_stop();
Patrick Williams73e278b2022-09-16 08:31:36 -0500129
130 if (first_stop)
131 {
Patrick Williams78e436f2022-09-21 10:06:20 -0500132 // Now that the workers have been requested to stop, we need to wait
133 // until they all drain and then stop the internal tasks.
Patrick Williams4cfc2842022-09-22 09:53:33 -0500134
135 auto complete = [this]() {
136 final_stop.request_stop();
137 caller_wait.notify_one();
138 event_loop.break_run();
139 };
140
141 internal_tasks.spawn(pending_tasks.empty() | execution::then(complete) |
142 execution::upon_error([=](auto&& e) {
143 complete();
144 std::rethrow_exception(e);
Patrick Williams78e436f2022-09-21 10:06:20 -0500145 }));
Patrick Williams73e278b2022-09-16 08:31:36 -0500146 }
147
148 return first_stop;
149}
150
Patrick Williams74187662022-08-26 19:28:24 -0500151void context::caller_run(task<> startup)
152{
153 // Start up the worker thread.
154 worker_thread = std::thread{[this, startup = std::move(startup)]() mutable {
155 worker_run(std::move(startup));
156 }};
157
Patrick Williams73e278b2022-09-16 08:31:36 -0500158 // Run until the context requested to stop.
Patrick Williams78e436f2022-09-21 10:06:20 -0500159 while (!final_stop.stop_requested())
Patrick Williams74187662022-08-26 19:28:24 -0500160 {
Patrick Williams73e278b2022-09-16 08:31:36 -0500161 // Handle waiting on all the sd-events.
Patrick Williams74187662022-08-26 19:28:24 -0500162 details::wait_process_completion::wait_once(*this);
163 }
164
Patrick Williams4cfc2842022-09-22 09:53:33 -0500165 std::exception_ptr pending_exception{};
166
Patrick Williams78e436f2022-09-21 10:06:20 -0500167 // Wait for all the internal tasks to complete.
Patrick Williams4cfc2842022-09-22 09:53:33 -0500168 std::this_thread::sync_wait(internal_tasks.empty() |
169 execution::upon_error([&](auto&& e) {
170 pending_exception = std::move(e);
171 }));
Patrick Williamsc5b5ff52022-09-21 08:34:22 -0500172
Patrick Williams73e278b2022-09-16 08:31:36 -0500173 // Stop has been requested, so finish up the loop.
Patrick Williams74187662022-08-26 19:28:24 -0500174 loop.finish();
175 if (worker_thread.joinable())
176 {
177 worker_thread.join();
178 }
Patrick Williams4cfc2842022-09-22 09:53:33 -0500179
180 // If there was an exception inside the context, rethrow it.
181 if (pending_exception)
182 {
183 std::rethrow_exception(std::move(pending_exception));
184 }
Patrick Williams74187662022-08-26 19:28:24 -0500185}
186
187void context::worker_run(task<> startup)
188{
189 // Begin the 'startup' task.
190 // This shouldn't start detached because we want to be able to forward
191 // failures back to the 'run'. execution::ensure_started isn't
192 // implemented yet, so we don't have a lot of other options.
Patrick Williams0139cac2022-09-20 17:04:17 -0500193 spawn(std::move(startup));
Patrick Williams74187662022-08-26 19:28:24 -0500194
Patrick Williams78e436f2022-09-21 10:06:20 -0500195 // Also start the sdbus 'wait/process' loop; treat it as an internal task.
196 internal_tasks.spawn(details::wait_process_completion::loop(*this));
Patrick Williams74187662022-08-26 19:28:24 -0500197
198 // Run the execution::run_loop to handle all the tasks.
199 loop.run();
200}
201
202void details::wait_process_completion::arm() noexcept
203{
204 // Call process. True indicates something was handled and we do not
205 // need to `wait`, because there might be yet another pending operation
206 // to process, so immediately signal the operation as complete.
207 if (ctx.get_bus().process_discard())
208 {
209 this->complete();
210 return;
211 }
212
Patrick Williams73e278b2022-09-16 08:31:36 -0500213 // We need to call wait now, get the current timeout and stage ourselves
214 // as the next completion.
Patrick Williams74187662022-08-26 19:28:24 -0500215
216 // Get the bus' timeout.
Patrick Williams73e278b2022-09-16 08:31:36 -0500217 uint64_t to_usec = 0;
218 sd_bus_get_timeout(get_busp(ctx.get_bus()), &to_usec);
Patrick Williams74187662022-08-26 19:28:24 -0500219
Patrick Williams73e278b2022-09-16 08:31:36 -0500220 if (to_usec == UINT64_MAX)
Patrick Williams74187662022-08-26 19:28:24 -0500221 {
222 // sd_bus_get_timeout returns UINT64_MAX to indicate 'wait forever'.
Patrick Williams73e278b2022-09-16 08:31:36 -0500223 // Turn this into -1 for sd-event.
224 timeout = std::chrono::microseconds{-1};
Patrick Williams74187662022-08-26 19:28:24 -0500225 }
226 else
227 {
Patrick Williams73e278b2022-09-16 08:31:36 -0500228 timeout = std::chrono::microseconds(to_usec);
Patrick Williams74187662022-08-26 19:28:24 -0500229 }
230
231 // Assign ourselves as the pending completion and release the caller.
232 std::lock_guard lock{ctx.lock};
Patrick Williams73e278b2022-09-16 08:31:36 -0500233 ctx.staged = this;
Patrick Williams74187662022-08-26 19:28:24 -0500234 ctx.caller_wait.notify_one();
235}
236
237void details::wait_process_completion::wait_once(context& ctx)
238{
Patrick Williams74187662022-08-26 19:28:24 -0500239 // Scope for lock.
240 {
241 std::unique_lock lock{ctx.lock};
242
243 // If there isn't a completion waiting already, wait on the condition
244 // variable for one to show up (we can't call `poll` yet because we
245 // don't have the required parameters).
Patrick Williams73e278b2022-09-16 08:31:36 -0500246 ctx.caller_wait.wait(lock, [&] {
247 return (ctx.pending != nullptr) || (ctx.staged != nullptr) ||
Patrick Williams78e436f2022-09-21 10:06:20 -0500248 (ctx.final_stop.stop_requested());
Patrick Williams73e278b2022-09-16 08:31:36 -0500249 });
Patrick Williams74187662022-08-26 19:28:24 -0500250
Patrick Williams73e278b2022-09-16 08:31:36 -0500251 // Save the waiter as pending.
252 if (ctx.pending == nullptr)
253 {
254 ctx.pending = std::exchange(ctx.staged, nullptr);
255 }
Patrick Williams74187662022-08-26 19:28:24 -0500256 }
257
Patrick Williams73e278b2022-09-16 08:31:36 -0500258 // If the context has been requested to be stopped, exit now instead of
259 // running the context event loop.
Patrick Williams78e436f2022-09-21 10:06:20 -0500260 if (ctx.final_stop.stop_requested())
Patrick Williams73e278b2022-09-16 08:31:36 -0500261 {
262 return;
263 }
264
265 // Run the event loop to process one request.
266 ctx.event_loop.run_one(ctx.pending->timeout);
267
268 // If there is a stop requested, we need to stop the pending operation.
Patrick Williams78e436f2022-09-21 10:06:20 -0500269 if (ctx.final_stop.stop_requested())
Patrick Williams73e278b2022-09-16 08:31:36 -0500270 {
271 decltype(ctx.pending) pending = nullptr;
272
273 {
274 std::lock_guard lock{ctx.lock};
275 pending = std::exchange(ctx.pending, nullptr);
276 }
277
278 // Do the stop outside the lock to prevent potential deadlocks due to
279 // the stop handler running.
280 if (pending != nullptr)
281 {
282 pending->stop();
283 }
284 }
285}
286
287int context::dbus_event_handle(sd_event_source*, int, uint32_t, void* data)
288{
289 auto self = static_cast<context*>(data);
290
291 decltype(self->pending) pending = nullptr;
292 {
293 std::lock_guard lock{self->lock};
294 pending = std::exchange(self->pending, nullptr);
295 }
296
297 // Outside the lock complete the pending operation.
298 //
299 // This can cause the Receiver task (the worker) to start executing (on
300 // this thread!), hence we do not want the lock held in order to avoid
301 // deadlocks.
302 if (pending != nullptr)
303 {
Patrick Williams78e436f2022-09-21 10:06:20 -0500304 if (self->final_stop.stop_requested())
Patrick Williams73e278b2022-09-16 08:31:36 -0500305 {
306 pending->stop();
307 }
308 else
309 {
310 pending->complete();
311 }
312 }
313
314 return 0;
Patrick Williams74187662022-08-26 19:28:24 -0500315}
316
317} // namespace sdbusplus::async