From 2770da027a761aecb17610374f4f60cc1c6e76cd Mon Sep 17 00:00:00 2001 From: Lennart Poettering Date: Tue, 13 Feb 2018 18:27:05 +0100 Subject: [PATCH 1/3] sd-bus: add APIs to query the current read and write queue size --- man/rules/meson.build | 1 + man/sd_bus_get_n_queued_read.xml | 129 +++++++++++++++++++++++++++++++ src/libsystemd/libsystemd.sym | 6 ++ src/libsystemd/sd-bus/sd-bus.c | 20 +++++ 4 files changed, 156 insertions(+) create mode 100644 man/sd_bus_get_n_queued_read.xml diff --git a/man/rules/meson.build b/man/rules/meson.build index bfc267b544..84b911b365 100644 --- a/man/rules/meson.build +++ b/man/rules/meson.build @@ -186,6 +186,7 @@ manpages = [ ['SD_BUS_ERROR_END', 'SD_BUS_ERROR_MAP', 'sd_bus_error_map'], ''], ['sd_bus_get_fd', '3', [], ''], + ['sd_bus_get_n_queued_read', '3', ['sd_bus_get_n_queued_write'], ''], ['sd_bus_is_open', '3', ['sd_bus_is_ready'], ''], ['sd_bus_message_append', '3', ['sd_bus_message_appendv'], ''], ['sd_bus_message_append_array', diff --git a/man/sd_bus_get_n_queued_read.xml b/man/sd_bus_get_n_queued_read.xml new file mode 100644 index 0000000000..b39e87f8dd --- /dev/null +++ b/man/sd_bus_get_n_queued_read.xml @@ -0,0 +1,129 @@ + + + + + + + + + sd_bus_get_fd + systemd + + + + Developer + Lennart + Poettering + lennart@poettering.net + + + + + + sd_bus_get_n_queued_read + 3 + + + + sd_bus_get_n_queued_read + sd_bus_get_n_queued_write + + Get the number of pending bus messages in the read and write queues of a bus connection object + + + + + #include <systemd/sd-bus.h> + + + int sd_bus_get_n_queued_read + sd_bus *bus + uint64_t *ret + + + + int sd_bus_get_n_queued_write + sd_bus *bus + uint64_t *ret + + + + + + Description + + + sd_bus_get_n_queued_read() may be used to query the number of bus messages in the read queue + of a bus connection object. The read queue contains all messages read from the transport medium (e.g. network + socket) but not yet processed locally. The function expects two arguments: the bus object to query the number of + queued messages of, and a pointer to a 64bit counter variable to write the current queue size to. Use + sd_bus_process() in order to process queued messages, i.e. to reduce the size of the read + queue (as well as, in fact, the write queue, see below) when it is non-zero. + + + + Similar, sd_bus_get_n_queued_write() may be used to query the number of currently pending + bus messages in the write queue of a bus connection object. The write queue contains all messages enqueued into + the connection with a call such as sd_bus_send() but not yet written to the transport + medium. The expected arguments are similar to the ones of sd_bus_get_n_queued_read(). Here + too, use sd_bus_process() to reduce the size of the write queue. Alternatively, use + sd_bus_flush() to synchronously write out any pending bus messages until the write queue is + empty. + + + + + Return Value + + On success, these functions return 0 or a positive integer. On failure, they return a negative errno-style + error code. + + + + Errors + + Returned errors may indicate the following problems: + + + + -ECHILD + + The bus connection has been created in a different process. + + + + + + + See Also + + + systemd1, + sd-bus3, + sd_bus_process3, + sd_bus_send3, + sd_bus_flush3 + + + + diff --git a/src/libsystemd/libsystemd.sym b/src/libsystemd/libsystemd.sym index 00aeefbe19..7ff5816180 100644 --- a/src/libsystemd/libsystemd.sym +++ b/src/libsystemd/libsystemd.sym @@ -549,3 +549,9 @@ global: sd_event_source_get_io_fd_own; sd_event_source_set_io_fd_own; } LIBSYSTEMD_236; + +LIBSYSTEMD_238 { +global: + sd_bus_get_n_queued_read; + sd_bus_get_n_queued_write; +} LIBSYSTEMD_237; diff --git a/src/libsystemd/sd-bus/sd-bus.c b/src/libsystemd/sd-bus/sd-bus.c index cbbc6b6f0c..2f5e483ae2 100644 --- a/src/libsystemd/sd-bus/sd-bus.c +++ b/src/libsystemd/sd-bus/sd-bus.c @@ -4091,3 +4091,23 @@ _public_ int sd_bus_get_sender(sd_bus *bus, const char **ret) { *ret = bus->patch_sender; return 0; } + +_public_ int sd_bus_get_n_queued_read(sd_bus *bus, uint64_t *ret) { + assert_return(bus, -EINVAL); + assert_return(bus = bus_resolve(bus), -ENOPKG); + assert_return(!bus_pid_changed(bus), -ECHILD); + assert_return(ret, -EINVAL); + + *ret = bus->rqueue_size; + return 0; +} + +_public_ int sd_bus_get_n_queued_write(sd_bus *bus, uint64_t *ret) { + assert_return(bus, -EINVAL); + assert_return(bus = bus_resolve(bus), -ENOPKG); + assert_return(!bus_pid_changed(bus), -ECHILD); + assert_return(ret, -EINVAL); + + *ret = bus->wqueue_size; + return 0; +} From 9fc677e3c997548e701f418e0b413441e1c5e891 Mon Sep 17 00:00:00 2001 From: Lennart Poettering Date: Tue, 13 Feb 2018 18:27:47 +0100 Subject: [PATCH 2/3] core: don't bother enqueuing signal messages into busses that aren't ready yet This is an optimization: there's no point in enqueuing unit and job change notificiation signal messages into bus connection that aren't fully set up yet. This doesn't fix #8166 but should lower the load of messages enqueued but not processed yet a bit. --- src/core/dbus.c | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/src/core/dbus.c b/src/core/dbus.c index 305824da9b..d023c8715c 100644 --- a/src/core/dbus.c +++ b/src/core/dbus.c @@ -1197,6 +1197,11 @@ int bus_foreach_bus( /* Send to all direct buses, unconditionally */ SET_FOREACH(b, m->private_buses, i) { + + /* Don't bother with enqueing these messages to clients that haven't started yet */ + if (sd_bus_is_ready(b) <= 0) + continue; + r = send_message(b, userdata); if (r < 0) ret = r; From e0a085811de1d1976c019afcfc2e4e74f590cc9f Mon Sep 17 00:00:00 2001 From: Lennart Poettering Date: Tue, 13 Feb 2018 18:30:34 +0100 Subject: [PATCH 3/3] core: don't process dbus unit and job queue when there are already too many messages pending We maintain a queue of units and jobs that we are supposed to generate change/new notifications for because they were either just created or some of their property has changed. Let's throttle processing of this queue a bit: as soon as > 1K of bus messages are queued for writing let's skip processing the queue, and then recheck on the next iteration again. Moreover, never process more than 100 units in one go, return to the event loop after that. Both limits together should put effective limits on both space and time usage of the function, delaying further operations until a later moment, when the queue is empty or the the event loop is sufficiently idle again. This should keep the number of generated messages much lower than before on busy systems or where some client is hanging. Note that this also means a bad client can slow down message dispatching substantially for up to 90s if it likes to, for all clients. But that should be acceptable as we only allow trusted bus clients, anyway. Fixes: #8166 --- src/core/dbus.c | 31 ++++++++++++++++++++++++++++ src/core/dbus.h | 2 ++ src/core/manager.c | 49 ++++++++++++++++++++++++++++++++++++-------- src/systemd/sd-bus.h | 3 +++ 4 files changed, 76 insertions(+), 9 deletions(-) diff --git a/src/core/dbus.c b/src/core/dbus.c index d023c8715c..56b43adcda 100644 --- a/src/core/dbus.c +++ b/src/core/dbus.c @@ -1279,3 +1279,34 @@ int bus_verify_reload_daemon_async(Manager *m, sd_bus_message *call, sd_bus_erro int bus_verify_set_environment_async(Manager *m, sd_bus_message *call, sd_bus_error *error) { return bus_verify_polkit_async(call, CAP_SYS_ADMIN, "org.freedesktop.systemd1.set-environment", NULL, false, UID_INVALID, &m->polkit_registry, error); } + +uint64_t manager_bus_n_queued_write(Manager *m) { + uint64_t c = 0; + Iterator i; + sd_bus *b; + int r; + + /* Returns the total number of messages queued for writing on all our direct and API busses. */ + + SET_FOREACH(b, m->private_buses, i) { + uint64_t k; + + r = sd_bus_get_n_queued_write(b, &k); + if (r < 0) + log_debug_errno(r, "Failed to query queued messages for private bus: %m"); + else + c += k; + } + + if (m->api_bus) { + uint64_t k; + + r = sd_bus_get_n_queued_write(m->api_bus, &k); + if (r < 0) + log_debug_errno(r, "Failed to query queued messages for API bus: %m"); + else + c += k; + } + + return c; +} diff --git a/src/core/dbus.h b/src/core/dbus.h index 3051ec0665..702bc48ae1 100644 --- a/src/core/dbus.h +++ b/src/core/dbus.h @@ -48,3 +48,5 @@ int bus_verify_reload_daemon_async(Manager *m, sd_bus_message *call, sd_bus_erro int bus_verify_set_environment_async(Manager *m, sd_bus_message *call, sd_bus_error *error); int bus_forward_agent_released(Manager *m, const char *path); + +uint64_t manager_bus_n_queued_write(Manager *m); diff --git a/src/core/manager.c b/src/core/manager.c index ebac47024c..84adb9c666 100644 --- a/src/core/manager.c +++ b/src/core/manager.c @@ -102,6 +102,13 @@ #define JOBS_IN_PROGRESS_PERIOD_USEC (USEC_PER_SEC / 3) #define JOBS_IN_PROGRESS_PERIOD_DIVISOR 3 +/* If there are more than 1K bus messages queue across our API and direct busses, then let's not add more on top until + * the queue gets more empty. */ +#define MANAGER_BUS_BUSY_THRESHOLD 1024LU + +/* How many units and jobs to process of the bus queue before returning to the event loop. */ +#define MANAGER_BUS_MESSAGE_BUDGET 100U + static int manager_dispatch_notify_fd(sd_event_source *source, int fd, uint32_t revents, void *userdata); static int manager_dispatch_cgroups_agent_fd(sd_event_source *source, int fd, uint32_t revents, void *userdata); static int manager_dispatch_signal_fd(sd_event_source *source, int fd, uint32_t revents, void *userdata); @@ -1886,41 +1893,65 @@ static int manager_dispatch_run_queue(sd_event_source *source, void *userdata) { } static unsigned manager_dispatch_dbus_queue(Manager *m) { - Job *j; + unsigned n = 0, budget; Unit *u; - unsigned n = 0; + Job *j; assert(m); if (m->dispatching_dbus_queue) return 0; + /* Anything to do at all? */ + if (!m->dbus_unit_queue && !m->dbus_job_queue && !m->send_reloading_done && !m->queued_message) + return 0; + + /* Do we have overly many messages queued at the moment? If so, let's not enqueue more on top, let's sit this + * cycle out, and process things in a later cycle when the queues got a bit emptier. */ + if (manager_bus_n_queued_write(m) > MANAGER_BUS_BUSY_THRESHOLD) + return 0; + + /* Only process a certain number of units/jobs per event loop iteration. Even if the bus queue wasn't overly + * full before this call we shouldn't increase it in size too wildly in one step, and we shouldn't monopolize + * CPU time with generating these messages. Note the difference in counting of this "budget" and the + * "threshold" above: the "budget" is decreased only once per generated message, regardless how many + * busses/direct connections it is enqueued on, while the "threshold" is applied to each queued instance of bus + * message, i.e. if the same message is enqueued to five busses/direct connections it will be counted five + * times. This difference in counting ("references" vs. "instances") is primarily a result of the fact that + * it's easier to implement it this way, however it also reflects the thinking that the "threshold" should put + * a limit on used queue memory, i.e. space, while the "budget" should put a limit on time. Also note that + * the "threshold" is currently chosen much higher than the "budget". */ + budget = MANAGER_BUS_MESSAGE_BUDGET; + m->dispatching_dbus_queue = true; - while ((u = m->dbus_unit_queue)) { + while (budget > 0 && (u = m->dbus_unit_queue)) { + assert(u->in_dbus_queue); bus_unit_send_change_signal(u); - n++; + n++, budget--; } - while ((j = m->dbus_job_queue)) { + while (budget > 0 && (j = m->dbus_job_queue)) { assert(j->in_dbus_queue); bus_job_send_change_signal(j); - n++; + n++, budget--; } m->dispatching_dbus_queue = false; - if (m->send_reloading_done) { + if (budget > 0 && m->send_reloading_done) { m->send_reloading_done = false; - bus_manager_send_reloading(m, false); + n++, budget--; } - if (m->queued_message) + if (budget > 0 && m->queued_message) { bus_send_queued_message(m); + n++; + } return n; } diff --git a/src/systemd/sd-bus.h b/src/systemd/sd-bus.h index 82e7d445d2..fff6798d0f 100644 --- a/src/systemd/sd-bus.h +++ b/src/systemd/sd-bus.h @@ -202,6 +202,9 @@ int sd_bus_attach_event(sd_bus *bus, sd_event *e, int priority); int sd_bus_detach_event(sd_bus *bus); sd_event *sd_bus_get_event(sd_bus *bus); +int sd_bus_get_n_queued_read(sd_bus *bus, uint64_t *ret); +int sd_bus_get_n_queued_write(sd_bus *bus, uint64_t *ret); + int sd_bus_add_filter(sd_bus *bus, sd_bus_slot **slot, sd_bus_message_handler_t callback, void *userdata); int sd_bus_add_match(sd_bus *bus, sd_bus_slot **slot, const char *match, sd_bus_message_handler_t callback, void *userdata); int sd_bus_add_match_async(sd_bus *bus, sd_bus_slot **slot, const char *match, sd_bus_message_handler_t callback, sd_bus_message_handler_t install_callback, void *userdata);