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 --- 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