sd-netlink: also manage received messages by serial

Then, we can easily find the received message matching with requested
serial.

3 files changed, 41 insertions, 13 deletions

diff --git a/src/libsystemd/sd-netlink/netlink-internal.h b/src/libsystemd/sd-netlink/netlink-internal.h
index 9ad4c03a3c..bca13bce57 100644
--- a/src/libsystemd/sd-netlink/netlink-internal.h
+++ b/src/libsystemd/sd-netlink/netlink-internal.h
@@ -74,6 +74,7 @@ struct sd_netlink {
         bool broadcast_group_dont_leave:1; /* until we can rely on 4.2 */
 
         OrderedSet *rqueue;
+        Hashmap *rqueue_by_serial;
         Hashmap *rqueue_partial_by_serial;
 
         struct nlmsghdr *rbuffer;
diff --git a/src/libsystemd/sd-netlink/netlink-socket.c b/src/libsystemd/sd-netlink/netlink-socket.c
index 95ac7bad1a..3c7447c73c 100644
--- a/src/libsystemd/sd-netlink/netlink-socket.c
+++ b/src/libsystemd/sd-netlink/netlink-socket.c
@@ -247,6 +247,7 @@ DEFINE_PRIVATE_HASH_OPS_WITH_VALUE_DESTRUCTOR(
         sd_netlink_message, sd_netlink_message_unref);
 
 static int netlink_queue_received_message(sd_netlink *nl, sd_netlink_message *m) {
+        uint32_t serial;
         int r;
 
         assert(nl);
@@ -261,6 +262,36 @@ static int netlink_queue_received_message(sd_netlink *nl, sd_netlink_message *m)
                 return r;
 
         sd_netlink_message_ref(m);
+
+        if (sd_netlink_message_is_broadcast(m))
+                return 0;
+
+        serial = message_get_serial(m);
+        if (serial == 0)
+                return 0;
+
+        if (sd_netlink_message_get_errno(m) < 0) {
+                _cleanup_(sd_netlink_message_unrefp) sd_netlink_message *old = NULL;
+
+                old = hashmap_remove(nl->rqueue_by_serial, UINT32_TO_PTR(serial));
+                if (old)
+                        log_debug("sd-netlink: received error message with serial %"PRIu32", but another message with "
+                                  "the same serial is already stored in the read queue, replacing.", serial);
+        }
+
+        r = hashmap_ensure_put(&nl->rqueue_by_serial, &netlink_message_hash_ops, UINT32_TO_PTR(serial), m);
+        if (r == -EEXIST) {
+                if (!sd_netlink_message_is_error(m))
+                        log_debug("sd-netlink: received message with serial %"PRIu32", but another message with "
+                                  "the same serial is already stored in the read queue, ignoring.", serial);
+                return 0;
+        }
+        if (r < 0) {
+                sd_netlink_message_unref(ordered_set_remove(nl->rqueue, m));
+                return r;
+        }
+
+        sd_netlink_message_ref(m);
         return 0;
 }
 
diff --git a/src/libsystemd/sd-netlink/sd-netlink.c b/src/libsystemd/sd-netlink/sd-netlink.c
index 1f4f8e9015..6eb7f659ae 100644
--- a/src/libsystemd/sd-netlink/sd-netlink.c
+++ b/src/libsystemd/sd-netlink/sd-netlink.c
@@ -120,6 +120,7 @@ static sd_netlink *netlink_free(sd_netlink *nl) {
         assert(nl);
 
         ordered_set_free(nl->rqueue);
+        hashmap_free(nl->rqueue_by_serial);
         hashmap_free(nl->rqueue_partial_by_serial);
         free(nl->rbuffer);
 
@@ -186,6 +187,7 @@ static int dispatch_rqueue(sd_netlink *nl, sd_netlink_message **ret) {
 
         /* Dispatch a queued message */
         m = ordered_set_steal_first(nl->rqueue);
+        sd_netlink_message_unref(hashmap_remove_value(nl->rqueue_by_serial, UINT32_TO_PTR(message_get_serial(m)), m));
         *ret = m;
         return !!m;
 }
@@ -528,27 +530,21 @@ int sd_netlink_read(
         timeout = calc_elapse(usec);
 
         for (;;) {
-                sd_netlink_message *m;
+                _cleanup_(sd_netlink_message_unrefp) sd_netlink_message *m = NULL;
                 usec_t left;
 
-                ORDERED_SET_FOREACH(m, nl->rqueue) {
-                        _cleanup_(sd_netlink_message_unrefp) sd_netlink_message *incoming = NULL;
-                        uint32_t received_serial;
+                m = hashmap_remove(nl->rqueue_by_serial, UINT32_TO_PTR(serial));
+                if (m) {
                         uint16_t type;
 
-                        received_serial = message_get_serial(m);
-                        if (received_serial != serial)
-                                continue;
-
                         /* found a match, remove from rqueue and return it */
-                        ordered_set_remove(nl->rqueue, m);
-                        incoming = TAKE_PTR(m);
+                        sd_netlink_message_unref(ordered_set_remove(nl->rqueue, m));
 
-                        r = sd_netlink_message_get_errno(incoming);
+                        r = sd_netlink_message_get_errno(m);
                         if (r < 0)
                                 return r;
 
-                        r = sd_netlink_message_get_type(incoming, &type);
+                        r = sd_netlink_message_get_type(m, &type);
                         if (r < 0)
                                 return r;
 
@@ -559,7 +555,7 @@ int sd_netlink_read(
                         }
 
                         if (ret)
-                                *ret = TAKE_PTR(incoming);
+                                *ret = TAKE_PTR(m);
                         return 1;
                 }