sd-bus: introduce new sd_bus_default_flush_close() call

If code enqueues a message on one of the default busses, but doesn't
sync on it, and immediately drops the reference to the bus again, it
will stay queued and consume memory. Intrdouce a new call
sd_bus_default_flush_close() that can be invoked at the end of programs
(or threads) and flushes out all unsent messages on any of the default
busses.

4 files changed, 31 insertions, 5 deletions

diff --git a/src/libsystemd/libsystemd.sym b/src/libsystemd/libsystemd.sym
index d5ad127bcb..518cbbb7ed 100644
--- a/src/libsystemd/libsystemd.sym
+++ b/src/libsystemd/libsystemd.sym
@@ -473,3 +473,8 @@ global:
         sd_pid_get_cgroup;
         sd_peer_get_cgroup;
 } LIBSYSTEMD_222;
+
+LIBSYSTEMD_227 {
+global:
+        sd_bus_default_flush_close;
+} LIBSYSTEMD_226;
diff --git a/src/libsystemd/sd-bus/sd-bus.c b/src/libsystemd/sd-bus/sd-bus.c
index 25fd3b5c52..53d1c6f61d 100644
--- a/src/libsystemd/sd-bus/sd-bus.c
+++ b/src/libsystemd/sd-bus/sd-bus.c
@@ -69,6 +69,10 @@ static int bus_poll(sd_bus *bus, bool need_more, uint64_t timeout_usec);
 static int attach_io_events(sd_bus *b);
 static void detach_io_events(sd_bus *b);
 
+static thread_local sd_bus *default_system_bus = NULL;
+static thread_local sd_bus *default_user_bus = NULL;
+static thread_local sd_bus *default_starter_bus = NULL;
+
 static void bus_close_fds(sd_bus *b) {
         assert(b);
 
@@ -3348,14 +3352,11 @@ static int bus_default(int (*bus_open)(sd_bus **), sd_bus **default_bus, sd_bus
 }
 
 _public_ int sd_bus_default_system(sd_bus **ret) {
-        static thread_local sd_bus *default_system_bus = NULL;
-
         return bus_default(sd_bus_open_system, &default_system_bus, ret);
 }
 
-_public_ int sd_bus_default_user(sd_bus **ret) {
-        static thread_local sd_bus *default_user_bus = NULL;
 
+_public_ int sd_bus_default_user(sd_bus **ret) {
         return bus_default(sd_bus_open_user, &default_user_bus, ret);
 }
 
@@ -3382,7 +3383,6 @@ _public_ int sd_bus_default(sd_bus **ret) {
 
         e = secure_getenv("DBUS_STARTER_ADDRESS");
         if (e) {
-                static thread_local sd_bus *default_starter_bus = NULL;
 
                 return bus_default(sd_bus_open, &default_starter_bus, ret);
         }
@@ -3605,3 +3605,20 @@ _public_ int sd_bus_is_monitor(sd_bus *bus) {
 
         return !!(bus->hello_flags & KDBUS_HELLO_MONITOR);
 }
+
+static void flush_close(sd_bus *bus) {
+        if (!bus)
+                return;
+
+        /* Flushes and closes the specified bus. We take a ref before,
+         * to ensure the flushing does not cause the bus to be
+         * unreferenced. */
+
+        sd_bus_flush_close_unref(sd_bus_ref(bus));
+}
+
+_public_ void sd_bus_default_flush_close(void) {
+        flush_close(default_starter_bus);
+        flush_close(default_user_bus);
+        flush_close(default_system_bus);
+}
diff --git a/src/systemctl/systemctl.c b/src/systemctl/systemctl.c
index e20fc1bbbb..10484a2b70 100644
--- a/src/systemctl/systemctl.c
+++ b/src/systemctl/systemctl.c
@@ -7705,5 +7705,7 @@ finish:
         strv_free(arg_states);
         strv_free(arg_properties);
 
+        sd_bus_default_flush_close();
+
         return r < 0 ? EXIT_FAILURE : r;
 }
diff --git a/src/systemd/sd-bus.h b/src/systemd/sd-bus.h
index 5439a1903b..0883203ae7 100644
--- a/src/systemd/sd-bus.h
+++ b/src/systemd/sd-bus.h
@@ -158,6 +158,8 @@ sd_bus *sd_bus_ref(sd_bus *bus);
 sd_bus *sd_bus_unref(sd_bus *bus);
 sd_bus *sd_bus_flush_close_unref(sd_bus *bus);
 
+void sd_bus_default_flush_close(void);
+
 int sd_bus_is_open(sd_bus *bus);
 
 int sd_bus_get_bus_id(sd_bus *bus, sd_id128_t *id);