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/core/dbus.c b/src/core/dbus.c index 305824da9b..56b43adcda 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; @@ -1274,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/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; +} 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);