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

[Ciprian Marian Costea <ciprianmarian.costea@oss.nxp.com>]

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