Merge 550f38cb14 into 69af4849aa

Send out dbus signals only on buses where Subscribe() method was called
in order to save resources.

src/core/dbus-job.c

@@ -247,7 +247,7 @@ void bus_job_send_change_signal(Job *j) {
                 job_add_to_gc_queue(j);
         }
 
-        r = bus_foreach_bus(j->manager, j->bus_track, j->sent_dbus_new_signal ? send_changed_signal : send_new_signal, j);
+        r = bus_foreach_bus_signal(j->manager, j->bus_track, j->sent_dbus_new_signal ? send_changed_signal : send_new_signal, j);
         if (r < 0)
                 log_debug_errno(r, "Failed to send job change signal for %u: %m", j->id);
 
@@ -308,7 +308,7 @@ void bus_job_send_removed_signal(Job *j) {
         /* Make sure that any change signal on the unit is reflected before we send out the change signal on the job */
         bus_unit_send_pending_change_signal(j->unit, true);
 
-        r = bus_foreach_bus(j->manager, j->bus_track, send_removed_signal, j);
+        r = bus_foreach_bus_signal(j->manager, j->bus_track, send_removed_signal, j);
         if (r < 0)
                 log_debug_errno(r, "Failed to send job remove signal for %u: %m", j->id);
 }

src/core/dbus-manager.c

@@ -1337,10 +1337,6 @@ static int method_subscribe(sd_bus_message *message, void *userdata, sd_bus_erro
                 return r;
 
         if (sd_bus_message_get_bus(message) == m->api_bus) {
-
-                /* Note that direct bus connection subscribe by
-                 * default, we only track peers on the API bus here */
-
                 if (!m->subscribed) {
                         r = sd_bus_track_new(sd_bus_message_get_bus(message), &m->subscribed, NULL, NULL);
                         if (r < 0)
@@ -1350,10 +1346,15 @@ static int method_subscribe(sd_bus_message *message, void *userdata, sd_bus_erro
                 r = sd_bus_track_add_sender(m->subscribed, message);
                 if (r < 0)
                         return r;
-                if (r == 0)
+        } else {
-                        return sd_bus_error_set(error, BUS_ERROR_ALREADY_SUBSCRIBED, "Client is already subscribed.");
+                r = set_ensure_put(&m->private_buses_subscribed, NULL, sd_bus_message_get_bus(message));
+                if (r < 0)
+                        return r;
         }
 
+        if (r == 0)
+                return sd_bus_error_set(error, BUS_ERROR_ALREADY_SUBSCRIBED, "Client is already subscribed.");
+
         return sd_bus_reply_method_return(message, NULL);
 }
 
@@ -1373,9 +1374,11 @@ static int method_unsubscribe(sd_bus_message *message, void *userdata, sd_bus_er
                 r = sd_bus_track_remove_sender(m->subscribed, message);
                 if (r < 0)
                         return r;
-                if (r == 0)
+        } else
-                        return sd_bus_error_set(error, BUS_ERROR_NOT_SUBSCRIBED, "Client is not subscribed.");
+                r = !!set_remove(m->private_buses_subscribed, sd_bus_message_get_bus(message));
-        }
+
+        if (r == 0)
+                return sd_bus_error_set(error, BUS_ERROR_NOT_SUBSCRIBED, "Client is not subscribed.");
 
         return sd_bus_reply_method_return(message, NULL);
 }
@@ -2339,7 +2342,7 @@ static void manager_unit_files_changed(Manager *m, const InstallChange *changes,
         /* See comments for this variable in manager.h */
         m->unit_file_state_outdated = true;
 
-        r = bus_foreach_bus(m, NULL, send_unit_files_changed, NULL);
+        r = bus_foreach_bus_signal(m, NULL, send_unit_files_changed, NULL);
         if (r < 0)
                 log_debug_errno(r, "Failed to send UnitFilesChanged signal, ignoring: %m");
 }
