1 files changed, 356 insertions, 513 deletions

diff --git a/net/tipc/link.c b/net/tipc/link.c
index a4cf364316de..a6b30df6ec02 100644
--- a/net/tipc/link.c
+++ b/net/tipc/link.c
@@ -1,7 +1,7 @@
 /*
  * net/tipc/link.c: TIPC link code
  *
- * Copyright (c) 1996-2007, 2012-2014, Ericsson AB
+ * Copyright (c) 1996-2007, 2012-2015, Ericsson AB
  * Copyright (c) 2004-2007, 2010-2013, Wind River Systems
  * All rights reserved.
  *
@@ -35,6 +35,7 @@
  */
 
 #include "core.h"
+#include "subscr.h"
 #include "link.h"
 #include "bcast.h"
 #include "socket.h"
@@ -88,24 +89,14 @@ static const struct nla_policy tipc_nl_prop_policy[TIPC_NLA_PROP_MAX + 1] = {
 #define  TIMEOUT_EVT     560817u        /* link timer expired */
 
 /*
- * The following two 'message types' is really just implementation
- * data conveniently stored in the message header.
- * They must not be considered part of the protocol
+ * State value stored in 'failover_pkts'
  */
-#define OPEN_MSG   0
-#define CLOSED_MSG 1
-
-/*
- * State value stored in 'exp_msg_count'
- */
-#define START_CHANGEOVER 100000u
+#define FIRST_FAILOVER 0xffffu
 
 static void link_handle_out_of_seq_msg(struct tipc_link *link,
                                        struct sk_buff *skb);
 static void tipc_link_proto_rcv(struct tipc_link *link,
                                 struct sk_buff *skb);
-static int  tipc_link_tunnel_rcv(struct tipc_node *node,
-                                 struct sk_buff **skb);
 static void link_set_supervision_props(struct tipc_link *l_ptr, u32 tol);
 static void link_state_event(struct tipc_link *l_ptr, u32 event);
 static void link_reset_statistics(struct tipc_link *l_ptr);
@@ -114,7 +105,7 @@ static void tipc_link_sync_xmit(struct tipc_link *l);
 static void tipc_link_sync_rcv(struct tipc_node *n, struct sk_buff *buf);
 static void tipc_link_input(struct tipc_link *l, struct sk_buff *skb);
 static bool tipc_data_input(struct tipc_link *l, struct sk_buff *skb);
-
+static bool tipc_link_failover_rcv(struct tipc_link *l, struct sk_buff **skb);
 /*
  *  Simple link routines
  */
@@ -138,32 +129,11 @@ static void tipc_link_put(struct tipc_link *l_ptr)
         kref_put(&l_ptr->ref, tipc_link_release);
 }
 
-static void link_init_max_pkt(struct tipc_link *l_ptr)
+static struct tipc_link *tipc_parallel_link(struct tipc_link *l)
 {
-        struct tipc_node *node = l_ptr->owner;
-        struct tipc_net *tn = net_generic(node->net, tipc_net_id);
-        struct tipc_bearer *b_ptr;
-        u32 max_pkt;
-
-        rcu_read_lock();
-        b_ptr = rcu_dereference_rtnl(tn->bearer_list[l_ptr->bearer_id]);
-        if (!b_ptr) {
-                rcu_read_unlock();
-                return;
-        }
-        max_pkt = (b_ptr->mtu & ~3);
-        rcu_read_unlock();
-
-        if (max_pkt > MAX_MSG_SIZE)
-                max_pkt = MAX_MSG_SIZE;
-
-        l_ptr->max_pkt_target = max_pkt;
-        if (l_ptr->max_pkt_target < MAX_PKT_DEFAULT)
-                l_ptr->max_pkt = l_ptr->max_pkt_target;
-        else
-                l_ptr->max_pkt = MAX_PKT_DEFAULT;
-
-        l_ptr->max_pkt_probes = 0;
+        if (l->owner->active_links[0] != l)
+                return l->owner->active_links[0];
+        return l->owner->active_links[1];
 }
 
 /*
@@ -194,10 +164,10 @@ static void link_timeout(unsigned long data)
         tipc_node_lock(l_ptr->owner);
 
         /* update counters used in statistical profiling of send traffic */
-        l_ptr->stats.accu_queue_sz += skb_queue_len(&l_ptr->outqueue);
+        l_ptr->stats.accu_queue_sz += skb_queue_len(&l_ptr->transmq);
         l_ptr->stats.queue_sz_counts++;
 
