sd-rtnl: fix self-reference leaks

Like sd-bus, sd-rtnl can have self-references through queued messages. In
particular, each queued message has the following self-ref loop:
  rtnl->wqueue[i]->rtnl == rtnl
Same is true for "rqueue".

When sd_rtnl_unref() gets called, we must therefore make sure we correctly
consider each self-reference when deciding to destroy the object. For each
queued message, there _might_ be one ref. However, rtnl-messages can be
created _without_ a bus-reference, therefore we need to verify the
actually required ref-count.

Once we know exactly how many self-refs exist, and we verified none of the
queued messages has external references, we can destruct the object.
We must immediately drop our own reference, then flush all queues and
destroy the bus object. Otherwise, each sd_rtnl_message_unref() call would
recurse into the same destruction logic as they enter with the same
rtnl-refcnt.

Note: We really should verify _all_ queued messages have m->rtnl set to
      the bus they're queued on. If that's given, we can change:
        if (REFCNT_GET(rtnl->n_ref) <= refs)
      to
        if (REFCNT_GET(rtnl->n_ref) == refs)
      and thus avoid recalculating the required refs for each message we
      remove from the queue during destruction.

src/libsystemd/sd-rtnl/sd-rtnl.c

@@ -111,31 +111,82 @@ sd_rtnl *sd_rtnl_ref(sd_rtnl *rtnl) {
 }
 
 sd_rtnl *sd_rtnl_unref(sd_rtnl *rtnl) {
+        unsigned long refs;
 
-        if (rtnl && REFCNT_DEC(rtnl->n_ref) <= 0) {
+        if (!rtnl)
+                return NULL;
+
+        /*
+         * If our ref-cnt is exactly the number of internally queued messages
+         * plus the ref-cnt to be dropped, then we know there's no external
+         * reference to us. Hence, we look through all queued messages and if
+         * they also have no external references, we're about to drop the last
+         * ref. Flush the queues so the REFCNT_DEC() below will drop to 0.
+         * We must be careful not to introduce inter-message references or this
+         * logic will fall apart..
+         */
+
+        refs = rtnl->rqueue_size + rtnl->wqueue_size + 1;
+
+        if (REFCNT_GET(rtnl->n_ref) <= refs) {
                 struct match_callback *f;
+                bool q = true;
                 unsigned i;
 
-                for (i = 0; i < rtnl->rqueue_size; i++)
+                for (i = 0; i < rtnl->rqueue_size; i++) {
-                        sd_rtnl_message_unref(rtnl->rqueue[i]);
+                        if (REFCNT_GET(rtnl->rqueue[i]->n_ref) > 1) {
-                free(rtnl->rqueue);
+                                q = false;
-
+                                break;
-                for (i = 0; i < rtnl->wqueue_size; i++)
+                        } else if (rtnl->rqueue[i]->rtnl != rtnl)
-                        sd_rtnl_message_unref(rtnl->wqueue[i]);
+                                --refs;
-                free(rtnl->wqueue);
-
-                hashmap_free_free(rtnl->reply_callbacks);
-                prioq_free(rtnl->reply_callbacks_prioq);
-
-                while ((f = rtnl->match_callbacks)) {
-                        LIST_REMOVE(match_callbacks, rtnl->match_callbacks, f);
-                        free(f);
                 }
 
-                safe_close(rtnl->fd);
+                if (q) {
-                free(rtnl);
+                        for (i = 0; i < rtnl->wqueue_size; i++) {
+                                if (REFCNT_GET(rtnl->wqueue[i]->n_ref) > 1) {
+                                        q = false;
+                                        break;
+                                } else if (rtnl->wqueue[i]->rtnl != rtnl)
+                                        --refs;
+                        }
+                }
+
+                if (q && REFCNT_GET(rtnl->n_ref) == refs) {
+                        /* Drop our own ref early to avoid recursion from:
+                         *   sd_rtnl_message_unref()
+                         *     sd_rtnl_unref()
+                         * These must enter sd_rtnl_unref() with a ref-cnt
+                         * smaller than us. */
+                        REFCNT_DEC(rtnl->n_ref);
+
+                        for (i = 0; i < rtnl->rqueue_size; i++)
+                                sd_rtnl_message_unref(rtnl->rqueue[i]);
+                        free(rtnl->rqueue);
+
+                        for (i = 0; i < rtnl->wqueue_size; i++)
+                                sd_rtnl_message_unref(rtnl->wqueue[i]);
+                        free(rtnl->wqueue);
+
+                        assert_se(REFCNT_GET(rtnl->n_ref) == 0);
+
+                        hashmap_free_free(rtnl->reply_callbacks);
+                        prioq_free(rtnl->reply_callbacks_prioq);
+
+                        while ((f = rtnl->match_callbacks)) {
+                                LIST_REMOVE(match_callbacks, rtnl->match_callbacks, f);
+                                free(f);
+                        }
+
+                        safe_close(rtnl->fd);
+                        free(rtnl);
+
+                        return NULL;
+                }
         }
 
+        assert_se(REFCNT_GET(rtnl->n_ref) > 0);
+        REFCNT_DEC(rtnl->n_ref);
+
         return NULL;
 }