[PATCH v4 1/3] can: rx-offload: make skb_irq_queue per-CPU
Bough Chen
haibo.chen at oss.nxp.com
Mon Sep 7 01:19:54 PDT 2026
On Tue, Sep 01, 2026 at 01:48:46PM +0200, Ciprian Costea wrote:
> From: Ciprian Marian Costea <ciprianmarian.costea at 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 at 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 at 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
>
More information about the linux-arm-kernel
mailing list