Re: [PATCH RFC 1/2] can: rx-offload: can_rx_offload_threaded_irq_flush(): add newfunction to be called from within loop of threaded interrupt handlers

Re: [PATCH RFC 1/2] can: rx-offload: can_rx_offload_threaded_irq_flush(): add newfunction to be called from within loop of threaded interrupt handlers

[netdev-bot+sashiko]

Thank you for your contribution! Sashiko AI review found 5 potential
issue(s) to consider.

Critical: 0 · High: 2 · Medium: 1 · Low: 2

If you believe none of the issues are legitimate, please reply to each
issue of Medium severity or higher and explain why the AI is wrong.
If any of them are legitimate and you plan to respin, please let patchwork
know by including "pw-bot: cr" as a separate line at the end of your reply
(one such reply per series is enough).

- [High] The unbounded skb_irq_queue growth that this commit sets out to
  stop can still happen at the end of the series, so the fix is…
- [High] can_rx_offload_threaded_irq_flush() corrupts the qlen counters
  when frames are sorted in ahead of the saved marker.
- [Medium] The kernel-doc for can_rx_offload_threaded_irq_flush() says
  that when skb_irq_queue 'exceeds 3/4' of the max, the function will…
- [Low] The kernel-doc for the new static helper __skb_cut_position() in
  drivers/net/can/dev/rx-offload.c says '@entry: an entry within head,…
- [Low] In can_rx_offload_threaded_irq_flush(), queue_len is declared as
  u32 (line 404) but printed with "%s: queue_len=%d\n" in netdev_dbg()…

--

can: rx-offload: can_rx_offload_threaded_irq_flush(): add newfunction to be called from within loop of threaded interrupt handlers

This adds can_rx_offload_threaded_irq_flush(), which is meant to be called
from inside the loop of a threaded IRQ handler. When skb_irq_queue reaches
1/2 of skb_queue_len_max, it saves the current tail. When the queue reaches
3/4, it moves everything up to that saved skb into skb_queue and schedules
NAPI, so the newest entries stay behind for later reordering.

> diff --git a/drivers/net/can/dev/rx-offload.c b/drivers/net/can/dev/rx-offload.c
> index 46e7b6db4a1ef..9f494ab7561ef 100644
> --- a/drivers/net/can/dev/rx-offload.c
> +++ b/drivers/net/can/dev/rx-offload.c

[ ... ]

