RE: [PATCH net v8 1/3] tipc: fix NULL deref in tipc_named_node_up() on empty publication list

@@ -61,6 +61,8 @@ static int __net_init tipc_init_net(struct net *net)
 	tn->trial_addr = 0;
 	tn->addr_trial_end = 0;
 	tn->capabilities = TIPC_NODE_CAPABILITIES;
+	atomic_set(&tn->node_wq_count, 0);
+	atomic_set(&tn->finalized, 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));

@@ -119,6 +121,7 @@ static void __net_exit tipc_exit_net(struct net *net)
 #ifdef CONFIG_TIPC_CRYPTO
 	tipc_crypto_stop(&tipc_net(net)->crypto_tx);
 #endif
+	wait_var_event(&tn->node_wq_count, !atomic_read(&tn->node_wq_count));
 	wait_var_event(&tn->wq_count, atomic_read(&tn->wq_count) == 0);
 }
 

@@ -145,6 +145,10 @@ struct tipc_net {
 	struct work_struct work;
 	/* The numbers of work queues in schedule */
 	atomic_t wq_count;
+	/* The numbers of works of created nodes in schedule */
+	atomic_t node_wq_count;
+	/* flag to wake up scheduled works of created nodes */
+	atomic_t finalized;
 };
 
 static inline struct tipc_net *tipc_net(struct net *net)

@@ -147,8 +147,8 @@ struct sk_buff *tipc_named_withdraw(struct net *net, struct publication *p)
  * @pls: linked list of publication items to be packed into buffer chain
  * @seqno: sequence number for this message
  */
-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;

@@ -164,8 +164,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 1;
 			}
 			hdr = buf_msg(skb);
 			msg_set_bc_ack_invalid(hdr, true);

@@ -195,15 +196,16 @@ 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_node_up - tell specified node about all publications by this node
+ * tipc_named_distribute - distribute all publications to specified node
  * @net: the associated network namespace
  * @dnode: destination node
- * @capabilities: peer node's capabilities
  */
