tipc: fix broadcast wakeup contention after congestion

[karo-tx-linux.git] / net / tipc / node.c

diff --git a/net/tipc/node.c b/net/tipc/node.c
index 5781634e957d7be47c2e572783f3af2a3db9bf91..8d353ec77a6661820066e4cbee0d74cfe534ea08 100644 (file)
--- a/net/tipc/node.c
+++ b/net/tipc/node.c
@@ -58,6 +58,12 @@ struct tipc_sock_conn {
        struct list_head list;
 };
 
+static const struct nla_policy tipc_nl_node_policy[TIPC_NLA_NODE_MAX + 1] = {
+       [TIPC_NLA_NODE_UNSPEC]          = { .type = NLA_UNSPEC },
+       [TIPC_NLA_NODE_ADDR]            = { .type = NLA_U32 },
+       [TIPC_NLA_NODE_UP]              = { .type = NLA_FLAG }
+};
+
 /*
  * A trivial power-of-two bitmask technique is used for speed, since this
  * operation is done for every incoming TIPC packet. The number of hash table
@@ -107,9 +113,10 @@ struct tipc_node *tipc_node_create(u32 addr)
        spin_lock_init(&n_ptr->lock);
        INIT_HLIST_NODE(&n_ptr->hash);
        INIT_LIST_HEAD(&n_ptr->list);
-       INIT_LIST_HEAD(&n_ptr->nsub);
+       INIT_LIST_HEAD(&n_ptr->publ_list);
        INIT_LIST_HEAD(&n_ptr->conn_sks);
-       __skb_queue_head_init(&n_ptr->waiting_sks);
+       skb_queue_head_init(&n_ptr->waiting_sks);
+       __skb_queue_head_init(&n_ptr->bclink.deferred_queue);
 
        hlist_add_head_rcu(&n_ptr->hash, &node_htable[tipc_hashfn(addr)]);
 
@@ -375,8 +382,7 @@ static void node_lost_contact(struct tipc_node *n_ptr)
 
        /* Flush broadcast link info associated with lost node */
        if (n_ptr->bclink.recv_permitted) {
-               kfree_skb_list(n_ptr->bclink.deferred_head);
-               n_ptr->bclink.deferred_size = 0;
+               __skb_queue_purge(&n_ptr->bclink.deferred_queue);
 
                if (n_ptr->bclink.reasm_buf) {
                        kfree_skb(n_ptr->bclink.reasm_buf);
@@ -568,7 +574,7 @@ void tipc_node_unlock(struct tipc_node *node)
                skb_queue_splice_init(&node->waiting_sks, &waiting_sks);
 
        if (flags & TIPC_NOTIFY_NODE_DOWN) {
-               list_replace_init(&node->nsub, &nsub_list);
+               list_replace_init(&node->publ_list, &nsub_list);
                list_replace_init(&node->conn_sks, &conn_sks);
        }
        node->action_flags &= ~(TIPC_WAKEUP_USERS | TIPC_NOTIFY_NODE_DOWN |
@@ -585,7 +591,7 @@ void tipc_node_unlock(struct tipc_node *node)
                tipc_node_abort_sock_conns(&conn_sks);
 
        if (!list_empty(&nsub_list))
-               tipc_nodesub_notify(&nsub_list);
+               tipc_publ_notify(&nsub_list, addr);
 
        if (flags & TIPC_WAKEUP_BCAST_USERS)
                tipc_bclink_wakeup_users();
@@ -601,3 +607,93 @@ void tipc_node_unlock(struct tipc_node *node)
                tipc_nametbl_withdraw(TIPC_LINK_STATE, addr,
                                      link_id, addr);
 }
+
+/* Caller should hold node lock for the passed node */
+static int __tipc_nl_add_node(struct tipc_nl_msg *msg, struct tipc_node *node)
+{
+       void *hdr;
+       struct nlattr *attrs;
+
+       hdr = genlmsg_put(msg->skb, msg->portid, msg->seq, &tipc_genl_v2_family,
+                         NLM_F_MULTI, TIPC_NL_NODE_GET);
+       if (!hdr)
+               return -EMSGSIZE;
+
+       attrs = nla_nest_start(msg->skb, TIPC_NLA_NODE);
+       if (!attrs)
+               goto msg_full;
+
+       if (nla_put_u32(msg->skb, TIPC_NLA_NODE_ADDR, node->addr))
+               goto attr_msg_full;
+       if (tipc_node_is_up(node))
+               if (nla_put_flag(msg->skb, TIPC_NLA_NODE_UP))
+                       goto attr_msg_full;
+
+       nla_nest_end(msg->skb, attrs);
+       genlmsg_end(msg->skb, hdr);
+
+       return 0;
+
+attr_msg_full:
+       nla_nest_cancel(msg->skb, attrs);
+msg_full:
+       genlmsg_cancel(msg->skb, hdr);
+
+       return -EMSGSIZE;
+}
+
+int tipc_nl_node_dump(struct sk_buff *skb, struct netlink_callback *cb)
+{
+       int err;
+       int done = cb->args[0];
+       int last_addr = cb->args[1];
+       struct tipc_node *node;
+       struct tipc_nl_msg msg;
+
+       if (done)
+               return 0;
+
+       msg.skb = skb;
+       msg.portid = NETLINK_CB(cb->skb).portid;
+       msg.seq = cb->nlh->nlmsg_seq;
+
+       rcu_read_lock();
+
+       if (last_addr && !tipc_node_find(last_addr)) {
+               rcu_read_unlock();
+               /* We never set seq or call nl_dump_check_consistent() this
+                * means that setting prev_seq here will cause the consistence
+                * check to fail in the netlink callback handler. Resulting in
+                * the NLMSG_DONE message having the NLM_F_DUMP_INTR flag set if
+                * the node state changed while we released the lock.
+                */
+               cb->prev_seq = 1;
+               return -EPIPE;
+       }
+
+       list_for_each_entry_rcu(node, &tipc_node_list, list) {
+               if (last_addr) {
+                       if (node->addr == last_addr)
+                               last_addr = 0;
+                       else
+                               continue;
+               }
+
+               tipc_node_lock(node);
+               err = __tipc_nl_add_node(&msg, node);
+               if (err) {
+                       last_addr = node->addr;
+                       tipc_node_unlock(node);
+                       goto out;
+               }
+
+               tipc_node_unlock(node);
+       }
+       done = 1;
+out:
+       cb->args[0] = done;
+       cb->args[1] = last_addr;
+       rcu_read_unlock();
+
+       return skb->len;
+}