Use timerfd to process bus timeouts on time

igo95862 · igo95862 · commit fe49edb17582 · 2024-09-22T21:43:01.000+01:00
sdbus has its own internal timeout system where normally the
`sd_bus_get_timeout` function should be used for the maximum time
passed to `poll()`. However, Python's asyncio loop does not work
with that very well as it has its own internal implementation of
polling.

Instead create a new timer file descriptor and use it to make
asyncio call `SdBus.process()` on given time. This provides a much
better performance compared to the `call_later()` implementation
as timerfd timeout can be adjusted without allocating any new Python
objects.
diff --git a/src/sdbus/sd_bus_internals.h b/src/sdbus/sd_bus_internals.h
@@ -334,7 +334,9 @@ typedef struct {
         sd_bus* sd_bus_ref;
         PyObject* bus_fd;
         PyObject* loop;
+        PyObject* timer_fd;
         int asyncio_watchers_last_state;
+        int timer_fd_int;
 } SdBusObject;
 
 extern PyType_Spec SdBusType;
diff --git a/src/sdbus/sd_bus_internals_bus.c b/src/sdbus/sd_bus_internals_bus.c
@@ -20,13 +20,20 @@
 */
 #include <errno.h>
 #include <poll.h>
+#include <sys/timerfd.h>
+#include <time.h>
 #include "sd_bus_internals.h"
 
 static void SdBus_dealloc(SdBusObject* self) {
         if (NULL != self->loop && NULL != self->bus_fd) {
                 Py_XDECREF(PyObject_CallMethodObjArgs(self->loop, remove_reader_str, self->bus_fd, NULL));
                 Py_XDECREF(PyObject_CallMethodObjArgs(self->loop, remove_writer_str, self->bus_fd, NULL));
         }
+        if (NULL != self->timer_fd) {
+                Py_XDECREF(PyObject_CallMethodObjArgs(self->loop, remove_reader_str, self->bus_fd, NULL));
+                Py_DECREF(self->timer_fd);
+                close(self->timer_fd_int);
+        }
         sd_bus_unref(self->sd_bus_ref);
         Py_XDECREF(self->bus_fd);
         Py_XDECREF(self->loop);
@@ -623,6 +630,10 @@ static PyObject* SdBus_close(SdBusObject* self, PyObject* Py_UNUSED(args)) {
                 Py_XDECREF(CALL_PYTHON_AND_CHECK(PyObject_CallMethodObjArgs(self->loop, remove_reader_str, self->bus_fd, NULL)));
                 Py_XDECREF(CALL_PYTHON_AND_CHECK(PyObject_CallMethodObjArgs(self->loop, remove_writer_str, self->bus_fd, NULL)));
         }
+        if (NULL != self->timer_fd) {
+                Py_XDECREF(PyObject_CallMethodObjArgs(self->loop, remove_reader_str, self->bus_fd, NULL));
+                // TODO: Close timerfd
+        }
         Py_RETURN_NONE;
 }
 
@@ -639,7 +650,45 @@ static inline int sd_bus_get_events_zero_on_closed(SdBusObject* self) {
         return events;
 };
 
+static inline int sd_bus_get_timeout_uint_max_on_closed(SdBusObject* self, uint64_t* timeout_usec) {
+        int r = sd_bus_get_timeout(self->sd_bus_ref, timeout_usec);
+        if (-ENOTCONN == r) {
+                *timeout_usec = UINT64_MAX;
+                return 0;
+        }
+        return r;
+}
+
 static PyObject* SdBus_asyncio_update_fd_watchers(SdBusObject* self) {
+        PyObject* running_loop = CALL_PYTHON_AND_CHECK(_get_or_bind_loop(self));
+        PyObject* drive_method CLEANUP_PY_OBJECT = CALL_PYTHON_AND_CHECK(PyObject_GetAttrString((PyObject*)self, "process"));
+
+        if (NULL == self->timer_fd) {
+                self->timer_fd_int = CALL_SD_BUS_AND_CHECK(timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC));
+                if (self->timer_fd_int < 0) {
+                        PyErr_SetFromErrno(PyExc_OSError);
+                }
+                PyObject* timer_fd CLEANUP_PY_OBJECT = PyLong_FromLong((int)self->timer_fd_int);
+                Py_XDECREF(CALL_PYTHON_AND_CHECK(PyObject_CallMethodObjArgs(running_loop, add_reader_str, timer_fd, drive_method, NULL)));
+                Py_INCREF(timer_fd);
+                self->timer_fd = timer_fd;
+        }
+
+        uint64_t timeout_usec = UINT64_MAX;
+        CALL_SD_BUS_AND_CHECK(sd_bus_get_timeout_uint_max_on_closed(self, &timeout_usec));
+
+        struct itimerspec bus_timer = {0};
+        if (timeout_usec == UINT64_MAX) {
+                // Setting bus_timer to zero disarms timer.
+        } else if (timeout_usec != 0) {
+                bus_timer.it_value.tv_sec = timeout_usec / 1000000;
+                bus_timer.it_value.tv_nsec = (timeout_usec % 1000000) * 1000;
+        } else if (timeout_usec == 0) {
+                Py_XDECREF(CALL_PYTHON_AND_CHECK(PyObject_CallMethodObjArgs(running_loop, call_soon_str, drive_method, NULL)));
+        }
+
+        CALL_SD_BUS_AND_CHECK(timerfd_settime(self->timer_fd_int, TFD_TIMER_ABSTIME, &bus_timer, NULL));
+
         int events_to_watch = CALL_SD_BUS_AND_CHECK(sd_bus_get_events_zero_on_closed(self));
         if (events_to_watch == self->asyncio_watchers_last_state) {
                 // Do not update the watchers because state is the same
@@ -648,9 +697,6 @@ static PyObject* SdBus_asyncio_update_fd_watchers(SdBusObject* self) {
                 self->asyncio_watchers_last_state = events_to_watch;
         }
 
-        PyObject* running_loop = CALL_PYTHON_AND_CHECK(_get_or_bind_loop(self));
-        PyObject* drive_method CLEANUP_PY_OBJECT = CALL_PYTHON_AND_CHECK(PyObject_GetAttrString((PyObject*)self, "process"));
-
         if (NULL == self->bus_fd) {
                 self->bus_fd = CALL_PYTHON_AND_CHECK(SdBus_get_fd(self, NULL));
         }
diff --git a/test/test_sdbus_async.py b/test/test_sdbus_async.py
@@ -29,6 +29,7 @@
 from sdbus.exceptions import (
     DbusFailedError,
     DbusFileExistsError,
+    DbusNoReplyError,
     DbusPropertyReadOnlyError,
     DbusUnknownObjectError,
     SdBusLibraryError,
@@ -721,6 +722,20 @@ async def too_long_wait() -> None:
         with self.assertRaises(SdBusLibraryError):
             await wait_for(too_long_wait(), timeout=1)
 
+    async def test_bus_timerfd(self) -> None:
+        test_object, test_object_connection = initialize_object()
+
+        self.bus.method_call_timeout_usec = 10_000  # 0.01 seconds
+
+        loop = get_running_loop()
+
+        start = loop.time()
+
+        with self.assertRaises(DbusNoReplyError):
+            await wait_for(test_object_connection.looong_method(), timeout=1)
+
+        self.assertAlmostEqual(loop.time() - start, 0.01, delta=0.01)
+
     async def test_signal_queue_wildcard_match(self) -> None:
         test_object, test_object_connection = initialize_object()
 
