* [PATCH net 1/2] tipc: fix self-deadlock in tipc_named_node_up
2026-10-28 3:36 [PATCH net 0/2] tipc: fix deadlock and NUll pointer dereference in tipc_named_node_up() Tung Nguyen
@ 2026-10-28 3:36 ` Tung Nguyen
2026-10-29 4:27 ` netdev-bot+sashiko
2026-10-28 3:36 ` [PATCH net 2/2] tipc: fix NULL pointer dereference in named_distribute() Tung Nguyen
1 sibling, 1 reply; 6+ messages in thread
From: Tung Nguyen @ 2026-10-28 3:36 UTC (permalink / raw)
To: netdev
Cc: davem, kuba, edumazet, pabeni, jmaloy, horms, tipc-discussion,
Tung Nguyen
The following deadlock can occur, as observed during code inspection:
tipc_named_node_up()
read_lock_bh(&nt->cluster_scope_lock) <-- acquire read lock
named_distribute()
tipc_node_xmit()
tipc_node_link_down()
tipc_node_write_unlock()
tipc_nametbl_withdraw()
spin_lock_bh(&tn->nametbl_lock)
tipc_named_withdraw()
write_lock_bh(&nt->cluster_scope_lock) <-- acquire write lock
Fix this issue by:
- Removing cluster_scope_lock, which protects the name table's cluster_scope
list. Use nametbl_lock instead to protect the list against races when multiple
threads concurrently call tipc_named_publish(), tipc_named_withdraw(),
and tipc_named_node_up().
- Introducing a new function, tipc_node_distribute(), which is similar to
tipc_node_xmit() but does not call tipc_node_link_down(). Both functions call
the newly introduced tipc_node_build_list(), which builds a list of messages
in sequence.
Fixes: 988f3f1603d4 ("tipc: eliminate message disordering during binding table update")
Signed-off-by: Tung Nguyen <tung.quang.nguyen@est.tech>
---
net/tipc/name_distr.c | 32 ++++++----
net/tipc/name_distr.h | 2 +-
net/tipc/name_table.c | 1 -
net/tipc/name_table.h | 2 -
net/tipc/node.c | 140 ++++++++++++++++++++++++++++++++----------
net/tipc/node.h | 2 +
6 files changed, 130 insertions(+), 49 deletions(-)
diff --git a/net/tipc/name_distr.c b/net/tipc/name_distr.c
index ba4f4906e13b..a60f213ed118 100644
--- a/net/tipc/name_distr.c
+++ b/net/tipc/name_distr.c
@@ -95,9 +95,7 @@ struct sk_buff *tipc_named_publish(struct net *net, struct publication *p)
list_add_tail_rcu(&p->binding_node, &nt->node_scope);
return NULL;
}
- write_lock_bh(&nt->cluster_scope_lock);
list_add_tail(&p->binding_node, &nt->cluster_scope);
- write_unlock_bh(&nt->cluster_scope_lock);
skb = named_prepare_buf(net, PUBLICATION, ITEM_SIZE, 0);
if (!skb) {
pr_warn("Publication distribution failure\n");
@@ -121,9 +119,7 @@ struct sk_buff *tipc_named_withdraw(struct net *net, struct publication *p)
struct distr_item *item;
struct sk_buff *skb;
- write_lock_bh(&nt->cluster_scope_lock);
list_del(&p->binding_node);
- write_unlock_bh(&nt->cluster_scope_lock);
if (p->scope == TIPC_NODE_SCOPE)
return NULL;
@@ -202,25 +198,39 @@ static void named_distribute(struct net *net, struct sk_buff_head *list,
* @net: the associated network namespace
* @dnode: destination node
* @capabilities: peer node's capabilities
+ *
+ * Return:
+ * * 0 - Success
+ * * -ENOBUFS - No buffer space is available
+ * * -EHOSTUNREACH - Destination node does not exist or cannot be reached
+ *
*/
-void tipc_named_node_up(struct net *net, u32 dnode, u16 capabilities)
+int tipc_named_node_up(struct net *net, u32 dnode, u16 capabilities)
{
struct name_table *nt = tipc_name_table(net);
struct tipc_net *tn = tipc_net(net);
struct sk_buff_head head;
- u16 seqno;
+ int bearer_id = 0;
+ int rc = 0;
__skb_queue_head_init(&head);
spin_lock_bh(&tn->nametbl_lock);
if (!(capabilities & TIPC_NAMED_BCAST))
nt->rc_dests++;
- seqno = nt->snd_nxt;
+ named_distribute(net, &head, dnode, &nt->cluster_scope, nt->snd_nxt);
+ if (!skb_queue_empty(&head))
+ rc = tipc_node_distribute(net, &head, dnode, &bearer_id, 0);
spin_unlock_bh(&tn->nametbl_lock);
- read_lock_bh(&nt->cluster_scope_lock);
- named_distribute(net, &head, dnode, &nt->cluster_scope, seqno);
- tipc_node_xmit(net, &head, dnode, 0);
- read_unlock_bh(&nt->cluster_scope_lock);
+ /* The link is congested after the list is inserted into
+ * the link's send queue. Return 0, as this is normal.
+ */
+ if (rc == -ELINKCONG)
+ rc = 0;
+ else if (rc == -EMSGSIZE)
+ rc = -ENOBUFS;
+
+ return rc;
}
/**
diff --git a/net/tipc/name_distr.h b/net/tipc/name_distr.h
index c677f6f082df..3f41e36efee3 100644
--- a/net/tipc/name_distr.h
+++ b/net/tipc/name_distr.h
@@ -69,7 +69,7 @@ struct distr_item {
struct sk_buff *tipc_named_publish(struct net *net, struct publication *publ);
struct sk_buff *tipc_named_withdraw(struct net *net, struct publication *publ);
-void tipc_named_node_up(struct net *net, u32 dnode, u16 capabilities);
+int tipc_named_node_up(struct net *net, u32 dnode, u16 capabilities);
void tipc_named_rcv(struct net *net, struct sk_buff_head *namedq,
u16 *rcv_nxt, bool *open);
void tipc_named_reinit(struct net *net);
diff --git a/net/tipc/name_table.c b/net/tipc/name_table.c
index 6fda36ab1766..8924bdb6a443 100644
--- a/net/tipc/name_table.c
+++ b/net/tipc/name_table.c
@@ -921,7 +921,6 @@ int tipc_nametbl_init(struct net *net)
INIT_LIST_HEAD(&nt->node_scope);
INIT_LIST_HEAD(&nt->cluster_scope);
- rwlock_init(&nt->cluster_scope_lock);
tn->nametbl = nt;
spin_lock_init(&tn->nametbl_lock);
return 0;
diff --git a/net/tipc/name_table.h b/net/tipc/name_table.h
index 7ff6eeebaae6..afe665e68bb1 100644
--- a/net/tipc/name_table.h
+++ b/net/tipc/name_table.h
@@ -97,7 +97,6 @@ struct publication {
* @cluster_scope: all local publications with cluster scope
* - used by name_distr to send bulk updates to new nodes
* - used by name_distr during re-init of name table
- * @cluster_scope_lock: lock for accessing @cluster_scope
* @local_publ_count: number of publications issued by this node
* @rc_dests: destination node counter
* @snd_nxt: next sequence number to be used
@@ -107,7 +106,6 @@ struct name_table {
struct hlist_head services[TIPC_NAMETBL_SIZE];
struct list_head node_scope;
struct list_head cluster_scope;
- rwlock_t cluster_scope_lock;
u32 local_publ_count;
u32 rc_dests;
u32 snd_nxt;
diff --git a/net/tipc/node.c b/net/tipc/node.c
index 53b0993c2c23..b19d6652dc73 100644
--- a/net/tipc/node.c
+++ b/net/tipc/node.c
@@ -402,6 +402,7 @@ static void tipc_node_write_unlock(struct tipc_node *n)
struct list_head *publ_list;
struct tipc_uaddr ua;
u32 bearer_id, node;
+ int rc = 0;
if (likely(!flags)) {
write_unlock_bh(&n->lock);
@@ -425,7 +426,7 @@ static void tipc_node_write_unlock(struct tipc_node *n)
tipc_publ_notify(net, publ_list, node, n->capabilities);
if (flags & TIPC_NOTIFY_NODE_UP)
- tipc_named_node_up(net, node, n->capabilities);
+ rc = tipc_named_node_up(net, node, n->capabilities);
if (flags & TIPC_NOTIFY_LINK_UP) {
tipc_mon_peer_up(net, node, bearer_id);
@@ -435,6 +436,9 @@ static void tipc_node_write_unlock(struct tipc_node *n)
tipc_mon_peer_down(net, node, bearer_id);
tipc_nametbl_withdraw(net, &ua, &sk, sk.ref);
}
+
+ if (rc == -ENOBUFS)
+ tipc_node_link_down(n, bearer_id, false);
}
static void tipc_node_assign_peer_net(struct tipc_node *n, u32 hash_mixes)
@@ -1682,6 +1686,51 @@ static void tipc_lxc_xmit(struct net *peer_net, struct sk_buff_head *list)
}
}
+static int tipc_node_build_list(struct tipc_node *n, struct sk_buff_head *list,
+ struct sk_buff_head *xmitq, int selector,
+ struct tipc_media_addr *maddr,
+ int *bearer_id, bool *lxc)
+{
+ struct tipc_link_entry *le = NULL;
+ bool node_up = false;
+ struct net *peer_net;
+ int rc = 0;
+
+ rcu_read_lock();
+ tipc_node_read_lock(n);
+ node_up = node_is_up(n);
+ peer_net = n->peer_net;
+ tipc_node_read_unlock(n);
+
+ if (node_up && peer_net && check_net(peer_net)) {
+ /* xmit inner linux container */
+ tipc_lxc_xmit(peer_net, list);
+ if (likely(skb_queue_empty(list))) {
+ rcu_read_unlock();
+ *lxc = true;
+ return 0;
+ }
+ }
+ rcu_read_unlock();
+
+ tipc_node_read_lock(n);
+ *bearer_id = n->active_links[selector & 1];
+ if (unlikely(*bearer_id == INVALID_BEARER_ID)) {
+ tipc_node_read_unlock(n);
+ __skb_queue_purge(list);
+ return -EHOSTUNREACH;
+ }
+
+ le = &n->links[*bearer_id];
+ *maddr = le->maddr;
+ spin_lock_bh(&le->lock);
+ rc = tipc_link_xmit(le->link, list, xmitq);
+ spin_unlock_bh(&le->lock);
+ tipc_node_read_unlock(n);
+
+ return rc;
+}
+
/**
* tipc_node_xmit() - general link level function for message sending
* @net: the applicable net namespace
@@ -1694,12 +1743,11 @@ static void tipc_lxc_xmit(struct net *peer_net, struct sk_buff_head *list)
int tipc_node_xmit(struct net *net, struct sk_buff_head *list,
u32 dnode, int selector)
{
- struct tipc_link_entry *le = NULL;
- struct tipc_node *n;
+ struct tipc_media_addr maddr;
struct sk_buff_head xmitq;
- bool node_up = false;
- struct net *peer_net;
- int bearer_id;
+ struct tipc_node *n;
+ int bearer_id = 0;
+ bool lxc = false;
int rc;
if (in_own_node(net, dnode)) {
@@ -1715,42 +1763,66 @@ int tipc_node_xmit(struct net *net, struct sk_buff_head *list,
return -EHOSTUNREACH;
}
- rcu_read_lock();
- tipc_node_read_lock(n);
- node_up = node_is_up(n);
- peer_net = n->peer_net;
- tipc_node_read_unlock(n);
- if (node_up && peer_net && check_net(peer_net)) {
- /* xmit inner linux container */
- tipc_lxc_xmit(peer_net, list);
- if (likely(skb_queue_empty(list))) {
- rcu_read_unlock();
- tipc_node_put(n);
- return 0;
- }
- }
- rcu_read_unlock();
+ __skb_queue_head_init(&xmitq);
+ rc = tipc_node_build_list(n, list, &xmitq, selector, &maddr,
+ &bearer_id, &lxc);
- tipc_node_read_lock(n);
- bearer_id = n->active_links[selector & 1];
- if (unlikely(bearer_id == INVALID_BEARER_ID)) {
- tipc_node_read_unlock(n);
+ if (lxc) {
tipc_node_put(n);
+ return 0;
+ }
+
+ if (unlikely(rc == -ENOBUFS))
+ tipc_node_link_down(n, bearer_id, false);
+ else
+ tipc_bearer_xmit(net, bearer_id, &xmitq, &maddr, n);
+
+ tipc_node_put(n);
+
+ return rc;
+}
+
+/**
+ * tipc_node_distribute() - distribute list of messages to specific destination
+ * @net: the applicable net namespace
+ * @list: chain of buffers containing message
+ * @dnode: address of destination node
+ * @bearer_id: id of the bearer
+ * @selector: a number used for deterministic link selection
+ *
+ * Return:
+ * * 0 - Success
+ * * -ELINKCONG - Link is congested
+ * * -EHOSTUNREACH - Host is not reachable
+ * * -EMSGSIZE - Message's size is too big
+ * * -ENOBUF - No buffer space is available
+ *
+ */
+int tipc_node_distribute(struct net *net, struct sk_buff_head *list,
+ u32 dnode, int *bearer_id, int selector)
+{
+ struct tipc_media_addr maddr;
+ struct sk_buff_head xmitq;
+ struct tipc_node *n;
+ bool lxc = false;
+ int rc;
+
+ n = tipc_node_find(net, dnode);
+ if (unlikely(!n)) {
__skb_queue_purge(list);
return -EHOSTUNREACH;
}
__skb_queue_head_init(&xmitq);
- le = &n->links[bearer_id];
- spin_lock_bh(&le->lock);
- rc = tipc_link_xmit(le->link, list, &xmitq);
- spin_unlock_bh(&le->lock);
- tipc_node_read_unlock(n);
+ rc = tipc_node_build_list(n, list, &xmitq, selector, &maddr,
+ bearer_id, &lxc);
+ if (lxc) {
+ tipc_node_put(n);
+ return 0;
+ }
- if (unlikely(rc == -ENOBUFS))
- tipc_node_link_down(n, bearer_id, false);
- else
- tipc_bearer_xmit(net, bearer_id, &xmitq, &le->maddr, n);
+ if (likely(rc != -ENOBUFS))
+ tipc_bearer_xmit(net, *bearer_id, &xmitq, &maddr, n);
tipc_node_put(n);
diff --git a/net/tipc/node.h b/net/tipc/node.h
index 154a5bbb0d29..775a8ec240fb 100644
--- a/net/tipc/node.h
+++ b/net/tipc/node.h
@@ -100,6 +100,8 @@ int tipc_node_get_linkname(struct net *net, u32 bearer_id, u32 node,
char *linkname, size_t len);
int tipc_node_xmit(struct net *net, struct sk_buff_head *list, u32 dnode,
int selector);
+int tipc_node_distribute(struct net *net, struct sk_buff_head *list,
+ u32 dnode, int *bearer_id, int selector);
int tipc_node_distr_xmit(struct net *net, struct sk_buff_head *list);
int tipc_node_xmit_skb(struct net *net, struct sk_buff *skb, u32 dest,
u32 selector);
--
2.53.0
^ permalink raw reply related [flat|nested] 6+ messages in thread* [PATCH net 2/2] tipc: fix NULL pointer dereference in named_distribute()
2026-10-28 3:36 [PATCH net 0/2] tipc: fix deadlock and NUll pointer dereference in tipc_named_node_up() Tung Nguyen
2026-10-28 3:36 ` [PATCH net 1/2] tipc: fix self-deadlock in tipc_named_node_up Tung Nguyen
@ 2026-10-28 3:36 ` Tung Nguyen
2026-10-29 4:27 ` netdev-bot+sashiko
2026-10-29 14:10 ` Paolo Abeni
1 sibling, 2 replies; 6+ messages in thread
From: Tung Nguyen @ 2026-10-28 3:36 UTC (permalink / raw)
To: netdev
Cc: davem, kuba, edumazet, pabeni, jmaloy, horms, tipc-discussion,
Tung Nguyen, syzbot+64197f7577a09686ffdb
Syzbot reports NULL pointer dereference in named_distribute():
...
RIP: 0010:buf_msg net/tipc/msg.h:202 [inline]
RIP: 0010:named_distribute net/tipc/name_distr.c:195 [inline]
RIP: 0010:tipc_named_node_up+0x8e0/0xb00 net/tipc/name_distr.c:221
...
tipc_node_write_unlock+0x2ce/0x4f0 net/tipc/node.c:428
tipc_rcv+0xb55/0x1020 net/tipc/node.c:2196
tipc_l2_rcv_msg+0x2d9/0x470 net/tipc/bearer.c:671
The root cause is racing between tipc_net_finalize() and
named_distribute() as follows:
Thread 1 (tipc_net_finalize) | Thread 2 (named_distribute)
-----------------------------|-----------------------------
| ...
| list_for_each_entry(publ, pls, binding_node) {
| ...
| __skb_queue_tail(list, skb);
| ...
| }
| ...
| hdr = buf_msg(skb_peek_tail(list));
... |
tipc_nametbl_publish(); |
If 'tipc_nametbl_publish()' (Thread 1) executes after 'named_distribute()'
, list will be empty. As a result, NULL is passed to 'buf_msg()', leading
to a NULL pointer dereference.
Fix the this issue by checking for the node's self-address publication on
either node-up or node-timeout event. If the publication does not exist, insert
it into the name table and publish it to the other nodes.
To facilitate the implementation, three major changes are made:
- Move the sc->lock acquisition out of tipc_service_insert_publ() and into its
callers.
- Add an output parameter (*err) to tipc_nametbl_publish() to detect memory
allocation errors.
- Add three fields (nt_stop, nt_self_node_exist, and node_addr_set) to struct
tipc_net to synchronize the insertion and deletion of the node's self-address
publication.
Reported-by: syzbot+64197f7577a09686ffdb@syzkaller.appspotmail.com
Closes: https://syzkaller.appspot.com/bug?extid=64197f7577a09686ffdb
Fixes: adba75be0d23 ("tipc: fix lockdep warning when reinitilaizing sockets")
Signed-off-by: Tung Nguyen <tung.quang.nguyen@est.tech>
---
net/tipc/core.c | 3 +
net/tipc/core.h | 6 ++
net/tipc/name_distr.c | 120 ++++++++++++++++++++++++++++++++++---
net/tipc/name_distr.h | 1 +
net/tipc/name_table.c | 134 ++++++++++++++++++++++++++++++++++++++----
net/tipc/name_table.h | 6 +-
net/tipc/net.c | 9 ++-
net/tipc/node.c | 61 ++++++++++++++++---
net/tipc/node.h | 1 +
net/tipc/socket.c | 2 +-
10 files changed, 314 insertions(+), 29 deletions(-)
diff --git a/net/tipc/core.c b/net/tipc/core.c
index 315975c3be81..0f1d3ddeae09 100644
--- a/net/tipc/core.c
+++ b/net/tipc/core.c
@@ -61,6 +61,9 @@ static int __net_init tipc_init_net(struct net *net)
tn->trial_addr = 0;
tn->addr_trial_end = 0;
tn->capabilities = TIPC_NODE_CAPABILITIES;
+ tn->nt_stop = false;
+ tn->nt_self_node_exist = false;
+ atomic_set(&tn->node_addr_set, 0);
INIT_WORK(&tn->work, tipc_net_finalize_work);
memset(tn->node_id, 0, sizeof(tn->node_id));
memset(tn->node_id_string, 0, sizeof(tn->node_id_string));
diff --git a/net/tipc/core.h b/net/tipc/core.h
index 9ce5f9ff6cc0..975b1d48186e 100644
--- a/net/tipc/core.h
+++ b/net/tipc/core.h
@@ -145,6 +145,12 @@ struct tipc_net {
struct work_struct work;
/* The numbers of work queues in schedule */
atomic_t wq_count;
+ /* Flag to indicate that the node's address is set */
+ atomic_t node_addr_set;
+ /* The name table has stopped working */
+ bool nt_stop;
+ /* Node's self address publication is available in the name table */
+ bool nt_self_node_exist;
};
static inline struct tipc_net *tipc_net(struct net *net)
diff --git a/net/tipc/name_distr.c b/net/tipc/name_distr.c
index a60f213ed118..345bd767983e 100644
--- a/net/tipc/name_distr.c
+++ b/net/tipc/name_distr.c
@@ -142,9 +142,14 @@ struct sk_buff *tipc_named_withdraw(struct net *net, struct publication *p)
* @dnode: node to be updated
* @pls: linked list of publication items to be packed into buffer chain
* @seqno: sequence number for this message
+ *
+ * Return:
+ * * 0 - Success
+ * * -ENOBUFS - No buffer space is available
+ *
*/
-static void named_distribute(struct net *net, struct sk_buff_head *list,
- u32 dnode, struct list_head *pls, u16 seqno)
+static int named_distribute(struct net *net, struct sk_buff_head *list,
+ u32 dnode, struct list_head *pls, u16 seqno)
{
struct publication *publ;
struct sk_buff *skb = NULL;
@@ -160,8 +165,9 @@ static void named_distribute(struct net *net, struct sk_buff_head *list,
skb = named_prepare_buf(net, PUBLICATION, msg_rem,
dnode);
if (!skb) {
+ __skb_queue_purge(list);
pr_warn("Bulk publication failure\n");
- return;
+ return -ENOBUFS;
}
hdr = buf_msg(skb);
msg_set_bc_ack_invalid(hdr, true);
@@ -191,6 +197,94 @@ static void named_distribute(struct net *net, struct sk_buff_head *list,
hdr = buf_msg(skb_peek_tail(list));
msg_set_last_bulk(hdr);
msg_set_named_seqno(hdr, seqno);
+
+ return 0;
+}
+
+/**
+ * __tipc_named_distribute - build a list of publications need to be distributed
+ * @net: the associated network namespace
+ * @dnode: destination node
+ * @head: list of publications
+ *
+ * Return:
+ * * 0 - Success
+ * * -ENOBUFS - No buffer space is available
+ *
+ */
+static int __tipc_named_distribute(struct net *net, u32 dnode,
+ struct sk_buff_head *head)
+{
+ struct name_table *nt = tipc_name_table(net);
+ struct tipc_net *tn = tipc_net(net);
+
+ /* Name table has been deleted after namespace teardown or
+ * TIPC module exit.
+ */
+ if (unlikely(tn->nt_stop))
+ return 0;
+
+ /* Node's self-address is not set yet */
+ if (!atomic_read(&tn->node_addr_set)) {
+ /* Defer the distribution to node's timer */
+ tipc_node_set_dist_bulk(net, dnode);
+ return 0;
+ }
+
+ /* Previous call to tipc_nametb_insert_self_node_pub() was successful.
+ * The node's self-address publication was added to 'nt->cluster_scope'.
+ * So, 'nt->cluster_scope' is not empty now. Go to distribution.
+ */
+ if (tn->nt_self_node_exist)
+ goto distribute;
+
+ if (tipc_nametb_insert_self_node_pub(net))
+ return -ENOBUFS;
+
+ tn->nt_self_node_exist = true;
+
+distribute:
+ if (named_distribute(net, head, dnode,
+ &nt->cluster_scope, nt->snd_nxt) == -ENOBUFS)
+ return -ENOBUFS;
+
+ return 0;
+}
+
+/**
+ * tipc_named_distribute - distribute all publications to specified node
+ * @net: the associated network namespace
+ * @dnode: destination node
+ *
+ * Return:
+ * * 0 - Success
+ * * -ENOBUFS - No buffer space is available
+ * * -EHOSTUNREACH - Destination node does not exist or cannot be reached
+ *
+ */
+int tipc_named_distribute(struct net *net, u32 dnode)
+{
+ struct tipc_net *tn = tipc_net(net);
+ struct sk_buff_head head;
+ int bearer_id = 0;
+ int rc = 0;
+
+ __skb_queue_head_init(&head);
+ spin_lock_bh(&tn->nametbl_lock);
+ rc = __tipc_named_distribute(net, dnode, &head);
+ if (!rc && !skb_queue_empty(&head))
+ rc = tipc_node_distribute(net, &head, dnode, &bearer_id, 0);
+ spin_unlock_bh(&tn->nametbl_lock);
+
+ /* The link is congested after the list is inserted into
+ * the link's send queue. Return 0, as this is normal.
+ */
+ if (rc == -ELINKCONG)
+ rc = 0;
+ else if (rc == -EMSGSIZE)
+ rc = -ENOBUFS;
+
+ return rc;
}
/**
@@ -217,8 +311,8 @@ int tipc_named_node_up(struct net *net, u32 dnode, u16 capabilities)
spin_lock_bh(&tn->nametbl_lock);
if (!(capabilities & TIPC_NAMED_BCAST))
nt->rc_dests++;
- named_distribute(net, &head, dnode, &nt->cluster_scope, nt->snd_nxt);
- if (!skb_queue_empty(&head))
+ rc = __tipc_named_distribute(net, dnode, &head);
+ if (!rc && !skb_queue_empty(&head))
rc = tipc_node_distribute(net, &head, dnode, &bearer_id, 0);
spin_unlock_bh(&tn->nametbl_lock);
@@ -289,10 +383,12 @@ void tipc_publ_notify(struct net *net, struct list_head *nsub_list,
static bool tipc_update_nametbl(struct net *net, struct distr_item *i,
u32 node, u32 dtype)
{
+ struct tipc_net *tn = tipc_net(net);
struct publication *p = NULL;
u32 lower = ntohl(i->lower);
u32 upper = ntohl(i->upper);
struct tipc_socket_addr sk;
+ u32 type = ntohl(i->type);
u32 key = ntohl(i->key);
struct tipc_uaddr ua;
@@ -303,13 +399,22 @@ static bool tipc_update_nametbl(struct net *net, struct distr_item *i,
if (lower > upper)
return false;
+ /* Not insert or remove duplicate node-up event publication sent from
+ * remote nodes
+ */
+ if (unlikely(type == TIPC_NODE_STATE && tn->node_addr &&
+ lower == tn->node_addr && upper == tn->node_addr)) {
+ pr_warn_ratelimited("Can't update duplicate node-up pub\n");
+ return false;
+ }
+
tipc_uaddr(&ua, TIPC_SERVICE_RANGE, TIPC_CLUSTER_SCOPE,
- ntohl(i->type), lower, upper);
+ type, lower, upper);
sk.ref = ntohl(i->port);
sk.node = node;
if (dtype == PUBLICATION) {
- p = tipc_nametbl_insert_publ(net, &ua, &sk, key);
+ p = tipc_nametbl_insert_publ(net, &ua, &sk, key, NULL);
if (p) {
tipc_node_subscribe(net, &p->binding_node, node);
return true;
@@ -426,5 +531,6 @@ void tipc_named_reinit(struct net *net)
list_for_each_entry_rcu(p, &nt->cluster_scope, binding_node)
p->sk.node = self;
nt->rc_dests = 0;
+ tn->nt_stop = false;
spin_unlock_bh(&tn->nametbl_lock);
}
diff --git a/net/tipc/name_distr.h b/net/tipc/name_distr.h
index 3f41e36efee3..2043096ae967 100644
--- a/net/tipc/name_distr.h
+++ b/net/tipc/name_distr.h
@@ -75,5 +75,6 @@ void tipc_named_rcv(struct net *net, struct sk_buff_head *namedq,
void tipc_named_reinit(struct net *net);
void tipc_publ_notify(struct net *net, struct list_head *nsub_list,
u32 addr, u16 capabilities);
+int tipc_named_distribute(struct net *net, u32 dnode);
#endif
diff --git a/net/tipc/name_table.c b/net/tipc/name_table.c
index 8924bdb6a443..3bcb8762bc9c 100644
--- a/net/tipc/name_table.c
+++ b/net/tipc/name_table.c
@@ -329,7 +329,7 @@ static struct service_range *tipc_service_create_range(struct tipc_service *sc,
static bool tipc_service_insert_publ(struct net *net,
struct tipc_service *sc,
- struct publication *p)
+ struct publication *p, int *err)
{
struct tipc_subscription *sub, *tmp;
struct service_range *sr;
@@ -339,10 +339,12 @@ static bool tipc_service_insert_publ(struct net *net,
bool res = false;
u32 key = p->key;
- spin_lock_bh(&sc->lock);
sr = tipc_service_create_range(sc, p);
- if (!sr)
+ if (!sr) {
+ if (err)
+ *err = -ENOBUFS;
goto exit;
+ }
first = list_empty(&sr->all_publ);
@@ -353,6 +355,8 @@ static bool tipc_service_insert_publ(struct net *net,
pr_debug("Failed to bind duplicate %u,%u,%u/%u:%u/%u\n",
p->sr.type, p->sr.lower, p->sr.upper,
node, p->sk.ref, key);
+ if (err)
+ *err = -EEXIST;
goto exit;
}
}
@@ -371,7 +375,6 @@ static bool tipc_service_insert_publ(struct net *net,
if (!res)
pr_warn("Failed to bind to %u,%u,%u\n",
p->sr.type, p->sr.lower, p->sr.upper);
- spin_unlock_bh(&sc->lock);
return res;
}
@@ -478,20 +481,33 @@ static struct tipc_service *tipc_service_find(struct net *net,
struct publication *tipc_nametbl_insert_publ(struct net *net,
struct tipc_uaddr *ua,
struct tipc_socket_addr *sk,
- u32 key)
+ u32 key, int *err)
{
struct tipc_service *sc;
struct publication *p;
p = tipc_publ_create(ua, sk, key);
- if (!p)
+ if (!p) {
+ if (err)
+ *err = -ENOBUFS;
return NULL;
+ }
sc = tipc_service_find(net, ua);
if (!sc)
sc = tipc_service_create(net, ua);
- if (sc && tipc_service_insert_publ(net, sc, p))
- return p;
+
+ if (sc) {
+ spin_lock_bh(&sc->lock);
+ if (tipc_service_insert_publ(net, sc, p, err)) {
+ spin_unlock_bh(&sc->lock);
+ return p;
+ }
+ spin_unlock_bh(&sc->lock);
+ } else if (err) {
+ *err = -ENOBUFS;
+ }
+
kfree(p);
return NULL;
}
@@ -760,7 +776,8 @@ void tipc_nametbl_build_group(struct net *net, struct tipc_group *grp,
/* tipc_nametbl_publish - add service binding to name table
*/
struct publication *tipc_nametbl_publish(struct net *net, struct tipc_uaddr *ua,
- struct tipc_socket_addr *sk, u32 key)
+ struct tipc_socket_addr *sk,
+ u32 key, int *err)
{
struct name_table *nt = tipc_name_table(net);
u32 max_user_pub = TIPC_MAX_PUBL - 1;
@@ -768,6 +785,7 @@ struct publication *tipc_nametbl_publish(struct net *net, struct tipc_uaddr *ua,
struct publication *p = NULL;
struct sk_buff *skb = NULL;
bool protocol_type = false;
+ int error = 0;
u32 rc_dests;
if (ua->sr.type == TIPC_NODE_STATE || ua->sr.type == TIPC_LINK_STATE ||
@@ -788,7 +806,7 @@ struct publication *tipc_nametbl_publish(struct net *net, struct tipc_uaddr *ua,
}
insert:
- p = tipc_nametbl_insert_publ(net, ua, sk, key);
+ p = tipc_nametbl_insert_publ(net, ua, sk, key, &error);
if (p) {
/* Not count node state, link state and topology server types
* so that maximum nt->local_publ_count does not prevent
@@ -798,6 +816,13 @@ struct publication *tipc_nametbl_publish(struct net *net, struct tipc_uaddr *ua,
if (!protocol_type)
nt->local_publ_count++;
skb = tipc_named_publish(net, p);
+ /* Local-scope publication is not published (skb is NULL), only
+ * cluster-cope one is.
+ */
+ if (!skb && p->scope == TIPC_CLUSTER_SCOPE && err)
+ *err = -ENOBUFS;
+ } else if ((error == -ENOBUFS) && err) {
+ *err = -ENOBUFS;
}
rc_dests = nt->rc_dests;
exit:
@@ -809,6 +834,83 @@ struct publication *tipc_nametbl_publish(struct net *net, struct tipc_uaddr *ua,
}
+/**
+ * tipc_nametb_insert_self_node_pub - insert publication of node's self address
+ * @net: network namespace
+ *
+ * Return:
+ * * 0 - Success
+ * * -ENOBUFS - No buffer space is available
+ *
+ */
+int tipc_nametb_insert_self_node_pub(struct net *net)
+{
+ struct name_table *nt = tipc_name_table(net);
+ struct tipc_net *tn = tipc_net(net);
+ struct tipc_socket_addr sk;
+ struct service_range *sr;
+ struct tipc_service *sc;
+ bool sc_created = false;
+ struct publication *p;
+ struct tipc_uaddr ua;
+ int err = 0;
+
+ sk.ref = 0;
+ sk.node = tn->node_addr;
+ tipc_uaddr(&ua, TIPC_SERVICE_RANGE, TIPC_CLUSTER_SCOPE,
+ TIPC_NODE_STATE, tn->node_addr, tn->node_addr);
+ sc = tipc_service_find(net, &ua);
+ if (!sc) {
+ sc = tipc_service_create(net, &ua);
+ if (!sc)
+ return -ENOBUFS;
+ sc_created = true;
+ }
+
+ spin_lock_bh(&sc->lock);
+ sr = tipc_service_find_range(sc, &ua);
+ /* 1. tipc_net_finalize_work() is not scheduled because of namespace
+ * teardown.
+ * 2. Or tipc_net_finalize() ---> tipc_nametbl_publish() has failed
+ * to insert node self address publication into nt->cluster_scope
+ * due to memory allocation failure.
+ * 3. Or tipc_net_finalize() ---> tipc_nametbl_publish() has not
+ * executed yet.
+ */
+ if (!sr) {
+ p = tipc_publ_create(&ua, &sk, tn->node_addr);
+ if (!p)
+ goto error;
+
+ if (!tipc_service_insert_publ(net, sc, p, &err)) {
+ kfree(p);
+ goto error;
+ }
+
+ list_add_tail(&p->binding_node, &nt->cluster_scope);
+ goto exit;
+error:
+ if (sc_created) {
+ hlist_del_init_rcu(&sc->service_list);
+ spin_unlock_bh(&sc->lock);
+ kfree_rcu(sc, rcu);
+ if (unlikely(err == -EEXIST))
+ return 0;
+ else
+ return -ENOBUFS;
+ }
+
+ spin_unlock_bh(&sc->lock);
+ if (unlikely(err == -EEXIST))
+ return 0;
+ else
+ return -ENOBUFS;
+ }
+exit:
+ spin_unlock_bh(&sc->lock);
+ return 0;
+}
+
/**
* tipc_nametbl_withdraw - withdraw a service binding
* @net: network namespace
@@ -933,6 +1035,7 @@ int tipc_nametbl_init(struct net *net)
*/
static void tipc_service_delete(struct net *net, struct tipc_service *sc)
{
+ struct tipc_net *tn = tipc_net(net);
struct service_range *sr, *tmpr;
struct publication *p, *tmp;
@@ -940,6 +1043,15 @@ static void tipc_service_delete(struct net *net, struct tipc_service *sc)
rbtree_postorder_for_each_entry_safe(sr, tmpr, &sc->ranges, tree_node) {
list_for_each_entry_safe(p, tmp, &sr->all_publ, all_publ) {
tipc_service_remove_publ(sr, &p->sk, p->key);
+ /* tipc_nametbl_withdraw() does not delete
+ * 'p->binding_node' with type TIPC_NODE_STATE and
+ * range [tn->node_addr, tn->node_addr] from
+ * 'tn->cluster_scope'. So, delete it here.
+ */
+ if (p->sr.type == TIPC_NODE_STATE &&
+ p->sr.lower == tn->node_addr &&
+ p->sr.upper == tn->node_addr)
+ list_del(&p->binding_node);
kfree_rcu(p, rcu);
}
rb_erase_augmented(&sr->tree_node, &sc->ranges, &sr_callbacks);
@@ -970,6 +1082,8 @@ void tipc_nametbl_stop(struct net *net)
tipc_service_delete(net, service);
}
}
+ tn->nt_stop = true;
+ tn->nt_self_node_exist = false;
spin_unlock_bh(&tn->nametbl_lock);
/* TODO: clear tn->nametbl, implement proper RCU rules ? */
diff --git a/net/tipc/name_table.h b/net/tipc/name_table.h
index afe665e68bb1..4b411b6e9e40 100644
--- a/net/tipc/name_table.h
+++ b/net/tipc/name_table.h
@@ -124,13 +124,15 @@ bool tipc_nametbl_lookup_group(struct net *net, struct tipc_uaddr *ua,
void tipc_nametbl_build_group(struct net *net, struct tipc_group *grp,
struct tipc_uaddr *ua);
struct publication *tipc_nametbl_publish(struct net *net, struct tipc_uaddr *ua,
- struct tipc_socket_addr *sk, u32 key);
+ struct tipc_socket_addr *sk,
+ u32 key, int *err);
void tipc_nametbl_withdraw(struct net *net, struct tipc_uaddr *ua,
struct tipc_socket_addr *sk, u32 key);
struct publication *tipc_nametbl_insert_publ(struct net *net,
struct tipc_uaddr *ua,
struct tipc_socket_addr *sk,
- u32 key);
+ u32 key, int *err);
+int tipc_nametb_insert_self_node_pub(struct net *net);
struct publication *tipc_nametbl_remove_publ(struct net *net,
struct tipc_uaddr *ua,
struct tipc_socket_addr *sk,
diff --git a/net/tipc/net.c b/net/tipc/net.c
index 7e65d0b0c4a8..6e8bf0c5ebde 100644
--- a/net/tipc/net.c
+++ b/net/tipc/net.c
@@ -110,11 +110,14 @@ static void tipc_net_finalize(struct net *net, u32 addr);
int tipc_net_init(struct net *net, u8 *node_id, u32 addr)
{
+ struct tipc_net *tn = tipc_net(net);
+
if (tipc_own_id(net)) {
pr_info("Cannot configure node identity twice\n");
return -1;
}
pr_info("Started in network mode\n");
+ atomic_set(&tn->node_addr_set, 0);
if (node_id)
tipc_set_node_id(net, node_id);
@@ -138,7 +141,11 @@ static void tipc_net_finalize(struct net *net, u32 addr)
tipc_named_reinit(net);
tipc_sk_reinit(net);
tipc_mon_reinit_self(net);
- tipc_nametbl_publish(net, &ua, &sk, addr);
+ atomic_set(&tn->node_addr_set, 1);
+ /* Redistribute the node state publication via node-up event
+ * or node's timer if tipc_nametbl_publish() fails.
+ */
+ tipc_nametbl_publish(net, &ua, &sk, addr, NULL);
}
void tipc_net_finalize_work(struct work_struct *work)
diff --git a/net/tipc/node.c b/net/tipc/node.c
index b19d6652dc73..f566472a411f 100644
--- a/net/tipc/node.c
+++ b/net/tipc/node.c
@@ -111,6 +111,7 @@ struct tipc_bclink_entry {
* @peer_net: peer's net namespace
* @peer_hash_mix: hash for this peer (FIXME)
* @crypto_rx: RX crypto handler
+ * @dist_bulk: flag to redistribute cluster-scope publications
*/
struct tipc_node {
u32 addr;
@@ -145,6 +146,7 @@ struct tipc_node {
#ifdef CONFIG_TIPC_CRYPTO
struct tipc_crypto *crypto_rx;
#endif
+ bool dist_bulk;
};
/* Node FSM states and events:
@@ -393,6 +395,14 @@ static void tipc_node_write_unlock_fast(struct tipc_node *n)
write_unlock_bh(&n->lock);
}
+static void tipc_node_down(struct tipc_node *n)
+{
+ int bearer_id;
+
+ for (bearer_id = 0; bearer_id < MAX_BEARERS; bearer_id++)
+ tipc_node_link_down(n, bearer_id, false);
+}
+
static void tipc_node_write_unlock(struct tipc_node *n)
__releases(n->lock)
{
@@ -402,7 +412,7 @@ static void tipc_node_write_unlock(struct tipc_node *n)
struct list_head *publ_list;
struct tipc_uaddr ua;
u32 bearer_id, node;
- int rc = 0;
+ int rc = 0, err = 0;
if (likely(!flags)) {
write_unlock_bh(&n->lock);
@@ -420,6 +430,8 @@ static void tipc_node_write_unlock(struct tipc_node *n)
n->action_flags &= ~(TIPC_NOTIFY_NODE_DOWN | TIPC_NOTIFY_NODE_UP |
TIPC_NOTIFY_LINK_DOWN | TIPC_NOTIFY_LINK_UP);
+ if (flags & TIPC_NOTIFY_NODE_DOWN)
+ n->dist_bulk = false;
write_unlock_bh(&n->lock);
if (flags & TIPC_NOTIFY_NODE_DOWN)
@@ -430,17 +442,35 @@ static void tipc_node_write_unlock(struct tipc_node *n)
if (flags & TIPC_NOTIFY_LINK_UP) {
tipc_mon_peer_up(net, node, bearer_id);
- tipc_nametbl_publish(net, &ua, &sk, sk.ref);
+ tipc_nametbl_publish(net, &ua, &sk, sk.ref, &err);
}
if (flags & TIPC_NOTIFY_LINK_DOWN) {
tipc_mon_peer_down(net, node, bearer_id);
tipc_nametbl_withdraw(net, &ua, &sk, sk.ref);
}
+ /* Memory allocation has failed. Bring the node down to start over bulk
+ * distribution when the first link is up again.
+ */
if (rc == -ENOBUFS)
+ tipc_node_down(n);
+ else if (err == -ENOBUFS)
tipc_node_link_down(n, bearer_id, false);
}
+void tipc_node_set_dist_bulk(struct net *net, u32 addr)
+{
+ struct tipc_node *node;
+
+ node = tipc_node_find(net, addr);
+ if (node) {
+ tipc_node_write_lock(node);
+ node->dist_bulk = true;
+ tipc_node_write_unlock_fast(node);
+ tipc_node_put(node);
+ }
+}
+
static void tipc_node_assign_peer_net(struct tipc_node *n, u32 hash_mixes)
{
int net_id = tipc_netid(n->net);
@@ -568,6 +598,7 @@ struct tipc_node *tipc_node_create(struct net *net, u32 addr, u8 *peer_id,
INIT_LIST_HEAD(&n->list);
INIT_LIST_HEAD(&n->publ_list);
INIT_LIST_HEAD(&n->conn_sks);
+ n->dist_bulk = false;
skb_queue_head_init(&n->bc_entry.namedq);
skb_queue_head_init(&n->bc_entry.inputq1);
__skb_queue_head_init(&n->bc_entry.arrvq);
@@ -807,9 +838,11 @@ static void tipc_node_timeout(struct timer_list *t)
struct tipc_node *n = timer_container_of(n, t, timer);
struct tipc_link_entry *le;
struct sk_buff_head xmitq;
- int remains = n->link_cnt;
+ bool dist_bulk = false;
+ int remains = 0;
+ int dist_rc = 0;
int bearer_id;
- int rc = 0;
+ int rc;
trace_tipc_node_timeout(n, false, " ");
if (!node_is_up(n) && tipc_node_cleanup(n)) {
@@ -824,13 +857,23 @@ static void tipc_node_timeout(struct timer_list *t)
#endif
__skb_queue_head_init(&xmitq);
+ tipc_node_write_lock(n);
/* Initial node interval to value larger (10 seconds), then it will be
* recalculated with link lowest tolerance
*/
- tipc_node_read_lock(n);
n->keepalive_intv = 10000;
- tipc_node_read_unlock(n);
+
+ remains = n->link_cnt;
+ if (n->dist_bulk) {
+ dist_bulk = true;
+ n->dist_bulk = false;
+ }
+ tipc_node_write_unlock_fast(n);
+ if (dist_bulk)
+ dist_rc = tipc_named_distribute(n->net, n->addr);
+
for (bearer_id = 0; remains && (bearer_id < MAX_BEARERS); bearer_id++) {
+ rc = 0;
tipc_node_read_lock(n);
le = &n->links[bearer_id];
if (le->link) {
@@ -839,11 +882,13 @@ static void tipc_node_timeout(struct timer_list *t)
tipc_node_calculate_timer(n, le->link);
rc = tipc_link_timeout(le->link, &xmitq);
spin_unlock_bh(&le->lock);
- remains--;
+ if (dist_rc != -ENOBUFS)
+ remains--;
}
tipc_node_read_unlock(n);
tipc_bearer_xmit(n->net, bearer_id, &xmitq, &le->maddr, n);
- if (rc & TIPC_LINK_DOWN_EVT)
+ /* Force the node down in case the redistribution failed */
+ if ((rc & TIPC_LINK_DOWN_EVT) || dist_rc == -ENOBUFS)
tipc_node_link_down(n, bearer_id, false);
}
mod_timer(&n->timer, jiffies + msecs_to_jiffies(n->keepalive_intv));
diff --git a/net/tipc/node.h b/net/tipc/node.h
index 775a8ec240fb..b312cefd4d41 100644
--- a/net/tipc/node.h
+++ b/net/tipc/node.h
@@ -130,4 +130,5 @@ int tipc_nl_node_set_key(struct sk_buff *skb, struct genl_info *info);
int tipc_nl_node_flush_key(struct sk_buff *skb, struct genl_info *info);
#endif
void tipc_node_pre_cleanup_net(struct net *exit_net);
+void tipc_node_set_dist_bulk(struct net *net, u32 addr);
#endif
diff --git a/net/tipc/socket.c b/net/tipc/socket.c
index 65cc0c4964b3..4f204bdcaf21 100644
--- a/net/tipc/socket.c
+++ b/net/tipc/socket.c
@@ -2929,7 +2929,7 @@ static int tipc_sk_publish(struct tipc_sock *tsk, struct tipc_uaddr *ua)
return -EADDRINUSE;
skaddr.ref = tsk->portid;
skaddr.node = tipc_own_addr(net);
- p = tipc_nametbl_publish(net, ua, &skaddr, key);
+ p = tipc_nametbl_publish(net, ua, &skaddr, key, NULL);
if (unlikely(!p))
return -EINVAL;
--
2.53.0
^ permalink raw reply related [flat|nested] 6+ messages in thread