[PATCH net-next v4 0/4] vsock/virtio: reduce RX per-packet socket overhead

[PATCH net-next v4 0/4] vsock/virtio: reduce RX per-packet socket overhead

[Jia Jia]

This series reduces per-packet socket overhead in the virtio-vsock RX
worker.

Patch 1: separate socket lookup from locked packet processing.

Patch 2: keep one socket lock across a bounded run of established
STREAM/RW packets for the same native socket, and reuse the locked socket
for later same-flow packets.

Patch 3: coalesce default write-space notifications while that
lock is held.

Patch 4: defer the default readable callback until after the batch
unlocks.

Performance:

Tested on a single-stream vsock connection (one Guest receiver thread on
CPU0, virtio-vsock IRQ on Guest CPU1, fresh QEMU Guest per state, 16
balanced AB/BA pairs per payload). The host uses vhost-vsock.

For throughput, vsock_perf runs on the host as the
sender, with a custom single-threaded Guest receiver pinned to Guest CPU0
using blocking recv(). The fixed-64K table uses a 64K buffer; the matched
table uses the sender's request size. The virtio-vsock IRQ/RX worker is
pinned to Guest CPU1. To normalize run times across payloads into a
bounded, deterministic window and avoid long runs being skewed by host
background noise or virtualization scheduling jitter, transfer sizes scale
proportionally with the request size (64B/64M, 256B/256M, etc.), capped at
2G for 4K–64K. The custom receiver keeps the recv() size and count
explicit; vsock_perf's receiver uses poll()+read().

Latency uses a custom user-space vsock ping-pong tool.

The measurements below were collected for the RX batching series.

This benchmark does not run vsock_perf on both ends. Instead, it uses
vsock_perf as the sender on the host (vhost-vsock) side and a custom
blocking recv() program without poll() on the Guest side, with the receiver
thread pinned to Guest CPU0 (and the virtio-vsock RX worker pinned to Guest
CPU1). Plain blocking recv() combined with strict core pinning gives
audit-level determinism for tracing a single-threaded TID.

I think it is necessary to do further testing on the large I/O throughput
peak observed at 4K to find the root cause (and rule out anything
introduced by the measurement method). I used kprobe and temporary
trace_printk in the source for mechanism diagnostics. The diagnostic
data shows that lock wait time and contention at 4K both decrease strictly
monotonically as payload size increases;
wire-level tracing also confirms that 4K is split into two packets, and
that all sizes can reach the 64K batching limit.

The nature of the prominent 4K throughput gain most likely lies in it
sitting exactly at the end-to-end pipeline sweet spot between host and
guest:

Compared with 1K and smaller payloads: 4K halves the packet intensity, so
upstream host-side syscall and interrupt injection overhead is no longer
the bottleneck across the whole path, allowing the host to supply data
adequately.

Compared with larger 8K to 64K payloads: 4K keeps the payload size
moderate, and combined with the observed syscall reduction and cycle
improvement, the batch-draining benefit from notification coalescing is
fully realized.

As payload size increases further from 8K to 64K, the overall gain shrinks
smoothly, consistent with Amdahl's law, because system execution time
shifts toward page copying and VQ processing.

With all patches applied:

End-to-end RX throughput (64K Guest recv() size, SO_RCVLOWAT=1; Gbit/s):

  payload  baseline RX  PATCH RX  throughput change (95% CI)
           (Gbit/s)      (Gbit/s)
  64B      0.1065       0.2503    +135.00% [+130.35%, +139.74%]
  256B     0.5722       1.0286     +79.76% [+75.40%, +84.23%]
  512B     1.1464       1.9779     +72.53% [+69.50%, +75.63%]
  1K       2.2714       3.6368     +60.12% [+56.92%, +63.37%]
  4K       3.9095       7.7160     +97.37% [+92.09%, +102.79%]
  8K       3.3436       5.3111     +58.84% [+53.49%, +64.38%]
  16K      3.8876       5.7456     +47.80% [+45.45%, +50.18%]
  64K      4.0056       5.3590     +33.79% [+29.37%, +38.36%]

Fixed-syscall stress test (receiver recv() size = sender write size;
Gbit/s):

  payload baseline PATCH    throughput change (95% CI)    recv()
          RX       RX                                     (base/patch)
          (Gbit/s) (Gbit/s)
  64B     0.1259   0.2651   +110.52% [+99.08%, +122.62%]  1048577/1048577
  256B    0.3790   0.9444   +149.20% [+134.70%, +164.60%] 1048577/1048577
  512B    0.7768   1.7009   +118.97% [+103.94%, +135.11%] 1048577/1048577
  1K      1.4886   2.9477   +98.02% [+84.82%, +112.17%]   1048577/1048577
  4K      2.1129   4.7331   +124.01% [+115.40%, +132.96%] 537436/525524
  8K      2.8698   5.3683   +87.06% [+80.96%, +93.37%]    276448/264109
  16K     3.5156   5.8463   +66.30% [+63.10%, +69.55%]    147849/132267
  64K     4.2459   5.8576   +37.96% [+35.26%, +40.71%]    48447/33144

Request-response latency (Ping-Pong RTT, 10,000 requests):

  payload  baseline mean RTT (us)  PATCH mean RTT (us)  change (95% CI)
  64B      221.968                  218.144              -1.72%
                                                        [-4.23%, +0.85%]
  256B     126.968                  124.541              -1.91%
                                                        [-3.35%, -0.45%]
  4K       130.362                  126.664              -2.84%
                                                        [-5.12%, -0.50%]

Diagnostic Guest CPU1 system-wide cycles (fixed-64K receiver, with perf):

  Buffer size  baseline cycles/B  PATCH cycles/B  cycles change (95% CI)
  64B      62.032            52.370          -15.58% [-17.16%, -13.96%]
  256B     15.287            13.630          -10.84% [-12.51%, -9.14%]
  512B      7.306             6.611           -9.51% [-10.92%, -8.08%]
  1K        3.811             3.458           -9.25% [-10.96%, -7.50%]
  4K        1.850             1.669           -9.75% [-11.78%, -7.67%]
  8K        1.504             1.363           -9.36% [-12.97%, -5.60%]
  16K       1.321             1.210           -8.37% [-10.61%, -6.07%]
  64K       1.342             1.227           -8.57% [-10.87%, -6.21%]

Patch-scope measurements and attribution:

The series was tested cumulatively in two ranges, using the same
single-stream setup and paired AB/BA runs. The table below shows results
from patches 1–2 only, before the notification changes in patches 3–4.

  payload  baseline RX  PATCH RX  throughput change (95% CI)
           (Gbit/s)      (Gbit/s)
  64B      0.1211       0.2534    +109.26% [+106.34%, +112.23%]
  256B     0.5671       0.9989     +76.15% [+70.63%, +81.85%]
  512B     1.1334       1.9000     +67.64% [+64.64%, +70.69%]
  1K       2.2650       3.5610     +57.22% [+50.91%, +63.79%]
  4K       3.8331       7.2888     +90.15% [+83.46%, +97.09%]
  8K       4.8244       7.6636     +58.85% [+53.71%, +64.16%]
  16K      5.4979       8.2469     +50.00% [+45.57%, +54.57%]
  64K      3.8875       4.7588     +22.41% [+17.77%, +27.24%]

Fixed-syscall stress test (receiver recv() size = sender write size;
no perf; Gbit/s):

  payload baseline PATCH    throughput change (95% CI)    recv()
          RX       RX                                     (base/patch)
          (Gbit/s) (Gbit/s)
  64B     0.1397   0.2646   +89.40% [+80.92%, +98.27%]    1048577/1048577
  256B    0.3960   0.9516   +140.34% [+117.48%, +165.60%] 1048577/1048577
  512B    0.7541   1.6496   +118.74% [+101.42%, +137.55%] 1048577/1048577
  1K      1.4903   2.8793   +93.21% [+77.06%, +110.82%]   1048577/1048577
  4K      2.9466   6.2284   +111.37% [+97.52%, +126.20%]  538192/525220
  8K      4.0043   6.9249   +72.94% [+60.18%, +86.71%]    276313/263852
  16K     4.8981   7.5599   +54.34% [+49.77%, +59.06%]    147576/131600
  64K     5.9294   7.7897   +31.37% [+27.18%, +35.70%]    50207/33014

Ping-Pong RTT (10,000 requests; no perf):

  payload  baseline mean RTT (us)  PATCH mean RTT (us)  change (95% CI)
  64B      119.688                  119.739              +0.04%
                                                        [-2.61%, +2.77%]
  256B     120.603                  124.485              +3.22%
                                                        [-0.44%, +7.01%]
  4K       121.921                  122.491              +0.47%
                                                        [-3.51%, +4.61%]

Guest CPU1 system-wide aggregate cycles (fixed-64K receiver, with perf;
cycles/byte):

  Buffer size  baseline cycles/B  PATCH cycles/B  cycles change (95% CI)
  64B      66.838             60.012           -10.21% [-13.03%, -7.30%]
  256B     13.446             12.305            -8.49% [-11.23%, -5.66%]
  512B      6.833              6.317            -7.54% [ -9.52%, -5.52%]
  1K        3.558              3.434            -3.48% [ -6.02%, -0.87%]
  4K        1.806              1.686            -6.64% [ -8.69%, -4.54%]
  8K        1.457              1.403            -3.73% [ -5.64%, -1.78%]
  16K       1.295              1.279            -1.24% [ -5.21%, +2.89%]
  64K       1.214              1.198            -1.30% [ -4.06%, +1.53%]

The split-patch tests show that PATCH 2 mainly contributes to the
improvement in I/O Throughput, while PATCH 3 and PATCH 4 mainly contribute
to the improvements in cycles and RTT latency (although PATCH 2 shows no
obvious regression here).

---
v4:
  - use only the connected socket lookup for batch admission and return
    misses to the normal receive path
  - read packet-specific addresses from the skb in the locked receive
    helper and use the matched socket namespace for reset replies
  - remove the extra address and network namespace arguments from the
    locked receive helper
v3:
  - restrict RX batching to sockets using the original protocol and
    data-ready/write-space callbacks
  - exclude sockets carrying the psock marker published before sockmap
    protocol/callback replacement, and recheck it at each packet boundary
  - preserve existing sockmap update return values; an attach racing a
    native batch makes the next packet end the batch without queued-skb
    handoff
  - remove a leftover blank line after an opening brace
  - remove callback handoff work and attach/detach kicks, preserving the
    existing per-packet behavior for sockmap and other non-native sockets
  - avoid making this series drain queued or partially consumed skbs when a
    socket is added to or removed from a sockmap
  - clarify that Patch 1 prepares for batching and intends no functional
    change
  - inline address decoding and receive tracing, simplify the socket lookup
    helper, and pass locked receive arguments directly
  - fold same-flow lookup reuse into Patch 2 and match packets against the
    locked socket state, reducing the series from five patches to four
v2:
  - stop a batch after it consumes the last skb metadata slot, sharing the
    admission predicate with virtio_transport_inc_rx_pkt()
  - move replacement-callback handoff into vsock, process queued skbs one
    at a time, and trigger the handoff from vsock_bpf_update_proto() on
    attach and detach
  - reduce the packet-context diff in the locked receive helper
  - keep RX batch counters in the batch state and reset them in the finish
    path
  - use the vsock/virtio subject prefix and add the requested measurements
v1: https://lore.kernel.org/virtualization/20261002074551.318789-1-physicalmtea@gmail.com/

Jia Jia (4):
  vsock/virtio: split socket lookup from locked RX processing
  vsock/virtio: amortize RX socket locking for stream packets
  vsock/virtio: coalesce RX write-space notifications in lock batches
  vsock/virtio: defer RX notifications until batch unlock

 include/linux/virtio_vsock.h            |  13 ++
 include/net/af_vsock.h                  |   6 +
 net/vmw_vsock/af_vsock.c                |   2 +
 net/vmw_vsock/virtio_transport.c        |  29 ++-
 net/vmw_vsock/virtio_transport_common.c | 353 ++++++++++++++++++++++++++------
 5 files changed, 341 insertions(+), 62 deletions(-)

base-commit: 8df0638138d3e0344fd1fb36cf2d1ca1cf5028f0
-- 
2.53.0

[PATCH net-next v4 1/4] vsock/virtio: split socket lookup from locked RX processing

[Jia Jia]

Split virtio_transport_recv_pkt() so socket lookup is separate from
receive state machine processing under the socket lock. This prepares the
receive path to keep a looked-up socket across multiple packets in a later
change.

Keep virtio_transport_recv_pkt() on its existing per-packet locking path.
The locked helper obtains packet-specific addresses from the skb and uses
the matched socket namespace for reset replies. Preserve source validation
and failure handling.

No functional change intended.

Signed-off-by: Jia Jia <physicalmtea@gmail.com>
---
 net/vmw_vsock/virtio_transport_common.c | 128 ++++++++++++++----------
 1 file changed, 75 insertions(+), 53 deletions(-)

diff --git a/net/vmw_vsock/virtio_transport_common.c b/net/vmw_vsock/virtio_transport_common.c
index f225f53ed4bab..61ee3bc205d5d 100644
--- a/net/vmw_vsock/virtio_transport_common.c
+++ b/net/vmw_vsock/virtio_transport_common.c
@@ -1774,81 +1774,60 @@ static bool virtio_transport_valid_type(u16 type)
 	       (type == VIRTIO_VSOCK_TYPE_SEQPACKET);
 }
 
-/* We are under the virtio-vsock's vsock->rx_lock or vhost-vsock's vq->mutex
- * lock.
- */
-void virtio_transport_recv_pkt(struct virtio_transport *t,
-			       struct sk_buff *skb, struct net *net)
+static struct sock *
+virtio_transport_find_socket(struct sk_buff *skb, struct sockaddr_vm *src,
+			     struct sockaddr_vm *dst, struct net *net)
 {
 	struct virtio_vsock_hdr *hdr = virtio_vsock_hdr(skb);
-	struct sockaddr_vm src, dst;
-	struct vsock_sock *vsk;
 	struct sock *sk;
-	bool space_available;
-
-	vsock_addr_init(&src, le64_to_cpu(hdr->src_cid),
-			le32_to_cpu(hdr->src_port));
-	vsock_addr_init(&dst, le64_to_cpu(hdr->dst_cid),
-			le32_to_cpu(hdr->dst_port));
-
-	trace_virtio_transport_recv_pkt(src.svm_cid, src.svm_port,
-					dst.svm_cid, dst.svm_port,
-					le32_to_cpu(hdr->len),
-					le16_to_cpu(hdr->type),
-					le16_to_cpu(hdr->op),
-					le32_to_cpu(hdr->flags),
-					le32_to_cpu(hdr->buf_alloc),
-					le32_to_cpu(hdr->fwd_cnt));
 
-	if (!virtio_transport_valid_type(le16_to_cpu(hdr->type))) {
-		(void)virtio_transport_reset_no_sock(t, skb, net);
-		goto free_pkt;
-	}
+	if (!virtio_transport_valid_type(le16_to_cpu(hdr->type)))
+		return NULL;
 
-	/* The socket must be in connected or bound table
-	 * otherwise send reset back
-	 */
-	sk = vsock_find_connected_socket_net(&src, &dst, net);
-	if (!sk) {
-		sk = vsock_find_bound_socket_net(&dst, net);
-		if (!sk) {
-			(void)virtio_transport_reset_no_sock(t, skb, net);
-			goto free_pkt;
-		}
-	}
+	sk = vsock_find_connected_socket_net(src, dst, net);
+	if (!sk)
+		sk = vsock_find_bound_socket_net(dst, net);
+	if (!sk)
+		return NULL;
 
 	if (virtio_transport_get_type(sk) != le16_to_cpu(hdr->type)) {
-		(void)virtio_transport_reset_no_sock(t, skb, net);
 		sock_put(sk);
-		goto free_pkt;
+		return NULL;
 	}
 
-	if (!skb_set_owner_sk_safe(skb, sk)) {
-		WARN_ONCE(1, "receiving vsock socket has sk_refcnt == 0\n");
-		goto free_pkt;
-	}
+	return sk;
+}
 
-	vsk = vsock_sk(sk);
+/*
+ * The caller holds sk's socket lock and must free skb if this returns true.
+ */
+static bool
+virtio_transport_recv_pkt_locked(struct virtio_transport *t,
+				 struct sk_buff *skb, struct sock *sk)
+{
+	struct virtio_vsock_hdr *hdr = virtio_vsock_hdr(skb);
+	struct sockaddr_vm src;
+	struct vsock_sock *vsk = vsock_sk(sk);
+	bool space_available;
 
-	lock_sock(sk);
+	vsock_addr_init(&src, le64_to_cpu(hdr->src_cid),
+			le32_to_cpu(hdr->src_port));
 
-	/* Check if sk has been closed or assigned to another transport before
-	 * lock_sock (note: listener sockets are not assigned to any transport)
+	/* Listener sockets accept packets from any source and are not assigned to
+	 * a transport.
 	 */
 	if (sock_flag(sk, SOCK_DONE) ||
 	    (sk->sk_state != TCP_LISTEN &&
 	     !vsock_check_source(vsk, &t->transport, &src))) {
-		(void)virtio_transport_reset_no_sock(t, skb, net);
-		release_sock(sk);
-		sock_put(sk);
-		goto free_pkt;
+		(void)virtio_transport_reset_no_sock(t, skb, sock_net(sk));
+		return true;
 	}
 
 	space_available = virtio_transport_space_update(sk, skb);
 
 	/* Update CID in case it has changed after a transport reset event */
 	if (vsk->local_addr.svm_cid != VMADDR_CID_ANY)
-		vsk->local_addr.svm_cid = dst.svm_cid;
+		vsk->local_addr.svm_cid = le64_to_cpu(hdr->dst_cid);
 
 	if (space_available)
 		sk->sk_write_space(sk);
@@ -1870,17 +1849,60 @@ void virtio_transport_recv_pkt(struct virtio_transport *t,
 		kfree_skb(skb);
 		break;
 	default:
-		(void)virtio_transport_reset_no_sock(t, skb, net);
+		(void)virtio_transport_reset_no_sock(t, skb, sock_net(sk));
 		kfree_skb(skb);
 		break;
 	}
 
+	return false;
+}
+
+/* We are under the virtio-vsock's vsock->rx_lock or vhost-vsock's vq->mutex
+ * lock.
+ */
+void virtio_transport_recv_pkt(struct virtio_transport *t,
+			       struct sk_buff *skb, struct net *net)
+{
+	struct virtio_vsock_hdr *hdr = virtio_vsock_hdr(skb);
+	struct sockaddr_vm src, dst;
+	struct sock *sk;
+	bool free_pkt;
+
+	vsock_addr_init(&src, le64_to_cpu(hdr->src_cid),
+			le32_to_cpu(hdr->src_port));
+	vsock_addr_init(&dst, le64_to_cpu(hdr->dst_cid),
+			le32_to_cpu(hdr->dst_port));
+
+	trace_virtio_transport_recv_pkt(src.svm_cid, src.svm_port,
+					dst.svm_cid, dst.svm_port,
+					le32_to_cpu(hdr->len),
+					le16_to_cpu(hdr->type),
+					le16_to_cpu(hdr->op),
+					le32_to_cpu(hdr->flags),
+					le32_to_cpu(hdr->buf_alloc),
+					le32_to_cpu(hdr->fwd_cnt));
+
+	sk = virtio_transport_find_socket(skb, &src, &dst, net);
+	if (!sk) {
+		(void)virtio_transport_reset_no_sock(t, skb, net);
+		goto free_pkt;
+	}
+
+	if (!skb_set_owner_sk_safe(skb, sk)) {
+		WARN_ONCE(1, "receiving vsock socket has sk_refcnt == 0\n");
+		goto free_pkt;
+	}
+
+	lock_sock(sk);
+	free_pkt = virtio_transport_recv_pkt_locked(t, skb, sk);
 	release_sock(sk);
 
 	/* Release refcnt obtained when we fetched this socket out of the
 	 * bound or connected list.
 	 */
 	sock_put(sk);
+	if (free_pkt)
+		kfree_skb(skb);
 	return;
 
 free_pkt:
-- 
2.53.0

[PATCH net-next v4 2/4] vsock/virtio: amortize RX socket locking for stream packets

[Jia Jia]

The virtio-vsock RX worker currently looks up the socket and takes its
lock for every packet. Keep the socket lock and lookup reference across a
bounded run of STREAM/RW packets for the same established socket.

Reuse the locked socket while packets continue to match the same flow.
End the batch for a different flow, non-RW traffic, malformed packets, or
a change in socket state, transport, protocol, or receive-buffer headroom.
A connected-table miss uses the existing receive path.

The reused path still traces every packet, takes an skb owner reference,
validates the source and transport, updates credit, and runs the existing
receive state machine.

Sockmap can replace the protocol without taking the socket lock. Check the
psock marker at each packet boundary and serialize initial admission with
sk_callback_lock, so a concurrent attach ends batching at the next packet.

Cap a batch at 64 packets and 64K of payload. The packet limit bounds
per-packet work, while the byte limit bounds how much data can queue before
a reader acquires the socket lock. Batching remains in the process-context
RX worker.

Signed-off-by: Jia Jia <physicalmtea@gmail.com>
---
 include/linux/virtio_vsock.h            |  11 ++
 net/vmw_vsock/virtio_transport.c        |  29 +++-
 net/vmw_vsock/virtio_transport_common.c | 183 +++++++++++++++++++++++-
 3 files changed, 216 insertions(+), 7 deletions(-)

diff --git a/include/linux/virtio_vsock.h b/include/linux/virtio_vsock.h
index f91704731057e..26dde6909c4ed 100644
--- a/include/linux/virtio_vsock.h
+++ b/include/linux/virtio_vsock.h
@@ -282,6 +282,17 @@ void virtio_transport_destruct(struct vsock_sock *vsk);
 
 void virtio_transport_recv_pkt(struct virtio_transport *t,
 			       struct sk_buff *skb, struct net *net);
+
+struct virtio_transport_rx_batch {
+	struct sock *sk;
+	unsigned int pkts;
+	size_t bytes;
+};
+
+void virtio_transport_recv_pkt_batch(struct virtio_transport *t,
+				     struct sk_buff *skb, struct net *net,
+				     struct virtio_transport_rx_batch *batch);
+void virtio_transport_rx_batch_finish(struct virtio_transport_rx_batch *batch);
 void virtio_transport_inc_tx_pkt(struct virtio_vsock_sock *vvs, struct sk_buff *skb);
 u32 virtio_transport_get_credit(struct virtio_vsock_sock *vvs, u32 wanted);
 void virtio_transport_put_credit(struct virtio_vsock_sock *vvs, u32 credit);
diff --git a/net/vmw_vsock/virtio_transport.c b/net/vmw_vsock/virtio_transport.c
index 4f9aa9c4c3aa5..ef076732158d4 100644
--- a/net/vmw_vsock/virtio_transport.c
+++ b/net/vmw_vsock/virtio_transport.c
@@ -629,8 +629,16 @@ virtio_transport_seqpacket_allow(struct vsock_sock *vsk, u32 remote_cid)
 	return seqpacket_allow;
 }
 