-void tipc_named_node_up(struct net *net, u32 dnode, u16 capabilities)
+static int tipc_named_distribute(struct net *net, u32 dnode)
 {
 	struct name_table *nt = tipc_name_table(net);
 	struct tipc_net *tn = tipc_net(net);

@@ -212,15 +214,86 @@ void tipc_named_node_up(struct net *net, u32 dnode, u16 capabilities)
 
 	__skb_queue_head_init(&head);
 	spin_lock_bh(&tn->nametbl_lock);
-	if (!(capabilities & TIPC_NAMED_BCAST))
-		nt->rc_dests++;
 	seqno = nt->snd_nxt;
 	spin_unlock_bh(&tn->nametbl_lock);
 
 	read_lock_bh(&nt->cluster_scope_lock);
-	named_distribute(net, &head, dnode, &nt->cluster_scope, seqno);
+	if (named_distribute(net, &head, dnode, &nt->cluster_scope, seqno)) {
+		read_unlock_bh(&nt->cluster_scope_lock);
+		return -ENOBUFS;
+	}
 	tipc_node_xmit(net, &head, dnode, 0);
 	read_unlock_bh(&nt->cluster_scope_lock);
+
+	return 0;
+}
+
+/**
+ * tipc_named_node_up - tell specified node about all publications by this node
+ * @net: the associated network namespace
+ * @dnode: destination node
+ * @capabilities: peer node's 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);
+
+	spin_lock_bh(&tn->nametbl_lock);
+	if (!(capabilities & TIPC_NAMED_BCAST))
+		nt->rc_dests++;
+	spin_unlock_bh(&tn->nametbl_lock);
+
+	read_lock_bh(&nt->cluster_scope_lock);
+	/* 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 to nt->cluster_scope
+	 *    due to memory allocation failure.
+	 * 3. Or tipc_net_finalize() ---> tipc_nametbl_publish() has not
+	 *    executed yet.
+	 */
+	if (unlikely(list_empty(&nt->cluster_scope))) {
+		read_unlock_bh(&nt->cluster_scope_lock);
+		return 1;
+	}
+	read_unlock_bh(&nt->cluster_scope_lock);
+
+	return tipc_named_distribute(net, dnode);
+}
+
+/**
+ * tipc_named_dist_cluster_scope - distribute all publications to specified node
+ * @net: the associated network namespace
+ * @dnode: destination node
+ */
+int tipc_named_dist_cluster_scope(struct net *net, u32 dnode)
+{
+	struct name_table *nt = tipc_name_table(net);
+	bool reinsert = false;
+
+	read_lock_bh(&nt->cluster_scope_lock);
+	if (unlikely(list_empty(&nt->cluster_scope)))
+		reinsert = true;
+	read_unlock_bh(&nt->cluster_scope_lock);
+	/* tipc_net_finalize() ---> tipc_nametbl_publish() has failed to insert
+	 * node self address publication to nt->cluster_scope due to memory
+	 * allocation failure. So, reinsert this publication.
+	 */
+	if (reinsert) {
+		struct tipc_net *tn = tipc_net(net);
+		struct tipc_socket_addr sk;
+		struct tipc_uaddr ua;
+
+		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);
+		if (!tipc_nametbl_publish(net, &ua, &sk, tn->node_addr))
+			return -ENOBUFS;
+	}
+
+	return tipc_named_distribute(net, dnode);
 }
 
 /**

@@ -69,7 +69,8 @@ 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);
+int tipc_named_dist_cluster_scope(struct net *net, u32 dnode);
 void tipc_named_rcv(struct net *net, struct sk_buff_head *namedq,
 		    u16 *rcv_nxt, bool *open);
 void tipc_named_reinit(struct net *net);

@@ -763,6 +763,7 @@ struct publication *tipc_nametbl_publish(struct net *net, struct tipc_uaddr *ua,
 					 struct tipc_socket_addr *sk, u32 key)
 {
 	struct name_table *nt = tipc_name_table(net);
+	u32 max_user_pub = TIPC_MAX_PUBL - 1;
 	struct tipc_net *tn = tipc_net(net);
 	struct publication *p = NULL;
 	struct sk_buff *skb = NULL;

@@ -770,14 +771,26 @@ struct publication *tipc_nametbl_publish(struct net *net, struct tipc_uaddr *ua,
 
 	spin_lock_bh(&tn->nametbl_lock);
 
-	if (nt->local_publ_count >= TIPC_MAX_PUBL) {
-		pr_warn("Bind failed, max limit %u reached\n", TIPC_MAX_PUBL);
+	/* Reserve one entry for node state service type because it has cluster
+	 * scope and it is distributed in bulk. So, the maximum number of user's
+	 * publications is (TIPC_MAX_PUBL - 1).
+	 */
+	if (nt->local_publ_count >= max_user_pub) {
+		pr_warn("Bind failed, max limit %u reached\n", max_user_pub);
 		goto exit;
 	}
 
 	p = tipc_nametbl_insert_publ(net, ua, sk, key);
 	if (p) {
-		nt->local_publ_count++;
+		/* Not count node state, link state and topology server types
+		 * so that maximum nt->local_publ_count does not prevent
+		 * protocol service types from being inserted to the name
+		 * table.
+		 */
+		if (ua->sr.type != TIPC_NODE_STATE &&
+		    ua->sr.type != TIPC_LINK_STATE &&
+		    ua->sr.type != TIPC_TOP_SRV)
+			nt->local_publ_count++;
 		skb = tipc_named_publish(net, p);
 	}
 	rc_dests = nt->rc_dests;

