sd-netlink: introduce sd_netlink_ignore_serial()

When we send a message with NLM_F_ACK, but if later we are not
interested in the reply and do not want to call sd_netlink_read(),
the reply will be stored in the rqueue forever.

Let's introduce a way to ignore received message without waiting reply.

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

@@ -55,6 +55,11 @@ typedef struct sd_netlink_slot {
         };
 } sd_netlink_slot;
 
+typedef struct NetlinkIgnoredSerial {
+        uint32_t serial;
+        usec_t timeout_usec; /* timestamp in CLOCK_MONOTONIC */
+} NetlinkIgnoredSerial;
+
 typedef struct sd_netlink {
         unsigned n_ref;
 
@@ -78,6 +83,7 @@ typedef struct sd_netlink {
         bool processing:1;
 
         uint32_t serial;
+        Hashmap *ignored_serials;
 
         struct Prioq *reply_callbacks_prioq;
         Hashmap *reply_callbacks;

src/libsystemd/sd-netlink/netlink-socket.c

@@ -222,6 +222,16 @@ static int netlink_queue_received_message(sd_netlink *nl, sd_netlink_message *m)
         assert(nl);
         assert(m);
 
+        serial = message_get_serial(m);
+        if (serial != 0) {
+                NetlinkIgnoredSerial *s = hashmap_remove(nl->ignored_serials, UINT32_TO_PTR(serial));
+                if (s) {
+                        /* We are not interested in the message anymore. */
+                        free(s);
+                        return 0;
+                }
+        }
+
         if (ordered_set_size(nl->rqueue) >= NETLINK_RQUEUE_MAX)
                 return log_debug_errno(SYNTHETIC_ERRNO(ENOBUFS),
                                        "sd-netlink: exhausted the read queue size (%d)", NETLINK_RQUEUE_MAX);
@@ -235,7 +245,6 @@ static int netlink_queue_received_message(sd_netlink *nl, sd_netlink_message *m)
         if (sd_netlink_message_is_broadcast(m))
                 return 0;
 
-        serial = message_get_serial(m);
         if (serial == 0)
                 return 0;
 

src/libsystemd/sd-netlink/sd-netlink.c

@@ -126,6 +126,8 @@ static sd_netlink *netlink_free(sd_netlink *nl) {
 
         assert(nl);
 
+        hashmap_free(nl->ignored_serials);
+
         ordered_set_free(nl->rqueue);
         hashmap_free(nl->rqueue_by_serial);
         hashmap_free(nl->rqueue_partial_by_serial);
@@ -190,6 +192,58 @@ static usec_t timespan_to_timestamp(sd_netlink *nl, usec_t usec) {
         return usec_add(netlink_now(nl, CLOCK_MONOTONIC), usec);
 }
 
+static void netlink_trim_ignored_serials(sd_netlink *nl) {
+        NetlinkIgnoredSerial *s;
+        usec_t now_usec = 0;
+
+        assert(nl);
+
+        HASHMAP_FOREACH(s, nl->ignored_serials) {
+                if (s->timeout_usec == USEC_INFINITY)
+                        continue;
+
+                if (now_usec == 0)
+                        now_usec = netlink_now(nl, CLOCK_MONOTONIC);
+
+                if (s->timeout_usec < now_usec)
+                        free(hashmap_remove(nl->ignored_serials, UINT32_TO_PTR(s->serial)));
+        }
+}
+
+int sd_netlink_ignore_serial(sd_netlink *nl, uint32_t serial, uint64_t timeout_usec) {
+        int r;
+
+        assert_return(nl, -EINVAL);
+        assert_return(!netlink_pid_changed(nl), -ECHILD);
+        assert_return(serial != 0, -EINVAL);
+
+        timeout_usec = timespan_to_timestamp(nl, timeout_usec);
+
+        NetlinkIgnoredSerial *existing = hashmap_get(nl->ignored_serials, UINT32_TO_PTR(serial));
+        if (existing) {
+                existing->timeout_usec = timeout_usec;
+                return 0;
+        }
+
+        netlink_trim_ignored_serials(nl);
+
+        _cleanup_free_ NetlinkIgnoredSerial *s = new(NetlinkIgnoredSerial, 1);
+        if (!s)
+                return -ENOMEM;
+
+        *s = (NetlinkIgnoredSerial) {
+                .serial = serial,
+                .timeout_usec = timeout_usec,
+        };
+
+        r = hashmap_ensure_put(&nl->ignored_serials, &trivial_hash_ops_value_free, UINT32_TO_PTR(s->serial), s);
+        if (r < 0)
+                return r;
+
+        TAKE_PTR(s);
+        return 0;
+}
+
 int sd_netlink_send(
                 sd_netlink *nl,
                 sd_netlink_message *message,
@@ -373,6 +427,8 @@ static int process_running(sd_netlink *nl, sd_netlink_message **ret) {
 
         assert(nl);
 
+        netlink_trim_ignored_serials(nl);
+
         r = process_timeout(nl);
         if (r != 0)
                 goto null_message;

src/systemd/sd-netlink.h

@@ -56,6 +56,8 @@ int sd_netlink_call_async(sd_netlink *nl, sd_netlink_slot **ret_slot, sd_netlink
 int sd_netlink_call(sd_netlink *nl, sd_netlink_message *message, uint64_t timeout, sd_netlink_message **ret);
 int sd_netlink_read(sd_netlink *nl, uint32_t serial, uint64_t timeout, sd_netlink_message **ret);
 
+int sd_netlink_ignore_serial(sd_netlink *nl, uint32_t serial, uint64_t timeout_usec);
+
 int sd_netlink_get_events(sd_netlink *nl);
 int sd_netlink_get_timeout(sd_netlink *nl, uint64_t *ret);
 int sd_netlink_process(sd_netlink *nl, sd_netlink_message **ret);