Merge branch 'tipc-next'

Jon Maloy says:

====================
tipc: resolve message disordering problem

When TIPC receives messages from multi-threaded device drivers it may
occasionally deliver messages to their destination sockets in the wrong
order. This happens despite correct resequencing at the link layer,
because the upcall path from link to socket is not protected by any
locks.

These commits solve this problem by introducing an 'input' message
queue in each link, through which messages must be delivered to the
upper layers.
====================

Signed-off-by: David S. Miller <davem@davemloft.net>

15 files changed, 740 insertions, 563 deletions

diff --git a/net/tipc/bcast.c b/net/tipc/bcast.c
index 53f8bf059fec..81b1fef1f5e0 100644
--- a/net/tipc/bcast.c
+++ b/net/tipc/bcast.c
@@ -1,7 +1,7 @@
 /*
  * net/tipc/bcast.c: TIPC broadcast code
  *
- * Copyright (c) 2004-2006, 2014, Ericsson AB
+ * Copyright (c) 2004-2006, 2014-2015, Ericsson AB
  * Copyright (c) 2004, Intel Corporation.
  * Copyright (c) 2005, 2010-2011, Wind River Systems
  * All rights reserved.
@@ -79,6 +79,13 @@ static void tipc_bclink_unlock(struct net *net)
                 tipc_link_reset_all(node);
 }
 
+void tipc_bclink_input(struct net *net)
+{
+        struct tipc_net *tn = net_generic(net, tipc_net_id);
+
+        tipc_sk_mcast_rcv(net, &tn->bclink->arrvq, &tn->bclink->inputq);
+}
+
 uint  tipc_bclink_get_mtu(void)
 {
         return MAX_PKT_DEFAULT_MCAST;
@@ -189,10 +196,8 @@ static void bclink_retransmit_pkt(struct tipc_net *tn, u32 after, u32 to)
 void tipc_bclink_wakeup_users(struct net *net)
 {
         struct tipc_net *tn = net_generic(net, tipc_net_id);
-        struct sk_buff *skb;
 
-        while ((skb = skb_dequeue(&tn->bclink->link.waiting_sks)))
-                tipc_sk_rcv(net, skb);
+        tipc_sk_rcv(net, &tn->bclink->link.wakeupq);
 }
 
 /**
@@ -271,9 +276,8 @@ void tipc_bclink_acknowledge(struct tipc_node *n_ptr, u32 acked)
                 tipc_link_push_packets(tn->bcl);
                 bclink_set_last_sent(net);
         }
-        if (unlikely(released && !skb_queue_empty(&tn->bcl->waiting_sks)))
+        if (unlikely(released && !skb_queue_empty(&tn->bcl->wakeupq)))
                 n_ptr->action_flags |= TIPC_WAKEUP_BCAST_USERS;
-
 exit:
         tipc_bclink_unlock(net);
 }
@@ -283,10 +287,11 @@ exit:
  *
  * RCU and node lock set
  */
-void tipc_bclink_update_link_state(struct net *net, struct tipc_node *n_ptr,
+void tipc_bclink_update_link_state(struct tipc_node *n_ptr,
                                    u32 last_sent)
 {
         struct sk_buff *buf;
+        struct net *net = n_ptr->net;
         struct tipc_net *tn = net_generic(net, tipc_net_id);
 
         /* Ignore "stale" link state info */
@@ -317,7 +322,7 @@ void tipc_bclink_update_link_state(struct net *net, struct tipc_node *n_ptr,
                 struct sk_buff *skb = skb_peek(&n_ptr->bclink.deferred_queue);
                 u32 to = skb ? buf_seqno(skb) - 1 : n_ptr->bclink.last_sent;
 
-                tipc_msg_init(net, msg, BCAST_PROTOCOL, STATE_MSG,
+                tipc_msg_init(tn->own_addr, msg, BCAST_PROTOCOL, STATE_MSG,
                               INT_H_SIZE, n_ptr->addr);
                 msg_set_non_seq(msg, 1);
                 msg_set_mc_netid(msg, tn->net_id);
@@ -358,7 +363,7 @@ static void bclink_peek_nack(struct net *net, struct tipc_msg *msg)
         tipc_node_unlock(n_ptr);
 }
 
-/* tipc_bclink_xmit - broadcast buffer chain to all nodes in cluster
+/* tipc_bclink_xmit - deliver buffer chain to all nodes in cluster
  *                    and to identified node local sockets
  * @net: the applicable net namespace
  * @list: chain of buffers containing message
@@ -373,6 +378,8 @@ int tipc_bclink_xmit(struct net *net, struct sk_buff_head *list)
         int rc = 0;
         int bc = 0;
         struct sk_buff *skb;
+        struct sk_buff_head arrvq;
+        struct sk_buff_head inputq;
 
         /* Prepare clone of message for local node */
         skb = tipc_msg_reassemble(list);
@@ -381,7 +388,7 @@ int tipc_bclink_xmit(struct net *net, struct sk_buff_head *list)
                 return -EHOSTUNREACH;
         }
 
-        /* Broadcast to all other nodes */
+        /* Broadcast to all nodes */
         if (likely(bclink)) {
                 tipc_bclink_lock(net);
                 if (likely(bclink->bcast_nodes.count)) {
@@ -401,12 +408,15 @@ int tipc_bclink_xmit(struct net *net, struct sk_buff_head *list)
         if (unlikely(!bc))
                 __skb_queue_purge(list);
 
-        /* Deliver message clone */
-        if (likely(!rc))
-                tipc_sk_mcast_rcv(net, skb);
-        else
+        if (unlikely(rc)) {
                 kfree_skb(skb);
-
+                return rc;
+        }
+        /* Deliver message clone */
+        __skb_queue_head_init(&arrvq);
+        skb_queue_head_init(&inputq);
+        __skb_queue_tail(&arrvq, skb);
+        tipc_sk_mcast_rcv(net, &arrvq, &inputq);
         return rc;
 }
 
@@ -449,6 +459,9 @@ void tipc_bclink_rcv(struct net *net, struct sk_buff *buf)
         u32 next_in;
         u32 seqno;
         int deferred = 0;
+        int pos = 0;
+        struct sk_buff *iskb;
+        struct sk_buff_head *arrvq, *inputq;
 
         /* Screen out unwanted broadcast messages */
         if (msg_mc_netid(msg) != tn->net_id)
@@ -485,6 +498,8 @@ void tipc_bclink_rcv(struct net *net, struct sk_buff *buf)
         /* Handle in-sequence broadcast message */
         seqno = msg_seqno(msg);
         next_in = mod(node->bclink.last_in + 1);
+        arrvq = &tn->bclink->arrvq;
+        inputq = &tn->bclink->inputq;
 
         if (likely(seqno == next_in)) {
 receive:
@@ -492,20 +507,26 @@ receive:
                 if (likely(msg_isdata(msg))) {
                         tipc_bclink_lock(net);
                         bclink_accept_pkt(node, seqno);
+                        spin_lock_bh(&inputq->lock);
+                        __skb_queue_tail(arrvq, buf);
+                        spin_unlock_bh(&inputq->lock);
+                        node->action_flags |= TIPC_BCAST_MSG_EVT;
                         tipc_bclink_unlock(net);