blob: 2f6454d9611dc8ab908eb09697bd426dc631f4ab [file] [log] [blame]
#include <systemd/sd-bus.h>
#include <sdbusplus/async/context.hpp>
#include <chrono>
namespace sdbusplus::async
{
context::context(bus_t&& b) : bus(std::move(b))
{
dbus_source = event_loop.add_io(bus.get_fd(), EPOLLIN, dbus_event_handle,
this);
}
namespace details
{
/* The sd_bus_wait/process completion event.
*
* The wait/process handshake is modelled as a Sender with the the worker
* task `co_await`ing Senders over and over. This class is the completion
* handling for the Sender (to get it back to the Receiver, ie. the worker).
*/
struct wait_process_completion : bus::details::bus_friend
{
explicit wait_process_completion(context& ctx) : ctx(ctx) {}
virtual ~wait_process_completion() = default;
// Called by the `caller` to indicate the Sender is completed.
virtual void complete() noexcept = 0;
// Called by the `caller` to indicate the Sender should be stopped.
virtual void stop() noexcept = 0;
// Arm the completion event.
void arm() noexcept;
// Data to share with the worker.
context& ctx;
event_t::time_resolution timeout{};
static task<> loop(context& ctx);
static void wait_once(context& ctx);
};
/* The completion template based on receiver type.
*
* The type of the receivers (typically the co_awaiter) is only known by
* a template, so we need a sub-class of completion to hold the receiver.
*/
template <execution::receiver R>
struct wait_process_operation : public wait_process_completion
{
wait_process_operation(context& ctx, R r) :
wait_process_completion(ctx), receiver(std::move(r))
{}
wait_process_operation(wait_process_operation&&) = delete;
void complete() noexcept override final
{
execution::set_value(std::move(this->receiver));
}
void stop() noexcept override final
{
// Stop can be called when the context is shutting down,
// so treat it as if the receiver completed.
execution::set_value(std::move(this->receiver));
}
friend void tag_invoke(execution::start_t,
wait_process_operation& self) noexcept
{
self.arm();
}
R receiver;
};
/* The sender for the wait/process event. */
struct wait_process_sender
{
explicit wait_process_sender(context& ctx) : ctx(ctx) {}
friend auto tag_invoke(execution::get_completion_signatures_t,
const wait_process_sender&, auto)
-> execution::completion_signatures<execution::set_value_t()>;
template <execution::receiver R>
friend auto tag_invoke(execution::connect_t, wait_process_sender&& self,
R r) -> wait_process_operation<R>
{
// Create the completion for the wait.
return {self.ctx, std::move(r)};
}
private:
context& ctx;
};
task<> wait_process_completion::loop(context& ctx)
{
while (!ctx.final_stop.stop_requested())
{
// Handle the next sdbus event.
co_await wait_process_sender(ctx);
// Completion likely happened on the context 'caller' thread, so
// we need to switch to the worker thread.
co_await execution::schedule(ctx.loop.get_scheduler());
}
{
std::lock_guard lock{ctx.lock};
ctx.wait_process_stopped = true;
}
}
} // namespace details
context::~context() noexcept(false)
{
if (worker_thread.joinable())
{
throw std::logic_error(
"sdbusplus::async::context destructed without completion.");
}
}
void context::run()
{
// Run the primary portion of the run-loop.
caller_run();
// Rethrow the pending exception (if it exists).
rethrow_pending_exception();
// Otherwise this should be final_stop...
// We need to wait for the pending wait process and stop it.
wait_for_wait_process_stopped();
// Wait for all the internal tasks to complete.
stdexec::this_thread::sync_wait(
internal_tasks.empty() | execution::upon_error([&](auto&& e) {
pending_exceptions.emplace_back(std::move(e));
}));
// Finish up the loop and join the thread.
// (There shouldn't be anything going on by this point anyhow.)
loop.finish();
if (worker_thread.joinable())
{
worker_thread.join();
}
// Check for one last exception.
rethrow_pending_exception();
}
void context::worker_run()
{
// Start the sdbus 'wait/process' loop; treat it as an internal task.
internal_tasks.spawn(details::wait_process_completion::loop(*this) |
execution::upon_stopped([]() {}));
// Run the execution::run_loop to handle all the tasks.
loop.run();
}
void context::spawn_complete(std::exception_ptr&& e)
{
{
std::lock_guard l{lock};
spawn_watcher_running = false;
if (e)
{
pending_exceptions.emplace_back(std::move(e));
}
}
if (stop_requested())
{
final_stop.request_stop();
}
caller_wait.notify_one();
event_loop.break_run();
}
void context::check_stop_requested()
{
if (stop_requested())
{
throw std::logic_error(
"sdbusplus::async::context spawn called while already stopped.");
}
}
void context::spawn_watcher()
{
{
std::lock_guard l{lock};
if (spawn_watcher_running)
{
return;
}
spawn_watcher_running = true;
}
// Spawn the watch for completion / exceptions.
internal_tasks.spawn(pending_tasks.empty() |
execution::then([this]() { spawn_complete(); }) |
execution::upon_error([this](auto&& e) {
spawn_complete(std::move(e));
}));
}
void context::caller_run()
{
// We are able to run the loop until the context is requested to stop or
// we get an exception.
auto keep_running = [this]() {
std::lock_guard l{lock};
return !final_stop.stop_requested() && pending_exceptions.empty();
};
// If we are suppose to keep running, start the run loop.
if (keep_running())
{
// Start up the worker thread.
if (!worker_thread.joinable())
{
worker_thread = std::thread{[this]() { worker_run(); }};
}
else
{
// We've already been running and there might an exception or
// completion pending. Spawn a new watcher that checks for these.
spawn_watcher();
}
while (keep_running())
{
// Handle waiting on all the sd-events.
details::wait_process_completion::wait_once(*this);
}
}
else
{
// There might be pending exceptions still, so spawn a watcher for them.
spawn_watcher();
}
}
void context::wait_for_wait_process_stopped()
{
auto worker = std::exchange(pending, nullptr);
while (worker == nullptr)
{
std::lock_guard l{lock};
if (wait_process_stopped)
{
break;
}
worker = std::exchange(staged, nullptr);
if (!worker)
{
std::this_thread::yield();
}
}
if (worker)
{
worker->stop();
wait_process_stopped = true;
}
}
void context::rethrow_pending_exception()
{
{
std::lock_guard l{lock};
if (!pending_exceptions.empty())
{
auto e = pending_exceptions.front();
pending_exceptions.pop_front();
std::rethrow_exception(std::move(e));
}
}
}
void details::wait_process_completion::arm() noexcept
{
// Call process. True indicates something was handled and we do not
// need to `wait`, because there might be yet another pending operation
// to process, so immediately signal the operation as complete.
if (ctx.get_bus().process_discard())
{
this->complete();
return;
}
// We need to call wait now, get the current timeout and stage ourselves
// as the next completion.
// Get the bus' timeout.
uint64_t to_usec = 0;
sd_bus_get_timeout(get_busp(ctx.get_bus()), &to_usec);
if (to_usec == UINT64_MAX)
{
// sd_bus_get_timeout returns UINT64_MAX to indicate 'wait forever'.
// Turn this into -1 for sd-event.
timeout = std::chrono::microseconds{-1};
}
else
{
timeout = std::chrono::microseconds(to_usec);
}
// Assign ourselves as the pending completion and release the caller.
std::lock_guard lock{ctx.lock};
ctx.staged = this;
ctx.caller_wait.notify_one();
}
void details::wait_process_completion::wait_once(context& ctx)
{
// Scope for lock.
{
std::unique_lock lock{ctx.lock};
// If there isn't a completion waiting already, wait on the condition
// variable for one to show up (we can't call `poll` yet because we
// don't have the required parameters).
ctx.caller_wait.wait(lock, [&] {
return (ctx.pending != nullptr) || (ctx.staged != nullptr) ||
ctx.final_stop.stop_requested() ||
!ctx.pending_exceptions.empty();
});
// Save the waiter as pending.
if (ctx.pending == nullptr)
{
ctx.pending = std::exchange(ctx.staged, nullptr);
}
}
// Run the event loop to process one request.
// If the context has been requested to be stopped, skip the event loop.
if (!ctx.final_stop.stop_requested() && ctx.pending)
{
ctx.event_loop.run_one(ctx.pending->timeout);
}
}
int context::dbus_event_handle(sd_event_source*, int, uint32_t, void* data)
{
auto self = static_cast<context*>(data);
auto pending = std::exchange(self->pending, nullptr);
if (pending != nullptr)
{
pending->complete();
}
return 0;
}
} // namespace sdbusplus::async