From: Todd C. Miller <Todd.Miller@sudo.ws>
Date: Tue, 6 May 2025 22:39:14 +0000 (-0600)
Subject: flush_ports: flush POSIX message queues properly
X-Git-Tag: v258-rc1~670
X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=ffb6adb76367d5ab7d43937ccaac5947717b5b78;p=thirdparty%2Fsystemd.git

flush_ports: flush POSIX message queues properly

On Linux, read() on a message queue descriptor returns the message
queue statistics, not the actual message queue data.  We need to use
mq_receive() to drain the queues instead.

Fixes a problem where a POSIX message queue socket unit with messages
in the queue at shutdown time could result in a hang on reboot/shutdown.
---

diff --git a/src/basic/socket-util.c b/src/basic/socket-util.c
index 61a77370eb4..5e6195e108e 100644
--- a/src/basic/socket-util.c
+++ b/src/basic/socket-util.c
@@ -5,6 +5,7 @@
 #include <limits.h>
 #include <linux/if.h>
 #include <linux/if_arp.h>
+#include <mqueue.h>
 #include <net/if.h>
 #include <netdb.h>
 #include <netinet/ip.h>
@@ -1319,6 +1320,54 @@ int flush_accept(int fd) {
         }
 }
 
+ssize_t flush_mqueue(int fd) {
+        _cleanup_free_ char *buf = NULL;
+        struct mq_attr attr;
+        ssize_t count = 0;
+        int r;
+
+        assert(fd >= 0);
+
+        /* Similar to flush_fd() but flushes all messages from a POSIX message queue. */
+
+        for (;;) {
+                ssize_t l;
+
+                r = fd_wait_for_event(fd, POLLIN, /* timeout= */ 0);
+                if (r < 0) {
+                        if (r == -EINTR)
+                                continue;
+
+                        return r;
+                }
+                if (r == 0)
+                        return count;
+
+                if (!buf) {
+                        /* Buffer must be at least as large as mq_msgsize. */
+                        if (mq_getattr(fd, &attr) < 0)
+                                return -errno;
+
+                        buf = malloc(attr.mq_msgsize);
+                        if (!buf)
+                                return -ENOMEM;
+                }
+
+                l = mq_receive(fd, buf, attr.mq_msgsize, /* msg_prio = */ NULL);
+                if (l < 0) {
+                        if (errno == EINTR)
+                                continue;
+
+                        if (errno == EAGAIN)
+                                return count;
+
+                        return -errno;
+                }
+
+                count += l;
+        }
+}
+
 struct cmsghdr* cmsg_find(struct msghdr *mh, int level, int type, socklen_t length) {
         struct cmsghdr *cmsg;
 
diff --git a/src/basic/socket-util.h b/src/basic/socket-util.h
index a69277c6396..e9671c38333 100644
--- a/src/basic/socket-util.h
+++ b/src/basic/socket-util.h
@@ -199,6 +199,7 @@ int receive_many_fds(int transport_fd, int **ret_fds_array, size_t *ret_n_fds_ar
 ssize_t next_datagram_size_fd(int fd);
 
 int flush_accept(int fd);
+ssize_t flush_mqueue(int fd);
 
 #define CMSG_FOREACH(cmsg, mh)                                          \
         for ((cmsg) = CMSG_FIRSTHDR(mh); (cmsg); (cmsg) = CMSG_NXTHDR((mh), (cmsg)))
diff --git a/src/core/socket.c b/src/core/socket.c
index 1583c33cf48..0de430b97ed 100644
--- a/src/core/socket.c
+++ b/src/core/socket.c
@@ -2188,8 +2188,12 @@ static void flush_ports(Socket *s) {
                 if (p->fd < 0)
                         continue;
 
-                (void) flush_accept(p->fd);
-                (void) flush_fd(p->fd);
+                if (p->type == SOCKET_MQUEUE)
+                        (void) flush_mqueue(p->fd);
+                else {
+                        (void) flush_accept(p->fd);
+                        (void) flush_fd(p->fd);
+                }
         }
 }