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@syzkaller.appspotmail.com Link: https://lore.kernel.org/r/GV1P189MB1988F6301FA670CA9818F7FCC6DB2@GV1P189MB1988.EURP189.PROD.OUTLOOK.COM Signed-off-by: Bart Chronowski --- 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@mail.kernel.org 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); + } } /**