[PATCH] tipc: defer local Nagle backlog xmit from receive path

[PATCH] tipc: defer local Nagle backlog xmit from receive path

[syzbot]

From: Bartosz Chronowski <immersa.bartosz.chronowski@gmail.com>

A local TIPC stream socket workload can trap a CPU in an endless receive
loop. The resulting soft lockup can make the system unavailable.

tipc_sk_push_backlog() can transmit delayed stream data while tipc_sk_rcv()
holds a destination socket's sk_lock.slock. Own-node transmission enters
tipc_sk_rcv() synchronously. A reply that reaches the ancestor socket
cannot acquire the still-held lock, and tipc_sk_rcv() retries without
consuming its input queue.

Pass the receive output queue to tipc_sk_push_backlog() from receive-side
callers. Queue own-node output there so the enclosing receive path sends it
after releasing the socket lock. Keep shutdown and remote-node transmission
on the existing direct path, preserving remote link congestion handling.

Fixes: c0bceb97db9e ("tipc: add smart nagle feature")
Assisted-by: Gemini:gemini-3.5-flash Gemini:gemini-3.1-pro-preview syzbot
Reported-by: syzbot+10a41dc44eef71aa9450@syzkaller.appspotmail.com
Closes: https://syzkaller.appspot.com/bug?extid=10a41dc44eef71aa9450
Link: https://syzkaller.appspot.com/ai_job?id=aa2dc129-33d3-42e7-bd0b-bf2198c9c9c2
Signed-off-by: Bartosz Chronowski <immersa.bartosz.chronowski@gmail.com>

---
diff --git a/net/tipc/socket.c b/net/tipc/socket.c
index e564341e0..33d744dc5 100644
--- a/net/tipc/socket.c
+++ b/net/tipc/socket.c
@@ -156,7 +156,8 @@ static int tipc_sk_insert(struct tipc_sock *tsk);
 static void tipc_sk_remove(struct tipc_sock *tsk);
 static int __tipc_sendstream(struct socket *sock, struct msghdr *m, size_t dsz);
 static int __tipc_sendmsg(struct socket *sock, struct msghdr *m, size_t dsz);
-static void tipc_sk_push_backlog(struct tipc_sock *tsk, bool nagle_ack);
+static void tipc_sk_push_backlog(struct tipc_sock *tsk, bool nagle_ack,
+				 struct sk_buff_head *xmitq);
 static int tipc_wait_for_connect(struct socket *sock, long *timeo_p);
 
 static const struct proto_ops packet_ops;
@@ -560,7 +561,7 @@ static void __tipc_shutdown(struct socket *sock, int error)
 					    !tsk_conn_cong(tsk)));
 
 	/* Push out delayed messages if in Nagle mode */
-	tipc_sk_push_backlog(tsk, false);
+	tipc_sk_push_backlog(tsk, false, NULL);
 	/* Remove pending SYN */
 	__skb_queue_purge(&sk->sk_write_queue);
 
