Re: [BUG] af_unix: SOCK_SEQPACKET recv() returns 0 while a message is still queued

[Kuniyuki Iwashima <kuniyu@google.com>]

From: Andrew Rogers <andrew.rogers@vineyardtechnologies.org>
Date: Tue, 13 Oct 2026 14:35:26 +0000
> Hi,
> 
> A blocking recv() on an AF_UNIX SOCK_SEQPACKET socket can return 0 (EOF)
> while a message the peer sent before closing is still in the receive
> queue. The message is then lost: it is freed with
> SKB_DROP_REASON_SOCKET_CLOSE when the receiver closes its socket, along
> with any file descriptors it carried in SCM_RIGHTS.
> 
> Where it shows up
> -----------------
> 
> Chrome's Linux sandbox brokers open() for its GPU process. For each
> request the client creates a SOCK_SEQPACKET socketpair, passes one end
> to the broker process and blocks in recvmsg() on the other; the broker
> replies with sendmsg(), passing the opened fd, and closes its end. With
> the NVIDIA driver, which opens /dev/nvidiactl about 11 times per frame,
> a WebGPU page makes around 8,000 of these round trips a second, and
> within minutes one recvmsg() returns 0 although the broker's sendmsg()
> succeeded. The open() fails and the GPU process loses its device or
> crashes. A trace of one failure, with t the moment Chrome's broker
> client gives up:
> 
>   t-42.8us  broker  openat("/dev/nvidiactl") = 57
>   t-40.6us  broker  sendmsg(reply socket, 8 bytes + SCM_RIGHTS fd 57) = 8
>   t-38.7us  broker  close(reply socket)        -> unix_release_sock()
>   t-11.7us  gpu     recvmsg(other end) = 0
>   t-4us     gpu     close(other end): kfree_skb reason=SOCKET_CLOSE
>                     in unix_release_sock(), the unread reply
> 
> Reproducer
> ----------
> 
> seqpacket-eof.c is below. Each pair of threads repeats: the receiver
> makes a socketpair, hands one end to the sender and blocks in recv();
> the sender send()s 8 bytes and close()s. When recv() returns 0, the
> receiver retries with MSG_DONTWAIT and counts a loss if the message is
> there. An optional SCHED_FIFO thread on each receiver's CPU wakes from a
> timer about every 20us and spins for 3us, which preempts the receiver
> at arbitrary points and makes the race much more frequent. Without it
> the machine's own interrupts are enough, just rarely.
> 
> Results on Ubuntu 7.0.0-38-generic (7.0.14), x86_64, 8 CPUs,
> PREEMPT_DYNAMIC (lazy), HZ=1000, 2 pairs, otherwise idle desktop:
> 
>   SOCK_SEQPACKET, SCHED_FIFO interrupter (root)   30 s   7.7M rounds  61 lost
>   SOCK_SEQPACKET, interrupter as SCHED_OTHER     120 s  36.6M rounds   4 lost
>   SOCK_SEQPACKET, no interrupter, unprivileged   120 s  38.9M rounds   3 lost
>   SOCK_STREAM,    SCHED_FIFO interrupter (root)   30 s   7.6M rounds   0 lost
> 
> No kprobes or other tracing were attached during these runs.
> 
> Analysis
> --------
> 
> unix_seqpacket_recvmsg() goes through __unix_dgram_recvmsg(), which
> waits in __skb_wait_for_more_packets(). That function tests the queue
> and the shutdown state with two separate, unlocked reads:
> 
> 	if (READ_ONCE(queue->prev) != skb)
> 		goto out;
> 
> 	/* Socket shut down? */
> 	if (sk->sk_shutdown & RCV_SHUTDOWN)
> 		goto out_noerr;
> 
> On the other side, unix_dgram_sendmsg() queues the skb under
> unix_state_lock(other), and the close() that follows sets the peer's
> sk_shutdown in unix_release_sock(), also under unix_state_lock(). If the
> receiver reads queue->prev before the skb is queued and sk_shutdown
> after the close, it returns 0 with the skb in its queue. That needs the
> receiver to be held up between the two reads for about as long as the
> sender takes from queueing to close, around a microsecond here, which
> an interrupt or a preemption provides. To confirm the window, a kprobe
> on the sk_shutdown test (__skb_wait_for_more_packets+0xcc in this
> build) that busy-waits 3us made 6.8M of 6.9M rounds lose the message.
> 
> SOCK_STREAM is not affected: unix_stream_read_generic() and
> unix_stream_data_wait() look at the queue and at RCV_SHUTDOWN under
> unix_state_lock(), the lock the sender queues under.
> 
> The code in question is unchanged in v7.3-rc7. I have only tested the
> Ubuntu 7.0 kernel above.
> 
> A fix could re-check the receive queue under unix_state_lock() before
> __unix_dgram_recvmsg() reports EOF on a SEQPACKET socket, or have
> __skb_wait_for_more_packets() look at the queue again after it sees
> RCV_SHUTDOWN.

