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

[Bart Chronowski]

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@xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
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;
 		}
@@ -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);
+	}
 }
 
 /**