[PATCH v4 0/3] can: rx-offload: make skb_irq_queue per-CPU

[PATCH v4 0/3] can: rx-offload: make skb_irq_queue per-CPU

[Ciprian Costea]

From: Ciprian Marian Costea <ciprianmarian.costea@oss.nxp.com>

can_rx_offload keeps a lockless irq_queue that the IRQ handlers fill and
that is later spliced under skb_queue.lock into the NAPI-facing skb_queue.
This works as long as a single context fills the irq_queue. flexcan with
FLEXCAN_QUIRK_SECONDARY_MB_IRQ and mcf5441x use two mailbox IRQ lines. When
those are affined to different CPUs the two handlers can enqueue into the
same list at the same time and corrupt it.

This series:
  1. Makes the irq_queue per-CPU so the handlers no longer share a list.
  2. Fixes at91_can rx-offload teardown.
  3. Checks the can_rx_offload_add_manual() return value in gs_usb.

Changes since v3:

- In gs_usb driver, check the can_rx_offload_add_manual() return value,
  the same NULL-deref the per-CPU change exposes.

Changes since v2:

- at91_can: also add can_rx_offload_del() on the register_candev() error
  path and check the can_rx_offload_add_timestamp() return value.

Changes since v1:

- The enqueue helpers used this_cpu_ptr() without disabling preemption.
  All four enqueue helpers now use get_cpu_ptr()/put_cpu_ptr().
- Guard can_rx_offload_del() against skb_irq_queue == NULL.
- Fix 'at91_can' memory leak by adding missing 'can_rx_offload_del'.

Ciprian Marian Costea (3):
  can: rx-offload: make skb_irq_queue per-CPU
  can: at91_can: fix rx-offload cleanup on unbind and probe errors
  can: gs_usb: check can_rx_offload_add_manual() return value

 drivers/net/can/at91_can.c       | 10 +++-
 drivers/net/can/dev/rx-offload.c | 83 ++++++++++++++++++++++++++------
 drivers/net/can/usb/gs_usb.c     |  5 +-
 include/linux/can/rx-offload.h   |  2 +-
 4 files changed, 82 insertions(+), 18 deletions(-)

-- 
2.43.0

[PATCH v4 1/3] can: rx-offload: make skb_irq_queue per-CPU

[Ciprian Costea]

From: Ciprian Marian Costea <ciprianmarian.costea@oss.nxp.com>

skb_irq_queue is filled by the IRQ handlers using the lockless
__skb_queue_add_sort() / __skb_queue_tail() helpers and later spliced
into skb_queue under skb_queue.lock by can_rx_offload_irq_finish() and
can_rx_offload_threaded_irq_finish().

This is only safe while a single context fills skb_irq_queue. FlexCAN
on NXP S32G2 (FLEXCAN_QUIRK_SECONDARY_MB_IRQ) uses two mailbox IRQ
lines, one for MB0-7 and one for MB8-63; MCF5441X similarly splits its
mailbox interrupt. When these lines are affined to different CPUs both
handlers can run at the same time and enqueue into the same sk_buff_head
concurrently, corrupting its list.

Allocate skb_irq_queue per-CPU so the handlers no longer share a list,
keeping the enqueue path lock-free. Access the per-CPU queue via
get_cpu_ptr()/put_cpu_ptr() in the enqueue helpers: this disables
preemption around the lockless __skb_queue_*() operation.

can_rx_offload_irq_finish() runs in the same context as its enqueues and
splices this_cpu_ptr(). can_rx_offload_threaded_irq_finish() may have
been migrated after its enqueues, so it splices every possible CPU's
queue; this is safe because each per-CPU queue has a single producer and
that producer runs with preemption disabled, so it cannot race the
splice.

Cross-line frames are now sorted by timestamp only within a CPU's queue
and appended across CPUs on splice; each skb keeps its own timestamp.

Fixes: c757096ea103 ("can: rx-offload: add skb queue for use during ISR")
Signed-off-by: Ciprian Marian Costea <ciprianmarian.costea@oss.nxp.com>
---
 drivers/net/can/dev/rx-offload.c | 83 ++++++++++++++++++++++++++------
 include/linux/can/rx-offload.h   |  2 +-
 2 files changed, 70 insertions(+), 15 deletions(-)

diff --git a/drivers/net/can/dev/rx-offload.c b/drivers/net/can/dev/rx-offload.c
index 46e7b6db4a1e..649bfda08b65 100644
--- a/drivers/net/can/dev/rx-offload.c
+++ b/drivers/net/can/dev/rx-offload.c
@@ -7,6 +7,7 @@
 
 #include <linux/can/dev.h>
 #include <linux/can/rx-offload.h>
