[RFC net v2 1/2] af_packet: prepare the virtio header before reserving an RX frame

[RFC net v2 1/2] af_packet: prepare the virtio header before reserving an RX frame

[Jingxuan Xie]

From: jingxuanxie <jingxuanxie@deeproute.ai>

An RX virtio-header conversion can fail after packet_current_rx_frame()
has reserved space in a TPACKET_V3 block and increased its packet count.
Releasing the copy lock on that error path does not remove the incomplete
record from the block.

Prepare the header in a local variable before reserving a frame, and copy
it to the reserved frame only after conversion succeeds. Keep drop
accounting, notification and copy_skb cleanup on conversion failure.
This also keeps conversion out of the receive queue critical section.

Assisted-by: LLM
Signed-off-by: Jingxuan Xie <jingxuanxie@deeproute.ai>
---
 net/packet/af_packet.c | 22 ++++++++++++----------
 1 file changed, 12 insertions(+), 10 deletions(-)

diff --git a/net/packet/af_packet.c b/net/packet/af_packet.c
index 823e0911c910..7b3e1e27f777 100644
--- a/net/packet/af_packet.c
+++ b/net/packet/af_packet.c
@@ -2273,6 +2273,7 @@ static int tpacket_rcv(struct sk_buff *skb, struct net_device *dev,
 	__u32 ts_status;
 	unsigned int slot_id = 0;
 	int vnet_hdr_sz = 0;
+	struct virtio_net_hdr vnet_hdr;
 
 	/* struct tpacket{2,3}_hdr is aligned to a multiple of TPACKET_ALIGNMENT.
 	 * We may add members to them until current aligned size without forcing
@@ -2375,6 +2376,14 @@ static int tpacket_rcv(struct sk_buff *skb, struct net_device *dev,
 			vnet_hdr_sz = 0;
 		}
 	}
+	if (vnet_hdr_sz &&
+	    virtio_net_hdr_from_skb(skb, &vnet_hdr, vio_le(), true, 0)) {
+		drop_reason = SKB_DROP_REASON_PACKET_SOCK_ERROR;
+		atomic_inc(&po->tp_drops);
+		sk->sk_data_ready(sk);
+		sk_skb_reason_drop(sk, copy_skb, drop_reason);
+		goto drop_n_restore;
+	}
 	spin_lock(&sk->sk_receive_queue.lock);
 	h.raw = packet_current_rx_frame(po, skb,
 					TP_STATUS_KERNEL, (macoff+snaplen));
@@ -2388,16 +2397,9 @@ static int tpacket_rcv(struct sk_buff *skb, struct net_device *dev,
 		__set_bit(slot_id, po->rx_ring.rx_owner_map);
 	}
 
-	if (vnet_hdr_sz &&
-	    virtio_net_hdr_from_skb(skb, h.raw + macoff -
-				    sizeof(struct virtio_net_hdr),
-				    vio_le(), true, 0)) {
-		if (po->tp_version <= TPACKET_V2)
-			__clear_bit(slot_id, po->rx_ring.rx_owner_map);
-		else
-			prb_clear_blk_fill_status(&po->rx_ring);
-		goto drop_n_account;
-	}
+	if (vnet_hdr_sz)
+		memcpy(h.raw + macoff - sizeof(vnet_hdr), &vnet_hdr,
+		       sizeof(vnet_hdr));
 
 	if (po->tp_version <= TPACKET_V2) {
 		packet_increment_rx_head(po, &po->rx_ring);
-- 
2.25.1

[RFC net v2 1/2] af_packet: prepare the virtio header before reserving an RX frame

[Jingxuan Xie]

From: jingxuanxie <jingxuanxie@deeproute.ai>

An RX virtio-header conversion can fail after packet_current_rx_frame()
has reserved space in a TPACKET_V3 block and increased its packet count.
Releasing the copy lock on that error path does not remove the incomplete
record from the block.

Prepare the header in a local variable before reserving a frame, and copy
it to the reserved frame only after conversion succeeds. Keep drop
accounting, notification and copy_skb cleanup on conversion failure.
This also keeps conversion out of the receive queue critical section.

Assisted-by: LLM
Signed-off-by: Jingxuan Xie <jingxuanxie@deeproute.ai>
---
 net/packet/af_packet.c | 22 ++++++++++++----------
 1 file changed, 12 insertions(+), 10 deletions(-)

diff --git a/net/packet/af_packet.c b/net/packet/af_packet.c
index 823e0911c910..7b3e1e27f777 100644
--- a/net/packet/af_packet.c
+++ b/net/packet/af_packet.c
@@ -2273,6 +2273,7 @@ static int tpacket_rcv(struct sk_buff *skb, struct net_device *dev,
        __u32 ts_status;
        unsigned int slot_id = 0;
        int vnet_hdr_sz = 0;
+       struct virtio_net_hdr vnet_hdr;

        /* struct tpacket{2,3}_hdr is aligned to a multiple of TPACKET_ALIGNMENT.
         * We may add members to them until current aligned size without forcing
@@ -2375,6 +2376,14 @@ static int tpacket_rcv(struct sk_buff *skb, struct net_device *dev,
                        vnet_hdr_sz = 0;
                }
        }
+       if (vnet_hdr_sz &&
+           virtio_net_hdr_from_skb(skb, &vnet_hdr, vio_le(), true, 0)) {
+               drop_reason = SKB_DROP_REASON_PACKET_SOCK_ERROR;
+               atomic_inc(&po->tp_drops);
+               sk->sk_data_ready(sk);
+               sk_skb_reason_drop(sk, copy_skb, drop_reason);
+               goto drop_n_restore;
+       }
        spin_lock(&sk->sk_receive_queue.lock);
        h.raw = packet_current_rx_frame(po, skb,
                                        TP_STATUS_KERNEL, (macoff+snaplen));
@@ -2388,16 +2397,9 @@ static int tpacket_rcv(struct sk_buff *skb, struct net_device *dev,
                __set_bit(slot_id, po->rx_ring.rx_owner_map);
        }

-       if (vnet_hdr_sz &&
-           virtio_net_hdr_from_skb(skb, h.raw + macoff -
-                                   sizeof(struct virtio_net_hdr),
-                                   vio_le(), true, 0)) {
-               if (po->tp_version <= TPACKET_V2)
-                       __clear_bit(slot_id, po->rx_ring.rx_owner_map);
-               else
-                       prb_clear_blk_fill_status(&po->rx_ring);
-               goto drop_n_account;
-       }
+       if (vnet_hdr_sz)
+               memcpy(h.raw + macoff - sizeof(vnet_hdr), &vnet_hdr,
+                      sizeof(vnet_hdr));

        if (po->tp_version <= TPACKET_V2) {
                packet_increment_rx_head(po, &po->rx_ring);
--
2.25.1

________________________________


本电子邮件及转载的任何文件都属绝密文件，并且仅供标明地址的个人或团体使用。如果您收到的电子邮件存在错误，请通知本公司，删除全部原始信息（包括任何原始信息的备份），并且禁止散布或分发全部或部分原始信息。本公司不担保本电子邮件中信息的准确性、适当性或完整性，并且对此产生的任何错误或疏忽不承担任何责任。在正式协议签订以前，这里提到的任何参考术语、与主题相关的情形以及随后的电子邮件中的谈判，都将严格“遵守合同”。接收方应在接收电子邮件或任何附件时检查有无病毒，本公司对由于转载本电子邮件而引发病毒产生的任何损坏不承担任何责任。


Reproduced in this e-mail and any files are considered top-secret documents, and only indicate the address of individuals or groups. If you receive an e-mail there is an error, please notify the Company, delete all the original information (including any backup of the original information), and the prohibition of the dissemination or distribution of all or part of the original information. The Company does not guarantee the accuracy of the information in this e-mail, adequacy, or completeness, and this produces any error or omission does not bear any responsibility. The formal agreement was signed prior to any reference to the term mentioned here, and the circumstances relevant to the subject and subsequent e-mail in the negotiations will be strictly "comply with the contract." The receiver should be receiving an e-mail or any attachment, check for viruses. The company reproduced this e-mail because of the virus caused any damage resulting from not bear any responsibility.

[RFC net v2 0/2] af_packet: retire TPACKET_V3 blocks without waiting for copiers

[Jingxuan Xie]

This revision keeps asynchronous block retirement but removes completion-side
queue locking, the four-state enum, the eight-block publication budget and
the workqueue used in v1. Private copy accounting now lives in the existing
pgv allocation. Packet allocation and the existing retirement timer publish
ready blocks in order under their existing receive queue lock.

A preempted copier in a sealed block does not prevent capture into the next
available block. Copy completion only performs a release decrement. In the
absence of further traffic, publication may wait for a later timer callback;
there is no hard bound under RT starvation. A finite ring can still exhaust.

Patch 1 independently moves virtio RX-header preparation ahead of frame
reservation. Patch 2 implements the retirement change. Neither patch changes
the mmap layout or adds a UAPI.

Changes since v1, in response to Willem de Bruijn:
- Split virtio header preparation into its own patch.
- Remove the second receive queue lock acquisition from copy completion.
- Remove the state enum, separate metadata allocation and worker lifecycle.
- Replace the fixed publication budget with a loop over existing ready blocks.
- Use release completion and acquire observation of per-block counts.
- Permit idle delivery to wait for the existing retirement timer.

Testing: ARM64 QEMU virt, four CPUs, PREEMPT_RT and lockdep. The unmodified
retirement logic blocks sender B behind a deliberately delayed copier A.
With v2, B completes capture before A is released, for full-block and timer
retirement; idle delivery, block sequence, lengths and payload are checked.
The delay is introduced by validation-only kernel instrumentation and is
not included in this series. Ordinary psock_tpacket and psock_fanout pass
on uninstrumented RT and non-RT builds. External validation tests exercise
the actual private helpers; LKMM tests check multi-copier ordering.

The ready-block loop has no fixed work budget: a large ready backlog can
lengthen the receive queue critical section. A validation case publishes
1000 blocks in one call. This tradeoff is explicitly retained for review.
Separate RT KASAN and strict KCSAN runs pass the helper checks, psock tests,
delayed-copy tests and concurrent-close test without sanitizer reports.
Patch 1 also passes an independent malformed-GSO runtime check. ARM64 W=1
packet-object builds pass for each patch and under allyesconfig/allmodconfig.
QEMU does not establish hardware NIC or Orin integration acceptance.

v1:
https://lore.kernel.org/netdev/20261028070528.139531-1-jingxuanxie@deeproute.ai/

Review:
https://lore.kernel.org/netdev/willemdebruijn.kernel.31ae9fde05c@gmail.com/

Base: fcc00b15cb21a6de76826e5012b2c758505f5e65 (net/main).

jingxuanxie (2):
  af_packet: prepare the virtio header before reserving an RX frame
  af_packet: retire TPACKET_V3 blocks without waiting for copiers

 net/packet/af_packet.c | 162 +++++++++++++++++++++----------------------
 net/packet/internal.h  |   9 ++-
 2 files changed, 88 insertions(+), 83 deletions(-)


base-commit: fcc00b15cb21a6de76826e5012b2c758505f5e65
-- 
2.25.1

[RFC net v2 0/2] af_packet: retire TPACKET_V3 blocks without waiting for copiers

[Jingxuan Xie]

This revision keeps asynchronous block retirement but removes completion-side
queue locking, the four-state enum, the eight-block publication budget and
the workqueue used in v1. Private copy accounting now lives in the existing
pgv allocation. Packet allocation and the existing retirement timer publish
ready blocks in order under their existing receive queue lock.

A preempted copier in a sealed block does not prevent capture into the next
available block. Copy completion only performs a release decrement. In the
absence of further traffic, publication may wait for a later timer callback;
there is no hard bound under RT starvation. A finite ring can still exhaust.

Patch 1 independently moves virtio RX-header preparation ahead of frame
reservation. Patch 2 implements the retirement change. Neither patch changes
the mmap layout or adds a UAPI.

Changes since v1, in response to Willem de Bruijn:
- Split virtio header preparation into its own patch.
- Remove the second receive queue lock acquisition from copy completion.
- Remove the state enum, separate metadata allocation and worker lifecycle.
- Replace the fixed publication budget with a loop over existing ready blocks.
- Use release completion and acquire observation of per-block counts.
- Permit idle delivery to wait for the existing retirement timer.

Testing: ARM64 QEMU virt, four CPUs, PREEMPT_RT and lockdep. The unmodified
retirement logic blocks sender B behind a deliberately delayed copier A.
With v2, B completes capture before A is released, for full-block and timer
retirement; idle delivery, block sequence, lengths and payload are checked.
The delay is introduced by validation-only kernel instrumentation and is
not included in this series. Ordinary psock_tpacket and psock_fanout pass
on uninstrumented RT and non-RT builds. External validation tests exercise
the actual private helpers; LKMM tests check multi-copier ordering.

The ready-block loop has no fixed work budget: a large ready backlog can
lengthen the receive queue critical section. A validation case publishes
1000 blocks in one call. This tradeoff is explicitly retained for review.
Separate RT KASAN and strict KCSAN runs pass the helper checks, psock tests,
delayed-copy tests and concurrent-close test without sanitizer reports.
Patch 1 also passes an independent malformed-GSO runtime check. ARM64 W=1
packet-object builds pass for each patch and under allyesconfig/allmodconfig.
QEMU does not establish hardware NIC or Orin integration acceptance.

v1:
https://lore.kernel.org/netdev/20261028070528.139531-1-jingxuanxie@deeproute.ai/

Review:
https://lore.kernel.org/netdev/willemdebruijn.kernel.31ae9fde05c@gmail.com/

Base: fcc00b15cb21a6de76826e5012b2c758505f5e65 (net/main).

jingxuanxie (2):
  af_packet: prepare the virtio header before reserving an RX frame
  af_packet: retire TPACKET_V3 blocks without waiting for copiers

 net/packet/af_packet.c | 162 +++++++++++++++++++++----------------------
 net/packet/internal.h  |   9 ++-
 2 files changed, 88 insertions(+), 83 deletions(-)


base-commit: fcc00b15cb21a6de76826e5012b2c758505f5e65
--
2.25.1

________________________________


本电子邮件及转载的任何文件都属绝密文件，并且仅供标明地址的个人或团体使用。如果您收到的电子邮件存在错误，请通知本公司，删除全部原始信息（包括任何原始信息的备份），并且禁止散布或分发全部或部分原始信息。本公司不担保本电子邮件中信息的准确性、适当性或完整性，并且对此产生的任何错误或疏忽不承担任何责任。在正式协议签订以前，这里提到的任何参考术语、与主题相关的情形以及随后的电子邮件中的谈判，都将严格“遵守合同”。接收方应在接收电子邮件或任何附件时检查有无病毒，本公司对由于转载本电子邮件而引发病毒产生的任何损坏不承担任何责任。


Reproduced in this e-mail and any files are considered top-secret documents, and only indicate the address of individuals or groups. If you receive an e-mail there is an error, please notify the Company, delete all the original information (including any backup of the original information), and the prohibition of the dissemination or distribution of all or part of the original information. The Company does not guarantee the accuracy of the information in this e-mail, adequacy, or completeness, and this produces any error or omission does not bear any responsibility. The formal agreement was signed prior to any reference to the term mentioned here, and the circumstances relevant to the subject and subsequent e-mail in the negotiations will be strictly "comply with the contract." The receiver should be receiving an e-mail or any attachment, check for viruses. The company reproduced this e-mail because of the virus caused any damage resulting from not bear any responsibility.

[RFC net v2 1/2] af_packet: prepare the virtio header before reserving an RX frame

[Jingxuan Xie]

From: jingxuanxie <jingxuanxie@deeproute.ai>

An RX virtio-header conversion can fail after packet_current_rx_frame()
has reserved space in a TPACKET_V3 block and increased its packet count.
Releasing the copy lock on that error path does not remove the incomplete
record from the block.

Prepare the header in a local variable before reserving a frame, and copy
it to the reserved frame only after conversion succeeds. Keep drop
accounting, notification and copy_skb cleanup on conversion failure.
This also keeps conversion out of the receive queue critical section.

Assisted-by: LLM
Signed-off-by: Jingxuan Xie <jingxuanxie@deeproute.ai>
---
 net/packet/af_packet.c | 22 ++++++++++++----------
 1 file changed, 12 insertions(+), 10 deletions(-)

diff --git a/net/packet/af_packet.c b/net/packet/af_packet.c
index 823e0911c910..7b3e1e27f777 100644
--- a/net/packet/af_packet.c
+++ b/net/packet/af_packet.c
@@ -2273,6 +2273,7 @@ static int tpacket_rcv(struct sk_buff *skb, struct net_device *dev,
 	__u32 ts_status;
 	unsigned int slot_id = 0;
 	int vnet_hdr_sz = 0;
+	struct virtio_net_hdr vnet_hdr;
 
 	/* struct tpacket{2,3}_hdr is aligned to a multiple of TPACKET_ALIGNMENT.
 	 * We may add members to them until current aligned size without forcing
@@ -2375,6 +2376,14 @@ static int tpacket_rcv(struct sk_buff *skb, struct net_device *dev,
 			vnet_hdr_sz = 0;
 		}
 	}
+	if (vnet_hdr_sz &&
+	    virtio_net_hdr_from_skb(skb, &vnet_hdr, vio_le(), true, 0)) {
+		drop_reason = SKB_DROP_REASON_PACKET_SOCK_ERROR;
+		atomic_inc(&po->tp_drops);
+		sk->sk_data_ready(sk);
+		sk_skb_reason_drop(sk, copy_skb, drop_reason);
+		goto drop_n_restore;
+	}
 	spin_lock(&sk->sk_receive_queue.lock);
 	h.raw = packet_current_rx_frame(po, skb,
 					TP_STATUS_KERNEL, (macoff+snaplen));
@@ -2388,16 +2397,9 @@ static int tpacket_rcv(struct sk_buff *skb, struct net_device *dev,
 		__set_bit(slot_id, po->rx_ring.rx_owner_map);
 	}
 
-	if (vnet_hdr_sz &&
-	    virtio_net_hdr_from_skb(skb, h.raw + macoff -
-				    sizeof(struct virtio_net_hdr),
-				    vio_le(), true, 0)) {
-		if (po->tp_version <= TPACKET_V2)
-			__clear_bit(slot_id, po->rx_ring.rx_owner_map);
-		else
-			prb_clear_blk_fill_status(&po->rx_ring);
-		goto drop_n_account;
-	}
+	if (vnet_hdr_sz)
+		memcpy(h.raw + macoff - sizeof(vnet_hdr), &vnet_hdr,
+		       sizeof(vnet_hdr));
 
 	if (po->tp_version <= TPACKET_V2) {
 		packet_increment_rx_head(po, &po->rx_ring);
-- 
2.25.1

[RFC net v2 1/2] af_packet: prepare the virtio header before reserving an RX frame

[Jingxuan Xie]

From: jingxuanxie <jingxuanxie@deeproute.ai>

An RX virtio-header conversion can fail after packet_current_rx_frame()
has reserved space in a TPACKET_V3 block and increased its packet count.
Releasing the copy lock on that error path does not remove the incomplete
record from the block.

Prepare the header in a local variable before reserving a frame, and copy
it to the reserved frame only after conversion succeeds. Keep drop
accounting, notification and copy_skb cleanup on conversion failure.
This also keeps conversion out of the receive queue critical section.

Assisted-by: LLM
Signed-off-by: Jingxuan Xie <jingxuanxie@deeproute.ai>
---
 net/packet/af_packet.c | 22 ++++++++++++----------
 1 file changed, 12 insertions(+), 10 deletions(-)

diff --git a/net/packet/af_packet.c b/net/packet/af_packet.c
index 823e0911c910..7b3e1e27f777 100644
--- a/net/packet/af_packet.c
+++ b/net/packet/af_packet.c
@@ -2273,6 +2273,7 @@ static int tpacket_rcv(struct sk_buff *skb, struct net_device *dev,
        __u32 ts_status;
        unsigned int slot_id = 0;
        int vnet_hdr_sz = 0;
+       struct virtio_net_hdr vnet_hdr;

        /* struct tpacket{2,3}_hdr is aligned to a multiple of TPACKET_ALIGNMENT.
         * We may add members to them until current aligned size without forcing
@@ -2375,6 +2376,14 @@ static int tpacket_rcv(struct sk_buff *skb, struct net_device *dev,
                        vnet_hdr_sz = 0;
                }
        }
+       if (vnet_hdr_sz &&
+           virtio_net_hdr_from_skb(skb, &vnet_hdr, vio_le(), true, 0)) {
+               drop_reason = SKB_DROP_REASON_PACKET_SOCK_ERROR;
+               atomic_inc(&po->tp_drops);
+               sk->sk_data_ready(sk);
+               sk_skb_reason_drop(sk, copy_skb, drop_reason);
+               goto drop_n_restore;
+       }
        spin_lock(&sk->sk_receive_queue.lock);
        h.raw = packet_current_rx_frame(po, skb,
                                        TP_STATUS_KERNEL, (macoff+snaplen));
@@ -2388,16 +2397,9 @@ static int tpacket_rcv(struct sk_buff *skb, struct net_device *dev,
                __set_bit(slot_id, po->rx_ring.rx_owner_map);
        }

-       if (vnet_hdr_sz &&
-           virtio_net_hdr_from_skb(skb, h.raw + macoff -
-                                   sizeof(struct virtio_net_hdr),
-                                   vio_le(), true, 0)) {
-               if (po->tp_version <= TPACKET_V2)
-                       __clear_bit(slot_id, po->rx_ring.rx_owner_map);
-               else
-                       prb_clear_blk_fill_status(&po->rx_ring);
-               goto drop_n_account;
-       }
+       if (vnet_hdr_sz)
+               memcpy(h.raw + macoff - sizeof(vnet_hdr), &vnet_hdr,
+                      sizeof(vnet_hdr));

        if (po->tp_version <= TPACKET_V2) {
                packet_increment_rx_head(po, &po->rx_ring);
--
2.25.1

________________________________


本电子邮件及转载的任何文件都属绝密文件，并且仅供标明地址的个人或团体使用。如果您收到的电子邮件存在错误，请通知本公司，删除全部原始信息（包括任何原始信息的备份），并且禁止散布或分发全部或部分原始信息。本公司不担保本电子邮件中信息的准确性、适当性或完整性，并且对此产生的任何错误或疏忽不承担任何责任。在正式协议签订以前，这里提到的任何参考术语、与主题相关的情形以及随后的电子邮件中的谈判，都将严格“遵守合同”。接收方应在接收电子邮件或任何附件时检查有无病毒，本公司对由于转载本电子邮件而引发病毒产生的任何损坏不承担任何责任。


Reproduced in this e-mail and any files are considered top-secret documents, and only indicate the address of individuals or groups. If you receive an e-mail there is an error, please notify the Company, delete all the original information (including any backup of the original information), and the prohibition of the dissemination or distribution of all or part of the original information. The Company does not guarantee the accuracy of the information in this e-mail, adequacy, or completeness, and this produces any error or omission does not bear any responsibility. The formal agreement was signed prior to any reference to the term mentioned here, and the circumstances relevant to the subject and subsequent e-mail in the negotiations will be strictly "comply with the contract." The receiver should be receiving an e-mail or any attachment, check for viruses. The company reproduced this e-mail because of the virus caused any damage resulting from not bear any responsibility.

[RFC net v2 2/2] af_packet: retire TPACKET_V3 blocks without waiting for copiers

[Jingxuan Xie]

From: jingxuanxie <jingxuanxie@deeproute.ai>

TPACKET_V3 waits for a ring-wide read lock while holding the receive
queue lock before retiring a block. On PREEMPT_RT a preempted copier
can delay another packet path or the retirement timer until it runs
again. This was investigated after an approximately 355 ms retirement
wait in a PREEMPT_RT deployment.

Keep copy accounting in each block's existing private pgv descriptor.
Seal a full or expired block and advance to the next block without
waiting. Retain the original descriptor across copying and release its
copy reference after all packet data and headers have been written.
Publish sealed blocks in order when an acquire read observes no copies.

Publication runs under the existing receive queue lock in packet
allocation and timer paths. Completion does not reacquire that lock.
No worker or publication budget is needed. When traffic stops, completed
blocks may remain unpublished until the next timer callback; this does
not promise a hard capture delivery deadline. Ring exhaustion can still
drop capture copies.

Use each retired block's saved last packet and its own cache-flush range,
and exclude sealed blocks from allocation and fanout space checks. Poll
uses the publication cursor instead of the producer cursor. Widen the
producer index to cover the same block count as the publication index.
Allow an exactly fitting record so a maximum-size first record cannot
cause an empty block to be reopened instead of advancing publication.

Reproduction uses a QEMU ARM64 SMP PREEMPT_RT kernel with validation-only
instrumentation delaying an outgoing copier after frame reservation.
The instrumentation is excluded from these patches.

Fixes: 632ca50f2cbd ("af_packet: TPACKET_V3: replace busy-wait loop")
Assisted-by: LLM
Signed-off-by: Jingxuan Xie <jingxuanxie@deeproute.ai>
---
 net/packet/af_packet.c | 140 ++++++++++++++++++++---------------------
 net/packet/internal.h  |   9 ++-
 2 files changed, 76 insertions(+), 73 deletions(-)

diff --git a/net/packet/af_packet.c b/net/packet/af_packet.c
index 7b3e1e27f777..5bc486326be9 100644
--- a/net/packet/af_packet.c
+++ b/net/packet/af_packet.c
@@ -197,7 +197,7 @@ static void *packet_previous_frame(struct packet_sock *po,
 		struct packet_ring_buffer *rb,
 		int status);
 static void packet_increment_head(struct packet_ring_buffer *buff);
-static int prb_curr_blk_in_use(struct tpacket_block_desc *);
+static int prb_curr_blk_in_use(struct tpacket_kbdq_core *);
 static void *prb_dispatch_next_block(struct tpacket_kbdq_core *,
 			struct packet_sock *);
 static void prb_retire_current_block(struct tpacket_kbdq_core *,
@@ -205,6 +205,7 @@ static void prb_retire_current_block(struct tpacket_kbdq_core *,
 static int prb_queue_frozen(struct tpacket_kbdq_core *);
 static void prb_open_block(struct tpacket_kbdq_core *,
 		struct tpacket_block_desc *);
+static void prb_publish_blocks(struct tpacket_kbdq_core *, struct packet_sock *);
 static enum hrtimer_restart prb_retire_rx_blk_timer_expired(struct hrtimer *);
 static void prb_fill_rxhash(struct tpacket_kbdq_core *, struct tpacket3_hdr *);
 static void prb_clear_rxhash(struct tpacket_kbdq_core *,
@@ -666,7 +667,6 @@ static void init_prb_bdqc(struct packet_sock *po,
 		p1->interval_ktime = ms_to_ktime(prb_calc_retire_blk_tmo(po,
 						 req_u->req3.tp_block_size));
 	p1->blk_sizeof_priv = req_u->req3.tp_sizeof_priv;
-	rwlock_init(&p1->blk_fill_in_prog_lock);
 
 	p1->max_frame_len = p1->kblk_size - BLK_PLUS_PRIV(p1->blk_sizeof_priv);
 	prb_init_ft_ops(p1, req_u);
@@ -697,25 +697,11 @@ static enum hrtimer_restart prb_retire_rx_blk_timer_expired(struct hrtimer *t)
 	struct tpacket_block_desc *pbd;
 
 	spin_lock(&po->sk.sk_receive_queue.lock);
+	prb_publish_blocks(pkc, po);
 
 	frozen = prb_queue_frozen(pkc);
 	pbd = GET_CURR_PBLOCK_DESC_FROM_CORE(pkc);
 
-	/* We only need to plug the race when the block is partially filled.
-	 * tpacket_rcv:
-	 *		lock(); increment BLOCK_NUM_PKTS; unlock()
-	 *		copy_bits() is in progress ...
-	 *		timer fires on other cpu:
-	 *		we can't retire the current block because copy_bits
-	 *		is in progress.
-	 *
-	 */
-	if (BLOCK_NUM_PKTS(pbd)) {
-		/* Waiting for skb_copy_bits to finish... */
-		write_lock(&pkc->blk_fill_in_prog_lock);
-		write_unlock(&pkc->blk_fill_in_prog_lock);
-	}
-
 	if (!frozen) {
 		if (BLOCK_NUM_PKTS(pbd)) {
 			/* Not an empty block. Need retire the block. */
@@ -726,7 +712,7 @@ static enum hrtimer_restart prb_retire_rx_blk_timer_expired(struct hrtimer *t)
 		/* Case 1. Queue was frozen because user-space was
 		 * lagging behind.
 		 */
-		if (!prb_curr_blk_in_use(pbd)) {
+		if (!prb_curr_blk_in_use(pkc)) {
 			/* Case 2. queue was frozen,user-space caught up,
 			 * now the link went idle && the timer fired.
 			 * We don't have a block to close.So we open this
@@ -756,7 +742,7 @@ static void prb_flush_block(struct tpacket_kbdq_core *pkc1,
 	/* Skip the block header(we know header WILL fit in 4K) */
 	start += PAGE_SIZE;
 
-	end = (u8 *)PAGE_ALIGN((unsigned long)pkc1->pkblk_end);
+	end = (u8 *)PAGE_ALIGN((unsigned long)pbd1 + pkc1->kblk_size);
 	for (; start < end; start += PAGE_SIZE)
 		flush_dcache_page(pgv_to_page(start));
 
@@ -777,15 +763,7 @@ static void prb_flush_block(struct tpacket_kbdq_core *pkc1,
 #endif
 }
 
-/*
- * Side effect:
- *
- * 1) flush the block
- * 2) Increment active_blk_num
- *
- * Note:We DONT refresh the timer on purpose.
- *	Because almost always the next block will be opened.
- */
+/* The producer may already be filling another block. */
 static void prb_close_block(struct tpacket_kbdq_core *pkc1,
 		struct tpacket_block_desc *pbd1,
 		struct packet_sock *po, unsigned int stat)
@@ -799,7 +777,7 @@ static void prb_close_block(struct tpacket_kbdq_core *pkc1,
 	if (atomic_read(&po->tp_drops))
 		status |= TP_STATUS_LOSING;
 
-	last_pkt = (struct tpacket3_hdr *)pkc1->prev;
+	last_pkt = pkc1->pkbdq[pkc1->publish_blk_num].last_pkt;
 	last_pkt->tp_next_offset = 0;
 
 	/* Get the ts of the last pkt */
@@ -824,8 +802,27 @@ static void prb_close_block(struct tpacket_kbdq_core *pkc1,
 	prb_flush_block(pkc1, pbd1, status);
 
 	sk->sk_data_ready(sk);
+}
+
+/* Caller holds the receive queue lock. Never wait for unfinished copies. */
+static void prb_publish_blocks(struct tpacket_kbdq_core *pkc,
+			       struct packet_sock *po)
+{
+	unsigned int pending = pkc->pending_blocks;
+
+	while (pending--) {
+		struct pgv *block = &pkc->pkbdq[pkc->publish_blk_num];
+		struct tpacket_block_desc *pbd = (void *)block->buffer;
 
-	pkc1->kactive_blk_num = GET_NEXT_PRB_BLK_NUM(pkc1);
+		/* Pairs with the release decrement after every completed copy. */
+		if (atomic_read_acquire(&block->in_flight))
+			break;
+		prb_close_block(pkc, pbd, po, block->retire_status);
+		WRITE_ONCE(block->last_pkt, NULL);
+		if (++pkc->publish_blk_num == pkc->knum_blocks)
+			pkc->publish_blk_num = 0;
+		pkc->pending_blocks--;
+	}
 }
 
 static void prb_thaw_queue(struct tpacket_kbdq_core *pkc)
@@ -852,6 +849,7 @@ static void prb_open_block(struct tpacket_kbdq_core *pkc1,
 {
 	struct timespec64 ts;
 	struct tpacket_hdr_v1 *h1 = &pbd1->hdr.bh1;
+	struct pgv *block = &pkc1->pkbdq[pkc1->kactive_blk_num];
 
 	smp_rmb();
 
@@ -877,6 +875,7 @@ static void prb_open_block(struct tpacket_kbdq_core *pkc1,
 	pbd1->version = pkc1->version;
 	pkc1->prev = pkc1->nxt_offset;
 	pkc1->pkblk_end = pkc1->pkblk_start + pkc1->kblk_size;
+	atomic_set(&block->in_flight, 0);
 
 	prb_thaw_queue(pkc1);
 
@@ -932,7 +931,7 @@ static void *prb_dispatch_next_block(struct tpacket_kbdq_core *pkc,
 	pbd = GET_CURR_PBLOCK_DESC_FROM_CORE(pkc);
 
 	/* 2. If this block is currently in_use then freeze the queue */
-	if (TP_STATUS_USER & BLOCK_STATUS(pbd)) {
+	if (prb_curr_blk_in_use(pkc)) {
 		prb_freeze_queue(pkc, po);
 		return NULL;
 	}
@@ -950,31 +949,24 @@ static void prb_retire_current_block(struct tpacket_kbdq_core *pkc,
 		struct packet_sock *po, unsigned int status)
 {
 	struct tpacket_block_desc *pbd = GET_CURR_PBLOCK_DESC_FROM_CORE(pkc);
+	struct pgv *block = &pkc->pkbdq[pkc->kactive_blk_num];
 
-	/* retire/close the current block */
-	if (likely(TP_STATUS_KERNEL == BLOCK_STATUS(pbd))) {
-		/*
-		 * Plug the case where copy_bits() is in progress on
-		 * cpu-0 and tpacket_rcv() got invoked on cpu-1, didn't
-		 * have space to copy the pkt in the current block and
-		 * called prb_retire_current_block()
-		 *
-		 * We don't need to worry about the TMO case because
-		 * the timer-handler already handled this case.
-		 */
-		if (!(status & TP_STATUS_BLK_TMO)) {
-			/* Waiting for skb_copy_bits to finish... */
-			write_lock(&pkc->blk_fill_in_prog_lock);
-			write_unlock(&pkc->blk_fill_in_prog_lock);
-		}
-		prb_close_block(pkc, pbd, po, status);
+	if (!BLOCK_NUM_PKTS(pbd))
 		return;
-	}
+	block->retire_status = status;
+	WRITE_ONCE(block->last_pkt, (struct tpacket3_hdr *)pkc->prev);
+	pkc->pending_blocks++;
+	pkc->kactive_blk_num = GET_NEXT_PRB_BLK_NUM(pkc);
+	prb_publish_blocks(pkc, po);
 }
 
-static int prb_curr_blk_in_use(struct tpacket_block_desc *pbd)
+static int prb_curr_blk_in_use(struct tpacket_kbdq_core *pkc)
 {
-	return TP_STATUS_USER & BLOCK_STATUS(pbd);
+	struct pgv *block = &pkc->pkbdq[pkc->kactive_blk_num];
+	struct tpacket_block_desc *pbd = (void *)block->buffer;
+
+	return READ_ONCE(block->last_pkt) ||
+	       (TP_STATUS_USER & READ_ONCE(BLOCK_STATUS(pbd)));
 }
 
 static int prb_queue_frozen(struct tpacket_kbdq_core *pkc)
@@ -982,14 +974,6 @@ static int prb_queue_frozen(struct tpacket_kbdq_core *pkc)
 	return pkc->reset_pending_on_curr_blk;
 }
 
-static void prb_clear_blk_fill_status(struct packet_ring_buffer *rb)
-	__releases(&pkc->blk_fill_in_prog_lock)
-{
-	struct tpacket_kbdq_core *pkc  = GET_PBDQC_FROM_RB(rb);
-
-	read_unlock(&pkc->blk_fill_in_prog_lock);
-}
-
 static void prb_fill_rxhash(struct tpacket_kbdq_core *pkc,
 			struct tpacket3_hdr *ppd)
 {
@@ -1038,7 +1022,6 @@ static void prb_fill_curr_block(char *curr,
 				struct tpacket_kbdq_core *pkc,
 				struct tpacket_block_desc *pbd,
 				unsigned int len)
-	__acquires(&pkc->blk_fill_in_prog_lock)
 {
 	struct tpacket3_hdr *ppd;
 
@@ -1048,7 +1031,7 @@ static void prb_fill_curr_block(char *curr,
 	pkc->nxt_offset += TOTAL_PKT_LEN_INCL_ALIGN(len);
 	BLOCK_LEN(pbd) += TOTAL_PKT_LEN_INCL_ALIGN(len);
 	BLOCK_NUM_PKTS(pbd) += 1;
-	read_lock(&pkc->blk_fill_in_prog_lock);
+	atomic_inc(&pkc->pkbdq[pkc->kactive_blk_num].in_flight);
 	prb_run_all_ft_ops(pkc, ppd);
 }
 
@@ -1063,6 +1046,7 @@ static void *__packet_lookup_frame_in_block(struct packet_sock *po,
 	char *curr, *end;
 
 	pkc = GET_PBDQC_FROM_RB(&po->rx_ring);
+	prb_publish_blocks(pkc, po);
 	pbd = GET_CURR_PBLOCK_DESC_FROM_CORE(pkc);
 
 	/* Queue is frozen when user space is lagging behind */
@@ -1071,7 +1055,7 @@ static void *__packet_lookup_frame_in_block(struct packet_sock *po,
 		 * Check if that last block which caused the queue to freeze,
 		 * is still in_use by user-space.
 		 */
-		if (prb_curr_blk_in_use(pbd)) {
+		if (prb_curr_blk_in_use(pkc)) {
 			/* Can't record this packet */
 			return NULL;
 		} else {
@@ -1091,7 +1075,7 @@ static void *__packet_lookup_frame_in_block(struct packet_sock *po,
 	end = (char *)pbd + pkc->kblk_size;
 
 	/* first try the current block */
-	if (curr+TOTAL_PKT_LEN_INCL_ALIGN(len) < end) {
+	if (curr + TOTAL_PKT_LEN_INCL_ALIGN(len) <= end) {
 		prb_fill_curr_block(curr, pkc, pbd, len);
 		return (void *)curr;
 	}
@@ -1142,7 +1126,9 @@ static void *prb_lookup_block(const struct packet_sock *po,
 	struct tpacket_kbdq_core *pkc  = GET_PBDQC_FROM_RB(rb);
 	struct tpacket_block_desc *pbd = GET_PBLOCK_DESC(pkc, idx);
 
-	if (status != BLOCK_STATUS(pbd))
+	if (status == TP_STATUS_KERNEL && READ_ONCE(pkc->pkbdq[idx].last_pkt))
+		return NULL;
+	if (status != (READ_ONCE(BLOCK_STATUS(pbd)) & TP_STATUS_USER))
 		return NULL;
 	return pbd;
 }
@@ -1150,8 +1136,8 @@ static void *prb_lookup_block(const struct packet_sock *po,
 static int prb_previous_blk_num(struct packet_ring_buffer *rb)
 {
 	unsigned int prev;
-	if (rb->prb_bdqc.kactive_blk_num)
-		prev = rb->prb_bdqc.kactive_blk_num-1;
+	if (rb->prb_bdqc.publish_blk_num)
+		prev = rb->prb_bdqc.publish_blk_num - 1;
 	else
 		prev = rb->prb_bdqc.knum_blocks-1;
 	return prev;
@@ -1269,6 +1255,8 @@ static bool __tpacket_v3_has_room(const struct packet_sock *po, int pow_off)
 
 	len = READ_ONCE(po->rx_ring.prb_bdqc.knum_blocks);
 	idx = READ_ONCE(po->rx_ring.prb_bdqc.kactive_blk_num);
+	if (!prb_lookup_block(po, &po->rx_ring, idx, TP_STATUS_KERNEL))
+		return false;
 	if (pow_off)
 		idx += len >> pow_off;
 	if (idx >= len)
@@ -2274,6 +2262,7 @@ static int tpacket_rcv(struct sk_buff *skb, struct net_device *dev,
 	unsigned int slot_id = 0;
 	int vnet_hdr_sz = 0;
 	struct virtio_net_hdr vnet_hdr;
+	struct pgv *copy_block = NULL;
 
 	/* struct tpacket{2,3}_hdr is aligned to a multiple of TPACKET_ALIGNMENT.
 	 * We may add members to them until current aligned size without forcing
@@ -2306,8 +2295,10 @@ static int tpacket_rcv(struct sk_buff *skb, struct net_device *dev,
 	if (!res)
 		goto drop_n_restore;
 
-	/* If we are flooded, just give up */
-	if (__packet_rcv_has_room(po, skb) == ROOM_NONE) {
+	/* V3 allocation must get a chance to publish completed pending blocks. */
+	if (__packet_rcv_has_room(po, skb) == ROOM_NONE &&
+	    (po->tp_version != TPACKET_V3 ||
+	     !READ_ONCE(po->rx_ring.prb_bdqc.pending_blocks))) {
 		atomic_inc(&po->tp_drops);
 		goto drop_n_restore;
 	}
@@ -2389,6 +2380,11 @@ static int tpacket_rcv(struct sk_buff *skb, struct net_device *dev,
 					TP_STATUS_KERNEL, (macoff+snaplen));
 	if (!h.raw)
 		goto drop_n_account;
+	if (po->tp_version == TPACKET_V3) {
+		struct tpacket_kbdq_core *pkc = &po->rx_ring.prb_bdqc;
+
+		copy_block = &pkc->pkbdq[pkc->kactive_blk_num];
+	}
 
 	if (po->tp_version <= TPACKET_V2) {
 		slot_id = po->rx_ring.head;
@@ -2518,7 +2514,8 @@ static int tpacket_rcv(struct sk_buff *skb, struct net_device *dev,
 		spin_unlock(&sk->sk_receive_queue.lock);
 		sk->sk_data_ready(sk);
 	} else if (po->tp_version == TPACKET_V3) {
-		prb_clear_blk_fill_status(&po->rx_ring);
+		/* No frame accesses may follow this release. */
+		atomic_dec_return_release(&copy_block->in_flight);
 	}
 
 drop_n_restore:
@@ -4376,8 +4373,9 @@ static __poll_t packet_poll(struct file *file, struct socket *sock,
 
 	spin_lock_bh(&sk->sk_receive_queue.lock);
 	if (po->rx_ring.pg_vec) {
-		if (!packet_previous_rx_frame(po, &po->rx_ring,
-			TP_STATUS_KERNEL))
+		if (po->tp_version == TPACKET_V3 ?
+		    !!packet_previous_rx_frame(po, &po->rx_ring, TP_STATUS_USER) :
+		    !packet_previous_rx_frame(po, &po->rx_ring, TP_STATUS_KERNEL))
 			mask |= EPOLLIN | EPOLLRDNORM;
 	}
 	__packet_rcv_try_clear_pressure(po);
diff --git a/net/packet/internal.h b/net/packet/internal.h
index f5c8cd0eed2c..bece88a9672c 100644
--- a/net/packet/internal.h
+++ b/net/packet/internal.h
@@ -20,7 +20,7 @@ struct tpacket_kbdq_core {
 	unsigned int	feature_req_word;
 	unsigned int	hdrlen;
 	unsigned char	reset_pending_on_curr_blk;
-	unsigned short	kactive_blk_num;
+	unsigned int	kactive_blk_num;
 	unsigned int	blk_sizeof_priv;
 
 	unsigned short  version;
@@ -36,7 +36,8 @@ struct tpacket_kbdq_core {
 
 	struct sk_buff	*skb;
 
-	rwlock_t	blk_fill_in_prog_lock;
+	unsigned int	publish_blk_num;
+	unsigned int	pending_blocks;
 
 	/* Default is set to 8ms */
 #define DEFAULT_PRB_RETIRE_TOV	(8)
@@ -49,6 +50,10 @@ struct tpacket_kbdq_core {
 
 struct pgv {
 	char *buffer;
+	/* RX V3 only; last_pkt marks a sealed, unpublished block. */
+	struct tpacket3_hdr *last_pkt;
+	atomic_t in_flight;
+	unsigned int retire_status;
 };
 
 struct packet_ring_buffer {
-- 
2.25.1

[RFC net v2 2/2] af_packet: retire TPACKET_V3 blocks without waiting for copiers

[Jingxuan Xie]

From: jingxuanxie <jingxuanxie@deeproute.ai>

TPACKET_V3 waits for a ring-wide read lock while holding the receive
queue lock before retiring a block. On PREEMPT_RT a preempted copier
can delay another packet path or the retirement timer until it runs
again. This was investigated after an approximately 355 ms retirement
wait in a PREEMPT_RT deployment.

Keep copy accounting in each block's existing private pgv descriptor.
Seal a full or expired block and advance to the next block without
waiting. Retain the original descriptor across copying and release its
copy reference after all packet data and headers have been written.
Publish sealed blocks in order when an acquire read observes no copies.

Publication runs under the existing receive queue lock in packet
allocation and timer paths. Completion does not reacquire that lock.
No worker or publication budget is needed. When traffic stops, completed
blocks may remain unpublished until the next timer callback; this does
not promise a hard capture delivery deadline. Ring exhaustion can still
drop capture copies.

Use each retired block's saved last packet and its own cache-flush range,
and exclude sealed blocks from allocation and fanout space checks. Poll
uses the publication cursor instead of the producer cursor. Widen the
producer index to cover the same block count as the publication index.
Allow an exactly fitting record so a maximum-size first record cannot
cause an empty block to be reopened instead of advancing publication.

Reproduction uses a QEMU ARM64 SMP PREEMPT_RT kernel with validation-only
instrumentation delaying an outgoing copier after frame reservation.
The instrumentation is excluded from these patches.

Fixes: 632ca50f2cbd ("af_packet: TPACKET_V3: replace busy-wait loop")
Assisted-by: LLM
Signed-off-by: Jingxuan Xie <jingxuanxie@deeproute.ai>
---
 net/packet/af_packet.c | 140 ++++++++++++++++++++---------------------
 net/packet/internal.h  |   9 ++-
 2 files changed, 76 insertions(+), 73 deletions(-)

diff --git a/net/packet/af_packet.c b/net/packet/af_packet.c
index 7b3e1e27f777..5bc486326be9 100644
--- a/net/packet/af_packet.c
+++ b/net/packet/af_packet.c
@@ -197,7 +197,7 @@ static void *packet_previous_frame(struct packet_sock *po,
                struct packet_ring_buffer *rb,
                int status);
 static void packet_increment_head(struct packet_ring_buffer *buff);
-static int prb_curr_blk_in_use(struct tpacket_block_desc *);
+static int prb_curr_blk_in_use(struct tpacket_kbdq_core *);
 static void *prb_dispatch_next_block(struct tpacket_kbdq_core *,
                        struct packet_sock *);
 static void prb_retire_current_block(struct tpacket_kbdq_core *,
@@ -205,6 +205,7 @@ static void prb_retire_current_block(struct tpacket_kbdq_core *,
 static int prb_queue_frozen(struct tpacket_kbdq_core *);
 static void prb_open_block(struct tpacket_kbdq_core *,
                struct tpacket_block_desc *);
+static void prb_publish_blocks(struct tpacket_kbdq_core *, struct packet_sock *);
 static enum hrtimer_restart prb_retire_rx_blk_timer_expired(struct hrtimer *);
 static void prb_fill_rxhash(struct tpacket_kbdq_core *, struct tpacket3_hdr *);
 static void prb_clear_rxhash(struct tpacket_kbdq_core *,
@@ -666,7 +667,6 @@ static void init_prb_bdqc(struct packet_sock *po,
                p1->interval_ktime = ms_to_ktime(prb_calc_retire_blk_tmo(po,
                                                 req_u->req3.tp_block_size));
        p1->blk_sizeof_priv = req_u->req3.tp_sizeof_priv;
-       rwlock_init(&p1->blk_fill_in_prog_lock);

        p1->max_frame_len = p1->kblk_size - BLK_PLUS_PRIV(p1->blk_sizeof_priv);
        prb_init_ft_ops(p1, req_u);
@@ -697,25 +697,11 @@ static enum hrtimer_restart prb_retire_rx_blk_timer_expired(struct hrtimer *t)
        struct tpacket_block_desc *pbd;

        spin_lock(&po->sk.sk_receive_queue.lock);
+       prb_publish_blocks(pkc, po);

        frozen = prb_queue_frozen(pkc);
        pbd = GET_CURR_PBLOCK_DESC_FROM_CORE(pkc);

-       /* We only need to plug the race when the block is partially filled.
-        * tpacket_rcv:
-        *              lock(); increment BLOCK_NUM_PKTS; unlock()
-        *              copy_bits() is in progress ...
-        *              timer fires on other cpu:
-        *              we can't retire the current block because copy_bits
-        *              is in progress.
-        *
-        */
-       if (BLOCK_NUM_PKTS(pbd)) {
-               /* Waiting for skb_copy_bits to finish... */
-               write_lock(&pkc->blk_fill_in_prog_lock);
-               write_unlock(&pkc->blk_fill_in_prog_lock);
-       }
-
        if (!frozen) {
                if (BLOCK_NUM_PKTS(pbd)) {
                        /* Not an empty block. Need retire the block. */
@@ -726,7 +712,7 @@ static enum hrtimer_restart prb_retire_rx_blk_timer_expired(struct hrtimer *t)
                /* Case 1. Queue was frozen because user-space was
                 * lagging behind.
                 */
-               if (!prb_curr_blk_in_use(pbd)) {
+               if (!prb_curr_blk_in_use(pkc)) {
                        /* Case 2. queue was frozen,user-space caught up,
                         * now the link went idle && the timer fired.
                         * We don't have a block to close.So we open this
@@ -756,7 +742,7 @@ static void prb_flush_block(struct tpacket_kbdq_core *pkc1,
        /* Skip the block header(we know header WILL fit in 4K) */
        start += PAGE_SIZE;

-       end = (u8 *)PAGE_ALIGN((unsigned long)pkc1->pkblk_end);
+       end = (u8 *)PAGE_ALIGN((unsigned long)pbd1 + pkc1->kblk_size);
        for (; start < end; start += PAGE_SIZE)
                flush_dcache_page(pgv_to_page(start));

@@ -777,15 +763,7 @@ static void prb_flush_block(struct tpacket_kbdq_core *pkc1,
 #endif
 }

-/*
- * Side effect:
- *
- * 1) flush the block
- * 2) Increment active_blk_num
- *
- * Note:We DONT refresh the timer on purpose.
- *     Because almost always the next block will be opened.
- */
+/* The producer may already be filling another block. */
 static void prb_close_block(struct tpacket_kbdq_core *pkc1,
                struct tpacket_block_desc *pbd1,
                struct packet_sock *po, unsigned int stat)
@@ -799,7 +777,7 @@ static void prb_close_block(struct tpacket_kbdq_core *pkc1,
        if (atomic_read(&po->tp_drops))
                status |= TP_STATUS_LOSING;

-       last_pkt = (struct tpacket3_hdr *)pkc1->prev;
+       last_pkt = pkc1->pkbdq[pkc1->publish_blk_num].last_pkt;
        last_pkt->tp_next_offset = 0;

        /* Get the ts of the last pkt */
@@ -824,8 +802,27 @@ static void prb_close_block(struct tpacket_kbdq_core *pkc1,
        prb_flush_block(pkc1, pbd1, status);

        sk->sk_data_ready(sk);
+}
+
+/* Caller holds the receive queue lock. Never wait for unfinished copies. */
+static void prb_publish_blocks(struct tpacket_kbdq_core *pkc,
+                              struct packet_sock *po)
+{
+       unsigned int pending = pkc->pending_blocks;
+
+       while (pending--) {
+               struct pgv *block = &pkc->pkbdq[pkc->publish_blk_num];
+               struct tpacket_block_desc *pbd = (void *)block->buffer;

-       pkc1->kactive_blk_num = GET_NEXT_PRB_BLK_NUM(pkc1);
+               /* Pairs with the release decrement after every completed copy. */
+               if (atomic_read_acquire(&block->in_flight))
+                       break;
+               prb_close_block(pkc, pbd, po, block->retire_status);
+               WRITE_ONCE(block->last_pkt, NULL);
+               if (++pkc->publish_blk_num == pkc->knum_blocks)
+                       pkc->publish_blk_num = 0;
+               pkc->pending_blocks--;
+       }
 }

 static void prb_thaw_queue(struct tpacket_kbdq_core *pkc)
@@ -852,6 +849,7 @@ static void prb_open_block(struct tpacket_kbdq_core *pkc1,
 {
        struct timespec64 ts;
        struct tpacket_hdr_v1 *h1 = &pbd1->hdr.bh1;
+       struct pgv *block = &pkc1->pkbdq[pkc1->kactive_blk_num];

        smp_rmb();

@@ -877,6 +875,7 @@ static void prb_open_block(struct tpacket_kbdq_core *pkc1,
        pbd1->version = pkc1->version;
        pkc1->prev = pkc1->nxt_offset;
        pkc1->pkblk_end = pkc1->pkblk_start + pkc1->kblk_size;
+       atomic_set(&block->in_flight, 0);

        prb_thaw_queue(pkc1);

@@ -932,7 +931,7 @@ static void *prb_dispatch_next_block(struct tpacket_kbdq_core *pkc,
        pbd = GET_CURR_PBLOCK_DESC_FROM_CORE(pkc);

        /* 2. If this block is currently in_use then freeze the queue */
-       if (TP_STATUS_USER & BLOCK_STATUS(pbd)) {
+       if (prb_curr_blk_in_use(pkc)) {
                prb_freeze_queue(pkc, po);
                return NULL;
        }
@@ -950,31 +949,24 @@ static void prb_retire_current_block(struct tpacket_kbdq_core *pkc,
                struct packet_sock *po, unsigned int status)
 {
        struct tpacket_block_desc *pbd = GET_CURR_PBLOCK_DESC_FROM_CORE(pkc);
+       struct pgv *block = &pkc->pkbdq[pkc->kactive_blk_num];

-       /* retire/close the current block */
-       if (likely(TP_STATUS_KERNEL == BLOCK_STATUS(pbd))) {
-               /*
-                * Plug the case where copy_bits() is in progress on
-                * cpu-0 and tpacket_rcv() got invoked on cpu-1, didn't
-                * have space to copy the pkt in the current block and
-                * called prb_retire_current_block()
-                *
-                * We don't need to worry about the TMO case because
-                * the timer-handler already handled this case.
-                */
-               if (!(status & TP_STATUS_BLK_TMO)) {
-                       /* Waiting for skb_copy_bits to finish... */
-                       write_lock(&pkc->blk_fill_in_prog_lock);
-                       write_unlock(&pkc->blk_fill_in_prog_lock);
-               }
-               prb_close_block(pkc, pbd, po, status);
+       if (!BLOCK_NUM_PKTS(pbd))
                return;
-       }
+       block->retire_status = status;
+       WRITE_ONCE(block->last_pkt, (struct tpacket3_hdr *)pkc->prev);
+       pkc->pending_blocks++;
+       pkc->kactive_blk_num = GET_NEXT_PRB_BLK_NUM(pkc);
+       prb_publish_blocks(pkc, po);
 }

-static int prb_curr_blk_in_use(struct tpacket_block_desc *pbd)
+static int prb_curr_blk_in_use(struct tpacket_kbdq_core *pkc)
 {
-       return TP_STATUS_USER & BLOCK_STATUS(pbd);
+       struct pgv *block = &pkc->pkbdq[pkc->kactive_blk_num];
+       struct tpacket_block_desc *pbd = (void *)block->buffer;
+
+       return READ_ONCE(block->last_pkt) ||
+              (TP_STATUS_USER & READ_ONCE(BLOCK_STATUS(pbd)));
 }

 static int prb_queue_frozen(struct tpacket_kbdq_core *pkc)
@@ -982,14 +974,6 @@ static int prb_queue_frozen(struct tpacket_kbdq_core *pkc)
        return pkc->reset_pending_on_curr_blk;
 }

-static void prb_clear_blk_fill_status(struct packet_ring_buffer *rb)
-       __releases(&pkc->blk_fill_in_prog_lock)
-{
-       struct tpacket_kbdq_core *pkc  = GET_PBDQC_FROM_RB(rb);
-
-       read_unlock(&pkc->blk_fill_in_prog_lock);
-}
-
 static void prb_fill_rxhash(struct tpacket_kbdq_core *pkc,
                        struct tpacket3_hdr *ppd)
 {
@@ -1038,7 +1022,6 @@ static void prb_fill_curr_block(char *curr,
                                struct tpacket_kbdq_core *pkc,
                                struct tpacket_block_desc *pbd,
                                unsigned int len)
-       __acquires(&pkc->blk_fill_in_prog_lock)
 {
        struct tpacket3_hdr *ppd;

@@ -1048,7 +1031,7 @@ static void prb_fill_curr_block(char *curr,
        pkc->nxt_offset += TOTAL_PKT_LEN_INCL_ALIGN(len);
        BLOCK_LEN(pbd) += TOTAL_PKT_LEN_INCL_ALIGN(len);
        BLOCK_NUM_PKTS(pbd) += 1;
-       read_lock(&pkc->blk_fill_in_prog_lock);
+       atomic_inc(&pkc->pkbdq[pkc->kactive_blk_num].in_flight);
        prb_run_all_ft_ops(pkc, ppd);
 }

@@ -1063,6 +1046,7 @@ static void *__packet_lookup_frame_in_block(struct packet_sock *po,
        char *curr, *end;

        pkc = GET_PBDQC_FROM_RB(&po->rx_ring);
+       prb_publish_blocks(pkc, po);
        pbd = GET_CURR_PBLOCK_DESC_FROM_CORE(pkc);

        /* Queue is frozen when user space is lagging behind */
@@ -1071,7 +1055,7 @@ static void *__packet_lookup_frame_in_block(struct packet_sock *po,
                 * Check if that last block which caused the queue to freeze,
                 * is still in_use by user-space.
                 */
-               if (prb_curr_blk_in_use(pbd)) {
+               if (prb_curr_blk_in_use(pkc)) {
                        /* Can't record this packet */
                        return NULL;
                } else {
@@ -1091,7 +1075,7 @@ static void *__packet_lookup_frame_in_block(struct packet_sock *po,
        end = (char *)pbd + pkc->kblk_size;

        /* first try the current block */
-       if (curr+TOTAL_PKT_LEN_INCL_ALIGN(len) < end) {
+       if (curr + TOTAL_PKT_LEN_INCL_ALIGN(len) <= end) {
                prb_fill_curr_block(curr, pkc, pbd, len);
                return (void *)curr;
        }
@@ -1142,7 +1126,9 @@ static void *prb_lookup_block(const struct packet_sock *po,
        struct tpacket_kbdq_core *pkc  = GET_PBDQC_FROM_RB(rb);
        struct tpacket_block_desc *pbd = GET_PBLOCK_DESC(pkc, idx);

-       if (status != BLOCK_STATUS(pbd))
+       if (status == TP_STATUS_KERNEL && READ_ONCE(pkc->pkbdq[idx].last_pkt))
+               return NULL;
+       if (status != (READ_ONCE(BLOCK_STATUS(pbd)) & TP_STATUS_USER))
                return NULL;
        return pbd;
 }
@@ -1150,8 +1136,8 @@ static void *prb_lookup_block(const struct packet_sock *po,
 static int prb_previous_blk_num(struct packet_ring_buffer *rb)
 {
        unsigned int prev;
-       if (rb->prb_bdqc.kactive_blk_num)
-               prev = rb->prb_bdqc.kactive_blk_num-1;
+       if (rb->prb_bdqc.publish_blk_num)
+               prev = rb->prb_bdqc.publish_blk_num - 1;
        else
                prev = rb->prb_bdqc.knum_blocks-1;
        return prev;
@@ -1269,6 +1255,8 @@ static bool __tpacket_v3_has_room(const struct packet_sock *po, int pow_off)

        len = READ_ONCE(po->rx_ring.prb_bdqc.knum_blocks);
        idx = READ_ONCE(po->rx_ring.prb_bdqc.kactive_blk_num);
+       if (!prb_lookup_block(po, &po->rx_ring, idx, TP_STATUS_KERNEL))
+               return false;
        if (pow_off)
                idx += len >> pow_off;
        if (idx >= len)
@@ -2274,6 +2262,7 @@ static int tpacket_rcv(struct sk_buff *skb, struct net_device *dev,
        unsigned int slot_id = 0;
        int vnet_hdr_sz = 0;
        struct virtio_net_hdr vnet_hdr;
+       struct pgv *copy_block = NULL;

        /* struct tpacket{2,3}_hdr is aligned to a multiple of TPACKET_ALIGNMENT.
         * We may add members to them until current aligned size without forcing
@@ -2306,8 +2295,10 @@ static int tpacket_rcv(struct sk_buff *skb, struct net_device *dev,
        if (!res)
                goto drop_n_restore;

-       /* If we are flooded, just give up */
-       if (__packet_rcv_has_room(po, skb) == ROOM_NONE) {
+       /* V3 allocation must get a chance to publish completed pending blocks. */
+       if (__packet_rcv_has_room(po, skb) == ROOM_NONE &&
+           (po->tp_version != TPACKET_V3 ||
+            !READ_ONCE(po->rx_ring.prb_bdqc.pending_blocks))) {
                atomic_inc(&po->tp_drops);
                goto drop_n_restore;
        }
@@ -2389,6 +2380,11 @@ static int tpacket_rcv(struct sk_buff *skb, struct net_device *dev,
                                        TP_STATUS_KERNEL, (macoff+snaplen));
        if (!h.raw)
                goto drop_n_account;
+       if (po->tp_version == TPACKET_V3) {
+               struct tpacket_kbdq_core *pkc = &po->rx_ring.prb_bdqc;
+
+               copy_block = &pkc->pkbdq[pkc->kactive_blk_num];
+       }

        if (po->tp_version <= TPACKET_V2) {
                slot_id = po->rx_ring.head;
@@ -2518,7 +2514,8 @@ static int tpacket_rcv(struct sk_buff *skb, struct net_device *dev,
                spin_unlock(&sk->sk_receive_queue.lock);
                sk->sk_data_ready(sk);
        } else if (po->tp_version == TPACKET_V3) {
-               prb_clear_blk_fill_status(&po->rx_ring);
+               /* No frame accesses may follow this release. */
+               atomic_dec_return_release(&copy_block->in_flight);
        }

 drop_n_restore:
@@ -4376,8 +4373,9 @@ static __poll_t packet_poll(struct file *file, struct socket *sock,

        spin_lock_bh(&sk->sk_receive_queue.lock);
        if (po->rx_ring.pg_vec) {
-               if (!packet_previous_rx_frame(po, &po->rx_ring,
-                       TP_STATUS_KERNEL))
+               if (po->tp_version == TPACKET_V3 ?
+                   !!packet_previous_rx_frame(po, &po->rx_ring, TP_STATUS_USER) :
+                   !packet_previous_rx_frame(po, &po->rx_ring, TP_STATUS_KERNEL))
                        mask |= EPOLLIN | EPOLLRDNORM;
        }
        __packet_rcv_try_clear_pressure(po);
diff --git a/net/packet/internal.h b/net/packet/internal.h
index f5c8cd0eed2c..bece88a9672c 100644
--- a/net/packet/internal.h
+++ b/net/packet/internal.h
@@ -20,7 +20,7 @@ struct tpacket_kbdq_core {
        unsigned int    feature_req_word;
        unsigned int    hdrlen;
        unsigned char   reset_pending_on_curr_blk;
-       unsigned short  kactive_blk_num;
+       unsigned int    kactive_blk_num;
        unsigned int    blk_sizeof_priv;

        unsigned short  version;
@@ -36,7 +36,8 @@ struct tpacket_kbdq_core {

        struct sk_buff  *skb;

-       rwlock_t        blk_fill_in_prog_lock;
+       unsigned int    publish_blk_num;
+       unsigned int    pending_blocks;

        /* Default is set to 8ms */
 #define DEFAULT_PRB_RETIRE_TOV (8)
@@ -49,6 +50,10 @@ struct tpacket_kbdq_core {

 struct pgv {
        char *buffer;
+       /* RX V3 only; last_pkt marks a sealed, unpublished block. */
+       struct tpacket3_hdr *last_pkt;
+       atomic_t in_flight;
+       unsigned int retire_status;
 };

 struct packet_ring_buffer {
--
2.25.1

________________________________


本电子邮件及转载的任何文件都属绝密文件，并且仅供标明地址的个人或团体使用。如果您收到的电子邮件存在错误，请通知本公司，删除全部原始信息（包括任何原始信息的备份），并且禁止散布或分发全部或部分原始信息。本公司不担保本电子邮件中信息的准确性、适当性或完整性，并且对此产生的任何错误或疏忽不承担任何责任。在正式协议签订以前，这里提到的任何参考术语、与主题相关的情形以及随后的电子邮件中的谈判，都将严格“遵守合同”。接收方应在接收电子邮件或任何附件时检查有无病毒，本公司对由于转载本电子邮件而引发病毒产生的任何损坏不承担任何责任。


Reproduced in this e-mail and any files are considered top-secret documents, and only indicate the address of individuals or groups. If you receive an e-mail there is an error, please notify the Company, delete all the original information (including any backup of the original information), and the prohibition of the dissemination or distribution of all or part of the original information. The Company does not guarantee the accuracy of the information in this e-mail, adequacy, or completeness, and this produces any error or omission does not bear any responsibility. The formal agreement was signed prior to any reference to the term mentioned here, and the circumstances relevant to the subject and subsequent e-mail in the negotiations will be strictly "comply with the contract." The receiver should be receiving an e-mail or any attachment, check for viruses. The company reproduced this e-mail because of the virus caused any damage resulting from not bear any responsibility.

Re: [RFC net v2 1/2] af_packet: prepare the virtio header before reserving an RX frame

[Willem de Bruijn]

Jingxuan Xie wrote:
> From: jingxuanxie <jingxuanxie@deeproute.ai>
> 
> An RX virtio-header conversion can fail after packet_current_rx_frame()
> has reserved space in a TPACKET_V3 block and increased its packet count.
> Releasing the copy lock on that error path does not remove the incomplete
> record from the block.

Can you give a little more detail on the exact bug: what state does
not get reverted on error, and how problematic is this.

> Prepare the header in a local variable before reserving a frame, and copy
> it to the reserved frame only after conversion succeeds. Keep drop
> accounting, notification and copy_skb cleanup on conversion failure.
> This also keeps conversion out of the receive queue critical section.
> 
> Assisted-by: LLM
> Signed-off-by: Jingxuan Xie <jingxuanxie@deeproute.ai>
> ---
>  net/packet/af_packet.c | 22 ++++++++++++----------
>  1 file changed, 12 insertions(+), 10 deletions(-)
> 
> diff --git a/net/packet/af_packet.c b/net/packet/af_packet.c
> index 823e0911c910..7b3e1e27f777 100644
> --- a/net/packet/af_packet.c
> +++ b/net/packet/af_packet.c
> @@ -2273,6 +2273,7 @@ static int tpacket_rcv(struct sk_buff *skb, struct net_device *dev,
>  	__u32 ts_status;
>  	unsigned int slot_id = 0;
>  	int vnet_hdr_sz = 0;
> +	struct virtio_net_hdr vnet_hdr;
>  
>  	/* struct tpacket{2,3}_hdr is aligned to a multiple of TPACKET_ALIGNMENT.
>  	 * We may add members to them until current aligned size without forcing
> @@ -2375,6 +2376,14 @@ static int tpacket_rcv(struct sk_buff *skb, struct net_device *dev,
>  			vnet_hdr_sz = 0;
>  		}
>  	}
> +	if (vnet_hdr_sz &&
> +	    virtio_net_hdr_from_skb(skb, &vnet_hdr, vio_le(), true, 0)) {
> +		drop_reason = SKB_DROP_REASON_PACKET_SOCK_ERROR;
> +		atomic_inc(&po->tp_drops);
> +		sk->sk_data_ready(sk);
> +		sk_skb_reason_drop(sk, copy_skb, drop_reason);
> +		goto drop_n_restore;

Probably prefer a goto drop_n_account_unlocked rather than open coding
that here.

> +	}
>  	spin_lock(&sk->sk_receive_queue.lock);
>  	h.raw = packet_current_rx_frame(po, skb,
>  					TP_STATUS_KERNEL, (macoff+snaplen));
> @@ -2388,16 +2397,9 @@ static int tpacket_rcv(struct sk_buff *skb, struct net_device *dev,
>  		__set_bit(slot_id, po->rx_ring.rx_owner_map);
>  	}
>  
> -	if (vnet_hdr_sz &&
> -	    virtio_net_hdr_from_skb(skb, h.raw + macoff -
> -				    sizeof(struct virtio_net_hdr),
> -				    vio_le(), true, 0)) {
> -		if (po->tp_version <= TPACKET_V2)
> -			__clear_bit(slot_id, po->rx_ring.rx_owner_map);
> -		else
> -			prb_clear_blk_fill_status(&po->rx_ring);
> -		goto drop_n_account;
> -	}
> +	if (vnet_hdr_sz)
> +		memcpy(h.raw + macoff - sizeof(vnet_hdr), &vnet_hdr,
> +		       sizeof(vnet_hdr));
>  
>  	if (po->tp_version <= TPACKET_V2) {
>  		packet_increment_rx_head(po, &po->rx_ring);
> -- 
> 2.25.1