-        skb = skb_peek(&l_ptr->outqueue);
+        skb = skb_peek(&l_ptr->transmq);
         if (skb) {
                 struct tipc_msg *msg = buf_msg(skb);
                 u32 length = msg_size(msg);
@@ -229,7 +199,7 @@ static void link_timeout(unsigned long data)
         /* do all other link processing performed on a periodic basis */
         link_state_event(l_ptr, TIMEOUT_EVT);
 
-        if (l_ptr->next_out)
+        if (skb_queue_len(&l_ptr->backlogq))
                 tipc_link_push_packets(l_ptr);
 
         tipc_node_unlock(l_ptr->owner);
@@ -305,16 +275,15 @@ struct tipc_link *tipc_link_create(struct tipc_node *n_ptr,
         msg_set_session(msg, (tn->random & 0xffff));
         msg_set_bearer_id(msg, b_ptr->identity);
         strcpy((char *)msg_data(msg), if_name);
-
+        l_ptr->net_plane = b_ptr->net_plane;
+        l_ptr->advertised_mtu = b_ptr->mtu;
+        l_ptr->mtu = l_ptr->advertised_mtu;
         l_ptr->priority = b_ptr->priority;
         tipc_link_set_queue_limits(l_ptr, b_ptr->window);
-
-        l_ptr->net_plane = b_ptr->net_plane;
-        link_init_max_pkt(l_ptr);
-
         l_ptr->next_out_no = 1;
-        __skb_queue_head_init(&l_ptr->outqueue);
-        __skb_queue_head_init(&l_ptr->deferred_queue);
+        __skb_queue_head_init(&l_ptr->transmq);
+        __skb_queue_head_init(&l_ptr->backlogq);
+        __skb_queue_head_init(&l_ptr->deferdq);
         skb_queue_head_init(&l_ptr->wakeupq);
         skb_queue_head_init(&l_ptr->inputq);
         skb_queue_head_init(&l_ptr->namedq);
@@ -327,15 +296,19 @@ struct tipc_link *tipc_link_create(struct tipc_node *n_ptr,
 }
 
 /**
- * link_delete - Conditional deletion of link.
- *               If timer still running, real delete is done when it expires
- * @link: link to be deleted
+ * tipc_link_delete - Delete a link
+ * @l: link to be deleted
  */
-void tipc_link_delete(struct tipc_link *link)
+void tipc_link_delete(struct tipc_link *l)
 {
-        tipc_link_reset_fragments(link);
-        tipc_node_detach_link(link->owner, link);
-        tipc_link_put(link);
+        tipc_link_reset(l);
+        if (del_timer(&l->timer))
+                tipc_link_put(l);
+        l->flags |= LINK_STOPPED;
+        /* Delete link now, or when timer is finished: */
+        tipc_link_reset_fragments(l);
+        tipc_node_detach_link(l->owner, l);
+        tipc_link_put(l);
 }
 
 void tipc_link_delete_list(struct net *net, unsigned int bearer_id,
@@ -349,16 +322,7 @@ void tipc_link_delete_list(struct net *net, unsigned int bearer_id,
         list_for_each_entry_rcu(node, &tn->node_list, list) {
                 tipc_node_lock(node);
                 link = node->links[bearer_id];
-                if (!link) {
-                        tipc_node_unlock(node);
-                        continue;
-                }
-                tipc_link_reset(link);
-                if (del_timer(&link->timer))
-                        tipc_link_put(link);
-                link->flags |= LINK_STOPPED;
-                /* Delete link now, or when failover is finished: */
-                if (shutting_down || !tipc_node_is_up(node))
+                if (link)
                         tipc_link_delete(link);
                 tipc_node_unlock(node);
         }
@@ -366,28 +330,43 @@ void tipc_link_delete_list(struct net *net, unsigned int bearer_id,
 }
 
 /**
- * link_schedule_user - schedule user for wakeup after congestion
+ * link_schedule_user - schedule a message sender for wakeup after congestion
  * @link: congested link
- * @oport: sending port
- * @chain_sz: size of buffer chain that was attempted sent
- * @imp: importance of message attempted sent
+ * @list: message that was attempted sent
  * Create pseudo msg to send back to user when congestion abates
+ * Only consumes message if there is an error
  */
-static bool link_schedule_user(struct tipc_link *link, u32 oport,
-                               uint chain_sz, uint imp)
+static int link_schedule_user(struct tipc_link *link, struct sk_buff_head *list)
 {
-        struct sk_buff *buf;
+        struct tipc_msg *msg = buf_msg(skb_peek(list));
+        int imp = msg_importance(msg);
+        u32 oport = msg_origport(msg);
+        u32 addr = link_own_addr(link);
+        struct sk_buff *skb;
 
-        buf = tipc_msg_create(SOCK_WAKEUP, 0, INT_H_SIZE, 0,
-                              link_own_addr(link), link_own_addr(link),
-                              oport, 0, 0);
-        if (!buf)
-                return false;
-        TIPC_SKB_CB(buf)->chain_sz = chain_sz;
-        TIPC_SKB_CB(buf)->chain_imp = imp;
-        skb_queue_tail(&link->wakeupq, buf);
+        /* This really cannot happen...  */
+        if (unlikely(imp > TIPC_CRITICAL_IMPORTANCE)) {
+                pr_warn("%s<%s>, send queue full", link_rst_msg, link->name);
+                tipc_link_reset(link);
+                goto err;
+        }
+        /* Non-blocking sender: */
+        if (TIPC_SKB_CB(skb_peek(list))->wakeup_pending)
+                return -ELINKCONG;
+
+        /* Create and schedule wakeup pseudo message */
+        skb = tipc_msg_create(SOCK_WAKEUP, 0, INT_H_SIZE, 0,
+                              addr, addr, oport, 0, 0);
+        if (!skb)
+                goto err;
+        TIPC_SKB_CB(skb)->chain_sz = skb_queue_len(list);
+        TIPC_SKB_CB(skb)->chain_imp = imp;
+        skb_queue_tail(&link->wakeupq, skb);
         link->stats.link_congs++;
-        return true;
+        return -ELINKCONG;
+err:
+        __skb_queue_purge(list);
+        return -ENOBUFS;
 }
 
 /**
@@ -396,19 +375,22 @@ static bool link_schedule_user(struct tipc_link *link, u32 oport,
  * Move a number of waiting users, as permitted by available space in
  * the send queue, from link wait queue to node wait queue for wakeup
  */
-void link_prepare_wakeup(struct tipc_link *link)
+void link_prepare_wakeup(struct tipc_link *l)
 {
-        uint pend_qsz = skb_queue_len(&link->outqueue);
+        int pnd[TIPC_SYSTEM_IMPORTANCE + 1] = {0,};
+        int imp, lim;
         struct sk_buff *skb, *tmp;
 
-        skb_queue_walk_safe(&link->wakeupq, skb, tmp) {
-                if (pend_qsz >= link->queue_limit[TIPC_SKB_CB(skb)->chain_imp])
+        skb_queue_walk_safe(&l->wakeupq, skb, tmp) {
+                imp = TIPC_SKB_CB(skb)->chain_imp;
+                lim = l->window + l->backlog[imp].limit;
+                pnd[imp] += TIPC_SKB_CB(skb)->chain_sz;
+                if ((pnd[imp] + l->backlog[imp].len) >= lim)
                         break;
-                pend_qsz += TIPC_SKB_CB(skb)->chain_sz;
-                skb_unlink(skb, &link->wakeupq);
-                skb_queue_tail(&link->inputq, skb);
-                link->owner->inputq = &link->inputq;
-                link->owner->action_flags |= TIPC_MSG_EVT;
+                skb_unlink(skb, &l->wakeupq);
+                skb_queue_tail(&l->inputq, skb);
+                l->owner->inputq = &l->inputq;
+                l->owner->action_flags |= TIPC_MSG_EVT;
         }
 }
 
@@ -422,31 +404,42 @@ void tipc_link_reset_fragments(struct tipc_link *l_ptr)
         l_ptr->reasm_buf = NULL;
 }
 
+static void tipc_link_purge_backlog(struct tipc_link *l)
+{
+        __skb_queue_purge(&l->backlogq);
+        l->backlog[TIPC_LOW_IMPORTANCE].len = 0;
+        l->backlog[TIPC_MEDIUM_IMPORTANCE].len = 0;
+        l->backlog[TIPC_HIGH_IMPORTANCE].len = 0;
+        l->backlog[TIPC_CRITICAL_IMPORTANCE].len = 0;
+        l->backlog[TIPC_SYSTEM_IMPORTANCE].len = 0;
+}
+
 /**
  * tipc_link_purge_queues - purge all pkt queues associated with link
  * @l_ptr: pointer to link
  */
 void tipc_link_purge_queues(struct tipc_link *l_ptr)
 {
-        __skb_queue_purge(&l_ptr->deferred_queue);
-        __skb_queue_purge(&l_ptr->outqueue);
+        __skb_queue_purge(&l_ptr->deferdq);
+        __skb_queue_purge(&l_ptr->transmq);
+        tipc_link_purge_backlog(l_ptr);
         tipc_link_reset_fragments(l_ptr);
 }
 
 void tipc_link_reset(struct tipc_link *l_ptr)
 {
         u32 prev_state = l_ptr->state;
-        u32 checkpoint = l_ptr->next_in_no;
         int was_active_link = tipc_link_is_active(l_ptr);
         struct tipc_node *owner = l_ptr->owner;
+        struct tipc_link *pl = tipc_parallel_link(l_ptr);
 
         msg_set_session(l_ptr->pmsg, ((msg_session(l_ptr->pmsg) + 1) & 0xffff));
 
         /* Link is down, accept any session */
         l_ptr->peer_session = INVALID_SESSION;
 
-        /* Prepare for max packet size negotiation */
-        link_init_max_pkt(l_ptr);
+        /* Prepare for renewed mtu size negotiation */
+        l_ptr->mtu = l_ptr->advertised_mtu;
 
         l_ptr->state = RESET_UNKNOWN;
 
@@ -456,20 +449,26 @@ void tipc_link_reset(struct tipc_link *l_ptr)
         tipc_node_link_down(l_ptr->owner, l_ptr);
         tipc_bearer_remove_dest(owner->net, l_ptr->bearer_id, l_ptr->addr);
 
-        if (was_active_link && tipc_node_active_links(l_ptr->owner)) {
-                l_ptr->reset_checkpoint = checkpoint;
-                l_ptr->exp_msg_count = START_CHANGEOVER;
+        if (was_active_link && tipc_node_is_up(l_ptr->owner) && (pl != l_ptr)) {
+                l_ptr->flags |= LINK_FAILINGOVER;
+                l_ptr->failover_checkpt = l_ptr->next_in_no;
+                pl->failover_pkts = FIRST_FAILOVER;
+                pl->failover_checkpt = l_ptr->next_in_no;
+                pl->failover_skb = l_ptr->reasm_buf;
+        } else {
+                kfree_skb(l_ptr->reasm_buf);
         }
-
         /* Clean up all queues, except inputq: */
-        __skb_queue_purge(&l_ptr->outqueue);
-        __skb_queue_purge(&l_ptr->deferred_queue);
-        skb_queue_splice_init(&l_ptr->wakeupq, &l_ptr->inputq);
-        if (!skb_queue_empty(&l_ptr->inputq))
+        __skb_queue_purge(&l_ptr->transmq);
+        __skb_queue_purge(&l_ptr->deferdq);
+        if (!owner->inputq)
+                owner->inputq = &l_ptr->inputq;
+        skb_queue_splice_init(&l_ptr->wakeupq, owner->inputq);
+        if (!skb_queue_empty(owner->inputq))
                 owner->action_flags |= TIPC_MSG_EVT;
-        owner->inputq = &l_ptr->inputq;
-        l_ptr->next_out = NULL;
-        l_ptr->unacked_window = 0;
+        tipc_link_purge_backlog(l_ptr);
+        l_ptr->reasm_buf = NULL;
+        l_ptr->rcv_unacked = 0;
         l_ptr->checkpoint = 1;
         l_ptr->next_out_no = 1;
         l_ptr->fsm_msg_cnt = 0;
@@ -520,8 +519,7 @@ static void link_state_event(struct tipc_link *l_ptr, unsigned int event)
         if (!(l_ptr->flags & LINK_STARTED) && (event != STARTING_EVT))
                 return;         /* Not yet. */
 
-        /* Check whether changeover is going on */
-        if (l_ptr->exp_msg_count) {
+        if (l_ptr->flags & LINK_FAILINGOVER) {
                 if (event == TIMEOUT_EVT)
                         link_set_timer(l_ptr, cont_intv);
                 return;
@@ -538,11 +536,7 @@ static void link_state_event(struct tipc_link *l_ptr, unsigned int event)
                                 l_ptr->checkpoint = l_ptr->next_in_no;
                                 if (tipc_bclink_acks_missing(l_ptr->owner)) {
                                         tipc_link_proto_xmit(l_ptr, STATE_MSG,
-                                                             0, 0, 0, 0, 0);
-                                        l_ptr->fsm_msg_cnt++;
-                                } else if (l_ptr->max_pkt < l_ptr->max_pkt_target) {
-                                        tipc_link_proto_xmit(l_ptr, STATE_MSG,
-                                                             1, 0, 0, 0, 0);
+                                                             0, 0, 0, 0);
                                         l_ptr->fsm_msg_cnt++;
                                 }
                                 link_set_timer(l_ptr, cont_intv);
@@ -550,7 +544,7 @@ static void link_state_event(struct tipc_link *l_ptr, unsigned int event)
                         }
                         l_ptr->state = WORKING_UNKNOWN;
                         l_ptr->fsm_msg_cnt = 0;
-                        tipc_link_proto_xmit(l_ptr, STATE_MSG, 1, 0, 0, 0, 0);
+                        tipc_link_proto_xmit(l_ptr, STATE_MSG, 1, 0, 0, 0);
                         l_ptr->fsm_msg_cnt++;
                         link_set_timer(l_ptr, cont_intv / 4);
                         break;
@@ -561,7 +555,7 @@ static void link_state_event(struct tipc_link *l_ptr, unsigned int event)
                         l_ptr->state = RESET_RESET;
                         l_ptr->fsm_msg_cnt = 0;
                         tipc_link_proto_xmit(l_ptr, ACTIVATE_MSG,
-                                             0, 0, 0, 0, 0);
+                                             0, 0, 0, 0);
                         l_ptr->fsm_msg_cnt++;
                         link_set_timer(l_ptr, cont_intv);
                         break;
@@ -584,7 +578,7 @@ static void link_state_event(struct tipc_link *l_ptr, unsigned int event)
                         l_ptr->state = RESET_RESET;
                         l_ptr->fsm_msg_cnt = 0;
                         tipc_link_proto_xmit(l_ptr, ACTIVATE_MSG,
-                                             0, 0, 0, 0, 0);
+                                             0, 0, 0, 0);
                         l_ptr->fsm_msg_cnt++;
                         link_set_timer(l_ptr, cont_intv);
                         break;
@@ -595,13 +589,13 @@ static void link_state_event(struct tipc_link *l_ptr, unsigned int event)
                                 l_ptr->checkpoint = l_ptr->next_in_no;
                                 if (tipc_bclink_acks_missing(l_ptr->owner)) {
                                         tipc_link_proto_xmit(l_ptr, STATE_MSG,
-                                                             0, 0, 0, 0, 0);
+                                                             0, 0, 0, 0);
                                         l_ptr->fsm_msg_cnt++;
                                 }
                                 link_set_timer(l_ptr, cont_intv);
                         } else if (l_ptr->fsm_msg_cnt < l_ptr->abort_limit) {
                                 tipc_link_proto_xmit(l_ptr, STATE_MSG,
-                                                     1, 0, 0, 0, 0);
+                                                     1, 0, 0, 0);
                                 l_ptr->fsm_msg_cnt++;
                                 link_set_timer(l_ptr, cont_intv / 4);
                         } else {        /* Link has failed */
@@ -611,7 +605,7 @@ static void link_state_event(struct tipc_link *l_ptr, unsigned int event)
                                 l_ptr->state = RESET_UNKNOWN;
                                 l_ptr->fsm_msg_cnt = 0;
                                 tipc_link_proto_xmit(l_ptr, RESET_MSG,
-                                                     0, 0, 0, 0, 0);
+                                                     0, 0, 0, 0);
                                 l_ptr->fsm_msg_cnt++;
                                 link_set_timer(l_ptr, cont_intv);
                         }