> @@ -349,6 +351,90 @@ void can_rx_offload_threaded_irq_finish(struct can_rx_offload *offload)
>  }
>  EXPORT_SYMBOL_GPL(can_rx_offload_threaded_irq_finish);
>  
> +/**
> + * __skb_cut_position - cut a skb list into two
> + * @list: a new list to add all removed entries
> + * @head: a list with entries
> + * @entry: an entry within head, could be the head itself
> + *	and if so we won't cut the list

[Severity: Low]
Does this helper actually handle entry == head? This text seems to be copied
from list_cut_position(), where the wrapper has the checks that make the
statement true:

include/linux/list.h:list_cut_position() {
	if (list_empty(head))
		return;
	if (list_is_singular(head) && !list_is_head(entry, head) && (entry != head->next))
		return;
	if (list_is_head(entry, head))
		INIT_LIST_HEAD(list);
	else
		__list_cut_position(list, head, entry);
}

__skb_cut_position() has none of these checks. It always does the cut and
adjusts both qlen values. If entry is the head, or head is empty, both lists
come out malformed.

The only caller passes the result of skb_peek_tail(), so this path is not
reachable today. Could the comment be corrected, though?

This isn't a bug, but the __skb_ prefix makes a private CAN helper look like
a core skbuff API, and it could clash with a future core helper of the same
name.

> + * @qlen: length up to @entry in @head
> + *
> + * This helper moves the initial part of @head, up to and including
> + * @entry, from @head to @list. You should pass on @entry an element
> + * you know is on @head and @qlen being the position of that element
> + * on @head. @list should be an empty list or a list you do not care
> + * about losing its data.
> + *
> + */
> +static inline void __skb_cut_position(struct sk_buff_head *list,
> +				      struct sk_buff_head *head,
> +				      struct sk_buff *entry,
> +				      __u32 qlen)
> +{
> +	struct sk_buff *new_first = entry->next;
> +	struct sk_buff *first = head->next;
> +
> +	WRITE_ONCE(list->next, first);
> +	WRITE_ONCE(list->next->prev, (struct sk_buff *)list);
> +	WRITE_ONCE(list->prev, entry);
> +	WRITE_ONCE(entry->next, (struct sk_buff *)list);
> +	list->qlen = qlen;
> +
> +	WRITE_ONCE(head->next, new_first);
> +	WRITE_ONCE(new_first->prev, (struct sk_buff *)head);
> +	head->qlen -= qlen;
> +}
> +
> +/**
> + * can_rx_offload_threaded_irq_flush() - partially flush the rx_offload queue
> + * @offload: pointer to rx_offload context
> + *
> + * If can_rx_offload->skb_irq_queue exceeds 3/4 of the maximal queue
> + * length (can_rx_offload->skb_queue_len_max) flush about 1/2 of the
> + * maximal queue length to NAPI.
> + *
> + * This function is intended to be called within the loop of threaded
> + * IRQ handlers, which run until all IRQs have been served.

[Severity: High]
Can skb_irq_queue still grow without bound once the series is applied?

This patch adds no caller. The follow-up commit "can: mcp251xfd:
mcp251xfd_irq(): add call to can_rx_offload_threaded_irq_flush()" adds one,
but only in the main intf loop of mcp251xfd_irq().

mcp251xfd_irq() also has an rx_int GPIO fast-path loop, and that loop never
calls the new helper:

drivers/net/can/spi/mcp251xfd/mcp251xfd-core.c:mcp251xfd_irq() {
	if (priv->rx_int)
		do {
			int rx_pending;

			rx_pending = gpiod_get_value_cansleep(priv->rx_int);
			if (!rx_pending)
				break;
			...
			err = mcp251xfd_handle(priv, rxif);
			...
		} while (priv->rx_ring_num == 1);
	...
}

Each frame read in that loop goes through can_rx_offload_queue_timestamp().
That function only compares skb_queue, not skb_irq_queue, against
skb_queue_len_max. NAPI is not scheduled until
can_rx_offload_threaded_irq_finish() runs.

With microchip,rx-int-gpios and a single RX ring, wouldn't the case the
commit message describes ("the threaded IRQ handler will not exit the while
loop") still leave skb_irq_queue growing without limit?

> + *
> + */
> +void can_rx_offload_threaded_irq_flush(struct can_rx_offload *offload)
> +{
> +	struct sk_buff_head tmp_queue;
> +	unsigned long flags;
> +	u32 irq_queue_len, queue_len;
> +
> +	irq_queue_len = skb_queue_len(&offload->skb_irq_queue);
> +	if (irq_queue_len < offload->skb_queue_len_max / 2)
> +		return;
> +
> +	if (!offload->flush_skb) {
> +		offload->flush_skb = skb_peek_tail(&offload->skb_irq_queue);
> +		offload->flush_len = irq_queue_len;
> +	}
> +
> +	if (irq_queue_len < offload->skb_queue_len_max * 3 / 4)
> +		return;

[Severity: Medium]
What happens when flush_skb is NULL and irq_queue_len is already at or above
3/4 of skb_queue_len_max on the first call?

In that case, the same call sets flush_skb to the current tail and
flush_len to the full queue length. It then skips this return and cuts the
entire skb_irq_queue.

That doesn't match the kernel-doc above ("flush about 1/2 of the maximal
queue length") or the commit message ("leave about 1/4 of the maximum queue
length in offload->skb_irq_queue").

This looks reachable in mcp251xfd. The rx_int loop in mcp251xfd_irq() fills
skb_irq_queue without calling the helper, so the first call from the main
loop can already see 384 or more entries (weight 32, max 512). Older frames
read afterwards from another RX ring or the TEF would then be delivered
after newer ones.

Should saving the marker be kept separate from making the cut?

As a minor point, the doc says "exceeds", but the early returns use <. That
means the code triggers at exactly 1/2 and exactly 3/4.

> +
> +	__skb_cut_position(&tmp_queue, &offload->skb_irq_queue,
> +			   offload->flush_skb, offload->flush_len);

[Severity: High]
Is offload->flush_len still the position of flush_skb at this point?

flush_len is saved once, when the queue reaches 1/2. After that, the IRQ
loop keeps adding frames through __skb_queue_add_sort(), which walks
backwards from the tail and inserts by timestamp:

drivers/net/can/dev/rx-offload.c:__skb_queue_add_sort() {
	skb_queue_reverse_walk(head, pos) {
		...
		if (compare(pos, new) < 0)
			continue;
		insert = pos;
		break;
	}
	...
}

Any frame older than flush_skb is therefore inserted in front of it. In
mcp251xfd, this can be a TEF echo skb queued with
can_rx_offload_get_echo_skb_queue_timestamp(), or a frame from another RX
ring. That is the late, older-frame case the commit message describes.

If N such frames were inserted, the cut moves flush_len + N skbs, but
__skb_cut_position() does:

	list->qlen = qlen;
	...
	head->qlen -= qlen;

As a result, skb_queue.qlen ends up N too low and skb_irq_queue.qlen ends up
N too high.

Once NAPI drains skb_queue, __skb_unlink() decrements qlen once per real
skb. skb_queue.qlen (a u32) then wraps to about 2^32 - N. After that, this
check always fires:

drivers/net/can/dev/rx-offload.c:can_rx_offload_queue_timestamp() {
	if (skb_queue_len(&offload->skb_queue) >
	    offload->skb_queue_len_max) {
		dev_kfree_skb_any(skb);
		return -ENOBUFS;
	}
	...
}

can_rx_offload_queue_tail() and can_rx_offload_offload_one() have the same
check. Every RX frame and every TX echo skb would then be dropped.

If no frame was queued after flush_skb, the cut leaves skb_irq_queue
physically empty with qlen == N. can_rx_offload_threaded_irq_finish() then
returns early on skb_queue_empty_lockless() and never splices the counts
back. New frames are dropped before they reach skb_irq_queue.

Could RX then stay dead until the driver is unbound? mcp251xfd only calls
can_rx_offload_del() in remove.

There may also be a NULL dereference. If skb_irq_queue is empty but its qlen
is at or above 3/4 of the max, skb_peek_tail() returns NULL, and
__skb_cut_position() would then dereference entry->next on that NULL
flush_skb.

> +
> +	spin_lock_irqsave(&offload->skb_queue.lock, flags);
> +	skb_queue_splice_tail_init(&tmp_queue, &offload->skb_queue);
> +	spin_unlock_irqrestore(&offload->skb_queue.lock, flags);
> +
> +	offload->flush_skb = NULL;
> +
> +	queue_len = skb_queue_len(&offload->skb_queue);
> +	if (queue_len > offload->skb_queue_len_max / 8)
> +		netdev_dbg(offload->dev, "%s: queue_len=%d\n",
> +			   __func__, queue_len);

[Severity: Low]
This isn't a bug, but queue_len is a u32 here, while the copy in
can_rx_offload_threaded_irq_finish() is an int. Should this use %u?

> +
> +	local_bh_disable();
> +	napi_schedule(&offload->napi);
> +	local_bh_enable();
> +}
> +EXPORT_SYMBOL_GPL(can_rx_offload_threaded_irq_flush);

