2 files changed, 68 insertions, 6 deletions

diff --git a/src/libsystemd/sd-bus/bus-message.c b/src/libsystemd/sd-bus/bus-message.c
index 306b6d6816..7fe8929f82 100644
--- a/src/libsystemd/sd-bus/bus-message.c
+++ b/src/libsystemd/sd-bus/bus-message.c
@@ -120,7 +120,8 @@ static sd_bus_message* message_free(sd_bus_message *m) {
 
         message_reset_parts(m);
 
-        sd_bus_unref(m->bus);
+        /* Note that we don't unref m->bus here. That's already done by sd_bus_message_unref() as each user
+         * reference to the bus message also is considered a reference to the bus connection itself. */
 
         if (m->free_fds) {
                 close_many(m->fds, m->n_fds);
@@ -893,27 +894,76 @@ int bus_message_new_synthetic_error(
 }
 
 _public_ sd_bus_message* sd_bus_message_ref(sd_bus_message *m) {
-
         if (!m)
                 return NULL;
 
-        assert(m->n_ref > 0);
+        /* We are fine if this message so far was either explicitly reffed or not reffed but queued into at
+         * least one bus connection object. */
+        assert(m->n_ref > 0 || m->n_queued > 0);
+
         m->n_ref++;
 
+        /* Each user reference to a bus message shall also be considered a ref on the bus */
+        sd_bus_ref(m->bus);
         return m;
 }
 
 _public_ sd_bus_message* sd_bus_message_unref(sd_bus_message *m) {
-
         if (!m)
                 return NULL;
 
         assert(m->n_ref > 0);
+
+        sd_bus_unref(m->bus); /* Each regular ref is also a ref on the bus connection. Let's hence drop it
+                               * here. Note we have to do this before decrementing our own n_ref here, since
+                               * otherwise, if this message is currently queued sd_bus_unref() might call
+                               * bus_message_unref_queued() for this which might then destroy the message
+                               * while we are still processing it. */
         m->n_ref--;
 
-        if (m->n_ref > 0)
+        if (m->n_ref > 0 || m->n_queued > 0)
                 return NULL;
 
+        /* Unset the bus field if neither the user has a reference nor this message is queued. We are careful
+         * to reset the field only after the last reference to the bus is dropped, after all we might keep
+         * multiple references to the bus, once for each reference kept on outselves. */
+        m->bus = NULL;
+
+        return message_free(m);
+}
+
+sd_bus_message* bus_message_ref_queued(sd_bus_message *m, sd_bus *bus) {
+        if (!m)
+                return NULL;
+
+        /* If this is a different bus than the message is associated with, then implicitly turn this into a
+         * regular reference. This means that you can create a memory leak by enqueuing a message generated
+         * on one bus onto another at the same time as enqueueing a message from the second one on the first,
+         * as we'll not detect the cyclic references there. */
+        if (bus != m->bus)
+                return sd_bus_message_ref(m);
+
+        assert(m->n_ref > 0 || m->n_queued > 0);
+        m->n_queued++;
+
+        return m;
+}
+
+sd_bus_message* bus_message_unref_queued(sd_bus_message *m, sd_bus *bus) {
+        if (!m)
+                return NULL;
+
+        if (bus != m->bus)
+                return sd_bus_message_unref(m);
+
+        assert(m->n_queued > 0);
+        m->n_queued--;
+
+        if (m->n_ref > 0 || m->n_queued > 0)
+                return NULL;
+
+        m->bus = NULL;
+
         return message_free(m);
 }
 
diff --git a/src/libsystemd/sd-bus/bus-message.h b/src/libsystemd/sd-bus/bus-message.h
index 97f6060e30..ded88005e2 100644
--- a/src/libsystemd/sd-bus/bus-message.h
+++ b/src/libsystemd/sd-bus/bus-message.h
@@ -51,7 +51,16 @@ struct bus_body_part {
 };
 
 struct sd_bus_message {
-        unsigned n_ref;
+        /* Caveat: a message can be referenced in two different ways: the main (user-facing) way will also
+         * pin the bus connection object the message is associated with. The secondary way ("queued") is used
+         * when a message is in the read or write queues of the bus connection object, which will not pin the
+         * bus connection object. This is necessary so that we don't have to have a pair of cyclic references
+         * between a message that is queued and its connection: as soon as a message is only referenced by
+         * the connection (by means of being queued) and the connection itself has no other references it
+         * will be freed. */
+
+        unsigned n_ref;     /* Counter of references that pin the connection */
+        unsigned n_queued;  /* Counter of references that do not pin the connection */
 
         sd_bus *bus;
 
@@ -216,3 +225,6 @@ int bus_message_append_sender(sd_bus_message *m, const char *sender);
 
 void bus_message_set_sender_driver(sd_bus *bus, sd_bus_message *m);
 void bus_message_set_sender_local(sd_bus *bus, sd_bus_message *m);
+
+sd_bus_message* bus_message_ref_queued(sd_bus_message *m, sd_bus *bus);
+sd_bus_message* bus_message_unref_queued(sd_bus_message *m, sd_bus *bus);