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

[Stefano Garzarella <sgarzare@redhat.com>]

On Sun, Oct 18, 2026 at 12:52:36PM +0800, Jia Jia wrote:
>>
>> 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.

A summary of this should be included in the commit message and where we 
define the cap in the code.

>
>> >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.

Please mention this in the commit.

>
>
>> > 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.

Yeah, exactly, thanks!

>
>> >+ */
>> >+#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.

Yes, please.

>
>> >+                              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.

Looks okay to me, but I need to review again to make sure.

Please add a comment in these cases.

>
>> >+              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.

Yeah!

Thanks,
Stefano

>
>
>> >+
>> >+      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
>> >
>>
>