[ ... ]

-- 
Sashiko AI review · https://netdev-ai.bots.linux.dev/sashiko/#/patchset/20261005-upstream-can-rx-offload-batching-alternative-v1-0-4c08ccb516ec%40pengutronix.de

[PATCH RFC 0/2] can: rx-offload: can_rx_offload_threaded_irq_flush(): partly flush skb_irq_queue

[Marc Kleine-Budde]

This is a proof of conecpt series to partly flush the skb_irq_queue. It
should avoid issues pointed out by the netdev sashiko bot in Ciprian
Costea's series.

When the system is under heavy load, flushing is required to prevent an RX
starvation.

Partial flushing is required to prevent out of order RX errors when
combining multiple RX FIFOs.

Signed-off-by: Marc Kleine-Budde <mkl@pengutronix.de>
---
Marc Kleine-Budde (2):
      can: rx-offload: can_rx_offload_threaded_irq_flush(): add newfunction to be called from within loop of threaded interrupt handlers
      can: mcp251xfd: mcp251xfd_irq(): add call to can_rx_offload_threaded_irq_flush()

 drivers/net/can/dev/rx-offload.c               | 88 ++++++++++++++++++++++++++
 drivers/net/can/spi/mcp251xfd/mcp251xfd-core.c |  1 +
 include/linux/can/rx-offload.h                 |  4 ++
 3 files changed, 93 insertions(+)