@@ -1267,8 +1268,10 @@ void tipc_sk_mcast_rcv(struct net *net, struct sk_buff_head *arrvq,
 
 /* tipc_sk_push_backlog(): send accumulated buffers in socket write queue
  *                         when socket is in Nagle mode
+ * @xmitq: receive output queue, or NULL outside receive context
  */
-static void tipc_sk_push_backlog(struct tipc_sock *tsk, bool nagle_ack)
+static void tipc_sk_push_backlog(struct tipc_sock *tsk, bool nagle_ack,
+				 struct sk_buff_head *xmitq)
 {
 	struct sk_buff_head *txq = &tsk->sk.sk_write_queue;
 	struct sk_buff *skb = skb_peek_tail(txq);
@@ -1310,6 +1313,12 @@ static void tipc_sk_push_backlog(struct tipc_sock *tsk, bool nagle_ack)
 		tsk->pkt_cnt += skb_queue_len(txq);
 	tsk->snt_unacked += tsk->snd_backlog;
 	tsk->snd_backlog = 0;
+
+	if (xmitq && in_own_node(net, dnode)) {
+		skb_queue_splice_tail_init(txq, xmitq);
+		return;
+	}
+
 	rc = tipc_node_xmit(net, txq, dnode, tsk->portid);
 	if (rc == -ELINKCONG)
 		tsk->cong_link_cnt = 1;
@@ -1367,7 +1376,7 @@ static void tipc_sk_conn_proto_rcv(struct tipc_sock *tsk, struct sk_buff *skb,
 			goto exit;
 
 		was_cong = tsk_conn_cong(tsk);
-		tipc_sk_push_backlog(tsk, msg_nagle_ack(hdr));
+		tipc_sk_push_backlog(tsk, msg_nagle_ack(hdr), xmitq);
 		tsk->snt_unacked -= msg_conn_ack(hdr);
 		if (tsk->peer_caps & TIPC_BLOCK_FLOWCTL)
 			tsk->snd_win = msg_adv_win(hdr);
@@ -2165,7 +2174,7 @@ static void tipc_sk_proto_rcv(struct sock *sk,
 		smp_wmb();
 		tsk->cong_link_cnt--;
 		wakeup = true;
-		tipc_sk_push_backlog(tsk, false);
+		tipc_sk_push_backlog(tsk, false, xmitq);
 		break;
 	case GROUP_PROTOCOL:
 		tipc_group_proto_rcv(grp, &wakeup, hdr, inputq, xmitq);
@@ -2256,7 +2265,7 @@ static bool tipc_sk_filter_connect(struct tipc_sock *tsk, struct sk_buff *skb,
 		return false;
 	case TIPC_ESTABLISHED:
 		if (!skb_queue_empty(&sk->sk_write_queue))
-			tipc_sk_push_backlog(tsk, false);
+			tipc_sk_push_backlog(tsk, false, xmitq);
 		/* Accept only connection-based messages sent by peer */
 		if (likely(con_msg && !err && pport == oport &&
 			   pnode == onode)) {


base-commit: 1590cf0329716306e948a8fc29f1d3ee87d3989f
-- 
See https://goo.gle/syzbot-ai-patches for information about AI-generated patches.
The person who has signed off on the patch is responsible for
addressing comments.
syzbot engineers can be reached at syzkaller@googlegroups.com.

RE: [PATCH] tipc: defer local Nagle backlog xmit from receive path

[Tung Quang Nguyen]

>Subject: [PATCH] tipc: defer local Nagle backlog xmit from receive path
>
>From: Bartosz Chronowski <immersa.bartosz.chronowski@gmail.com>
>
>A local TIPC stream socket workload can trap a CPU in an endless receive loop.
>The resulting soft lockup can make the system unavailable.
>
>tipc_sk_push_backlog() can transmit delayed stream data while tipc_sk_rcv()
>holds a destination socket's sk_lock.slock. Own-node transmission enters
>tipc_sk_rcv() synchronously. A reply that reaches the ancestor socket cannot
>acquire the still-held lock, and tipc_sk_rcv() retries without consuming its input
>queue.
>
>Pass the receive output queue to tipc_sk_push_backlog() from receive-side
>callers. Queue own-node output there so the enclosing receive path sends it
>after releasing the socket lock. Keep shutdown and remote-node transmission
>on the existing direct path, preserving remote link congestion handling.
>
>Fixes: c0bceb97db9e ("tipc: add smart nagle feature")
>Assisted-by: Gemini:gemini-3.5-flash Gemini:gemini-3.1-pro-preview syzbot
>Reported-by: syzbot+10a41dc44eef71aa9450@syzkaller.appspotmail.com
>Closes: https://syzkaller.appspot.com/bug?extid=10a41dc44eef71aa9450
>Link: https://syzkaller.appspot.com/ai_job?id=aa2dc129-33d3-42e7-bd0b-
>bf2198c9c9c2
>Signed-off-by: Bartosz Chronowski <immersa.bartosz.chronowski@gmail.com>
>
>---
>diff --git a/net/tipc/socket.c b/net/tipc/socket.c index e564341e0..33d744dc5
>100644
>--- a/net/tipc/socket.c
>+++ b/net/tipc/socket.c
>@@ -156,7 +156,8 @@ static int tipc_sk_insert(struct tipc_sock *tsk);  static
>void tipc_sk_remove(struct tipc_sock *tsk);  static int __tipc_sendstream(struct
>socket *sock, struct msghdr *m, size_t dsz);  static int __tipc_sendmsg(struct
>socket *sock, struct msghdr *m, size_t dsz); -static void
>tipc_sk_push_backlog(struct tipc_sock *tsk, bool nagle_ack);
>+static void tipc_sk_push_backlog(struct tipc_sock *tsk, bool nagle_ack,
>+				 struct sk_buff_head *xmitq);
> static int tipc_wait_for_connect(struct socket *sock, long *timeo_p);
>
> static const struct proto_ops packet_ops; @@ -560,7 +561,7 @@ static void
>__tipc_shutdown(struct socket *sock, int error)
> 					    !tsk_conn_cong(tsk)));
>
> 	/* Push out delayed messages if in Nagle mode */
>-	tipc_sk_push_backlog(tsk, false);
>+	tipc_sk_push_backlog(tsk, false, NULL);

This is wrong because it breaks user applications (send()/close()) as well as current implementation.
All pending messages in the socket queue need to be delivered before closing that socket.

> 	/* Remove pending SYN */
> 	__skb_queue_purge(&sk->sk_write_queue);
>
>@@ -1267,8 +1268,10 @@ void tipc_sk_mcast_rcv(struct net *net, struct
>sk_buff_head *arrvq,
>
> /* tipc_sk_push_backlog(): send accumulated buffers in socket write queue
>  *                         when socket is in Nagle mode
>+ * @xmitq: receive output queue, or NULL outside receive context
>  */
>-static void tipc_sk_push_backlog(struct tipc_sock *tsk, bool nagle_ack)
>+static void tipc_sk_push_backlog(struct tipc_sock *tsk, bool nagle_ack,
>+				 struct sk_buff_head *xmitq)
> {
> 	struct sk_buff_head *txq = &tsk->sk.sk_write_queue;
> 	struct sk_buff *skb = skb_peek_tail(txq); @@ -1310,6 +1313,12 @@
>static void tipc_sk_push_backlog(struct tipc_sock *tsk, bool nagle_ack)
> 		tsk->pkt_cnt += skb_queue_len(txq);
> 	tsk->snt_unacked += tsk->snd_backlog;
> 	tsk->snd_backlog = 0;
>+
>+	if (xmitq && in_own_node(net, dnode)) {
>+		skb_queue_splice_tail_init(txq, xmitq);
>+		return;
>+	}
>+
> 	rc = tipc_node_xmit(net, txq, dnode, tsk->portid);
> 	if (rc == -ELINKCONG)
> 		tsk->cong_link_cnt = 1;
>@@ -1367,7 +1376,7 @@ static void tipc_sk_conn_proto_rcv(struct tipc_sock
>*tsk, struct sk_buff *skb,
> 			goto exit;
>
> 		was_cong = tsk_conn_cong(tsk);
>-		tipc_sk_push_backlog(tsk, msg_nagle_ack(hdr));
>+		tipc_sk_push_backlog(tsk, msg_nagle_ack(hdr), xmitq);

Passing a stack-based queue (local variable) to tipc_sk_push_backlog() is a wrong approach.
This will break user applications because it will create out-of-order messages at receiver.

' tsk->sk.sk_write_queue' must always be sent under lock protection. After you copy its skbs to the local ' xmitq', 
there could be 2 threads sending ' tsk->sk.sk_write_queue' and 'xmitq'. This causes disordered messages.

> 		tsk->snt_unacked -= msg_conn_ack(hdr);
> 		if (tsk->peer_caps & TIPC_BLOCK_FLOWCTL)
> 			tsk->snd_win = msg_adv_win(hdr);
>@@ -2165,7 +2174,7 @@ static void tipc_sk_proto_rcv(struct sock *sk,
> 		smp_wmb();
> 		tsk->cong_link_cnt--;
> 		wakeup = true;
>-		tipc_sk_push_backlog(tsk, false);
>+		tipc_sk_push_backlog(tsk, false, xmitq);
> 		break;
> 	case GROUP_PROTOCOL:
> 		tipc_group_proto_rcv(grp, &wakeup, hdr, inputq, xmitq); @@ -
>2256,7 +2265,7 @@ static bool tipc_sk_filter_connect(struct tipc_sock *tsk,
>struct sk_buff *skb,
> 		return false;
> 	case TIPC_ESTABLISHED:
> 		if (!skb_queue_empty(&sk->sk_write_queue))
>-			tipc_sk_push_backlog(tsk, false);
>+			tipc_sk_push_backlog(tsk, false, xmitq);
> 		/* Accept only connection-based messages sent by peer */
> 		if (likely(con_msg && !err && pport == oport &&
> 			   pnode == onode)) {
>
>
>base-commit: 1590cf0329716306e948a8fc29f1d3ee87d3989f
>--
>See https://goo.gle/syzbot-ai-patches for information about AI-generated
>patches.
>The person who has signed off on the patch is responsible for addressing
>comments.
>syzbot engineers can be reached at syzkaller@googlegroups.com.

Re: [PATCH] tipc: defer local Nagle backlog xmit from receive path

[Bartosz Chronowski]

#syz test: https://git.kernel.org/pub/scm/linux/kernel/git/torvalds/linux.git 1590cf0329716306e948a8fc29f1d3ee87d3989f

diff --git a/net/tipc/socket.c b/net/tipc/socket.c
index e564341e0216d35ae4fe539576b6e0be89b84ddb..0bf259672827db095046cda5646df047002dd663 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);
@@ -520,6 +523,7 @@ static int tipc_sk_create(struct net *net, struct socket *sock,
 	sk->sk_data_ready = tipc_data_ready;
 	sk->sk_write_space = tipc_write_space;
 	sk->sk_destruct = tipc_sock_destruct;
+	INIT_WORK(&tsk->backlog_work, tipc_sk_backlog_work);
 	tsk->conn_timeout = CONN_TIMEOUT_DEFAULT;
 	tsk->group_is_open = true;
 	atomic_set(&tsk->dupl_rcvcnt, 0);
@@ -2417,6 +2421,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
@@ -2434,18 +2449,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 = !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;
 		}
@@ -2456,6 +2483,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;
@@ -2472,6 +2500,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);
+	}
 }
 
 /**

Re: [syzbot] [tipc?] BUG: soft lockup in do_sock_setsockopt

[syzbot]

Hello,

syzbot has tested the proposed patch and the reproducer did not trigger any issue:

Reported-by: syzbot+10a41dc44eef71aa9450@syzkaller.appspotmail.com
Tested-by: syzbot+10a41dc44eef71aa9450@syzkaller.appspotmail.com

Tested on:

commit:         1590cf03 Linux 7.2-rc4
git tree:       https://git.kernel.org/pub/scm/linux/kernel/git/torvalds/linux.git
console output: https://syzkaller.appspot.com/x/log.txt?x=17a00949580000
kernel config:  https://syzkaller.appspot.com/x/.config?x=8e9390a601b4f3bd
dashboard link: https://syzkaller.appspot.com/bug?extid=10a41dc44eef71aa9450
compiler:       gcc (Debian 14.2.0-19) 14.2.0, GNU ld (GNU Binutils for Debian) 2.44
patch:          https://syzkaller.appspot.com/x/patch.diff?x=10baba9e580000

Note: testing is done by a robot and is best-effort only.