@@ -3706,7 +3709,7 @@ void bus_manager_send_finished(
 
         assert(m);
 
-        r = bus_foreach_bus(
+        r = bus_foreach_bus_signal(
                         m,
                         NULL,
                         send_finished,
@@ -3744,7 +3747,7 @@ void bus_manager_send_reloading(Manager *m, bool active) {
 
         assert(m);
 
-        r = bus_foreach_bus(m, NULL, send_reloading, INT_TO_PTR(active));
+        r = bus_foreach_bus_signal(m, NULL, send_reloading, INT_TO_PTR(active));
         if (r < 0)
                 log_debug_errno(r, "Failed to send reloading signal: %m");
 }
@@ -3763,7 +3766,7 @@ void bus_manager_send_change_signal(Manager *m) {
 
         assert(m);
 
-        r = bus_foreach_bus(m, NULL, send_changed_signal, NULL);
+        r = bus_foreach_bus_signal(m, NULL, send_changed_signal, NULL);
         if (r < 0)
                 log_debug_errno(r, "Failed to send manager change signal: %m");
 }

src/core/dbus-unit.c

@@ -1705,7 +1705,7 @@ void bus_unit_send_change_signal(Unit *u) {
         if (!u->id)
                 return;
 
-        r = bus_foreach_bus(u->manager, u->bus_track, u->sent_dbus_new_signal ? send_changed_signal : send_new_signal, u);
+        r = bus_foreach_bus_signal(u->manager, u->bus_track, u->sent_dbus_new_signal ? send_changed_signal : send_new_signal, u);
         if (r < 0)
                 log_unit_debug_errno(u, r, "Failed to send unit change signal for %s: %m", u->id);
 
@@ -1800,7 +1800,7 @@ void bus_unit_send_removed_signal(Unit *u) {
         if (!u->id)
                 return;
 
-        r = bus_foreach_bus(u->manager, u->bus_track, send_removed_signal, u);
+        r = bus_foreach_bus_signal(u->manager, u->bus_track, send_removed_signal, u);
         if (r < 0)
                 log_unit_debug_errno(u, r, "Failed to send unit remove signal for %s: %m", u->id);
 }

src/core/dbus.c

@@ -138,6 +138,9 @@ static int signal_disconnected(sd_bus_message *message, void *userdata, sd_bus_e
 
         if (set_remove(m->private_buses, bus)) {
                 log_debug("Got disconnect on private connection.");
+
+                /* Don't bother checking if the bus was subscribed; try to remove it opportunistically. */
+                set_remove(m->private_buses_subscribed, bus);
                 destroy_bus(m, &bus);
         }
 
@@ -761,6 +764,9 @@ static int bus_on_connection(sd_event_source *s, int fd, uint32_t revents, void
                 return 0;
         }
 
+        /* If this bus (i.e. object address) was subscribed previously let's drop it from that set. */
+        set_remove(m->private_buses_subscribed, bus);
+
         TAKE_PTR(bus);
 
         log_debug("Accepted new private connection.");
@@ -1044,6 +1050,8 @@ void bus_done_private(Manager *m) {
                 destroy_bus(m, &b);
 
         m->private_buses = set_free(m->private_buses);
+        m->private_buses_subscribed = set_free(m->private_buses_subscribed);
+
 
         m->private_listen_event_source = sd_event_source_disable_unref(m->private_listen_event_source);
         m->private_listen_fd = safe_close(m->private_listen_fd);
@@ -1101,33 +1109,33 @@ int bus_fdset_add_all(Manager *m, FDSet *fds) {
         return 0;
 }
 
-int bus_foreach_bus(
+int bus_foreach_bus_signal(
                 Manager *m,
                 sd_bus_track *subscribed2,
-                int (*send_message)(sd_bus *bus, void *userdata),
+                int (*send_signal)(sd_bus *bus, void *userdata),
                 void *userdata) {
 
         int r = 0;
 
         assert(m);
-        assert(send_message);
+        assert(send_signal);
 
         /* Send to all direct buses, unconditionally */
         sd_bus *b;
-        SET_FOREACH(b, m->private_buses) {
+        SET_FOREACH(b, m->private_buses_subscribed) {
 
                 /* Don't bother with enqueuing these messages to clients that haven't started yet */
                 if (sd_bus_is_ready(b) <= 0)
                         continue;
 
-                RET_GATHER(r, send_message(b, userdata));
+                RET_GATHER(r, send_signal(b, userdata));
         }
 
         /* Send to API bus, but only if somebody is subscribed */
         if (m->api_bus &&
             (sd_bus_track_count(m->subscribed) > 0 ||
              sd_bus_track_count(subscribed2) > 0))
-                RET_GATHER(r, send_message(m->api_bus, userdata));
+                RET_GATHER(r, send_signal(m->api_bus, userdata));
 
         return r;
 }

src/core/dbus.h

@@ -21,7 +21,7 @@ int bus_fdset_add_all(Manager *m, FDSet *fds);
 void bus_track_serialize(sd_bus_track *t, FILE *f, const char *prefix);
 int bus_track_coldplug(Manager *m, sd_bus_track **t, bool recursive, char **l);
 
-int bus_foreach_bus(Manager *m, sd_bus_track *subscribed2, int (*send_message)(sd_bus *bus, void *userdata), void *userdata);
+int bus_foreach_bus_signal(Manager *m, sd_bus_track *subscribed2, int (*send_signal)(sd_bus *bus, void *userdata), void *userdata);
 
 int bus_forward_agent_released(Manager *m, const char *path);
 

src/core/manager.h

@@ -332,6 +332,8 @@ struct Manager {
         /* Data specific to the D-Bus subsystem */
         sd_bus *api_bus, *system_bus;
         Set *private_buses;
+        /* Private buses on which client called Subscribe() method */
+        Set *private_buses_subscribed;
         int private_listen_fd;
         sd_event_source *private_listen_event_source;
 

src/libsystemd/sd-bus/bus-internal.h

@@ -327,6 +327,8 @@ struct sd_bus {
 
         /* zero means use value specified by $SYSTEMD_BUS_TIMEOUT= environment variable or built-in default */
         usec_t method_call_timeout;
+
+        bool queue_toggle;
 };
 
 /* For method calls we timeout at 25s, like in the D-Bus reference implementation */

src/libsystemd/sd-bus/sd-bus.c

@@ -3064,9 +3064,14 @@ static int process_running(sd_bus *bus, sd_bus_message **ret) {
         if (r != 0)
                 goto null_message;
 
-        r = dispatch_wqueue(bus);
+        /* Toggle which queue we process preferably in this iteration, the rq or the wq */
-        if (r != 0)
+        bus->queue_toggle = !bus->queue_toggle;
-                goto null_message;
+
+        if (bus->queue_toggle) { /* wqueue preferably */
+                r = dispatch_wqueue(bus);
+                if (r != 0)
+                        goto null_message;
+        }
 
         r = dispatch_track(bus);
         if (r != 0)
@@ -3075,6 +3080,12 @@ static int process_running(sd_bus *bus, sd_bus_message **ret) {
         r = dispatch_rqueue(bus, &m);
         if (r < 0)
                 return r;
+        if (r == 0 && !bus->queue_toggle) {
+                /* Nothing happened on the rq, and we didn't check the wq in this iteration, hence do that now. */
+                r = dispatch_wqueue(bus);
+                goto null_message;
+        }
+
         if (!m)
                 goto null_message;
 

src/shared/bus-wait-for-jobs.c

@@ -112,6 +112,19 @@ int bus_wait_for_jobs_new(sd_bus *bus, BusWaitForJobs **ret) {
         if (r < 0)
                 return r;
 
+        r = sd_bus_call_method_async(
+                        bus,
+                        /* slot= */ NULL,
+                        "org.freedesktop.systemd1",
+                        "/org/freedesktop/systemd1",
+                        "org.freedesktop.systemd1.Manager",
+                        "Subscribe",
+                        /* callback= */ NULL,
+                        /* userdata= */ NULL,
+                        /* types= */ NULL);
+        if (r < 0)
+                return r;
+
         *ret = TAKE_PTR(d);
 
         return 0;