RE: [PATCH net v2] tipc: defer local Nagle backlog receive processing

[Tung Quang Nguyen]

>Subject: [PATCH net v2] tipc: defer local Nagle backlog receive processing
>
>A local TIPC stream can soft-lock a CPU when receiving a packet sends queued
>Nagle data while the socket's spinlock is held. Local delivery enters the receive
>path again before that lock is released. The nested call cannot take the lock, so
>it retries the input queue without making progress.
>
>When an unowned socket has queued data for a local peer, put the incoming
>packet on sk_backlog. Queue work to take socket ownership and drain the
>backlog through release_sock(). Leave the older data on sk_write_queue,
>where later send and shutdown cannot overtake it.
>
>The owned-socket path already puts incoming packets on the backlog.
>Only inspect sk_write_queue and the peer node when no user owns the
>socket. tipc_sk_enqueue() holds sk_lock.slock at that point, so a user cannot
>take ownership and change those fields during the check.
>
>Hold a socket reference for the work, and drop the extra reference if it is
>already pending. Queue the work after leaving the receive loop, even when the
>loop stops at its time limit or finds no matching packet. Set up the work before
>publishing the socket in the TIPC port hash.
>
>Fixes: c0bceb97db9e ("tipc: add smart nagle feature")
>Reported-by: syzbot+10a41dc44eef71aa9450@xxxxxxxxxxxxxxxxxxxxxxxxx
>Link:
>https://lore.kernel.org/r/GV1P189MB1988F6301FA670CA9818F7FCC6DB2@GV
>1P189MB1988.EURP189.PROD.OUTLOOK.COM
>Signed-off-by: Bart Chronowski <immersa.bartosz.chronowski@xxxxxxxxx>
>---
>Changes in v2 (since the posted v1):
>  - Ordering update: keep older output on
>    sk_write_queue, where later sends append and shutdown pushes the
>    queue before FIN. Defer incoming local packets to sk_backlog
>    instead of detaching queued output into a separate TX queue.
>  - Initialize the backlog work before publishing the socket, and
>    inspect the TX queue and peer node only when no user owns the socket.
>
>v1: https://lore.kernel.org/r/631b75b4-d1e3-40ea-a129-
>5ed090f914bf@xxxxxxxxxxxxxxx
>
>Validation:
>  For the reported soft lockup, the C reproducer stalled all eight
>  unpatched guests. None of the eight patched guests crashed during
>  the 270-second run, with the same base and config.
>
>  For send/close ordering, a framed AF_TIPC stream test kept accepted
>  data in order in 300 later-send, 300 shutdown, and 300 last-close
>  cases on the patched image.
>
>  A complete workqueue trace recorded 547 balanced callbacks and
>  four requeues while work was active, with no overlapping callbacks.
>
> net/tipc/socket.c | 42 ++++++++++++++++++++++++++++++++++++++----
> 1 file changed, 38 insertions(+), 4 deletions(-)
>
>diff --git a/net/tipc/socket.c b/net/tipc/socket.c index
>65cc0c4964b31..b03dd4a3d3ea2 100644
>--- a/net/tipc/socket.c
>+++ b/net/tipc/socket.c
>@@ -94,6 +94,7 @@ struct sockaddr_pair {
>  * @peer: 'connected' peer for dgram/rdm
>  * @node: hash table node
>  * @mc_method: cookie for use between socket and broadcast layer
>+ * @backlog_work: processes receive backlog under socket ownership
>  * @rcu: rcu struct for tipc_sock
>  * @group: TIPC communications group
>  * @oneway: message count in one direction (FIXME) @@ -128,6 +129,7 @@
>struct tipc_sock {
> 	struct sockaddr_tipc peer;
> 	struct rhash_head node;
> 	struct tipc_mc_method mc_method;
>+	struct work_struct backlog_work;
> 	struct rcu_head rcu;
> 	struct tipc_group *group;
> 	u32 oneway;
>@@ -143,6 +145,7 @@ struct tipc_sock {
> };
>
> static int tipc_sk_backlog_rcv(struct sock *sk, struct sk_buff *skb);
>+static void tipc_sk_backlog_work(struct work_struct *work);
> static void tipc_data_ready(struct sock *sk);  static void tipc_write_space(struct
>sock *sk);  static void tipc_sock_destruct(struct sock *sk); @@ -499,6 +502,7
>@@ static int tipc_sk_create(struct net *net, struct socket *sock,
> 	/* Finish initializing socket data structures */
> 	sock->ops = ops;
> 	sock_init_data(sock, sk);
>+	INIT_WORK(&tsk->backlog_work, tipc_sk_backlog_work);
> 	tipc_set_sk_state(sk, TIPC_OPEN);
> 	if (tipc_sk_insert(tsk)) {
> 		sk_free(sk);
>@@ -2428,6 +2432,17 @@ static int tipc_sk_backlog_rcv(struct sock *sk, struct
>sk_buff *skb)
> 	return 0;
> }
>
>+static void tipc_sk_backlog_work(struct work_struct *work) {
>+	struct tipc_sock *tsk = container_of(work, struct tipc_sock,
>+					     backlog_work);
>+	struct sock *sk = &tsk->sk;
>+
>+	lock_sock(sk);
>+	release_sock(sk);
>+	sock_put(sk);
>+}
>+
> /**
>  * tipc_sk_enqueue - extract all buffers with destination 'dport' from
>  *                   inputq and try adding them to socket or backlog queue
>@@ -2445,18 +2460,30 @@ static void tipc_sk_enqueue(struct sk_buff_head
>*inputq, struct sock *sk,
> 	struct sk_buff *skb;
> 	unsigned int lim;
> 	atomic_t *dcnt;
>+	bool deferred = false;
>+	bool owned;
>+	bool local_backlog;
> 	u32 onode;
>
> 	while (skb_queue_len(inputq)) {
> 		if (unlikely(time_after_eq(jiffies, time_limit)))
>-			return;
>+			break;
>
> 		skb = tipc_skb_dequeue(inputq, dport);
> 		if (unlikely(!skb))
>-			return;
>+			break;
>
>-		/* Add message directly to receive queue if possible */
>-		if (!sock_owned_by_user(sk)) {
>+		/*
>+		 * A local Nagle backlog may loop back into this socket while its
>+		 * spinlock is held.  Queue the triggering input for processing
>under
>+		 * socket ownership, which keeps a concurrent sender from
>overtaking
>+		 * the already queued output.
>+		 */
>+		owned = sock_owned_by_user(sk);
>+		local_backlog = !owned && !skb_queue_empty(&sk-
>>sk_write_queue) &&
>+				in_own_node(sock_net(sk),
>+					    tsk_peer_node(tipc_sk(sk)));
>+		if (!owned && !local_backlog) {
> 			tipc_sk_filter_rcv(sk, skb, xmitq);
> 			continue;
> 		}

Bypassing tipc_sk_filter_rcv () and adding many skb to the backlog queue will quickly overload it and cause packet drop.

>@@ -2467,6 +2494,7 @@ static void tipc_sk_enqueue(struct sk_buff_head
>*inputq, struct sock *sk,
> 			atomic_set(dcnt, 0);
> 		lim = rcvbuf_limit(sk, skb) + atomic_read(dcnt);
> 		if (likely(!sk_add_backlog(sk, skb, lim))) {
>+			deferred |= !owned;
> 			trace_tipc_sk_overlimit1(sk, skb,
>TIPC_DUMP_SK_BKLGQ,
> 						 "bklg & rcvq >90%
>allocated!");
> 			continue;
>@@ -2483,6 +2511,12 @@ static void tipc_sk_enqueue(struct sk_buff_head
>*inputq, struct sock *sk,
> 		}
> 		break;
> 	}
>+
>+	if (deferred) {
>+		sock_hold(sk);
>+		if (!schedule_work(&tipc_sk(sk)->backlog_work))
>+			sock_put(sk);
>+	}

This approach is not good. It creates many work queues and could impact the performance.
Further more, as reported by sashiko, it does not cover the net namespace's communications via tipc_lxc_xmit().

--
pw-bot: cr