+/*
+ * Keep a bounded run of packets for one socket under a single socket lock.
+ * Limit packet count and payload size to bound the work done while locked.
+ */
+#define VIRTIO_TRANSPORT_RX_BATCH_MAX_PKTS	64
+#define VIRTIO_TRANSPORT_RX_BATCH_MAX_BYTES	(64 * 1024)
+
 static void virtio_transport_rx_work(struct work_struct *work)
 {
+	struct virtio_transport_rx_batch batch = {};
 	struct virtio_vsock *vsock =
 		container_of(work, struct virtio_vsock, rx_work);
 	struct virtqueue *vq;
@@ -666,6 +674,7 @@ static void virtio_transport_rx_work(struct work_struct *work)
 			/* Drop short/long packets */
 			if (unlikely(len < sizeof(*hdr) ||
 				     len > virtio_vsock_skb_len(skb))) {
+				virtio_transport_rx_batch_finish(&batch);
 				kfree_skb(skb);
 				continue;
 			}
@@ -673,6 +682,7 @@ static void virtio_transport_rx_work(struct work_struct *work)
 			hdr = virtio_vsock_hdr(skb);
 			payload_len = le32_to_cpu(hdr->len);
 			if (unlikely(payload_len > len - sizeof(*hdr))) {
+				virtio_transport_rx_batch_finish(&batch);
 				kfree_skb(skb);
 				continue;
 			}
@@ -680,16 +690,33 @@ static void virtio_transport_rx_work(struct work_struct *work)
 			if (payload_len)
 				virtio_vsock_skb_put(skb, payload_len);
 
+			if (batch.sk &&
+			    payload_len > VIRTIO_TRANSPORT_RX_BATCH_MAX_BYTES -
+					  batch.bytes) {
+				virtio_transport_rx_batch_finish(&batch);
+			}
+
 			virtio_transport_deliver_tap_pkt(skb);
 
 			/* Force virtio-transport into global mode since it
 			 * does not yet support local-mode namespacing.
 			 */
-			virtio_transport_recv_pkt(&virtio_transport, skb, NULL);
+			virtio_transport_recv_pkt_batch(&virtio_transport, skb, NULL,
+							&batch);
+
+			if (batch.sk) {
+				batch.pkts++;
+				batch.bytes += payload_len;
+				if (batch.pkts >= VIRTIO_TRANSPORT_RX_BATCH_MAX_PKTS ||
+				    batch.bytes >= VIRTIO_TRANSPORT_RX_BATCH_MAX_BYTES) {
+					virtio_transport_rx_batch_finish(&batch);
+				}
+			}
 		}
 	} while (!virtqueue_enable_cb(vq));
 
 out:
+	virtio_transport_rx_batch_finish(&batch);
 	if (vsock->rx_buf_nr < vsock->rx_buf_max_nr / 2)
 		virtio_vsock_rx_fill(vsock);
 out_nofill:
diff --git a/net/vmw_vsock/virtio_transport_common.c b/net/vmw_vsock/virtio_transport_common.c
index 61ee3bc205d5d..ff74e805527a5 100644
--- a/net/vmw_vsock/virtio_transport_common.c
+++ b/net/vmw_vsock/virtio_transport_common.c
@@ -576,11 +576,16 @@ virtio_transport_collapse_rx_queue(struct virtio_vsock_sock *vvs,
 	skb_queue_splice(&new_queue, &vvs->rx_queue);
 }
 
+static bool
+virtio_transport_rx_skb_has_headroom(struct virtio_vsock_sock *vvs)
+{
+	return (u64)(skb_queue_len(&vvs->rx_queue) + 1) * SKB_TRUESIZE(0) <=
+		vvs->buf_alloc;
+}
+
 static bool virtio_transport_inc_rx_pkt(struct virtio_vsock_sock *vvs,
 					u32 len)
 {
-	u64 skb_overhead = (skb_queue_len(&vvs->rx_queue) + 1) * SKB_TRUESIZE(0);
-
 	/* Allow at most buf_alloc * 2 total budget (payload + overhead),
 	 * similar to how SO_RCVBUF is doubled to reserve space for sk_buff
 	 * metadata. Check payload against buf_alloc to be sure the other
@@ -588,7 +593,7 @@ static bool virtio_transport_inc_rx_pkt(struct virtio_vsock_sock *vvs,
 	 * queue growth.
 	 */
 	if ((u64)vvs->buf_used + len > vvs->buf_alloc ||
-	    skb_overhead > vvs->buf_alloc)
+	    !virtio_transport_rx_skb_has_headroom(vvs))
 		return false;
 
 	vvs->rx_bytes += len;
@@ -1798,12 +1803,38 @@ virtio_transport_find_socket(struct sk_buff *skb, struct sockaddr_vm *src,
 	return sk;
 }
 
+static bool
+virtio_transport_sock_has_psock(const struct sock *sk)
+{
+	return (uintptr_t)rcu_access_pointer(__sk_user_data(sk)) &
+	       SK_USER_DATA_PSOCK;
+}
+
+static bool
+virtio_transport_recv_pkt_batchable(struct virtio_transport *t,
+				    struct sock *sk)
+{
+	struct vsock_sock *vsk = vsock_sk(sk);
+	struct virtio_vsock_sock *vvs = vsk->trans;
+
+	return sk->sk_state == TCP_ESTABLISHED &&
+	       sk->sk_type == SOCK_STREAM &&
+	       READ_ONCE(sk->sk_prot) == sk->sk_prot_creator &&
+	       !sock_flag(sk, SOCK_DONE) &&
+	       vsk->transport == &t->transport &&
+	       vvs && virtio_transport_rx_skb_has_headroom(vvs) &&
+	       !virtio_transport_sock_has_psock(sk);
+}
+
 /*
- * The caller holds sk's socket lock and must free skb if this returns true.
+ * 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.
  */
 static bool
 virtio_transport_recv_pkt_locked(struct virtio_transport *t,
-				 struct sk_buff *skb, struct sock *sk)
+				 struct sk_buff *skb, struct sock *sk,
+				 bool *batchable)
 {
 	struct virtio_vsock_hdr *hdr = virtio_vsock_hdr(skb);
 	struct sockaddr_vm src;
@@ -1813,6 +1844,9 @@ virtio_transport_recv_pkt_locked(struct virtio_transport *t,
 	vsock_addr_init(&src, le64_to_cpu(hdr->src_cid),
 			le32_to_cpu(hdr->src_port));
 
+	if (batchable)
+		*batchable = false;
+
 	/* Listener sockets accept packets from any source and are not assigned to
 	 * a transport.
 	 */
@@ -1854,6 +1888,9 @@ virtio_transport_recv_pkt_locked(struct virtio_transport *t,
 		break;
 	}
 
+	if (batchable)
+		*batchable = virtio_transport_recv_pkt_batchable(t, sk);
+
 	return false;
 }
 
@@ -1894,7 +1931,7 @@ void virtio_transport_recv_pkt(struct virtio_transport *t,
 	}
 
 	lock_sock(sk);
-	free_pkt = virtio_transport_recv_pkt_locked(t, skb, sk);
+	free_pkt = virtio_transport_recv_pkt_locked(t, skb, sk, NULL);
 	release_sock(sk);
 
 	/* Release refcnt obtained when we fetched this socket out of the
@@ -1910,6 +1947,140 @@ void virtio_transport_recv_pkt(struct virtio_transport *t,
 }
 EXPORT_SYMBOL_GPL(virtio_transport_recv_pkt);
 
+/*
+ * Finish the RX batch.
+ * For a non-empty batch, the caller must hold the socket lock acquired with
+ * lock_sock(). This function releases the lock and the batch's lookup
+ * reference.
+ * An empty batch is a no-op.
+ */
+void virtio_transport_rx_batch_finish(struct virtio_transport_rx_batch *batch)
+{
+	struct sock *sk = batch->sk;
+
+	batch->sk = NULL;
+	batch->pkts = 0;
+	batch->bytes = 0;
+
+	if (!sk)
+		return;
+
+	release_sock(sk);
+	sock_put(sk);
+}
+EXPORT_SYMBOL_GPL(virtio_transport_rx_batch_finish);
+
+void virtio_transport_recv_pkt_batch(struct virtio_transport *t,
+				     struct sk_buff *skb, struct net *net,
+				     struct virtio_transport_rx_batch *batch)
+{
+	struct virtio_vsock_hdr *hdr = virtio_vsock_hdr(skb);
+	struct sockaddr_vm src, dst;
+	struct vsock_sock *vsk;
+	bool batchable, start_batch;
+	struct sock *sk;
+	bool free_pkt;
+
+	/* Only STREAM/RW packets can share a socket lock. */
+	if (le16_to_cpu(hdr->type) != VIRTIO_VSOCK_TYPE_STREAM ||
+	    le16_to_cpu(hdr->op) != VIRTIO_VSOCK_OP_RW) {
+		virtio_transport_rx_batch_finish(batch);
+		virtio_transport_recv_pkt(t, skb, net);
+		return;
+	}
+
+	vsock_addr_init(&src, le64_to_cpu(hdr->src_cid),
+			le32_to_cpu(hdr->src_port));
+	vsock_addr_init(&dst, le64_to_cpu(hdr->dst_cid),
+			le32_to_cpu(hdr->dst_port));
+
+	if (batch->sk) {
+		vsk = vsock_sk(batch->sk);
+		if (vsock_addr_equals_addr(&src, &vsk->remote_addr) &&
+		    dst.svm_port == vsk->local_addr.svm_port &&
+		    vsock_net_check_mode(sock_net(batch->sk), net) &&
+		    virtio_transport_recv_pkt_batchable(t, batch->sk)) {
+			sk = batch->sk;
+			trace_virtio_transport_recv_pkt(src.svm_cid,
+							src.svm_port,
+							dst.svm_cid,
+							dst.svm_port,
+							le32_to_cpu(hdr->len),
+							le16_to_cpu(hdr->type),
+							le16_to_cpu(hdr->op),
+							le32_to_cpu(hdr->flags),
+							le32_to_cpu(hdr->buf_alloc),
+							le32_to_cpu(hdr->fwd_cnt));
+			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);
+				kfree_skb(skb);
+				return;
+			}
+
+			free_pkt = virtio_transport_recv_pkt_locked(t, skb, sk,
+								    &batchable);
+			if (!batchable)
+				virtio_transport_rx_batch_finish(batch);
+			if (free_pkt)
+				kfree_skb(skb);
+			return;
+		}
+
+		virtio_transport_rx_batch_finish(batch);
+	}
+
+	sk = vsock_find_connected_socket_net(&src, &dst, net);
+	if (!sk) {
+		virtio_transport_recv_pkt(t, skb, net);
+		return;
+	}
+
+	if (virtio_transport_get_type(sk) != le16_to_cpu(hdr->type)) {
+		sock_put(sk);
+		virtio_transport_recv_pkt(t, skb, net);
+		return;
+	}
+
+	trace_virtio_transport_recv_pkt(src.svm_cid, src.svm_port,
+					dst.svm_cid, dst.svm_port,
+					le32_to_cpu(hdr->len),
+					le16_to_cpu(hdr->type),
+					le16_to_cpu(hdr->op),
+					le32_to_cpu(hdr->flags),
+					le32_to_cpu(hdr->buf_alloc),
+					le32_to_cpu(hdr->fwd_cnt));
+
+	if (!skb_set_owner_sk_safe(skb, sk)) {
+		WARN_ONCE(1, "receiving vsock socket has sk_refcnt == 0\n");
+		kfree_skb(skb);
+		return;
+	}
+
+	lock_sock(sk);
+	/*
+	 * Sockmap removal restores the native protocol under sk_callback_lock.
+	 * Serialize this initial eligibility check with that update.
+	 */
+	read_lock_bh(&sk->sk_callback_lock);
+	start_batch = virtio_transport_recv_pkt_batchable(t, sk);
+	read_unlock_bh(&sk->sk_callback_lock);
+
+	free_pkt = virtio_transport_recv_pkt_locked(t, skb, sk,
+						    start_batch ? &batchable : NULL);
+	if (start_batch && batchable) {
+		/* Keep the lookup reference until the batch is released. */
+		batch->sk = sk;
+		return;
+	}
+
+	release_sock(sk);
+	sock_put(sk);
+	if (free_pkt)
+		kfree_skb(skb);
+}
+EXPORT_SYMBOL_GPL(virtio_transport_recv_pkt_batch);
+
 /* Remove skbs found in a queue that have a vsk that matches.
  *
  * Each skb is freed.
-- 
2.53.0

[PATCH net-next v4 3/4] vsock/virtio: coalesce RX write-space notifications in lock batches

[Jia Jia]

Each received packet updates peer credit and may call sk_write_space().
A writer cannot use that credit until the batch releases the socket lock,
so one notification at batch finish is sufficient.

Save the initial sk_write_space callback when the AF_VSOCK socket is
created and require it to remain installed for batching. This lets the
transport module deliver the deferred notification without referring to
the unexported default callback. Sockets with replaced callbacks retain
their per-packet behavior.

Record the batch socket before processing its first packet so a batch that
ends on that packet cannot lose the pending notification.

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

diff --git a/include/linux/virtio_vsock.h b/include/linux/virtio_vsock.h
index 26dde6909c4ed..bf077ee0ded88 100644
--- a/include/linux/virtio_vsock.h
+++ b/include/linux/virtio_vsock.h
@@ -287,6 +287,7 @@ struct virtio_transport_rx_batch {
 	struct sock *sk;
 	unsigned int pkts;
 	size_t bytes;
+	bool write_space_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 5549298c1ec6b..9a40960d10cbe 100644
--- a/include/net/af_vsock.h
+++ b/include/net/af_vsock.h
@@ -63,6 +63,11 @@ struct vsock_sock {
 	u32 peer_shutdown;
 	bool sent_request;
 	bool ignore_connecting_rst;
+	/*
+	 * Initial callback, used to identify replacements and deliver deferred
+	 * write-space notifications.
+	 */
+	void (*default_write_space)(struct sock *sk);
 
 	/* Protected by lock_sock(sk) */
 	u64 buffer_size;