@@ -631,7 +625,7 @@ static void link_state_event(struct tipc_link *l_ptr, unsigned int event)
                         l_ptr->state = WORKING_WORKING;
                         l_ptr->fsm_msg_cnt = 0;
                         link_activate(l_ptr);
-                        tipc_link_proto_xmit(l_ptr, STATE_MSG, 1, 0, 0, 0, 0);
+                        tipc_link_proto_xmit(l_ptr, STATE_MSG, 1, 0, 0, 0);
                         l_ptr->fsm_msg_cnt++;
                         if (l_ptr->owner->working_links == 1)
                                 tipc_link_sync_xmit(l_ptr);
@@ -641,7 +635,7 @@ static void link_state_event(struct tipc_link *l_ptr, unsigned int event)
                         l_ptr->state = RESET_RESET;
                         l_ptr->fsm_msg_cnt = 0;
                         tipc_link_proto_xmit(l_ptr, ACTIVATE_MSG,
-                                             1, 0, 0, 0, 0);
+                                             1, 0, 0, 0);
                         l_ptr->fsm_msg_cnt++;
                         link_set_timer(l_ptr, cont_intv);
                         break;
@@ -651,7 +645,7 @@ static void link_state_event(struct tipc_link *l_ptr, unsigned int event)
                         link_set_timer(l_ptr, cont_intv);
                         break;
                 case TIMEOUT_EVT:
-                        tipc_link_proto_xmit(l_ptr, RESET_MSG, 0, 0, 0, 0, 0);
+                        tipc_link_proto_xmit(l_ptr, RESET_MSG, 0, 0, 0, 0);
                         l_ptr->fsm_msg_cnt++;
                         link_set_timer(l_ptr, cont_intv);
                         break;
@@ -669,7 +663,7 @@ static void link_state_event(struct tipc_link *l_ptr, unsigned int event)
                         l_ptr->state = WORKING_WORKING;
                         l_ptr->fsm_msg_cnt = 0;
                         link_activate(l_ptr);
-                        tipc_link_proto_xmit(l_ptr, STATE_MSG, 1, 0, 0, 0, 0);
+                        tipc_link_proto_xmit(l_ptr, STATE_MSG, 1, 0, 0, 0);
                         l_ptr->fsm_msg_cnt++;
                         if (l_ptr->owner->working_links == 1)
                                 tipc_link_sync_xmit(l_ptr);
@@ -679,7 +673,7 @@ static void link_state_event(struct tipc_link *l_ptr, unsigned int event)
                         break;
                 case TIMEOUT_EVT:
                         tipc_link_proto_xmit(l_ptr, ACTIVATE_MSG,
-                                             0, 0, 0, 0, 0);
+                                             0, 0, 0, 0);
                         l_ptr->fsm_msg_cnt++;
                         link_set_timer(l_ptr, cont_intv);
                         break;
@@ -692,101 +686,65 @@ static void link_state_event(struct tipc_link *l_ptr, unsigned int event)
         }
 }
 
