Add single-threaded EventLoop that multiplexes D-Bus and fd sources

CMakeLists.txt

@@ -14,6 +14,11 @@ set(CMAKE_CXX_STANDARD_REQUIRED ON)
 set(CMAKE_CXX_EXTENSIONS OFF)
 
 option(ENABLE_LTO "Enable Link Time Optimization" ON)
+option(SDBUS_CPP_EXAMPLES_BUILD_TESTS "Build the test targets" OFF)
+
+if (SDBUS_CPP_EXAMPLES_BUILD_TESTS)
+    enable_testing()
+endif ()
 
 #
 # Toolchain target

src/utils/CMakeLists.txt

@@ -1,5 +1,7 @@
 
 add_library(utils STATIC
+        event_loop.cc
+        event_loop.h
         utils.cc
         utils.h
 )
@@ -9,3 +11,9 @@ target_link_libraries(utils
         glaze::glaze
         spdlog::spdlog
 )
+
+if (SDBUS_CPP_EXAMPLES_BUILD_TESTS)
+    add_executable(event_loop_test event_loop_test.cc)
+    target_link_libraries(event_loop_test PRIVATE utils sdbus-c++ spdlog::spdlog)
+    add_test(NAME event_loop_test COMMAND event_loop_test)
+endif ()

src/utils/event_loop.cc

@@ -0,0 +1,157 @@
+// Copyright (c) 2026 Joel Winarske
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "event_loop.h"
+
+#include <algorithm>
+#include <cerrno>
+#include <cstdint>
+#include <cstring>
+
+#include <sys/eventfd.h>
+#include <unistd.h>
+
+#include <sdbus-c++/sdbus-c++.h>
+
+#include "logging.h"
+
+namespace {
+constexpr std::size_t kNotPresent = static_cast<std::size_t>(-1);
+}  // namespace
+
+EventLoop::EventLoop() : wake_fd_(::eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK)) {
+  if (!wake_fd_.valid()) {
+    LOG_ERROR("EventLoop: failed to create eventfd: {}", strerror(errno));
+  }
+}
+
+void EventLoop::wake() noexcept {
+  if (!wake_fd_.valid()) {
+    return;
+  }
+  constexpr std::uint64_t one = 1;
+  if (::write(wake_fd_.get(), &one, sizeof(one)) < 0 && errno != EAGAIN) {
+    LOG_ERROR("EventLoop: failed to signal wake eventfd: {}", strerror(errno));
+  }
+}
+
+void EventLoop::add(EventSource* source) {
+  to_add_.push_back(source);
+  wake();
+}
+
+void EventLoop::remove(EventSource* source) {
+  to_remove_.push_back(source);
+  wake();
+}
+
+void EventLoop::stop(const int exit_code) noexcept {
+  exit_code_.store(exit_code, std::memory_order_relaxed);
+  running_.store(false, std::memory_order_relaxed);
+  wake();  // write() is async-signal-safe
+}
+
+void EventLoop::apply_pending() {
+  for (auto* source : to_remove_) {
+    std::erase(sources_, source);
+  }
+  to_remove_.clear();
+
+  for (auto* source : to_add_) {
+    if (std::ranges::find(sources_, source) == sources_.end()) {
+      sources_.push_back(source);
+    }
+  }
+  to_add_.clear();
+}
+
+int EventLoop::run(sdbus::IConnection& bus) {
+  running_.store(true, std::memory_order_relaxed);
+
+  while (running_.load(std::memory_order_relaxed)) {
+    apply_pending();
+    if (!running_.load(std::memory_order_relaxed)) {
+      break;
+    }
+
+    const auto pd = bus.getEventLoopPollData();
+
+    std::vector<pollfd> pfds;
+    pfds.reserve(sources_.size() + 3);
+
+    const std::size_t dbus_idx = pfds.size();
+    pfds.push_back({pd.fd, pd.events, 0});
+
+    std::size_t dbus_event_idx = kNotPresent;
+    if (pd.eventFd >= 0) {
+      dbus_event_idx = pfds.size();
+      pfds.push_back({pd.eventFd, POLLIN, 0});
+    }
+
+    const std::size_t wake_idx = pfds.size();
+    pfds.push_back({wake_fd_.get(), POLLIN, 0});
+
+    const std::size_t first_source_idx = pfds.size();
+    for (const auto* source : sources_) {
+      pfds.push_back({source->fd(), source->events(), 0});
+    }
+
+    const int n = ::poll(pfds.data(), pfds.size(), pd.getPollTimeout());
+    if (n < 0) {
+      if (errno == EINTR) {
+        continue;
+      }
+      LOG_ERROR("EventLoop: poll failed: {}", strerror(errno));
+      exit_code_.store(1, std::memory_order_relaxed);
+      break;
+    }
+
+    // Drain the wake eventfd so it doesn't stay readable.
+    if ((pfds[wake_idx].revents & POLLIN) != 0) {
+      std::uint64_t drained = 0;
+      while (::read(wake_fd_.get(), &drained, sizeof(drained)) ==
+             sizeof(drained)) {
+      }
+    }
+
+    // Drive D-Bus whenever its fd/eventFd is ready, or on timeout so sd-bus can
+    // recalculate and fire its own timers.
+    const bool dbus_ready =
+        n == 0 || pfds[dbus_idx].revents != 0 ||
+        (dbus_event_idx != kNotPresent && pfds[dbus_event_idx].revents != 0);
+    if (dbus_ready) {
+      try {
+        while (bus.processPendingEvent()) {
+        }
+      } catch (const sdbus::Error& e) {
+        LOG_ERROR("EventLoop: D-Bus processing error: {} - {}", e.getName(),
+                  e.getMessage());
+        exit_code_.store(1, std::memory_order_relaxed);
+        break;
+      }
+    }
+
+    // Dispatch ready sources. sources_ is only mutated in apply_pending() (top
+    // of the loop), so indexing the snapshot built above stays valid even if a
+    // dispatch() calls add()/remove().
+    for (std::size_t i = 0; i < sources_.size(); ++i) {
+      if (const short revents = pfds[first_source_idx + i].revents;
+          revents != 0) {
+        sources_[i]->dispatch(revents);
+      }
+    }
+  }
+
+  return exit_code_.load(std::memory_order_relaxed);
+}