---
base-commit: cfb7793d1bc0f7d90571611979654cf1b3886b29
change-id: 20261001-upstream-can-rx-offload-batching-alternative-c2a1d0b85a63

Best regards,
--  
Marc Kleine-Budde <mkl@pengutronix.de>

[PATCH RFC 1/2] can: rx-offload: can_rx_offload_threaded_irq_flush(): add newfunction to be called from within loop of threaded interrupt handlers

[Marc Kleine-Budde]

CAN drivers that use threaded IRQ handlers usually have a while loop that
runs until all IRQs have been processed. In this loop, received CAN frames
are placed in the offload->skb_irq_queue.

At the end of the IRQ handler, the offload->skb_irq_queue is spliced into
the offload->skb_queue and NAPI is started, which then pushes the CAN
frames into the network stack.

Under certain load situations, the threaded IRQ handler will not exit the
while loop, resulting in an unlimited growth of the offload->skb_irq_queue.

In order to avoid this situation add a new function
can_rx_offload_threaded_irq_flush(). It's meant to be called from within
the threaded IRQ while loop.

Check in can_rx_offload_threaded_irq_flush() whether
offload->skb_irq_queue exceeds 1/2 of the maximum queue length
(can_rx_offload->skb_queue_len_max) and save the last element of the queue.
If the queue exceeds 3/4 of the maximum queue length, spliced up to the
previously saved element into offload->skb_queue and start NAPI.

Why is not the whole offload->skb_irq_queue flushed to NAPI?

Some CAN-IP cores use more than one FIFO or even independent mailboxes. In
situations where reception from the CAN bus and reading of the
FIFOs/mailboxes take place simultaneously, older CAN frames may be present
in the chip at the end of the current IRQ handler loop than in the
offload->skb_irq_queue.

As the order of the CAN frames is decisive for most CAN protocols, leave
about 1/4 of the maximum queue length in offload->skb_irq_queue so that
older CAN frames can be added to the queue at the correct position in the
next loop of the IRQ handler.

Signed-off-by: Marc Kleine-Budde <mkl@pengutronix.de>
---
 drivers/net/can/dev/rx-offload.c | 88 ++++++++++++++++++++++++++++++++++++++++
 include/linux/can/rx-offload.h   |  4 ++
 2 files changed, 92 insertions(+)

diff --git a/drivers/net/can/dev/rx-offload.c b/drivers/net/can/dev/rx-offload.c
index 46e7b6db4a1e..9f494ab7561e 100644
--- a/drivers/net/can/dev/rx-offload.c
+++ b/drivers/net/can/dev/rx-offload.c
@@ -338,6 +338,8 @@ void can_rx_offload_threaded_irq_finish(struct can_rx_offload *offload)
 	skb_queue_splice_tail_init(&offload->skb_irq_queue, &offload->skb_queue);
 	spin_unlock_irqrestore(&offload->skb_queue.lock, flags);
 
+	offload->flush_skb = NULL;
+
 	queue_len = skb_queue_len(&offload->skb_queue);
 	if (queue_len > offload->skb_queue_len_max / 8)
 		netdev_dbg(offload->dev, "%s: queue_len=%d\n",
@@ -349,6 +351,90 @@ void can_rx_offload_threaded_irq_finish(struct can_rx_offload *offload)
 }
 EXPORT_SYMBOL_GPL(can_rx_offload_threaded_irq_finish);
 