diff --git a/net/vmw_vsock/af_vsock.c b/net/vmw_vsock/af_vsock.c
index 44cee74519555..fe7f49da6c4bc 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_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 ff74e805527a5..f4b5f0230493e 100644
--- a/net/vmw_vsock/virtio_transport_common.c
+++ b/net/vmw_vsock/virtio_transport_common.c
@@ -1820,6 +1820,7 @@ virtio_transport_recv_pkt_batchable(struct virtio_transport *t,
 	return sk->sk_state == TCP_ESTABLISHED &&
 	       sk->sk_type == SOCK_STREAM &&
 	       READ_ONCE(sk->sk_prot) == sk->sk_prot_creator &&
+	       READ_ONCE(sk->sk_write_space) == vsk->default_write_space &&
 	       !sock_flag(sk, SOCK_DONE) &&
 	       vsk->transport == &t->transport &&
 	       vvs && virtio_transport_rx_skb_has_headroom(vvs) &&
@@ -1828,13 +1829,15 @@ 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.
+ * locked for another ordinary STREAM/RW packet.  When @batch is non-NULL,
+ * record a default write-space notification for batch finish instead of
+ * delivering it immediately.  Return true if the caller must free @skb.
  */
 static bool
 virtio_transport_recv_pkt_locked(struct virtio_transport *t,
 				 struct sk_buff *skb, struct sock *sk,
-				 bool *batchable)
+				 bool *batchable,
+				 struct virtio_transport_rx_batch *batch)
 {
 	struct virtio_vsock_hdr *hdr = virtio_vsock_hdr(skb);
 	struct sockaddr_vm src;
@@ -1863,8 +1866,12 @@ virtio_transport_recv_pkt_locked(struct virtio_transport *t,
 	if (vsk->local_addr.svm_cid != VMADDR_CID_ANY)
 		vsk->local_addr.svm_cid = le64_to_cpu(hdr->dst_cid);
 
-	if (space_available)
-		sk->sk_write_space(sk);
+	if (space_available) {
+		if (batch)
+			batch->write_space_pending = true;
+		else
+			sk->sk_write_space(sk);
+	}
 
 	switch (sk->sk_state) {
 	case TCP_LISTEN:
@@ -1931,7 +1938,7 @@ void virtio_transport_recv_pkt(struct virtio_transport *t,
 	}
 
 	lock_sock(sk);
-	free_pkt = virtio_transport_recv_pkt_locked(t, skb, sk, NULL);
+	free_pkt = virtio_transport_recv_pkt_locked(t, skb, sk, NULL, NULL);
 	release_sock(sk);
 
 	/* Release refcnt obtained when we fetched this socket out of the
@@ -1949,22 +1956,27 @@ EXPORT_SYMBOL_GPL(virtio_transport_recv_pkt);
 
 /*
  * Finish the RX batch.
- * For a non-empty batch, the caller must hold the socket lock acquired with
- * lock_sock(). This function releases the lock and the batch's lookup
- * reference.
+ * For a non-empty batch, the caller holds the socket lock acquired with
+ * lock_sock(). Deliver any deferred write-space notification while the lock
+ * is held, then release the lock and the batch's lookup reference.
  * An empty batch is a no-op.
  */
 void virtio_transport_rx_batch_finish(struct virtio_transport_rx_batch *batch)
 {
+	bool write_space_pending = batch->write_space_pending;
 	struct sock *sk = batch->sk;
 
 	batch->sk = NULL;
 	batch->pkts = 0;
 	batch->bytes = 0;
+	batch->write_space_pending = false;
 
 	if (!sk)
 		return;
 
+	if (write_space_pending)
+		vsock_sk(sk)->default_write_space(sk);
+
 	release_sock(sk);
 	sock_put(sk);
 }
@@ -2019,7 +2031,8 @@ void virtio_transport_recv_pkt_batch(struct virtio_transport *t,
 			}
 
 			free_pkt = virtio_transport_recv_pkt_locked(t, skb, sk,
-								    &batchable);
+								    &batchable,
+								    batch);
 			if (!batchable)
 				virtio_transport_rx_batch_finish(batch);
 			if (free_pkt)
@@ -2059,23 +2072,31 @@ void virtio_transport_recv_pkt_batch(struct virtio_transport *t,
 
 	lock_sock(sk);
 	/*
-	 * Sockmap removal restores the native protocol under sk_callback_lock.
-	 * Serialize this initial eligibility check with that update.
+	 * Sockmap teardown restores callbacks under sk_callback_lock. Keep the
+	 * initial eligibility check consistent with that update.
 	 */
 	read_lock_bh(&sk->sk_callback_lock);
 	start_batch = virtio_transport_recv_pkt_batchable(t, sk);
 	read_unlock_bh(&sk->sk_callback_lock);
 
+	if (start_batch)
+		batch->sk = sk;
+
 	free_pkt = virtio_transport_recv_pkt_locked(t, skb, sk,
-						    start_batch ? &batchable : NULL);
+						    start_batch ? &batchable : NULL,
+						    start_batch ? batch : NULL);
 	if (start_batch && batchable) {
 		/* Keep the lookup reference until the batch is released. */
-		batch->sk = sk;
 		return;
 	}
 
-	release_sock(sk);
-	sock_put(sk);
+	if (start_batch) {
+		virtio_transport_rx_batch_finish(batch);
+	} else {
+		release_sock(sk);
+		sock_put(sk);
+	}
+
 	if (free_pkt)
 		kfree_skb(skb);
 }
-- 
2.53.0

[PATCH net-next v4 4/4] vsock/virtio: defer RX notifications until batch unlock

[Jia Jia]

An RX batch can make a socket readable while the worker still holds its
socket lock. Waking a blocking reader then only moves it to waiting for
the same lock.

Record the existing low-watermark condition and notify once after
release_sock(). Move the coalesced write-space notification after the
unlock as well, since both callbacks wake sk_sleep(sk). Deliver
write-space before data-ready and keep the lookup reference until both
callbacks complete.

As with write-space, save the initial sk_data_ready callback and batch only
while both saved callbacks remain installed. Sockets with replaced
callbacks retain their per-packet behavior.

The existing psock-marker check also covers a concurrent sockmap attach:
at most the already admitted packet completes on the native path, and the
next packet ends the batch.

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

diff --git a/include/linux/virtio_vsock.h b/include/linux/virtio_vsock.h
index bf077ee0ded88..f1396ca3ad9b5 100644
--- a/include/linux/virtio_vsock.h
+++ b/include/linux/virtio_vsock.h
@@ -288,6 +288,7 @@ struct virtio_transport_rx_batch {
 	unsigned int pkts;
 	size_t bytes;
 	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 9a40960d10cbe..78bab5820f528 100644
--- a/include/net/af_vsock.h
+++ b/include/net/af_vsock.h
@@ -64,9 +64,10 @@ struct vsock_sock {
 	bool sent_request;
 	bool ignore_connecting_rst;
 	/*
-	 * Initial callback, used to identify replacements and deliver deferred
-	 * write-space notifications.
+	 * Initial callbacks, used to identify replacements and deliver deferred
+	 * notifications.
 	 */
+	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 fe7f49da6c4bc..f40659ec29df0 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 f4b5f0230493e..8b62b789b71c5 100644
--- a/net/vmw_vsock/virtio_transport_common.c
+++ b/net/vmw_vsock/virtio_transport_common.c
@@ -1588,7 +1588,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);
@@ -1607,7 +1608,13 @@ virtio_transport_recv_connected(struct sock *sk,
 			vsock_remove_sock(vsk);
 			break;
 		}
-		vsock_data_ready(sk);
+		if (!data_ready_pending) {
+			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);
@@ -1820,6 +1827,7 @@ virtio_transport_recv_pkt_batchable(struct virtio_transport *t,
 	return sk->sk_state == TCP_ESTABLISHED &&
 	       sk->sk_type == SOCK_STREAM &&
 	       READ_ONCE(sk->sk_prot) == sk->sk_prot_creator &&
+	       READ_ONCE(sk->sk_data_ready) == vsk->default_data_ready &&
 	       READ_ONCE(sk->sk_write_space) == vsk->default_write_space &&
 	       !sock_flag(sk, SOCK_DONE) &&
 	       vsk->transport == &t->transport &&
@@ -1830,8 +1838,9 @@ 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.  When @batch is non-NULL,
- * record a default write-space notification for batch finish instead of
- * delivering it immediately.  Return true if the caller must free @skb.
+ * record default write-space and data-ready notifications for batch finish
+ * instead of delivering them immediately.  Return true if the caller must
+ * free @skb.
  */
 static bool
 virtio_transport_recv_pkt_locked(struct virtio_transport *t,
@@ -1883,7 +1892,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,
+						batch ?
+						&batch->data_ready_pending : NULL);
 		break;
 	case TCP_CLOSING:
 		virtio_transport_recv_disconnecting(sk, skb);
@@ -1957,27 +1968,33 @@ EXPORT_SYMBOL_GPL(virtio_transport_recv_pkt);
 /*
  * Finish the RX batch.
  * For a non-empty batch, the caller holds the socket lock acquired with
- * lock_sock(). Deliver any deferred write-space notification while the lock
- * is held, then release the lock and the batch's lookup reference.
- * An empty batch is a no-op.
+ * lock_sock(). Release the lock, deliver any deferred write-space and
+ * data-ready notifications in that order, then drop the batch's lookup
+ * reference. An empty batch is a no-op.
  */
 void virtio_transport_rx_batch_finish(struct virtio_transport_rx_batch *batch)
 {
 	bool write_space_pending = batch->write_space_pending;
+	bool data_ready_pending = batch->data_ready_pending;
 	struct sock *sk = batch->sk;
 
 	batch->sk = NULL;
 	batch->pkts = 0;
 	batch->bytes = 0;
 	batch->write_space_pending = false;
+	batch->data_ready_pending = false;
 
 	if (!sk)
 		return;
 
+	release_sock(sk);
+
 	if (write_space_pending)
 		vsock_sk(sk)->default_write_space(sk);
 
-	release_sock(sk);
+	if (data_ready_pending)
+		vsock_sk(sk)->default_data_ready(sk);
+
 	sock_put(sk);
 }
 EXPORT_SYMBOL_GPL(virtio_transport_rx_batch_finish);
-- 
2.53.0

Re: [PATCH net-next v4 1/4] vsock/virtio: split socket lookup from locked RX processing

[Stefano Garzarella]

On Wed, Oct 14, 2026 at 07:56:37PM +0800, Jia Jia wrote:
>Split virtio_transport_recv_pkt() so socket lookup is separate from
>receive state machine processing under the socket lock. This prepares the
>receive path to keep a looked-up socket across multiple packets in a later
>change.
>
>Keep virtio_transport_recv_pkt() on its existing per-packet locking path.
>The locked helper obtains packet-specific addresses from the skb and uses
>the matched socket namespace for reset replies. Preserve source validation
>and failure handling.
>
>No functional change intended.
>
>Signed-off-by: Jia Jia <physicalmtea@gmail.com>
>---
> net/vmw_vsock/virtio_transport_common.c | 128 ++++++++++++++----------
> 1 file changed, 75 insertions(+), 53 deletions(-)
>
>diff --git a/net/vmw_vsock/virtio_transport_common.c b/net/vmw_vsock/virtio_transport_common.c
>index f225f53ed4bab..61ee3bc205d5d 100644
>--- a/net/vmw_vsock/virtio_transport_common.c
>+++ b/net/vmw_vsock/virtio_transport_common.c
>@@ -1774,81 +1774,60 @@ static bool virtio_transport_valid_type(u16 type)
> 	       (type == VIRTIO_VSOCK_TYPE_SEQPACKET);
> }
>
>-/* We are under the virtio-vsock's vsock->rx_lock or vhost-vsock's vq->mutex
>- * lock.
>- */
>-void virtio_transport_recv_pkt(struct virtio_transport *t,
>-			       struct sk_buff *skb, struct net *net)
>+static struct sock *
>+virtio_transport_find_socket(struct sk_buff *skb, struct sockaddr_vm *src,
>+			     struct sockaddr_vm *dst, struct net *net)
> {
> 	struct virtio_vsock_hdr *hdr = virtio_vsock_hdr(skb);
>-	struct sockaddr_vm src, dst;
>-	struct vsock_sock *vsk;
> 	struct sock *sk;
>-	bool space_available;
>-
>-	vsock_addr_init(&src, le64_to_cpu(hdr->src_cid),
>-			le32_to_cpu(hdr->src_port));
>-	vsock_addr_init(&dst, le64_to_cpu(hdr->dst_cid),
>-			le32_to_cpu(hdr->dst_port));
>-
>-	trace_virtio_transport_recv_pkt(src.svm_cid, src.svm_port,
>-					dst.svm_cid, dst.svm_port,
>-					le32_to_cpu(hdr->len),
>-					le16_to_cpu(hdr->type),
>-					le16_to_cpu(hdr->op),
>-					le32_to_cpu(hdr->flags),
>-					le32_to_cpu(hdr->buf_alloc),
>-					le32_to_cpu(hdr->fwd_cnt));
>
>-	if (!virtio_transport_valid_type(le16_to_cpu(hdr->type))) {
>-		(void)virtio_transport_reset_no_sock(t, skb, net);
>-		goto free_pkt;
>-	}
>+	if (!virtio_transport_valid_type(le16_to_cpu(hdr->type)))
>+		return NULL;
>
>-	/* The socket must be in connected or bound table
>-	 * otherwise send reset back
>-	 */
>-	sk = vsock_find_connected_socket_net(&src, &dst, net);
>-	if (!sk) {
>-		sk = vsock_find_bound_socket_net(&dst, net);
>-		if (!sk) {
>-			(void)virtio_transport_reset_no_sock(t, skb, net);
>-			goto free_pkt;
>-		}
>-	}
>+	sk = vsock_find_connected_socket_net(src, dst, net);
>+	if (!sk)
>+		sk = vsock_find_bound_socket_net(dst, net);
>+	if (!sk)
>+		return NULL;
>
> 	if (virtio_transport_get_type(sk) != le16_to_cpu(hdr->type)) {
>-		(void)virtio_transport_reset_no_sock(t, skb, net);
> 		sock_put(sk);
>-		goto free_pkt;
>+		return NULL;
> 	}
>
>-	if (!skb_set_owner_sk_safe(skb, sk)) {
>-		WARN_ONCE(1, "receiving vsock socket has sk_refcnt == 0\n");
>-		goto free_pkt;
>-	}
>+	return sk;
>+}
>
>-	vsk = vsock_sk(sk);
>+/*
>+ * The caller holds sk's socket lock and must free skb if this returns true.

Why the skb must be freed by the caller and not by this function itself?

>+ */
>+static bool
>+virtio_transport_recv_pkt_locked(struct virtio_transport *t,
>+				 struct sk_buff *skb, struct sock *sk)
>+{
>+	struct virtio_vsock_hdr *hdr = virtio_vsock_hdr(skb);
>+	struct sockaddr_vm src;
>+	struct vsock_sock *vsk = vsock_sk(sk);
>+	bool space_available;

Please, follow the netdev convection for ordering local variables: 
https://www.kernel.org/doc/html/v7.2/process/maintainer-netdev.html#local-variable-ordering-reverse-xmas-tree-rcs

>
>-	lock_sock(sk);
>+	vsock_addr_init(&src, le64_to_cpu(hdr->src_cid),
>+			le32_to_cpu(hdr->src_port));
>
>-	/* Check if sk has been closed or assigned to another transport before
>-	 * lock_sock (note: listener sockets are not assigned to any transport)
>+	/* Listener sockets accept packets from any source and are not assigned to
>+	 * a transport.
> 	 */

Why changing this comment in this commit?

Oh because we moved out lock_sock, right?

I think we should keep in some way the old info that was much more 
informative that the current comment that just say what the check is 
doing and not why.

> 	if (sock_flag(sk, SOCK_DONE) ||
> 	    (sk->sk_state != TCP_LISTEN &&
> 	     !vsock_check_source(vsk, &t->transport, &src))) {
>-		(void)virtio_transport_reset_no_sock(t, skb, net);
>-		release_sock(sk);
>-		sock_put(sk);
>-		goto free_pkt;
>+		(void)virtio_transport_reset_no_sock(t, skb, sock_net(sk));

We need to be careful here, IIRC Bobby used the net from the caller for 
a reason.

Are you sure it's fine this change? Why not passing net as parameter and 
minimize changes?

>+		return true;
> 	}
>
> 	space_available = virtio_transport_space_update(sk, skb);
>
> 	/* Update CID in case it has changed after a transport reset event */
> 	if (vsk->local_addr.svm_cid != VMADDR_CID_ANY)
>-		vsk->local_addr.svm_cid = dst.svm_cid;
>+		vsk->local_addr.svm_cid = le64_to_cpu(hdr->dst_cid);
>
> 	if (space_available)
> 		sk->sk_write_space(sk);
>@@ -1870,17 +1849,60 @@ void virtio_transport_recv_pkt(struct virtio_transport *t,
> 		kfree_skb(skb);
> 		break;
> 	default:
>-		(void)virtio_transport_reset_no_sock(t, skb, net);
>+		(void)virtio_transport_reset_no_sock(t, skb, sock_net(sk));
> 		kfree_skb(skb);
> 		break;
> 	}
>
>+	return false;
>+}
>+
>+/* We are under the virtio-vsock's vsock->rx_lock or vhost-vsock's vq->mutex
>+ * lock.
>+ */
>+void virtio_transport_recv_pkt(struct virtio_transport *t,
>+			       struct sk_buff *skb, struct net *net)
>+{
>+	struct virtio_vsock_hdr *hdr = virtio_vsock_hdr(skb);
>+	struct sockaddr_vm src, dst;
>+	struct sock *sk;
>+	bool free_pkt;
>+
>+	vsock_addr_init(&src, le64_to_cpu(hdr->src_cid),
>+			le32_to_cpu(hdr->src_port));
>+	vsock_addr_init(&dst, le64_to_cpu(hdr->dst_cid),
>+			le32_to_cpu(hdr->dst_port));
>+
>+	trace_virtio_transport_recv_pkt(src.svm_cid, src.svm_port,
>+					dst.svm_cid, dst.svm_port,
>+					le32_to_cpu(hdr->len),
>+					le16_to_cpu(hdr->type),
>+					le16_to_cpu(hdr->op),
>+					le32_to_cpu(hdr->flags),
>+					le32_to_cpu(hdr->buf_alloc),
>+					le32_to_cpu(hdr->fwd_cnt));
>+
>+	sk = virtio_transport_find_socket(skb, &src, &dst, net);
>+	if (!sk) {
>+		(void)virtio_transport_reset_no_sock(t, skb, net);
>+		goto free_pkt;
>+	}
>+
>+	if (!skb_set_owner_sk_safe(skb, sk)) {
>+		WARN_ONCE(1, "receiving vsock socket has sk_refcnt == 0\n");
>+		goto free_pkt;
>+	}
>+
>+	lock_sock(sk);
>+	free_pkt = virtio_transport_recv_pkt_locked(t, skb, sk);
> 	release_sock(sk);
>
> 	/* Release refcnt obtained when we fetched this socket out of the
> 	 * bound or connected list.
> 	 */
> 	sock_put(sk);
>+	if (free_pkt)
>+		kfree_skb(skb);
> 	return;
>
> free_pkt:
>-- 
>2.53.0
>

Re: [PATCH net-next v4 3/4] vsock/virtio: coalesce RX write-space notifications in lock batches

[Bobby Eshleman]

On Wed, Oct 14, 2026 at 07:56:39PM +0800, Jia Jia wrote:
> Each received packet updates peer credit and may call sk_write_space().
> A writer cannot use that credit until the batch releases the socket lock,
> so one notification at batch finish is sufficient.
> 
> Save the initial sk_write_space callback when the AF_VSOCK socket is
> created and require it to remain installed for batching. This lets the
> transport module deliver the deferred notification without referring to
> the unexported default callback. Sockets with replaced callbacks retain
> their per-packet behavior.
> 
> Record the batch socket before processing its first packet so a batch that
> ends on that packet cannot lose the pending notification.
> 
> Signed-off-by: Jia Jia <physicalmtea@gmail.com>
> ---
>  include/linux/virtio_vsock.h            |  1 +
>  include/net/af_vsock.h                  |  5 +++
>  net/vmw_vsock/af_vsock.c                |  1 +
>  net/vmw_vsock/virtio_transport_common.c | 53 +++++++++++++++++--------
>  4 files changed, 44 insertions(+), 16 deletions(-)
> 
> diff --git a/include/linux/virtio_vsock.h b/include/linux/virtio_vsock.h
> index 26dde6909c4ed..bf077ee0ded88 100644
> --- a/include/linux/virtio_vsock.h
> +++ b/include/linux/virtio_vsock.h
> @@ -287,6 +287,7 @@ struct virtio_transport_rx_batch {
>  	struct sock *sk;
>  	unsigned int pkts;
>  	size_t bytes;
> +	bool write_space_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 5549298c1ec6b..9a40960d10cbe 100644
> --- a/include/net/af_vsock.h
> +++ b/include/net/af_vsock.h
> @@ -63,6 +63,11 @@ struct vsock_sock {
>  	u32 peer_shutdown;
>  	bool sent_request;
>  	bool ignore_connecting_rst;
> +	/*
> +	 * Initial callback, used to identify replacements and deliver deferred
> +	 * write-space notifications.
> +	 */
> +	void (*default_write_space)(struct sock *sk);
>  
>  	/* Protected by lock_sock(sk) */
>  	u64 buffer_size;
> diff --git a/net/vmw_vsock/af_vsock.c b/net/vmw_vsock/af_vsock.c
> index 44cee74519555..fe7f49da6c4bc 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_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 ff74e805527a5..f4b5f0230493e 100644
> --- a/net/vmw_vsock/virtio_transport_common.c
> +++ b/net/vmw_vsock/virtio_transport_common.c
> @@ -1820,6 +1820,7 @@ virtio_transport_recv_pkt_batchable(struct virtio_transport *t,
>  	return sk->sk_state == TCP_ESTABLISHED &&
>  	       sk->sk_type == SOCK_STREAM &&
>  	       READ_ONCE(sk->sk_prot) == sk->sk_prot_creator &&
> +	       READ_ONCE(sk->sk_write_space) == vsk->default_write_space &&
>  	       !sock_flag(sk, SOCK_DONE) &&
>  	       vsk->transport == &t->transport &&
>  	       vvs && virtio_transport_rx_skb_has_headroom(vvs) &&
> @@ -1828,13 +1829,15 @@ 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.
> + * locked for another ordinary STREAM/RW packet.  When @batch is non-NULL,
> + * record a default write-space notification for batch finish instead of
> + * delivering it immediately.  Return true if the caller must free @skb.
>   */
>  static bool
>  virtio_transport_recv_pkt_locked(struct virtio_transport *t,
>  				 struct sk_buff *skb, struct sock *sk,
> -				 bool *batchable)
> +				 bool *batchable,
> +				 struct virtio_transport_rx_batch *batch)
>  {
>  	struct virtio_vsock_hdr *hdr = virtio_vsock_hdr(skb);
>  	struct sockaddr_vm src;
> @@ -1863,8 +1866,12 @@ virtio_transport_recv_pkt_locked(struct virtio_transport *t,
>  	if (vsk->local_addr.svm_cid != VMADDR_CID_ANY)
>  		vsk->local_addr.svm_cid = le64_to_cpu(hdr->dst_cid);
>  
> -	if (space_available)
> -		sk->sk_write_space(sk);
> +	if (space_available) {
> +		if (batch)
> +			batch->write_space_pending = true;
> +		else
> +			sk->sk_write_space(sk);
> +	}
>  
>  	switch (sk->sk_state) {
>  	case TCP_LISTEN:
> @@ -1931,7 +1938,7 @@ void virtio_transport_recv_pkt(struct virtio_transport *t,
>  	}
>  
>  	lock_sock(sk);
> -	free_pkt = virtio_transport_recv_pkt_locked(t, skb, sk, NULL);
> +	free_pkt = virtio_transport_recv_pkt_locked(t, skb, sk, NULL, NULL);
>  	release_sock(sk);
>  
>  	/* Release refcnt obtained when we fetched this socket out of the
> @@ -1949,22 +1956,27 @@ EXPORT_SYMBOL_GPL(virtio_transport_recv_pkt);
>  
>  /*
>   * Finish the RX batch.
> - * For a non-empty batch, the caller must hold the socket lock acquired with
> - * lock_sock(). This function releases the lock and the batch's lookup
> - * reference.
> + * For a non-empty batch, the caller holds the socket lock acquired with
> + * lock_sock(). Deliver any deferred write-space notification while the lock
> + * is held, then release the lock and the batch's lookup reference.
>   * An empty batch is a no-op.
>   */
>  void virtio_transport_rx_batch_finish(struct virtio_transport_rx_batch *batch)
>  {
> +	bool write_space_pending = batch->write_space_pending;
>  	struct sock *sk = batch->sk;
>  
>  	batch->sk = NULL;
>  	batch->pkts = 0;
>  	batch->bytes = 0;
> +	batch->write_space_pending = false;
>  
>  	if (!sk)
>  		return;
>  
> +	if (write_space_pending)
> +		vsock_sk(sk)->default_write_space(sk);

Any reason not to just do READ_ONCE(sk->sk_write_space)(sk) instead of
storing this for later? If we setup the socket in a sockmap program
mid-batch, doesn't this and sk_data_ready (from the next patch) go
stale?

> +
>  	release_sock(sk);
>  	sock_put(sk);
>  }
> @@ -2019,7 +2031,8 @@ void virtio_transport_recv_pkt_batch(struct virtio_transport *t,
>  			}
>  
>  			free_pkt = virtio_transport_recv_pkt_locked(t, skb, sk,
> -								    &batchable);
> +								    &batchable,
> +								    batch);
>  			if (!batchable)
>  				virtio_transport_rx_batch_finish(batch);
>  			if (free_pkt)
> @@ -2059,23 +2072,31 @@ void virtio_transport_recv_pkt_batch(struct virtio_transport *t,
>  
>  	lock_sock(sk);
>  	/*
> -	 * Sockmap removal restores the native protocol under sk_callback_lock.
> -	 * Serialize this initial eligibility check with that update.
> +	 * Sockmap teardown restores callbacks under sk_callback_lock. Keep the
> +	 * initial eligibility check consistent with that update.

Why reword here?

>  	 */
>  	read_lock_bh(&sk->sk_callback_lock);
>  	start_batch = virtio_transport_recv_pkt_batchable(t, sk);
>  	read_unlock_bh(&sk->sk_callback_lock);
>  
> +	if (start_batch)
> +		batch->sk = sk;
> +
>  	free_pkt = virtio_transport_recv_pkt_locked(t, skb, sk,
> -						    start_batch ? &batchable : NULL);
> +						    start_batch ? &batchable : NULL,
> +						    start_batch ? batch : NULL);
>  	if (start_batch && batchable) {
>  		/* Keep the lookup reference until the batch is released. */
> -		batch->sk = sk;

Seems like `batch->sk = sk` is moving around between patches?

I don't know, feels like if the batching patch introduces
sk_write_space/sk_data_ready bugs, then that patch should probably
contain these patches too. Should minimize churn too?


>  		return;
>  	}
>  
> -	release_sock(sk);
> -	sock_put(sk);
> +	if (start_batch) {
> +		virtio_transport_rx_batch_finish(batch);
> +	} else {
> +		release_sock(sk);
> +		sock_put(sk);
> +	}
> +
>  	if (free_pkt)
>  		kfree_skb(skb);
>  }
> -- 
> 2.53.0
> 

Best,
Bobby

Re: [PATCH net-next v4 1/4] vsock/virtio: split socket lookup from locked RX processing

[Bobby Eshleman]

On Fri, Oct 16, 2026 at 03:47:33PM +0200, Stefano Garzarella wrote:
> On Wed, Oct 14, 2026 at 07:56:37PM +0800, Jia Jia wrote:
> > Split virtio_transport_recv_pkt() so socket lookup is separate from
> > receive state machine processing under the socket lock. This prepares the
> > receive path to keep a looked-up socket across multiple packets in a later
> > change.
> > 
> > Keep virtio_transport_recv_pkt() on its existing per-packet locking path.
> > The locked helper obtains packet-specific addresses from the skb and uses
> > the matched socket namespace for reset replies. Preserve source validation
> > and failure handling.
> > 
> > No functional change intended.
> > 
> > Signed-off-by: Jia Jia <physicalmtea@gmail.com>
> > ---
> > net/vmw_vsock/virtio_transport_common.c | 128 ++++++++++++++----------
> > 1 file changed, 75 insertions(+), 53 deletions(-)
> > 
> > diff --git a/net/vmw_vsock/virtio_transport_common.c b/net/vmw_vsock/virtio_transport_common.c
> > index f225f53ed4bab..61ee3bc205d5d 100644
> > --- a/net/vmw_vsock/virtio_transport_common.c
> > +++ b/net/vmw_vsock/virtio_transport_common.c
> > @@ -1774,81 +1774,60 @@ static bool virtio_transport_valid_type(u16 type)
> > 	       (type == VIRTIO_VSOCK_TYPE_SEQPACKET);
> > }
> > 
> > -/* We are under the virtio-vsock's vsock->rx_lock or vhost-vsock's vq->mutex
> > - * lock.
> > - */
> > -void virtio_transport_recv_pkt(struct virtio_transport *t,
> > -			       struct sk_buff *skb, struct net *net)
> > +static struct sock *
> > +virtio_transport_find_socket(struct sk_buff *skb, struct sockaddr_vm *src,
> > +			     struct sockaddr_vm *dst, struct net *net)
> > {
> > 	struct virtio_vsock_hdr *hdr = virtio_vsock_hdr(skb);
> > -	struct sockaddr_vm src, dst;
> > -	struct vsock_sock *vsk;
> > 	struct sock *sk;
> > -	bool space_available;
> > -
> > -	vsock_addr_init(&src, le64_to_cpu(hdr->src_cid),
> > -			le32_to_cpu(hdr->src_port));
> > -	vsock_addr_init(&dst, le64_to_cpu(hdr->dst_cid),
> > -			le32_to_cpu(hdr->dst_port));
> > -
> > -	trace_virtio_transport_recv_pkt(src.svm_cid, src.svm_port,
> > -					dst.svm_cid, dst.svm_port,
> > -					le32_to_cpu(hdr->len),
> > -					le16_to_cpu(hdr->type),
> > -					le16_to_cpu(hdr->op),
> > -					le32_to_cpu(hdr->flags),
> > -					le32_to_cpu(hdr->buf_alloc),
> > -					le32_to_cpu(hdr->fwd_cnt));
> > 
> > -	if (!virtio_transport_valid_type(le16_to_cpu(hdr->type))) {
> > -		(void)virtio_transport_reset_no_sock(t, skb, net);
> > -		goto free_pkt;
> > -	}
> > +	if (!virtio_transport_valid_type(le16_to_cpu(hdr->type)))
> > +		return NULL;
> > 
> > -	/* The socket must be in connected or bound table
> > -	 * otherwise send reset back
> > -	 */
> > -	sk = vsock_find_connected_socket_net(&src, &dst, net);
> > -	if (!sk) {
> > -		sk = vsock_find_bound_socket_net(&dst, net);
> > -		if (!sk) {
> > -			(void)virtio_transport_reset_no_sock(t, skb, net);
> > -			goto free_pkt;
> > -		}
> > -	}
> > +	sk = vsock_find_connected_socket_net(src, dst, net);
> > +	if (!sk)
> > +		sk = vsock_find_bound_socket_net(dst, net);
> > +	if (!sk)
> > +		return NULL;
> > 
> > 	if (virtio_transport_get_type(sk) != le16_to_cpu(hdr->type)) {
> > -		(void)virtio_transport_reset_no_sock(t, skb, net);
> > 		sock_put(sk);
> > -		goto free_pkt;
> > +		return NULL;
> > 	}
> > 
> > -	if (!skb_set_owner_sk_safe(skb, sk)) {
> > -		WARN_ONCE(1, "receiving vsock socket has sk_refcnt == 0\n");
> > -		goto free_pkt;
> > -	}
> > +	return sk;
> > +}
> > 
> > -	vsk = vsock_sk(sk);
> > +/*
> > + * The caller holds sk's socket lock and must free skb if this returns true.
> 
> Why the skb must be freed by the caller and not by this function itself?
> 
> > + */
> > +static bool
> > +virtio_transport_recv_pkt_locked(struct virtio_transport *t,
> > +				 struct sk_buff *skb, struct sock *sk)
> > +{
> > +	struct virtio_vsock_hdr *hdr = virtio_vsock_hdr(skb);
> > +	struct sockaddr_vm src;
> > +	struct vsock_sock *vsk = vsock_sk(sk);
> > +	bool space_available;
> 
> Please, follow the netdev convection for ordering local variables: https://www.kernel.org/doc/html/v7.2/process/maintainer-netdev.html#local-variable-ordering-reverse-xmas-tree-rcs
> 
> > 
> > -	lock_sock(sk);
> > +	vsock_addr_init(&src, le64_to_cpu(hdr->src_cid),
> > +			le32_to_cpu(hdr->src_port));
> > 
> > -	/* Check if sk has been closed or assigned to another transport before
> > -	 * lock_sock (note: listener sockets are not assigned to any transport)
> > +	/* Listener sockets accept packets from any source and are not assigned to
> > +	 * a transport.
> > 	 */
> 
> Why changing this comment in this commit?
> 
> Oh because we moved out lock_sock, right?
> 
> I think we should keep in some way the old info that was much more
> informative that the current comment that just say what the check is doing
> and not why.
> 
> > 	if (sock_flag(sk, SOCK_DONE) ||
> > 	    (sk->sk_state != TCP_LISTEN &&
> > 	     !vsock_check_source(vsk, &t->transport, &src))) {
> > -		(void)virtio_transport_reset_no_sock(t, skb, net);
> > -		release_sock(sk);
> > -		sock_put(sk);
> > -		goto free_pkt;
> > +		(void)virtio_transport_reset_no_sock(t, skb, sock_net(sk));
> 
> We need to be careful here, IIRC Bobby used the net from the caller for a
> reason.
> 
> Are you sure it's fine this change? Why not passing net as parameter and
> minimize changes?

I think you may be remembering this issue:

a6ae12a599e0 ("virtio: set skb owner of virtio_transport_reset_no_sock() reply")

?

If so, sock_net(sk) might still be okay, but I 100% agree we shouldn't
change things haphazardly if avoidable.

> 
> > +		return true;
> > 	}
> > 
> > 	space_available = virtio_transport_space_update(sk, skb);
> > 
> > 	/* Update CID in case it has changed after a transport reset event */
> > 	if (vsk->local_addr.svm_cid != VMADDR_CID_ANY)
> > -		vsk->local_addr.svm_cid = dst.svm_cid;
> > +		vsk->local_addr.svm_cid = le64_to_cpu(hdr->dst_cid);
> > 
> > 	if (space_available)
> > 		sk->sk_write_space(sk);
> > @@ -1870,17 +1849,60 @@ void virtio_transport_recv_pkt(struct virtio_transport *t,
> > 		kfree_skb(skb);
> > 		break;
> > 	default:
> > -		(void)virtio_transport_reset_no_sock(t, skb, net);
> > +		(void)virtio_transport_reset_no_sock(t, skb, sock_net(sk));
> > 		kfree_skb(skb);
> > 		break;
> > 	}
> > 
> > +	return false;
> > +}
> > +
> > +/* We are under the virtio-vsock's vsock->rx_lock or vhost-vsock's vq->mutex
> > + * lock.
> > + */
> > +void virtio_transport_recv_pkt(struct virtio_transport *t,
> > +			       struct sk_buff *skb, struct net *net)
> > +{
> > +	struct virtio_vsock_hdr *hdr = virtio_vsock_hdr(skb);
> > +	struct sockaddr_vm src, dst;
> > +	struct sock *sk;
> > +	bool free_pkt;
> > +
> > +	vsock_addr_init(&src, le64_to_cpu(hdr->src_cid),
> > +			le32_to_cpu(hdr->src_port));
> > +	vsock_addr_init(&dst, le64_to_cpu(hdr->dst_cid),
> > +			le32_to_cpu(hdr->dst_port));
> > +
> > +	trace_virtio_transport_recv_pkt(src.svm_cid, src.svm_port,
> > +					dst.svm_cid, dst.svm_port,
> > +					le32_to_cpu(hdr->len),
> > +					le16_to_cpu(hdr->type),
> > +					le16_to_cpu(hdr->op),
> > +					le32_to_cpu(hdr->flags),
> > +					le32_to_cpu(hdr->buf_alloc),
> > +					le32_to_cpu(hdr->fwd_cnt));
> > +
> > +	sk = virtio_transport_find_socket(skb, &src, &dst, net);
> > +	if (!sk) {
> > +		(void)virtio_transport_reset_no_sock(t, skb, net);
> > +		goto free_pkt;
> > +	}
> > +
> > +	if (!skb_set_owner_sk_safe(skb, sk)) {
> > +		WARN_ONCE(1, "receiving vsock socket has sk_refcnt == 0\n");
> > +		goto free_pkt;
> > +	}
> > +
> > +	lock_sock(sk);
> > +	free_pkt = virtio_transport_recv_pkt_locked(t, skb, sk);
> > 	release_sock(sk);
> > 
> > 	/* Release refcnt obtained when we fetched this socket out of the
> > 	 * bound or connected list.
> > 	 */
> > 	sock_put(sk);
> > +	if (free_pkt)
> > +		kfree_skb(skb);
> > 	return;
> > 
> > free_pkt:
> > -- 
> > 2.53.0
> > 
> 

Re: [PATCH net-next v4 2/4] vsock/virtio: amortize RX socket locking for stream packets

[Stefano Garzarella]

On Wed, Oct 14, 2026 at 07:56:38PM +0800, Jia Jia wrote:
>The virtio-vsock RX worker currently looks up the socket and takes its
>lock for every packet. Keep the socket lock and lookup reference across a
>bounded run of STREAM/RW packets for the same established socket.
>
>Reuse the locked socket while packets continue to match the same flow.
>End the batch for a different flow, non-RW traffic, malformed packets, or
>a change in socket state, transport, protocol, or receive-buffer headroom.
>A connected-table miss uses the existing receive path.
>
>The reused path still traces every packet, takes an skb owner reference,
>validates the source and transport, updates credit, and runs the existing
>receive state machine.
>
>Sockmap can replace the protocol without taking the socket lock. Check the
>psock marker at each packet boundary and serialize initial admission with
>sk_callback_lock, so a concurrent attach ends batching at the next packet.
>
>Cap a batch at 64 packets and 64K of payload. The packet limit bounds

Why this cap?

>per-packet work, while the byte limit bounds how much data can queue before
>a reader acquires the socket lock. Batching remains in the process-context
>RX worker.

IMHO this patch should carry performance results.

>
>Signed-off-by: Jia Jia <physicalmtea@gmail.com>
>---
> include/linux/virtio_vsock.h            |  11 ++
> net/vmw_vsock/virtio_transport.c        |  29 +++-

Why no changes in vhost-vsock or vsock_loopback?

I'm not asking to do that, I'd like to understand why only in the 
driver.

> net/vmw_vsock/virtio_transport_common.c | 183 +++++++++++++++++++++++-
> 3 files changed, 216 insertions(+), 7 deletions(-)
>
>diff --git a/include/linux/virtio_vsock.h b/include/linux/virtio_vsock.h
>index f91704731057e..26dde6909c4ed 100644
>--- a/include/linux/virtio_vsock.h
>+++ b/include/linux/virtio_vsock.h
>@@ -282,6 +282,17 @@ void virtio_transport_destruct(struct vsock_sock *vsk);
>
> void virtio_transport_recv_pkt(struct virtio_transport *t,
> 			       struct sk_buff *skb, struct net *net);
>+
>+struct virtio_transport_rx_batch {
>+	struct sock *sk;
>+	unsigned int pkts;
>+	size_t bytes;
>+};
>+
>+void virtio_transport_recv_pkt_batch(struct virtio_transport *t,
>+				     struct sk_buff *skb, struct net *net,
>+				     struct virtio_transport_rx_batch *batch);
>+void virtio_transport_rx_batch_finish(struct virtio_transport_rx_batch *batch);
> void virtio_transport_inc_tx_pkt(struct virtio_vsock_sock *vvs, struct sk_buff *skb);
> u32 virtio_transport_get_credit(struct virtio_vsock_sock *vvs, u32 wanted);
> void virtio_transport_put_credit(struct virtio_vsock_sock *vvs, u32 credit);
>diff --git a/net/vmw_vsock/virtio_transport.c b/net/vmw_vsock/virtio_transport.c
>index 4f9aa9c4c3aa5..ef076732158d4 100644
>--- a/net/vmw_vsock/virtio_transport.c
>+++ b/net/vmw_vsock/virtio_transport.c
>@@ -629,8 +629,16 @@ virtio_transport_seqpacket_allow(struct vsock_sock *vsk, u32 remote_cid)
> 	return seqpacket_allow;
> }
>
>+/*
>+ * Keep a bounded run of packets for one socket under a single socket lock.
>+ * Limit packet count and payload size to bound the work done while locked.

Here we should explain why we choose these values.

>+ */
>+#define VIRTIO_TRANSPORT_RX_BATCH_MAX_PKTS	64
>+#define VIRTIO_TRANSPORT_RX_BATCH_MAX_BYTES	(64 * 1024)
>+
> static void virtio_transport_rx_work(struct work_struct *work)
> {
>+	struct virtio_transport_rx_batch batch = {};
> 	struct virtio_vsock *vsock =
> 		container_of(work, struct virtio_vsock, rx_work);
> 	struct virtqueue *vq;
>@@ -666,6 +674,7 @@ static void virtio_transport_rx_work(struct work_struct *work)
> 			/* Drop short/long packets */
> 			if (unlikely(len < sizeof(*hdr) ||
> 				     len > virtio_vsock_skb_len(skb))) {
>+				virtio_transport_rx_batch_finish(&batch);
> 				kfree_skb(skb);
> 				continue;
> 			}
>@@ -673,6 +682,7 @@ static void virtio_transport_rx_work(struct work_struct *work)
> 			hdr = virtio_vsock_hdr(skb);
> 			payload_len = le32_to_cpu(hdr->len);
> 			if (unlikely(payload_len > len - sizeof(*hdr))) {
>+				virtio_transport_rx_batch_finish(&batch);
> 				kfree_skb(skb);
> 				continue;
> 			}
>@@ -680,16 +690,33 @@ static void virtio_transport_rx_work(struct work_struct *work)
> 			if (payload_len)
> 				virtio_vsock_skb_put(skb, payload_len);
>
>+			if (batch.sk &&
>+			    payload_len > VIRTIO_TRANSPORT_RX_BATCH_MAX_BYTES -
>+					  batch.bytes) {

Why this?


>+				virtio_transport_rx_batch_finish(&batch);
>+			}
>+
> 			virtio_transport_deliver_tap_pkt(skb);
>
> 			/* Force virtio-transport into global mode since it
> 			 * does not yet support local-mode namespacing.
> 			 */
>-			virtio_transport_recv_pkt(&virtio_transport, skb, NULL);
>+			virtio_transport_recv_pkt_batch(&virtio_transport, skb, NULL,
>+							&batch);
>+
>+			if (batch.sk) {
>+				batch.pkts++;
>+				batch.bytes += payload_len;
>+				if (batch.pkts >= VIRTIO_TRANSPORT_RX_BATCH_MAX_PKTS ||
>+				    batch.bytes >= VIRTIO_TRANSPORT_RX_BATCH_MAX_BYTES) {
>+					virtio_transport_rx_batch_finish(&batch);
>+				}
>+			}

Why this is not in the common part?

> 		}
> 	} while (!virtqueue_enable_cb(vq));
>
> out:
>+	virtio_transport_rx_batch_finish(&batch);
> 	if (vsock->rx_buf_nr < vsock->rx_buf_max_nr / 2)
> 		virtio_vsock_rx_fill(vsock);
> out_nofill:
>diff --git a/net/vmw_vsock/virtio_transport_common.c b/net/vmw_vsock/virtio_transport_common.c
>index 61ee3bc205d5d..ff74e805527a5 100644
>--- a/net/vmw_vsock/virtio_transport_common.c
>+++ b/net/vmw_vsock/virtio_transport_common.c
>@@ -576,11 +576,16 @@ virtio_transport_collapse_rx_queue(struct virtio_vsock_sock *vvs,
> 	skb_queue_splice(&new_queue, &vvs->rx_queue);
> }
>
>+static bool
>+virtio_transport_rx_skb_has_headroom(struct virtio_vsock_sock *vvs)
>+{
>+	return (u64)(skb_queue_len(&vvs->rx_queue) + 1) * SKB_TRUESIZE(0) <=
>+		vvs->buf_alloc;
>+}

Please move this change in a separate patch.

skb_has_headroom() is confusing IMO, maybe 
virtio_transport_rx_buf_full(), but reversing the return value, so 
keeping the same check as before.

>+
> static bool virtio_transport_inc_rx_pkt(struct virtio_vsock_sock *vvs,
> 					u32 len)
> {
>-	u64 skb_overhead = (skb_queue_len(&vvs->rx_queue) + 1) * SKB_TRUESIZE(0);
>-
> 	/* Allow at most buf_alloc * 2 total budget (payload + overhead),
> 	 * similar to how SO_RCVBUF is doubled to reserve space for sk_buff
> 	 * metadata. Check payload against buf_alloc to be sure the other
>@@ -588,7 +593,7 @@ static bool virtio_transport_inc_rx_pkt(struct virtio_vsock_sock *vvs,
> 	 * queue growth.
> 	 */
> 	if ((u64)vvs->buf_used + len > vvs->buf_alloc ||
>-	    skb_overhead > vvs->buf_alloc)
>+	    !virtio_transport_rx_skb_has_headroom(vvs))
> 		return false;
>
> 	vvs->rx_bytes += len;
>@@ -1798,12 +1803,38 @@ virtio_transport_find_socket(struct sk_buff *skb, struct sockaddr_vm *src,
> 	return sk;
> }
>
>+static bool
>+virtio_transport_sock_has_psock(const struct sock *sk)
>+{
>+	return (uintptr_t)rcu_access_pointer(__sk_user_data(sk)) &
>+	       SK_USER_DATA_PSOCK;
>+}
>+
>+static bool
>+virtio_transport_recv_pkt_batchable(struct virtio_transport *t,
>+				    struct sock *sk)
>+{
>+	struct vsock_sock *vsk = vsock_sk(sk);
>+	struct virtio_vsock_sock *vvs = vsk->trans;
>+

A comment here to explain why some checks like...

>+	return sk->sk_state == TCP_ESTABLISHED &&
>+	       sk->sk_type == SOCK_STREAM &&
>+	       READ_ONCE(sk->sk_prot) == sk->sk_prot_creator &&
... this       ^

I mean in general to explain better these conditions (I mean not a 
description of what we are doing, but why we are doing these checks).

>+	       !sock_flag(sk, SOCK_DONE) &&
>+	       vsk->transport == &t->transport &&
>+	       vvs && virtio_transport_rx_skb_has_headroom(vvs) &&
>+	       !virtio_transport_sock_has_psock(sk);
>+}
>+
> /*
>- * The caller holds sk's socket lock and must free skb if this returns true.
>+ * 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.
>  */
> static bool
> virtio_transport_recv_pkt_locked(struct virtio_transport *t,
>-				 struct sk_buff *skb, struct sock *sk)
>+				 struct sk_buff *skb, struct sock *sk,
>+				 bool *batchable)
> {
> 	struct virtio_vsock_hdr *hdr = virtio_vsock_hdr(skb);
> 	struct sockaddr_vm src;
>@@ -1813,6 +1844,9 @@ virtio_transport_recv_pkt_locked(struct virtio_transport *t,
> 	vsock_addr_init(&src, le64_to_cpu(hdr->src_cid),
> 			le32_to_cpu(hdr->src_port));
>
>+	if (batchable)
>+		*batchable = false;
>+
> 	/* Listener sockets accept packets from any source and are not assigned to
> 	 * a transport.
> 	 */
>@@ -1854,6 +1888,9 @@ virtio_transport_recv_pkt_locked(struct virtio_transport *t,
> 		break;
> 	}
>
>+	if (batchable)
>+		*batchable = virtio_transport_recv_pkt_batchable(t, sk);
>+
> 	return false;
> }
>
>@@ -1894,7 +1931,7 @@ void virtio_transport_recv_pkt(struct virtio_transport *t,
> 	}
>
> 	lock_sock(sk);
>-	free_pkt = virtio_transport_recv_pkt_locked(t, skb, sk);
>+	free_pkt = virtio_transport_recv_pkt_locked(t, skb, sk, NULL);
> 	release_sock(sk);
>
> 	/* Release refcnt obtained when we fetched this socket out of the
>@@ -1910,6 +1947,140 @@ void virtio_transport_recv_pkt(struct virtio_transport *t,
> }
> EXPORT_SYMBOL_GPL(virtio_transport_recv_pkt);
>
>+/*
>+ * Finish the RX batch.
>+ * For a non-empty batch, the caller must hold the socket lock acquired with
>+ * lock_sock(). This function releases the lock and the batch's lookup
>+ * reference.
>+ * An empty batch is a no-op.

What empty batch means, I mean how the caller can understand that a 
batch is empty?

Maybe we should provide a function for that...

>+ */
>+void virtio_transport_rx_batch_finish(struct virtio_transport_rx_batch *batch)
>+{
>+	struct sock *sk = batch->sk;
>+
>+	batch->sk = NULL;
>+	batch->pkts = 0;
>+	batch->bytes = 0;
>+
>+	if (!sk)
>+		return;
>+
>+	release_sock(sk);
>+	sock_put(sk);
>+}
>+EXPORT_SYMBOL_GPL(virtio_transport_rx_batch_finish);
>+
>+void virtio_transport_recv_pkt_batch(struct virtio_transport *t,
>+				     struct sk_buff *skb, struct net *net,
>+				     struct virtio_transport_rx_batch *batch)
>+{
>+	struct virtio_vsock_hdr *hdr = virtio_vsock_hdr(skb);
>+	struct sockaddr_vm src, dst;
>+	struct vsock_sock *vsk;
>+	bool batchable, start_batch;

Ditto about declarations order, please check in the entire series.

>+	struct sock *sk;
>+	bool free_pkt;
>+
>+	/* Only STREAM/RW packets can share a socket lock. */

Why?

Should the caller holds the lock for the entire batch?
Should this be documented?

>+	if (le16_to_cpu(hdr->type) != VIRTIO_VSOCK_TYPE_STREAM ||
>+	    le16_to_cpu(hdr->op) != VIRTIO_VSOCK_OP_RW) {
>+		virtio_transport_rx_batch_finish(batch);
>+		virtio_transport_recv_pkt(t, skb, net);
>+		return;
>+	}
>+
>+	vsock_addr_init(&src, le64_to_cpu(hdr->src_cid),
>+			le32_to_cpu(hdr->src_port));
>+	vsock_addr_init(&dst, le64_to_cpu(hdr->dst_cid),
>+			le32_to_cpu(hdr->dst_port));
>+
>+	if (batch->sk) {
>+		vsk = vsock_sk(batch->sk);
>+		if (vsock_addr_equals_addr(&src, &vsk->remote_addr) &&
>+		    dst.svm_port == vsk->local_addr.svm_port &&
>+		    vsock_net_check_mode(sock_net(batch->sk), net) &&
>+		    virtio_transport_recv_pkt_batchable(t, batch->sk)) {
>+			sk = batch->sk;
>+			trace_virtio_transport_recv_pkt(src.svm_cid,
>+							src.svm_port,
>+							dst.svm_cid,
>+							dst.svm_port,
>+							le32_to_cpu(hdr->len),
>+							le16_to_cpu(hdr->type),
>+							le16_to_cpu(hdr->op),
>+							le32_to_cpu(hdr->flags),
>+							le32_to_cpu(hdr->buf_alloc),
>+							le32_to_cpu(hdr->fwd_cnt));
>+			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);
>+				kfree_skb(skb);
>+				return;
>+			}
>+

This code is unreadable TBH, there isn't a blank line till here...
and too much nested blocks.

Maybe you should move this in a function, and use a bit more blank lines.


>+			free_pkt = virtio_transport_recv_pkt_locked(t, skb, sk,
>+								    &batchable);
>+			if (!batchable)
>+				virtio_transport_rx_batch_finish(batch);
>+			if (free_pkt)
>+				kfree_skb(skb);
>+			return;
>+		}
>+
>+		virtio_transport_rx_batch_finish(batch);
>+	}
>+
>+	sk = vsock_find_connected_socket_net(&src, &dst, net);
>+	if (!sk) {
>+		virtio_transport_recv_pkt(t, skb, net);

Why this?

>+		return;
>+	}
>+
>+	if (virtio_transport_get_type(sk) != le16_to_cpu(hdr->type)) {
>+		sock_put(sk);
>+		virtio_transport_recv_pkt(t, skb, net);
>+		return;
>+	}
>+
>+	trace_virtio_transport_recv_pkt(src.svm_cid, src.svm_port,
>+					dst.svm_cid, dst.svm_port,
>+					le32_to_cpu(hdr->len),
>+					le16_to_cpu(hdr->type),
>+					le16_to_cpu(hdr->op),
>+					le32_to_cpu(hdr->flags),
>+					le32_to_cpu(hdr->buf_alloc),
>+					le32_to_cpu(hdr->fwd_cnt));
>+
>+	if (!skb_set_owner_sk_safe(skb, sk)) {
>+		WARN_ONCE(1, "receiving vsock socket has sk_refcnt == 0\n");
>+		kfree_skb(skb);
>+		return;
>+	}

We have this code duplicated in several place now, can this be avoided?

>+
>+	lock_sock(sk);
>+	/*
>+	 * Sockmap removal restores the native protocol under sk_callback_lock.
>+	 * Serialize this initial eligibility check with that update.
>+	 */
>+	read_lock_bh(&sk->sk_callback_lock);
>+	start_batch = virtio_transport_recv_pkt_batchable(t, sk);
>+	read_unlock_bh(&sk->sk_callback_lock);
>+
>+	free_pkt = virtio_transport_recv_pkt_locked(t, skb, sk,
>+						    start_batch ? &batchable : NULL);
>+	if (start_batch && batchable) {
>+		/* Keep the lookup reference until the batch is released. */
>+		batch->sk = sk;
>+		return;
>+	}

Ah, this is for the first run, when the batch is empty, right?

Can you merge in some way both branches?

BTW please add some more useful comments (explaining why)

>+
>+	release_sock(sk);
>+	sock_put(sk);
>+	if (free_pkt)
>+		kfree_skb(skb);
>+}
>+EXPORT_SYMBOL_GPL(virtio_transport_recv_pkt_batch);
>+
> /* Remove skbs found in a queue that have a vsk that matches.
>  *
>  * Each skb is freed.
>-- 
>2.53.0
>

Re: [PATCH net-next v4 3/4] vsock/virtio: coalesce RX write-space notifications in lock batches

[Stefano Garzarella]

On Wed, Oct 14, 2026 at 07:56:39PM +0800, Jia Jia wrote:
>Each received packet updates peer credit and may call sk_write_space().
>A writer cannot use that credit until the batch releases the socket lock,
>so one notification at batch finish is sufficient.
>
>Save the initial sk_write_space callback when the AF_VSOCK socket is
>created and require it to remain installed for batching. This lets the
>transport module deliver the deferred notification without referring to
>the unexported default callback. Sockets with replaced callbacks retain
>their per-packet behavior.
>
>Record the batch socket before processing its first packet so a batch that
>ends on that packet cannot lose the pending notification.

mmm, is this a fix of the previous patch?

>
>Signed-off-by: Jia Jia <physicalmtea@gmail.com>
>---
> include/linux/virtio_vsock.h            |  1 +
> include/net/af_vsock.h                  |  5 +++
> net/vmw_vsock/af_vsock.c                |  1 +
> net/vmw_vsock/virtio_transport_common.c | 53 +++++++++++++++++--------
> 4 files changed, 44 insertions(+), 16 deletions(-)
>
>diff --git a/include/linux/virtio_vsock.h b/include/linux/virtio_vsock.h
>index 26dde6909c4ed..bf077ee0ded88 100644
>--- a/include/linux/virtio_vsock.h
>+++ b/include/linux/virtio_vsock.h
>@@ -287,6 +287,7 @@ struct virtio_transport_rx_batch {
> 	struct sock *sk;
> 	unsigned int pkts;
> 	size_t bytes;
>+	bool write_space_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 5549298c1ec6b..9a40960d10cbe 100644
>--- a/include/net/af_vsock.h
>+++ b/include/net/af_vsock.h
>@@ -63,6 +63,11 @@ struct vsock_sock {
> 	u32 peer_shutdown;
> 	bool sent_request;
> 	bool ignore_connecting_rst;
>+	/*
>+	 * Initial callback, used to identify replacements and deliver deferred
>+	 * write-space notifications.
>+	 */
>+	void (*default_write_space)(struct sock *sk);

This is kind of virtio-vsock specific, why need to be here?

>
> 	/* Protected by lock_sock(sk) */
> 	u64 buffer_size;
>diff --git a/net/vmw_vsock/af_vsock.c b/net/vmw_vsock/af_vsock.c
>index 44cee74519555..fe7f49da6c4bc 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_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 ff74e805527a5..f4b5f0230493e 100644
>--- a/net/vmw_vsock/virtio_transport_common.c
>+++ b/net/vmw_vsock/virtio_transport_common.c
>@@ -1820,6 +1820,7 @@ virtio_transport_recv_pkt_batchable(struct virtio_transport *t,
> 	return sk->sk_state == TCP_ESTABLISHED &&
> 	       sk->sk_type == SOCK_STREAM &&
> 	       READ_ONCE(sk->sk_prot) == sk->sk_prot_creator &&
>+	       READ_ONCE(sk->sk_write_space) == vsk->default_write_space &&

Why?

> 	       !sock_flag(sk, SOCK_DONE) &&
> 	       vsk->transport == &t->transport &&
> 	       vvs && virtio_transport_rx_skb_has_headroom(vvs) &&
>@@ -1828,13 +1829,15 @@ 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.
>+ * locked for another ordinary STREAM/RW packet.  When @batch is non-NULL,
>+ * record a default write-space notification for batch finish instead of
>+ * delivering it immediately.  Return true if the caller must free @skb.
>  */
> static bool
> virtio_transport_recv_pkt_locked(struct virtio_transport *t,
> 				 struct sk_buff *skb, struct sock *sk,
>-				 bool *batchable)
>+				 bool *batchable,
>+				 struct virtio_transport_rx_batch *batch)
> {
> 	struct virtio_vsock_hdr *hdr = virtio_vsock_hdr(skb);
> 	struct sockaddr_vm src;
>@@ -1863,8 +1866,12 @@ virtio_transport_recv_pkt_locked(struct virtio_transport *t,
> 	if (vsk->local_addr.svm_cid != VMADDR_CID_ANY)
> 		vsk->local_addr.svm_cid = le64_to_cpu(hdr->dst_cid);
>
>-	if (space_available)
>-		sk->sk_write_space(sk);
>+	if (space_available) {
>+		if (batch)
>+			batch->write_space_pending = true;
>+		else
>+			sk->sk_write_space(sk);
>+	}
>
> 	switch (sk->sk_state) {
> 	case TCP_LISTEN:
>@@ -1931,7 +1938,7 @@ void virtio_transport_recv_pkt(struct virtio_transport *t,
> 	}
>
> 	lock_sock(sk);
>-	free_pkt = virtio_transport_recv_pkt_locked(t, skb, sk, NULL);
>+	free_pkt = virtio_transport_recv_pkt_locked(t, skb, sk, NULL, NULL);
> 	release_sock(sk);
>
> 	/* Release refcnt obtained when we fetched this socket out of the
>@@ -1949,22 +1956,27 @@ EXPORT_SYMBOL_GPL(virtio_transport_recv_pkt);
>
> /*
>  * Finish the RX batch.
>- * For a non-empty batch, the caller must hold the socket lock acquired with
>- * lock_sock(). This function releases the lock and the batch's lookup
>- * reference.
>+ * For a non-empty batch, the caller holds the socket lock acquired with
>+ * lock_sock(). Deliver any deferred write-space notification while the lock
>+ * is held, then release the lock and the batch's lookup reference.
>  * An empty batch is a no-op.
>  */
> void virtio_transport_rx_batch_finish(struct virtio_transport_rx_batch *batch)
> {
>+	bool write_space_pending = batch->write_space_pending;
> 	struct sock *sk = batch->sk;
>
> 	batch->sk = NULL;
> 	batch->pkts = 0;
> 	batch->bytes = 0;
>+	batch->write_space_pending = false;
>
> 	if (!sk)
> 		return;
>
>+	if (write_space_pending)
>+		vsock_sk(sk)->default_write_space(sk);
>+
> 	release_sock(sk);
> 	sock_put(sk);
> }
>@@ -2019,7 +2031,8 @@ void virtio_transport_recv_pkt_batch(struct virtio_transport *t,
> 			}
>
> 			free_pkt = virtio_transport_recv_pkt_locked(t, skb, sk,
>-								    &batchable);
>+								    &batchable,
>+								    batch);
> 			if (!batchable)
> 				virtio_transport_rx_batch_finish(batch);
> 			if (free_pkt)
>@@ -2059,23 +2072,31 @@ void virtio_transport_recv_pkt_batch(struct virtio_transport *t,
>
> 	lock_sock(sk);
> 	/*
>-	 * Sockmap removal restores the native protocol under sk_callback_lock.
>-	 * Serialize this initial eligibility check with that update.
>+	 * Sockmap teardown restores callbacks under sk_callback_lock. Keep the
>+	 * initial eligibility check consistent with that update.
> 	 */

Unrelated change?

BTW if you are assisted by AI, please read https://docs.kernel.org/process/coding-assistants.html 

> 	read_lock_bh(&sk->sk_callback_lock);
> 	start_batch = virtio_transport_recv_pkt_batchable(t, sk);
> 	read_unlock_bh(&sk->sk_callback_lock);
>
>+	if (start_batch)
>+		batch->sk = sk;
>+
> 	free_pkt = virtio_transport_recv_pkt_locked(t, skb, sk,
>-						    start_batch ? &batchable : NULL);
>+						    start_batch ? &batchable : NULL,
>+						    start_batch ? batch : NULL);
> 	if (start_batch && batchable) {
> 		/* Keep the lookup reference until the batch is released. */
>-		batch->sk = sk;
> 		return;
> 	}
>
>-	release_sock(sk);
>-	sock_put(sk);
>+	if (start_batch) {
>+		virtio_transport_rx_batch_finish(batch);
>+	} else {
>+		release_sock(sk);
>+		sock_put(sk);
>+	}
>+
> 	if (free_pkt)
> 		kfree_skb(skb);
> }
>-- 
>2.53.0
>

Re: [PATCH net-next v4 4/4] vsock/virtio: defer RX notifications until batch unlock

[Stefano Garzarella]

On Wed, Oct 14, 2026 at 07:56:40PM +0800, Jia Jia wrote:
>An RX batch can make a socket readable while the worker still holds its
>socket lock. Waking a blocking reader then only moves it to waiting for
>the same lock.
>
>Record the existing low-watermark condition and notify once after
>release_sock(). Move the coalesced write-space notification after the
>unlock as well, since both callbacks wake sk_sleep(sk). Deliver
>write-space before data-ready and keep the lookup reference until both
>callbacks complete.
>
>As with write-space, save the initial sk_data_ready callback and batch only
>while both saved callbacks remain installed. Sockets with replaced
>callbacks retain their per-packet behavior.
>
>The existing psock-marker check also covers a concurrent sockmap attach:
>at most the already admitted packet completes on the native path, and the
>next packet ends the batch.
>
>Signed-off-by: Jia Jia <physicalmtea@gmail.com>
>---
> include/linux/virtio_vsock.h            |  1 +
> include/net/af_vsock.h                  |  5 ++--
> net/vmw_vsock/af_vsock.c                |  1 +
> net/vmw_vsock/virtio_transport_common.c | 35 ++++++++++++++++++-------
> 4 files changed, 31 insertions(+), 11 deletions(-)
>
>diff --git a/include/linux/virtio_vsock.h b/include/linux/virtio_vsock.h
>index bf077ee0ded88..f1396ca3ad9b5 100644
>--- a/include/linux/virtio_vsock.h
>+++ b/include/linux/virtio_vsock.h
>@@ -288,6 +288,7 @@ struct virtio_transport_rx_batch {
> 	unsigned int pkts;
> 	size_t bytes;
> 	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 9a40960d10cbe..78bab5820f528 100644
>--- a/include/net/af_vsock.h
>+++ b/include/net/af_vsock.h
>@@ -64,9 +64,10 @@ struct vsock_sock {
> 	bool sent_request;
> 	bool ignore_connecting_rst;
> 	/*
>-	 * Initial callback, used to identify replacements and deliver deferred
>-	 * write-space notifications.
>+	 * Initial callbacks, used to identify replacements and deliver deferred
>+	 * notifications.
> 	 */
>+	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 fe7f49da6c4bc..f40659ec29df0 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 f4b5f0230493e..8b62b789b71c5 100644
>--- a/net/vmw_vsock/virtio_transport_common.c
>+++ b/net/vmw_vsock/virtio_transport_common.c
>@@ -1588,7 +1588,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);
>@@ -1607,7 +1608,13 @@ virtio_transport_recv_connected(struct sock *sk,
> 			vsock_remove_sock(vsk);
> 			break;
> 		}
>-		vsock_data_ready(sk);
>+		if (!data_ready_pending) {
>+			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);
>@@ -1820,6 +1827,7 @@ virtio_transport_recv_pkt_batchable(struct virtio_transport *t,
> 	return sk->sk_state == TCP_ESTABLISHED &&
> 	       sk->sk_type == SOCK_STREAM &&
> 	       READ_ONCE(sk->sk_prot) == sk->sk_prot_creator &&
>+	       READ_ONCE(sk->sk_data_ready) == vsk->default_data_ready &&
> 	       READ_ONCE(sk->sk_write_space) == vsk->default_write_space &&
> 	       !sock_flag(sk, SOCK_DONE) &&
> 	       vsk->transport == &t->transport &&
>@@ -1830,8 +1838,9 @@ 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.  When @batch is non-NULL,
>- * record a default write-space notification for batch finish instead of
>- * delivering it immediately.  Return true if the caller must free @skb.
>+ * record default write-space and data-ready notifications for batch finish
>+ * instead of delivering them immediately.  Return true if the caller must
>+ * free @skb.
>  */
> static bool
> virtio_transport_recv_pkt_locked(struct virtio_transport *t,
>@@ -1883,7 +1892,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,
>+						batch ?
>+						&batch->data_ready_pending : NULL);
> 		break;
> 	case TCP_CLOSING:
> 		virtio_transport_recv_disconnecting(sk, skb);
>@@ -1957,27 +1968,33 @@ EXPORT_SYMBOL_GPL(virtio_transport_recv_pkt);
> /*
>  * Finish the RX batch.
>  * For a non-empty batch, the caller holds the socket lock acquired with
>- * lock_sock(). Deliver any deferred write-space notification while the lock
>- * is held, then release the lock and the batch's lookup reference.
>- * An empty batch is a no-op.
>+ * lock_sock(). Release the lock, deliver any deferred write-space and
>+ * data-ready notifications in that order, then drop the batch's lookup
>+ * reference. An empty batch is a no-op.
>  */
> void virtio_transport_rx_batch_finish(struct virtio_transport_rx_batch *batch)
> {
> 	bool write_space_pending = batch->write_space_pending;
>+	bool data_ready_pending = batch->data_ready_pending;
> 	struct sock *sk = batch->sk;
>
> 	batch->sk = NULL;
> 	batch->pkts = 0;
> 	batch->bytes = 0;
> 	batch->write_space_pending = false;
>+	batch->data_ready_pending = false;
>
> 	if (!sk)
> 		return;
>
>+	release_sock(sk);
>+

Why in the previous patch we release the socket after calling 
default_write_space() and now we are moving before?

> 	if (write_space_pending)
> 		vsock_sk(sk)->default_write_space(sk);
>
>-	release_sock(sk);
>+	if (data_ready_pending)
>+		vsock_sk(sk)->default_data_ready(sk);
>+
> 	sock_put(sk);
> }
> EXPORT_SYMBOL_GPL(virtio_transport_rx_batch_finish);
>-- 
>2.53.0
>

Re: [PATCH net-next v4 1/4] vsock/virtio: split socket lookup from locked RX processing

[Jia Jia]

>
> On Wed, Oct 14, 2026 at 07:56:37PM +0800, Jia Jia wrote:
> >Split virtio_transport_recv_pkt() so socket lookup is separate from
> >receive state machine processing under the socket lock. This prepares the
> >receive path to keep a looked-up socket across multiple packets in a later
> >change.
> >
> >Keep virtio_transport_recv_pkt() on its existing per-packet locking path.
> >The locked helper obtains packet-specific addresses from the skb and uses
> >the matched socket namespace for reset replies. Preserve source validation
> >and failure handling.
> >
> >No functional change intended.
> >
> >Signed-off-by: Jia Jia <physicalmtea@gmail.com>
> >---
> > net/vmw_vsock/virtio_transport_common.c | 128 ++++++++++++++----------
> > 1 file changed, 75 insertions(+), 53 deletions(-)
> >
> >diff --git a/net/vmw_vsock/virtio_transport_common.c b/net/vmw_vsock/virtio_transport_common.c
> >index f225f53ed4bab..61ee3bc205d5d 100644
> >--- a/net/vmw_vsock/virtio_transport_common.c
> >+++ b/net/vmw_vsock/virtio_transport_common.c
> >@@ -1774,81 +1774,60 @@ static bool virtio_transport_valid_type(u16 type)
> >              (type == VIRTIO_VSOCK_TYPE_SEQPACKET);
> > }
> >
> >-/* We are under the virtio-vsock's vsock->rx_lock or vhost-vsock's vq->mutex
> >- * lock.
> >- */
> >-void virtio_transport_recv_pkt(struct virtio_transport *t,
> >-                             struct sk_buff *skb, struct net *net)
> >+static struct sock *
> >+virtio_transport_find_socket(struct sk_buff *skb, struct sockaddr_vm *src,
> >+                           struct sockaddr_vm *dst, struct net *net)
> > {
> >       struct virtio_vsock_hdr *hdr = virtio_vsock_hdr(skb);
> >-      struct sockaddr_vm src, dst;
> >-      struct vsock_sock *vsk;
> >       struct sock *sk;
> >-      bool space_available;
> >-
> >-      vsock_addr_init(&src, le64_to_cpu(hdr->src_cid),
> >-                      le32_to_cpu(hdr->src_port));
> >-      vsock_addr_init(&dst, le64_to_cpu(hdr->dst_cid),
> >-                      le32_to_cpu(hdr->dst_port));
> >-
> >-      trace_virtio_transport_recv_pkt(src.svm_cid, src.svm_port,
> >-                                      dst.svm_cid, dst.svm_port,
> >-                                      le32_to_cpu(hdr->len),
> >-                                      le16_to_cpu(hdr->type),
> >-                                      le16_to_cpu(hdr->op),
> >-                                      le32_to_cpu(hdr->flags),
> >-                                      le32_to_cpu(hdr->buf_alloc),
> >-                                      le32_to_cpu(hdr->fwd_cnt));
> >
> >-      if (!virtio_transport_valid_type(le16_to_cpu(hdr->type))) {
> >-              (void)virtio_transport_reset_no_sock(t, skb, net);
> >-              goto free_pkt;
> >-      }
> >+      if (!virtio_transport_valid_type(le16_to_cpu(hdr->type)))
> >+              return NULL;
> >
> >-      /* The socket must be in connected or bound table
> >-       * otherwise send reset back
> >-       */
> >-      sk = vsock_find_connected_socket_net(&src, &dst, net);
> >-      if (!sk) {
> >-              sk = vsock_find_bound_socket_net(&dst, net);
> >-              if (!sk) {
> >-                      (void)virtio_transport_reset_no_sock(t, skb, net);
> >-                      goto free_pkt;
> >-              }
> >-      }
> >+      sk = vsock_find_connected_socket_net(src, dst, net);
> >+      if (!sk)
> >+              sk = vsock_find_bound_socket_net(dst, net);
> >+      if (!sk)
> >+              return NULL;
> >
> >       if (virtio_transport_get_type(sk) != le16_to_cpu(hdr->type)) {
> >-              (void)virtio_transport_reset_no_sock(t, skb, net);
> >               sock_put(sk);
> >-              goto free_pkt;
> >+              return NULL;
> >       }
> >
> >-      if (!skb_set_owner_sk_safe(skb, sk)) {
> >-              WARN_ONCE(1, "receiving vsock socket has sk_refcnt == 0\n");
> >-              goto free_pkt;
> >-      }
> >+      return sk;
> >+}
> >
> >-      vsk = vsock_sk(sk);
> >+/*
> >+ * The caller holds sk's socket lock and must free skb if this returns true.
>
> Why the skb must be freed by the caller and not by this function itself?
>

You're right, it doesn't have to be. I originally left the skb cleanup
to the caller so it could unwind either per-packet or batch state
while preserving
the existing error-path order (release_sock() -> drop the lookup
reference -> kfree_skb()).

However, that distinction does not require skb ownership to escape the
helper, and the resulting ownership contract is confusing since the
other paths
consume the skb internally. I'll make the helper consume it directly in v5.


> >+ */
> >+static bool
> >+virtio_transport_recv_pkt_locked(struct virtio_transport *t,
> >+                               struct sk_buff *skb, struct sock *sk)
> >+{
> >+      struct virtio_vsock_hdr *hdr = virtio_vsock_hdr(skb);
> >+      struct sockaddr_vm src;
> >+      struct vsock_sock *vsk = vsock_sk(sk);
> >+      bool space_available;
>
> Please, follow the netdev convection for ordering local variables:
> https://www.kernel.org/doc/html/v7.2/process/maintainer-netdev.html#local-variable-ordering-reverse-xmas-tree-rcs
>

Sorry about that. You had pointed me to this guideline before, and I
missed this during the final preparation. That's entirely my
oversight, and I will fix the declaration ordering throughout the
series in v5.


> >
> >-      lock_sock(sk);
> >+      vsock_addr_init(&src, le64_to_cpu(hdr->src_cid),
> >+                      le32_to_cpu(hdr->src_port));
> >
> >-      /* Check if sk has been closed or assigned to another transport before
> >-       * lock_sock (note: listener sockets are not assigned to any transport)
> >+      /* Listener sockets accept packets from any source and are not assigned to
> >+       * a transport.
> >        */
>
> Why changing this comment in this commit?
>
> Oh because we moved out lock_sock, right?
>
> I think we should keep in some way the old info that was much more
> informative that the current comment that just say what the check is
> doing and not why.
>

Yes, I changed it because lock_sock() moved to the caller, so the old
wording no longer read clearly at that location. But I over-shortened
it and dropped the important reason for the recheck. I'll restore that
explanation:


> >       if (sock_flag(sk, SOCK_DONE) ||
> >           (sk->sk_state != TCP_LISTEN &&
> >            !vsock_check_source(vsk, &t->transport, &src))) {
> >-              (void)virtio_transport_reset_no_sock(t, skb, net);
> >-              release_sock(sk);
> >-              sock_put(sk);
> >-              goto free_pkt;
> >+              (void)virtio_transport_reset_no_sock(t, skb, sock_net(sk));
>
> We need to be careful here, IIRC Bobby used the net from the caller for
> a reason.
>
> Are you sure it's fine this change? Why not passing net as parameter and
> minimize changes?
>

I assumed sock_net(sk) was interchangeable with the caller's net after
a successful lookup, but I took that equivalence for granted. I'll
change this in v5.

> >+              return true;
> >       }
> >
> >       space_available = virtio_transport_space_update(sk, skb);
> >
> >       /* Update CID in case it has changed after a transport reset event */
> >       if (vsk->local_addr.svm_cid != VMADDR_CID_ANY)
> >-              vsk->local_addr.svm_cid = dst.svm_cid;
> >+              vsk->local_addr.svm_cid = le64_to_cpu(hdr->dst_cid);
> >
> >       if (space_available)
> >               sk->sk_write_space(sk);
> >@@ -1870,17 +1849,60 @@ void virtio_transport_recv_pkt(struct virtio_transport *t,
> >               kfree_skb(skb);
> >               break;
> >       default:
> >-              (void)virtio_transport_reset_no_sock(t, skb, net);
> >+              (void)virtio_transport_reset_no_sock(t, skb, sock_net(sk));
> >               kfree_skb(skb);
> >               break;
> >       }
> >
> >+      return false;
> >+}
> >+
> >+/* We are under the virtio-vsock's vsock->rx_lock or vhost-vsock's vq->mutex
> >+ * lock.
> >+ */
> >+void virtio_transport_recv_pkt(struct virtio_transport *t,
> >+                             struct sk_buff *skb, struct net *net)
> >+{
> >+      struct virtio_vsock_hdr *hdr = virtio_vsock_hdr(skb);
> >+      struct sockaddr_vm src, dst;
> >+      struct sock *sk;
> >+      bool free_pkt;
> >+
> >+      vsock_addr_init(&src, le64_to_cpu(hdr->src_cid),
> >+                      le32_to_cpu(hdr->src_port));
> >+      vsock_addr_init(&dst, le64_to_cpu(hdr->dst_cid),
> >+                      le32_to_cpu(hdr->dst_port));
> >+
> >+      trace_virtio_transport_recv_pkt(src.svm_cid, src.svm_port,
> >+                                      dst.svm_cid, dst.svm_port,
> >+                                      le32_to_cpu(hdr->len),
> >+                                      le16_to_cpu(hdr->type),
> >+                                      le16_to_cpu(hdr->op),
> >+                                      le32_to_cpu(hdr->flags),
> >+                                      le32_to_cpu(hdr->buf_alloc),
> >+                                      le32_to_cpu(hdr->fwd_cnt));
> >+
> >+      sk = virtio_transport_find_socket(skb, &src, &dst, net);
> >+      if (!sk) {
> >+              (void)virtio_transport_reset_no_sock(t, skb, net);
> >+              goto free_pkt;
> >+      }
> >+
> >+      if (!skb_set_owner_sk_safe(skb, sk)) {
> >+              WARN_ONCE(1, "receiving vsock socket has sk_refcnt == 0\n");
> >+              goto free_pkt;
> >+      }
> >+
> >+      lock_sock(sk);
> >+      free_pkt = virtio_transport_recv_pkt_locked(t, skb, sk);
> >       release_sock(sk);
> >
> >       /* Release refcnt obtained when we fetched this socket out of the
> >        * bound or connected list.
> >        */
> >       sock_put(sk);
> >+      if (free_pkt)
> >+              kfree_skb(skb);
> >       return;
> >
> > free_pkt:
> >--
> >2.53.0
> >
>

Re: [PATCH net-next v4 2/4] vsock/virtio: amortize RX socket locking for stream packets

[Jia Jia]

>
> On Wed, Oct 14, 2026 at 07:56:38PM +0800, Jia Jia wrote:
> >The virtio-vsock RX worker currently looks up the socket and takes its
> >lock for every packet. Keep the socket lock and lookup reference across a
> >bounded run of STREAM/RW packets for the same established socket.
> >
> >Reuse the locked socket while packets continue to match the same flow.
> >End the batch for a different flow, non-RW traffic, malformed packets, or
> >a change in socket state, transport, protocol, or receive-buffer headroom.
> >A connected-table miss uses the existing receive path.
> >
> >The reused path still traces every packet, takes an skb owner reference,
> >validates the source and transport, updates credit, and runs the existing
> >receive state machine.
> >
> >Sockmap can replace the protocol without taking the socket lock. Check the
> >psock marker at each packet boundary and serialize initial admission with
> >sk_callback_lock, so a concurrent attach ends batching at the next packet.
> >
> >Cap a batch at 64 packets and 64K of payload. The packet limit bounds
>
> Why this cap?
>

The 64 KiB limit was chosen to bound one uninterrupted socket-lock
ownership interval to one VIRTIO_VSOCK_MAX_PKT_BUF_SIZE worth of
payload. For small or zero-length packets, the byte limit alone would
not bound the per-packet work, so a separate packet limit is needed. I
chose 64 packets as an initial batch size, using VHOST_NET_BATCH as a
reference. It is a policy choice rather than a protocol requirement.

Beyond these limits, a larger batch would only further amortize the
socket lookup and lock/unlock operations, while the virtqueue,
validation, tracing, skb ownership, credit update, and queueing work
remains per-packet. I therefore expect diminishing returns while the
uninterrupted socket-lock interval continues to grow, but I have not
measured larger limits yet.

> >per-packet work, while the byte limit bounds how much data can queue before
> >a reader acquires the socket lock. Batching remains in the process-context
> >RX worker.
>
> IMHO this patch should carry performance results.
>

Sure, I'll include the relevant Patch 1 - Patch 2 cumulative
performance results in this commit message in v5.

> >
> >Signed-off-by: Jia Jia <physicalmtea@gmail.com>
> >---
> > include/linux/virtio_vsock.h            |  11 ++
> > net/vmw_vsock/virtio_transport.c        |  29 +++-
>
> Why no changes in vhost-vsock or vsock_loopback?
>
> I'm not asking to do that, I'd like to understand why only in the
> driver.
>

I have not yet measured the corresponding worker and socket-lock
overhead in vhost-vsock or vsock_loopback, and they have different
worker and locking models, so I left those paths unchanged. I plan to
evaluate vhost-vsock separately as follow-up work.


> > net/vmw_vsock/virtio_transport_common.c | 183 +++++++++++++++++++++++-
> > 3 files changed, 216 insertions(+), 7 deletions(-)
> >
> >diff --git a/include/linux/virtio_vsock.h b/include/linux/virtio_vsock.h
> >index f91704731057e..26dde6909c4ed 100644
> >--- a/include/linux/virtio_vsock.h
> >+++ b/include/linux/virtio_vsock.h
> >@@ -282,6 +282,17 @@ void virtio_transport_destruct(struct vsock_sock *vsk);
> >
> > void virtio_transport_recv_pkt(struct virtio_transport *t,
> >                              struct sk_buff *skb, struct net *net);
> >+
> >+struct virtio_transport_rx_batch {
> >+      struct sock *sk;
> >+      unsigned int pkts;
> >+      size_t bytes;
> >+};
> >+
> >+void virtio_transport_recv_pkt_batch(struct virtio_transport *t,
> >+                                   struct sk_buff *skb, struct net *net,
> >+                                   struct virtio_transport_rx_batch *batch);
> >+void virtio_transport_rx_batch_finish(struct virtio_transport_rx_batch *batch);
> > void virtio_transport_inc_tx_pkt(struct virtio_vsock_sock *vvs, struct sk_buff *skb);
> > u32 virtio_transport_get_credit(struct virtio_vsock_sock *vvs, u32 wanted);
> > void virtio_transport_put_credit(struct virtio_vsock_sock *vvs, u32 credit);
> >diff --git a/net/vmw_vsock/virtio_transport.c b/net/vmw_vsock/virtio_transport.c
> >index 4f9aa9c4c3aa5..ef076732158d4 100644
> >--- a/net/vmw_vsock/virtio_transport.c
> >+++ b/net/vmw_vsock/virtio_transport.c
> >@@ -629,8 +629,16 @@ virtio_transport_seqpacket_allow(struct vsock_sock *vsk, u32 remote_cid)
> >       return seqpacket_allow;
> > }
> >
> >+/*
> >+ * Keep a bounded run of packets for one socket under a single socket lock.
> >+ * Limit packet count and payload size to bound the work done while locked.
>
> Here we should explain why we choose these values.
>

I'll update the comment here in v5 to explain why the 64-packet and 64
KiB thresholds were chosen.

> >+ */
> >+#define VIRTIO_TRANSPORT_RX_BATCH_MAX_PKTS    64
> >+#define VIRTIO_TRANSPORT_RX_BATCH_MAX_BYTES   (64 * 1024)
> >+
> > static void virtio_transport_rx_work(struct work_struct *work)
> > {
> >+      struct virtio_transport_rx_batch batch = {};
> >       struct virtio_vsock *vsock =
> >               container_of(work, struct virtio_vsock, rx_work);
> >       struct virtqueue *vq;
> >@@ -666,6 +674,7 @@ static void virtio_transport_rx_work(struct work_struct *work)
> >                       /* Drop short/long packets */
> >                       if (unlikely(len < sizeof(*hdr) ||
> >                                    len > virtio_vsock_skb_len(skb))) {
> >+                              virtio_transport_rx_batch_finish(&batch);
> >                               kfree_skb(skb);
> >                               continue;
> >                       }
> >@@ -673,6 +682,7 @@ static void virtio_transport_rx_work(struct work_struct *work)
> >                       hdr = virtio_vsock_hdr(skb);
> >                       payload_len = le32_to_cpu(hdr->len);
> >                       if (unlikely(payload_len > len - sizeof(*hdr))) {
> >+                              virtio_transport_rx_batch_finish(&batch);
> >                               kfree_skb(skb);
> >                               continue;
> >                       }
> >@@ -680,16 +690,33 @@ static void virtio_transport_rx_work(struct work_struct *work)
> >                       if (payload_len)
> >                               virtio_vsock_skb_put(skb, payload_len);
> >
> >+                      if (batch.sk &&
> >+                          payload_len > VIRTIO_TRANSPORT_RX_BATCH_MAX_BYTES -
> >+                                        batch.bytes) {
>
> Why this?
>
>

This pre-check ends the current batch before admitting a packet whose
payload would push it past the byte limit. The post-processing check
can end the
batch only after that packet has already been processed, so it cannot
enforce the limit by itself. I'll encapsulate and document this in the
common
batching code.

> >+                              virtio_transport_rx_batch_finish(&batch);
> >+                      }
> >+
> >                       virtio_transport_deliver_tap_pkt(skb);
> >
> >                       /* Force virtio-transport into global mode since it
> >                        * does not yet support local-mode namespacing.
> >                        */
> >-                      virtio_transport_recv_pkt(&virtio_transport, skb, NULL);
> >+                      virtio_transport_recv_pkt_batch(&virtio_transport, skb, NULL,
> >+                                                      &batch);
> >+
> >+                      if (batch.sk) {
> >+                              batch.pkts++;
> >+                              batch.bytes += payload_len;
> >+                              if (batch.pkts >= VIRTIO_TRANSPORT_RX_BATCH_MAX_PKTS ||
> >+                                  batch.bytes >= VIRTIO_TRANSPORT_RX_BATCH_MAX_BYTES) {
> >+                                      virtio_transport_rx_batch_finish(&batch);
> >+                              }
> >+                      }
>
> Why this is not in the common part?
>

I put this in the driver because it owns the RX loop and its work
budget. But I agree that having the caller manipulate the batch
counters and active
state makes the interface harder to follow. I'll move the accounting
and limit checks into the common batching code in v5.

> >               }
> >       } while (!virtqueue_enable_cb(vq));
> >
> > out:
> >+      virtio_transport_rx_batch_finish(&batch);
> >       if (vsock->rx_buf_nr < vsock->rx_buf_max_nr / 2)
> >               virtio_vsock_rx_fill(vsock);
> > out_nofill:
> >diff --git a/net/vmw_vsock/virtio_transport_common.c b/net/vmw_vsock/virtio_transport_common.c
> >index 61ee3bc205d5d..ff74e805527a5 100644
> >--- a/net/vmw_vsock/virtio_transport_common.c
> >+++ b/net/vmw_vsock/virtio_transport_common.c
> >@@ -576,11 +576,16 @@ virtio_transport_collapse_rx_queue(struct virtio_vsock_sock *vvs,
> >       skb_queue_splice(&new_queue, &vvs->rx_queue);
> > }
> >
> >+static bool
> >+virtio_transport_rx_skb_has_headroom(struct virtio_vsock_sock *vvs)
> >+{
> >+      return (u64)(skb_queue_len(&vvs->rx_queue) + 1) * SKB_TRUESIZE(0) <=
> >+              vvs->buf_alloc;
> >+}
>
> Please move this change in a separate patch.
>
> skb_has_headroom() is confusing IMO, maybe
> virtio_transport_rx_buf_full(), but reversing the return value, so
> keeping the same check as before.
>

Sure, I'll move this into a separate preparatory patch

> >+
> > static bool virtio_transport_inc_rx_pkt(struct virtio_vsock_sock *vvs,
> >                                       u32 len)
> > {
> >-      u64 skb_overhead = (skb_queue_len(&vvs->rx_queue) + 1) * SKB_TRUESIZE(0);
> >-
> >       /* Allow at most buf_alloc * 2 total budget (payload + overhead),
> >        * similar to how SO_RCVBUF is doubled to reserve space for sk_buff
> >        * metadata. Check payload against buf_alloc to be sure the other
> >@@ -588,7 +593,7 @@ static bool virtio_transport_inc_rx_pkt(struct virtio_vsock_sock *vvs,
> >        * queue growth.
> >        */
> >       if ((u64)vvs->buf_used + len > vvs->buf_alloc ||
> >-          skb_overhead > vvs->buf_alloc)
> >+          !virtio_transport_rx_skb_has_headroom(vvs))
> >               return false;
> >
> >       vvs->rx_bytes += len;
> >@@ -1798,12 +1803,38 @@ virtio_transport_find_socket(struct sk_buff *skb, struct sockaddr_vm *src,
> >       return sk;
> > }
> >
> >+static bool
> >+virtio_transport_sock_has_psock(const struct sock *sk)
> >+{
> >+      return (uintptr_t)rcu_access_pointer(__sk_user_data(sk)) &
> >+             SK_USER_DATA_PSOCK;
> >+}
> >+
> >+static bool
> >+virtio_transport_recv_pkt_batchable(struct virtio_transport *t,
> >+                                  struct sock *sk)
> >+{
> >+      struct vsock_sock *vsk = vsock_sk(sk);
> >+      struct virtio_vsock_sock *vvs = vsk->trans;
> >+
>
> A comment here to explain why some checks like...
>

I'll add why-focused comments for the condition groups in v5.

> >+      return sk->sk_state == TCP_ESTABLISHED &&
> >+             sk->sk_type == SOCK_STREAM &&
> >+             READ_ONCE(sk->sk_prot) == sk->sk_prot_creator &&
> ... this       ^
>
> I mean in general to explain better these conditions (I mean not a
> description of what we are doing, but why we are doing these checks).
>

I'll add comments explaining these reasons.

> >+             !sock_flag(sk, SOCK_DONE) &&
> >+             vsk->transport == &t->transport &&
> >+             vvs && virtio_transport_rx_skb_has_headroom(vvs) &&
> >+             !virtio_transport_sock_has_psock(sk);
> >+}
> >+
> > /*
> >- * The caller holds sk's socket lock and must free skb if this returns true.
> >+ * 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.
> >  */
> > static bool
> > virtio_transport_recv_pkt_locked(struct virtio_transport *t,
> >-                               struct sk_buff *skb, struct sock *sk)
> >+                               struct sk_buff *skb, struct sock *sk,
> >+                               bool *batchable)
> > {
> >       struct virtio_vsock_hdr *hdr = virtio_vsock_hdr(skb);
> >       struct sockaddr_vm src;
> >@@ -1813,6 +1844,9 @@ virtio_transport_recv_pkt_locked(struct virtio_transport *t,
> >       vsock_addr_init(&src, le64_to_cpu(hdr->src_cid),
> >                       le32_to_cpu(hdr->src_port));
> >
> >+      if (batchable)
> >+              *batchable = false;
> >+
> >       /* Listener sockets accept packets from any source and are not assigned to
> >        * a transport.
> >        */
> >@@ -1854,6 +1888,9 @@ virtio_transport_recv_pkt_locked(struct virtio_transport *t,
> >               break;
> >       }
> >
> >+      if (batchable)
> >+              *batchable = virtio_transport_recv_pkt_batchable(t, sk);
> >+
> >       return false;
> > }
> >
> >@@ -1894,7 +1931,7 @@ void virtio_transport_recv_pkt(struct virtio_transport *t,
> >       }
> >
> >       lock_sock(sk);
> >-      free_pkt = virtio_transport_recv_pkt_locked(t, skb, sk);
> >+      free_pkt = virtio_transport_recv_pkt_locked(t, skb, sk, NULL);
> >       release_sock(sk);
> >
> >       /* Release refcnt obtained when we fetched this socket out of the
> >@@ -1910,6 +1947,140 @@ void virtio_transport_recv_pkt(struct virtio_transport *t,
> > }
> > EXPORT_SYMBOL_GPL(virtio_transport_recv_pkt);
> >
> >+/*
> >+ * Finish the RX batch.
> >+ * For a non-empty batch, the caller must hold the socket lock acquired with
> >+ * lock_sock(). This function releases the lock and the batch's lookup
> >+ * reference.
> >+ * An empty batch is a no-op.
>
> What empty batch means, I mean how the caller can understand that a
> batch is empty?
>
> Maybe we should provide a function for that...
>

I'll add virtio_transport_rx_batch_empty() and document the
associated socket-lock and lookup-reference ownership.

> >+ */
> >+void virtio_transport_rx_batch_finish(struct virtio_transport_rx_batch *batch)
> >+{
> >+      struct sock *sk = batch->sk;
> >+
> >+      batch->sk = NULL;
> >+      batch->pkts = 0;
> >+      batch->bytes = 0;
> >+
> >+      if (!sk)
> >+              return;
> >+
> >+      release_sock(sk);
> >+      sock_put(sk);
> >+}
> >+EXPORT_SYMBOL_GPL(virtio_transport_rx_batch_finish);
> >+
> >+void virtio_transport_recv_pkt_batch(struct virtio_transport *t,
> >+                                   struct sk_buff *skb, struct net *net,
> >+                                   struct virtio_transport_rx_batch *batch)
> >+{
> >+      struct virtio_vsock_hdr *hdr = virtio_vsock_hdr(skb);
> >+      struct sockaddr_vm src, dst;
> >+      struct vsock_sock *vsk;
> >+      bool batchable, start_batch;
>
> Ditto about declarations order, please check in the entire series.
>

Right, I missed this one too. I'll check and fix the declaration
ordering throughout the entire series in v5.

> >+      struct sock *sk;
> >+      bool free_pkt;
> >+
> >+      /* Only STREAM/RW packets can share a socket lock. */
>
> Why?
>
> Should the caller holds the lock for the entire batch?
> Should this be documented?
>

Yes. Once a batch is started, the receive helper returns with the
socket lock and lookup reference held, and the same caller retains
them until it calls the finish helper. That ownership contract is not
clear enough in the current interface, and I'll document it in the
public API in v5.


> >+      if (le16_to_cpu(hdr->type) != VIRTIO_VSOCK_TYPE_STREAM ||
> >+          le16_to_cpu(hdr->op) != VIRTIO_VSOCK_OP_RW) {
> >+              virtio_transport_rx_batch_finish(batch);
> >+              virtio_transport_recv_pkt(t, skb, net);
> >+              return;
> >+      }
> >+
> >+      vsock_addr_init(&src, le64_to_cpu(hdr->src_cid),
> >+                      le32_to_cpu(hdr->src_port));
> >+      vsock_addr_init(&dst, le64_to_cpu(hdr->dst_cid),
> >+                      le32_to_cpu(hdr->dst_port));
> >+
> >+      if (batch->sk) {
> >+              vsk = vsock_sk(batch->sk);
> >+              if (vsock_addr_equals_addr(&src, &vsk->remote_addr) &&
> >+                  dst.svm_port == vsk->local_addr.svm_port &&
> >+                  vsock_net_check_mode(sock_net(batch->sk), net) &&
> >+                  virtio_transport_recv_pkt_batchable(t, batch->sk)) {
> >+                      sk = batch->sk;
> >+                      trace_virtio_transport_recv_pkt(src.svm_cid,
> >+                                                      src.svm_port,
> >+                                                      dst.svm_cid,
> >+                                                      dst.svm_port,
> >+                                                      le32_to_cpu(hdr->len),
> >+                                                      le16_to_cpu(hdr->type),
> >+                                                      le16_to_cpu(hdr->op),
> >+                                                      le32_to_cpu(hdr->flags),
> >+                                                      le32_to_cpu(hdr->buf_alloc),
> >+                                                      le32_to_cpu(hdr->fwd_cnt));
> >+                      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);
> >+                              kfree_skb(skb);
> >+                              return;
> >+                      }
> >+
>
> This code is unreadable TBH, there isn't a blank line till here...
> and too much nested blocks.
>
> Maybe you should move this in a function, and use a bit more blank lines.
>
>

I'll move the active-batch path into a helper, reduce the nesting with
early returns, and add blank lines between the logical steps.


> >+                      free_pkt = virtio_transport_recv_pkt_locked(t, skb, sk,
> >+                                                                  &batchable);
> >+                      if (!batchable)
> >+                              virtio_transport_rx_batch_finish(batch);
> >+                      if (free_pkt)
> >+                              kfree_skb(skb);
> >+                      return;
> >+              }
> >+
> >+              virtio_transport_rx_batch_finish(batch);
> >+      }
> >+
> >+      sk = vsock_find_connected_socket_net(&src, &dst, net);
> >+      if (!sk) {
> >+              virtio_transport_recv_pkt(t, skb, net);
>
> Why this?
>

If you mean the fallback to virtio_transport_recv_pkt(), my
understanding of the lookup semantics is that it is needed because the
batching fast path only
looks in the connected table, as only an established connected socket
is eligible for batching. A miss there can still match a bound socket,
so the packet
must follow the existing receive path rather than being treated as
having no socket. Please correct me if I have misunderstood this.

> >+              return;
> >+      }
> >+
> >+      if (virtio_transport_get_type(sk) != le16_to_cpu(hdr->type)) {
> >+              sock_put(sk);
> >+              virtio_transport_recv_pkt(t, skb, net);
> >+              return;
> >+      }
> >+
> >+      trace_virtio_transport_recv_pkt(src.svm_cid, src.svm_port,
> >+                                      dst.svm_cid, dst.svm_port,
> >+                                      le32_to_cpu(hdr->len),
> >+                                      le16_to_cpu(hdr->type),
> >+                                      le16_to_cpu(hdr->op),
> >+                                      le32_to_cpu(hdr->flags),
> >+                                      le32_to_cpu(hdr->buf_alloc),
> >+                                      le32_to_cpu(hdr->fwd_cnt));
> >+
> >+      if (!skb_set_owner_sk_safe(skb, sk)) {
> >+              WARN_ONCE(1, "receiving vsock socket has sk_refcnt == 0\n");
> >+              kfree_skb(skb);
> >+              return;
> >+      }
>
> We have this code duplicated in several place now, can this be avoided?
>

Yes. I'll extract the trace and skb owner setup into a small shared helper.

> >+
> >+      lock_sock(sk);
> >+      /*
> >+       * Sockmap removal restores the native protocol under sk_callback_lock.
> >+       * Serialize this initial eligibility check with that update.
> >+       */
> >+      read_lock_bh(&sk->sk_callback_lock);
> >+      start_batch = virtio_transport_recv_pkt_batchable(t, sk);
> >+      read_unlock_bh(&sk->sk_callback_lock);
> >+
> >+      free_pkt = virtio_transport_recv_pkt_locked(t, skb, sk,
> >+                                                  start_batch ? &batchable : NULL);
> >+      if (start_batch && batchable) {
> >+              /* Keep the lookup reference until the batch is released. */
> >+              batch->sk = sk;
> >+              return;
> >+      }
>
> Ah, this is for the first run, when the batch is empty, right?
>
> Can you merge in some way both branches?
>
> BTW please add some more useful comments (explaining why)
>

Yes. I'll merge the common parts of the two paths and add comments
explaining the remaining differences and why they are necessary.


> >+
> >+      release_sock(sk);
> >+      sock_put(sk);
> >+      if (free_pkt)
> >+              kfree_skb(skb);
> >+}
> >+EXPORT_SYMBOL_GPL(virtio_transport_recv_pkt_batch);
> >+
> > /* Remove skbs found in a queue that have a vsk that matches.
> >  *
> >  * Each skb is freed.
> >--
> >2.53.0
> >
>

Re: [PATCH net-next v4 3/4] vsock/virtio: coalesce RX write-space notifications in lock batches

[Jia Jia]

>
> On Wed, Oct 14, 2026 at 07:56:39PM +0800, Jia Jia wrote:
> >Each received packet updates peer credit and may call sk_write_space().
> >A writer cannot use that credit until the batch releases the socket lock,
> >so one notification at batch finish is sufficient.
> >
> >Save the initial sk_write_space callback when the AF_VSOCK socket is
> >created and require it to remain installed for batching. This lets the
> >transport module deliver the deferred notification without referring to
> >the unexported default callback. Sockets with replaced callbacks retain
> >their per-packet behavior.
> >
> >Record the batch socket before processing its first packet so a batch that
> >ends on that packet cannot lose the pending notification.
>
> mmm, is this a fix of the previous patch?
>

No. Patch 2 preserves the existing per-packet sk_write_space()
behavior, so Patch 3 is not a correctness fix for it. Patch 3 only
coalesces the native/ default notifications.

Moving batch->sk earlier is required only after the deferred state is
introduced here: the socket must already be recorded before the first
packet can set write_space_pending, so the notification can still be
delivered if that packet immediately ends the batch. Sorry, the commit
message did not make this distinction clear.

> >
> >Signed-off-by: Jia Jia <physicalmtea@gmail.com>
> >---
> > include/linux/virtio_vsock.h            |  1 +
> > include/net/af_vsock.h                  |  5 +++
> > net/vmw_vsock/af_vsock.c                |  1 +
> > net/vmw_vsock/virtio_transport_common.c | 53 +++++++++++++++++--------
> > 4 files changed, 44 insertions(+), 16 deletions(-)
> >
> >diff --git a/include/linux/virtio_vsock.h b/include/linux/virtio_vsock.h
> >index 26dde6909c4ed..bf077ee0ded88 100644
> >--- a/include/linux/virtio_vsock.h
> >+++ b/include/linux/virtio_vsock.h
> >@@ -287,6 +287,7 @@ struct virtio_transport_rx_batch {
> >       struct sock *sk;
> >       unsigned int pkts;
> >       size_t bytes;
> >+      bool write_space_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 5549298c1ec6b..9a40960d10cbe 100644
> >--- a/include/net/af_vsock.h
> >+++ b/include/net/af_vsock.h
> >@@ -63,6 +63,11 @@ struct vsock_sock {
> >       u32 peer_shutdown;
> >       bool sent_request;
> >       bool ignore_connecting_rst;
> >+      /*
> >+       * Initial callback, used to identify replacements and deliver deferred
> >+       * write-space notifications.
> >+       */
> >+      void (*default_write_space)(struct sock *sk);
>
> This is kind of virtio-vsock specific, why need to be here?
>

I stored this state here because __vsock_create() can capture the
native callback before the socket is exposed or any callback
replacement occurs, and the transport cannot directly reference the
unexported default callback. However, I agree that exposing this
virtio-specific state directly in struct vsock_sock is not ideal. I'll
look at hiding it behind an AF_VSOCK helper in v5 while preserving the
capture point and the lifetime required by deferred callbacks.


> >
> >       /* Protected by lock_sock(sk) */
> >       u64 buffer_size;
> >diff --git a/net/vmw_vsock/af_vsock.c b/net/vmw_vsock/af_vsock.c
> >index 44cee74519555..fe7f49da6c4bc 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_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 ff74e805527a5..f4b5f0230493e 100644
> >--- a/net/vmw_vsock/virtio_transport_common.c
> >+++ b/net/vmw_vsock/virtio_transport_common.c
> >@@ -1820,6 +1820,7 @@ virtio_transport_recv_pkt_batchable(struct virtio_transport *t,
> >       return sk->sk_state == TCP_ESTABLISHED &&
> >              sk->sk_type == SOCK_STREAM &&
> >              READ_ONCE(sk->sk_prot) == sk->sk_prot_creator &&
> >+             READ_ONCE(sk->sk_write_space) == vsk->default_write_space &&
>
> Why?
>

I believe this check is necessary because notification coalescing is
only safe for the original native sk_write_space callback. Sockmap can
replace this callback under sk_callback_lock, which is independent of
the socket lock held across the batch, and the replacement may require
per-packet handling. Comparing it with the callback captured at socket
creation lets us fall back to the existing per-packet behavior when it
has been replaced. I'll add a comment explaining this.


> >              !sock_flag(sk, SOCK_DONE) &&
> >              vsk->transport == &t->transport &&
> >              vvs && virtio_transport_rx_skb_has_headroom(vvs) &&
> >@@ -1828,13 +1829,15 @@ 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.
> >+ * locked for another ordinary STREAM/RW packet.  When @batch is non-NULL,
> >+ * record a default write-space notification for batch finish instead of
> >+ * delivering it immediately.  Return true if the caller must free @skb.
> >  */
> > static bool
> > virtio_transport_recv_pkt_locked(struct virtio_transport *t,
> >                                struct sk_buff *skb, struct sock *sk,
> >-                               bool *batchable)
> >+                               bool *batchable,
> >+                               struct virtio_transport_rx_batch *batch)
> > {
> >       struct virtio_vsock_hdr *hdr = virtio_vsock_hdr(skb);
> >       struct sockaddr_vm src;
> >@@ -1863,8 +1866,12 @@ virtio_transport_recv_pkt_locked(struct virtio_transport *t,
> >       if (vsk->local_addr.svm_cid != VMADDR_CID_ANY)
> >               vsk->local_addr.svm_cid = le64_to_cpu(hdr->dst_cid);
> >
> >-      if (space_available)
> >-              sk->sk_write_space(sk);
> >+      if (space_available) {
> >+              if (batch)
> >+                      batch->write_space_pending = true;
> >+              else
> >+                      sk->sk_write_space(sk);
> >+      }
> >
> >       switch (sk->sk_state) {
> >       case TCP_LISTEN:
> >@@ -1931,7 +1938,7 @@ void virtio_transport_recv_pkt(struct virtio_transport *t,
> >       }
> >
> >       lock_sock(sk);
> >-      free_pkt = virtio_transport_recv_pkt_locked(t, skb, sk, NULL);
> >+      free_pkt = virtio_transport_recv_pkt_locked(t, skb, sk, NULL, NULL);
> >       release_sock(sk);
> >
> >       /* Release refcnt obtained when we fetched this socket out of the
> >@@ -1949,22 +1956,27 @@ EXPORT_SYMBOL_GPL(virtio_transport_recv_pkt);
> >
> > /*
> >  * Finish the RX batch.
> >- * For a non-empty batch, the caller must hold the socket lock acquired with
> >- * lock_sock(). This function releases the lock and the batch's lookup
> >- * reference.
> >+ * For a non-empty batch, the caller holds the socket lock acquired with
> >+ * lock_sock(). Deliver any deferred write-space notification while the lock
> >+ * is held, then release the lock and the batch's lookup reference.
> >  * An empty batch is a no-op.
> >  */
> > void virtio_transport_rx_batch_finish(struct virtio_transport_rx_batch *batch)
> > {
> >+      bool write_space_pending = batch->write_space_pending;
> >       struct sock *sk = batch->sk;
> >
> >       batch->sk = NULL;
> >       batch->pkts = 0;
> >       batch->bytes = 0;
> >+      batch->write_space_pending = false;
> >
> >       if (!sk)
> >               return;
> >
> >+      if (write_space_pending)
> >+              vsock_sk(sk)->default_write_space(sk);
> >+
> >       release_sock(sk);
> >       sock_put(sk);
> > }
> >@@ -2019,7 +2031,8 @@ void virtio_transport_recv_pkt_batch(struct virtio_transport *t,
> >                       }
> >
> >                       free_pkt = virtio_transport_recv_pkt_locked(t, skb, sk,
> >-                                                                  &batchable);
> >+                                                                  &batchable,
> >+                                                                  batch);
> >                       if (!batchable)
> >                               virtio_transport_rx_batch_finish(batch);
> >                       if (free_pkt)
> >@@ -2059,23 +2072,31 @@ void virtio_transport_recv_pkt_batch(struct virtio_transport *t,
> >
> >       lock_sock(sk);
> >       /*
> >-       * Sockmap removal restores the native protocol under sk_callback_lock.
> >-       * Serialize this initial eligibility check with that update.
> >+       * Sockmap teardown restores callbacks under sk_callback_lock. Keep the
> >+       * initial eligibility check consistent with that update.
> >        */
>
> Unrelated change?
>

This is related to the new callback eligibility check, but I agree
that the original rewrite was unnecessary. I’ll keep the original
serialization wording and only extend the comment to mention both the
native protocol and callbacks.

> BTW if you are assisted by AI, please read https://docs.kernel.org/process/coding-assistants.html
>

Thanks for the reminder. I read the document when you mentioned it
previously.  take responsibility for every change in it. As this
review has shown, I missed some netdev conventions, and the patch
organization can also be improved; those are my mistakes, and I’ll
address them in v5.

I used an LLM to help set up the test environment, translate and
polish the English text, and review the series. I ran and verified the
tests myself. I’ll
disclose that assistance in v5 as described in the documentation.


> >       read_lock_bh(&sk->sk_callback_lock);
> >       start_batch = virtio_transport_recv_pkt_batchable(t, sk);
> >       read_unlock_bh(&sk->sk_callback_lock);
> >
> >+      if (start_batch)
> >+              batch->sk = sk;
> >+
> >       free_pkt = virtio_transport_recv_pkt_locked(t, skb, sk,
> >-                                                  start_batch ? &batchable : NULL);
> >+                                                  start_batch ? &batchable : NULL,
> >+                                                  start_batch ? batch : NULL);
> >       if (start_batch && batchable) {
> >               /* Keep the lookup reference until the batch is released. */
> >-              batch->sk = sk;
> >               return;
> >       }
> >
> >-      release_sock(sk);
> >-      sock_put(sk);
> >+      if (start_batch) {
> >+              virtio_transport_rx_batch_finish(batch);
> >+      } else {
> >+              release_sock(sk);
> >+              sock_put(sk);
> >+      }
> >+
> >       if (free_pkt)
> >               kfree_skb(skb);
> > }
> >--
> >2.53.0
> >
>

Re: [PATCH net-next v4 4/4] vsock/virtio: defer RX notifications until batch unlock

[Jia Jia]

>
> On Wed, Oct 14, 2026 at 07:56:40PM +0800, Jia Jia wrote:
> >An RX batch can make a socket readable while the worker still holds its
> >socket lock. Waking a blocking reader then only moves it to waiting for
> >the same lock.
> >
> >Record the existing low-watermark condition and notify once after
> >release_sock(). Move the coalesced write-space notification after the
> >unlock as well, since both callbacks wake sk_sleep(sk). Deliver
> >write-space before data-ready and keep the lookup reference until both
> >callbacks complete.
> >
> >As with write-space, save the initial sk_data_ready callback and batch only
> >while both saved callbacks remain installed. Sockets with replaced
> >callbacks retain their per-packet behavior.
> >
> >The existing psock-marker check also covers a concurrent sockmap attach:
> >at most the already admitted packet completes on the native path, and the
> >next packet ends the batch.
> >
> >Signed-off-by: Jia Jia <physicalmtea@gmail.com>
> >---
> > include/linux/virtio_vsock.h            |  1 +
> > include/net/af_vsock.h                  |  5 ++--
> > net/vmw_vsock/af_vsock.c                |  1 +
> > net/vmw_vsock/virtio_transport_common.c | 35 ++++++++++++++++++-------
> > 4 files changed, 31 insertions(+), 11 deletions(-)
> >
> >diff --git a/include/linux/virtio_vsock.h b/include/linux/virtio_vsock.h
> >index bf077ee0ded88..f1396ca3ad9b5 100644
> >--- a/include/linux/virtio_vsock.h
> >+++ b/include/linux/virtio_vsock.h
> >@@ -288,6 +288,7 @@ struct virtio_transport_rx_batch {
> >       unsigned int pkts;
> >       size_t bytes;
> >       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 9a40960d10cbe..78bab5820f528 100644
> >--- a/include/net/af_vsock.h
> >+++ b/include/net/af_vsock.h
> >@@ -64,9 +64,10 @@ struct vsock_sock {
> >       bool sent_request;
> >       bool ignore_connecting_rst;
> >       /*
> >-       * Initial callback, used to identify replacements and deliver deferred
> >-       * write-space notifications.
> >+       * Initial callbacks, used to identify replacements and deliver deferred
> >+       * notifications.
> >        */
> >+      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 fe7f49da6c4bc..f40659ec29df0 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 f4b5f0230493e..8b62b789b71c5 100644
> >--- a/net/vmw_vsock/virtio_transport_common.c
> >+++ b/net/vmw_vsock/virtio_transport_common.c
> >@@ -1588,7 +1588,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);
> >@@ -1607,7 +1608,13 @@ virtio_transport_recv_connected(struct sock *sk,
> >                       vsock_remove_sock(vsk);
> >                       break;
> >               }
> >-              vsock_data_ready(sk);
> >+              if (!data_ready_pending) {
> >+                      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);
> >@@ -1820,6 +1827,7 @@ virtio_transport_recv_pkt_batchable(struct virtio_transport *t,
> >       return sk->sk_state == TCP_ESTABLISHED &&
> >              sk->sk_type == SOCK_STREAM &&
> >              READ_ONCE(sk->sk_prot) == sk->sk_prot_creator &&
> >+             READ_ONCE(sk->sk_data_ready) == vsk->default_data_ready &&
> >              READ_ONCE(sk->sk_write_space) == vsk->default_write_space &&
> >              !sock_flag(sk, SOCK_DONE) &&
> >              vsk->transport == &t->transport &&
> >@@ -1830,8 +1838,9 @@ 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.  When @batch is non-NULL,
> >- * record a default write-space notification for batch finish instead of
> >- * delivering it immediately.  Return true if the caller must free @skb.
> >+ * record default write-space and data-ready notifications for batch finish
> >+ * instead of delivering them immediately.  Return true if the caller must
> >+ * free @skb.
> >  */
> > static bool
> > virtio_transport_recv_pkt_locked(struct virtio_transport *t,
> >@@ -1883,7 +1892,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,
> >+                                              batch ?
> >+                                              &batch->data_ready_pending : NULL);
> >               break;
> >       case TCP_CLOSING:
> >               virtio_transport_recv_disconnecting(sk, skb);
> >@@ -1957,27 +1968,33 @@ EXPORT_SYMBOL_GPL(virtio_transport_recv_pkt);
> > /*
> >  * Finish the RX batch.
> >  * For a non-empty batch, the caller holds the socket lock acquired with
> >- * lock_sock(). Deliver any deferred write-space notification while the lock
> >- * is held, then release the lock and the batch's lookup reference.
> >- * An empty batch is a no-op.
> >+ * lock_sock(). Release the lock, deliver any deferred write-space and
> >+ * data-ready notifications in that order, then drop the batch's lookup
> >+ * reference. An empty batch is a no-op.
> >  */
> > void virtio_transport_rx_batch_finish(struct virtio_transport_rx_batch *batch)
> > {
> >       bool write_space_pending = batch->write_space_pending;
> >+      bool data_ready_pending = batch->data_ready_pending;
> >       struct sock *sk = batch->sk;
> >
> >       batch->sk = NULL;
> >       batch->pkts = 0;
> >       batch->bytes = 0;
> >       batch->write_space_pending = false;
> >+      batch->data_ready_pending = false;
> >
> >       if (!sk)
> >               return;
> >
> >+      release_sock(sk);
> >+
>
> Why in the previous patch we release the socket after calling
> default_write_space() and now we are moving before?
>

My intention was to keep the two behavioral changes separately
testable. Patch 3 is a complete intermediate state: it coalesces
repeated sk_write_space()
notifications while preserving the existing under-lock callback
context. Patch 4 then adds data-ready coalescing and moves both
deferred notifications
after release_sock(). Patch 4 therefore builds on Patch 3 rather than fixing it.

> >       if (write_space_pending)
> >               vsock_sk(sk)->default_write_space(sk);
> >
> >-      release_sock(sk);
> >+      if (data_ready_pending)
> >+              vsock_sk(sk)->default_data_ready(sk);
> >+
> >       sock_put(sk);
> > }
> > EXPORT_SYMBOL_GPL(virtio_transport_rx_batch_finish);
> >--
> >2.53.0
> >
>

Re: [PATCH net-next v4 3/4] vsock/virtio: coalesce RX write-space notifications in lock batches

[Jia Jia]

>
> On Wed, Oct 14, 2026 at 07:56:39PM +0800, Jia Jia wrote:
> > Each received packet updates peer credit and may call sk_write_space().
> > A writer cannot use that credit until the batch releases the socket lock,
> > so one notification at batch finish is sufficient.
> >
> > Save the initial sk_write_space callback when the AF_VSOCK socket is
> > created and require it to remain installed for batching. This lets the
> > transport module deliver the deferred notification without referring to
> > the unexported default callback. Sockets with replaced callbacks retain
> > their per-packet behavior.
> >
> > Record the batch socket before processing its first packet so a batch that
> > ends on that packet cannot lose the pending notification.
> >
> > Signed-off-by: Jia Jia <physicalmtea@gmail.com>
> > ---
> >  include/linux/virtio_vsock.h            |  1 +
> >  include/net/af_vsock.h                  |  5 +++
> >  net/vmw_vsock/af_vsock.c                |  1 +
> >  net/vmw_vsock/virtio_transport_common.c | 53 +++++++++++++++++--------
> >  4 files changed, 44 insertions(+), 16 deletions(-)
> >
> > diff --git a/include/linux/virtio_vsock.h b/include/linux/virtio_vsock.h
> > index 26dde6909c4ed..bf077ee0ded88 100644
> > --- a/include/linux/virtio_vsock.h
> > +++ b/include/linux/virtio_vsock.h
> > @@ -287,6 +287,7 @@ struct virtio_transport_rx_batch {
> >       struct sock *sk;
> >       unsigned int pkts;
> >       size_t bytes;
> > +     bool write_space_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 5549298c1ec6b..9a40960d10cbe 100644
> > --- a/include/net/af_vsock.h
> > +++ b/include/net/af_vsock.h
> > @@ -63,6 +63,11 @@ struct vsock_sock {
> >       u32 peer_shutdown;
> >       bool sent_request;
> >       bool ignore_connecting_rst;
> > +     /*
> > +      * Initial callback, used to identify replacements and deliver deferred
> > +      * write-space notifications.
> > +      */
> > +     void (*default_write_space)(struct sock *sk);
> >
> >       /* Protected by lock_sock(sk) */
> >       u64 buffer_size;
> > diff --git a/net/vmw_vsock/af_vsock.c b/net/vmw_vsock/af_vsock.c
> > index 44cee74519555..fe7f49da6c4bc 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_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 ff74e805527a5..f4b5f0230493e 100644
> > --- a/net/vmw_vsock/virtio_transport_common.c
> > +++ b/net/vmw_vsock/virtio_transport_common.c
> > @@ -1820,6 +1820,7 @@ virtio_transport_recv_pkt_batchable(struct virtio_transport *t,
> >       return sk->sk_state == TCP_ESTABLISHED &&
> >              sk->sk_type == SOCK_STREAM &&
> >              READ_ONCE(sk->sk_prot) == sk->sk_prot_creator &&
> > +            READ_ONCE(sk->sk_write_space) == vsk->default_write_space &&
> >              !sock_flag(sk, SOCK_DONE) &&
> >              vsk->transport == &t->transport &&
> >              vvs && virtio_transport_rx_skb_has_headroom(vvs) &&
> > @@ -1828,13 +1829,15 @@ 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.
> > + * locked for another ordinary STREAM/RW packet.  When @batch is non-NULL,
> > + * record a default write-space notification for batch finish instead of
> > + * delivering it immediately.  Return true if the caller must free @skb.
> >   */
> >  static bool
> >  virtio_transport_recv_pkt_locked(struct virtio_transport *t,
> >                                struct sk_buff *skb, struct sock *sk,
> > -                              bool *batchable)
> > +                              bool *batchable,
> > +                              struct virtio_transport_rx_batch *batch)
> >  {
> >       struct virtio_vsock_hdr *hdr = virtio_vsock_hdr(skb);
> >       struct sockaddr_vm src;
> > @@ -1863,8 +1866,12 @@ virtio_transport_recv_pkt_locked(struct virtio_transport *t,
> >       if (vsk->local_addr.svm_cid != VMADDR_CID_ANY)
> >               vsk->local_addr.svm_cid = le64_to_cpu(hdr->dst_cid);
> >
> > -     if (space_available)
> > -             sk->sk_write_space(sk);
> > +     if (space_available) {
> > +             if (batch)
> > +                     batch->write_space_pending = true;
> > +             else
> > +                     sk->sk_write_space(sk);
> > +     }
> >
> >       switch (sk->sk_state) {
> >       case TCP_LISTEN:
> > @@ -1931,7 +1938,7 @@ void virtio_transport_recv_pkt(struct virtio_transport *t,
> >       }
> >
> >       lock_sock(sk);
> > -     free_pkt = virtio_transport_recv_pkt_locked(t, skb, sk, NULL);
> > +     free_pkt = virtio_transport_recv_pkt_locked(t, skb, sk, NULL, NULL);
> >       release_sock(sk);
> >
> >       /* Release refcnt obtained when we fetched this socket out of the
> > @@ -1949,22 +1956,27 @@ EXPORT_SYMBOL_GPL(virtio_transport_recv_pkt);
> >
> >  /*
> >   * Finish the RX batch.
> > - * For a non-empty batch, the caller must hold the socket lock acquired with
> > - * lock_sock(). This function releases the lock and the batch's lookup
> > - * reference.
> > + * For a non-empty batch, the caller holds the socket lock acquired with
> > + * lock_sock(). Deliver any deferred write-space notification while the lock
> > + * is held, then release the lock and the batch's lookup reference.
> >   * An empty batch is a no-op.
> >   */
> >  void virtio_transport_rx_batch_finish(struct virtio_transport_rx_batch *batch)
> >  {
> > +     bool write_space_pending = batch->write_space_pending;
> >       struct sock *sk = batch->sk;
> >
> >       batch->sk = NULL;
> >       batch->pkts = 0;
> >       batch->bytes = 0;
> > +     batch->write_space_pending = false;
> >
> >       if (!sk)
> >               return;
> >
> > +     if (write_space_pending)
> > +             vsock_sk(sk)->default_write_space(sk);
>
> Any reason not to just do READ_ONCE(sk->sk_write_space)(sk) instead of
> storing this for later? If we setup the socket in a sockmap program
> mid-batch, doesn't this and sk_data_ready (from the next patch) go
> stale?
>

Thanks for the review, I tried loading the current callback in an
earlier version, but READ_ONCE() does not synchronize the load and
invocation with sockmap attach or teardown.
For example, it could load sk_psock_write_space, then invoke it after
the psock has been removed, in which case the wrapper would no longer
forward the
native wakeup. The sk_data_ready case is more serious: the stale
sockmap wrapper can dequeue a native skb and then drop it when its
actor finds no psock.

There is also an attach-side semantic issue: one deferred data-ready
notification may represent several native skbs, while one invocation
of the sockmap
callback consumes only one skb. The saved callbacks are therefore
intentional and represent the native path at packet admission, rather
than the callback
installed at batch finish. A packet boundary that observes a callback,
protocol, or psock change terminates the batch. I’ll document this
admission-time
rule more clearly and add sockmap attach/detach stress coverage.


> > +
> >       release_sock(sk);
> >       sock_put(sk);
> >  }
> > @@ -2019,7 +2031,8 @@ void virtio_transport_recv_pkt_batch(struct virtio_transport *t,
> >                       }
> >
> >                       free_pkt = virtio_transport_recv_pkt_locked(t, skb, sk,
> > -                                                                 &batchable);
> > +                                                                 &batchable,
> > +                                                                 batch);
> >                       if (!batchable)
> >                               virtio_transport_rx_batch_finish(batch);
> >                       if (free_pkt)
> > @@ -2059,23 +2072,31 @@ void virtio_transport_recv_pkt_batch(struct virtio_transport *t,
> >
> >       lock_sock(sk);
> >       /*
> > -      * Sockmap removal restores the native protocol under sk_callback_lock.
> > -      * Serialize this initial eligibility check with that update.
> > +      * Sockmap teardown restores callbacks under sk_callback_lock. Keep the
> > +      * initial eligibility check consistent with that update.
>
> Why reword here?
>
> >        */
> >       read_lock_bh(&sk->sk_callback_lock);
> >       start_batch = virtio_transport_recv_pkt_batchable(t, sk);
> >       read_unlock_bh(&sk->sk_callback_lock);
> >
> > +     if (start_batch)
> > +             batch->sk = sk;
> > +
> >       free_pkt = virtio_transport_recv_pkt_locked(t, skb, sk,
> > -                                                 start_batch ? &batchable : NULL);
> > +                                                 start_batch ? &batchable : NULL,
> > +                                                 start_batch ? batch : NULL);
> >       if (start_batch && batchable) {
> >               /* Keep the lookup reference until the batch is released. */
> > -             batch->sk = sk;
>
> Seems like `batch->sk = sk` is moving around between patches?
>
> I don't know, feels like if the batching patch introduces
> sk_write_space/sk_data_ready bugs, then that patch should probably
> contain these patches too. Should minimize churn too?
>
>

My intention was to structure the series in two stages. Patches 1 and
2 form a complete batching change while retaining the existing
per-packet
notification behavior. Patches 3 and 4 then add notification
coalescing and after-unlock delivery. Therefore, Patch 3 is not a
correctness fix for Patch
2; the series through Patch 2 remains functional on its own.

The batch->sk assignment moves because Patch 3 introduces pending
notification state. Once the first packet can record a deferred
notification, the batch
must record its socket before processing that packet, even if the
packet subsequently ends the batch.

I agree, however, that this split causes avoidable churn.


> >               return;
> >       }
> >
> > -     release_sock(sk);
> > -     sock_put(sk);
> > +     if (start_batch) {
> > +             virtio_transport_rx_batch_finish(batch);
> > +     } else {
> > +             release_sock(sk);
> > +             sock_put(sk);
> > +     }
> > +
> >       if (free_pkt)
> >               kfree_skb(skb);
> >  }
> > --
> > 2.53.0
> >
>
> Best,
> Bobby