I remember this was reported before.

This behaviour predates the git era and I don't think this
is worth backporting, so I'll post this to net-next after
a few more testing and cleanup.

Thanks

---8<---
diff --git a/net/unix/af_unix.c b/net/unix/af_unix.c
index a1f8042508a8..47cca66241b4 100644
--- a/net/unix/af_unix.c
+++ b/net/unix/af_unix.c
@@ -2483,6 +2483,84 @@ static void unix_copy_addr(struct msghdr *msg, struct sock *sk)
 	}
 }
 
+static int receiver_wake_function(wait_queue_entry_t *wait, unsigned int mode, int sync,
+				  void *key)
+{
+	/*
+	 * Avoid a wakeup if event not interesting for us
+	 */
+	if (key && !(key_to_poll(key) & (EPOLLIN | EPOLLERR)))
+		return 0;
+	return autoremove_wake_function(wait, mode, sync, key);
+}
+
+static int unix_datagram_data_wait(struct sock *sk, struct sk_buff_head *queue,
+				   int *err, long *timeo_p,
+				   const struct sk_buff *skb)
+{
+	DEFINE_WAIT_FUNC(wait, receiver_wake_function);
+	int error = 0;
+
+	prepare_to_wait_exclusive(sk_sleep(sk), &wait, TASK_INTERRUPTIBLE);
+
+	if (READ_ONCE(queue->prev) != skb)
+		goto out;
+
+	if (!READ_ONCE(sk->sk_err) &&
+	    !(READ_ONCE(sk->sk_shutdown) & RCV_SHUTDOWN) &&
+	    !(sk->sk_type == SOCK_SEQPACKET && sk->sk_state == TCP_CLOSE) &&
+	    !signal_pending(current)) {
+		*timeo_p = schedule_timeout(*timeo_p);
+		goto out;
+	}
+
+	unix_state_lock(sk);
+
+	if (READ_ONCE(queue->prev) != skb)
+		goto out_unlock;
+
+	/* Socket errors? */
+	error = sock_error(sk);
+	if (error) {
+		*err = error;
+		goto out_unlock;
+	}
+
+	/* Socket shut down? */
+	if (unlikely(sk->sk_shutdown & RCV_SHUTDOWN)) {
+		*err = 0;
+		error = 1;
+		goto out_unlock;
+	}
+
+	/* Sequenced packets can come disconnected.
+	 * If so we report the problem
+	 */
+	if (sk->sk_type == SOCK_SEQPACKET &&
+	    !(sk->sk_state == TCP_ESTABLISHED || sk->sk_state == TCP_LISTEN)) {
+		*err = error = -ENOTCONN;
+		goto out_unlock;
+	}
+
+	/* handle signals */
+	if (signal_pending(current)) {
+		*err = error = sock_intr_errno(*timeo_p);
+		goto out_unlock;
+	}
+
+	unix_state_unlock(sk);
+
+	/* someone stole sock_error(). */
+	error = 0;
+	*timeo_p = schedule_timeout(*timeo_p);
+out:
+	finish_wait(sk_sleep(sk), &wait);
+	return error;
+out_unlock:
+	unix_state_unlock(sk);
+	goto out;
+}
+
 int __unix_dgram_recvmsg(struct sock *sk, struct msghdr *msg, size_t size,
 			 int flags)
 {
@@ -2517,8 +2595,8 @@ int __unix_dgram_recvmsg(struct sock *sk, struct msghdr *msg, size_t size,
 		if (err != -EAGAIN)
 			break;
 	} while (timeo &&
-		 !__skb_wait_for_more_packets(sk, &sk->sk_receive_queue,
-					      &err, &timeo, last));
+		 !unix_datagram_data_wait(sk, &sk->sk_receive_queue,
+					  &err, &timeo, last));
 
 	if (!skb) { /* implies iolock unlocked */
 		/* Signal EOF on disconnected non-blocking SEQPACKET socket. */
---8<---



> 
> Thanks,
> Andrew Rogers
> 
> ----------------------------- seqpacket-eof.c -----------------------------
> // SPDX-License-Identifier: GPL-2.0
> // AF_UNIX SOCK_SEQPACKET: a blocking recv() can return 0 (EOF) while a
> // message the peer sent before closing is still in the receive queue.
> //
> // Each pair of threads repeats: the receiver makes a socketpair, hands one
> // end to the sender and blocks in recv() on the other; the sender send()s
> // an 8-byte message and close()s its end. If recv() returns 0, the receiver
> // retries with MSG_DONTWAIT; getting the message then shows EOF was
> // reported while it was queued.
> //
> // The loss needs the receiver to be held up for about a microsecond inside
> // __skb_wait_for_more_packets(), after it has seen the queue unchanged and
> // before it tests sk_shutdown. Interrupts and preemption do that. An
> // optional SCHED_FIFO thread on each receiver's CPU wakes from a timer
> // about every PERIOD us and spins for SPIN us, preempting the receiver
> // wherever it is, which makes the race much more frequent. Without root it
> // runs as an ordinary thread; the race still happens, rarely.
> //
> //   cc -O2 -Wall -pthread -o seqpacket-eof seqpacket-eof.c
> //   sudo ./seqpacket-eof [pairs [seconds [period_us spin_us [stream]]]]
> //
> // Defaults: 2 pairs, 30 s, period 20 us, spin 3 us. "0 0" disables the
> // interrupter; "stream" uses SOCK_STREAM instead, for comparison. Needs two
> // CPUs per pair. Exits 2 if any message was lost.
> #define _GNU_SOURCE
> #include <pthread.h>
> #include <sched.h>
> #include <stdatomic.h>
> #include <stdint.h>
> #include <stdio.h>
> #include <stdlib.h>
> #include <string.h>
> #include <sys/socket.h>
> #include <time.h>
> #include <unistd.h>
> 
> struct pair {
>     int recv_cpu, send_cpu;
>     _Atomic int handoff;    // the sender's socket fd, or -1
>     _Atomic int stop;
>     unsigned long rounds, eof_with_data, eof_empty;
> };
> 
> static int period_us = 20, spin_us = 3;
> static int sock_type = SOCK_SEQPACKET;
> static _Atomic int stop_all;
> 
> static void die(const char *what)
> {
>     perror(what);
>     exit(1);
> }
> 
> static void pin(int cpu)
> {
>     cpu_set_t set;
> 
>     CPU_ZERO(&set);
>     CPU_SET(cpu, &set);
>     pthread_setaffinity_np(pthread_self(), sizeof(set), &set);
> }
> 
> static uint64_t now_ns(void)
> {
>     struct timespec ts;
> 
>     clock_gettime(CLOCK_MONOTONIC, &ts);
>     return (uint64_t)ts.tv_sec * 1000000000u + ts.tv_nsec;
> }
> 
> static void *sender(void *arg)
> {
>     struct pair *p = arg;
>     const char msg[8] = "reply!!";
> 
>     pin(p->send_cpu);
>     while (!atomic_load(&p->stop)) {
>         int fd = atomic_exchange(&p->handoff, -1);
> 
>         if (fd < 0)
>             continue;
>         if (send(fd, msg, sizeof(msg), 0) != sizeof(msg))
>             die("send");
>         close(fd);
>     }
>     return NULL;
> }
> 
> static void *receiver(void *arg)
> {
>     struct pair *p = arg;
>     char buf[64];
>     int sv[2];
> 
>     pin(p->recv_cpu);
>     while (!atomic_load(&p->stop)) {
>         ssize_t n;
> 
>         if (socketpair(AF_UNIX, sock_type, 0, sv) < 0)
>             die("socketpair");
>         atomic_store(&p->handoff, sv[1]);
>         n = recv(sv[0], buf, sizeof(buf), 0);
>         if (n == 0) {
>             if (recv(sv[0], buf, sizeof(buf), MSG_DONTWAIT) > 0)
>                 p->eof_with_data++;
>             else
>                 p->eof_empty++;
>         } else if (n < 0) {
>             die("recv");
>         }
>         close(sv[0]);
>         p->rounds++;
>     }
>     return NULL;
> }
> 
> static void *interrupter(void *arg)
> {
>     int cpu = (int)(intptr_t)arg;
>     unsigned int seed = (unsigned int)cpu * 2654435761u;
>     long period_ns = (long)period_us * 1000;
> 
>     pin(cpu);
>     while (!atomic_load(&stop_all)) {
>         // Jitter the period so the wakeups land at every point of the
>         // receiver's path.
>         struct timespec ts = {
>             0, period_ns / 2 + rand_r(&seed) % (period_ns + 1)
>         };
>         uint64_t until;
> 
>         clock_nanosleep(CLOCK_MONOTONIC, 0, &ts, NULL);
>         until = now_ns() + (uint64_t)spin_us * 1000;
>         while (now_ns() < until)
>             ;
>     }
>     return NULL;
> }
> 
> static int start_interrupter(int cpu)
> {
>     struct sched_param sp = { .sched_priority = 1 };
>     void *arg = (void *)(intptr_t)cpu;
>     pthread_attr_t attr;
>     pthread_t t;
> 
>     pthread_attr_init(&attr);
>     pthread_attr_setinheritsched(&attr, PTHREAD_EXPLICIT_SCHED);
>     pthread_attr_setschedpolicy(&attr, SCHED_FIFO);
>     pthread_attr_setschedparam(&attr, &sp);
>     if (pthread_create(&t, &attr, interrupter, arg) == 0)
>         return 1;
>     pthread_create(&t, NULL, interrupter, arg);
>     return 0;
> }
> 
> int main(int argc, char **argv)
> {
>     int pairs = argc > 1 ? atoi(argv[1]) : 2;
>     int seconds = argc > 2 ? atoi(argv[2]) : 30;
>     int ncpu = (int)sysconf(_SC_NPROCESSORS_ONLN);
>     unsigned long rounds = 0, bad = 0, empty = 0;
>     struct pair *ps;
>     int rt = 0;
> 
>     if (argc > 4) {
>         period_us = atoi(argv[3]);
>         spin_us = atoi(argv[4]);
>     }
>     if (argc > 5 && strcmp(argv[5], "stream") == 0)
>         sock_type = SOCK_STREAM;
>     if (pairs * 2 > ncpu) {
>         fprintf(stderr, "need two CPUs per pair (%d online)\n", ncpu);
>         return 1;
>     }
> 
>     ps = calloc(pairs, sizeof(*ps));
>     for (int i = 0; i < pairs; i++) {
>         pthread_t t;
> 
>         ps[i].recv_cpu = 2 * i;
>         ps[i].send_cpu = 2 * i + 1;
>         atomic_store(&ps[i].handoff, -1);
>         pthread_create(&t, NULL, sender, &ps[i]);
>         pthread_create(&t, NULL, receiver, &ps[i]);
>         if (period_us > 0)
>             rt = start_interrupter(ps[i].recv_cpu);
>     }
> 
>     printf("type=%s pairs=%d seconds=%d period=%dus spin=%dus "
>            "interrupter=%s\n",
>            sock_type == SOCK_STREAM ? "SOCK_STREAM" : "SOCK_SEQPACKET",
>            pairs, seconds, period_us, spin_us,
>            period_us <= 0 ? "none" :
>            rt ? "SCHED_FIFO" : "SCHED_OTHER (SCHED_FIFO not permitted)");
>     fflush(stdout);
> 
>     for (int s = 1; s <= seconds; s++) {
>         sleep(1);
>         rounds = bad = empty = 0;
>         for (int i = 0; i < pairs; i++) {
>             rounds += ps[i].rounds;
>             bad += ps[i].eof_with_data;
>             empty += ps[i].eof_empty;
>         }
>         if (s % 5 == 0 || s == seconds) {
>             printf("t=%ds rounds=%lu eof_with_message_queued=%lu "
>                    "eof_empty=%lu\n", s, rounds, bad, empty);
>             fflush(stdout);
>         }
>     }
>     atomic_store(&stop_all, 1);
>     return bad ? 2 : 0;
> }
>