RE: [PATCH net] tipc: serialize publication purging with name table updates

[Tung Quang Nguyen]

>Subject: [PATCH net] tipc: serialize publication purging with name table updates
>
>tipc_publ_notify() walks a failed node's publication list after the node lock has
>been released. Its list iterator is not protected by the name table lock, which is
>only acquired inside tipc_publ_purge().
>
>CPU A can save the next publication before entering tipc_publ_purge().
>CPU B then takes nametbl_lock in tipc_named_rcv(), processes a
>WITHDRAWAL for that publication, unlinks it with list_del_init(), and queues it
>for freeing with kfree_rcu(). CPU A advances to the removed publication and
>loops on its self-linked binding_node, stalling the CPU.
>
>The kernel reported:
>
>  rcu: INFO: rcu_sched self-detected stall on CPU
>  Call Trace:
>   tipc_publ_notify+0x3b5/0x650
>   tipc_node_write_unlock+0x49d/0x5d0
>   tipc_node_link_down+0x15c/0x4a0
>   tipc_node_delete_links+0xfc/0x190
>   bearer_disable+0x111/0x270
>   __tipc_nl_bearer_disable+0x1db/0x2f0
>   tipc_nl_bearer_disable+0x1c/0x30
>

Please update your changelog with decoded stack trace.

>Locking the entire traversal would also prevent the race, but would hold
>nametbl_lock with bottom halves disabled while purging every publication.
>A node with many publications could therefore cause excessive lock hold times
>and delay other name-table operations.
>
>Move the failed node's publications to a private list under nametbl_lock, then
>select and unlink its first entry under the same lock before purging it.
>Concurrent withdrawals can still unlink entries from this private list, while
>publications arriving after node recovery stay on the node's live list. No
>publication pointer is carried across an unlocked interval.
>
>Unlink before the table lookup to make progress even if the lookup fails, and
>acquire and release the lock for each publication. Check the private list under
>the lock on every iteration, since concurrent withdrawals can still remove
>entries. Keep withdrawal notifications and the final rc_dests update in their
>existing order.
>
>Remove the failed-node address argument from tipc_publ_notify() and its
>purge helper since unlinking no longer needs a node lookup.
>
>Fixes: 9db9fdd1983e ("tipc: avoid to asynchronously notify subscriptions")
>Cc: stable@xxxxxxxxxxxxxxx
>Assisted-by: GPT-6-Astra
>Signed-off-by: Chengfeng Ye <nicoyip.dev@xxxxxxxxx>
>---
> net/tipc/name_distr.c | 34 +++++++++++++++++++++++-----------
> net/tipc/name_distr.h |  2 +-
> net/tipc/node.c       |  2 +-
> 3 files changed, 25 insertions(+), 13 deletions(-)
>
>diff --git a/net/tipc/name_distr.c b/net/tipc/name_distr.c index
>ba4f4906e13b..34867b69044c 100644
>--- a/net/tipc/name_distr.c
>+++ b/net/tipc/name_distr.c
>@@ -227,38 +227,50 @@ void tipc_named_node_up(struct net *net, u32
>dnode, u16 capabilities)
>  * tipc_publ_purge - remove publication associated with a failed node
>  * @net: the associated network namespace
>  * @p: the publication to remove
>- * @addr: failed node's address
>  *
>  * Invoked for each publication issued by a newly failed node.
>  * Removes publication structure from name table & deletes it.
>+ * The caller must hold nametbl_lock and unlink the node subscription.
>  */
>-static void tipc_publ_purge(struct net *net, struct publication *p, u32 addr)
>+static void tipc_publ_purge(struct net *net, struct publication *p)
> {
>-	struct tipc_net *tn = tipc_net(net);
> 	struct publication *_p;
> 	struct tipc_uaddr ua;
>
> 	tipc_uaddr(&ua, TIPC_SERVICE_RANGE, p->scope, p->sr.type,
> 		   p->sr.lower, p->sr.upper);
>-	spin_lock_bh(&tn->nametbl_lock);
> 	_p = tipc_nametbl_remove_publ(net, &ua, &p->sk, p->key);
>-	if (_p)
>-		tipc_node_unsubscribe(net, &_p->binding_node, addr);
>-	spin_unlock_bh(&tn->nametbl_lock);
> 	if (_p)
> 		kfree_rcu(_p, rcu);
> }
>
> void tipc_publ_notify(struct net *net, struct list_head *nsub_list,
>-		      u32 addr, u16 capabilities)
>+		      u16 capabilities)
> {
> 	struct name_table *nt = tipc_name_table(net);
> 	struct tipc_net *tn = tipc_net(net);
>
>-	struct publication *publ, *tmp;
>+	struct publication *publ;
>+	LIST_HEAD(purge_list);
>+
>+	spin_lock_bh(&tn->nametbl_lock);
>+	/* Leave new publications on the node's list during the purge. */
>+	list_splice_init(nsub_list, &purge_list);
>+	spin_unlock_bh(&tn->nametbl_lock);

After this lock is released, new publications can be inserted into node->publ_list. Does this defeat the purpose of current publication release ?

>+
>+	for (;;) {
>+		spin_lock_bh(&tn->nametbl_lock);
>+		if (list_empty(&purge_list)) {
>+			spin_unlock_bh(&tn->nametbl_lock);
>+			break;
>+		}
>+		publ = list_first_entry(&purge_list, struct publication,
>+					binding_node);
>+		list_del_init(&publ->binding_node);
>+		tipc_publ_purge(net, publ);
>+		spin_unlock_bh(&tn->nametbl_lock);
>+	}
>
>-	list_for_each_entry_safe(publ, tmp, nsub_list, binding_node)
>-		tipc_publ_purge(net, publ, addr);
> 	spin_lock_bh(&tn->nametbl_lock);
> 	if (!(capabilities & TIPC_NAMED_BCAST))
> 		nt->rc_dests--;
>diff --git a/net/tipc/name_distr.h b/net/tipc/name_distr.h index
>c677f6f082df..8debe23469b2 100644
>--- a/net/tipc/name_distr.h
>+++ b/net/tipc/name_distr.h
>@@ -74,6 +74,6 @@ void tipc_named_rcv(struct net *net, struct sk_buff_head
>*namedq,
> 		    u16 *rcv_nxt, bool *open);
> void tipc_named_reinit(struct net *net);  void tipc_publ_notify(struct net *net,
>struct list_head *nsub_list,
>-		      u32 addr, u16 capabilities);
>+		      u16 capabilities);
>
> #endif
>diff --git a/net/tipc/node.c b/net/tipc/node.c index
>bd91378b7540..9a218d137c45 100644
>--- a/net/tipc/node.c
>+++ b/net/tipc/node.c
>@@ -422,7 +422,7 @@ static void tipc_node_write_unlock(struct tipc_node
>*n)
> 	write_unlock_bh(&n->lock);
>
> 	if (flags & TIPC_NOTIFY_NODE_DOWN)
>-		tipc_publ_notify(net, publ_list, node, n->capabilities);
>+		tipc_publ_notify(net, publ_list, n->capabilities);
>
> 	if (flags & TIPC_NOTIFY_NODE_UP)
> 		tipc_named_node_up(net, node, n->capabilities);
>--
>2.43.0