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@xxxxxxxxxxx>
> 
> 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@xxxxxxxxxxx>
> ---
>  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@xxxxxxx>

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
>