@@ -810,7 +823,10 @@ void tipc_nametbl_withdraw(struct net *net, struct tipc_uaddr *ua,
 
 	p = tipc_nametbl_remove_publ(net, ua, sk, key);
 	if (p) {
-		nt->local_publ_count--;
+		if (p->sr.type != TIPC_NODE_STATE &&
+		    p->sr.type != TIPC_LINK_STATE &&
+		    p->sr.type != TIPC_TOP_SRV)
+			nt->local_publ_count--;
 		skb = tipc_named_withdraw(net, p);
 		list_del_init(&p->binding_sock);
 		kfree_rcu(p, rcu);

@@ -139,6 +139,11 @@ static void tipc_net_finalize(struct net *net, u32 addr)
 	tipc_sk_reinit(net);
 	tipc_mon_reinit_self(net);
 	tipc_nametbl_publish(net, &ua, &sk, addr);
+	atomic_set(&tn->finalized, 1);
+	/* Wake up nodes waiting for distributing bulk of publications
+	 * if applicable.
+	 */
+	tipc_node_wakeup(net);
 }
 
 void tipc_net_finalize_work(struct work_struct *work)

@@ -111,6 +111,9 @@ struct tipc_bclink_entry {
  * @peer_net: peer's net namespace
  * @peer_hash_mix: hash for this peer (FIXME)
  * @crypto_rx: RX crypto handler
+ * @work: work item for bulk distribution of cluster scope publications
+ * @work_scheduled: flag to indicate the work has been scheduled
+ * @finalized: flag to wake up scheduled work
  */
 struct tipc_node {
 	u32 addr;

@@ -145,6 +148,9 @@ struct tipc_node {
 #ifdef CONFIG_TIPC_CRYPTO
 	struct tipc_crypto *crypto_rx;
 #endif
+	struct work_struct work;
+	atomic_t work_scheduled;
+	atomic_t finalized;
 };
 
 /* Node FSM states and events:

@@ -393,6 +399,62 @@ 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)
+{
+	u32 bearer_id, bearer_cnt;
+
+	tipc_node_read_lock(n);
+	bearer_cnt = n->link_cnt;
+	tipc_node_read_unlock(n);
+	for (bearer_id = 0; bearer_id < bearer_cnt; bearer_id++)
+		tipc_node_link_down(n, bearer_id, false);
+}
+
+static void tipc_node_dist_bulk(struct work_struct *work)
+{
+	struct tipc_node *node = container_of(work, struct tipc_node, work);
+	struct tipc_net *tn = tipc_net(node->net);
+
+	wait_var_event(&node->finalized, atomic_read(&node->finalized));
+
+	/* node is down */
+	if (!atomic_read(&node->work_scheduled))
+		goto exit;
+
+	/* Stop waking up work in tipc_node_timeout() */
+	atomic_set(&node->work_scheduled, 0);
+
+	if (tipc_named_dist_cluster_scope(node->net, node->addr) < 0)
+		tipc_node_down(node);
+
+exit:
+	if (atomic_dec_and_test(&tn->node_wq_count))
+		wake_up_var(&tn->node_wq_count);
+
+	tipc_node_put(node);
+}
+
+static void tipc_node_cancel_work(struct tipc_node *node)
+{
+	if (atomic_read(&node->work_scheduled)) {
+		struct tipc_net *tn = tipc_net(node->net);
+
+		/* Wake up scheduled work if it is waiting. */
+		if (!atomic_read(&node->finalized)) {
+			atomic_set(&node->work_scheduled, 0);
+			atomic_set(&node->finalized, 1);
+			wake_up_var(&node->finalized);
+		}
+
+		/* Work is pending */
+		if (cancel_work(&node->work)) {
+			if (atomic_dec_and_test(&tn->node_wq_count))
+				wake_up_var(&tn->node_wq_count);
+			tipc_node_put(node);
+		}
+	}
+}
+
 static void tipc_node_write_unlock(struct tipc_node *n)
 	__releases(n->lock)
 {

@@ -421,11 +483,30 @@ static void tipc_node_write_unlock(struct tipc_node *n)
 
 	write_unlock_bh(&n->lock);
 
-	if (flags & TIPC_NOTIFY_NODE_DOWN)
+	if (flags & TIPC_NOTIFY_NODE_DOWN) {
 		tipc_publ_notify(net, publ_list, node, n->capabilities);
-
-	if (flags & TIPC_NOTIFY_NODE_UP)
-		tipc_named_node_up(net, node, n->capabilities);
+		tipc_node_cancel_work(n);
+	}
+
+	if (flags & TIPC_NOTIFY_NODE_UP) {
+		struct tipc_net *tn = tipc_net(net);
+		int rc = 0;
+
+		rc = tipc_named_node_up(net, node, n->capabilities);
+		/* Defer bulk distribution to work queue */
+		if (rc > 0) {
+			atomic_set(&n->work_scheduled, 1);
+			atomic_inc(&tn->node_wq_count);
+			tipc_node_get(n);
+			if (!schedule_work(&n->work))
+				tipc_node_put(n);
+		} else if (rc < 0) {
+			/* Bring the node down to start over bulk distribution
+			 * when the first link is up again.
+			 */
+			tipc_node_down(n);
+		}
+	}
 
 	if (flags & TIPC_NOTIFY_LINK_UP) {
 		tipc_mon_peer_up(net, node, bearer_id);

@@ -564,6 +645,9 @@ 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);
+	INIT_WORK(&n->work, tipc_node_dist_bulk);
+	atomic_set(&n->work_scheduled, 0);
+	atomic_set(&n->finalized, 0);
 	skb_queue_head_init(&n->bc_entry.namedq);
 	skb_queue_head_init(&n->bc_entry.inputq1);
 	__skb_queue_head_init(&n->bc_entry.arrvq);

@@ -653,6 +737,21 @@ void tipc_node_stop(struct net *net)
 	spin_unlock_bh(&tn->node_list_lock);
 }
 
+void tipc_node_wakeup(struct net *net)
+{
+	struct tipc_net *tn = tipc_net(net);
+	struct tipc_node *node;
+
+	rcu_read_lock();
+	list_for_each_entry_rcu(node, &tn->node_list, list) {
+		if (atomic_read(&node->work_scheduled)) {
+			atomic_set(&node->finalized, 1);
+			wake_up_var(&node->finalized);
+		}
+	}
+	rcu_read_unlock();
+}
+
 void tipc_node_subscribe(struct net *net, struct list_head *subscr, u32 addr)
 {
 	struct tipc_node *n;

@@ -801,6 +900,7 @@ static bool tipc_node_cleanup(struct tipc_node *peer)
 static void tipc_node_timeout(struct timer_list *t)
 {
 	struct tipc_node *n = timer_container_of(n, t, timer);
+	struct tipc_net *tn = tipc_net(n->net);
 	struct tipc_link_entry *le;
 	struct sk_buff_head xmitq;
 	int remains = n->link_cnt;

@@ -814,6 +914,11 @@ static void tipc_node_timeout(struct timer_list *t)
 		return;
 	}
 
+	if (atomic_read(&tn->finalized) && atomic_read(&n->work_scheduled)) {
+		atomic_set(&n->finalized, 1);
+		wake_up_var(&n->finalized);
+	}
+
 #ifdef CONFIG_TIPC_CRYPTO
 	/* Take any crypto key related actions first */
 	tipc_crypto_timeout(n->crypto_rx);

@@ -75,6 +75,7 @@ enum {
 #define INVALID_BEARER_ID -1
 
 void tipc_node_stop(struct net *net);
+void tipc_node_wakeup(struct net *net);
 bool tipc_node_get_id(struct net *net, u32 addr, u8 *id);
 u32 tipc_node_get_addr(struct tipc_node *node);
 char *tipc_node_get_id_str(struct tipc_node *node);