[PATCH 5/5] vsock: defer RX readable notifications until batch unlock

[PATCH 5/5] vsock: defer RX readable notifications until batch unlock

[physicalmtea]

From: Jia Jia <physicalmtea@gmail.com>

An RX batch can make a socket readable while the worker still owns its
socket lock. Waking a reader at that point can make it immediately wait
for the same lock.

For the callback installed by sock_init_data(), record when the existing
low-watermark condition is met and issue one notification after the batch
releases the lock. Save the initial sock_def_readable() callback when the
AF_VSOCK socket is created. virtio_transport_common can be built as a
module and cannot refer to that unexported symbol directly.

Custom data-ready callbacks retain per-packet notification behavior. If
the callback changes while a default notification is pending, finish the
batch before returning to the per-packet path.

Keep the socket reference until the deferred callback completes. A sockmap
attachment takes the socket lock, so data queued before the attachment is
reported through the saved default callback.

Control packets, state transitions, transport changes and non-batched
sockets retain their existing behavior.

Signed-off-by: Jia Jia <physicalmtea@gmail.com>
---
 include/linux/virtio_vsock.h            |  1 +
 include/net/af_vsock.h                  |  3 +-
 net/vmw_vsock/af_vsock.c                |  1 +
 net/vmw_vsock/virtio_transport_common.c | 53 ++++++++++++++++++++++++++-----
 4 files changed, 49 insertions(+), 9 deletions(-)