src/utils/event_loop.h

@@ -0,0 +1,99 @@
+// Copyright (c) 2026 Joel Winarske
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef SRC_UTILS_EVENT_LOOP_H
+#define SRC_UTILS_EVENT_LOOP_H
+
+#include <atomic>
+#include <vector>
+
+#include <poll.h>
+
+#include "unique_fd.h"
+
+namespace sdbus {
+class IConnection;
+}
+
+/// A file descriptor that can be multiplexed by EventLoop.
+///
+/// fd() is polled for events() (POLLIN by default); when ready, dispatch() is
+/// invoked on the loop thread with the reported revents. Because every source
+/// is dispatched from the single loop thread, state touched only from
+/// dispatch() needs no locking.
+class EventSource {
+ public:
+  virtual ~EventSource() = default;
+
+  [[nodiscard]] virtual int fd() const = 0;
+  [[nodiscard]] virtual short events() const { return POLLIN; }
+
+  /// Called on the loop thread when fd() is ready. Must not block.
+  virtual void dispatch(short revents) = 0;
+};
+
+/// Single-threaded event loop that multiplexes an sdbus::IConnection together
+/// with a set of EventSource objects, so D-Bus callbacks and device/fd I/O all
+/// run on one thread. This is the alternative to enterEventLoopAsync(): drive
+/// the bus from our own poll() instead of a bus-owned thread, which removes the
+/// need to synchronise state shared between the D-Bus and device callbacks.
+///
+/// Threading: run() and all EventSource::dispatch() callbacks execute on the
+/// thread that calls run(). add()/remove() are intended to be called from that
+/// same thread (typically from within a dispatch()); they take effect on the
+/// next loop iteration. stop() is the sole exception — it is safe to call from
+/// a signal handler or another thread.
+class EventLoop {
+ public:
+  EventLoop();
+
+  EventLoop(const EventLoop&) = delete;
+  EventLoop& operator=(const EventLoop&) = delete;
+  EventLoop(EventLoop&&) = delete;
+  EventLoop& operator=(EventLoop&&) = delete;
+  ~EventLoop() = default;
+
+  /// Register a source. Takes effect on the next loop iteration, so it is safe
+  /// to call from within a dispatch() callback.
+  void add(EventSource* source);
+
+  /// Unregister a source. Takes effect on the next loop iteration. The source
+  /// object must remain alive until then, so do not destroy it inside its own
+  /// dispatch() — reset it after the next iteration (or after run() returns).
+  void remove(EventSource* source);
+
+  /// Drive `bus` and every registered source until stop() is called or an
+  /// unrecoverable error occurs. Returns the code passed to stop() (0 on a
+  /// clean external stop, non-zero on internal error).
+  int run(sdbus::IConnection& bus);
+
+  /// Request the loop to exit with `exit_code`. Async-signal-safe and
+  /// thread-safe: writes an eventfd to break out of poll().
+  void stop(int exit_code = 0) noexcept;
+
+ private:
+  void wake() noexcept;
+  void apply_pending();
+
+  UniqueFd
+      wake_fd_;  // eventfd used to break out of poll() for stop()/add/remove
+  std::atomic<bool> running_{false};
+  std::atomic<int> exit_code_{0};
+
+  std::vector<EventSource*> sources_;
+  std::vector<EventSource*> to_add_;
+  std::vector<EventSource*> to_remove_;
+};
+
+#endif  // SRC_UTILS_EVENT_LOOP_H