+#include <linux/percpu.h>
 
 struct can_rx_offload_cb {
 	u32 timestamp;
@@ -175,9 +176,18 @@ can_rx_offload_offload_one(struct can_rx_offload *offload, unsigned int n)
 int can_rx_offload_irq_offload_timestamp(struct can_rx_offload *offload,
 					 u64 pending)
 {
+	struct sk_buff_head *irq_queue;
 	unsigned int i;
 	int received = 0;
 
+	/*
+	 * get_cpu_ptr() disables preemption so that the lockless
+	 * __skb_queue_*() below operate on the current CPU's queue without
+	 * racing a migration. This also keeps this_cpu_ptr() valid when a
+	 * driver enqueues from a preemptible (threaded IRQ) context.
+	 */
+	irq_queue = get_cpu_ptr(offload->skb_irq_queue);
+
 	for (i = offload->mb_first;
 	     can_rx_offload_le(offload, i, offload->mb_last);
 	     can_rx_offload_inc(offload, &i)) {
@@ -190,20 +200,25 @@ int can_rx_offload_irq_offload_timestamp(struct can_rx_offload *offload,
 		if (IS_ERR_OR_NULL(skb))
 			continue;
 
-		__skb_queue_add_sort(&offload->skb_irq_queue, skb,
+		__skb_queue_add_sort(irq_queue, skb,
 				     can_rx_offload_compare);
 		received++;
 	}
 
+	put_cpu_ptr(offload->skb_irq_queue);
+
 	return received;
 }
 EXPORT_SYMBOL_GPL(can_rx_offload_irq_offload_timestamp);
 
 int can_rx_offload_irq_offload_fifo(struct can_rx_offload *offload)
 {
+	struct sk_buff_head *irq_queue;
 	struct sk_buff *skb;
 	int received = 0;
 
+	irq_queue = get_cpu_ptr(offload->skb_irq_queue);
+
 	while (1) {
 		skb = can_rx_offload_offload_one(offload, 0);
 		if (IS_ERR(skb))
@@ -211,10 +226,12 @@ int can_rx_offload_irq_offload_fifo(struct can_rx_offload *offload)
 		if (!skb)
 			break;
 
-		__skb_queue_tail(&offload->skb_irq_queue, skb);
+		__skb_queue_tail(irq_queue, skb);
 		received++;
 	}
 
+	put_cpu_ptr(offload->skb_irq_queue);
+
 	return received;
 }
 EXPORT_SYMBOL_GPL(can_rx_offload_irq_offload_fifo);
@@ -222,6 +239,7 @@ EXPORT_SYMBOL_GPL(can_rx_offload_irq_offload_fifo);
 int can_rx_offload_queue_timestamp(struct can_rx_offload *offload,
 				   struct sk_buff *skb, u32 timestamp)
 {
+	struct sk_buff_head *irq_queue;
 	struct can_rx_offload_cb *cb;
 
 	if (skb_queue_len(&offload->skb_queue) >
@@ -233,8 +251,9 @@ int can_rx_offload_queue_timestamp(struct can_rx_offload *offload,
 	cb = can_rx_offload_get_cb(skb);
 	cb->timestamp = timestamp;
 
-	__skb_queue_add_sort(&offload->skb_irq_queue, skb,
-			     can_rx_offload_compare);
+	irq_queue = get_cpu_ptr(offload->skb_irq_queue);
+	__skb_queue_add_sort(irq_queue, skb, can_rx_offload_compare);
+	put_cpu_ptr(offload->skb_irq_queue);
 
 	return 0;
 }
@@ -268,13 +287,17 @@ EXPORT_SYMBOL_GPL(can_rx_offload_get_echo_skb_queue_timestamp);
 int can_rx_offload_queue_tail(struct can_rx_offload *offload,
 			      struct sk_buff *skb)
 {
+	struct sk_buff_head *irq_queue;
+
 	if (skb_queue_len(&offload->skb_queue) >
 	    offload->skb_queue_len_max) {
 		dev_kfree_skb_any(skb);
 		return -ENOBUFS;
 	}
 
-	__skb_queue_tail(&offload->skb_irq_queue, skb);
+	irq_queue = get_cpu_ptr(offload->skb_irq_queue);
+	__skb_queue_tail(irq_queue, skb);
+	put_cpu_ptr(offload->skb_irq_queue);
 
 	return 0;
 }
@@ -307,14 +330,15 @@ EXPORT_SYMBOL_GPL(can_rx_offload_get_echo_skb_queue_tail);
 
 void can_rx_offload_irq_finish(struct can_rx_offload *offload)
 {
+	struct sk_buff_head *irq_queue = this_cpu_ptr(offload->skb_irq_queue);
 	unsigned long flags;
 	int queue_len;
 
-	if (skb_queue_empty_lockless(&offload->skb_irq_queue))
+	if (skb_queue_empty_lockless(irq_queue))
 		return;
 
 	spin_lock_irqsave(&offload->skb_queue.lock, flags);
-	skb_queue_splice_tail_init(&offload->skb_irq_queue, &offload->skb_queue);
+	skb_queue_splice_tail_init(irq_queue, &offload->skb_queue);
 	spin_unlock_irqrestore(&offload->skb_queue.lock, flags);
 
 	queue_len = skb_queue_len(&offload->skb_queue);
@@ -330,15 +354,29 @@ void can_rx_offload_threaded_irq_finish(struct can_rx_offload *offload)
 {
 	unsigned long flags;
 	int queue_len;
-
-	if (skb_queue_empty_lockless(&offload->skb_irq_queue))
-		return;
-
+	int cpu;
+
+	/*
+	 * Splice every CPU's queue: unlike the non-threaded
+	 * can_rx_offload_irq_finish(), a threaded handler may be migrated
+	 * between the enqueue and this splice, so the frames may sit on a
+	 * different CPU's queue. This is only safe because a given per-CPU
+	 * queue has a single producer (the enqueue on that CPU is
+	 * non-preemptible), so no producer can race this splice.
+	 */
 	spin_lock_irqsave(&offload->skb_queue.lock, flags);
-	skb_queue_splice_tail_init(&offload->skb_irq_queue, &offload->skb_queue);
+	for_each_possible_cpu(cpu) {
+		struct sk_buff_head *irq_queue;
+
+		irq_queue = per_cpu_ptr(offload->skb_irq_queue, cpu);
+		skb_queue_splice_tail_init(irq_queue, &offload->skb_queue);
+	}
 	spin_unlock_irqrestore(&offload->skb_queue.lock, flags);
 
 	queue_len = skb_queue_len(&offload->skb_queue);
+	if (!queue_len)
+		return;
+
 	if (queue_len > offload->skb_queue_len_max / 8)
 		netdev_dbg(offload->dev, "%s: queue_len=%d\n",
 			   __func__, queue_len);
@@ -353,13 +391,21 @@ static int can_rx_offload_init_queue(struct net_device *dev,
 				     struct can_rx_offload *offload,
 				     unsigned int weight)
 {
+	int cpu;
+
 	offload->dev = dev;
 
 	/* Limit queue len to 4x the weight (rounded to next power of two) */
 	offload->skb_queue_len_max = 2 << fls(weight);
 	offload->skb_queue_len_max *= 4;
 	skb_queue_head_init(&offload->skb_queue);
-	__skb_queue_head_init(&offload->skb_irq_queue);
+
+	offload->skb_irq_queue = alloc_percpu(struct sk_buff_head);
+	if (!offload->skb_irq_queue)
+		return -ENOMEM;
+
+	for_each_possible_cpu(cpu)
+		__skb_queue_head_init(per_cpu_ptr(offload->skb_irq_queue, cpu));
 
 	netif_napi_add_weight(dev, &offload->napi, can_rx_offload_napi_poll,
 			      weight);
@@ -420,8 +466,17 @@ EXPORT_SYMBOL_GPL(can_rx_offload_enable);
 
 void can_rx_offload_del(struct can_rx_offload *offload)
 {
+	int cpu;
+
 	netif_napi_del(&offload->napi);
 	skb_queue_purge(&offload->skb_queue);
-	__skb_queue_purge(&offload->skb_irq_queue);
+
+	if (!offload->skb_irq_queue)
+		return;
+
+	for_each_possible_cpu(cpu)
+		__skb_queue_purge(per_cpu_ptr(offload->skb_irq_queue, cpu));
+
+	free_percpu(offload->skb_irq_queue);
 }
 EXPORT_SYMBOL_GPL(can_rx_offload_del);
diff --git a/include/linux/can/rx-offload.h b/include/linux/can/rx-offload.h
index d29bb4521947..1b9e2a8ab39a 100644
--- a/include/linux/can/rx-offload.h
+++ b/include/linux/can/rx-offload.h
@@ -20,7 +20,7 @@ struct can_rx_offload {
 					bool drop);
 
 	struct sk_buff_head skb_queue;
-	struct sk_buff_head skb_irq_queue;
+	struct sk_buff_head __percpu *skb_irq_queue;
 	u32 skb_queue_len_max;
 
 	unsigned int mb_first;
-- 
2.43.0

[PATCH v4 2/3] can: at91_can: fix rx-offload cleanup on unbind and probe errors

[Ciprian Costea]

From: Ciprian Marian Costea <ciprianmarian.costea@oss.nxp.com>

Making can_rx_offload's skb_irq_queue per-CPU (previous patch) adds an
alloc_percpu() to can_rx_offload_add_*(). That allocation has to be freed
on teardown and can fail with -ENOMEM, which exposes three problems in
at91_can:

- at91_can_remove() does not call can_rx_offload_del(), so the NAPI
  instance and the per-CPU queues are leaked on unbind and module removal.

- The probe error path after a failed register_candev() jumps straight to
  free_candev() without can_rx_offload_del() and leaks the same objects.

- The return value of can_rx_offload_add_timestamp() is ignored. It can now
  return -ENOMEM with offload->skb_irq_queue == NULL, probe still succeeds,
  and the first RX interrupt dereferences that NULL pointer in
  can_rx_offload_irq_offload_timestamp() via get_cpu_ptr().

Check the return value and call can_rx_offload_del() from at91_can_remove()
and from a new error label taken when register_candev() fails.

Fixes: 137f59d5dab4 ("can: at91_can: switch to rx-offload implementation")
Signed-off-by: Ciprian Marian Costea <ciprianmarian.costea@oss.nxp.com>
---
 drivers/net/can/at91_can.c | 10 ++++++++--
 1 file changed, 8 insertions(+), 2 deletions(-)

diff --git a/drivers/net/can/at91_can.c b/drivers/net/can/at91_can.c
index 58da323f14d7..3f6c5bb373d3 100644
--- a/drivers/net/can/at91_can.c
+++ b/drivers/net/can/at91_can.c
@@ -1123,7 +1123,9 @@ static int at91_can_probe(struct platform_device *pdev)
 	priv->offload.mb_first = devtype_data->rx_first;
 	priv->offload.mb_last = devtype_data->rx_last;
 
-	can_rx_offload_add_timestamp(dev, &priv->offload);
+	err = can_rx_offload_add_timestamp(dev, &priv->offload);
+	if (err)
+		goto exit_free;
 
 	if (transceiver)
 		priv->can.bitrate_max = transceiver->attrs.max_link_rate;
@@ -1137,7 +1139,7 @@ static int at91_can_probe(struct platform_device *pdev)
 	err = register_candev(dev);
 	if (err) {
 		dev_err(&pdev->dev, "registering netdev failed\n");
-		goto exit_free;
+		goto exit_offload;
 	}
 
 	dev_info(&pdev->dev, "device registered (reg_base=%p, irq=%d)\n",
@@ -1145,6 +1147,8 @@ static int at91_can_probe(struct platform_device *pdev)
 
 	return 0;
 
+ exit_offload:
+	can_rx_offload_del(&priv->offload);
  exit_free:
 	free_candev(dev);
  exit_iounmap:
@@ -1165,6 +1169,8 @@ static void at91_can_remove(struct platform_device *pdev)
 
 	unregister_netdev(dev);
 
+	can_rx_offload_del(&priv->offload);
+
 	iounmap(priv->reg_base);
 
 	res = platform_get_resource(pdev, IORESOURCE_MEM, 0);
-- 
2.43.0

[PATCH v4 3/3] can: gs_usb: check can_rx_offload_add_manual() return value

[Ciprian Costea]

From: Ciprian Marian Costea <ciprianmarian.costea@oss.nxp.com>

gs_make_candev() ignores the return value of can_rx_offload_add_manual().
Once can_rx_offload's skb_irq_queue is made per-CPU, that call allocates a
per-CPU queue and can fail with -ENOMEM, leaving offload->skb_irq_queue ==
NULL.

Check the return value and exit via the existing free_candev() on error
path.

Fixes: 24bc41b45583 ("can: gs_usb: convert to NAPI/rx-offload to avoid OoO reception")
Signed-off-by: Ciprian Marian Costea <ciprianmarian.costea@oss.nxp.com>
---
 drivers/net/can/usb/gs_usb.c | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/drivers/net/can/usb/gs_usb.c b/drivers/net/can/usb/gs_usb.c
index 3b9b2f104d86..cac4ce323a1d 100644
--- a/drivers/net/can/usb/gs_usb.c
+++ b/drivers/net/can/usb/gs_usb.c
@@ -1487,7 +1487,10 @@ static struct gs_can *gs_make_candev(unsigned int channel,
 		dev->can.fd.data_bittiming_const = &dev->data_bt_const;
 	}
 
-	can_rx_offload_add_manual(netdev, &dev->offload, GS_NAPI_WEIGHT);
+	rc = can_rx_offload_add_manual(netdev, &dev->offload, GS_NAPI_WEIGHT);
+	if (rc)
+		goto out_free_candev;
+
 	SET_NETDEV_DEV(netdev, &intf->dev);
 
 	rc = register_candev(dev->netdev);
-- 
2.43.0

Re: [PATCH v4 1/3] can: rx-offload: make skb_irq_queue per-CPU

[Bough Chen]

On Tue, Sep 01, 2026 at 01:48:46PM +0200, Ciprian Costea wrote:
> From: Ciprian Marian Costea <ciprianmarian.costea@oss.nxp.com>
> 
> skb_irq_queue is filled by the IRQ handlers using the lockless
> __skb_queue_add_sort() / __skb_queue_tail() helpers and later spliced
> into skb_queue under skb_queue.lock by can_rx_offload_irq_finish() and
> can_rx_offload_threaded_irq_finish().
> 
> This is only safe while a single context fills skb_irq_queue. FlexCAN
> on NXP S32G2 (FLEXCAN_QUIRK_SECONDARY_MB_IRQ) uses two mailbox IRQ
> lines, one for MB0-7 and one for MB8-63; MCF5441X similarly splits its
> mailbox interrupt. When these lines are affined to different CPUs both
> handlers can run at the same time and enqueue into the same sk_buff_head
> concurrently, corrupting its list.
> 
> Allocate skb_irq_queue per-CPU so the handlers no longer share a list,
> keeping the enqueue path lock-free. Access the per-CPU queue via
> get_cpu_ptr()/put_cpu_ptr() in the enqueue helpers: this disables
> preemption around the lockless __skb_queue_*() operation.
> 
> can_rx_offload_irq_finish() runs in the same context as its enqueues and
> splices this_cpu_ptr(). can_rx_offload_threaded_irq_finish() may have
> been migrated after its enqueues, so it splices every possible CPU's
> queue; this is safe because each per-CPU queue has a single producer and
> that producer runs with preemption disabled, so it cannot race the
> splice.
> 
> Cross-line frames are now sorted by timestamp only within a CPU's queue
> and appended across CPUs on splice; each skb keeps its own timestamp.
> 
> Fixes: c757096ea103 ("can: rx-offload: add skb queue for use during ISR")
> Signed-off-by: Ciprian Marian Costea <ciprianmarian.costea@oss.nxp.com>
> ---
>  drivers/net/can/dev/rx-offload.c | 83 ++++++++++++++++++++++++++------
>  include/linux/can/rx-offload.h   |  2 +-
>  2 files changed, 70 insertions(+), 15 deletions(-)
> 
> diff --git a/drivers/net/can/dev/rx-offload.c b/drivers/net/can/dev/rx-offload.c
> index 46e7b6db4a1e..649bfda08b65 100644
> --- a/drivers/net/can/dev/rx-offload.c
> +++ b/drivers/net/can/dev/rx-offload.c
> @@ -7,6 +7,7 @@
>  
>  #include <linux/can/dev.h>
>  #include <linux/can/rx-offload.h>
> +#include <linux/percpu.h>
>  
>  struct can_rx_offload_cb {
>  	u32 timestamp;
> @@ -175,9 +176,18 @@ can_rx_offload_offload_one(struct can_rx_offload *offload, unsigned int n)
>  int can_rx_offload_irq_offload_timestamp(struct can_rx_offload *offload,
>  					 u64 pending)
>  {
> +	struct sk_buff_head *irq_queue;
>  	unsigned int i;
>  	int received = 0;
>  
> +	/*
> +	 * get_cpu_ptr() disables preemption so that the lockless
> +	 * __skb_queue_*() below operate on the current CPU's queue without
> +	 * racing a migration. This also keeps this_cpu_ptr() valid when a
> +	 * driver enqueues from a preemptible (threaded IRQ) context.
> +	 */
> +	irq_queue = get_cpu_ptr(offload->skb_irq_queue);
> +
>  	for (i = offload->mb_first;
>  	     can_rx_offload_le(offload, i, offload->mb_last);
>  	     can_rx_offload_inc(offload, &i)) {
> @@ -190,20 +200,25 @@ int can_rx_offload_irq_offload_timestamp(struct can_rx_offload *offload,
>  		if (IS_ERR_OR_NULL(skb))
>  			continue;
>  
> -		__skb_queue_add_sort(&offload->skb_irq_queue, skb,
> +		__skb_queue_add_sort(irq_queue, skb,
>  				     can_rx_offload_compare);
>  		received++;
>  	}
>  
> +	put_cpu_ptr(offload->skb_irq_queue);
> +
>  	return received;
>  }
>  EXPORT_SYMBOL_GPL(can_rx_offload_irq_offload_timestamp);
>  
>  int can_rx_offload_irq_offload_fifo(struct can_rx_offload *offload)
>  {
> +	struct sk_buff_head *irq_queue;
>  	struct sk_buff *skb;
>  	int received = 0;
>  
> +	irq_queue = get_cpu_ptr(offload->skb_irq_queue);
> +
>  	while (1) {
>  		skb = can_rx_offload_offload_one(offload, 0);
>  		if (IS_ERR(skb))
> @@ -211,10 +226,12 @@ int can_rx_offload_irq_offload_fifo(struct can_rx_offload *offload)
>  		if (!skb)
>  			break;
>  
> -		__skb_queue_tail(&offload->skb_irq_queue, skb);
> +		__skb_queue_tail(irq_queue, skb);
>  		received++;
>  	}
>  
> +	put_cpu_ptr(offload->skb_irq_queue);
> +
>  	return received;
>  }
>  EXPORT_SYMBOL_GPL(can_rx_offload_irq_offload_fifo);
> @@ -222,6 +239,7 @@ EXPORT_SYMBOL_GPL(can_rx_offload_irq_offload_fifo);
>  int can_rx_offload_queue_timestamp(struct can_rx_offload *offload,
>  				   struct sk_buff *skb, u32 timestamp)
>  {
> +	struct sk_buff_head *irq_queue;
>  	struct can_rx_offload_cb *cb;
>  
>  	if (skb_queue_len(&offload->skb_queue) >
> @@ -233,8 +251,9 @@ int can_rx_offload_queue_timestamp(struct can_rx_offload *offload,
>  	cb = can_rx_offload_get_cb(skb);
>  	cb->timestamp = timestamp;
>  
> -	__skb_queue_add_sort(&offload->skb_irq_queue, skb,
> -			     can_rx_offload_compare);
> +	irq_queue = get_cpu_ptr(offload->skb_irq_queue);
> +	__skb_queue_add_sort(irq_queue, skb, can_rx_offload_compare);
> +	put_cpu_ptr(offload->skb_irq_queue);
>  
>  	return 0;
>  }
> @@ -268,13 +287,17 @@ EXPORT_SYMBOL_GPL(can_rx_offload_get_echo_skb_queue_timestamp);
>  int can_rx_offload_queue_tail(struct can_rx_offload *offload,
>  			      struct sk_buff *skb)
>  {
> +	struct sk_buff_head *irq_queue;
> +
>  	if (skb_queue_len(&offload->skb_queue) >
>  	    offload->skb_queue_len_max) {
>  		dev_kfree_skb_any(skb);
>  		return -ENOBUFS;
>  	}
>  
> -	__skb_queue_tail(&offload->skb_irq_queue, skb);
> +	irq_queue = get_cpu_ptr(offload->skb_irq_queue);
> +	__skb_queue_tail(irq_queue, skb);
> +	put_cpu_ptr(offload->skb_irq_queue);
>  
>  	return 0;
>  }
> @@ -307,14 +330,15 @@ EXPORT_SYMBOL_GPL(can_rx_offload_get_echo_skb_queue_tail);
>  
>  void can_rx_offload_irq_finish(struct can_rx_offload *offload)
>  {
> +	struct sk_buff_head *irq_queue = this_cpu_ptr(offload->skb_irq_queue);
>  	unsigned long flags;
>  	int queue_len;
>  
> -	if (skb_queue_empty_lockless(&offload->skb_irq_queue))
> +	if (skb_queue_empty_lockless(irq_queue))
>  		return;
>  
>  	spin_lock_irqsave(&offload->skb_queue.lock, flags);
> -	skb_queue_splice_tail_init(&offload->skb_irq_queue, &offload->skb_queue);
> +	skb_queue_splice_tail_init(irq_queue, &offload->skb_queue);
>  	spin_unlock_irqrestore(&offload->skb_queue.lock, flags);
>  
>  	queue_len = skb_queue_len(&offload->skb_queue);
> @@ -330,15 +354,29 @@ void can_rx_offload_threaded_irq_finish(struct can_rx_offload *offload)
>  {
>  	unsigned long flags;
>  	int queue_len;
> -
> -	if (skb_queue_empty_lockless(&offload->skb_irq_queue))
> -		return;
> -
> +	int cpu;
> +
> +	/*
> +	 * Splice every CPU's queue: unlike the non-threaded
> +	 * can_rx_offload_irq_finish(), a threaded handler may be migrated
> +	 * between the enqueue and this splice, so the frames may sit on a
> +	 * different CPU's queue. This is only safe because a given per-CPU
> +	 * queue has a single producer (the enqueue on that CPU is
> +	 * non-preemptible), so no producer can race this splice.
> +	 */
>  	spin_lock_irqsave(&offload->skb_queue.lock, flags);
> -	skb_queue_splice_tail_init(&offload->skb_irq_queue, &offload->skb_queue);
> +	for_each_possible_cpu(cpu) {
> +		struct sk_buff_head *irq_queue;
> +
> +		irq_queue = per_cpu_ptr(offload->skb_irq_queue, cpu);
> +		skb_queue_splice_tail_init(irq_queue, &offload->skb_queue);
> +	}
>  	spin_unlock_irqrestore(&offload->skb_queue.lock, flags);
>  
Hi Ciprian,
The fix looks correct to me. I checked all rx-offload users in
drivers/net/can/ and the change is safe for every current driver.

One suggestion:
the cross-CPU splice in can_rx_offload_threaded_irq_finish() is safe
only as long as there is a single threaded handler context per offload
instance, so each per-CPU queue has exactly one producer.
This isn't new (the old shared skb_irq_queue relied on the same
"single context fills the queue" assumption), and for the current
threaded users it's actually enforced by genirq:
they all use request_threaded_irq(irq, NULL, handler, ...),
which mandates IRQF_ONESHOT, so the handler can't re-enter.

Only the threaded finish path cares about this, and all three such
drivers request the IRQ with IRQF_ONESHOT:
  - m_can (peripheral)
  - mcp251xfd
  - nct6694_canfd
They all use the manual enqueue path (queue_timestamp/queue_tail), not
irq_offload_*(). Everyone else uses the non-threaded irq_finish()
(this_cpu_ptr only) and is safe by construction.

Could you spell out this assumption in the comment above the
for_each_possible_cpu() loop? e.g.:

This assumes a single threaded handler context per offload instance
(IRQ requested with IRQF_ONESHOT / handler non-reentrant), so each
per-CPU queue has exactly one producer. If that changes, this
cross-CPU splice of lockless queues would need additional locking.

Minor, non-blocking: get_cpu_ptr() only wraps a single enqueue, so a
handler that drains several frames per IRQ (e.g. mcp251xfd) can migrate
mid-batch and split a burst across CPU queues, losing intra-batch
timestamp order after the splice. Just as the Sashiko reveiw in your
V2. I think it is harmless, and SocketCAN doesn't guarantee delivery
order anyway, so I'm fine with it as-is.
Point it out just in case other people may have comment on it.

With the comment clarification:

Reviewed-by: Haibo Chen <haibo.chen@nxp.com>

Regards
Haibo Chen

>  	queue_len = skb_queue_len(&offload->skb_queue);
> +	if (!queue_len)
> +		return;
> +
>  	if (queue_len > offload->skb_queue_len_max / 8)
>  		netdev_dbg(offload->dev, "%s: queue_len=%d\n",
>  			   __func__, queue_len);
> @@ -353,13 +391,21 @@ static int can_rx_offload_init_queue(struct net_device *dev,
>  				     struct can_rx_offload *offload,
>  				     unsigned int weight)
>  {
> +	int cpu;
> +
>  	offload->dev = dev;
>  
>  	/* Limit queue len to 4x the weight (rounded to next power of two) */
>  	offload->skb_queue_len_max = 2 << fls(weight);
>  	offload->skb_queue_len_max *= 4;
>  	skb_queue_head_init(&offload->skb_queue);
> -	__skb_queue_head_init(&offload->skb_irq_queue);
> +
> +	offload->skb_irq_queue = alloc_percpu(struct sk_buff_head);
> +	if (!offload->skb_irq_queue)
> +		return -ENOMEM;
> +
> +	for_each_possible_cpu(cpu)
> +		__skb_queue_head_init(per_cpu_ptr(offload->skb_irq_queue, cpu));
>  
>  	netif_napi_add_weight(dev, &offload->napi, can_rx_offload_napi_poll,
>  			      weight);
> @@ -420,8 +466,17 @@ EXPORT_SYMBOL_GPL(can_rx_offload_enable);
>  
>  void can_rx_offload_del(struct can_rx_offload *offload)
>  {
> +	int cpu;
> +
>  	netif_napi_del(&offload->napi);
>  	skb_queue_purge(&offload->skb_queue);
> -	__skb_queue_purge(&offload->skb_irq_queue);
> +
> +	if (!offload->skb_irq_queue)
> +		return;
> +
> +	for_each_possible_cpu(cpu)
> +		__skb_queue_purge(per_cpu_ptr(offload->skb_irq_queue, cpu));
> +
> +	free_percpu(offload->skb_irq_queue);
>  }
>  EXPORT_SYMBOL_GPL(can_rx_offload_del);
> diff --git a/include/linux/can/rx-offload.h b/include/linux/can/rx-offload.h
> index d29bb4521947..1b9e2a8ab39a 100644
> --- a/include/linux/can/rx-offload.h
> +++ b/include/linux/can/rx-offload.h
> @@ -20,7 +20,7 @@ struct can_rx_offload {
>  					bool drop);
>  
>  	struct sk_buff_head skb_queue;
> -	struct sk_buff_head skb_irq_queue;
> +	struct sk_buff_head __percpu *skb_irq_queue;
>  	u32 skb_queue_len_max;
>  
>  	unsigned int mb_first;
> -- 
> 2.43.0
> 

Re: [PATCH v4 1/3] can: rx-offload: make skb_irq_queue per-CPU

[Ciprian Marian Costea]

On 9/7/2026 11:19 AM, Bough Chen wrote:
> On Tue, Sep 01, 2026 at 01:48:46PM +0200, Ciprian Costea wrote:
>> From: Ciprian Marian Costea <ciprianmarian.costea@oss.nxp.com>
>>
>> skb_irq_queue is filled by the IRQ handlers using the lockless
>> __skb_queue_add_sort() / __skb_queue_tail() helpers and later spliced
>> into skb_queue under skb_queue.lock by can_rx_offload_irq_finish() and
>> can_rx_offload_threaded_irq_finish().
>>
>> This is only safe while a single context fills skb_irq_queue. FlexCAN
>> on NXP S32G2 (FLEXCAN_QUIRK_SECONDARY_MB_IRQ) uses two mailbox IRQ
>> lines, one for MB0-7 and one for MB8-63; MCF5441X similarly splits its
>> mailbox interrupt. When these lines are affined to different CPUs both
>> handlers can run at the same time and enqueue into the same sk_buff_head
>> concurrently, corrupting its list.
>>
>> Allocate skb_irq_queue per-CPU so the handlers no longer share a list,
>> keeping the enqueue path lock-free. Access the per-CPU queue via
>> get_cpu_ptr()/put_cpu_ptr() in the enqueue helpers: this disables
>> preemption around the lockless __skb_queue_*() operation.
>>
>> can_rx_offload_irq_finish() runs in the same context as its enqueues and
>> splices this_cpu_ptr(). can_rx_offload_threaded_irq_finish() may have
>> been migrated after its enqueues, so it splices every possible CPU's
>> queue; this is safe because each per-CPU queue has a single producer and
>> that producer runs with preemption disabled, so it cannot race the
>> splice.
>>
>> Cross-line frames are now sorted by timestamp only within a CPU's queue
>> and appended across CPUs on splice; each skb keeps its own timestamp.
>>
>> Fixes: c757096ea103 ("can: rx-offload: add skb queue for use during ISR")
>> Signed-off-by: Ciprian Marian Costea <ciprianmarian.costea@oss.nxp.com>
>> ---
>>   drivers/net/can/dev/rx-offload.c | 83 ++++++++++++++++++++++++++------
>>   include/linux/can/rx-offload.h   |  2 +-
>>   2 files changed, 70 insertions(+), 15 deletions(-)
>>
>> diff --git a/drivers/net/can/dev/rx-offload.c b/drivers/net/can/dev/rx-offload.c
>> index 46e7b6db4a1e..649bfda08b65 100644
>> --- a/drivers/net/can/dev/rx-offload.c
>> +++ b/drivers/net/can/dev/rx-offload.c
>> @@ -7,6 +7,7 @@
>>   
>>   #include <linux/can/dev.h>
>>   #include <linux/can/rx-offload.h>
>> +#include <linux/percpu.h>
>>   
>>   struct can_rx_offload_cb {
>>   	u32 timestamp;
>> @@ -175,9 +176,18 @@ can_rx_offload_offload_one(struct can_rx_offload *offload, unsigned int n)
>>   int can_rx_offload_irq_offload_timestamp(struct can_rx_offload *offload,
>>   					 u64 pending)
>>   {
>> +	struct sk_buff_head *irq_queue;
>>   	unsigned int i;
>>   	int received = 0;
>>   
>> +	/*
>> +	 * get_cpu_ptr() disables preemption so that the lockless
>> +	 * __skb_queue_*() below operate on the current CPU's queue without
>> +	 * racing a migration. This also keeps this_cpu_ptr() valid when a
>> +	 * driver enqueues from a preemptible (threaded IRQ) context.
>> +	 */
>> +	irq_queue = get_cpu_ptr(offload->skb_irq_queue);
>> +
>>   	for (i = offload->mb_first;
>>   	     can_rx_offload_le(offload, i, offload->mb_last);
>>   	     can_rx_offload_inc(offload, &i)) {
>> @@ -190,20 +200,25 @@ int can_rx_offload_irq_offload_timestamp(struct can_rx_offload *offload,
>>   		if (IS_ERR_OR_NULL(skb))
>>   			continue;
>>   
>> -		__skb_queue_add_sort(&offload->skb_irq_queue, skb,
>> +		__skb_queue_add_sort(irq_queue, skb,
>>   				     can_rx_offload_compare);
>>   		received++;
>>   	}
>>   
>> +	put_cpu_ptr(offload->skb_irq_queue);
>> +
>>   	return received;
>>   }
>>   EXPORT_SYMBOL_GPL(can_rx_offload_irq_offload_timestamp);
>>   
>>   int can_rx_offload_irq_offload_fifo(struct can_rx_offload *offload)
>>   {
>> +	struct sk_buff_head *irq_queue;
>>   	struct sk_buff *skb;
>>   	int received = 0;
>>   
>> +	irq_queue = get_cpu_ptr(offload->skb_irq_queue);
>> +
>>   	while (1) {
>>   		skb = can_rx_offload_offload_one(offload, 0);
>>   		if (IS_ERR(skb))
>> @@ -211,10 +226,12 @@ int can_rx_offload_irq_offload_fifo(struct can_rx_offload *offload)
>>   		if (!skb)
>>   			break;
>>   
>> -		__skb_queue_tail(&offload->skb_irq_queue, skb);
>> +		__skb_queue_tail(irq_queue, skb);
>>   		received++;
>>   	}
>>   
>> +	put_cpu_ptr(offload->skb_irq_queue);
>> +
>>   	return received;
>>   }
>>   EXPORT_SYMBOL_GPL(can_rx_offload_irq_offload_fifo);
>> @@ -222,6 +239,7 @@ EXPORT_SYMBOL_GPL(can_rx_offload_irq_offload_fifo);
>>   int can_rx_offload_queue_timestamp(struct can_rx_offload *offload,
>>   				   struct sk_buff *skb, u32 timestamp)
>>   {
>> +	struct sk_buff_head *irq_queue;
>>   	struct can_rx_offload_cb *cb;
>>   
>>   	if (skb_queue_len(&offload->skb_queue) >
>> @@ -233,8 +251,9 @@ int can_rx_offload_queue_timestamp(struct can_rx_offload *offload,
>>   	cb = can_rx_offload_get_cb(skb);
>>   	cb->timestamp = timestamp;
>>   
>> -	__skb_queue_add_sort(&offload->skb_irq_queue, skb,
>> -			     can_rx_offload_compare);
>> +	irq_queue = get_cpu_ptr(offload->skb_irq_queue);
>> +	__skb_queue_add_sort(irq_queue, skb, can_rx_offload_compare);
>> +	put_cpu_ptr(offload->skb_irq_queue);
>>   
>>   	return 0;
>>   }
>> @@ -268,13 +287,17 @@ EXPORT_SYMBOL_GPL(can_rx_offload_get_echo_skb_queue_timestamp);
>>   int can_rx_offload_queue_tail(struct can_rx_offload *offload,
>>   			      struct sk_buff *skb)
>>   {
>> +	struct sk_buff_head *irq_queue;
>> +
>>   	if (skb_queue_len(&offload->skb_queue) >
>>   	    offload->skb_queue_len_max) {
>>   		dev_kfree_skb_any(skb);
>>   		return -ENOBUFS;
>>   	}
>>   
>> -	__skb_queue_tail(&offload->skb_irq_queue, skb);
>> +	irq_queue = get_cpu_ptr(offload->skb_irq_queue);
>> +	__skb_queue_tail(irq_queue, skb);
>> +	put_cpu_ptr(offload->skb_irq_queue);
>>   
>>   	return 0;
>>   }
>> @@ -307,14 +330,15 @@ EXPORT_SYMBOL_GPL(can_rx_offload_get_echo_skb_queue_tail);
>>   
>>   void can_rx_offload_irq_finish(struct can_rx_offload *offload)
>>   {
>> +	struct sk_buff_head *irq_queue = this_cpu_ptr(offload->skb_irq_queue);
>>   	unsigned long flags;
>>   	int queue_len;
>>   
>> -	if (skb_queue_empty_lockless(&offload->skb_irq_queue))
>> +	if (skb_queue_empty_lockless(irq_queue))
>>   		return;
>>   
>>   	spin_lock_irqsave(&offload->skb_queue.lock, flags);
>> -	skb_queue_splice_tail_init(&offload->skb_irq_queue, &offload->skb_queue);
>> +	skb_queue_splice_tail_init(irq_queue, &offload->skb_queue);
>>   	spin_unlock_irqrestore(&offload->skb_queue.lock, flags);
>>   
>>   	queue_len = skb_queue_len(&offload->skb_queue);
>> @@ -330,15 +354,29 @@ void can_rx_offload_threaded_irq_finish(struct can_rx_offload *offload)
>>   {
>>   	unsigned long flags;
>>   	int queue_len;
>> -
>> -	if (skb_queue_empty_lockless(&offload->skb_irq_queue))
>> -		return;
>> -
>> +	int cpu;
>> +
>> +	/*
>> +	 * Splice every CPU's queue: unlike the non-threaded
>> +	 * can_rx_offload_irq_finish(), a threaded handler may be migrated
>> +	 * between the enqueue and this splice, so the frames may sit on a
>> +	 * different CPU's queue. This is only safe because a given per-CPU
>> +	 * queue has a single producer (the enqueue on that CPU is
>> +	 * non-preemptible), so no producer can race this splice.
>> +	 */
>>   	spin_lock_irqsave(&offload->skb_queue.lock, flags);
>> -	skb_queue_splice_tail_init(&offload->skb_irq_queue, &offload->skb_queue);
>> +	for_each_possible_cpu(cpu) {
>> +		struct sk_buff_head *irq_queue;
>> +
>> +		irq_queue = per_cpu_ptr(offload->skb_irq_queue, cpu);
>> +		skb_queue_splice_tail_init(irq_queue, &offload->skb_queue);
>> +	}
>>   	spin_unlock_irqrestore(&offload->skb_queue.lock, flags);
>>   
> Hi Ciprian,
> The fix looks correct to me. I checked all rx-offload users in
> drivers/net/can/ and the change is safe for every current driver.
> 
> One suggestion:
> the cross-CPU splice in can_rx_offload_threaded_irq_finish() is safe
> only as long as there is a single threaded handler context per offload
> instance, so each per-CPU queue has exactly one producer.
> This isn't new (the old shared skb_irq_queue relied on the same
> "single context fills the queue" assumption), and for the current
> threaded users it's actually enforced by genirq:
> they all use request_threaded_irq(irq, NULL, handler, ...),
> which mandates IRQF_ONESHOT, so the handler can't re-enter.
> 
> Only the threaded finish path cares about this, and all three such
> drivers request the IRQ with IRQF_ONESHOT:
>    - m_can (peripheral)
>    - mcp251xfd
>    - nct6694_canfd
> They all use the manual enqueue path (queue_timestamp/queue_tail), not
> irq_offload_*(). Everyone else uses the non-threaded irq_finish()
> (this_cpu_ptr only) and is safe by construction.
> 
> Could you spell out this assumption in the comment above the
> for_each_possible_cpu() loop? e.g.:
> 
> This assumes a single threaded handler context per offload instance
> (IRQ requested with IRQF_ONESHOT / handler non-reentrant), so each
> per-CPU queue has exactly one producer. If that changes, this
> cross-CPU splice of lockless queues would need additional locking.
> 
> Minor, non-blocking: get_cpu_ptr() only wraps a single enqueue, so a
> handler that drains several frames per IRQ (e.g. mcp251xfd) can migrate
> mid-batch and split a burst across CPU queues, losing intra-batch
> timestamp order after the splice. Just as the Sashiko reveiw in your
> V2. I think it is harmless, and SocketCAN doesn't guarantee delivery
> order anyway, so I'm fine with it as-is.
> Point it out just in case other people may have comment on it.
> 
> With the comment clarification:
> 
> Reviewed-by: Haibo Chen <haibo.chen@nxp.com>
> 
> Regards
> Haibo Chen
> 

Hello Haibo,

Thank you for taking time into reviewing this patchset.

I will add your comment clarifying the single-producer context in V5.

Best Regards,
Ciprian

>>   	queue_len = skb_queue_len(&offload->skb_queue);
>> +	if (!queue_len)
>> +		return;
>> +
>>   	if (queue_len > offload->skb_queue_len_max / 8)
>>   		netdev_dbg(offload->dev, "%s: queue_len=%d\n",
>>   			   __func__, queue_len);
>> @@ -353,13 +391,21 @@ static int can_rx_offload_init_queue(struct net_device *dev,
>>   				     struct can_rx_offload *offload,
>>   				     unsigned int weight)
>>   {
>> +	int cpu;
>> +
>>   	offload->dev = dev;
>>   
>>   	/* Limit queue len to 4x the weight (rounded to next power of two) */
>>   	offload->skb_queue_len_max = 2 << fls(weight);
>>   	offload->skb_queue_len_max *= 4;
>>   	skb_queue_head_init(&offload->skb_queue);
>> -	__skb_queue_head_init(&offload->skb_irq_queue);
>> +
>> +	offload->skb_irq_queue = alloc_percpu(struct sk_buff_head);
>> +	if (!offload->skb_irq_queue)
>> +		return -ENOMEM;
>> +
>> +	for_each_possible_cpu(cpu)
>> +		__skb_queue_head_init(per_cpu_ptr(offload->skb_irq_queue, cpu));
>>   
>>   	netif_napi_add_weight(dev, &offload->napi, can_rx_offload_napi_poll,
>>   			      weight);
>> @@ -420,8 +466,17 @@ EXPORT_SYMBOL_GPL(can_rx_offload_enable);
>>   
>>   void can_rx_offload_del(struct can_rx_offload *offload)
>>   {
>> +	int cpu;
>> +
>>   	netif_napi_del(&offload->napi);
>>   	skb_queue_purge(&offload->skb_queue);
>> -	__skb_queue_purge(&offload->skb_irq_queue);
>> +
>> +	if (!offload->skb_irq_queue)
>> +		return;
>> +
>> +	for_each_possible_cpu(cpu)
>> +		__skb_queue_purge(per_cpu_ptr(offload->skb_irq_queue, cpu));
>> +
>> +	free_percpu(offload->skb_irq_queue);
>>   }
>>   EXPORT_SYMBOL_GPL(can_rx_offload_del);
>> diff --git a/include/linux/can/rx-offload.h b/include/linux/can/rx-offload.h
>> index d29bb4521947..1b9e2a8ab39a 100644
>> --- a/include/linux/can/rx-offload.h
>> +++ b/include/linux/can/rx-offload.h
>> @@ -20,7 +20,7 @@ struct can_rx_offload {
>>   					bool drop);
>>   
>>   	struct sk_buff_head skb_queue;
>> -	struct sk_buff_head skb_irq_queue;
>> +	struct sk_buff_head __percpu *skb_irq_queue;
>>   	u32 skb_queue_len_max;
>>   
>>   	unsigned int mb_first;
>> -- 
>> 2.43.0
>>