-/* tipc_link_cong: determine return value and how to treat the
- * sent buffer during link congestion.
- * - For plain, errorless user data messages we keep the buffer and
- *   return -ELINKONG.
- * - For all other messages we discard the buffer and return -EHOSTUNREACH
- * - For TIPC internal messages we also reset the link
- */
-static int tipc_link_cong(struct tipc_link *link, struct sk_buff_head *list)
-{
-        struct sk_buff *skb = skb_peek(list);
-        struct tipc_msg *msg = buf_msg(skb);
-        uint imp = tipc_msg_tot_importance(msg);
-        u32 oport = msg_tot_origport(msg);
-
-        if (unlikely(imp > TIPC_CRITICAL_IMPORTANCE)) {
-                pr_warn("%s<%s>, send queue full", link_rst_msg, link->name);
-                tipc_link_reset(link);
-                goto drop;
-        }
-        if (unlikely(msg_errcode(msg)))
-                goto drop;
-        if (unlikely(msg_reroute_cnt(msg)))
-                goto drop;
-        if (TIPC_SKB_CB(skb)->wakeup_pending)
-                return -ELINKCONG;
-        if (link_schedule_user(link, oport, skb_queue_len(list), imp))
-                return -ELINKCONG;
-drop:
-        __skb_queue_purge(list);
-        return -EHOSTUNREACH;
-}
-
 /**
  * __tipc_link_xmit(): same as tipc_link_xmit, but destlink is known & locked
  * @link: link to use
  * @list: chain of buffers containing message
  *
- * Consumes the buffer chain, except when returning -ELINKCONG
- * Returns 0 if success, otherwise errno: -ELINKCONG, -EMSGSIZE (plain socket
- * user data messages) or -EHOSTUNREACH (all other messages/senders)
- * Only the socket functions tipc_send_stream() and tipc_send_packet() need
- * to act on the return value, since they may need to do more send attempts.
+ * Consumes the buffer chain, except when returning -ELINKCONG,
+ * since the caller then may want to make more send attempts.
+ * Returns 0 if success, or errno: -ELINKCONG, -EMSGSIZE or -ENOBUFS
+ * Messages at TIPC_SYSTEM_IMPORTANCE are always accepted
  */
 int __tipc_link_xmit(struct net *net, struct tipc_link *link,
                      struct sk_buff_head *list)
 {
         struct tipc_msg *msg = buf_msg(skb_peek(list));
-        uint psz = msg_size(msg);
-        uint sndlim = link->queue_limit[0];
-        uint imp = tipc_msg_tot_importance(msg);
-        uint mtu = link->max_pkt;
+        unsigned int maxwin = link->window;
+        unsigned int imp = msg_importance(msg);
+        uint mtu = link->mtu;
         uint ack = mod(link->next_in_no - 1);
         uint seqno = link->next_out_no;
         uint bc_last_in = link->owner->bclink.last_in;
         struct tipc_media_addr *addr = &link->media_addr;
-        struct sk_buff_head *outqueue = &link->outqueue;
+        struct sk_buff_head *transmq = &link->transmq;
+        struct sk_buff_head *backlogq = &link->backlogq;
         struct sk_buff *skb, *tmp;
 
-        /* Match queue limits against msg importance: */
-        if (unlikely(skb_queue_len(outqueue) >= link->queue_limit[imp]))
-                return tipc_link_cong(link, list);
+        /* Match backlog limit against msg importance: */
+        if (unlikely(link->backlog[imp].len >= link->backlog[imp].limit))
+                return link_schedule_user(link, list);
 
-        /* Has valid packet limit been used ? */
-        if (unlikely(psz > mtu)) {
+        if (unlikely(msg_size(msg) > mtu)) {
                 __skb_queue_purge(list);
                 return -EMSGSIZE;
         }
-
-        /* Prepare each packet for sending, and add to outqueue: */
+        /* Prepare each packet for sending, and add to relevant queue: */
         skb_queue_walk_safe(list, skb, tmp) {
                 __skb_unlink(skb, list);
                 msg = buf_msg(skb);
-                msg_set_word(msg, 2, ((ack << 16) | mod(seqno)));
+                msg_set_seqno(msg, seqno);
+                msg_set_ack(msg, ack);
                 msg_set_bcast_ack(msg, bc_last_in);
 
-                if (skb_queue_len(outqueue) < sndlim) {
-                        __skb_queue_tail(outqueue, skb);
-                        tipc_bearer_send(net, link->bearer_id,
-                                         skb, addr);
-                        link->next_out = NULL;
-                        link->unacked_window = 0;
-                } else if (tipc_msg_bundle(outqueue, skb, mtu)) {
+                if (likely(skb_queue_len(transmq) < maxwin)) {
+                        __skb_queue_tail(transmq, skb);
+                        tipc_bearer_send(net, link->bearer_id, skb, addr);
+                        link->rcv_unacked = 0;
+                        seqno++;
+                        continue;
+                }
+                if (tipc_msg_bundle(skb_peek_tail(backlogq), skb, mtu)) {
                         link->stats.sent_bundled++;
                         continue;
-                } else if (tipc_msg_make_bundle(outqueue, skb, mtu,
-                                                link->addr)) {
+                }
+                if (tipc_msg_make_bundle(&skb, mtu, link->addr)) {
                         link->stats.sent_bundled++;
                         link->stats.sent_bundles++;
-                        if (!link->next_out)
-                                link->next_out = skb_peek_tail(outqueue);
-                } else {
-                        __skb_queue_tail(outqueue, skb);
-                        if (!link->next_out)
-                                link->next_out = skb;
+                        imp = msg_importance(buf_msg(skb));
                 }
+                __skb_queue_tail(backlogq, skb);
+                link->backlog[imp].len++;
                 seqno++;
         }
         link->next_out_no = seqno;
@@ -807,13 +765,25 @@ static int __tipc_link_xmit_skb(struct tipc_link *link, struct sk_buff *skb)
         return __tipc_link_xmit(link->owner->net, link, &head);
 }
 
+/* tipc_link_xmit_skb(): send single buffer to destination
+ * Buffers sent via this functon are generally TIPC_SYSTEM_IMPORTANCE
+ * messages, which will not be rejected
+ * The only exception is datagram messages rerouted after secondary
+ * lookup, which are rare and safe to dispose of anyway.
+ * TODO: Return real return value, and let callers use
+ * tipc_wait_for_sendpkt() where applicable
+ */
 int tipc_link_xmit_skb(struct net *net, struct sk_buff *skb, u32 dnode,
                        u32 selector)
 {
         struct sk_buff_head head;
+        int rc;
 
         skb2list(skb, &head);
-        return tipc_link_xmit(net, &head, dnode, selector);
+        rc = tipc_link_xmit(net, &head, dnode, selector);
+        if (rc == -ELINKCONG)
+                kfree_skb(skb);
+        return 0;
 }
 
 /**
@@ -840,12 +810,15 @@ int tipc_link_xmit(struct net *net, struct sk_buff_head *list, u32 dnode,
                 if (link)
                         rc = __tipc_link_xmit(net, link, list);
                 tipc_node_unlock(node);
+                tipc_node_put(node);
         }
         if (link)
                 return rc;
 
-        if (likely(in_own_node(net, dnode)))
-                return tipc_sk_rcv(net, list);
+        if (likely(in_own_node(net, dnode))) {
+                tipc_sk_rcv(net, list);
+                return 0;
+        }
 
         __skb_queue_purge(list);
         return rc;
@@ -892,14 +865,6 @@ static void tipc_link_sync_rcv(struct tipc_node *n, struct sk_buff *buf)
         kfree_skb(buf);
 }
 
-struct sk_buff *tipc_skb_queue_next(const struct sk_buff_head *list,
-                                    const struct sk_buff *skb)
-{
-        if (skb_queue_is_last(list, skb))
-                return NULL;
-        return skb->next;
-}
-
 /*
  * tipc_link_push_packets - push unsent packets to bearer
  *
@@ -908,30 +873,24 @@ struct sk_buff *tipc_skb_queue_next(const struct sk_buff_head *list,
  *
  * Called with node locked
  */
-void tipc_link_push_packets(struct tipc_link *l_ptr)
+void tipc_link_push_packets(struct tipc_link *link)
 {
-        struct sk_buff_head *outqueue = &l_ptr->outqueue;
-        struct sk_buff *skb = l_ptr->next_out;
+        struct sk_buff *skb;
         struct tipc_msg *msg;
-        u32 next, first;
+        unsigned int ack = mod(link->next_in_no - 1);
 
-        skb_queue_walk_from(outqueue, skb) {
-                msg = buf_msg(skb);
-                next = msg_seqno(msg);
-                first = buf_seqno(skb_peek(outqueue));
-
-                if (mod(next - first) < l_ptr->queue_limit[0]) {
-                        msg_set_ack(msg, mod(l_ptr->next_in_no - 1));
-                        msg_set_bcast_ack(msg, l_ptr->owner->bclink.last_in);
-                        if (msg_user(msg) == MSG_BUNDLER)
-                                TIPC_SKB_CB(skb)->bundling = false;
-                        tipc_bearer_send(l_ptr->owner->net,
-                                         l_ptr->bearer_id, skb,
-                                         &l_ptr->media_addr);
-                        l_ptr->next_out = tipc_skb_queue_next(outqueue, skb);
-                } else {
+        while (skb_queue_len(&link->transmq) < link->window) {
+                skb = __skb_dequeue(&link->backlogq);
+                if (!skb)
                         break;
-                }
+                msg = buf_msg(skb);
+                link->backlog[msg_importance(msg)].len--;
+                msg_set_ack(msg, ack);
+                msg_set_bcast_ack(msg, link->owner->bclink.last_in);
+                link->rcv_unacked = 0;
+                __skb_queue_tail(&link->transmq, skb);
+                tipc_bearer_send(link->owner->net, link->bearer_id,
+                                 skb, &link->media_addr);
         }
 }
 
@@ -978,7 +937,6 @@ static void link_retransmit_failure(struct tipc_link *l_ptr,
                         (unsigned long) TIPC_SKB_CB(buf)->handle);
 
                 n_ptr = tipc_bclink_retransmit_to(net);
-                tipc_node_lock(n_ptr);
 
                 tipc_addr_string_fill(addr_string, n_ptr->addr);
                 pr_info("Broadcast link info for %s\n", addr_string);
@@ -990,9 +948,7 @@ static void link_retransmit_failure(struct tipc_link *l_ptr,
                         n_ptr->bclink.oos_state,
                         n_ptr->bclink.last_sent);
 
-                tipc_node_unlock(n_ptr);
-
-                tipc_bclink_set_flags(net, TIPC_BCLINK_RESET);
+                n_ptr->action_flags |= TIPC_BCAST_RESET;
                 l_ptr->stale_count = 0;
         }
 }
@@ -1018,8 +974,8 @@ void tipc_link_retransmit(struct tipc_link *l_ptr, struct sk_buff *skb,
                 l_ptr->stale_count = 1;
         }
 
-        skb_queue_walk_from(&l_ptr->outqueue, skb) {
-                if (!retransmits || skb == l_ptr->next_out)
+        skb_queue_walk_from(&l_ptr->transmq, skb) {
+                if (!retransmits)
                         break;
                 msg = buf_msg(skb);
                 msg_set_ack(msg, mod(l_ptr->next_in_no - 1));
@@ -1031,72 +987,43 @@ void tipc_link_retransmit(struct tipc_link *l_ptr, struct sk_buff *skb,
         }
 }
 
-static void link_retrieve_defq(struct tipc_link *link,
-                               struct sk_buff_head *list)
-{
-        u32 seq_no;
-
-        if (skb_queue_empty(&link->deferred_queue))
-                return;
-
-        seq_no = buf_seqno(skb_peek(&link->deferred_queue));
-        if (seq_no == mod(link->next_in_no))
-                skb_queue_splice_tail_init(&link->deferred_queue, list);
-}
-
-/**
- * link_recv_buf_validate - validate basic format of received message
- *
- * This routine ensures a TIPC message has an acceptable header, and at least
- * as much data as the header indicates it should.  The routine also ensures
- * that the entire message header is stored in the main fragment of the message
- * buffer, to simplify future access to message header fields.
- *
- * Note: Having extra info present in the message header or data areas is OK.
- * TIPC will ignore the excess, under the assumption that it is optional info
- * introduced by a later release of the protocol.
+/* link_synch(): check if all packets arrived before the synch
+ *               point have been consumed
+ * Returns true if the parallel links are synched, otherwise false
  */
-static int link_recv_buf_validate(struct sk_buff *buf)
+static bool link_synch(struct tipc_link *l)
 {
-        static u32 min_data_hdr_size[8] = {
-                SHORT_H_SIZE, MCAST_H_SIZE, NAMED_H_SIZE, BASIC_H_SIZE,
-                MAX_H_SIZE, MAX_H_SIZE, MAX_H_SIZE, MAX_H_SIZE
-                };
+        unsigned int post_synch;
+        struct tipc_link *pl;
 
-        struct tipc_msg *msg;
-        u32 tipc_hdr[2];
-        u32 size;
-        u32 hdr_size;
-        u32 min_hdr_size;
+        pl  = tipc_parallel_link(l);
+        if (pl == l)
+                goto synched;
 
-        /* If this packet comes from the defer queue, the skb has already
-         * been validated
-         */
-        if (unlikely(TIPC_SKB_CB(buf)->deferred))
-                return 1;
-
-        if (unlikely(buf->len < MIN_H_SIZE))
-                return 0;
-
-        msg = skb_header_pointer(buf, 0, sizeof(tipc_hdr), tipc_hdr);
-        if (msg == NULL)
-                return 0;
+        /* Was last pre-synch packet added to input queue ? */
+        if (less_eq(pl->next_in_no, l->synch_point))
+                return false;
 
-        if (unlikely(msg_version(msg) != TIPC_VERSION))
-                return 0;
+        /* Is it still in the input queue ? */
+        post_synch = mod(pl->next_in_no - l->synch_point) - 1;
+        if (skb_queue_len(&pl->inputq) > post_synch)
+                return false;
+synched:
+        l->flags &= ~LINK_SYNCHING;
+        return true;
+}
 
-        size = msg_size(msg);
-        hdr_size = msg_hdr_sz(msg);
-        min_hdr_size = msg_isdata(msg) ?
-                min_data_hdr_size[msg_type(msg)] : INT_H_SIZE;
+static void link_retrieve_defq(struct tipc_link *link,
+                               struct sk_buff_head *list)
+{
+        u32 seq_no;
 
-        if (unlikely((hdr_size < min_hdr_size) ||
-                     (size < hdr_size) ||
-                     (buf->len < size) ||
-                     (size - hdr_size > TIPC_MAX_USER_MSG_SIZE)))
-                return 0;
+        if (skb_queue_empty(&link->deferdq))
+                return;
 
-        return pskb_may_pull(buf, hdr_size);
+        seq_no = buf_seqno(skb_peek(&link->deferdq));
+        if (seq_no == mod(link->next_in_no))
+                skb_queue_splice_tail_init(&link->deferdq, list);
 }
 
 /**
@@ -1124,16 +1051,11 @@ void tipc_rcv(struct net *net, struct sk_buff *skb, struct tipc_bearer *b_ptr)
 
         while ((skb = __skb_dequeue(&head))) {
                 /* Ensure message is well-formed */
-                if (unlikely(!link_recv_buf_validate(skb)))
-                        goto discard;
-
-                /* Ensure message data is a single contiguous unit */
-                if (unlikely(skb_linearize(skb)))
+                if (unlikely(!tipc_msg_validate(skb)))
                         goto discard;
 
                 /* Handle arrival of a non-unicast link message */
                 msg = buf_msg(skb);
-
                 if (unlikely(msg_non_seq(msg))) {
                         if (msg_user(msg) ==  LINK_CONFIG)
                                 tipc_disc_rcv(net, skb, b_ptr);
@@ -1151,8 +1073,8 @@ void tipc_rcv(struct net *net, struct sk_buff *skb, struct tipc_bearer *b_ptr)
                 n_ptr = tipc_node_find(net, msg_prevnode(msg));
                 if (unlikely(!n_ptr))
                         goto discard;
-                tipc_node_lock(n_ptr);
 
+                tipc_node_lock(n_ptr);
                 /* Locate unicast link endpoint that should handle message */
                 l_ptr = n_ptr->links[b_ptr->identity];
                 if (unlikely(!l_ptr))
@@ -1174,21 +1096,20 @@ void tipc_rcv(struct net *net, struct sk_buff *skb, struct tipc_bearer *b_ptr)
                 ackd = msg_ack(msg);
 
                 /* Release acked messages */
-                if (n_ptr->bclink.recv_permitted)
+                if (unlikely(n_ptr->bclink.acked != msg_bcast_ack(msg)))
                         tipc_bclink_acknowledge(n_ptr, msg_bcast_ack(msg));
 
                 released = 0;
-                skb_queue_walk_safe(&l_ptr->outqueue, skb1, tmp) {
-                        if (skb1 == l_ptr->next_out ||
-                            more(buf_seqno(skb1), ackd))
+                skb_queue_walk_safe(&l_ptr->transmq, skb1, tmp) {
+                        if (more(buf_seqno(skb1), ackd))
                                 break;
-                         __skb_unlink(skb1, &l_ptr->outqueue);
+                         __skb_unlink(skb1, &l_ptr->transmq);
                          kfree_skb(skb1);
                          released = 1;
                 }
 
                 /* Try sending any messages link endpoint has pending */
-                if (unlikely(l_ptr->next_out))
+                if (unlikely(skb_queue_len(&l_ptr->backlogq)))
                         tipc_link_push_packets(l_ptr);
 
                 if (released && !skb_queue_empty(&l_ptr->wakeupq))
@@ -1222,18 +1143,26 @@ void tipc_rcv(struct net *net, struct sk_buff *skb, struct tipc_bearer *b_ptr)
                         skb = NULL;
                         goto unlock;
                 }
+                /* Synchronize with parallel link if applicable */
+                if (unlikely((l_ptr->flags & LINK_SYNCHING) && !msg_dup(msg))) {
+                        link_handle_out_of_seq_msg(l_ptr, skb);
+                        if (link_synch(l_ptr))
+                                link_retrieve_defq(l_ptr, &head);
+                        skb = NULL;
+                        goto unlock;
+                }
                 l_ptr->next_in_no++;
-                if (unlikely(!skb_queue_empty(&l_ptr->deferred_queue)))
+                if (unlikely(!skb_queue_empty(&l_ptr->deferdq)))
                         link_retrieve_defq(l_ptr, &head);
-
-                if (unlikely(++l_ptr->unacked_window >= TIPC_MIN_LINK_WIN)) {
+                if (unlikely(++l_ptr->rcv_unacked >= TIPC_MIN_LINK_WIN)) {
                         l_ptr->stats.sent_acks++;
-                        tipc_link_proto_xmit(l_ptr, STATE_MSG, 0, 0, 0, 0, 0);
+                        tipc_link_proto_xmit(l_ptr, STATE_MSG, 0, 0, 0, 0);
                 }
                 tipc_link_input(l_ptr, skb);
                 skb = NULL;
 unlock:
                 tipc_node_unlock(n_ptr);
+                tipc_node_put(n_ptr);
 discard:
                 if (unlikely(skb))
                         kfree_skb(skb);
@@ -1270,7 +1199,7 @@ static bool tipc_data_input(struct tipc_link *link, struct sk_buff *skb)
                         node->action_flags |= TIPC_NAMED_MSG_EVT;
                 return true;
         case MSG_BUNDLER:
-        case CHANGEOVER_PROTOCOL:
+        case TUNNEL_PROTOCOL:
         case MSG_FRAGMENTER:
         case BCAST_PROTOCOL:
                 return false;
@@ -1297,8 +1226,14 @@ static void tipc_link_input(struct tipc_link *link, struct sk_buff *skb)
                 return;
 
         switch (msg_user(msg)) {
-        case CHANGEOVER_PROTOCOL:
-                if (!tipc_link_tunnel_rcv(node, &skb))
+        case TUNNEL_PROTOCOL:
+                if (msg_dup(msg)) {
+                        link->flags |= LINK_SYNCHING;
+                        link->synch_point = msg_seqno(msg_get_wrapped(msg));
+                        kfree_skb(skb);
+                        break;
+                }
+                if (!tipc_link_failover_rcv(link, &skb))
                         break;
                 if (msg_user(buf_msg(skb)) != MSG_BUNDLER) {
                         tipc_data_input(link, skb);
@@ -1393,11 +1328,10 @@ static void link_handle_out_of_seq_msg(struct tipc_link *l_ptr,
                 return;
         }
 
-        if (tipc_link_defer_pkt(&l_ptr->deferred_queue, buf)) {
+        if (tipc_link_defer_pkt(&l_ptr->deferdq, buf)) {
                 l_ptr->stats.deferred_recv++;
-                TIPC_SKB_CB(buf)->deferred = true;
-                if ((skb_queue_len(&l_ptr->deferred_queue) % 16) == 1)
-                        tipc_link_proto_xmit(l_ptr, STATE_MSG, 0, 0, 0, 0, 0);
+                if ((skb_queue_len(&l_ptr->deferdq) % TIPC_MIN_LINK_WIN) == 1)
+                        tipc_link_proto_xmit(l_ptr, STATE_MSG, 0, 0, 0, 0);
         } else {
                 l_ptr->stats.duplicates++;
         }
@@ -1407,15 +1341,15 @@ static void link_handle_out_of_seq_msg(struct tipc_link *l_ptr,
  * Send protocol message to the other endpoint.
  */
 void tipc_link_proto_xmit(struct tipc_link *l_ptr, u32 msg_typ, int probe_msg,
-                          u32 gap, u32 tolerance, u32 priority, u32 ack_mtu)
+                          u32 gap, u32 tolerance, u32 priority)
 {
         struct sk_buff *buf = NULL;
         struct tipc_msg *msg = l_ptr->pmsg;
         u32 msg_size = sizeof(l_ptr->proto_msg);
         int r_flag;
 
-        /* Don't send protocol message during link changeover */
-        if (l_ptr->exp_msg_count)
+        /* Don't send protocol message during link failover */
+        if (l_ptr->flags & LINK_FAILINGOVER)
                 return;
 
         /* Abort non-RESET send if communication with node is prohibited */
@@ -1433,11 +1367,11 @@ void tipc_link_proto_xmit(struct tipc_link *l_ptr, u32 msg_typ, int probe_msg,
 
                 if (!tipc_link_is_up(l_ptr))
                         return;
-                if (l_ptr->next_out)
-                        next_sent = buf_seqno(l_ptr->next_out);
+                if (skb_queue_len(&l_ptr->backlogq))
+                        next_sent = buf_seqno(skb_peek(&l_ptr->backlogq));
                 msg_set_next_sent(msg, next_sent);
-                if (!skb_queue_empty(&l_ptr->deferred_queue)) {
-                        u32 rec = buf_seqno(skb_peek(&l_ptr->deferred_queue));
+                if (!skb_queue_empty(&l_ptr->deferdq)) {
+                        u32 rec = buf_seqno(skb_peek(&l_ptr->deferdq));
                         gap = mod(rec - mod(l_ptr->next_in_no));
                 }
                 msg_set_seq_gap(msg, gap);
@@ -1445,35 +1379,20 @@ void tipc_link_proto_xmit(struct tipc_link *l_ptr, u32 msg_typ, int probe_msg,
                         l_ptr->stats.sent_nacks++;
                 msg_set_link_tolerance(msg, tolerance);
                 msg_set_linkprio(msg, priority);
-                msg_set_max_pkt(msg, ack_mtu);
+                msg_set_max_pkt(msg, l_ptr->mtu);
                 msg_set_ack(msg, mod(l_ptr->next_in_no - 1));
                 msg_set_probe(msg, probe_msg != 0);
-                if (probe_msg) {
-                        u32 mtu = l_ptr->max_pkt;
-
-                        if ((mtu < l_ptr->max_pkt_target) &&
-                            link_working_working(l_ptr) &&
-                            l_ptr->fsm_msg_cnt) {
-                                msg_size = (mtu + (l_ptr->max_pkt_target - mtu)/2 + 2) & ~3;
-                                if (l_ptr->max_pkt_probes == 10) {
-                                        l_ptr->max_pkt_target = (msg_size - 4);
-                                        l_ptr->max_pkt_probes = 0;
-                                        msg_size = (mtu + (l_ptr->max_pkt_target - mtu)/2 + 2) & ~3;
-                                }
-                                l_ptr->max_pkt_probes++;
-                        }
-
+                if (probe_msg)
                         l_ptr->stats.sent_probes++;
-                }
                 l_ptr->stats.sent_states++;
         } else {                /* RESET_MSG or ACTIVATE_MSG */
-                msg_set_ack(msg, mod(l_ptr->reset_checkpoint - 1));
+                msg_set_ack(msg, mod(l_ptr->failover_checkpt - 1));
                 msg_set_seq_gap(msg, 0);
                 msg_set_next_sent(msg, 1);
                 msg_set_probe(msg, 0);
                 msg_set_link_tolerance(msg, l_ptr->tolerance);
                 msg_set_linkprio(msg, l_ptr->priority);
-                msg_set_max_pkt(msg, l_ptr->max_pkt_target);
+                msg_set_max_pkt(msg, l_ptr->advertised_mtu);
         }
 
         r_flag = (l_ptr->owner->working_links > tipc_link_is_up(l_ptr));
@@ -1489,10 +1408,9 @@ void tipc_link_proto_xmit(struct tipc_link *l_ptr, u32 msg_typ, int probe_msg,
 
         skb_copy_to_linear_data(buf, msg, sizeof(l_ptr->proto_msg));
         buf->priority = TC_PRIO_CONTROL;
-
         tipc_bearer_send(l_ptr->owner->net, l_ptr->bearer_id, buf,
                          &l_ptr->media_addr);
-        l_ptr->unacked_window = 0;
+        l_ptr->rcv_unacked = 0;
         kfree_skb(buf);
 }
 
@@ -1505,13 +1423,10 @@ static void tipc_link_proto_rcv(struct tipc_link *l_ptr,
                                 struct sk_buff *buf)
 {
         u32 rec_gap = 0;
-        u32 max_pkt_info;
-        u32 max_pkt_ack;
         u32 msg_tol;
         struct tipc_msg *msg = buf_msg(buf);
 
-        /* Discard protocol message during link changeover */
-        if (l_ptr->exp_msg_count)
+        if (l_ptr->flags & LINK_FAILINGOVER)
                 goto exit;
 
         if (l_ptr->net_plane != msg_net_plane(msg))
@@ -1550,15 +1465,8 @@ static void tipc_link_proto_rcv(struct tipc_link *l_ptr,
                 if (msg_linkprio(msg) > l_ptr->priority)
                         l_ptr->priority = msg_linkprio(msg);
 
-                max_pkt_info = msg_max_pkt(msg);
-                if (max_pkt_info) {
-                        if (max_pkt_info < l_ptr->max_pkt_target)
-                                l_ptr->max_pkt_target = max_pkt_info;
-                        if (l_ptr->max_pkt > l_ptr->max_pkt_target)
-                                l_ptr->max_pkt = l_ptr->max_pkt_target;
-                } else {
-                        l_ptr->max_pkt = l_ptr->max_pkt_target;
-                }
+                if (l_ptr->mtu > msg_max_pkt(msg))
+                        l_ptr->mtu = msg_max_pkt(msg);
 
                 /* Synchronize broadcast link info, if not done previously */
                 if (!tipc_node_is_up(l_ptr->owner)) {
@@ -1603,18 +1511,8 @@ static void tipc_link_proto_rcv(struct tipc_link *l_ptr,
                                       mod(l_ptr->next_in_no));
                 }
 
-                max_pkt_ack = msg_max_pkt(msg);
-                if (max_pkt_ack > l_ptr->max_pkt) {
-                        l_ptr->max_pkt = max_pkt_ack;
-                        l_ptr->max_pkt_probes = 0;
-                }
-
-                max_pkt_ack = 0;
-                if (msg_probe(msg)) {
+                if (msg_probe(msg))
                         l_ptr->stats.recv_probes++;
-                        if (msg_size(msg) > sizeof(l_ptr->proto_msg))
-                                max_pkt_ack = msg_size(msg);
-                }
 
                 /* Protocol message before retransmits, reduce loss risk */
                 if (l_ptr->owner->bclink.recv_permitted)
@@ -1622,12 +1520,12 @@ static void tipc_link_proto_rcv(struct tipc_link *l_ptr,
                                                       msg_last_bcast(msg));
 
                 if (rec_gap || (msg_probe(msg))) {
-                        tipc_link_proto_xmit(l_ptr, STATE_MSG, 0, rec_gap, 0,
-                                             0, max_pkt_ack);
+                        tipc_link_proto_xmit(l_ptr, STATE_MSG, 0,
+                                             rec_gap, 0, 0);
                 }
                 if (msg_seq_gap(msg)) {
                         l_ptr->stats.recv_nacks++;
-                        tipc_link_retransmit(l_ptr, skb_peek(&l_ptr->outqueue),
+                        tipc_link_retransmit(l_ptr, skb_peek(&l_ptr->transmq),
                                              msg_seq_gap(msg));
                 }
                 break;
@@ -1674,7 +1572,7 @@ static void tipc_link_tunnel_xmit(struct tipc_link *l_ptr,
  */
 void tipc_link_failover_send_queue(struct tipc_link *l_ptr)
 {
-        u32 msgcount = skb_queue_len(&l_ptr->outqueue);
+        int msgcount;
         struct tipc_link *tunnel = l_ptr->owner->active_links[0];
         struct tipc_msg tunnel_hdr;
         struct sk_buff *skb;
@@ -1683,12 +1581,15 @@ void tipc_link_failover_send_queue(struct tipc_link *l_ptr)
         if (!tunnel)
                 return;
 
-        tipc_msg_init(link_own_addr(l_ptr), &tunnel_hdr, CHANGEOVER_PROTOCOL,
-                      ORIGINAL_MSG, INT_H_SIZE, l_ptr->addr);
+        tipc_msg_init(link_own_addr(l_ptr), &tunnel_hdr, TUNNEL_PROTOCOL,
+                      FAILOVER_MSG, INT_H_SIZE, l_ptr->addr);
+        skb_queue_splice_tail_init(&l_ptr->backlogq, &l_ptr->transmq);
+        tipc_link_purge_backlog(l_ptr);
+        msgcount = skb_queue_len(&l_ptr->transmq);
         msg_set_bearer_id(&tunnel_hdr, l_ptr->peer_bearer_id);
         msg_set_msgcnt(&tunnel_hdr, msgcount);
 
-        if (skb_queue_empty(&l_ptr->outqueue)) {
+        if (skb_queue_empty(&l_ptr->transmq)) {
                 skb = tipc_buf_acquire(INT_H_SIZE);
                 if (skb) {
                         skb_copy_to_linear_data(skb, &tunnel_hdr, INT_H_SIZE);
@@ -1704,7 +1605,7 @@ void tipc_link_failover_send_queue(struct tipc_link *l_ptr)
         split_bundles = (l_ptr->owner->active_links[0] !=
                          l_ptr->owner->active_links[1]);
 
-        skb_queue_walk(&l_ptr->outqueue, skb) {
+        skb_queue_walk(&l_ptr->transmq, skb) {
                 struct tipc_msg *msg = buf_msg(skb);
 
                 if ((msg_user(msg) == MSG_BUNDLER) && split_bundles) {
@@ -1735,157 +1636,105 @@ void tipc_link_failover_send_queue(struct tipc_link *l_ptr)
  * and sequence order is preserved per sender/receiver socket pair.
  * Owner node is locked.
  */
-void tipc_link_dup_queue_xmit(struct tipc_link *l_ptr,
-                              struct tipc_link *tunnel)
+void tipc_link_dup_queue_xmit(struct tipc_link *link,
+                              struct tipc_link *tnl)
 {
         struct sk_buff *skb;
-        struct tipc_msg tunnel_hdr;
-
-        tipc_msg_init(link_own_addr(l_ptr), &tunnel_hdr, CHANGEOVER_PROTOCOL,
-                      DUPLICATE_MSG, INT_H_SIZE, l_ptr->addr);
-        msg_set_msgcnt(&tunnel_hdr, skb_queue_len(&l_ptr->outqueue));
-        msg_set_bearer_id(&tunnel_hdr, l_ptr->peer_bearer_id);
-        skb_queue_walk(&l_ptr->outqueue, skb) {
+        struct tipc_msg tnl_hdr;
+        struct sk_buff_head *queue = &link->transmq;
+        int mcnt;
+
+        tipc_msg_init(link_own_addr(link), &tnl_hdr, TUNNEL_PROTOCOL,
+                      SYNCH_MSG, INT_H_SIZE, link->addr);
+        mcnt = skb_queue_len(&link->transmq) + skb_queue_len(&link->backlogq);
+        msg_set_msgcnt(&tnl_hdr, mcnt);
+        msg_set_bearer_id(&tnl_hdr, link->peer_bearer_id);
+
+tunnel_queue:
+        skb_queue_walk(queue, skb) {
                 struct sk_buff *outskb;
                 struct tipc_msg *msg = buf_msg(skb);
-                u32 length = msg_size(msg);
+                u32 len = msg_size(msg);
 
-                if (msg_user(msg) == MSG_BUNDLER)
-                        msg_set_type(msg, CLOSED_MSG);
-                msg_set_ack(msg, mod(l_ptr->next_in_no - 1));   /* Update */
-                msg_set_bcast_ack(msg, l_ptr->owner->bclink.last_in);
-                msg_set_size(&tunnel_hdr, length + INT_H_SIZE);
-                outskb = tipc_buf_acquire(length + INT_H_SIZE);
+                msg_set_ack(msg, mod(link->next_in_no - 1));
+                msg_set_bcast_ack(msg, link->owner->bclink.last_in);
+                msg_set_size(&tnl_hdr, len + INT_H_SIZE);
+                outskb = tipc_buf_acquire(len + INT_H_SIZE);
                 if (outskb == NULL) {
                         pr_warn("%sunable to send duplicate msg\n",
                                 link_co_err);
                         return;
                 }
-                skb_copy_to_linear_data(outskb, &tunnel_hdr, INT_H_SIZE);
-                skb_copy_to_linear_data_offset(outskb, INT_H_SIZE, skb->data,
-                                               length);
-                __tipc_link_xmit_skb(tunnel, outskb);
-                if (!tipc_link_is_up(l_ptr))
+                skb_copy_to_linear_data(outskb, &tnl_hdr, INT_H_SIZE);
+                skb_copy_to_linear_data_offset(outskb, INT_H_SIZE,
+                                               skb->data, len);
+                __tipc_link_xmit_skb(tnl, outskb);
+                if (!tipc_link_is_up(link))
                         return;
         }
-}
-
-/**
- * buf_extract - extracts embedded TIPC message from another message
- * @skb: encapsulating message buffer
- * @from_pos: offset to extract from
- *
- * Returns a new message buffer containing an embedded message.  The
- * encapsulating buffer is left unchanged.
- */
-static struct sk_buff *buf_extract(struct sk_buff *skb, u32 from_pos)
-{
-        struct tipc_msg *msg = (struct tipc_msg *)(skb->data + from_pos);
-        u32 size = msg_size(msg);
-        struct sk_buff *eb;
-
-        eb = tipc_buf_acquire(size);
-        if (eb)
-                skb_copy_to_linear_data(eb, msg, size);
-        return eb;
-}
-
-/* tipc_link_dup_rcv(): Receive a tunnelled DUPLICATE_MSG packet.
- * Owner node is locked.
- */
-static void tipc_link_dup_rcv(struct tipc_link *l_ptr,
-                              struct sk_buff *t_buf)
-{
-        struct sk_buff *buf;
-
-        if (!tipc_link_is_up(l_ptr))
+        if (queue == &link->backlogq)
                 return;
-
-        buf = buf_extract(t_buf, INT_H_SIZE);
-        if (buf == NULL) {
-                pr_warn("%sfailed to extract inner dup pkt\n", link_co_err);
-                return;
-        }
-
-        /* Add buffer to deferred queue, if applicable: */
-        link_handle_out_of_seq_msg(l_ptr, buf);
+        queue = &link->backlogq;
+        goto tunnel_queue;
 }
 
-/*  tipc_link_failover_rcv(): Receive a tunnelled ORIGINAL_MSG packet
+/*  tipc_link_failover_rcv(): Receive a tunnelled FAILOVER_MSG packet
  *  Owner node is locked.
  */
-static struct sk_buff *tipc_link_failover_rcv(struct tipc_link *l_ptr,
-                                              struct sk_buff *t_buf)
+static bool tipc_link_failover_rcv(struct tipc_link *link,
+                                   struct sk_buff **skb)
 {
-        struct tipc_msg *t_msg = buf_msg(t_buf);
-        struct sk_buff *buf = NULL;
-        struct tipc_msg *msg;
-
-        if (tipc_link_is_up(l_ptr))
-                tipc_link_reset(l_ptr);
-
-        /* First failover packet? */
-        if (l_ptr->exp_msg_count == START_CHANGEOVER)
-                l_ptr->exp_msg_count = msg_msgcnt(t_msg);
-
-        /* Should there be an inner packet? */
-        if (l_ptr->exp_msg_count) {
-                l_ptr->exp_msg_count--;
-                buf = buf_extract(t_buf, INT_H_SIZE);
-                if (buf == NULL) {
-                        pr_warn("%sno inner failover pkt\n", link_co_err);
-                        goto exit;
-                }
-                msg = buf_msg(buf);
+        struct tipc_msg *msg = buf_msg(*skb);
+        struct sk_buff *iskb = NULL;
+        struct tipc_link *pl = NULL;
+        int bearer_id = msg_bearer_id(msg);
+        int pos = 0;
 
-                if (less(msg_seqno(msg), l_ptr->reset_checkpoint)) {
-                        kfree_skb(buf);
-                        buf = NULL;
-                        goto exit;
-                }
-                if (msg_user(msg) == MSG_FRAGMENTER) {
-                        l_ptr->stats.recv_fragments++;
-                        tipc_buf_append(&l_ptr->reasm_buf, &buf);
-                }
+        if (msg_type(msg) != FAILOVER_MSG) {
+                pr_warn("%sunknown tunnel pkt received\n", link_co_err);
+                goto exit;
         }
-exit:
-        if ((!l_ptr->exp_msg_count) && (l_ptr->flags & LINK_STOPPED))
-                tipc_link_delete(l_ptr);
-        return buf;
-}
+        if (bearer_id >= MAX_BEARERS)
+                goto exit;
 
-/*  tipc_link_tunnel_rcv(): Receive a tunnelled packet, sent
- *  via other link as result of a failover (ORIGINAL_MSG) or
- *  a new active link (DUPLICATE_MSG). Failover packets are
- *  returned to the active link for delivery upwards.
- *  Owner node is locked.
- */
-static int tipc_link_tunnel_rcv(struct tipc_node *n_ptr,
-                                struct sk_buff **buf)
-{
-        struct sk_buff *t_buf = *buf;
-        struct tipc_link *l_ptr;
-        struct tipc_msg *t_msg = buf_msg(t_buf);
-        u32 bearer_id = msg_bearer_id(t_msg);
+        if (bearer_id == link->bearer_id)
+                goto exit;
 
-        *buf = NULL;
+        pl = link->owner->links[bearer_id];
+        if (pl && tipc_link_is_up(pl))
+                tipc_link_reset(pl);
 
-        if (bearer_id >= MAX_BEARERS)
+        if (link->failover_pkts == FIRST_FAILOVER)
+                link->failover_pkts = msg_msgcnt(msg);
+
+        /* Should we expect an inner packet? */
+        if (!link->failover_pkts)
                 goto exit;
 
-        l_ptr = n_ptr->links[bearer_id];
-        if (!l_ptr)
+        if (!tipc_msg_extract(*skb, &iskb, &pos)) {
+                pr_warn("%sno inner failover pkt\n", link_co_err);
+                *skb = NULL;
                 goto exit;
+        }
+        link->failover_pkts--;
+        *skb = NULL;
 
-        if (msg_type(t_msg) == DUPLICATE_MSG)
-                tipc_link_dup_rcv(l_ptr, t_buf);
-        else if (msg_type(t_msg) == ORIGINAL_MSG)
-                *buf = tipc_link_failover_rcv(l_ptr, t_buf);
-        else
-                pr_warn("%sunknown tunnel pkt received\n", link_co_err);
+        /* Was this packet already delivered? */
+        if (less(buf_seqno(iskb), link->failover_checkpt)) {
+                kfree_skb(iskb);
+                iskb = NULL;
+                goto exit;
+        }
+        if (msg_user(buf_msg(iskb)) == MSG_FRAGMENTER) {
+                link->stats.recv_fragments++;
+                tipc_buf_append(&link->failover_skb, &iskb);
+        }
 exit:
-        kfree_skb(t_buf);
-        return *buf != NULL;
+        if (!link->failover_pkts && pl)
+                pl->flags &= ~LINK_FAILINGOVER;
+        kfree_skb(*skb);
+        *skb = iskb;
+        return *skb;
 }
 
 static void link_set_supervision_props(struct tipc_link *l_ptr, u32 tol)
@@ -1900,23 +1749,16 @@ static void link_set_supervision_props(struct tipc_link *l_ptr, u32 tol)
         l_ptr->abort_limit = tol / (jiffies_to_msecs(l_ptr->cont_intv) / 4);
 }
 
-void tipc_link_set_queue_limits(struct tipc_link *l_ptr, u32 window)
+void tipc_link_set_queue_limits(struct tipc_link *l, u32 win)
 {
-        /* Data messages from this node, inclusive FIRST_FRAGM */
-        l_ptr->queue_limit[TIPC_LOW_IMPORTANCE] = window;
-        l_ptr->queue_limit[TIPC_MEDIUM_IMPORTANCE] = (window / 3) * 4;
-        l_ptr->queue_limit[TIPC_HIGH_IMPORTANCE] = (window / 3) * 5;
-        l_ptr->queue_limit[TIPC_CRITICAL_IMPORTANCE] = (window / 3) * 6;
-        /* Transiting data messages,inclusive FIRST_FRAGM */
-        l_ptr->queue_limit[TIPC_LOW_IMPORTANCE + 4] = 300;
-        l_ptr->queue_limit[TIPC_MEDIUM_IMPORTANCE + 4] = 600;
-        l_ptr->queue_limit[TIPC_HIGH_IMPORTANCE + 4] = 900;
-        l_ptr->queue_limit[TIPC_CRITICAL_IMPORTANCE + 4] = 1200;
-        l_ptr->queue_limit[CONN_MANAGER] = 1200;
-        l_ptr->queue_limit[CHANGEOVER_PROTOCOL] = 2500;
-        l_ptr->queue_limit[NAME_DISTRIBUTOR] = 3000;
-        /* FRAGMENT and LAST_FRAGMENT packets */
-        l_ptr->queue_limit[MSG_FRAGMENTER] = 4000;
+        int max_bulk = TIPC_MAX_PUBLICATIONS / (l->mtu / ITEM_SIZE);
+
+        l->window = win;
+        l->backlog[TIPC_LOW_IMPORTANCE].limit      = win / 2;
+        l->backlog[TIPC_MEDIUM_IMPORTANCE].limit   = win;
+        l->backlog[TIPC_HIGH_IMPORTANCE].limit     = win / 2 * 3;
+        l->backlog[TIPC_CRITICAL_IMPORTANCE].limit = win * 2;
+        l->backlog[TIPC_SYSTEM_IMPORTANCE].limit   = max_bulk;
 }
 
 /* tipc_link_find_owner - locate owner node of link by link's name
@@ -2081,14 +1923,14 @@ int tipc_nl_link_set(struct sk_buff *skb, struct genl_info *info)
 
                         tol = nla_get_u32(props[TIPC_NLA_PROP_TOL]);
                         link_set_supervision_props(link, tol);
-                        tipc_link_proto_xmit(link, STATE_MSG, 0, 0, tol, 0, 0);
+                        tipc_link_proto_xmit(link, STATE_MSG, 0, 0, tol, 0);
                 }
                 if (props[TIPC_NLA_PROP_PRIO]) {
                         u32 prio;
 
                         prio = nla_get_u32(props[TIPC_NLA_PROP_PRIO]);
                         link->priority = prio;
-                        tipc_link_proto_xmit(link, STATE_MSG, 0, 0, 0, prio, 0);
+                        tipc_link_proto_xmit(link, STATE_MSG, 0, 0, 0, prio);
                 }
                 if (props[TIPC_NLA_PROP_WIN]) {
                         u32 win;
@@ -2193,7 +2035,7 @@ static int __tipc_nl_add_link(struct net *net, struct tipc_nl_msg *msg,
         if (nla_put_u32(msg->skb, TIPC_NLA_LINK_DEST,
                         tipc_cluster_mask(tn->own_addr)))
                 goto attr_msg_full;
-        if (nla_put_u32(msg->skb, TIPC_NLA_LINK_MTU, link->max_pkt))
+        if (nla_put_u32(msg->skb, TIPC_NLA_LINK_MTU, link->mtu))
                 goto attr_msg_full;
         if (nla_put_u32(msg->skb, TIPC_NLA_LINK_RX, link->next_in_no))
                 goto attr_msg_full;
@@ -2215,7 +2057,7 @@ static int __tipc_nl_add_link(struct net *net, struct tipc_nl_msg *msg,
         if (nla_put_u32(msg->skb, TIPC_NLA_PROP_TOL, link->tolerance))
                 goto prop_msg_full;
         if (nla_put_u32(msg->skb, TIPC_NLA_PROP_WIN,
-                        link->queue_limit[TIPC_LOW_IMPORTANCE]))
+                        link->window))
                 goto prop_msg_full;
         if (nla_put_u32(msg->skb, TIPC_NLA_PROP_PRIO, link->priority))
                 goto prop_msg_full;
@@ -2281,7 +2123,6 @@ int tipc_nl_link_dump(struct sk_buff *skb, struct netlink_callback *cb)
         msg.seq = cb->nlh->nlmsg_seq;
 
         rcu_read_lock();
-
         if (prev_node) {
                 node = tipc_node_find(net, prev_node);
                 if (!node) {
@@ -2294,6 +2135,7 @@ int tipc_nl_link_dump(struct sk_buff *skb, struct netlink_callback *cb)
                         cb->prev_seq = 1;
                         goto out;
                 }
+                tipc_node_put(node);
 
                 list_for_each_entry_continue_rcu(node, &tn->node_list,
                                                  list) {
@@ -2301,6 +2143,7 @@ int tipc_nl_link_dump(struct sk_buff *skb, struct netlink_callback *cb)
                         err = __tipc_nl_add_node_links(net, &msg, node,
                                                        &prev_link);
                         tipc_node_unlock(node);
+                        tipc_node_put(node);
                         if (err)
                                 goto out;