diff --git a/include/linux/virtio_vsock.h b/include/linux/virtio_vsock.h
index f67fa99ec..d69c09d5c 100644
--- a/include/linux/virtio_vsock.h
+++ b/include/linux/virtio_vsock.h
@@ -289,6 +289,7 @@ struct virtio_transport_rx_batch {
 	struct sockaddr_vm src;
 	struct sockaddr_vm dst;
 	bool write_space_pending;
+	bool data_ready_pending;
 };
 
 void virtio_transport_recv_pkt_batch(struct virtio_transport *t,
diff --git a/include/net/af_vsock.h b/include/net/af_vsock.h
index 9d8ae6220..a799f0855 100644
--- a/include/net/af_vsock.h
+++ b/include/net/af_vsock.h
@@ -63,7 +63,8 @@ struct vsock_sock {
 	u32 peer_shutdown;
 	bool sent_request;
 	bool ignore_connecting_rst;
-	/* Initial callback, used to identify replacements. */
+	/* Initial callbacks, used to identify replacements. */
+	void (*default_data_ready)(struct sock *sk);
 	void (*default_write_space)(struct sock *sk);
 
 	/* Protected by lock_sock(sk) */
diff --git a/net/vmw_vsock/af_vsock.c b/net/vmw_vsock/af_vsock.c
index e5290a3bb..1a25b5393 100644
--- a/net/vmw_vsock/af_vsock.c
+++ b/net/vmw_vsock/af_vsock.c
@@ -958,6 +958,7 @@ static struct sock *__vsock_create(struct net *net,
 		sk->sk_type = type;
 
 	vsk = vsock_sk(sk);
+	vsk->default_data_ready = sk->sk_data_ready;
 	vsk->default_write_space = sk->sk_write_space;
 	vsock_addr_init(&vsk->local_addr, VMADDR_CID_ANY, VMADDR_PORT_ANY);
 	vsock_addr_init(&vsk->remote_addr, VMADDR_CID_ANY, VMADDR_PORT_ANY);
diff --git a/net/vmw_vsock/virtio_transport_common.c b/net/vmw_vsock/virtio_transport_common.c
index 78c4e2f9e..e337a7aaf 100644
--- a/net/vmw_vsock/virtio_transport_common.c
+++ b/net/vmw_vsock/virtio_transport_common.c
@@ -1583,7 +1583,8 @@ virtio_transport_recv_enqueue(struct vsock_sock *vsk,
 
 static int
 virtio_transport_recv_connected(struct sock *sk,
-				struct sk_buff *skb)
+				struct sk_buff *skb,
+				bool *data_ready_pending)
 {
 	struct virtio_vsock_hdr *hdr = virtio_vsock_hdr(skb);
 	struct vsock_sock *vsk = vsock_sk(sk);
@@ -1602,7 +1603,14 @@ virtio_transport_recv_connected(struct sock *sk,
 			vsock_remove_sock(vsk);
 			break;
 		}
-		vsock_data_ready(sk);
+		if (!data_ready_pending ||
+		    READ_ONCE(sk->sk_data_ready) != vsk->default_data_ready) {
+			vsock_data_ready(sk);
+		} else if (!*data_ready_pending &&
+			   (vsock_stream_has_data(vsk) >= sk->sk_rcvlowat ||
+			    sock_flag(sk, SOCK_DONE))) {
+			*data_ready_pending = true;
+		}
 		return err;
 	case VIRTIO_VSOCK_OP_CREDIT_REQUEST:
 		virtio_transport_send_credit_update(vsk);
@@ -1836,6 +1844,7 @@ struct virtio_transport_rx_pkt_ctx {
 	const struct sockaddr_vm *dst;
 	bool *batchable;
 	struct virtio_transport_rx_batch *batch;
+	bool defer_data_ready;
 };
 
 static bool
@@ -1852,9 +1861,11 @@ virtio_transport_recv_pkt_batchable(struct virtio_transport *t,
 }
 
 /*
- * The caller holds sk's socket lock.  Set @batchable if the socket can remain
- * locked for another ordinary STREAM/RW packet.  Return true if the caller
- * must free @skb.
+ * The caller holds sk's socket lock. The packet context optionally records
+ * callbacks to deliver when the batch finishes. If defer_data_ready is true,
+ * defer the default data-ready callback in the batch. Set batchable if the
+ * socket can remain locked for another ordinary STREAM/RW packet. Return
+ * true if the caller must free skb.
  */
 static bool
 virtio_transport_recv_pkt_locked(struct virtio_transport *t,
@@ -1902,7 +1913,9 @@ virtio_transport_recv_pkt_locked(struct virtio_transport *t,
 		kfree_skb(skb);
 		break;
 	case TCP_ESTABLISHED:
-		virtio_transport_recv_connected(sk, skb);
+		virtio_transport_recv_connected(sk, skb,
+						ctx->defer_data_ready && ctx->batch ?
+						&ctx->batch->data_ready_pending : NULL);
 		break;
 	case TCP_CLOSING:
 		virtio_transport_recv_disconnecting(sk, skb);
@@ -1971,10 +1984,12 @@ void virtio_transport_rx_batch_finish(struct virtio_transport_rx_batch *batch)
 {
 	struct sock *sk = batch->sk;
 	bool write_space_pending = batch->write_space_pending;
+	bool data_ready_pending = batch->data_ready_pending;
 
 	batch->sk = NULL;
 	batch->net = NULL;
 	batch->write_space_pending = false;
+	batch->data_ready_pending = false;
 
 	if (!sk)
 		return;
@@ -1983,6 +1998,16 @@ void virtio_transport_rx_batch_finish(struct virtio_transport_rx_batch *batch)
 	if (write_space_pending)
 		vsock_sk(sk)->default_write_space(sk);
+
 	release_sock(sk);
+
+	/*
+	 * This event covers data queued while the default callback was installed.
+	 * A sockmap attachment takes the socket lock, so it is ordered after that
+	 * enqueue even if it replaces the callback before this call.
+	 */
+	if (data_ready_pending)
+		vsock_sk(sk)->default_data_ready(sk);
+
 	sock_put(sk);
 }
 EXPORT_SYMBOL_GPL(virtio_transport_rx_batch_finish);
@@ -1995,7 +2017,7 @@ void virtio_transport_recv_pkt_batch(struct virtio_transport *t,
 	struct sockaddr_vm src, dst;
 	struct sock *sk;
 	struct virtio_transport_rx_pkt_ctx ctx;
-	bool batchable, start_batch;
+	bool batchable, defer_data_ready, start_batch;
 	bool free_pkt;
 
 	/* Only STREAM/RW packets can share a socket lock. */
@@ -2015,6 +2037,13 @@ void virtio_transport_recv_pkt_batch(struct virtio_transport *t,
 		    vsock_addr_equals_addr(&batch->dst, &dst) &&
 		    virtio_transport_recv_pkt_batchable(t, batch->sk)) {
 			sk = batch->sk;
+			defer_data_ready = READ_ONCE(sk->sk_data_ready) ==
+					   vsock_sk(sk)->default_data_ready;
+			if (batch->data_ready_pending && !defer_data_ready) {
+				virtio_transport_rx_batch_finish(batch);
+				goto lookup;
+			}
+
 			if (!skb_set_owner_sk_safe(skb, sk)) {
 				WARN_ONCE(1, "receiving vsock socket has sk_refcnt == 0\n");
 				virtio_transport_rx_batch_finish(batch);
@@ -2028,6 +2057,7 @@ void virtio_transport_recv_pkt_batch(struct virtio_transport *t,
 				.dst = &dst,
 				.batchable = &batchable,
 				.batch = batch,
+				.defer_data_ready = defer_data_ready,
 			};
 			free_pkt = virtio_transport_recv_pkt_locked(t, skb, sk, &ctx);
 			if (!batchable)
@@ -2040,6 +2070,7 @@ void virtio_transport_recv_pkt_batch(struct virtio_transport *t,
 		virtio_transport_rx_batch_finish(batch);
 	}
 
+lookup:
 	sk = virtio_transport_recv_pkt_find_socket(skb, &src, &dst, net);
 	if (!sk) {
 		(void)virtio_transport_reset_no_sock(t, skb, net);
@@ -2060,6 +2091,8 @@ void virtio_transport_recv_pkt_batch(struct virtio_transport *t,
 	 */
 	read_lock_bh(&sk->sk_callback_lock);
 	start_batch = virtio_transport_recv_pkt_batchable(t, sk);
+	defer_data_ready = start_batch &&
+		READ_ONCE(sk->sk_data_ready) == vsock_sk(sk)->default_data_ready;
 	read_unlock_bh(&sk->sk_callback_lock);
 
 	if (start_batch)
@@ -2070,10 +2103,11 @@ void virtio_transport_recv_pkt_batch(struct virtio_transport *t,
 		.dst = &dst,
 		.batchable = start_batch ? &batchable : NULL,
 		.batch = start_batch ? batch : NULL,
+		.defer_data_ready = defer_data_ready,
 	};
 	free_pkt = virtio_transport_recv_pkt_locked(t, skb, sk, &ctx);
 	if (start_batch && batchable) {
-		/* Keep the lookup reference until the batch is released. */
+		/* Keep the lookup reference until the batch finishes. */
 		batch->net = net;
 		batch->src = src;
 		batch->dst = dst;
-- 
2.53.0

Re: [PATCH 5/5] vsock: defer RX readable notifications until batch unlock

[sashiko-bot]

Thank you for your contribution! Sashiko AI review found 1 potential issue(s) to consider:
- [Low] vsock: unreachable callback replacement abort logic in batching
--

commit 811453fb266663fdded1db37d3b4cf3bde547c9f
Author: Jia Jia <physicalmtea@gmail.com>

vsock: defer RX readable notifications until batch unlock

This commit defers vsock RX data-ready notifications until the batch finishes
and releases the socket lock. It records the initial callback to handle
mid-batch replacements.

> diff --git a/net/vmw_vsock/virtio_transport_common.c b/net/vmw_vsock/virtio_transport_common.c
> --- a/net/vmw_vsock/virtio_transport_common.c
> +++ b/net/vmw_vsock/virtio_transport_common.c

[ ... ]

> @@ -2015,6 +2040,13 @@ void virtio_transport_recv_pkt_batch(struct virtio_transport *t,
>  		    vsock_addr_equals_addr(&batch->dst, &dst) &&
>  		    virtio_transport_recv_pkt_batchable(t, batch->sk)) {
>  			sk = batch->sk;
> +			defer_data_ready = READ_ONCE(sk->sk_data_ready) ==
> +					   vsock_sk(sk)->default_data_ready;
> +			if (batch->data_ready_pending && !defer_data_ready) {
> +				virtio_transport_rx_batch_finish(batch);
> +				goto lookup;
> +			}
> +

[Severity: Low]
Can this branch ever execute?

The condition checks if a custom callback was installed (!defer_data_ready)
while a default notification is already pending.

Since custom callback attachments (such as BPF sockmap insertion) require
acquiring the exact same socket lock, and the lock is held continuously while
the batch is active, is it possible for the callback to change from default
to custom mid-batch?

Could this bailout branch be unreachable dead code due to the structural mutual
exclusion provided by the socket lock?

-- 
Sashiko AI review · https://sashiko.dev/#/patchset/20261002074551.318789-1-physicalmtea@gmail.com?part=5

Re: [PATCH 5/5] vsock: defer RX readable notifications until batch unlock

[Jia Jia]

>
> Thank you for your contribution! Sashiko AI review found 1 potential issue(s) to consider:
> - [Low] vsock: unreachable callback replacement abort logic in batching
> --
>
> commit 811453fb266663fdded1db37d3b4cf3bde547c9f
> Author: Jia Jia <physicalmtea@gmail.com>
>
> vsock: defer RX readable notifications until batch unlock
>
> This commit defers vsock RX data-ready notifications until the batch finishes
> and releases the socket lock. It records the initial callback to handle
> mid-batch replacements.
>
> > diff --git a/net/vmw_vsock/virtio_transport_common.c b/net/vmw_vsock/virtio_transport_common.c
> > --- a/net/vmw_vsock/virtio_transport_common.c
> > +++ b/net/vmw_vsock/virtio_transport_common.c
>
> [ ... ]
>
> > @@ -2015,6 +2040,13 @@ void virtio_transport_recv_pkt_batch(struct virtio_transport *t,
> >                   vsock_addr_equals_addr(&batch->dst, &dst) &&
> >                   virtio_transport_recv_pkt_batchable(t, batch->sk)) {
> >                       sk = batch->sk;
> > +                     defer_data_ready = READ_ONCE(sk->sk_data_ready) ==
> > +                                        vsock_sk(sk)->default_data_ready;
> > +                     if (batch->data_ready_pending && !defer_data_ready) {
> > +                             virtio_transport_rx_batch_finish(batch);
> > +                             goto lookup;
> > +                     }
> > +
>
> [Severity: Low]
> Can this branch ever execute?
>

Unless I'm mistaken, this branch can actually be reached.

> The condition checks if a custom callback was installed (!defer_data_ready)
> while a default notification is already pending.
>
> Since custom callback attachments (such as BPF sockmap insertion) require
> acquiring the exact same socket lock, and the lock is held continuously while
> the batch is active, is it possible for the callback to change from default
> to custom mid-batch?
>
> Could this bailout branch be unreachable dead code due to the structural mutual
> exclusion provided by the socket lock?
>

The premise that sockmap updates always take lock_sock(sk) applies only to
the syscall path. When a BPF program calls bpf_map_update_elem(), for example
from SK_LOOKUP or BPF_TRACE_ITER, sock_map_update_elem() takes
bh_lock_sock(sk),

The batch reuse path checks sk->sk_prot in
virtio_transport_recv_pkt_batchable() before reading sk->sk_data_ready. A
concurrent BPF update can replace both callbacks between those reads while
the RX worker owns the socket.

> --
> Sashiko AI review · https://sashiko.dev/#/patchset/20261002074551.318789-1-physicalmtea@gmail.com?part=5