[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@xxxxxxxxxxxxxx>
---
 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