+/**
+ * __skb_cut_position - cut a skb list into two
+ * @list: a new list to add all removed entries
+ * @head: a list with entries
+ * @entry: an entry within head, could be the head itself
+ *	and if so we won't cut the list
+ * @qlen: length up to @entry in @head
+ *
+ * This helper moves the initial part of @head, up to and including
+ * @entry, from @head to @list. You should pass on @entry an element
+ * you know is on @head and @qlen being the position of that element
+ * on @head. @list should be an empty list or a list you do not care
+ * about losing its data.
+ *
+ */
+static inline void __skb_cut_position(struct sk_buff_head *list,
+				      struct sk_buff_head *head,
+				      struct sk_buff *entry,
+				      __u32 qlen)
+{
+	struct sk_buff *new_first = entry->next;
+	struct sk_buff *first = head->next;
+
+	WRITE_ONCE(list->next, first);
+	WRITE_ONCE(list->next->prev, (struct sk_buff *)list);
+	WRITE_ONCE(list->prev, entry);
+	WRITE_ONCE(entry->next, (struct sk_buff *)list);
+	list->qlen = qlen;
+
+	WRITE_ONCE(head->next, new_first);
+	WRITE_ONCE(new_first->prev, (struct sk_buff *)head);
+	head->qlen -= qlen;
+}
+
+/**
+ * can_rx_offload_threaded_irq_flush() - partially flush the rx_offload queue
+ * @offload: pointer to rx_offload context
+ *
+ * If can_rx_offload->skb_irq_queue exceeds 3/4 of the maximal queue
+ * length (can_rx_offload->skb_queue_len_max) flush about 1/2 of the
+ * maximal queue length to NAPI.
+ *
+ * This function is intended to be called within the loop of threaded
+ * IRQ handlers, which run until all IRQs have been served.
+ *
+ */
+void can_rx_offload_threaded_irq_flush(struct can_rx_offload *offload)
+{
+	struct sk_buff_head tmp_queue;
+	unsigned long flags;
+	u32 irq_queue_len, queue_len;
+
+	irq_queue_len = skb_queue_len(&offload->skb_irq_queue);
+	if (irq_queue_len < offload->skb_queue_len_max / 2)
+		return;
+
+	if (!offload->flush_skb) {
+		offload->flush_skb = skb_peek_tail(&offload->skb_irq_queue);
+		offload->flush_len = irq_queue_len;
+	}
+
+	if (irq_queue_len < offload->skb_queue_len_max * 3 / 4)
+		return;
+
+	__skb_cut_position(&tmp_queue, &offload->skb_irq_queue,
+			   offload->flush_skb, offload->flush_len);
+
+	spin_lock_irqsave(&offload->skb_queue.lock, flags);
+	skb_queue_splice_tail_init(&tmp_queue, &offload->skb_queue);
+	spin_unlock_irqrestore(&offload->skb_queue.lock, flags);
+
+	offload->flush_skb = NULL;
+
+	queue_len = skb_queue_len(&offload->skb_queue);
+	if (queue_len > offload->skb_queue_len_max / 8)
+		netdev_dbg(offload->dev, "%s: queue_len=%d\n",
+			   __func__, queue_len);
+
+	local_bh_disable();
+	napi_schedule(&offload->napi);
+	local_bh_enable();
+}
+EXPORT_SYMBOL_GPL(can_rx_offload_threaded_irq_flush);
+
 static int can_rx_offload_init_queue(struct net_device *dev,
 				     struct can_rx_offload *offload,
 				     unsigned int weight)
@@ -360,6 +446,8 @@ static int can_rx_offload_init_queue(struct net_device *dev,
 	offload->skb_queue_len_max *= 4;
 	skb_queue_head_init(&offload->skb_queue);
 	__skb_queue_head_init(&offload->skb_irq_queue);
+	offload->flush_skb = NULL;
+	offload->flush_len = 0;
 
 	netif_napi_add_weight(dev, &offload->napi, can_rx_offload_napi_poll,
 			      weight);
diff --git a/include/linux/can/rx-offload.h b/include/linux/can/rx-offload.h
index d29bb4521947..18dcced516a5 100644
--- a/include/linux/can/rx-offload.h
+++ b/include/linux/can/rx-offload.h
@@ -23,6 +23,9 @@ struct can_rx_offload {
 	struct sk_buff_head skb_irq_queue;
 	u32 skb_queue_len_max;
 
+	struct sk_buff *flush_skb;
+	u32 flush_len;
+
 	unsigned int mb_first;
 	unsigned int mb_last;
 
@@ -54,6 +57,7 @@ unsigned int can_rx_offload_get_echo_skb_queue_tail(struct can_rx_offload *offlo
 						    unsigned int *frame_len_ptr);
 void can_rx_offload_irq_finish(struct can_rx_offload *offload);
 void can_rx_offload_threaded_irq_finish(struct can_rx_offload *offload);
+void can_rx_offload_threaded_irq_flush(struct can_rx_offload *offload);
 void can_rx_offload_del(struct can_rx_offload *offload);
 void can_rx_offload_enable(struct can_rx_offload *offload);
 

-- 
2.53.0