From 9d13ec65ede775f896c3da1cfa35283afe2f796c Mon Sep 17 00:00:00 2001
From: Jon Paul Maloy <jon.maloy@ericsson.com>
Date: Thu, 16 Jul 2015 16:54:19 -0400
Subject: tipc: introduce link entry structure to struct tipc_node

struct 'tipc_node' currently contains two arrays for link attributes,
one for the link pointers, and one for the usable link MTUs.

We now group those into a new struct 'tipc_link_entry', and intoduce
one single array consisting of such enties. Apart from being a cosmetic
improvement, this is a starting point for the strict master-slave
relation between node and link that we will introduce in the following
commits.

Reviewed-by: Ying Xue <ying.xue@windriver.com>
Signed-off-by: Jon Maloy <jon.maloy@ericsson.com>
Signed-off-by: David S. Miller <davem@davemloft.net>
---
 net/tipc/link.c | 60 ++++++++++++++++++++++++++++++---------------------------
 1 file changed, 32 insertions(+), 28 deletions(-)

(limited to 'net/tipc/link.c')

diff --git a/net/tipc/link.c b/net/tipc/link.c
index eaa9fe54b4ae..03372a7e98df 100644
--- a/net/tipc/link.c
+++ b/net/tipc/link.c
@@ -132,9 +132,11 @@ static void tipc_link_put(struct tipc_link *l_ptr)
 
 static struct tipc_link *tipc_parallel_link(struct tipc_link *l)
 {
-	if (l->owner->active_links[0] != l)
-		return l->owner->active_links[0];
-	return l->owner->active_links[1];
+	struct tipc_node *n = l->owner;
+
+	if (node_active_link(n, 0) != l)
+		return node_active_link(n, 0);
+	return node_active_link(n, 1);
 }
 
 /*
@@ -147,10 +149,11 @@ int tipc_link_is_up(struct tipc_link *l_ptr)
 	return link_working_working(l_ptr) || link_working_unknown(l_ptr);
 }
 
-int tipc_link_is_active(struct tipc_link *l_ptr)
+int tipc_link_is_active(struct tipc_link *l)
 {
-	return	(l_ptr->owner->active_links[0] == l_ptr) ||
-		(l_ptr->owner->active_links[1] == l_ptr);
+	struct tipc_node *n = l->owner;
+
+	return (node_active_link(n, 0) == l) || (node_active_link(n, 1) == l);
 }
 
 /**
@@ -240,7 +243,7 @@ struct tipc_link *tipc_link_create(struct tipc_node *n_ptr,
 		return NULL;
 	}
 
-	if (n_ptr->links[b_ptr->identity]) {
+	if (n_ptr->links[b_ptr->identity].link) {
 		tipc_addr_string_fill(addr_string, n_ptr->addr);
 		pr_err("Attempt to establish second link on <%s> to %s\n",
 		       b_ptr->name, addr_string);
@@ -321,7 +324,7 @@ void tipc_link_delete_list(struct net *net, unsigned int bearer_id)
 	rcu_read_lock();
 	list_for_each_entry_rcu(node, &tn->node_list, list) {
 		tipc_node_lock(node);
-		link = node->links[bearer_id];
+		link = node->links[bearer_id].link;
 		if (link)
 			tipc_link_delete(link);
 		tipc_node_unlock(node);
@@ -446,7 +449,7 @@ void tipc_link_reset(struct tipc_link *l_ptr)
 	if ((prev_state == RESET_UNKNOWN) || (prev_state == RESET_RESET))
 		return;
 
-	tipc_node_link_down(l_ptr->owner, l_ptr);
+	tipc_node_link_down(l_ptr->owner, l_ptr->bearer_id);
 	tipc_bearer_remove_dest(owner->net, l_ptr->bearer_id, l_ptr->addr);
 
 	if (was_active_link && tipc_node_is_up(l_ptr->owner) && (pl != l_ptr)) {
@@ -482,7 +485,7 @@ static void link_activate(struct tipc_link *link)
 	link->rcv_nxt = 1;
 	link->stats.recv_info = 1;
 	link->silent_intv_cnt = 0;
-	tipc_node_link_up(node, link);
+	tipc_node_link_up(node, link->bearer_id);
 	tipc_bearer_add_dest(node->net, link->bearer_id, link->addr);
 }
 
@@ -577,7 +580,7 @@ static void link_state_event(struct tipc_link *l_ptr, unsigned int event)
 		case TRAFFIC_MSG_EVT:
 			break;
 		case ACTIVATE_MSG:
-			other = l_ptr->owner->active_links[0];
+			other = node_active_link(l_ptr->owner, 0);
 			if (other && link_working_unknown(other))
 				break;
 			l_ptr->state = WORKING_WORKING;
@@ -606,7 +609,7 @@ static void link_state_event(struct tipc_link *l_ptr, unsigned int event)
 		switch (event) {
 		case TRAFFIC_MSG_EVT:
 		case ACTIVATE_MSG:
-			other = l_ptr->owner->active_links[0];
+			other = node_active_link(l_ptr->owner, 0);
 			if (other && link_working_unknown(other))
 				break;
 			l_ptr->state = WORKING_WORKING;
@@ -755,7 +758,7 @@ int tipc_link_xmit(struct net *net, struct sk_buff_head *list, u32 dnode,
 	node = tipc_node_find(net, dnode);
 	if (node) {
 		tipc_node_lock(node);
-		link = node->active_links[selector & 1];
+		link = node_active_link(node, selector & 1);
 		if (link)
 			rc = __tipc_link_xmit(net, link, list);
 		tipc_node_unlock(node);
@@ -858,9 +861,9 @@ void tipc_link_reset_all(struct tipc_node *node)
 		tipc_addr_string_fill(addr_string, node->addr));
 
 	for (i = 0; i < MAX_BEARERS; i++) {
-		if (node->links[i]) {
-			link_print(node->links[i], "Resetting link\n");
-			tipc_link_reset(node->links[i]);
+		if (node->links[i].link) {
+			link_print(node->links[i].link, "Resetting link\n");
+			tipc_link_reset(node->links[i].link);
 		}
 	}
 
@@ -1029,7 +1032,7 @@ void tipc_rcv(struct net *net, struct sk_buff *skb, struct tipc_bearer *b_ptr)
 
 		tipc_node_lock(n_ptr);
 		/* Locate unicast link endpoint that should handle message */
-		l_ptr = n_ptr->links[b_ptr->identity];
+		l_ptr = n_ptr->links[b_ptr->identity].link;
 		if (unlikely(!l_ptr))
 			goto unlock;
 
@@ -1496,7 +1499,7 @@ static void tipc_link_tunnel_xmit(struct tipc_link *l_ptr,
 	struct sk_buff *skb;
 	u32 length = msg_size(msg);
 
-	tunnel = l_ptr->owner->active_links[selector & 1];
+	tunnel = node_active_link(l_ptr->owner, selector & 1);
 	if (!tipc_link_is_up(tunnel)) {
 		pr_warn("%stunnel link no longer available\n", link_co_err);
 		return;
@@ -1522,7 +1525,7 @@ static void tipc_link_tunnel_xmit(struct tipc_link *l_ptr,
 void tipc_link_failover_send_queue(struct tipc_link *l_ptr)
 {
 	int msgcount;
-	struct tipc_link *tunnel = l_ptr->owner->active_links[0];
+	struct tipc_link *tunnel = node_active_link(l_ptr->owner, 0);
 	struct tipc_msg tunnel_hdr;
 	struct sk_buff *skb;
 	int split_bundles;
@@ -1556,8 +1559,8 @@ void tipc_link_failover_send_queue(struct tipc_link *l_ptr)
 		return;
 	}
 
-	split_bundles = (l_ptr->owner->active_links[0] !=
-			 l_ptr->owner->active_links[1]);
+	split_bundles = (node_active_link(l_ptr->owner, 0) !=
+			 node_active_link(l_ptr->owner, 0));
 
 	skb_queue_walk(&l_ptr->transmq, skb) {
 		struct tipc_msg *msg = buf_msg(skb);
@@ -1660,7 +1663,7 @@ static bool tipc_link_failover_rcv(struct tipc_link *link,
 	if (bearer_id == link->bearer_id)
 		goto exit;
 
-	pl = link->owner->links[bearer_id];
+	pl = link->owner->links[bearer_id].link;
 	if (pl && tipc_link_is_up(pl))
 		tipc_link_reset(pl);
 
@@ -1743,7 +1746,7 @@ static struct tipc_node *tipc_link_find_owner(struct net *net,
 	list_for_each_entry_rcu(n_ptr, &tn->node_list, list) {
 		tipc_node_lock(n_ptr);
 		for (i = 0; i < MAX_BEARERS; i++) {
-			l_ptr = n_ptr->links[i];
+			l_ptr = n_ptr->links[i].link;
 			if (l_ptr && !strcmp(l_ptr->name, link_name)) {
 				*bearer_id = i;
 				found_node = n_ptr;
@@ -1865,7 +1868,7 @@ int tipc_nl_link_set(struct sk_buff *skb, struct genl_info *info)
 
 	tipc_node_lock(node);
 
-	link = node->links[bearer_id];
+	link = node->links[bearer_id].link;
 	if (!link) {
 		res = -EINVAL;
 		goto out;
@@ -2055,10 +2058,11 @@ static int __tipc_nl_add_node_links(struct net *net, struct tipc_nl_msg *msg,
 	for (i = *prev_link; i < MAX_BEARERS; i++) {
 		*prev_link = i;
 
-		if (!node->links[i])
+		if (!node->links[i].link)
 			continue;
 
-		err = __tipc_nl_add_link(net, msg, node->links[i], NLM_F_MULTI);
+		err = __tipc_nl_add_link(net, msg,
+					 node->links[i].link, NLM_F_MULTI);
 		if (err)
 			return err;
 	}
@@ -2172,7 +2176,7 @@ int tipc_nl_link_get(struct sk_buff *skb, struct genl_info *info)
 			return -EINVAL;
 
 		tipc_node_lock(node);
-		link = node->links[bearer_id];
+		link = node->links[bearer_id].link;
 		if (!link) {
 			tipc_node_unlock(node);
 			nlmsg_free(msg.skb);
@@ -2227,7 +2231,7 @@ int tipc_nl_link_reset_stats(struct sk_buff *skb, struct genl_info *info)
 
 	tipc_node_lock(node);
 
-	link = node->links[bearer_id];
+	link = node->links[bearer_id].link;
 	if (!link) {
 		tipc_node_unlock(node);
 		return -EINVAL;
-- 
cgit v1.2.3


From d39bbd445dc44259c77bbbc8aadcce7dcdba39cc Mon Sep 17 00:00:00 2001
From: Jon Paul Maloy <jon.maloy@ericsson.com>
Date: Thu, 16 Jul 2015 16:54:21 -0400
Subject: tipc: move link input queue to tipc_node

At present, the link input queue and the name distributor receive
queues are fields aggregated in struct tipc_link. This is a hazard,
because a link might be deleted while a receiving socket still keeps
reference to one of the queues.

This commit fixes this bug. However, rather than adding yet another
reference counter to the critical data path, we move the two queues
to safe ground inside struct tipc_node, which is already protected, and
let the link code only handle references to the queues. This is also
in line with planned later changes in this area.

Reviewed-by: Ying Xue <ying.xue@windriver.com>
Signed-off-by: Jon Maloy <jon.maloy@ericsson.com>
Signed-off-by: David S. Miller <davem@davemloft.net>
---
 net/tipc/link.c | 27 +++++++++++++++------------
 1 file changed, 15 insertions(+), 12 deletions(-)

(limited to 'net/tipc/link.c')

diff --git a/net/tipc/link.c b/net/tipc/link.c
index 03372a7e98df..f8e0e2ceceb4 100644
--- a/net/tipc/link.c
+++ b/net/tipc/link.c
@@ -227,7 +227,9 @@ static void link_set_timer(struct tipc_link *link, unsigned long time)
  */
 struct tipc_link *tipc_link_create(struct tipc_node *n_ptr,
 				   struct tipc_bearer *b_ptr,
-				   const struct tipc_media_addr *media_addr)
+				   const struct tipc_media_addr *media_addr,
+				   struct sk_buff_head *inputq,
+				   struct sk_buff_head *namedq)
 {
 	struct tipc_net *tn = net_generic(n_ptr->net, tipc_net_id);
 	struct tipc_link *l_ptr;
@@ -289,8 +291,9 @@ struct tipc_link *tipc_link_create(struct tipc_node *n_ptr,
 	__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);
+	l_ptr->inputq = inputq;
+	l_ptr->namedq = namedq;
+	skb_queue_head_init(l_ptr->inputq);
 	link_reset_statistics(l_ptr);
 	tipc_node_attach_link(n_ptr, l_ptr);
 	setup_timer(&l_ptr->timer, link_timeout, (unsigned long)l_ptr);
@@ -391,8 +394,8 @@ void link_prepare_wakeup(struct tipc_link *l)
 		if ((pnd[imp] + l->backlog[imp].len) >= lim)
 			break;
 		skb_unlink(skb, &l->wakeupq);
-		skb_queue_tail(&l->inputq, skb);
-		l->owner->inputq = &l->inputq;
+		skb_queue_tail(l->inputq, skb);
+		l->owner->inputq = l->inputq;
 		l->owner->action_flags |= TIPC_MSG_EVT;
 	}
 }
@@ -465,7 +468,7 @@ void tipc_link_reset(struct tipc_link *l_ptr)
 	__skb_queue_purge(&l_ptr->transmq);
 	__skb_queue_purge(&l_ptr->deferdq);
 	if (!owner->inputq)
-		owner->inputq = &l_ptr->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;
@@ -962,7 +965,7 @@ static bool link_synch(struct tipc_link *l)
 
 	/* Is it still in the input queue ? */
 	post_synch = mod(pl->rcv_nxt - l->synch_point) - 1;
-	if (skb_queue_len(&pl->inputq) > post_synch)
+	if (skb_queue_len(pl->inputq) > post_synch)
 		return false;
 synched:
 	l->flags &= ~LINK_SYNCHING;
@@ -1141,16 +1144,16 @@ static bool tipc_data_input(struct tipc_link *link, struct sk_buff *skb)
 	case TIPC_HIGH_IMPORTANCE:
 	case TIPC_CRITICAL_IMPORTANCE:
 	case CONN_MANAGER:
-		if (tipc_skb_queue_tail(&link->inputq, skb, dport)) {
-			node->inputq = &link->inputq;
+		if (tipc_skb_queue_tail(link->inputq, skb, dport)) {
+			node->inputq = link->inputq;
 			node->action_flags |= TIPC_MSG_EVT;
 		}
 		return true;
 	case NAME_DISTRIBUTOR:
 		node->bclink.recv_permitted = true;
-		node->namedq = &link->namedq;
-		skb_queue_tail(&link->namedq, skb);
-		if (skb_queue_len(&link->namedq) == 1)
+		node->namedq = link->namedq;
+		skb_queue_tail(link->namedq, skb);
+		if (skb_queue_len(link->namedq) == 1)
 			node->action_flags |= TIPC_NAMED_MSG_EVT;
 		return true;
 	case MSG_BUNDLER:
-- 
cgit v1.2.3


From 22d85c79428b8ca9a01623aa3e3a1fe29a30a119 Mon Sep 17 00:00:00 2001
From: Jon Paul Maloy <jon.maloy@ericsson.com>
Date: Thu, 16 Jul 2015 16:54:23 -0400
Subject: tipc: change sk_buffer handling in tipc_link_xmit()

When the function tipc_link_xmit() is given a buffer list for
transmission, it currently consumes the list both when transmission
is successful and when it fails, except for the special case when
it encounters link congestion.

This behavior is inconsistent, and needs to be corrected if we want
to avoid problems in later commits in this series.

In this commit, we change this to let the function consume the list
only when transmission is successful, and leave the list with the
sender in all other cases. We also modifiy the socket code so that
it adapts to this change, i.e., purges the list when a non-congestion
error code is returned.

Reviewed-by: Ying Xue <ying.xue@windriver.com>
Signed-off-by: Jon Maloy <jon.maloy@ericsson.com>
Signed-off-by: David S. Miller <davem@davemloft.net>
---
 net/tipc/link.c | 23 +++++++++--------------
 1 file changed, 9 insertions(+), 14 deletions(-)

(limited to 'net/tipc/link.c')

diff --git a/net/tipc/link.c b/net/tipc/link.c
index f8e0e2ceceb4..ea32679b6737 100644
--- a/net/tipc/link.c
+++ b/net/tipc/link.c
@@ -340,7 +340,7 @@ void tipc_link_delete_list(struct net *net, unsigned int bearer_id)
  * @link: congested link
  * @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
+ * Does not consume buffer list
  */
 static int link_schedule_user(struct tipc_link *link, struct sk_buff_head *list)
 {
@@ -354,7 +354,7 @@ static int link_schedule_user(struct tipc_link *link, struct sk_buff_head *list)
 	if (unlikely(imp > TIPC_CRITICAL_IMPORTANCE)) {
 		pr_warn("%s<%s>, send queue full", link_rst_msg, link->name);
 		tipc_link_reset(link);
-		goto err;
+		return -ENOBUFS;
 	}
 	/* Non-blocking sender: */
 	if (TIPC_SKB_CB(skb_peek(list))->wakeup_pending)
@@ -364,15 +364,12 @@ static int link_schedule_user(struct tipc_link *link, struct sk_buff_head *list)
 	skb = tipc_msg_create(SOCK_WAKEUP, 0, INT_H_SIZE, 0,
 			      addr, addr, oport, 0, 0);
 	if (!skb)
-		goto err;
+		return -ENOBUFS;
 	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 -ELINKCONG;
-err:
-	__skb_queue_purge(list);
-	return -ENOBUFS;
 }
 
 /**
@@ -641,8 +638,7 @@ static void link_state_event(struct tipc_link *l_ptr, unsigned int event)
  * @link: link to use
  * @list: chain of buffers containing message
  *
- * Consumes the buffer chain, except when returning -ELINKCONG,
- * since the caller then may want to make more send attempts.
+ * Consumes the buffer chain, except when returning an error code,
  * Returns 0 if success, or errno: -ELINKCONG, -EMSGSIZE or -ENOBUFS
  * Messages at TIPC_SYSTEM_IMPORTANCE are always accepted
  */
@@ -666,10 +662,9 @@ int __tipc_link_xmit(struct net *net, struct tipc_link *link,
 		if (unlikely(link->backlog[i].len >= link->backlog[i].limit))
 			return link_schedule_user(link, list);
 	}
-	if (unlikely(msg_size(msg) > mtu)) {
-		__skb_queue_purge(list);
+	if (unlikely(msg_size(msg) > mtu))
 		return -EMSGSIZE;
-	}
+
 	/* Prepare each packet for sending, and add to relevant queue: */
 	while (skb_queue_len(list)) {
 		skb = skb_peek(list);
@@ -722,7 +717,7 @@ static int __tipc_link_xmit_skb(struct tipc_link *link, struct sk_buff *skb)
 
 /* 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
+ * messages, which will not cause link congestion
  * 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
@@ -736,7 +731,7 @@ int tipc_link_xmit_skb(struct net *net, struct sk_buff *skb, u32 dnode,
 
 	skb2list(skb, &head);
 	rc = tipc_link_xmit(net, &head, dnode, selector);
-	if (rc == -ELINKCONG)
+	if (rc)
 		kfree_skb(skb);
 	return 0;
 }
@@ -748,7 +743,7 @@ int tipc_link_xmit_skb(struct net *net, struct sk_buff *skb, u32 dnode,
  * @dsz: amount of user data to be sent
  * @dnode: address of destination node
  * @selector: a number used for deterministic link selection
- * Consumes the buffer chain, except when returning -ELINKCONG
+ * Consumes the buffer chain, except when returning error
  * Returns 0 if success, otherwise errno: -ELINKCONG,-EHOSTUNREACH,-EMSGSIZE
  */
 int tipc_link_xmit(struct net *net, struct sk_buff_head *list, u32 dnode,
-- 
cgit v1.2.3


From af9b028e270fda6fb812d70d17d902297df1ceb5 Mon Sep 17 00:00:00 2001
From: Jon Paul Maloy <jon.maloy@ericsson.com>
Date: Thu, 16 Jul 2015 16:54:24 -0400
Subject: tipc: make media xmit call outside node spinlock context

Currently, message sending is performed through a deep call chain,
where the node spinlock is grabbed and held during a significant
part of the transmission time. This is clearly detrimental to
overall throughput performance; it would be better if we could send
the message after the spinlock has been released.

In this commit, we do instead let the call revert on the stack after
the buffer chain has been added to the transmission queue, whereafter
clones of the buffers are transmitted to the device layer outside the
spinlock scope.

As a further step in our effort to separate the roles of the node
and link entities we also move the function tipc_link_xmit() to
node.c, and rename it to tipc_node_xmit().

Reviewed-by: Ying Xue <ying.xue@windriver.com>
Signed-off-by: Jon Maloy <jon.maloy@ericsson.com>
Signed-off-by: David S. Miller <davem@davemloft.net>
---
 net/tipc/link.c | 132 ++++++++++++++++++++++++++++++--------------------------
 1 file changed, 72 insertions(+), 60 deletions(-)

(limited to 'net/tipc/link.c')

diff --git a/net/tipc/link.c b/net/tipc/link.c
index ea32679b6737..c052437a7cfa 100644
--- a/net/tipc/link.c
+++ b/net/tipc/link.c
@@ -353,7 +353,6 @@ static int link_schedule_user(struct tipc_link *link, struct sk_buff_head *list)
 	/* 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);
 		return -ENOBUFS;
 	}
 	/* Non-blocking sender: */
@@ -701,6 +700,78 @@ int __tipc_link_xmit(struct net *net, struct tipc_link *link,
 	return 0;
 }
 
+/**
+ * tipc_link_xmit(): enqueue buffer list according to queue situation
+ * @link: link to use
+ * @list: chain of buffers containing message
+ * @xmitq: returned list of packets to be sent by caller
+ *
+ * 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 tipc_link *l, struct sk_buff_head *list,
+		   struct sk_buff_head *xmitq)
+{
+	struct tipc_msg *hdr = buf_msg(skb_peek(list));
+	unsigned int maxwin = l->window;
+	unsigned int i, imp = msg_importance(hdr);
+	unsigned int mtu = l->mtu;
+	u16 ack = l->rcv_nxt - 1;
+	u16 seqno = l->snd_nxt;
+	u16 bc_last_in = l->owner->bclink.last_in;
+	struct sk_buff_head *transmq = &l->transmq;
+	struct sk_buff_head *backlogq = &l->backlogq;
+	struct sk_buff *skb, *_skb, *bskb;
+
+	/* Match msg importance against this and all higher backlog limits: */
+	for (i = imp; i <= TIPC_SYSTEM_IMPORTANCE; i++) {
+		if (unlikely(l->backlog[i].len >= l->backlog[i].limit))
+			return link_schedule_user(l, list);
+	}
+	if (unlikely(msg_size(hdr) > mtu))
+		return -EMSGSIZE;
+
+	/* Prepare each packet for sending, and add to relevant queue: */
+	while (skb_queue_len(list)) {
+		skb = skb_peek(list);
+		hdr = buf_msg(skb);
+		msg_set_seqno(hdr, seqno);
+		msg_set_ack(hdr, ack);
+		msg_set_bcast_ack(hdr, bc_last_in);
+
+		if (likely(skb_queue_len(transmq) < maxwin)) {
+			_skb = skb_clone(skb, GFP_ATOMIC);
+			if (!_skb)
+				return -ENOBUFS;
+			__skb_dequeue(list);
+			__skb_queue_tail(transmq, skb);
+			__skb_queue_tail(xmitq, _skb);
+			l->rcv_unacked = 0;
+			seqno++;
+			continue;
+		}
+		if (tipc_msg_bundle(skb_peek_tail(backlogq), hdr, mtu)) {
+			kfree_skb(__skb_dequeue(list));
+			l->stats.sent_bundled++;
+			continue;
+		}
+		if (tipc_msg_make_bundle(&bskb, hdr, mtu, l->addr)) {
+			kfree_skb(__skb_dequeue(list));
+			__skb_queue_tail(backlogq, bskb);
+			l->backlog[msg_importance(buf_msg(bskb))].len++;
+			l->stats.sent_bundled++;
+			l->stats.sent_bundles++;
+			continue;
+		}
+		l->backlog[imp].len += skb_queue_len(list);
+		skb_queue_splice_tail_init(list, backlogq);
+	}
+	l->snd_nxt = seqno;
+	return 0;
+}
+
 static void skb2list(struct sk_buff *skb, struct sk_buff_head *list)
 {
 	skb_queue_head_init(list);
@@ -715,65 +786,6 @@ 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 cause link congestion
- * 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);
-	rc = tipc_link_xmit(net, &head, dnode, selector);
-	if (rc)
-		kfree_skb(skb);
-	return 0;
-}
-
-/**
- * tipc_link_xmit() is the general link level function for message sending
- * @net: the applicable net namespace
- * @list: chain of buffers containing message
- * @dsz: amount of user data to be sent
- * @dnode: address of destination node
- * @selector: a number used for deterministic link selection
- * Consumes the buffer chain, except when returning error
- * Returns 0 if success, otherwise errno: -ELINKCONG,-EHOSTUNREACH,-EMSGSIZE
- */
-int tipc_link_xmit(struct net *net, struct sk_buff_head *list, u32 dnode,
-		   u32 selector)
-{
-	struct tipc_link *link = NULL;
-	struct tipc_node *node;
-	int rc = -EHOSTUNREACH;
-
-	node = tipc_node_find(net, dnode);
-	if (node) {
-		tipc_node_lock(node);
-		link = node_active_link(node, selector & 1);
-		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))) {
-		tipc_sk_rcv(net, list);
-		return 0;
-	}
-
-	__skb_queue_purge(list);
-	return rc;
-}
-
 /*
  * tipc_link_sync_xmit - synchronize broadcast link endpoints.
  *
-- 
cgit v1.2.3


From d3504c3449fead545e5254bfb11da916f72c4734 Mon Sep 17 00:00:00 2001
From: Jon Paul Maloy <jon.maloy@ericsson.com>
Date: Thu, 16 Jul 2015 16:54:25 -0400
Subject: tipc: clean up definitions and usage of link flags

The status flag LINK_STOPPED is not needed any more, since the
mechanism for delayed deletion of links has been removed.
Likewise, LINK_STARTED and LINK_START_EVT are unnecessary,
because we can just as well start the link timer directly from
inside tipc_link_create().

We eliminate these flags in this commit.

Instead of the above flags, we now introduce three new link modes,
TIPC_LINK_OPEN, TIPC_LINK_BLOCKED and TIPC_LINK_TUNNEL. The values
indicate whether, and in the case of TIPC_LINK_TUNNEL, which, messages
the link is allowed to receive in this state. TIPC_LINK_BLOCKED also
blocks timer-driven protocol messages to be sent out, and any change
to the link FSM. Since the modes are mutually exclusive, we convert
them to state values, and rename the 'flags' field in struct tipc_link
to 'exec_mode'.

Finally, we move the #defines for link FSM states and events from link.h
into enums inside the file link.c, which is the real usage scope of
these definitions.

Reviewed-by: Ying Xue <ying.xue@windriver.com>
Signed-off-by: Jon Maloy <jon.maloy@ericsson.com>
Signed-off-by: David S. Miller <davem@davemloft.net>
---
 net/tipc/link.c | 98 ++++++++++++++++++++++++++++++++++-----------------------
 1 file changed, 58 insertions(+), 40 deletions(-)

(limited to 'net/tipc/link.c')

diff --git a/net/tipc/link.c b/net/tipc/link.c
index c052437a7cfa..35a2da688db1 100644
--- a/net/tipc/link.c
+++ b/net/tipc/link.c
@@ -79,19 +79,49 @@ static const struct nla_policy tipc_nl_prop_policy[TIPC_NLA_PROP_MAX + 1] = {
 /*
  * Out-of-range value for link session numbers
  */
-#define INVALID_SESSION 0x10000
+#define WILDCARD_SESSION 0x10000
 
-/*
- * Link state events:
+/* State value stored in 'failover_pkts'
  */
-#define  STARTING_EVT    856384768	/* link processing trigger */
-#define  TRAFFIC_MSG_EVT 560815u	/* rx'd ??? */
-#define  SILENCE_EVT     560817u	/* timer dicovered silence from peer */
+#define FIRST_FAILOVER 0xffffu
 
-/*
- * State value stored in 'failover_pkts'
+/* Link FSM states and events:
  */
-#define FIRST_FAILOVER 0xffffu
+enum {
+	WORKING_WORKING,
+	WORKING_UNKNOWN,
+	RESET_RESET,
+	RESET_UNKNOWN
+};
+
+enum {
+	PEER_RESET_EVT    = RESET_MSG,
+	ACTIVATE_EVT      = ACTIVATE_MSG,
+	TRAFFIC_EVT,      /* Any other valid msg from peer */
+	SILENCE_EVT       /* Peer was silent during last timer interval*/
+};
+
+/* Link FSM state checking routines
+ */
+static int link_working_working(struct tipc_link *l)
+{
+	return l->state == WORKING_WORKING;
+}
+
+static int link_working_unknown(struct tipc_link *l)
+{
+	return l->state == WORKING_UNKNOWN;
+}
+
+static int link_reset_unknown(struct tipc_link *l)
+{
+	return l->state == RESET_UNKNOWN;
+}
+
+static int link_reset_reset(struct tipc_link *l)
+{
+	return l->state == RESET_RESET;
+}
 
 static void link_handle_out_of_seq_msg(struct tipc_link *link,
 				       struct sk_buff *skb);
@@ -268,7 +298,7 @@ struct tipc_link *tipc_link_create(struct tipc_node *n_ptr,
 		/* note: peer i/f name is updated by reset/activate message */
 	memcpy(&l_ptr->media_addr, media_addr, sizeof(*media_addr));
 	l_ptr->owner = n_ptr;
-	l_ptr->peer_session = INVALID_SESSION;
+	l_ptr->peer_session = WILDCARD_SESSION;
 	l_ptr->bearer_id = b_ptr->identity;
 	link_set_supervision_props(l_ptr, b_ptr->tolerance);
 	l_ptr->state = RESET_UNKNOWN;
@@ -297,8 +327,7 @@ struct tipc_link *tipc_link_create(struct tipc_node *n_ptr,
 	link_reset_statistics(l_ptr);
 	tipc_node_attach_link(n_ptr, l_ptr);
 	setup_timer(&l_ptr->timer, link_timeout, (unsigned long)l_ptr);
-	link_state_event(l_ptr, STARTING_EVT);
-
+	link_set_timer(l_ptr, l_ptr->keepalive_intv);
 	return l_ptr;
 }
 
@@ -311,7 +340,6 @@ void tipc_link_delete(struct tipc_link *l)
 	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);
@@ -438,7 +466,7 @@ void tipc_link_reset(struct tipc_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;
+	l_ptr->peer_session = WILDCARD_SESSION;
 
 	/* Prepare for renewed mtu size negotiation */
 	l_ptr->mtu = l_ptr->advertised_mtu;
@@ -452,7 +480,7 @@ void tipc_link_reset(struct tipc_link *l_ptr)
 	tipc_bearer_remove_dest(owner->net, l_ptr->bearer_id, l_ptr->addr);
 
 	if (was_active_link && tipc_node_is_up(l_ptr->owner) && (pl != l_ptr)) {
-		l_ptr->flags |= LINK_FAILINGOVER;
+		l_ptr->exec_mode = TIPC_LINK_BLOCKED;
 		l_ptr->failover_checkpt = l_ptr->rcv_nxt;
 		pl->failover_pkts = FIRST_FAILOVER;
 		pl->failover_checkpt = l_ptr->rcv_nxt;
@@ -496,21 +524,14 @@ static void link_activate(struct tipc_link *link)
 static void link_state_event(struct tipc_link *l_ptr, unsigned int event)
 {
 	struct tipc_link *other;
-	unsigned long timer_intv = l_ptr->keepalive_intv;
-
-	if (l_ptr->flags & LINK_STOPPED)
-		return;
-
-	if (!(l_ptr->flags & LINK_STARTED) && (event != STARTING_EVT))
-		return;		/* Not yet. */
 
-	if (l_ptr->flags & LINK_FAILINGOVER)
+	if (l_ptr->exec_mode == TIPC_LINK_BLOCKED)
 		return;
 
 	switch (l_ptr->state) {
 	case WORKING_WORKING:
 		switch (event) {
-		case TRAFFIC_MSG_EVT:
+		case TRAFFIC_EVT:
 		case ACTIVATE_MSG:
 			l_ptr->silent_intv_cnt = 0;
 			break;
@@ -538,7 +559,7 @@ static void link_state_event(struct tipc_link *l_ptr, unsigned int event)
 		break;
 	case WORKING_UNKNOWN:
 		switch (event) {
-		case TRAFFIC_MSG_EVT:
+		case TRAFFIC_EVT:
 		case ACTIVATE_MSG:
 			l_ptr->state = WORKING_WORKING;
 			l_ptr->silent_intv_cnt = 0;
@@ -576,7 +597,7 @@ static void link_state_event(struct tipc_link *l_ptr, unsigned int event)
 		break;
 	case RESET_UNKNOWN:
 		switch (event) {
-		case TRAFFIC_MSG_EVT:
+		case TRAFFIC_EVT:
 			break;
 		case ACTIVATE_MSG:
 			other = node_active_link(l_ptr->owner, 0);
@@ -593,10 +614,6 @@ static void link_state_event(struct tipc_link *l_ptr, unsigned int event)
 			tipc_link_proto_xmit(l_ptr, ACTIVATE_MSG,
 					     1, 0, 0, 0);
 			break;
-		case STARTING_EVT:
-			l_ptr->flags |= LINK_STARTED;
-			link_set_timer(l_ptr, timer_intv);
-			break;
 		case SILENCE_EVT:
 			tipc_link_proto_xmit(l_ptr, RESET_MSG, 0, 0, 0, 0);
 			break;
@@ -606,7 +623,7 @@ static void link_state_event(struct tipc_link *l_ptr, unsigned int event)
 		break;
 	case RESET_RESET:
 		switch (event) {
-		case TRAFFIC_MSG_EVT:
+		case TRAFFIC_EVT:
 		case ACTIVATE_MSG:
 			other = node_active_link(l_ptr->owner, 0);
 			if (other && link_working_unknown(other))
@@ -975,7 +992,7 @@ static bool link_synch(struct tipc_link *l)
 	if (skb_queue_len(pl->inputq) > post_synch)
 		return false;
 synched:
-	l->flags &= ~LINK_SYNCHING;
+	l->exec_mode = TIPC_LINK_OPEN;
 	return true;
 }
 
@@ -1091,7 +1108,7 @@ void tipc_rcv(struct net *net, struct sk_buff *skb, struct tipc_bearer *b_ptr)
 			}
 
 			/* Traffic message. Conditionally activate link */
-			link_state_event(l_ptr, TRAFFIC_MSG_EVT);
+			link_state_event(l_ptr, TRAFFIC_EVT);
 
 			if (link_working_working(l_ptr)) {
 				/* Re-insert buffer in front of queue */
@@ -1112,7 +1129,8 @@ void tipc_rcv(struct net *net, struct sk_buff *skb, struct tipc_bearer *b_ptr)
 		l_ptr->silent_intv_cnt = 0;
 
 		/* Synchronize with parallel link if applicable */
-		if (unlikely((l_ptr->flags & LINK_SYNCHING) && !msg_dup(msg))) {
+		if (unlikely((l_ptr->exec_mode == TIPC_LINK_TUNNEL) &&
+			     !msg_dup(msg))) {
 			if (!link_synch(l_ptr))
 				goto unlock;
 		}
@@ -1193,7 +1211,7 @@ static void tipc_link_input(struct tipc_link *link, struct sk_buff *skb)
 	switch (msg_user(msg)) {
 	case TUNNEL_PROTOCOL:
 		if (msg_dup(msg)) {
-			link->flags |= LINK_SYNCHING;
+			link->exec_mode = TIPC_LINK_TUNNEL;
 			link->synch_point = msg_seqno(msg_get_wrapped(msg));
 			kfree_skb(skb);
 			break;
@@ -1315,7 +1333,7 @@ void tipc_link_proto_xmit(struct tipc_link *l_ptr, u32 msg_typ, int probe_msg,
 	u16 last_rcv;
 
 	/* Don't send protocol message during link failover */
-	if (l_ptr->flags & LINK_FAILINGOVER)
+	if (l_ptr->exec_mode == TIPC_LINK_BLOCKED)
 		return;
 
 	/* Abort non-RESET send if communication with node is prohibited */
@@ -1390,7 +1408,7 @@ static void tipc_link_proto_rcv(struct tipc_link *l_ptr,
 	u32 msg_tol;
 	struct tipc_msg *msg = buf_msg(buf);
 
-	if (l_ptr->flags & LINK_FAILINGOVER)
+	if (l_ptr->exec_mode == TIPC_LINK_BLOCKED)
 		goto exit;
 
 	if (l_ptr->net_plane != msg_net_plane(msg))
@@ -1401,7 +1419,7 @@ static void tipc_link_proto_rcv(struct tipc_link *l_ptr,
 
 	case RESET_MSG:
 		if (!link_working_unknown(l_ptr) &&
-		    (l_ptr->peer_session != INVALID_SESSION)) {
+		    (l_ptr->peer_session != WILDCARD_SESSION)) {
 			if (less_eq(msg_session(msg), l_ptr->peer_session))
 				break; /* duplicate or old reset: ignore */
 		}
@@ -1465,7 +1483,7 @@ static void tipc_link_proto_rcv(struct tipc_link *l_ptr,
 		/* Record reception; force mismatch at next timeout: */
 		l_ptr->silent_intv_cnt = 0;
 
-		link_state_event(l_ptr, TRAFFIC_MSG_EVT);
+		link_state_event(l_ptr, TRAFFIC_EVT);
 		l_ptr->stats.recv_states++;
 		if (link_reset_unknown(l_ptr))
 			break;
@@ -1704,7 +1722,7 @@ static bool tipc_link_failover_rcv(struct tipc_link *link,
 	}
 exit:
 	if (!link->failover_pkts && pl)
-		pl->flags &= ~LINK_FAILINGOVER;
+		pl->exec_mode = TIPC_LINK_OPEN;
 	kfree_skb(*skb);
 	*skb = iskb;
 	return *skb;
-- 
cgit v1.2.3


From 426cc2b86d1813959497d608dcb52c32df2d448a Mon Sep 17 00:00:00 2001
From: Jon Paul Maloy <jon.maloy@ericsson.com>
Date: Thu, 16 Jul 2015 16:54:26 -0400
Subject: tipc: introduce new link protocol msg create function

As a preparation for later changes, we introduce a new function
tipc_link_build_proto_msg(). Instead of actually sending the created
protocol message, it only creates it and adds it to the head of a
skb queue provided by the caller.

Since we still need the existing function tipc_link_protocol_xmit()
for a while, we redesign it to make use of the new function.

Reviewed-by: Ying Xue <ying.xue@windriver.com>
Signed-off-by: Jon Maloy <jon.maloy@ericsson.com>
Signed-off-by: David S. Miller <davem@davemloft.net>
---
 net/tipc/link.c | 144 ++++++++++++++++++++++++++++++--------------------------
 1 file changed, 77 insertions(+), 67 deletions(-)

(limited to 'net/tipc/link.c')

diff --git a/net/tipc/link.c b/net/tipc/link.c
index 35a2da688db1..657ba91fde41 100644
--- a/net/tipc/link.c
+++ b/net/tipc/link.c
@@ -129,6 +129,9 @@ static void tipc_link_proto_rcv(struct tipc_link *link,
 				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 tipc_link_build_proto_msg(struct tipc_link *l, int mtyp, bool probe,
+				      u16 rcvgap, int tolerance, int priority,
+				      struct sk_buff_head *xmitq);
 static void link_reset_statistics(struct tipc_link *l_ptr);
 static void link_print(struct tipc_link *l_ptr, const char *str);
 static void tipc_link_sync_xmit(struct tipc_link *l);
@@ -1323,77 +1326,21 @@ 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,
+void tipc_link_proto_xmit(struct tipc_link *l, u32 msg_typ, int probe_msg,
 			  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;
-	u16 last_rcv;
+	struct sk_buff *skb = NULL;
+	struct sk_buff_head xmitq;
 
-	/* Don't send protocol message during link failover */
-	if (l_ptr->exec_mode == TIPC_LINK_BLOCKED)
-		return;
-
-	/* Abort non-RESET send if communication with node is prohibited */
-	if ((tipc_node_blocked(l_ptr->owner)) && (msg_typ != RESET_MSG))
-		return;
-
-	/* Create protocol message with "out-of-sequence" sequence number */
-	msg_set_type(msg, msg_typ);
-	msg_set_net_plane(msg, l_ptr->net_plane);
-	msg_set_bcast_ack(msg, l_ptr->owner->bclink.last_in);
-	msg_set_last_bcast(msg, tipc_bclink_get_last_sent(l_ptr->owner->net));
-
-	if (msg_typ == STATE_MSG) {
-		u16 next_sent = l_ptr->snd_nxt;
-
-		if (!tipc_link_is_up(l_ptr))
-			return;
-		msg_set_next_sent(msg, next_sent);
-		if (!skb_queue_empty(&l_ptr->deferdq)) {
-			last_rcv = buf_seqno(skb_peek(&l_ptr->deferdq));
-			gap = mod(last_rcv - l_ptr->rcv_nxt);
-		}
-		msg_set_seq_gap(msg, gap);
-		if (gap)
-			l_ptr->stats.sent_nacks++;
-		msg_set_link_tolerance(msg, tolerance);
-		msg_set_linkprio(msg, priority);
-		msg_set_max_pkt(msg, l_ptr->mtu);
-		msg_set_ack(msg, mod(l_ptr->rcv_nxt - 1));
-		msg_set_probe(msg, probe_msg != 0);
-		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->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->advertised_mtu);
-	}
-
-	r_flag = (l_ptr->owner->working_links > tipc_link_is_up(l_ptr));
-	msg_set_redundant_link(msg, r_flag);
-	msg_set_linkprio(msg, l_ptr->priority);
-	msg_set_size(msg, msg_size);
-
-	msg_set_seqno(msg, mod(l_ptr->snd_nxt + (0xffff / 2)));
-
-	buf = tipc_buf_acquire(msg_size);
-	if (!buf)
+	__skb_queue_head_init(&xmitq);
+	tipc_link_build_proto_msg(l, msg_typ, probe_msg, gap,
+				  tolerance, priority, &xmitq);
+	skb = __skb_dequeue(&xmitq);
+	if (!skb)
 		return;
-
-	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->rcv_unacked = 0;
-	kfree_skb(buf);
+	tipc_bearer_send(l->owner->net, l->bearer_id, skb, &l->media_addr);
+	l->rcv_unacked = 0;
+	kfree_skb(skb);
 }
 
 /*
@@ -1514,6 +1461,69 @@ exit:
 	kfree_skb(buf);
 }
 
+/* tipc_link_build_proto_msg: prepare link protocol message for transmission
+ */
+static void tipc_link_build_proto_msg(struct tipc_link *l, int mtyp, bool probe,
+				      u16 rcvgap, int tolerance, int priority,
+				      struct sk_buff_head *xmitq)
+{
+	struct sk_buff *skb = NULL;
+	struct tipc_msg *hdr = l->pmsg;
+	u16 snd_nxt = l->snd_nxt;
+	u16 rcv_nxt = l->rcv_nxt;
+	u16 rcv_last = rcv_nxt - 1;
+	int node_up = l->owner->bclink.recv_permitted;
+
+	/* Don't send protocol message during reset or link failover */
+	if (l->exec_mode == TIPC_LINK_BLOCKED)
+		return;
+
+	/* Abort non-RESET send if communication with node is prohibited */
+	if ((tipc_node_blocked(l->owner)) && (mtyp != RESET_MSG))
+		return;
+
+	msg_set_type(hdr, mtyp);
+	msg_set_net_plane(hdr, l->net_plane);
+	msg_set_bcast_ack(hdr, l->owner->bclink.last_in);
+	msg_set_last_bcast(hdr, tipc_bclink_get_last_sent(l->owner->net));
+	msg_set_link_tolerance(hdr, tolerance);
+	msg_set_linkprio(hdr, priority);
+	msg_set_redundant_link(hdr, node_up);
+	msg_set_seq_gap(hdr, 0);
+
+	/* Compatibility: created msg must not be in sequence with pkt flow */
+	msg_set_seqno(hdr, snd_nxt + U16_MAX / 2);
+
+	if (mtyp == STATE_MSG) {
+		if (!tipc_link_is_up(l))
+			return;
+		msg_set_next_sent(hdr, snd_nxt);
+
+		/* Override rcvgap if there are packets in deferred queue */
+		if (!skb_queue_empty(&l->deferdq))
+			rcvgap = buf_seqno(skb_peek(&l->deferdq)) - rcv_nxt;
+		if (rcvgap) {
+			msg_set_seq_gap(hdr, rcvgap);
+			l->stats.sent_nacks++;
+		}
+		msg_set_ack(hdr, rcv_last);
+		msg_set_probe(hdr, probe);
+		if (probe)
+			l->stats.sent_probes++;
+		l->stats.sent_states++;
+	} else {
+		/* RESET_MSG or ACTIVATE_MSG */
+		msg_set_max_pkt(hdr, l->advertised_mtu);
+		msg_set_ack(hdr, l->failover_checkpt - 1);
+		msg_set_next_sent(hdr, 1);
+	}
+	skb = tipc_buf_acquire(msg_size(hdr));
+	if (!skb)
+		return;
+	skb_copy_to_linear_data(skb, hdr, msg_size(hdr));
+	skb->priority = TC_PRIO_CONTROL;
+	__skb_queue_head(xmitq, skb);
+}
 
 /* tipc_link_tunnel_xmit(): Tunnel one packet via a link belonging to
  * a different bearer. Owner node is locked.
-- 
cgit v1.2.3


From 6ab30f9cbe134d19559f48dc748587d036529aaf Mon Sep 17 00:00:00 2001
From: Jon Paul Maloy <jon.maloy@ericsson.com>
Date: Thu, 16 Jul 2015 16:54:27 -0400
Subject: tipc: improve link FSM implementation

The link FSM implementation is currently unnecessarily complex.
It sometimes checks for conditional state outside the FSM data
before deciding next state, and often performs actions directly
inside the FSM logics.

In this commit, we create a second, simpler FSM implementation,
that as far as possible acts only on states and events that it is
strictly defined for, and postpone any actions until it is finished
with its decisions. It also returns an event flag field and an a
buffer queue which may potentially contain a protocol message to
be sent by the caller.

Unfortunately, we cannot yet make the FSM "clean", in the sense
that its decisions are only based on FSM state and event, and that
state changes happen only here. That will have to wait until the
activate/reset logics has been cleaned up in a future commit.

We also rename the link states as follows:

WORKING_WORKING -> TIPC_LINK_WORKING
WORKING_UNKNOWN -> TIPC_LINK_PROBING
RESET_UNKNOWN   -> TIPC_LINK_RESETTING
RESET_RESET     -> TIPC_LINK_ESTABLISHING

The existing FSM function, link_state_event(), is still needed for
a while, so we redesign it to make use of the new function.

Reviewed-by: Ying Xue <ying.xue@windriver.com>
Signed-off-by: Jon Maloy <jon.maloy@ericsson.com>
Signed-off-by: David S. Miller <davem@davemloft.net>
---
 net/tipc/link.c | 344 +++++++++++++++++++++++++++++++-------------------------
 1 file changed, 188 insertions(+), 156 deletions(-)

(limited to 'net/tipc/link.c')

diff --git a/net/tipc/link.c b/net/tipc/link.c
index 657ba91fde41..5d2f9198c6bc 100644
--- a/net/tipc/link.c
+++ b/net/tipc/link.c
@@ -88,10 +88,10 @@ static const struct nla_policy tipc_nl_prop_policy[TIPC_NLA_PROP_MAX + 1] = {
 /* Link FSM states and events:
  */
 enum {
-	WORKING_WORKING,
-	WORKING_UNKNOWN,
-	RESET_RESET,
-	RESET_UNKNOWN
+	TIPC_LINK_WORKING,
+	TIPC_LINK_PROBING,
+	TIPC_LINK_RESETTING,
+	TIPC_LINK_ESTABLISHING
 };
 
 enum {
@@ -103,24 +103,24 @@ enum {
 
 /* Link FSM state checking routines
  */
-static int link_working_working(struct tipc_link *l)
+static int link_working(struct tipc_link *l)
 {
-	return l->state == WORKING_WORKING;
+	return l->state == TIPC_LINK_WORKING;
 }
 
-static int link_working_unknown(struct tipc_link *l)
+static int link_probing(struct tipc_link *l)
 {
-	return l->state == WORKING_UNKNOWN;
+	return l->state == TIPC_LINK_PROBING;
 }
 
-static int link_reset_unknown(struct tipc_link *l)
+static int link_resetting(struct tipc_link *l)
 {
-	return l->state == RESET_UNKNOWN;
+	return l->state == TIPC_LINK_RESETTING;
 }
 
-static int link_reset_reset(struct tipc_link *l)
+static int link_establishing(struct tipc_link *l)
 {
-	return l->state == RESET_RESET;
+	return l->state == TIPC_LINK_ESTABLISHING;
 }
 
 static void link_handle_out_of_seq_msg(struct tipc_link *link,
@@ -140,6 +140,8 @@ 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);
 static void link_set_timer(struct tipc_link *link, unsigned long time);
+static void link_activate(struct tipc_link *link);
+
 /*
  *  Simple link routines
  */
@@ -179,7 +181,7 @@ int tipc_link_is_up(struct tipc_link *l_ptr)
 {
 	if (!l_ptr)
 		return 0;
-	return link_working_working(l_ptr) || link_working_unknown(l_ptr);
+	return link_working(l_ptr) || link_probing(l_ptr);
 }
 
 int tipc_link_is_active(struct tipc_link *l)
@@ -234,8 +236,11 @@ static void link_timeout(unsigned long data)
 	}
 
 	/* do all other link processing performed on a periodic basis */
-	if (l_ptr->silent_intv_cnt || tipc_bclink_acks_missing(l_ptr->owner))
+	if (l_ptr->silent_intv_cnt)
 		link_state_event(l_ptr, SILENCE_EVT);
+	else if (link_working(l_ptr) && tipc_bclink_acks_missing(l_ptr->owner))
+		tipc_link_proto_xmit(l_ptr, STATE_MSG, 0, 0, 0, 0);
+
 	l_ptr->silent_intv_cnt++;
 	if (skb_queue_len(&l_ptr->backlogq))
 		tipc_link_push_packets(l_ptr);
@@ -304,7 +309,7 @@ struct tipc_link *tipc_link_create(struct tipc_node *n_ptr,
 	l_ptr->peer_session = WILDCARD_SESSION;
 	l_ptr->bearer_id = b_ptr->identity;
 	link_set_supervision_props(l_ptr, b_ptr->tolerance);
-	l_ptr->state = RESET_UNKNOWN;
+	l_ptr->state = TIPC_LINK_RESETTING;
 
 	l_ptr->pmsg = (struct tipc_msg *)&l_ptr->proto_msg;
 	msg = l_ptr->pmsg;
@@ -366,6 +371,134 @@ void tipc_link_delete_list(struct net *net, unsigned int bearer_id)
 	rcu_read_unlock();
 }
 
+/**
+ * tipc_link_fsm_evt - link finite state machine
+ * @l: pointer to link
+ * @evt: state machine event to be processed
+ * @xmitq: queue to prepend created protocol message, if any
+ */
+static int tipc_link_fsm_evt(struct tipc_link *l, int evt,
+			     struct sk_buff_head *xmitq)
+{
+	int mtyp = 0, rc = 0;
+	struct tipc_link *pl;
+	enum {
+		LINK_RESET    = 1,
+		LINK_ACTIVATE = (1 << 1),
+		SND_PROBE     = (1 << 2),
+		SND_STATE     = (1 << 3),
+		SND_RESET     = (1 << 4),
+		SND_ACTIVATE  = (1 << 5)
+	} actions = 0;
+
+	if (l->exec_mode == TIPC_LINK_BLOCKED)
+		return rc;
+
+	switch (l->state) {
+	case TIPC_LINK_WORKING:
+		switch (evt) {
+		case TRAFFIC_EVT:
+		case ACTIVATE_EVT:
+			break;
+		case SILENCE_EVT:
+			l->state = TIPC_LINK_PROBING;
+			actions |= SND_PROBE;
+			break;
+		case PEER_RESET_EVT:
+			actions |= LINK_RESET | SND_ACTIVATE;
+			break;
+		default:
+			pr_debug("%s%u WORKING\n", link_unk_evt, evt);
+		}
+		break;
+	case TIPC_LINK_PROBING:
+		switch (evt) {
+		case TRAFFIC_EVT:
+		case ACTIVATE_EVT:
+			l->state = TIPC_LINK_WORKING;
+			break;
+		case PEER_RESET_EVT:
+			actions |= LINK_RESET | SND_ACTIVATE;
+			break;
+		case SILENCE_EVT:
+			if (l->silent_intv_cnt <= l->abort_limit) {
+				actions |= SND_PROBE;
+				break;
+			}
+			actions |= LINK_RESET | SND_RESET;
+			break;
+		default:
+			pr_err("%s%u PROBING\n", link_unk_evt, evt);
+		}
+		break;
+	case TIPC_LINK_RESETTING:
+		switch (evt) {
+		case TRAFFIC_EVT:
+			break;
+		case ACTIVATE_EVT:
+			pl = node_active_link(l->owner, 0);
+			if (pl && link_probing(pl))
+				break;
+			actions |= LINK_ACTIVATE;
+			if (l->owner->working_links == 1)
+				tipc_link_sync_xmit(l);
+			break;
+		case PEER_RESET_EVT:
+			l->state = TIPC_LINK_ESTABLISHING;
+			actions |= SND_ACTIVATE;
+			break;
+		case SILENCE_EVT:
+			actions |= SND_RESET;
+			break;
+		default:
+			pr_err("%s%u in RESETTING\n", link_unk_evt, evt);
+		}
+		break;
+	case TIPC_LINK_ESTABLISHING:
+		switch (evt) {
+		case TRAFFIC_EVT:
+		case ACTIVATE_EVT:
+			pl = node_active_link(l->owner, 0);
+			if (pl && link_probing(pl))
+				break;
+			actions |= LINK_ACTIVATE;
+			if (l->owner->working_links == 1)
+				tipc_link_sync_xmit(l);
+			break;
+		case PEER_RESET_EVT:
+			break;
+		case SILENCE_EVT:
+			actions |= SND_ACTIVATE;
+			break;
+		default:
+			pr_err("%s%u ESTABLISHING\n", link_unk_evt, evt);
+		}
+		break;
+	default:
+		pr_err("Unknown link state %u/%u\n", l->state, evt);
+	}
+
+	/* Perform actions as decided by FSM */
+	if (actions & LINK_RESET) {
+		l->exec_mode = TIPC_LINK_BLOCKED;
+		rc |= TIPC_LINK_DOWN_EVT;
+	}
+	if (actions & LINK_ACTIVATE) {
+		l->exec_mode = TIPC_LINK_OPEN;
+		rc |= TIPC_LINK_UP_EVT;
+	}
+	if (actions & (SND_STATE | SND_PROBE))
+		mtyp = STATE_MSG;
+	if (actions & SND_RESET)
+		mtyp = RESET_MSG;
+	if (actions & SND_ACTIVATE)
+		mtyp = ACTIVATE_MSG;
+	if (actions & (SND_PROBE | SND_STATE | SND_RESET | SND_ACTIVATE))
+		tipc_link_build_proto_msg(l, mtyp, actions & SND_PROBE,
+					  0, 0, 0, xmitq);
+	return rc;
+}
+
 /**
  * link_schedule_user - schedule a message sender for wakeup after congestion
  * @link: congested link
@@ -474,9 +607,10 @@ void tipc_link_reset(struct tipc_link *l_ptr)
 	/* Prepare for renewed mtu size negotiation */
 	l_ptr->mtu = l_ptr->advertised_mtu;
 
-	l_ptr->state = RESET_UNKNOWN;
+	l_ptr->state = TIPC_LINK_RESETTING;
 
-	if ((prev_state == RESET_UNKNOWN) || (prev_state == RESET_RESET))
+	if ((prev_state == TIPC_LINK_RESETTING) ||
+	    (prev_state == TIPC_LINK_ESTABLISHING))
 		return;
 
 	tipc_node_link_down(l_ptr->owner, l_ptr->bearer_id);
@@ -515,6 +649,8 @@ static void link_activate(struct tipc_link *link)
 	link->rcv_nxt = 1;
 	link->stats.recv_info = 1;
 	link->silent_intv_cnt = 0;
+	link->state = TIPC_LINK_WORKING;
+	link->exec_mode = TIPC_LINK_OPEN;
 	tipc_node_link_up(node, link->bearer_id);
 	tipc_bearer_add_dest(node->net, link->bearer_id, link->addr);
 }
@@ -524,132 +660,29 @@ static void link_activate(struct tipc_link *link)
  * @l_ptr: pointer to link
  * @event: state machine event to process
  */
-static void link_state_event(struct tipc_link *l_ptr, unsigned int event)
+static void link_state_event(struct tipc_link *l, unsigned int evt)
 {
-	struct tipc_link *other;
+	int rc;
+	struct sk_buff_head xmitq;
+	struct sk_buff *skb;
 
-	if (l_ptr->exec_mode == TIPC_LINK_BLOCKED)
+	if (l->exec_mode == TIPC_LINK_BLOCKED)
 		return;
 
-	switch (l_ptr->state) {
-	case WORKING_WORKING:
-		switch (event) {
-		case TRAFFIC_EVT:
-		case ACTIVATE_MSG:
-			l_ptr->silent_intv_cnt = 0;
-			break;
-		case SILENCE_EVT:
-			if (!l_ptr->silent_intv_cnt) {
-				if (tipc_bclink_acks_missing(l_ptr->owner))
-					tipc_link_proto_xmit(l_ptr, STATE_MSG,
-							     0, 0, 0, 0);
-				break;
-			}
-			l_ptr->state = WORKING_UNKNOWN;
-			tipc_link_proto_xmit(l_ptr, STATE_MSG, 1, 0, 0, 0);
-			break;
-		case RESET_MSG:
-			pr_debug("%s<%s>, requested by peer\n",
-				 link_rst_msg, l_ptr->name);
-			tipc_link_reset(l_ptr);
-			l_ptr->state = RESET_RESET;
-			tipc_link_proto_xmit(l_ptr, ACTIVATE_MSG,
-					     0, 0, 0, 0);
-			break;
-		default:
-			pr_debug("%s%u in WW state\n", link_unk_evt, event);
-		}
-		break;
-	case WORKING_UNKNOWN:
-		switch (event) {
-		case TRAFFIC_EVT:
-		case ACTIVATE_MSG:
-			l_ptr->state = WORKING_WORKING;
-			l_ptr->silent_intv_cnt = 0;
-			break;
-		case RESET_MSG:
-			pr_debug("%s<%s>, requested by peer while probing\n",
-				 link_rst_msg, l_ptr->name);
-			tipc_link_reset(l_ptr);
-			l_ptr->state = RESET_RESET;
-			tipc_link_proto_xmit(l_ptr, ACTIVATE_MSG,
-					     0, 0, 0, 0);
-			break;
-		case SILENCE_EVT:
-			if (!l_ptr->silent_intv_cnt) {
-				l_ptr->state = WORKING_WORKING;
-				if (tipc_bclink_acks_missing(l_ptr->owner))
-					tipc_link_proto_xmit(l_ptr, STATE_MSG,
-							     0, 0, 0, 0);
-			} else if (l_ptr->silent_intv_cnt <
-				   l_ptr->abort_limit) {
-				tipc_link_proto_xmit(l_ptr, STATE_MSG,
-						     1, 0, 0, 0);
-			} else {	/* Link has failed */
-				pr_debug("%s<%s>, peer not responding\n",
-					 link_rst_msg, l_ptr->name);
-				tipc_link_reset(l_ptr);
-				l_ptr->state = RESET_UNKNOWN;
-				tipc_link_proto_xmit(l_ptr, RESET_MSG,
-						     0, 0, 0, 0);
-			}
-			break;
-		default:
-			pr_err("%s%u in WU state\n", link_unk_evt, event);
-		}
-		break;
-	case RESET_UNKNOWN:
-		switch (event) {
-		case TRAFFIC_EVT:
-			break;
-		case ACTIVATE_MSG:
-			other = node_active_link(l_ptr->owner, 0);
-			if (other && link_working_unknown(other))
-				break;
-			l_ptr->state = WORKING_WORKING;
-			link_activate(l_ptr);
-			tipc_link_proto_xmit(l_ptr, STATE_MSG, 1, 0, 0, 0);
-			if (l_ptr->owner->working_links == 1)
-				tipc_link_sync_xmit(l_ptr);
-			break;
-		case RESET_MSG:
-			l_ptr->state = RESET_RESET;
-			tipc_link_proto_xmit(l_ptr, ACTIVATE_MSG,
-					     1, 0, 0, 0);
-			break;
-		case SILENCE_EVT:
-			tipc_link_proto_xmit(l_ptr, RESET_MSG, 0, 0, 0, 0);
-			break;
-		default:
-			pr_err("%s%u in RU state\n", link_unk_evt, event);
-		}
-		break;
-	case RESET_RESET:
-		switch (event) {
-		case TRAFFIC_EVT:
-		case ACTIVATE_MSG:
-			other = node_active_link(l_ptr->owner, 0);
-			if (other && link_working_unknown(other))
-				break;
-			l_ptr->state = WORKING_WORKING;
-			link_activate(l_ptr);
-			tipc_link_proto_xmit(l_ptr, STATE_MSG, 1, 0, 0, 0);
-			if (l_ptr->owner->working_links == 1)
-				tipc_link_sync_xmit(l_ptr);
-			break;
-		case RESET_MSG:
-			break;
-		case SILENCE_EVT:
-			tipc_link_proto_xmit(l_ptr, ACTIVATE_MSG,
-					     0, 0, 0, 0);
-			break;
-		default:
-			pr_err("%s%u in RR state\n", link_unk_evt, event);
-		}
-		break;
-	default:
-		pr_err("Unknown link state %u/%u\n", l_ptr->state, event);
-	}
+	__skb_queue_head_init(&xmitq);
+
+	rc = tipc_link_fsm_evt(l, evt, &xmitq);
+
+	if (rc & TIPC_LINK_UP_EVT)
+		link_activate(l);
+
+	if (rc & TIPC_LINK_DOWN_EVT)
+		tipc_link_reset(l);
+
+	skb = __skb_dequeue(&xmitq);
+	if (!skb)
+		return;
+	tipc_bearer_send(l->owner->net, l->bearer_id, skb, &l->media_addr);
 }
 
 /**
@@ -1102,7 +1135,7 @@ void tipc_rcv(struct net *net, struct sk_buff *skb, struct tipc_bearer *b_ptr)
 			link_prepare_wakeup(l_ptr);
 
 		/* Process the incoming packet */
-		if (unlikely(!link_working_working(l_ptr))) {
+		if (unlikely(!link_working(l_ptr))) {
 			if (msg_user(msg) == LINK_PROTOCOL) {
 				tipc_link_proto_rcv(l_ptr, skb);
 				link_retrieve_defq(l_ptr, &head);
@@ -1113,7 +1146,7 @@ void tipc_rcv(struct net *net, struct sk_buff *skb, struct tipc_bearer *b_ptr)
 			/* Traffic message. Conditionally activate link */
 			link_state_event(l_ptr, TRAFFIC_EVT);
 
-			if (link_working_working(l_ptr)) {
+			if (link_working(l_ptr)) {
 				/* Re-insert buffer in front of queue */
 				__skb_queue_head(&head, skb);
 				skb = NULL;
@@ -1122,7 +1155,7 @@ void tipc_rcv(struct net *net, struct sk_buff *skb, struct tipc_bearer *b_ptr)
 			goto unlock;
 		}
 
-		/* Link is now in state WORKING_WORKING */
+		/* Link is now in state TIPC_LINK_WORKING */
 		if (unlikely(seq_no != l_ptr->rcv_nxt)) {
 			link_handle_out_of_seq_msg(l_ptr, skb);
 			link_retrieve_defq(l_ptr, &head);
@@ -1365,16 +1398,15 @@ static void tipc_link_proto_rcv(struct tipc_link *l_ptr,
 	switch (msg_type(msg)) {
 
 	case RESET_MSG:
-		if (!link_working_unknown(l_ptr) &&
+		if (!link_probing(l_ptr) &&
 		    (l_ptr->peer_session != WILDCARD_SESSION)) {
 			if (less_eq(msg_session(msg), l_ptr->peer_session))
 				break; /* duplicate or old reset: ignore */
 		}
 
-		if (!msg_redundant_link(msg) && (link_working_working(l_ptr) ||
-				link_working_unknown(l_ptr))) {
-			/*
-			 * peer has lost contact -- don't allow peer's links
+		if (!msg_redundant_link(msg) && (link_working(l_ptr) ||
+						 link_probing(l_ptr))) {
+			/* peer has lost contact -- don't allow peer's links
 			 * to reactivate before we recognize loss & clean up
 			 */
 			l_ptr->owner->action_flags |= TIPC_WAIT_OWN_LINKS_DOWN;
@@ -1432,7 +1464,7 @@ static void tipc_link_proto_rcv(struct tipc_link *l_ptr,
 
 		link_state_event(l_ptr, TRAFFIC_EVT);
 		l_ptr->stats.recv_states++;
-		if (link_reset_unknown(l_ptr))
+		if (link_resetting(l_ptr))
 			break;
 
 		if (less_eq(l_ptr->rcv_nxt, msg_next_sent(msg)))
@@ -1822,14 +1854,14 @@ static void link_print(struct tipc_link *l_ptr, const char *str)
 		pr_info("%s Link %x<%s>:", str, l_ptr->addr, b_ptr->name);
 	rcu_read_unlock();
 
-	if (link_working_unknown(l_ptr))
-		pr_cont(":WU\n");
-	else if (link_reset_reset(l_ptr))
-		pr_cont(":RR\n");
-	else if (link_reset_unknown(l_ptr))
-		pr_cont(":RU\n");
-	else if (link_working_working(l_ptr))
-		pr_cont(":WW\n");
+	if (link_probing(l_ptr))
+		pr_cont(":P\n");
+	else if (link_establishing(l_ptr))
+		pr_cont(":E\n");
+	else if (link_resetting(l_ptr))
+		pr_cont(":R\n");
+	else if (link_working(l_ptr))
+		pr_cont(":W\n");
 	else
 		pr_cont("\n");
 }
-- 
cgit v1.2.3


From 333ef69ed2121f535e00ceb26e095d3745584c6e Mon Sep 17 00:00:00 2001
From: Jon Paul Maloy <jon.maloy@ericsson.com>
Date: Thu, 16 Jul 2015 16:54:28 -0400
Subject: tipc: simplify link timer implementation

We create a second, simpler, link timer function, tipc_link_timeout().
The new function  makes use of the new FSM function introduced in the
previous commit, and just like it, takes a buffer queue as parameter.
It returns an event bit field and potentially a link protocol packet
to the caller.

The existing timer function, link_timeout(), is still needed for a
while, so we redesign it to become a wrapper around the new function.

Reviewed-by: Ying Xue <ying.xue@windriver.com>
Signed-off-by: Jon Maloy <jon.maloy@ericsson.com>
Signed-off-by: David S. Miller <davem@davemloft.net>
---
 net/tipc/link.c | 116 ++++++++++++++++++++++++++++++++++----------------------
 1 file changed, 71 insertions(+), 45 deletions(-)

(limited to 'net/tipc/link.c')

diff --git a/net/tipc/link.c b/net/tipc/link.c
index 5d2f9198c6bc..f58bb434d1c8 100644
--- a/net/tipc/link.c
+++ b/net/tipc/link.c
@@ -193,60 +193,30 @@ int tipc_link_is_active(struct tipc_link *l)
 
 /**
  * link_timeout - handle expiration of link timer
- * @l_ptr: pointer to link
  */
 static void link_timeout(unsigned long data)
 {
-	struct tipc_link *l_ptr = (struct tipc_link *)data;
+	struct tipc_link *l = (struct tipc_link *)data;
+	struct sk_buff_head xmitq;
 	struct sk_buff *skb;
+	int rc;
 
-	tipc_node_lock(l_ptr->owner);
+	__skb_queue_head_init(&xmitq);
 
-	/* update counters used in statistical profiling of send traffic */
-	l_ptr->stats.accu_queue_sz += skb_queue_len(&l_ptr->transmq);
-	l_ptr->stats.queue_sz_counts++;
+	tipc_node_lock(l->owner);
 
-	skb = skb_peek(&l_ptr->transmq);
-	if (skb) {
-		struct tipc_msg *msg = buf_msg(skb);
-		u32 length = msg_size(msg);
-
-		if ((msg_user(msg) == MSG_FRAGMENTER) &&
-		    (msg_type(msg) == FIRST_FRAGMENT)) {
-			length = msg_size(msg_get_wrapped(msg));
-		}
-		if (length) {
-			l_ptr->stats.msg_lengths_total += length;
-			l_ptr->stats.msg_length_counts++;
-			if (length <= 64)
-				l_ptr->stats.msg_length_profile[0]++;
-			else if (length <= 256)
-				l_ptr->stats.msg_length_profile[1]++;
-			else if (length <= 1024)
-				l_ptr->stats.msg_length_profile[2]++;
-			else if (length <= 4096)
-				l_ptr->stats.msg_length_profile[3]++;
-			else if (length <= 16384)
-				l_ptr->stats.msg_length_profile[4]++;
-			else if (length <= 32768)
-				l_ptr->stats.msg_length_profile[5]++;
-			else
-				l_ptr->stats.msg_length_profile[6]++;
-		}
-	}
+	rc = tipc_link_timeout(l, &xmitq);
 
-	/* do all other link processing performed on a periodic basis */
-	if (l_ptr->silent_intv_cnt)
-		link_state_event(l_ptr, SILENCE_EVT);
-	else if (link_working(l_ptr) && tipc_bclink_acks_missing(l_ptr->owner))
-		tipc_link_proto_xmit(l_ptr, STATE_MSG, 0, 0, 0, 0);
+	if (rc & TIPC_LINK_DOWN_EVT)
+		tipc_link_reset(l);
 
-	l_ptr->silent_intv_cnt++;
-	if (skb_queue_len(&l_ptr->backlogq))
-		tipc_link_push_packets(l_ptr);
-	link_set_timer(l_ptr, l_ptr->keepalive_intv);
-	tipc_node_unlock(l_ptr->owner);
-	tipc_link_put(l_ptr);
+	skb = __skb_dequeue(&xmitq);
+	if (skb)
+		tipc_bearer_send(l->owner->net, l->bearer_id,
+				 skb, &l->media_addr);
+	link_set_timer(l, l->keepalive_intv);
+	tipc_node_unlock(l->owner);
+	tipc_link_put(l);
 }
 
 static void link_set_timer(struct tipc_link *link, unsigned long time)
@@ -499,6 +469,62 @@ static int tipc_link_fsm_evt(struct tipc_link *l, int evt,
 	return rc;
 }
 
+/* link_profile_stats - update statistical profiling of traffic
+ */
+static void link_profile_stats(struct tipc_link *l)
+{
+	struct sk_buff *skb;
+	struct tipc_msg *msg;
+	int length;
+
+	/* Update counters used in statistical profiling of send traffic */
+	l->stats.accu_queue_sz += skb_queue_len(&l->transmq);
+	l->stats.queue_sz_counts++;
+
+	skb = skb_peek(&l->transmq);
+	if (!skb)
+		return;
+	msg = buf_msg(skb);
+	length = msg_size(msg);
+
+	if (msg_user(msg) == MSG_FRAGMENTER) {
+		if (msg_type(msg) != FIRST_FRAGMENT)
+			return;
+		length = msg_size(msg_get_wrapped(msg));
+	}
+	l->stats.msg_lengths_total += length;
+	l->stats.msg_length_counts++;
+	if (length <= 64)
+		l->stats.msg_length_profile[0]++;
+	else if (length <= 256)
+		l->stats.msg_length_profile[1]++;
+	else if (length <= 1024)
+		l->stats.msg_length_profile[2]++;
+	else if (length <= 4096)
+		l->stats.msg_length_profile[3]++;
+	else if (length <= 16384)
+		l->stats.msg_length_profile[4]++;
+	else if (length <= 32768)
+		l->stats.msg_length_profile[5]++;
+	else
+		l->stats.msg_length_profile[6]++;
+}
+
+/* tipc_link_timeout - perform periodic task as instructed from node timeout
+ */
+int tipc_link_timeout(struct tipc_link *l, struct sk_buff_head *xmitq)
+{
+	int rc = 0;
+
+	link_profile_stats(l);
+	if (l->silent_intv_cnt)
+		rc = tipc_link_fsm_evt(l, SILENCE_EVT, xmitq);
+	else if (link_working(l) && tipc_bclink_acks_missing(l->owner))
+		tipc_link_build_proto_msg(l, STATE_MSG, 0, 0, 0, 0, xmitq);
+	l->silent_intv_cnt++;
+	return rc;
+}
+
 /**
  * link_schedule_user - schedule a message sender for wakeup after congestion
  * @link: congested link
-- 
cgit v1.2.3


From 8a1577c96f122308ac9b5f195f9f9a7dd74ac541 Mon Sep 17 00:00:00 2001
From: Jon Paul Maloy <jon.maloy@ericsson.com>
Date: Thu, 16 Jul 2015 16:54:29 -0400
Subject: tipc: move link supervision timer to node level

In our effort to move control of the links to the link aggregation
layer, we move the perodic link supervision timer to struct tipc_node.
The new timer is shared between all links belonging to the node, thus
saving resources, while still kicking the FSM on both its pertaining
links at each expiration.

The current link timer and corresponding functions are removed.

Reviewed-by: Ying Xue <ying.xue@windriver.com>
Signed-off-by: Jon Maloy <jon.maloy@ericsson.com>
Signed-off-by: David S. Miller <davem@davemloft.net>
---
 net/tipc/link.c | 78 +++------------------------------------------------------
 1 file changed, 4 insertions(+), 74 deletions(-)

(limited to 'net/tipc/link.c')

diff --git a/net/tipc/link.c b/net/tipc/link.c
index f58bb434d1c8..5b4609bd0ddc 100644
--- a/net/tipc/link.c
+++ b/net/tipc/link.c
@@ -127,7 +127,6 @@ 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 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 tipc_link_build_proto_msg(struct tipc_link *l, int mtyp, bool probe,
 				      u16 rcvgap, int tolerance, int priority,
@@ -139,7 +138,6 @@ 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);
-static void link_set_timer(struct tipc_link *link, unsigned long time);
 static void link_activate(struct tipc_link *link);
 
 /*
@@ -150,21 +148,6 @@ static unsigned int align(unsigned int i)
 	return (i + 3) & ~3u;
 }
 
-static void tipc_link_release(struct kref *kref)
-{
-	kfree(container_of(kref, struct tipc_link, ref));
-}
-
-static void tipc_link_get(struct tipc_link *l_ptr)
-{
-	kref_get(&l_ptr->ref);
-}
-
-static void tipc_link_put(struct tipc_link *l_ptr)
-{
-	kref_put(&l_ptr->ref, tipc_link_release);
-}
-
 static struct tipc_link *tipc_parallel_link(struct tipc_link *l)
 {
 	struct tipc_node *n = l->owner;
@@ -191,40 +174,6 @@ int tipc_link_is_active(struct tipc_link *l)
 	return (node_active_link(n, 0) == l) || (node_active_link(n, 1) == l);
 }
 
-/**
- * link_timeout - handle expiration of link timer
- */
-static void link_timeout(unsigned long data)
-{
-	struct tipc_link *l = (struct tipc_link *)data;
-	struct sk_buff_head xmitq;
-	struct sk_buff *skb;
-	int rc;
-
-	__skb_queue_head_init(&xmitq);
-
-	tipc_node_lock(l->owner);
-
-	rc = tipc_link_timeout(l, &xmitq);
-
-	if (rc & TIPC_LINK_DOWN_EVT)
-		tipc_link_reset(l);
-
-	skb = __skb_dequeue(&xmitq);
-	if (skb)
-		tipc_bearer_send(l->owner->net, l->bearer_id,
-				 skb, &l->media_addr);
-	link_set_timer(l, l->keepalive_intv);
-	tipc_node_unlock(l->owner);
-	tipc_link_put(l);
-}
-
-static void link_set_timer(struct tipc_link *link, unsigned long time)
-{
-	if (!mod_timer(&link->timer, jiffies + time))
-		tipc_link_get(link);
-}
-
 /**
  * tipc_link_create - create a new link
  * @n_ptr: pointer to associated node
@@ -265,7 +214,6 @@ struct tipc_link *tipc_link_create(struct tipc_node *n_ptr,
 		pr_warn("Link creation failed, no memory\n");
 		return NULL;
 	}
-	kref_init(&l_ptr->ref);
 	l_ptr->addr = peer;
 	if_name = strchr(b_ptr->name, ':') + 1;
 	sprintf(l_ptr->name, "%u.%u.%u:%s-%u.%u.%u:unknown",
@@ -278,7 +226,7 @@ struct tipc_link *tipc_link_create(struct tipc_node *n_ptr,
 	l_ptr->owner = n_ptr;
 	l_ptr->peer_session = WILDCARD_SESSION;
 	l_ptr->bearer_id = b_ptr->identity;
-	link_set_supervision_props(l_ptr, b_ptr->tolerance);
+	l_ptr->tolerance = b_ptr->tolerance;
 	l_ptr->state = TIPC_LINK_RESETTING;
 
 	l_ptr->pmsg = (struct tipc_msg *)&l_ptr->proto_msg;
@@ -304,8 +252,6 @@ struct tipc_link *tipc_link_create(struct tipc_node *n_ptr,
 	skb_queue_head_init(l_ptr->inputq);
 	link_reset_statistics(l_ptr);
 	tipc_node_attach_link(n_ptr, l_ptr);
-	setup_timer(&l_ptr->timer, link_timeout, (unsigned long)l_ptr);
-	link_set_timer(l_ptr, l_ptr->keepalive_intv);
 	return l_ptr;
 }
 
@@ -316,12 +262,8 @@ struct tipc_link *tipc_link_create(struct tipc_node *n_ptr,
 void tipc_link_delete(struct tipc_link *l)
 {
 	tipc_link_reset(l);
-	if (del_timer(&l->timer))
-		tipc_link_put(l);
-	/* 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)
@@ -1447,7 +1389,7 @@ static void tipc_link_proto_rcv(struct tipc_link *l_ptr,
 
 		msg_tol = msg_link_tolerance(msg);
 		if (msg_tol > l_ptr->tolerance)
-			link_set_supervision_props(l_ptr, msg_tol);
+			l_ptr->tolerance = msg_tol;
 
 		if (msg_linkprio(msg) > l_ptr->priority)
 			l_ptr->priority = msg_linkprio(msg);
@@ -1473,7 +1415,7 @@ static void tipc_link_proto_rcv(struct tipc_link *l_ptr,
 
 		msg_tol = msg_link_tolerance(msg);
 		if (msg_tol)
-			link_set_supervision_props(l_ptr, msg_tol);
+			l_ptr->tolerance = msg_tol;
 
 		if (msg_linkprio(msg) &&
 		    (msg_linkprio(msg) != l_ptr->priority)) {
@@ -1796,18 +1738,6 @@ exit:
 	return *skb;
 }
 
-static void link_set_supervision_props(struct tipc_link *l_ptr, u32 tol)
-{
-	unsigned long intv = ((tol / 4) > 500) ? 500 : tol / 4;
-
-	if ((tol < TIPC_MIN_LINK_TOL) || (tol > TIPC_MAX_LINK_TOL))
-		return;
-
-	l_ptr->tolerance = tol;
-	l_ptr->keepalive_intv = msecs_to_jiffies(intv);
-	l_ptr->abort_limit = tol / (jiffies_to_msecs(l_ptr->keepalive_intv));
-}
-
 void tipc_link_set_queue_limits(struct tipc_link *l, u32 win)
 {
 	int max_bulk = TIPC_MAX_PUBLICATIONS / (l->mtu / ITEM_SIZE);
@@ -1984,7 +1914,7 @@ int tipc_nl_link_set(struct sk_buff *skb, struct genl_info *info)
 			u32 tol;
 
 			tol = nla_get_u32(props[TIPC_NLA_PROP_TOL]);
-			link_set_supervision_props(link, tol);
+			link->tolerance = tol;
 			tipc_link_proto_xmit(link, STATE_MSG, 0, 0, tol, 0);
 		}
 		if (props[TIPC_NLA_PROP_PRIO]) {
-- 
cgit v1.2.3


From 1a20cc254e60e79929ef7edb5cf784df86b46e42 Mon Sep 17 00:00:00 2001
From: Jon Paul Maloy <jon.maloy@ericsson.com>
Date: Thu, 16 Jul 2015 16:54:30 -0400
Subject: tipc: introduce node contact FSM

The logics for determining when a node is permitted to establish
and maintain contact with its peer node becomes non-trivial in the
presence of multiple parallel links that may come and go independently.

A known failure scenario is that one endpoint registers both its links
to the peer lost, cleans up it binding table, and prepares for a table
update once contact is re-establihed, while the other endpoint may
see its links reset and re-established one by one, hence seeing
no need to re-synchronize the binding table. To avoid this, a node
must not allow re-establishing contact until it has confirmation that
even the peer has lost both links.

Currently, the mechanism for handling this consists of setting and
resetting two state flags from different locations in the code. This
solution is hard to understand and maintain. A closer analysis even
reveals that it is not completely safe.

In this commit we do instead introduce an FSM that keeps track of
the conditions for when the node can establish and maintain links.
It has six states and four events, and is strictly based on explicit
knowledge about the own node's and the peer node's contact states.
Only events leading to state change are shown as edges in the figure
below.

                             +--------------+
                             | SELF_UP/     |
           +---------------->| PEER_COMING  |-----------------+
    SELF_  |                 +--------------+                 |PEER_
    ESTBL_ |                        |                         |ESTBL_
    CONTACT|      SELF_LOST_CONTACT |                         |CONTACT
           |                        v                         |
           |                 +--------------+                 |
           |      PEER_      | SELF_DOWN/   |     SELF_       |
           |      LOST_   +--| PEER_LEAVING |<--+ LOST_       v
+-------------+   CONTACT |  +--------------+   | CONTACT  +-----------+
| SELF_DOWN/  |<----------+                     +----------| SELF_UP/  |
| PEER_DOWN   |<----------+                     +----------| PEER_UP   |
+-------------+   SELF_   |  +--------------+   | PEER_    +-----------+
           |      LOST_   +--| SELF_LEAVING/|<--+ LOST_       A
           |      CONTACT    | PEER_DOWN    |     CONTACT     |
           |                 +--------------+                 |
           |                         A                        |
    PEER_  |       PEER_LOST_CONTACT |                        |SELF_
    ESTBL_ |                         |                        |ESTBL_
    CONTACT|                 +--------------+                 |CONTACT
           +---------------->| PEER_UP/     |-----------------+
                             | SELF_COMING  |
                             +--------------+

Reviewed-by: Ying Xue <ying.xue@windriver.com>
Signed-off-by: Jon Maloy <jon.maloy@ericsson.com>
Signed-off-by: David S. Miller <davem@davemloft.net>
---
 net/tipc/link.c | 74 +++++++++++++++++++++++++--------------------------------
 1 file changed, 32 insertions(+), 42 deletions(-)

(limited to 'net/tipc/link.c')

diff --git a/net/tipc/link.c b/net/tipc/link.c
index 5b4609bd0ddc..eaccf4552d15 100644
--- a/net/tipc/link.c
+++ b/net/tipc/link.c
@@ -911,9 +911,13 @@ static void link_retransmit_failure(struct tipc_link *l_ptr,
 
 	if (l_ptr->addr) {
 		/* Handle failure on standard link */
-		link_print(l_ptr, "Resetting link\n");
+		link_print(l_ptr, "Resetting link ");
+		pr_info("Failed msg: usr %u, typ %u, len %u, err %u\n",
+			msg_user(msg), msg_type(msg), msg_size(msg),
+			msg_errcode(msg));
+		pr_info("sqno %u, prev: %x, src: %x\n",
+			msg_seqno(msg), msg_prevnode(msg), msg_orignode(msg));
 		tipc_link_reset(l_ptr);
-
 	} else {
 		/* Handle failure on broadcast link */
 		struct tipc_node *n_ptr;
@@ -1067,15 +1071,8 @@ void tipc_rcv(struct net *net, struct sk_buff *skb, struct tipc_bearer *b_ptr)
 		if (unlikely(!l_ptr))
 			goto unlock;
 
-		/* Verify that communication with node is currently allowed */
-		if ((n_ptr->action_flags & TIPC_WAIT_PEER_LINKS_DOWN) &&
-		    msg_user(msg) == LINK_PROTOCOL &&
-		    (msg_type(msg) == RESET_MSG ||
-		    msg_type(msg) == ACTIVATE_MSG) &&
-		    !msg_redundant_link(msg))
-			n_ptr->action_flags &= ~TIPC_WAIT_PEER_LINKS_DOWN;
-
-		if (tipc_node_blocked(n_ptr))
+		/* Is reception of this pkt permitted at the moment ? */
+		if (!tipc_node_filter_skb(n_ptr, msg))
 			goto unlock;
 
 		/* Validate message sequence number info */
@@ -1371,15 +1368,6 @@ static void tipc_link_proto_rcv(struct tipc_link *l_ptr,
 			if (less_eq(msg_session(msg), l_ptr->peer_session))
 				break; /* duplicate or old reset: ignore */
 		}
-
-		if (!msg_redundant_link(msg) && (link_working(l_ptr) ||
-						 link_probing(l_ptr))) {
-			/* peer has lost contact -- don't allow peer's links
-			 * to reactivate before we recognize loss & clean up
-			 */
-			l_ptr->owner->action_flags |= TIPC_WAIT_OWN_LINKS_DOWN;
-		}
-
 		link_state_event(l_ptr, RESET_MSG);
 
 		/* fall thru' */
@@ -1408,6 +1396,8 @@ static void tipc_link_proto_rcv(struct tipc_link *l_ptr,
 		l_ptr->peer_session = msg_session(msg);
 		l_ptr->peer_bearer_id = msg_bearer_id(msg);
 
+		if (!msg_peer_is_up(msg))
+			tipc_node_fsm_evt(l_ptr->owner, PEER_LOST_CONTACT_EVT);
 		if (msg_type(msg) == ACTIVATE_MSG)
 			link_state_event(l_ptr, ACTIVATE_MSG);
 		break;
@@ -1419,11 +1409,11 @@ static void tipc_link_proto_rcv(struct tipc_link *l_ptr,
 
 		if (msg_linkprio(msg) &&
 		    (msg_linkprio(msg) != l_ptr->priority)) {
-			pr_debug("%s<%s>, priority change %u->%u\n",
-				 link_rst_msg, l_ptr->name,
-				 l_ptr->priority, msg_linkprio(msg));
+			pr_info("%s<%s>, priority change %u->%u\n",
+				link_rst_msg, l_ptr->name,
+				l_ptr->priority, msg_linkprio(msg));
 			l_ptr->priority = msg_linkprio(msg);
-			tipc_link_reset(l_ptr); /* Enforce change to take effect */
+			tipc_link_reset(l_ptr);
 			break;
 		}
 
@@ -1446,15 +1436,18 @@ static void tipc_link_proto_rcv(struct tipc_link *l_ptr,
 			tipc_bclink_update_link_state(l_ptr->owner,
 						      msg_last_bcast(msg));
 
-		if (rec_gap || (msg_probe(msg))) {
+		if (rec_gap || (msg_probe(msg)))
 			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->transmq),
 					     msg_seq_gap(msg));
 		}
+		if (tipc_link_is_up(l_ptr))
+			tipc_node_fsm_evt(l_ptr->owner,
+					  PEER_ESTABL_CONTACT_EVT);
 		break;
 	}
 exit:
@@ -1478,10 +1471,6 @@ static void tipc_link_build_proto_msg(struct tipc_link *l, int mtyp, bool probe,
 	if (l->exec_mode == TIPC_LINK_BLOCKED)
 		return;
 
-	/* Abort non-RESET send if communication with node is prohibited */
-	if ((tipc_node_blocked(l->owner)) && (mtyp != RESET_MSG))
-		return;
-
 	msg_set_type(hdr, mtyp);
 	msg_set_net_plane(hdr, l->net_plane);
 	msg_set_bcast_ack(hdr, l->owner->bclink.last_in);
@@ -1799,27 +1788,28 @@ static void link_reset_statistics(struct tipc_link *l_ptr)
 	l_ptr->stats.recv_info = l_ptr->rcv_nxt;
 }
 
-static void link_print(struct tipc_link *l_ptr, const char *str)
+static void link_print(struct tipc_link *l, const char *str)
 {
-	struct tipc_net *tn = net_generic(l_ptr->owner->net, tipc_net_id);
-	struct tipc_bearer *b_ptr;
+	struct sk_buff *hskb = skb_peek(&l->transmq);
+	u16 head = hskb ? msg_seqno(buf_msg(hskb)) : l->snd_nxt;
+	u16 tail = l->snd_nxt - 1;
 
-	rcu_read_lock();
-	b_ptr = rcu_dereference_rtnl(tn->bearer_list[l_ptr->bearer_id]);
-	if (b_ptr)
-		pr_info("%s Link %x<%s>:", str, l_ptr->addr, b_ptr->name);
-	rcu_read_unlock();
+	pr_info("%s Link <%s>:", str, l->name);
 
-	if (link_probing(l_ptr))
+	if (link_probing(l))
 		pr_cont(":P\n");
-	else if (link_establishing(l_ptr))
+	else if (link_establishing(l))
 		pr_cont(":E\n");
-	else if (link_resetting(l_ptr))
+	else if (link_resetting(l))
 		pr_cont(":R\n");
-	else if (link_working(l_ptr))
+	else if (link_working(l))
 		pr_cont(":W\n");
 	else
 		pr_cont("\n");
+
+	pr_info("XMTQ: %u [%u-%u], BKLGQ: %u, SNDNX: %u, RCVNX: %u\n",
+		skb_queue_len(&l->transmq), head, tail,
+		skb_queue_len(&l->backlogq), l->snd_nxt, l->rcv_nxt);
 }
 
 /* Parse and validate nested (link) properties valid for media, bearer and link
-- 
cgit v1.2.3


From d999297c3dbbe7fdd832f7fa4ec84301e170b3e6 Mon Sep 17 00:00:00 2001
From: Jon Paul Maloy <jon.maloy@ericsson.com>
Date: Thu, 16 Jul 2015 16:54:31 -0400
Subject: tipc: reduce locking scope during packet reception

We convert packet/message reception according to the same principle
we have been using for message sending and timeout handling:

We move the function tipc_rcv() to node.c, hence handling the initial
packet reception at the link aggregation level. The function grabs
the node lock, selects the receiving link, and accesses it via a new
call tipc_link_rcv(). This function appends buffers to the input
queue for delivery upwards, but it may also append outgoing packets
to the xmit queue, just as we do during regular message sending. The
latter will happen when buffers are forwarded from the link backlog,
or when retransmission is requested.

Upon return of this function, and after having released the node lock,
tipc_rcv() delivers/tranmsits the contents of those queues, but it may
also perform actions such as link activation or reset, as indicated by
the return flags from the link.

This reduces the number of cpu cycles spent inside the node spinlock,
and reduces contention on that lock.

Reviewed-by: Ying Xue <ying.xue@windriver.com>
Signed-off-by: Jon Maloy <jon.maloy@ericsson.com>
Signed-off-by: David S. Miller <davem@davemloft.net>
---
 net/tipc/link.c | 673 +++++++++++++++++++++++++-------------------------------
 1 file changed, 301 insertions(+), 372 deletions(-)

(limited to 'net/tipc/link.c')

diff --git a/net/tipc/link.c b/net/tipc/link.c
index eaccf4552d15..55b675d20de8 100644
--- a/net/tipc/link.c
+++ b/net/tipc/link.c
@@ -76,6 +76,10 @@ static const struct nla_policy tipc_nl_prop_policy[TIPC_NLA_PROP_MAX + 1] = {
 	[TIPC_NLA_PROP_WIN]		= { .type = NLA_U32 }
 };
 
+/*
+ * Interval between NACKs when packets arrive out of order
+ */
+#define TIPC_NACK_INTV (TIPC_MIN_LINK_WIN * 2)
 /*
  * Out-of-range value for link session numbers
  */
@@ -123,22 +127,19 @@ static int link_establishing(struct tipc_link *l)
 	return l->state == TIPC_LINK_ESTABLISHING;
 }
 
-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 void link_state_event(struct tipc_link *l_ptr, u32 event);
+static int tipc_link_proto_rcv(struct tipc_link *l, struct sk_buff *skb,
+			       struct sk_buff_head *xmitq);
 static void tipc_link_build_proto_msg(struct tipc_link *l, int mtyp, bool probe,
 				      u16 rcvgap, int tolerance, int priority,
 				      struct sk_buff_head *xmitq);
 static void link_reset_statistics(struct tipc_link *l_ptr);
 static void link_print(struct tipc_link *l_ptr, const char *str);
-static void tipc_link_sync_xmit(struct tipc_link *l);
+static void tipc_link_build_bcast_sync_msg(struct tipc_link *l,
+					   struct sk_buff_head *xmitq);
 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);
-static void link_activate(struct tipc_link *link);
 
 /*
  *  Simple link routines
@@ -283,6 +284,26 @@ void tipc_link_delete_list(struct net *net, unsigned int bearer_id)
 	rcu_read_unlock();
 }
 
+/* tipc_link_build_bcast_sync_msg() - synchronize broadcast link endpoints.
+ *
+ * Give a newly added peer node the sequence number where it should
+ * start receiving and acking broadcast packets.
+ */
+static void tipc_link_build_bcast_sync_msg(struct tipc_link *l,
+					   struct sk_buff_head *xmitq)
+{
+	struct sk_buff *skb;
+	struct sk_buff_head list;
+
+	skb = tipc_msg_create(BCAST_PROTOCOL, STATE_MSG, INT_H_SIZE,
+			      0, l->addr, link_own_addr(l), 0, 0, 0);
+	if (!skb)
+		return;
+	__skb_queue_head_init(&list);
+	__skb_queue_tail(&list, skb);
+	tipc_link_xmit(l, &list, xmitq);
+}
+
 /**
  * tipc_link_fsm_evt - link finite state machine
  * @l: pointer to link
@@ -295,12 +316,13 @@ static int tipc_link_fsm_evt(struct tipc_link *l, int evt,
 	int mtyp = 0, rc = 0;
 	struct tipc_link *pl;
 	enum {
-		LINK_RESET    = 1,
-		LINK_ACTIVATE = (1 << 1),
-		SND_PROBE     = (1 << 2),
-		SND_STATE     = (1 << 3),
-		SND_RESET     = (1 << 4),
-		SND_ACTIVATE  = (1 << 5)
+		LINK_RESET     = 1,
+		LINK_ACTIVATE  = (1 << 1),
+		SND_PROBE      = (1 << 2),
+		SND_STATE      = (1 << 3),
+		SND_RESET      = (1 << 4),
+		SND_ACTIVATE   = (1 << 5),
+		SND_BCAST_SYNC = (1 << 6)
 	} actions = 0;
 
 	if (l->exec_mode == TIPC_LINK_BLOCKED)
@@ -352,8 +374,8 @@ static int tipc_link_fsm_evt(struct tipc_link *l, int evt,
 			if (pl && link_probing(pl))
 				break;
 			actions |= LINK_ACTIVATE;
-			if (l->owner->working_links == 1)
-				tipc_link_sync_xmit(l);
+			if (!l->owner->working_links)
+				actions |= SND_BCAST_SYNC;
 			break;
 		case PEER_RESET_EVT:
 			l->state = TIPC_LINK_ESTABLISHING;
@@ -374,8 +396,8 @@ static int tipc_link_fsm_evt(struct tipc_link *l, int evt,
 			if (pl && link_probing(pl))
 				break;
 			actions |= LINK_ACTIVATE;
-			if (l->owner->working_links == 1)
-				tipc_link_sync_xmit(l);
+			if (!l->owner->working_links)
+				actions |= SND_BCAST_SYNC;
 			break;
 		case PEER_RESET_EVT:
 			break;
@@ -408,6 +430,8 @@ static int tipc_link_fsm_evt(struct tipc_link *l, int evt,
 	if (actions & (SND_PROBE | SND_STATE | SND_RESET | SND_ACTIVATE))
 		tipc_link_build_proto_msg(l, mtyp, actions & SND_PROBE,
 					  0, 0, 0, xmitq);
+	if (actions & SND_BCAST_SYNC)
+		tipc_link_build_bcast_sync_msg(l, xmitq);
 	return rc;
 }
 
@@ -605,12 +629,14 @@ void tipc_link_reset(struct tipc_link *l_ptr)
 	l_ptr->reasm_buf = NULL;
 	l_ptr->rcv_unacked = 0;
 	l_ptr->snd_nxt = 1;
+	l_ptr->rcv_nxt = 1;
 	l_ptr->silent_intv_cnt = 0;
+	l_ptr->stats.recv_info = 0;
 	l_ptr->stale_count = 0;
 	link_reset_statistics(l_ptr);
 }
 
-static void link_activate(struct tipc_link *link)
+void tipc_link_activate(struct tipc_link *link)
 {
 	struct tipc_node *node = link->owner;
 
@@ -623,36 +649,6 @@ static void link_activate(struct tipc_link *link)
 	tipc_bearer_add_dest(node->net, link->bearer_id, link->addr);
 }
 
-/**
- * link_state_event - link finite state machine
- * @l_ptr: pointer to link
- * @event: state machine event to process
- */
-static void link_state_event(struct tipc_link *l, unsigned int evt)
-{
-	int rc;
-	struct sk_buff_head xmitq;
-	struct sk_buff *skb;
-
-	if (l->exec_mode == TIPC_LINK_BLOCKED)
-		return;
-
-	__skb_queue_head_init(&xmitq);
-
-	rc = tipc_link_fsm_evt(l, evt, &xmitq);
-
-	if (rc & TIPC_LINK_UP_EVT)
-		link_activate(l);
-
-	if (rc & TIPC_LINK_DOWN_EVT)
-		tipc_link_reset(l);
-
-	skb = __skb_dequeue(&xmitq);
-	if (!skb)
-		return;
-	tipc_bearer_send(l->owner->net, l->bearer_id, skb, &l->media_addr);
-}
-
 /**
  * __tipc_link_xmit(): same as tipc_link_xmit, but destlink is known & locked
  * @link: link to use
@@ -807,30 +803,6 @@ 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_sync_xmit - synchronize broadcast link endpoints.
- *
- * Give a newly added peer node the sequence number where it should
- * start receiving and acking broadcast packets.
- *
- * Called with node locked
- */
-static void tipc_link_sync_xmit(struct tipc_link *link)
-{
-	struct sk_buff *skb;
-	struct tipc_msg *msg;
-
-	skb = tipc_buf_acquire(INT_H_SIZE);
-	if (!skb)
-		return;
-
-	msg = buf_msg(skb);
-	tipc_msg_init(link_own_addr(link), msg, BCAST_PROTOCOL, STATE_MSG,
-		      INT_H_SIZE, link->addr);
-	msg_set_last_bcast(msg, link->owner->bclink.acked);
-	__tipc_link_xmit_skb(link, skb);
-}
-
 /*
  * tipc_link_sync_rcv - synchronize broadcast link endpoints.
  * Receive the sequence number where we should start receiving and
@@ -881,6 +853,34 @@ void tipc_link_push_packets(struct tipc_link *link)
 	link->snd_nxt = seqno;
 }
 
+void tipc_link_advance_backlog(struct tipc_link *l, struct sk_buff_head *xmitq)
+{
+	struct sk_buff *skb, *_skb;
+	struct tipc_msg *hdr;
+	u16 seqno = l->snd_nxt;
+	u16 ack = l->rcv_nxt - 1;
+
+	while (skb_queue_len(&l->transmq) < l->window) {
+		skb = skb_peek(&l->backlogq);
+		if (!skb)
+			break;
+		_skb = skb_clone(skb, GFP_ATOMIC);
+		if (!_skb)
+			break;
+		__skb_dequeue(&l->backlogq);
+		hdr = buf_msg(skb);
+		l->backlog[msg_importance(hdr)].len--;
+		__skb_queue_tail(&l->transmq, skb);
+		__skb_queue_tail(xmitq, _skb);
+		msg_set_ack(hdr, ack);
+		msg_set_seqno(hdr, seqno);
+		msg_set_bcast_ack(hdr, l->owner->bclink.last_in);
+		l->rcv_unacked = 0;
+		seqno++;
+	}
+	l->snd_nxt = seqno;
+}
+
 void tipc_link_reset_all(struct tipc_node *node)
 {
 	char addr_string[16];
@@ -978,6 +978,41 @@ void tipc_link_retransmit(struct tipc_link *l_ptr, struct sk_buff *skb,
 	}
 }
 
+static int tipc_link_retransm(struct tipc_link *l, int retransm,
+			      struct sk_buff_head *xmitq)
+{
+	struct sk_buff *_skb, *skb = skb_peek(&l->transmq);
+	struct tipc_msg *hdr;
+
+	if (!skb)
+		return 0;
+
+	/* Detect repeated retransmit failures on same packet */
+	if (likely(l->last_retransm != buf_seqno(skb))) {
+		l->last_retransm = buf_seqno(skb);
+		l->stale_count = 1;
+	} else if (++l->stale_count > 100) {
+		link_retransmit_failure(l, skb);
+		return TIPC_LINK_DOWN_EVT;
+	}
+	skb_queue_walk(&l->transmq, skb) {
+		if (!retransm)
+			return 0;
+		hdr = buf_msg(skb);
+		_skb = __pskb_copy(skb, MIN_H_SIZE, GFP_ATOMIC);
+		if (!_skb)
+			return 0;
+		hdr = buf_msg(_skb);
+		msg_set_ack(hdr, l->rcv_nxt - 1);
+		msg_set_bcast_ack(hdr, l->owner->bclink.last_in);
+		_skb->priority = TC_PRIO_CONTROL;
+		__skb_queue_tail(xmitq, _skb);
+		retransm--;
+		l->stats.retransmitted++;
+	}
+	return 0;
+}
+
 /* link_synch(): check if all packets arrived before the synch
  *               point have been consumed
  * Returns true if the parallel links are synched, otherwise false
@@ -1004,155 +1039,6 @@ synched:
 	return true;
 }
 
-static void link_retrieve_defq(struct tipc_link *link,
-			       struct sk_buff_head *list)
-{
-	u16 seq_no;
-
-	if (skb_queue_empty(&link->deferdq))
-		return;
-
-	seq_no = buf_seqno(skb_peek(&link->deferdq));
-	if (seq_no == link->rcv_nxt)
-		skb_queue_splice_tail_init(&link->deferdq, list);
-}
-
-/**
- * tipc_rcv - process TIPC packets/messages arriving from off-node
- * @net: the applicable net namespace
- * @skb: TIPC packet
- * @b_ptr: pointer to bearer message arrived on
- *
- * Invoked with no locks held.  Bearer pointer must point to a valid bearer
- * structure (i.e. cannot be NULL), but bearer can be inactive.
- */
-void tipc_rcv(struct net *net, struct sk_buff *skb, struct tipc_bearer *b_ptr)
-{
-	struct tipc_net *tn = net_generic(net, tipc_net_id);
-	struct sk_buff_head head;
-	struct tipc_node *n_ptr;
-	struct tipc_link *l_ptr;
-	struct sk_buff *skb1, *tmp;
-	struct tipc_msg *msg;
-	u16 seq_no;
-	u16 ackd;
-	u32 released;
-
-	skb2list(skb, &head);
-
-	while ((skb = __skb_dequeue(&head))) {
-		/* Ensure message is well-formed */
-		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);
-			else
-				tipc_bclink_rcv(net, skb);
-			continue;
-		}
-
-		/* Discard unicast link messages destined for another node */
-		if (unlikely(!msg_short(msg) &&
-			     (msg_destnode(msg) != tn->own_addr)))
-			goto discard;
-
-		/* Locate neighboring node that sent message */
-		n_ptr = tipc_node_find(net, msg_prevnode(msg));
-		if (unlikely(!n_ptr))
-			goto discard;
-
-		tipc_node_lock(n_ptr);
-		/* Locate unicast link endpoint that should handle message */
-		l_ptr = n_ptr->links[b_ptr->identity].link;
-		if (unlikely(!l_ptr))
-			goto unlock;
-
-		/* Is reception of this pkt permitted at the moment ? */
-		if (!tipc_node_filter_skb(n_ptr, msg))
-			goto unlock;
-
-		/* Validate message sequence number info */
-		seq_no = msg_seqno(msg);
-		ackd = msg_ack(msg);
-
-		/* Release acked messages */
-		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->transmq, skb1, tmp) {
-			if (more(buf_seqno(skb1), ackd))
-				break;
-			 __skb_unlink(skb1, &l_ptr->transmq);
-			 kfree_skb(skb1);
-			 released = 1;
-		}
-
-		/* Try sending any messages link endpoint has pending */
-		if (unlikely(skb_queue_len(&l_ptr->backlogq)))
-			tipc_link_push_packets(l_ptr);
-
-		if (released && !skb_queue_empty(&l_ptr->wakeupq))
-			link_prepare_wakeup(l_ptr);
-
-		/* Process the incoming packet */
-		if (unlikely(!link_working(l_ptr))) {
-			if (msg_user(msg) == LINK_PROTOCOL) {
-				tipc_link_proto_rcv(l_ptr, skb);
-				link_retrieve_defq(l_ptr, &head);
-				skb = NULL;
-				goto unlock;
-			}
-
-			/* Traffic message. Conditionally activate link */
-			link_state_event(l_ptr, TRAFFIC_EVT);
-
-			if (link_working(l_ptr)) {
-				/* Re-insert buffer in front of queue */
-				__skb_queue_head(&head, skb);
-				skb = NULL;
-				goto unlock;
-			}
-			goto unlock;
-		}
-
-		/* Link is now in state TIPC_LINK_WORKING */
-		if (unlikely(seq_no != l_ptr->rcv_nxt)) {
-			link_handle_out_of_seq_msg(l_ptr, skb);
-			link_retrieve_defq(l_ptr, &head);
-			skb = NULL;
-			goto unlock;
-		}
-		l_ptr->silent_intv_cnt = 0;
-
-		/* Synchronize with parallel link if applicable */
-		if (unlikely((l_ptr->exec_mode == TIPC_LINK_TUNNEL) &&
-			     !msg_dup(msg))) {
-			if (!link_synch(l_ptr))
-				goto unlock;
-		}
-		l_ptr->rcv_nxt++;
-		if (unlikely(!skb_queue_empty(&l_ptr->deferdq)))
-			link_retrieve_defq(l_ptr, &head);
-		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);
-		}
-		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);
-	}
-}
-
 /* tipc_data_input - deliver data and name distr msgs to upper layer
  *
  * Consumes buffer if message is of right type
@@ -1206,9 +1092,6 @@ static void tipc_link_input(struct tipc_link *link, struct sk_buff *skb)
 	struct sk_buff *iskb;
 	int pos = 0;
 
-	if (likely(tipc_data_input(link, skb)))
-		return;
-
 	switch (msg_user(msg)) {
 	case TUNNEL_PROTOCOL:
 		if (msg_dup(msg)) {
@@ -1247,6 +1130,110 @@ static void tipc_link_input(struct tipc_link *link, struct sk_buff *skb)
 	};
 }
 
+static bool tipc_link_release_pkts(struct tipc_link *l, u16 acked)
+{
+	bool released = false;
+	struct sk_buff *skb, *tmp;
+
+	skb_queue_walk_safe(&l->transmq, skb, tmp) {
+		if (more(buf_seqno(skb), acked))
+			break;
+		__skb_unlink(skb, &l->transmq);
+		kfree_skb(skb);
+		released = true;
+	}
+	return released;
+}
+
+/* tipc_link_rcv - process TIPC packets/messages arriving from off-node
+ * @link: the link that should handle the message
+ * @skb: TIPC packet
+ * @xmitq: queue to place packets to be sent after this call
+ */
+int tipc_link_rcv(struct tipc_link *l, struct sk_buff *skb,
+		  struct sk_buff_head *xmitq)
+{
+	struct sk_buff_head *arrvq = &l->deferdq;
+	struct sk_buff *tmp;
+	struct tipc_msg *hdr;
+	u16 seqno, rcv_nxt;
+	int rc = 0;
+
+	if (unlikely(!__tipc_skb_queue_sorted(arrvq, skb))) {
+		if (!(skb_queue_len(arrvq) % TIPC_NACK_INTV))
+			tipc_link_build_proto_msg(l, STATE_MSG, 0,
+						  0, 0, 0, xmitq);
+		return rc;
+	}
+
+	skb_queue_walk_safe(arrvq, skb, tmp) {
+		hdr = buf_msg(skb);
+
+		/* Verify and update link state */
+		if (unlikely(msg_user(hdr) == LINK_PROTOCOL)) {
+			__skb_dequeue(arrvq);
+			rc |= tipc_link_proto_rcv(l, skb, xmitq);
+			continue;
+		}
+
+		if (unlikely(!link_working(l))) {
+			rc |= tipc_link_fsm_evt(l, TRAFFIC_EVT, xmitq);
+			if (!link_working(l)) {
+				kfree_skb(__skb_dequeue(arrvq));
+				return rc;
+			}
+		}
+
+		l->silent_intv_cnt = 0;
+
+		/* Forward queues and wake up waiting users */
+		if (likely(tipc_link_release_pkts(l, msg_ack(hdr)))) {
+			tipc_link_advance_backlog(l, xmitq);
+			if (unlikely(!skb_queue_empty(&l->wakeupq)))
+				link_prepare_wakeup(l);
+		}
+
+		/* Defer reception if there is a gap in the sequence */
+		seqno = msg_seqno(hdr);
+		rcv_nxt = l->rcv_nxt;
+		if (unlikely(less(rcv_nxt, seqno))) {
+			l->stats.deferred_recv++;
+			return rc;
+		}
+
+		__skb_dequeue(arrvq);
+
+		/* Drop if packet already received */
+		if (unlikely(more(rcv_nxt, seqno))) {
+			l->stats.duplicates++;
+			kfree_skb(skb);
+			return rc;
+		}
+
+		/* Synchronize with parallel link if applicable */
+		if (unlikely(l->exec_mode == TIPC_LINK_TUNNEL))
+			if (!msg_dup(hdr) && !link_synch(l)) {
+				kfree_skb(skb);
+				return rc;
+			}
+
+		/* Packet can be delivered */
+		l->rcv_nxt++;
+		l->stats.recv_info++;
+		if (unlikely(!tipc_data_input(l, skb)))
+			tipc_link_input(l, skb);
+
+		/* Ack at regular intervals */
+		if (unlikely(++l->rcv_unacked >= TIPC_MIN_LINK_WIN)) {
+			l->rcv_unacked = 0;
+			l->stats.sent_acks++;
+			tipc_link_build_proto_msg(l, STATE_MSG,
+						  0, 0, 0, 0, xmitq);
+		}
+	}
+	return rc;
+}
+
 /**
  * tipc_link_defer_pkt - Add out-of-sequence message to deferred reception queue
  *
@@ -1286,41 +1273,6 @@ u32 tipc_link_defer_pkt(struct sk_buff_head *list, struct sk_buff *skb)
 	return 1;
 }
 
-/*
- * link_handle_out_of_seq_msg - handle arrival of out-of-sequence packet
- */
-static void link_handle_out_of_seq_msg(struct tipc_link *l_ptr,
-				       struct sk_buff *buf)
-{
-	u32 seq_no = buf_seqno(buf);
-
-	if (likely(msg_user(buf_msg(buf)) == LINK_PROTOCOL)) {
-		tipc_link_proto_rcv(l_ptr, buf);
-		return;
-	}
-
-	/* Record OOS packet arrival */
-	l_ptr->silent_intv_cnt = 0;
-
-	/*
-	 * Discard packet if a duplicate; otherwise add it to deferred queue
-	 * and notify peer of gap as per protocol specification
-	 */
-	if (less(seq_no, l_ptr->rcv_nxt)) {
-		l_ptr->stats.duplicates++;
-		kfree_skb(buf);
-		return;
-	}
-
-	if (tipc_link_defer_pkt(&l_ptr->deferdq, buf)) {
-		l_ptr->stats.deferred_recv++;
-		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++;
-	}
-}
-
 /*
  * Send protocol message to the other endpoint.
  */
@@ -1341,119 +1293,6 @@ void tipc_link_proto_xmit(struct tipc_link *l, u32 msg_typ, int probe_msg,
 	kfree_skb(skb);
 }
 
-/*
- * Receive protocol message :
- * Note that network plane id propagates through the network, and may
- * change at any time. The node with lowest address rules
- */
-static void tipc_link_proto_rcv(struct tipc_link *l_ptr,
-				struct sk_buff *buf)
-{
-	u32 rec_gap = 0;
-	u32 msg_tol;
-	struct tipc_msg *msg = buf_msg(buf);
-
-	if (l_ptr->exec_mode == TIPC_LINK_BLOCKED)
-		goto exit;
-
-	if (l_ptr->net_plane != msg_net_plane(msg))
-		if (link_own_addr(l_ptr) > msg_prevnode(msg))
-			l_ptr->net_plane = msg_net_plane(msg);
-
-	switch (msg_type(msg)) {
-
-	case RESET_MSG:
-		if (!link_probing(l_ptr) &&
-		    (l_ptr->peer_session != WILDCARD_SESSION)) {
-			if (less_eq(msg_session(msg), l_ptr->peer_session))
-				break; /* duplicate or old reset: ignore */
-		}
-		link_state_event(l_ptr, RESET_MSG);
-
-		/* fall thru' */
-	case ACTIVATE_MSG:
-		/* Update link settings according other endpoint's values */
-		strcpy((strrchr(l_ptr->name, ':') + 1), (char *)msg_data(msg));
-
-		msg_tol = msg_link_tolerance(msg);
-		if (msg_tol > l_ptr->tolerance)
-			l_ptr->tolerance = msg_tol;
-
-		if (msg_linkprio(msg) > l_ptr->priority)
-			l_ptr->priority = msg_linkprio(msg);
-
-		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)) {
-			l_ptr->owner->bclink.last_sent =
-				l_ptr->owner->bclink.last_in =
-				msg_last_bcast(msg);
-			l_ptr->owner->bclink.oos_state = 0;
-		}
-
-		l_ptr->peer_session = msg_session(msg);
-		l_ptr->peer_bearer_id = msg_bearer_id(msg);
-
-		if (!msg_peer_is_up(msg))
-			tipc_node_fsm_evt(l_ptr->owner, PEER_LOST_CONTACT_EVT);
-		if (msg_type(msg) == ACTIVATE_MSG)
-			link_state_event(l_ptr, ACTIVATE_MSG);
-		break;
-	case STATE_MSG:
-
-		msg_tol = msg_link_tolerance(msg);
-		if (msg_tol)
-			l_ptr->tolerance = msg_tol;
-
-		if (msg_linkprio(msg) &&
-		    (msg_linkprio(msg) != l_ptr->priority)) {
-			pr_info("%s<%s>, priority change %u->%u\n",
-				link_rst_msg, l_ptr->name,
-				l_ptr->priority, msg_linkprio(msg));
-			l_ptr->priority = msg_linkprio(msg);
-			tipc_link_reset(l_ptr);
-			break;
-		}
-
-		/* Record reception; force mismatch at next timeout: */
-		l_ptr->silent_intv_cnt = 0;
-
-		link_state_event(l_ptr, TRAFFIC_EVT);
-		l_ptr->stats.recv_states++;
-		if (link_resetting(l_ptr))
-			break;
-
-		if (less_eq(l_ptr->rcv_nxt, msg_next_sent(msg)))
-			rec_gap = mod(msg_next_sent(msg) - l_ptr->rcv_nxt);
-
-		if (msg_probe(msg))
-			l_ptr->stats.recv_probes++;
-
-		/* Protocol message before retransmits, reduce loss risk */
-		if (l_ptr->owner->bclink.recv_permitted)
-			tipc_bclink_update_link_state(l_ptr->owner,
-						      msg_last_bcast(msg));
-
-		if (rec_gap || (msg_probe(msg)))
-			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->transmq),
-					     msg_seq_gap(msg));
-		}
-		if (tipc_link_is_up(l_ptr))
-			tipc_node_fsm_evt(l_ptr->owner,
-					  PEER_ESTABL_CONTACT_EVT);
-		break;
-	}
-exit:
-	kfree_skb(buf);
-}
-
 /* tipc_link_build_proto_msg: prepare link protocol message for transmission
  */
 static void tipc_link_build_proto_msg(struct tipc_link *l, int mtyp, bool probe,
@@ -1727,6 +1566,96 @@ exit:
 	return *skb;
 }
 
+/* tipc_link_proto_rcv(): receive link level protocol message :
+ * Note that network plane id propagates through the network, and may
+ * change at any time. The node with lowest numerical id determines
+ * network plane
+ */
+static int tipc_link_proto_rcv(struct tipc_link *l, struct sk_buff *skb,
+			       struct sk_buff_head *xmitq)
+{
+	struct tipc_msg *hdr = buf_msg(skb);
+	u16 rcvgap = 0;
+	u16 nacked_gap = msg_seq_gap(hdr);
+	u16 peers_snd_nxt =  msg_next_sent(hdr);
+	u16 peers_tol = msg_link_tolerance(hdr);
+	u16 peers_prio = msg_linkprio(hdr);
+	char *if_name;
+	int rc = 0;
+
+	if (l->exec_mode == TIPC_LINK_BLOCKED)
+		goto exit;
+
+	if (link_own_addr(l) > msg_prevnode(hdr))
+		l->net_plane = msg_net_plane(hdr);
+
+	switch (msg_type(hdr)) {
+	case RESET_MSG:
+
+		/* Ignore duplicate RESET with old session number */
+		if ((less_eq(msg_session(hdr), l->peer_session)) &&
+		    (l->peer_session != WILDCARD_SESSION))
+			break;
+		/* fall thru' */
+	case ACTIVATE_MSG:
+
+		/* Complete own link name with peer's interface name */
+		if_name =  strrchr(l->name, ':') + 1;
+		if (sizeof(l->name) - (if_name - l->name) <= TIPC_MAX_IF_NAME)
+			break;
+		if (msg_data_sz(hdr) < TIPC_MAX_IF_NAME)
+			break;
+		strncpy(if_name, msg_data(hdr),	TIPC_MAX_IF_NAME);
+
+		/* Update own tolerance if peer indicates a non-zero value */
+		if (in_range(peers_tol, TIPC_MIN_LINK_TOL, TIPC_MAX_LINK_TOL))
+			l->tolerance = peers_tol;
+
+		/* Update own priority if peer's priority is higher */
+		if (in_range(peers_prio, l->priority + 1, TIPC_MAX_LINK_PRI))
+			l->priority = peers_prio;
+
+		l->peer_session = msg_session(hdr);
+		l->peer_bearer_id = msg_bearer_id(hdr);
+		rc = tipc_link_fsm_evt(l, msg_type(hdr), xmitq);
+		if (l->mtu > msg_max_pkt(hdr))
+			l->mtu = msg_max_pkt(hdr);
+		break;
+	case STATE_MSG:
+		/* Update own tolerance if peer indicates a non-zero value */
+		if (in_range(peers_tol, TIPC_MIN_LINK_TOL, TIPC_MAX_LINK_TOL))
+			l->tolerance = peers_tol;
+
+		l->silent_intv_cnt = 0;
+		l->stats.recv_states++;
+		if (msg_probe(hdr))
+			l->stats.recv_probes++;
+		rc = tipc_link_fsm_evt(l, TRAFFIC_EVT, xmitq);
+		if (!tipc_link_is_up(l))
+			break;
+
+		/* Has peer sent packets we haven't received yet ? */
+		if (more(peers_snd_nxt, l->rcv_nxt))
+			rcvgap = peers_snd_nxt - l->rcv_nxt;
+		if (rcvgap || (msg_probe(hdr)))
+			tipc_link_build_proto_msg(l, STATE_MSG, 0, rcvgap,
+						  0, l->mtu, xmitq);
+		tipc_link_release_pkts(l, msg_ack(hdr));
+
+		/* If NACK, retransmit will now start at right position */
+		if (nacked_gap) {
+			rc |= tipc_link_retransm(l, nacked_gap, xmitq);
+			l->stats.recv_nacks++;
+		}
+		tipc_link_advance_backlog(l, xmitq);
+		if (unlikely(!skb_queue_empty(&l->wakeupq)))
+			link_prepare_wakeup(l);
+	}
+exit:
+	kfree_skb(skb);
+	return rc;
+}
+
 void tipc_link_set_queue_limits(struct tipc_link *l, u32 win)
 {
 	int max_bulk = TIPC_MAX_PUBLICATIONS / (l->mtu / ITEM_SIZE);
-- 
cgit v1.2.3


From 16040894b26af9f85d9395f072c53d76a44eba21 Mon Sep 17 00:00:00 2001
From: Jon Paul Maloy <jon.maloy@ericsson.com>
Date: Tue, 21 Jul 2015 06:42:28 -0400
Subject: tipc: fix compatibility bug

In commit d999297c3dbbe7fdd832f7fa4ec84301e170b3e6
("tipc: reduce locking scope during packet reception") we introduced
a new function tipc_link_proto_rcv(). This function contains a bug,
so that it sometimes by error sends out a non-zero link priority value
in created protocol messages.

The bug may lead to an extra link reset at initial link establising
with older nodes. This will never happen more than once, whereafter
the link will work as intended.

We fix this bug in this commit.

Signed-off-by: Jon Maloy <jon.maloy@ericsson.com>
Signed-off-by: David S. Miller <davem@davemloft.net>
---
 net/tipc/link.c | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

(limited to 'net/tipc/link.c')

diff --git a/net/tipc/link.c b/net/tipc/link.c
index 55b675d20de8..b63d57390bb7 100644
--- a/net/tipc/link.c
+++ b/net/tipc/link.c
@@ -1639,7 +1639,7 @@ static int tipc_link_proto_rcv(struct tipc_link *l, struct sk_buff *skb,
 			rcvgap = peers_snd_nxt - l->rcv_nxt;
 		if (rcvgap || (msg_probe(hdr)))
 			tipc_link_build_proto_msg(l, STATE_MSG, 0, rcvgap,
-						  0, l->mtu, xmitq);
+						  0, 0, xmitq);
 		tipc_link_release_pkts(l, msg_ack(hdr));
 
 		/* If NACK, retransmit will now start at right position */
-- 
cgit v1.2.3


From 5a4c355229da12558b5ded0775f4d0bc6650d28d Mon Sep 17 00:00:00 2001
From: Jon Maloy <jon.maloy@ericsson.com>
Date: Wed, 29 Jul 2015 18:28:01 -0400
Subject: tipc: fix bug in broadcast synch message create function

In commit d999297c3dbbe7fdd832f7fa4ec84301e170b3e6
("tipc: reduce locking scope during packet reception") we introduced
a new function tipc_build_bcast_sync_msg(), which carries initial
synchronization data between two nodes at first contact and at
re-contact. In this function, we missed to add synchronization data,
with the effect that the broadcast link endpoints will fail to
synchronize correctly at re-contact between a running and a restarted
node. All other cases work as intended.

With this commit, we fix this bug.

Signed-off-by: Jon Maloy <jon.maloy@ericsson.com>
Signed-off-by: David S. Miller <davem@davemloft.net>
---
 net/tipc/link.c | 3 +++
 1 file changed, 3 insertions(+)

(limited to 'net/tipc/link.c')

diff --git a/net/tipc/link.c b/net/tipc/link.c
index b63d57390bb7..cc40aa6eb66c 100644
--- a/net/tipc/link.c
+++ b/net/tipc/link.c
@@ -294,11 +294,14 @@ static void tipc_link_build_bcast_sync_msg(struct tipc_link *l,
 {
 	struct sk_buff *skb;
 	struct sk_buff_head list;
+	u16 last_sent;
 
 	skb = tipc_msg_create(BCAST_PROTOCOL, STATE_MSG, INT_H_SIZE,
 			      0, l->addr, link_own_addr(l), 0, 0, 0);
 	if (!skb)
 		return;
+	last_sent = tipc_bclink_get_last_sent(l->owner->net);
+	msg_set_last_bcast(buf_msg(skb), last_sent);
 	__skb_queue_head_init(&list);
 	__skb_queue_tail(&list, skb);
 	tipc_link_xmit(l, &list, xmitq);
-- 
cgit v1.2.3


From cbeb83ca68dcedf69b336fd1c5263658cbe5b51e Mon Sep 17 00:00:00 2001
From: Jon Paul Maloy <jon.maloy@ericsson.com>
Date: Thu, 30 Jul 2015 18:24:15 -0400
Subject: tipc: eliminate function tipc_link_activate()

The function tipc_link_activate() is redundant, since it mostly performs
settings that have already been done in a preceding tipc_link_reset().

There are three exceptions to this:
- The actual state change to TIPC_LINK_WORKING. This should anyway be done
  in the FSM, and not in a separate function.
- Registration of the link with the bearer. This should be done by the
  node, since we don't want the link to have any knowledge about its
  specific bearer.
- Call to tipc_node_link_up() for user access registration. With the new
  role distribution between link aggregation and link level this becomes
  the wrong call order; tipc_node_link_up() should instead be called
  directly as a result of a TIPC_LINK_UP event, hence by the node itself.

This commit implements those changes.

Tested-by: Ying Xue <ying.xue@windriver.com>
Signed-off-by: Jon Maloy <jon.maloy@ericsson.com>
Signed-off-by: David S. Miller <davem@davemloft.net>
---
 net/tipc/link.c | 17 ++++-------------
 1 file changed, 4 insertions(+), 13 deletions(-)

(limited to 'net/tipc/link.c')

diff --git a/net/tipc/link.c b/net/tipc/link.c
index cc40aa6eb66c..05837ba7b68c 100644
--- a/net/tipc/link.c
+++ b/net/tipc/link.c
@@ -228,6 +228,8 @@ struct tipc_link *tipc_link_create(struct tipc_node *n_ptr,
 	l_ptr->peer_session = WILDCARD_SESSION;
 	l_ptr->bearer_id = b_ptr->identity;
 	l_ptr->tolerance = b_ptr->tolerance;
+	l_ptr->snd_nxt = 1;
+	l_ptr->rcv_nxt = 1;
 	l_ptr->state = TIPC_LINK_RESETTING;
 
 	l_ptr->pmsg = (struct tipc_msg *)&l_ptr->proto_msg;
@@ -376,6 +378,7 @@ static int tipc_link_fsm_evt(struct tipc_link *l, int evt,
 			pl = node_active_link(l->owner, 0);
 			if (pl && link_probing(pl))
 				break;
+			l->state = TIPC_LINK_WORKING;
 			actions |= LINK_ACTIVATE;
 			if (!l->owner->working_links)
 				actions |= SND_BCAST_SYNC;
@@ -398,6 +401,7 @@ static int tipc_link_fsm_evt(struct tipc_link *l, int evt,
 			pl = node_active_link(l->owner, 0);
 			if (pl && link_probing(pl))
 				break;
+			l->state = TIPC_LINK_WORKING;
 			actions |= LINK_ACTIVATE;
 			if (!l->owner->working_links)
 				actions |= SND_BCAST_SYNC;
@@ -639,19 +643,6 @@ void tipc_link_reset(struct tipc_link *l_ptr)
 	link_reset_statistics(l_ptr);
 }
 
-void tipc_link_activate(struct tipc_link *link)
-{
-	struct tipc_node *node = link->owner;
-
-	link->rcv_nxt = 1;
-	link->stats.recv_info = 1;
-	link->silent_intv_cnt = 0;
-	link->state = TIPC_LINK_WORKING;
-	link->exec_mode = TIPC_LINK_OPEN;
-	tipc_node_link_up(node, link->bearer_id);
-	tipc_bearer_add_dest(node->net, link->bearer_id, link->addr);
-}
-
 /**
  * __tipc_link_xmit(): same as tipc_link_xmit, but destlink is known & locked
  * @link: link to use
-- 
cgit v1.2.3


From 6144a996a65199480eed7521c1c50590c282e78e Mon Sep 17 00:00:00 2001
From: Jon Paul Maloy <jon.maloy@ericsson.com>
Date: Thu, 30 Jul 2015 18:24:16 -0400
Subject: tipc: move all link_reset() calls to link aggregation level

In line with our effort to let the node level have full control over
its links, we want to move all link reset calls from link.c to node.c.
Some of the calls can be moved by simply moving the calling function,
when this is the right thing to do. For the remaining calls we use
the now established technique of returning a TIPC_LINK_DOWN_EVT
flag from tipc_link_rcv(), whereafter we perform the reset call when
the call returns.

This change serves as a preparation for the coming commits.

Tested-by: Ying Xue <ying.xue@windriver.com>
Signed-off-by: Jon Maloy <jon.maloy@ericsson.com>
Signed-off-by: David S. Miller <davem@davemloft.net>
---
 net/tipc/link.c | 81 ++++++++++++++-------------------------------------------
 1 file changed, 20 insertions(+), 61 deletions(-)

(limited to 'net/tipc/link.c')

diff --git a/net/tipc/link.c b/net/tipc/link.c
index 05837ba7b68c..8c81db7b17f9 100644
--- a/net/tipc/link.c
+++ b/net/tipc/link.c
@@ -137,9 +137,9 @@ static void link_print(struct tipc_link *l_ptr, const char *str);
 static void tipc_link_build_bcast_sync_msg(struct tipc_link *l,
 					   struct sk_buff_head *xmitq);
 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 int 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);
+static int tipc_link_failover_rcv(struct tipc_link *l, struct sk_buff **skb);
 
 /*
  *  Simple link routines
@@ -258,34 +258,6 @@ struct tipc_link *tipc_link_create(struct tipc_node *n_ptr,
 	return l_ptr;
 }
 
-/**
- * tipc_link_delete - Delete a link
- * @l: link to be deleted
- */
-void tipc_link_delete(struct tipc_link *l)
-{
-	tipc_link_reset(l);
-	tipc_link_reset_fragments(l);
-	tipc_node_detach_link(l->owner, l);
-}
-
-void tipc_link_delete_list(struct net *net, unsigned int bearer_id)
-{
-	struct tipc_net *tn = net_generic(net, tipc_net_id);
-	struct tipc_link *link;
-	struct tipc_node *node;
-
-	rcu_read_lock();
-	list_for_each_entry_rcu(node, &tn->node_list, list) {
-		tipc_node_lock(node);
-		link = node->links[bearer_id].link;
-		if (link)
-			tipc_link_delete(link);
-		tipc_node_unlock(node);
-	}
-	rcu_read_unlock();
-}
-
 /* tipc_link_build_bcast_sync_msg() - synchronize broadcast link endpoints.
  *
  * Give a newly added peer node the sequence number where it should
@@ -875,26 +847,6 @@ void tipc_link_advance_backlog(struct tipc_link *l, struct sk_buff_head *xmitq)
 	l->snd_nxt = seqno;
 }
 
-void tipc_link_reset_all(struct tipc_node *node)
-{
-	char addr_string[16];
-	u32 i;
-
-	tipc_node_lock(node);
-
-	pr_warn("Resetting all links to %s\n",
-		tipc_addr_string_fill(addr_string, node->addr));
-
-	for (i = 0; i < MAX_BEARERS; i++) {
-		if (node->links[i].link) {
-			link_print(node->links[i].link, "Resetting link\n");
-			tipc_link_reset(node->links[i].link);
-		}
-	}
-
-	tipc_node_unlock(node);
-}
-
 static void link_retransmit_failure(struct tipc_link *l_ptr,
 				    struct sk_buff *buf)
 {
@@ -911,7 +863,6 @@ static void link_retransmit_failure(struct tipc_link *l_ptr,
 			msg_errcode(msg));
 		pr_info("sqno %u, prev: %x, src: %x\n",
 			msg_seqno(msg), msg_prevnode(msg), msg_orignode(msg));
-		tipc_link_reset(l_ptr);
 	} else {
 		/* Handle failure on broadcast link */
 		struct tipc_node *n_ptr;
@@ -987,6 +938,7 @@ static int tipc_link_retransm(struct tipc_link *l, int retransm,
 		l->stale_count = 1;
 	} else if (++l->stale_count > 100) {
 		link_retransmit_failure(l, skb);
+		l->exec_mode = TIPC_LINK_BLOCKED;
 		return TIPC_LINK_DOWN_EVT;
 	}
 	skb_queue_walk(&l->transmq, skb) {
@@ -1079,12 +1031,13 @@ static bool tipc_data_input(struct tipc_link *link, struct sk_buff *skb)
  * Consumes buffer
  * Node lock must be held
  */
-static void tipc_link_input(struct tipc_link *link, struct sk_buff *skb)
+static int tipc_link_input(struct tipc_link *link, struct sk_buff *skb)
 {
 	struct tipc_node *node = link->owner;
 	struct tipc_msg *msg = buf_msg(skb);
 	struct sk_buff *iskb;
 	int pos = 0;
+	int rc = 0;
 
 	switch (msg_user(msg)) {
 	case TUNNEL_PROTOCOL:
@@ -1094,7 +1047,8 @@ static void tipc_link_input(struct tipc_link *link, struct sk_buff *skb)
 			kfree_skb(skb);
 			break;
 		}
-		if (!tipc_link_failover_rcv(link, &skb))
+		rc |= tipc_link_failover_rcv(link, &skb);
+		if (!skb)
 			break;
 		if (msg_user(buf_msg(skb)) != MSG_BUNDLER) {
 			tipc_data_input(link, skb);
@@ -1113,7 +1067,8 @@ static void tipc_link_input(struct tipc_link *link, struct sk_buff *skb)
 			link->stats.recv_fragmented++;
 			tipc_data_input(link, skb);
 		} else if (!link->reasm_buf) {
-			tipc_link_reset(link);
+			link->exec_mode = TIPC_LINK_BLOCKED;
+			rc |= TIPC_LINK_DOWN_EVT;
 		}
 		break;
 	case BCAST_PROTOCOL:
@@ -1122,6 +1077,7 @@ static void tipc_link_input(struct tipc_link *link, struct sk_buff *skb)
 	default:
 		break;
 	};
+	return rc;
 }
 
 static bool tipc_link_release_pkts(struct tipc_link *l, u16 acked)
@@ -1215,7 +1171,7 @@ int tipc_link_rcv(struct tipc_link *l, struct sk_buff *skb,
 		l->rcv_nxt++;
 		l->stats.recv_info++;
 		if (unlikely(!tipc_data_input(l, skb)))
-			tipc_link_input(l, skb);
+			rc |= tipc_link_input(l, skb);
 
 		/* Ack at regular intervals */
 		if (unlikely(++l->rcv_unacked >= TIPC_MIN_LINK_WIN)) {
@@ -1504,14 +1460,15 @@ tunnel_queue:
 /*  tipc_link_failover_rcv(): Receive a tunnelled FAILOVER_MSG packet
  *  Owner node is locked.
  */
-static bool tipc_link_failover_rcv(struct tipc_link *link,
-				   struct sk_buff **skb)
+static int tipc_link_failover_rcv(struct tipc_link *link,
+				  struct sk_buff **skb)
 {
 	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;
+	int rc = 0;
 
 	if (msg_type(msg) != FAILOVER_MSG) {
 		pr_warn("%sunknown tunnel pkt received\n", link_co_err);
@@ -1524,8 +1481,6 @@ static bool tipc_link_failover_rcv(struct tipc_link *link,
 		goto exit;
 
 	pl = link->owner->links[bearer_id].link;
-	if (pl && tipc_link_is_up(pl))
-		tipc_link_reset(pl);
 
 	if (link->failover_pkts == FIRST_FAILOVER)
 		link->failover_pkts = msg_msgcnt(msg);
@@ -1550,14 +1505,18 @@ static bool tipc_link_failover_rcv(struct tipc_link *link,
 	}
 	if (msg_user(buf_msg(iskb)) == MSG_FRAGMENTER) {
 		link->stats.recv_fragments++;
-		tipc_buf_append(&link->failover_skb, &iskb);
+		if (!tipc_buf_append(&link->failover_skb, &iskb) &&
+		    !link->failover_skb) {
+			link->exec_mode = TIPC_LINK_BLOCKED;
+			rc |= TIPC_LINK_DOWN_EVT;
+		}
 	}
 exit:
 	if (!link->failover_pkts && pl)
 		pl->exec_mode = TIPC_LINK_OPEN;
 	kfree_skb(*skb);
 	*skb = iskb;
-	return *skb;
+	return rc;
 }
 
 /* tipc_link_proto_rcv(): receive link level protocol message :
-- 
cgit v1.2.3


From 655fb243b8ae5e652f744311bcb6e806e83cea1e Mon Sep 17 00:00:00 2001
From: Jon Paul Maloy <jon.maloy@ericsson.com>
Date: Thu, 30 Jul 2015 18:24:17 -0400
Subject: tipc: reverse call order for link_reset()->node_link_down()

In many cases the call order when a link is reset goes as follows:
tipc_node_xx()->tipc_link_reset()->tipc_node_link_down()

This is not the right order if we want the node to be in control,
so in this commit we change the order to:
tipc_node_xx()->tipc_node_link_down()->tipc_link_reset()

The fact that tipc_link_reset() now is called from only one
location with a well-defined state will also facilitate later
simplifications of tipc_link_reset() and the link FSM.

Tested-by: Ying Xue <ying.xue@windriver.com>
Signed-off-by: Jon Maloy <jon.maloy@ericsson.com>
Signed-off-by: David S. Miller <davem@davemloft.net>
---
 net/tipc/link.c | 6 +-----
 1 file changed, 1 insertion(+), 5 deletions(-)

(limited to 'net/tipc/link.c')

diff --git a/net/tipc/link.c b/net/tipc/link.c
index 8c81db7b17f9..2ccdb6ffd5c8 100644
--- a/net/tipc/link.c
+++ b/net/tipc/link.c
@@ -566,7 +566,6 @@ void tipc_link_purge_queues(struct tipc_link *l_ptr)
 void tipc_link_reset(struct tipc_link *l_ptr)
 {
 	u32 prev_state = l_ptr->state;
-	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);
 
@@ -584,10 +583,7 @@ void tipc_link_reset(struct tipc_link *l_ptr)
 	    (prev_state == TIPC_LINK_ESTABLISHING))
 		return;
 
-	tipc_node_link_down(l_ptr->owner, l_ptr->bearer_id);
-	tipc_bearer_remove_dest(owner->net, l_ptr->bearer_id, l_ptr->addr);
-
-	if (was_active_link && tipc_node_is_up(l_ptr->owner) && (pl != l_ptr)) {
+	if (tipc_node_is_up(l_ptr->owner) && (pl != l_ptr)) {
 		l_ptr->exec_mode = TIPC_LINK_BLOCKED;
 		l_ptr->failover_checkpt = l_ptr->rcv_nxt;
 		pl->failover_pkts = FIRST_FAILOVER;
-- 
cgit v1.2.3


From 6e498158a827fd515b514842e9a06bdf0f75ab86 Mon Sep 17 00:00:00 2001
From: Jon Paul Maloy <jon.maloy@ericsson.com>
Date: Thu, 30 Jul 2015 18:24:19 -0400
Subject: tipc: move link synch and failover to link aggregation level

Link failover and synchronization have until now been handled by the
links themselves, forcing them to have knowledge about and to access
parallel links in order to make the two algorithms work correctly.

In this commit, we move the control part of this functionality to the
link aggregation level in node.c, which is the right location for this.
As a result, the two algorithms become easier to follow, and the link
implementation becomes simpler.

Tested-by: Ying Xue <ying.xue@windriver.com>
Signed-off-by: Jon Maloy <jon.maloy@ericsson.com>
Signed-off-by: David S. Miller <davem@davemloft.net>
---
 net/tipc/link.c | 482 ++++++++++++++++----------------------------------------
 1 file changed, 132 insertions(+), 350 deletions(-)

(limited to 'net/tipc/link.c')

diff --git a/net/tipc/link.c b/net/tipc/link.c
index 2ccdb6ffd5c8..d5f4005f388f 100644
--- a/net/tipc/link.c
+++ b/net/tipc/link.c
@@ -48,7 +48,7 @@
 /*
  * Error message prefixes
  */
-static const char *link_co_err = "Link changeover error, ";
+static const char *link_co_err = "Link tunneling error, ";
 static const char *link_rst_msg = "Resetting link ";
 static const char *link_unk_evt = "Unknown link event ";
 
@@ -139,24 +139,6 @@ static void tipc_link_build_bcast_sync_msg(struct tipc_link *l,
 static void tipc_link_sync_rcv(struct tipc_node *n, struct sk_buff *buf);
 static int 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 int tipc_link_failover_rcv(struct tipc_link *l, struct sk_buff **skb);
-
-/*
- *  Simple link routines
- */
-static unsigned int align(unsigned int i)
-{
-	return (i + 3) & ~3u;
-}
-
-static struct tipc_link *tipc_parallel_link(struct tipc_link *l)
-{
-	struct tipc_node *n = l->owner;
-
-	if (node_active_link(n, 0) != l)
-		return node_active_link(n, 0);
-	return node_active_link(n, 1);
-}
 
 /*
  *  Simple non-static link routines (i.e. referenced outside this file)
@@ -394,12 +376,10 @@ static int tipc_link_fsm_evt(struct tipc_link *l, int evt,
 	/* Perform actions as decided by FSM */
 	if (actions & LINK_RESET) {
 		l->exec_mode = TIPC_LINK_BLOCKED;
-		rc |= TIPC_LINK_DOWN_EVT;
-	}
-	if (actions & LINK_ACTIVATE) {
-		l->exec_mode = TIPC_LINK_OPEN;
-		rc |= TIPC_LINK_UP_EVT;
+		rc = TIPC_LINK_DOWN_EVT;
 	}
+	if (actions & LINK_ACTIVATE)
+		rc = TIPC_LINK_UP_EVT;
 	if (actions & (SND_STATE | SND_PROBE))
 		mtyp = STATE_MSG;
 	if (actions & SND_RESET)
@@ -461,6 +441,9 @@ int tipc_link_timeout(struct tipc_link *l, struct sk_buff_head *xmitq)
 {
 	int rc = 0;
 
+	if (l->exec_mode == TIPC_LINK_BLOCKED)
+		return rc;
+
 	link_profile_stats(l);
 	if (l->silent_intv_cnt)
 		rc = tipc_link_fsm_evt(l, SILENCE_EVT, xmitq);
@@ -563,52 +546,42 @@ void tipc_link_purge_queues(struct tipc_link *l_ptr)
 	tipc_link_reset_fragments(l_ptr);
 }
 
-void tipc_link_reset(struct tipc_link *l_ptr)
+void tipc_link_reset(struct tipc_link *l)
 {
-	u32 prev_state = l_ptr->state;
-	struct tipc_node *owner = l_ptr->owner;
-	struct tipc_link *pl = tipc_parallel_link(l_ptr);
+	struct tipc_node *owner = l->owner;
 
-	msg_set_session(l_ptr->pmsg, ((msg_session(l_ptr->pmsg) + 1) & 0xffff));
+	l->state = TIPC_LINK_RESETTING;
 
 	/* Link is down, accept any session */
-	l_ptr->peer_session = WILDCARD_SESSION;
+	l->peer_session = WILDCARD_SESSION;
 
-	/* Prepare for renewed mtu size negotiation */
-	l_ptr->mtu = l_ptr->advertised_mtu;
-
-	l_ptr->state = TIPC_LINK_RESETTING;
+	/* If peer is up, it only accepts an incremented session number */
+	msg_set_session(l->pmsg, msg_session(l->pmsg) + 1);
 
-	if ((prev_state == TIPC_LINK_RESETTING) ||
-	    (prev_state == TIPC_LINK_ESTABLISHING))
-		return;
+	/* Prepare for renewed mtu size negotiation */
+	l->mtu = l->advertised_mtu;
 
-	if (tipc_node_is_up(l_ptr->owner) && (pl != l_ptr)) {
-		l_ptr->exec_mode = TIPC_LINK_BLOCKED;
-		l_ptr->failover_checkpt = l_ptr->rcv_nxt;
-		pl->failover_pkts = FIRST_FAILOVER;
-		pl->failover_checkpt = l_ptr->rcv_nxt;
-		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->transmq);
-	__skb_queue_purge(&l_ptr->deferdq);
+	__skb_queue_purge(&l->transmq);
+	__skb_queue_purge(&l->deferdq);
 	if (!owner->inputq)
-		owner->inputq = l_ptr->inputq;
-	skb_queue_splice_init(&l_ptr->wakeupq, owner->inputq);
+		owner->inputq = l->inputq;
+	skb_queue_splice_init(&l->wakeupq, owner->inputq);
 	if (!skb_queue_empty(owner->inputq))
 		owner->action_flags |= TIPC_MSG_EVT;
-	tipc_link_purge_backlog(l_ptr);
-	l_ptr->reasm_buf = NULL;
-	l_ptr->rcv_unacked = 0;
-	l_ptr->snd_nxt = 1;
-	l_ptr->rcv_nxt = 1;
-	l_ptr->silent_intv_cnt = 0;
-	l_ptr->stats.recv_info = 0;
-	l_ptr->stale_count = 0;
-	link_reset_statistics(l_ptr);
+
+	tipc_link_purge_backlog(l);
+	kfree_skb(l->reasm_buf);
+	kfree_skb(l->failover_reasm_skb);
+	l->reasm_buf = NULL;
+	l->failover_reasm_skb = NULL;
+	l->rcv_unacked = 0;
+	l->snd_nxt = 1;
+	l->rcv_nxt = 1;
+	l->silent_intv_cnt = 0;
+	l->stats.recv_info = 0;
+	l->stale_count = 0;
+	link_reset_statistics(l);
 }
 
 /**
@@ -751,20 +724,6 @@ int tipc_link_xmit(struct tipc_link *l, struct sk_buff_head *list,
 	return 0;
 }
 
-static void skb2list(struct sk_buff *skb, struct sk_buff_head *list)
-{
-	skb_queue_head_init(list);
-	__skb_queue_tail(list, skb);
-}
-
-static int __tipc_link_xmit_skb(struct tipc_link *link, struct sk_buff *skb)
-{
-	struct sk_buff_head head;
-
-	skb2list(skb, &head);
-	return __tipc_link_xmit(link->owner->net, link, &head);
-}
-
 /*
  * tipc_link_sync_rcv - synchronize broadcast link endpoints.
  * Receive the sequence number where we should start receiving and
@@ -955,32 +914,6 @@ static int tipc_link_retransm(struct tipc_link *l, int retransm,
 	return 0;
 }
 
-/* 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 bool link_synch(struct tipc_link *l)
-{
-	unsigned int post_synch;
-	struct tipc_link *pl;
-
-	pl  = tipc_parallel_link(l);
-	if (pl == l)
-		goto synched;
-
-	/* Was last pre-synch packet added to input queue ? */
-	if (less_eq(pl->rcv_nxt, l->synch_point))
-		return false;
-
-	/* Is it still in the input queue ? */
-	post_synch = mod(pl->rcv_nxt - l->synch_point) - 1;
-	if (skb_queue_len(pl->inputq) > post_synch)
-		return false;
-synched:
-	l->exec_mode = TIPC_LINK_OPEN;
-	return true;
-}
-
 /* tipc_data_input - deliver data and name distr msgs to upper layer
  *
  * Consumes buffer if message is of right type
@@ -1025,54 +958,59 @@ static bool tipc_data_input(struct tipc_link *link, struct sk_buff *skb)
 /* tipc_link_input - process packet that has passed link protocol check
  *
  * Consumes buffer
- * Node lock must be held
  */
-static int tipc_link_input(struct tipc_link *link, struct sk_buff *skb)
+static int tipc_link_input(struct tipc_link *l, struct sk_buff *skb)
 {
-	struct tipc_node *node = link->owner;
-	struct tipc_msg *msg = buf_msg(skb);
+	struct tipc_node *node = l->owner;
+	struct tipc_msg *hdr = buf_msg(skb);
+	struct sk_buff **reasm_skb = &l->reasm_buf;
 	struct sk_buff *iskb;
-	int pos = 0;
+	int usr = msg_user(hdr);
 	int rc = 0;
+	int pos = 0;
+	int ipos = 0;
 
-	switch (msg_user(msg)) {
-	case TUNNEL_PROTOCOL:
-		if (msg_dup(msg)) {
-			link->exec_mode = TIPC_LINK_TUNNEL;
-			link->synch_point = msg_seqno(msg_get_wrapped(msg));
-			kfree_skb(skb);
-			break;
+	if (unlikely(usr == TUNNEL_PROTOCOL)) {
+		if (msg_type(hdr) == SYNCH_MSG) {
+			__skb_queue_purge(&l->deferdq);
+			goto drop;
 		}
-		rc |= tipc_link_failover_rcv(link, &skb);
-		if (!skb)
-			break;
-		if (msg_user(buf_msg(skb)) != MSG_BUNDLER) {
-			tipc_data_input(link, skb);
-			break;
-		}
-	case MSG_BUNDLER:
-		link->stats.recv_bundles++;
-		link->stats.recv_bundled += msg_msgcnt(msg);
+		if (!tipc_msg_extract(skb, &iskb, &ipos))
+			return rc;
+		kfree_skb(skb);
+		skb = iskb;
+		hdr = buf_msg(skb);
+		if (less(msg_seqno(hdr), l->drop_point))
+			goto drop;
+		if (tipc_data_input(l, skb))
+			return rc;
+		usr = msg_user(hdr);
+		reasm_skb = &l->failover_reasm_skb;
+	}
 
+	if (usr == MSG_BUNDLER) {
+		l->stats.recv_bundles++;
+		l->stats.recv_bundled += msg_msgcnt(hdr);
 		while (tipc_msg_extract(skb, &iskb, &pos))
-			tipc_data_input(link, iskb);
-		break;
-	case MSG_FRAGMENTER:
-		link->stats.recv_fragments++;
-		if (tipc_buf_append(&link->reasm_buf, &skb)) {
-			link->stats.recv_fragmented++;
-			tipc_data_input(link, skb);
-		} else if (!link->reasm_buf) {
-			link->exec_mode = TIPC_LINK_BLOCKED;
-			rc |= TIPC_LINK_DOWN_EVT;
+			tipc_data_input(l, iskb);
+		return rc;
+	} else if (usr == MSG_FRAGMENTER) {
+		l->stats.recv_fragments++;
+		if (tipc_buf_append(reasm_skb, &skb)) {
+			l->stats.recv_fragmented++;
+			tipc_data_input(l, skb);
+		} else if (!*reasm_skb) {
+			l->exec_mode = TIPC_LINK_BLOCKED;
+			l->state = TIPC_LINK_RESETTING;
+			rc = TIPC_LINK_DOWN_EVT;
 		}
-		break;
-	case BCAST_PROTOCOL:
+		return rc;
+	} else if (usr == BCAST_PROTOCOL) {
 		tipc_link_sync_rcv(node, skb);
-		break;
-	default:
-		break;
-	};
+		return rc;
+	}
+drop:
+	kfree_skb(skb);
 	return rc;
 }
 
@@ -1100,7 +1038,6 @@ int tipc_link_rcv(struct tipc_link *l, struct sk_buff *skb,
 		  struct sk_buff_head *xmitq)
 {
 	struct sk_buff_head *arrvq = &l->deferdq;
-	struct sk_buff *tmp;
 	struct tipc_msg *hdr;
 	u16 seqno, rcv_nxt;
 	int rc = 0;
@@ -1112,18 +1049,18 @@ int tipc_link_rcv(struct tipc_link *l, struct sk_buff *skb,
 		return rc;
 	}
 
-	skb_queue_walk_safe(arrvq, skb, tmp) {
+	while ((skb = skb_peek(arrvq))) {
 		hdr = buf_msg(skb);
 
 		/* Verify and update link state */
 		if (unlikely(msg_user(hdr) == LINK_PROTOCOL)) {
 			__skb_dequeue(arrvq);
-			rc |= tipc_link_proto_rcv(l, skb, xmitq);
+			rc = tipc_link_proto_rcv(l, skb, xmitq);
 			continue;
 		}
 
 		if (unlikely(!link_working(l))) {
-			rc |= tipc_link_fsm_evt(l, TRAFFIC_EVT, xmitq);
+			rc = tipc_link_fsm_evt(l, TRAFFIC_EVT, xmitq);
 			if (!link_working(l)) {
 				kfree_skb(__skb_dequeue(arrvq));
 				return rc;
@@ -1156,18 +1093,11 @@ int tipc_link_rcv(struct tipc_link *l, struct sk_buff *skb,
 			return rc;
 		}
 
-		/* Synchronize with parallel link if applicable */
-		if (unlikely(l->exec_mode == TIPC_LINK_TUNNEL))
-			if (!msg_dup(hdr) && !link_synch(l)) {
-				kfree_skb(skb);
-				return rc;
-			}
-
 		/* Packet can be delivered */
 		l->rcv_nxt++;
 		l->stats.recv_info++;
 		if (unlikely(!tipc_data_input(l, skb)))
-			rc |= tipc_link_input(l, skb);
+			rc = tipc_link_input(l, skb);
 
 		/* Ack at regular intervals */
 		if (unlikely(++l->rcv_unacked >= TIPC_MIN_LINK_WIN)) {
@@ -1288,7 +1218,7 @@ static void tipc_link_build_proto_msg(struct tipc_link *l, int mtyp, bool probe,
 	} else {
 		/* RESET_MSG or ACTIVATE_MSG */
 		msg_set_max_pkt(hdr, l->advertised_mtu);
-		msg_set_ack(hdr, l->failover_checkpt - 1);
+		msg_set_ack(hdr, l->rcv_nxt - 1);
 		msg_set_next_sent(hdr, 1);
 	}
 	skb = tipc_buf_acquire(msg_size(hdr));
@@ -1296,223 +1226,75 @@ static void tipc_link_build_proto_msg(struct tipc_link *l, int mtyp, bool probe,
 		return;
 	skb_copy_to_linear_data(skb, hdr, msg_size(hdr));
 	skb->priority = TC_PRIO_CONTROL;
-	__skb_queue_head(xmitq, skb);
-}
-
-/* tipc_link_tunnel_xmit(): Tunnel one packet via a link belonging to
- * a different bearer. Owner node is locked.
- */
-static void tipc_link_tunnel_xmit(struct tipc_link *l_ptr,
-				  struct tipc_msg *tunnel_hdr,
-				  struct tipc_msg *msg,
-				  u32 selector)
-{
-	struct tipc_link *tunnel;
-	struct sk_buff *skb;
-	u32 length = msg_size(msg);
-
-	tunnel = node_active_link(l_ptr->owner, selector & 1);
-	if (!tipc_link_is_up(tunnel)) {
-		pr_warn("%stunnel link no longer available\n", link_co_err);
-		return;
-	}
-	msg_set_size(tunnel_hdr, length + INT_H_SIZE);
-	skb = tipc_buf_acquire(length + INT_H_SIZE);
-	if (!skb) {
-		pr_warn("%sunable to send tunnel msg\n", link_co_err);
-		return;
-	}
-	skb_copy_to_linear_data(skb, tunnel_hdr, INT_H_SIZE);
-	skb_copy_to_linear_data_offset(skb, INT_H_SIZE, msg, length);
-	__tipc_link_xmit_skb(tunnel, skb);
+	__skb_queue_tail(xmitq, skb);
 }
 
-
-/* tipc_link_failover_send_queue(): A link has gone down, but a second
- * link is still active. We can do failover. Tunnel the failing link's
- * whole send queue via the remaining link. This way, we don't lose
- * any packets, and sequence order is preserved for subsequent traffic
- * sent over the remaining link. Owner node is locked.
+/* tipc_link_tnl_prepare(): prepare and return a list of tunnel packets
+ * with contents of the link's tranmsit and backlog queues.
  */
-void tipc_link_failover_send_queue(struct tipc_link *l_ptr)
+void tipc_link_tnl_prepare(struct tipc_link *l, struct tipc_link *tnl,
+			   int mtyp, struct sk_buff_head *xmitq)
 {
-	int msgcount;
-	struct tipc_link *tunnel = node_active_link(l_ptr->owner, 0);
-	struct tipc_msg tunnel_hdr;
-	struct sk_buff *skb;
-	int split_bundles;
+	struct sk_buff *skb, *tnlskb;
+	struct tipc_msg *hdr, tnlhdr;
+	struct sk_buff_head *queue = &l->transmq;
+	struct sk_buff_head tmpxq, tnlq;
+	u16 pktlen, pktcnt, seqno = l->snd_nxt;
 
-	if (!tunnel)
+	if (!tnl)
 		return;
 
-	tipc_msg_init(link_own_addr(l_ptr), &tunnel_hdr, TUNNEL_PROTOCOL,
-		      FAILOVER_MSG, INT_H_SIZE, l_ptr->addr);
+	skb_queue_head_init(&tnlq);
+	skb_queue_head_init(&tmpxq);
 
-	skb_queue_walk(&l_ptr->backlogq, skb) {
-		msg_set_seqno(buf_msg(skb), l_ptr->snd_nxt);
-		l_ptr->snd_nxt = mod(l_ptr->snd_nxt + 1);
-	}
-	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->transmq)) {
-		skb = tipc_buf_acquire(INT_H_SIZE);
-		if (skb) {
-			skb_copy_to_linear_data(skb, &tunnel_hdr, INT_H_SIZE);
-			msg_set_size(&tunnel_hdr, INT_H_SIZE);
-			__tipc_link_xmit_skb(tunnel, skb);
-		} else {
-			pr_warn("%sunable to send changeover msg\n",
-				link_co_err);
-		}
+	/* At least one packet required for safe algorithm => add dummy */
+	skb = tipc_msg_create(TIPC_LOW_IMPORTANCE, TIPC_DIRECT_MSG,
+			      BASIC_H_SIZE, 0, l->addr, link_own_addr(l),
+			      0, 0, TIPC_ERR_NO_PORT);
+	if (!skb) {
+		pr_warn("%sunable to create tunnel packet\n", link_co_err);
 		return;
 	}
-
-	split_bundles = (node_active_link(l_ptr->owner, 0) !=
-			 node_active_link(l_ptr->owner, 0));
-
-	skb_queue_walk(&l_ptr->transmq, skb) {
-		struct tipc_msg *msg = buf_msg(skb);
-
-		if ((msg_user(msg) == MSG_BUNDLER) && split_bundles) {
-			struct tipc_msg *m = msg_get_wrapped(msg);
-			unchar *pos = (unchar *)m;
-
-			msgcount = msg_msgcnt(msg);
-			while (msgcount--) {
-				msg_set_seqno(m, msg_seqno(msg));
-				tipc_link_tunnel_xmit(l_ptr, &tunnel_hdr, m,
-						      msg_link_selector(m));
-				pos += align(msg_size(m));
-				m = (struct tipc_msg *)pos;
-			}
-		} else {
-			tipc_link_tunnel_xmit(l_ptr, &tunnel_hdr, msg,
-					      msg_link_selector(msg));
-		}
-	}
-}
-
-/* tipc_link_dup_queue_xmit(): A second link has become active. Tunnel a
- * duplicate of the first link's send queue via the new link. This way, we
- * are guaranteed that currently queued packets from a socket are delivered
- * before future traffic from the same socket, even if this is using the
- * new link. The last arriving copy of each duplicate packet is dropped at
- * the receiving end by the regular protocol check, so packet cardinality
- * and sequence order is preserved per sender/receiver socket pair.
- * Owner node is locked.
- */
-void tipc_link_dup_queue_xmit(struct tipc_link *link,
-			      struct tipc_link *tnl)
-{
-	struct sk_buff *skb;
-	struct tipc_msg tnl_hdr;
-	struct sk_buff_head *queue = &link->transmq;
-	int mcnt;
-	u16 seqno;
-
-	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_tail(&tnlq, skb);
+	tipc_link_xmit(l, &tnlq, &tmpxq);
+	__skb_queue_purge(&tmpxq);
+
+	/* Initialize reusable tunnel packet header */
+	tipc_msg_init(link_own_addr(l), &tnlhdr, TUNNEL_PROTOCOL,
+		      mtyp, INT_H_SIZE, l->addr);
+	pktcnt = skb_queue_len(&l->transmq) + skb_queue_len(&l->backlogq);
+	msg_set_msgcnt(&tnlhdr, pktcnt);
+	msg_set_bearer_id(&tnlhdr, l->peer_bearer_id);
+tnl:
+	/* Wrap each packet into a tunnel packet */
 	skb_queue_walk(queue, skb) {
-		struct sk_buff *outskb;
-		struct tipc_msg *msg = buf_msg(skb);
-		u32 len = msg_size(msg);
-
-		msg_set_ack(msg, mod(link->rcv_nxt - 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);
+		hdr = buf_msg(skb);
+		if (queue == &l->backlogq)
+			msg_set_seqno(hdr, seqno++);
+		pktlen = msg_size(hdr);
+		msg_set_size(&tnlhdr, pktlen + INT_H_SIZE);
+		tnlskb = tipc_buf_acquire(pktlen + INT_H_SIZE);
+		if (!tnlskb) {
+			pr_warn("%sunable to send packet\n", link_co_err);
 			return;
 		}
-		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;
+		skb_copy_to_linear_data(tnlskb, &tnlhdr, INT_H_SIZE);
+		skb_copy_to_linear_data_offset(tnlskb, INT_H_SIZE, hdr, pktlen);
+		__skb_queue_tail(&tnlq, tnlskb);
 	}
-	if (queue == &link->backlogq)
-		return;
-	seqno = link->snd_nxt;
-	skb_queue_walk(&link->backlogq, skb) {
-		msg_set_seqno(buf_msg(skb), seqno);
-		seqno = mod(seqno + 1);
-	}
-	queue = &link->backlogq;
-	goto tunnel_queue;
-}
-
-/*  tipc_link_failover_rcv(): Receive a tunnelled FAILOVER_MSG packet
- *  Owner node is locked.
- */
-static int tipc_link_failover_rcv(struct tipc_link *link,
-				  struct sk_buff **skb)
-{
-	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;
-	int rc = 0;
-
-	if (msg_type(msg) != FAILOVER_MSG) {
-		pr_warn("%sunknown tunnel pkt received\n", link_co_err);
-		goto exit;
+	if (queue != &l->backlogq) {
+		queue = &l->backlogq;
+		goto tnl;
 	}
-	if (bearer_id >= MAX_BEARERS)
-		goto exit;
-
-	if (bearer_id == link->bearer_id)
-		goto exit;
-
-	pl = link->owner->links[bearer_id].link;
-
-	if (link->failover_pkts == FIRST_FAILOVER)
-		link->failover_pkts = msg_msgcnt(msg);
-
-	/* Should we expect an inner packet? */
-	if (!link->failover_pkts)
-		goto exit;
 
-	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;
+	tipc_link_xmit(tnl, &tnlq, xmitq);
 
-	/* 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++;
-		if (!tipc_buf_append(&link->failover_skb, &iskb) &&
-		    !link->failover_skb) {
-			link->exec_mode = TIPC_LINK_BLOCKED;
-			rc |= TIPC_LINK_DOWN_EVT;
-		}
+	if (mtyp == FAILOVER_MSG) {
+		tnl->drop_point = l->rcv_nxt;
+		tnl->failover_reasm_skb = l->reasm_buf;
+		l->reasm_buf = NULL;
+		l->exec_mode = TIPC_LINK_BLOCKED;
 	}
-exit:
-	if (!link->failover_pkts && pl)
-		pl->exec_mode = TIPC_LINK_OPEN;
-	kfree_skb(*skb);
-	*skb = iskb;
-	return rc;
 }
 
 /* tipc_link_proto_rcv(): receive link level protocol message :
@@ -1593,7 +1375,7 @@ static int tipc_link_proto_rcv(struct tipc_link *l, struct sk_buff *skb,
 
 		/* If NACK, retransmit will now start at right position */
 		if (nacked_gap) {
-			rc |= tipc_link_retransm(l, nacked_gap, xmitq);
+			rc = tipc_link_retransm(l, nacked_gap, xmitq);
 			l->stats.recv_nacks++;
 		}
 		tipc_link_advance_backlog(l, xmitq);
-- 
cgit v1.2.3


From 5045f7b9009f1455268b98cecbcc271663934c85 Mon Sep 17 00:00:00 2001
From: Jon Paul Maloy <jon.maloy@ericsson.com>
Date: Thu, 30 Jul 2015 18:24:20 -0400
Subject: tipc: move protocol message sending away from link FSM

The implementation of the link FSM currently takes decisions about and
sends out link protocol messages. This is unnecessary, since such
actions are not the result of any link state change, and are even
decided based on non-FSM state information ("silent_intv_cnt").

We now move the sending of unicast link protocol messages to the
function tipc_link_timeout(), and the initial broadcast synchronization
message to tipc_node_link_up(). The latter is done because a link
instance should not need to know whether it is the first or second
link to a destination. Such information is now restricted to and
handled by the link aggregation layer in node.c

Tested-by: Ying Xue <ying.xue@windriver.com>
Signed-off-by: Jon Maloy <jon.maloy@ericsson.com>
Signed-off-by: David S. Miller <davem@davemloft.net>
---
 net/tipc/link.c | 51 ++++++++++++++++++++++++++++++---------------------
 1 file changed, 30 insertions(+), 21 deletions(-)

(limited to 'net/tipc/link.c')

diff --git a/net/tipc/link.c b/net/tipc/link.c
index d5f4005f388f..9a3ccf910c49 100644
--- a/net/tipc/link.c
+++ b/net/tipc/link.c
@@ -134,8 +134,6 @@ static void tipc_link_build_proto_msg(struct tipc_link *l, int mtyp, bool probe,
 				      struct sk_buff_head *xmitq);
 static void link_reset_statistics(struct tipc_link *l_ptr);
 static void link_print(struct tipc_link *l_ptr, const char *str);
-static void tipc_link_build_bcast_sync_msg(struct tipc_link *l,
-					   struct sk_buff_head *xmitq);
 static void tipc_link_sync_rcv(struct tipc_node *n, struct sk_buff *buf);
 static int tipc_link_input(struct tipc_link *l, struct sk_buff *skb);
 static bool tipc_data_input(struct tipc_link *l, struct sk_buff *skb);
@@ -245,8 +243,8 @@ struct tipc_link *tipc_link_create(struct tipc_node *n_ptr,
  * Give a newly added peer node the sequence number where it should
  * start receiving and acking broadcast packets.
  */
-static void tipc_link_build_bcast_sync_msg(struct tipc_link *l,
-					   struct sk_buff_head *xmitq)
+void tipc_link_build_bcast_sync_msg(struct tipc_link *l,
+				    struct sk_buff_head *xmitq)
 {
 	struct sk_buff *skb;
 	struct sk_buff_head list;
@@ -272,7 +270,7 @@ static void tipc_link_build_bcast_sync_msg(struct tipc_link *l,
 static int tipc_link_fsm_evt(struct tipc_link *l, int evt,
 			     struct sk_buff_head *xmitq)
 {
-	int mtyp = 0, rc = 0;
+	int rc = 0;
 	struct tipc_link *pl;
 	enum {
 		LINK_RESET     = 1,
@@ -380,17 +378,7 @@ static int tipc_link_fsm_evt(struct tipc_link *l, int evt,
 	}
 	if (actions & LINK_ACTIVATE)
 		rc = TIPC_LINK_UP_EVT;
-	if (actions & (SND_STATE | SND_PROBE))
-		mtyp = STATE_MSG;
-	if (actions & SND_RESET)
-		mtyp = RESET_MSG;
-	if (actions & SND_ACTIVATE)
-		mtyp = ACTIVATE_MSG;
-	if (actions & (SND_PROBE | SND_STATE | SND_RESET | SND_ACTIVATE))
-		tipc_link_build_proto_msg(l, mtyp, actions & SND_PROBE,
-					  0, 0, 0, xmitq);
-	if (actions & SND_BCAST_SYNC)
-		tipc_link_build_bcast_sync_msg(l, xmitq);
+
 	return rc;
 }
 
@@ -440,16 +428,37 @@ static void link_profile_stats(struct tipc_link *l)
 int tipc_link_timeout(struct tipc_link *l, struct sk_buff_head *xmitq)
 {
 	int rc = 0;
+	int mtyp = STATE_MSG;
+	bool xmit = false;
+	bool prb = false;
 
 	if (l->exec_mode == TIPC_LINK_BLOCKED)
 		return rc;
 
 	link_profile_stats(l);
-	if (l->silent_intv_cnt)
-		rc = tipc_link_fsm_evt(l, SILENCE_EVT, xmitq);
-	else if (link_working(l) && tipc_bclink_acks_missing(l->owner))
-		tipc_link_build_proto_msg(l, STATE_MSG, 0, 0, 0, 0, xmitq);
-	l->silent_intv_cnt++;
+
+	if (l->state == TIPC_LINK_WORKING) {
+		if (!l->silent_intv_cnt) {
+			if (tipc_bclink_acks_missing(l->owner))
+				xmit = true;
+		} else if (l->silent_intv_cnt <= l->abort_limit) {
+			xmit = true;
+			prb = true;
+		} else {
+			l->exec_mode = TIPC_LINK_BLOCKED;
+			rc |= TIPC_LINK_DOWN_EVT;
+		}
+		l->silent_intv_cnt++;
+	} else if (l->state == TIPC_LINK_RESETTING) {
+		xmit = true;
+		mtyp = RESET_MSG;
+	} else if (l->state == TIPC_LINK_ESTABLISHING) {
+		xmit = true;
+		mtyp = ACTIVATE_MSG;
+	}
+	if (xmit)
+		tipc_link_build_proto_msg(l, mtyp, prb, 0, 0, 0, xmitq);
+
 	return rc;
 }
 
-- 
cgit v1.2.3


From 662921cd0a53db4504838dfbb7d996f9e6e94001 Mon Sep 17 00:00:00 2001
From: Jon Paul Maloy <jon.maloy@ericsson.com>
Date: Thu, 30 Jul 2015 18:24:21 -0400
Subject: tipc: merge link->exec_mode and link->state into one FSM

Until now, we have been handling link failover and synchronization
by using an additional link state variable, "exec_mode". This variable
is not independent of the link FSM state, something causing a risk of
inconsistencies, apart from the fact that it clutters the code.

The conditions are now in place to define a new link FSM that covers
all existing use cases, including failover and synchronization, and
eliminate the "exec_mode" field altogether. The FSM must also support
non-atomic resetting of links, which will be introduced later.

The new link FSM is shown below, with 7 states and 8 events.
Only events leading to state change are shown as edges.

+------------------------------------+
|RESET_EVT                           |
|                                    |
|                             +--------------+
|           +-----------------|   SYNCHING   |-----------------+
|           |FAILURE_EVT      +--------------+   PEER_RESET_EVT|
|           |                  A            |                  |
|           |                  |            |                  |
|           |                  |            |                  |
|           |                  |SYNCH_      |SYNCH_            |
|           |                  |BEGIN_EVT   |END_EVT           |
|           |                  |            |                  |
|           V                  |            V                  V
|    +-------------+          +--------------+          +------------+
|    |  RESETTING  |<---------|  ESTABLISHED |--------->| PEER_RESET |
|    +-------------+ FAILURE_ +--------------+ PEER_    +------------+
|           |        EVT        |    A         RESET_EVT       |
|           |                   |    |                         |
|           |                   |    |                         |
|           |    +--------------+    |                         |
|  RESET_EVT|    |RESET_EVT          |ESTABLISH_EVT            |
|           |    |                   |                         |
|           |    |                   |                         |
|           V    V                   |                         |
|    +-------------+          +--------------+        RESET_EVT|
+--->|    RESET    |--------->| ESTABLISHING |<----------------+
     +-------------+ PEER_    +--------------+
      |           A  RESET_EVT       |
      |           |                  |
      |           |                  |
      |FAILOVER_  |FAILOVER_         |FAILOVER_
      |BEGIN_EVT  |END_EVT           |BEGIN_EVT
      |           |                  |
      V           |                  |
     +-------------+                 |
     | FAILINGOVER |<----------------+
     +-------------+

These changes are fully backwards compatible.

Tested-by: Ying Xue <ying.xue@windriver.com>
Signed-off-by: Jon Maloy <jon.maloy@ericsson.com>
Signed-off-by: David S. Miller <davem@davemloft.net>
---
 net/tipc/link.c | 350 +++++++++++++++++++++++++++++++-------------------------
 1 file changed, 192 insertions(+), 158 deletions(-)

(limited to 'net/tipc/link.c')

diff --git a/net/tipc/link.c b/net/tipc/link.c
index 9a3ccf910c49..9840b03348e1 100644
--- a/net/tipc/link.c
+++ b/net/tipc/link.c
@@ -50,7 +50,6 @@
  */
 static const char *link_co_err = "Link tunneling error, ";
 static const char *link_rst_msg = "Resetting link ";
-static const char *link_unk_evt = "Unknown link event ";
 
 static const struct nla_policy tipc_nl_link_policy[TIPC_NLA_LINK_MAX + 1] = {
 	[TIPC_NLA_LINK_UNSPEC]		= { .type = NLA_UNSPEC },
@@ -85,46 +84,23 @@ static const struct nla_policy tipc_nl_prop_policy[TIPC_NLA_PROP_MAX + 1] = {
  */
 #define WILDCARD_SESSION 0x10000
 
-/* State value stored in 'failover_pkts'
+/* Link FSM states:
  */
-#define FIRST_FAILOVER 0xffffu
-
-/* Link FSM states and events:
- */
-enum {
-	TIPC_LINK_WORKING,
-	TIPC_LINK_PROBING,
-	TIPC_LINK_RESETTING,
-	TIPC_LINK_ESTABLISHING
-};
-
 enum {
-	PEER_RESET_EVT    = RESET_MSG,
-	ACTIVATE_EVT      = ACTIVATE_MSG,
-	TRAFFIC_EVT,      /* Any other valid msg from peer */
-	SILENCE_EVT       /* Peer was silent during last timer interval*/
+	LINK_ESTABLISHED     = 0xe,
+	LINK_ESTABLISHING    = 0xe  << 4,
+	LINK_RESET           = 0x1  << 8,
+	LINK_RESETTING       = 0x2  << 12,
+	LINK_PEER_RESET      = 0xd  << 16,
+	LINK_FAILINGOVER     = 0xf  << 20,
+	LINK_SYNCHING        = 0xc  << 24
 };
 
 /* Link FSM state checking routines
  */
-static int link_working(struct tipc_link *l)
-{
-	return l->state == TIPC_LINK_WORKING;
-}
-
-static int link_probing(struct tipc_link *l)
-{
-	return l->state == TIPC_LINK_PROBING;
-}
-
-static int link_resetting(struct tipc_link *l)
+static int link_is_up(struct tipc_link *l)
 {
-	return l->state == TIPC_LINK_RESETTING;
-}
-
-static int link_establishing(struct tipc_link *l)
-{
-	return l->state == TIPC_LINK_ESTABLISHING;
+	return l->state & (LINK_ESTABLISHED | LINK_SYNCHING);
 }
 
 static int tipc_link_proto_rcv(struct tipc_link *l, struct sk_buff *skb,
@@ -141,11 +117,29 @@ static bool tipc_data_input(struct tipc_link *l, struct sk_buff *skb);
 /*
  *  Simple non-static link routines (i.e. referenced outside this file)
  */
-int tipc_link_is_up(struct tipc_link *l_ptr)
+bool tipc_link_is_up(struct tipc_link *l)
 {
-	if (!l_ptr)
-		return 0;
-	return link_working(l_ptr) || link_probing(l_ptr);
+	return link_is_up(l);
+}
+
+bool tipc_link_is_reset(struct tipc_link *l)
+{
+	return l->state & (LINK_RESET | LINK_FAILINGOVER | LINK_ESTABLISHING);
+}
+
+bool tipc_link_is_synching(struct tipc_link *l)
+{
+	return l->state == LINK_SYNCHING;
+}
+
+bool tipc_link_is_failingover(struct tipc_link *l)
+{
+	return l->state == LINK_FAILINGOVER;
+}
+
+bool tipc_link_is_blocked(struct tipc_link *l)
+{
+	return l->state & (LINK_RESETTING | LINK_PEER_RESET | LINK_FAILINGOVER);
 }
 
 int tipc_link_is_active(struct tipc_link *l)
@@ -210,7 +204,7 @@ struct tipc_link *tipc_link_create(struct tipc_node *n_ptr,
 	l_ptr->tolerance = b_ptr->tolerance;
 	l_ptr->snd_nxt = 1;
 	l_ptr->rcv_nxt = 1;
-	l_ptr->state = TIPC_LINK_RESETTING;
+	l_ptr->state = LINK_RESET;
 
 	l_ptr->pmsg = (struct tipc_msg *)&l_ptr->proto_msg;
 	msg = l_ptr->pmsg;
@@ -265,120 +259,159 @@ void tipc_link_build_bcast_sync_msg(struct tipc_link *l,
  * tipc_link_fsm_evt - link finite state machine
  * @l: pointer to link
  * @evt: state machine event to be processed
- * @xmitq: queue to prepend created protocol message, if any
  */
-static int tipc_link_fsm_evt(struct tipc_link *l, int evt,
-			     struct sk_buff_head *xmitq)
+int tipc_link_fsm_evt(struct tipc_link *l, int evt)
 {
 	int rc = 0;
-	struct tipc_link *pl;
-	enum {
-		LINK_RESET     = 1,
-		LINK_ACTIVATE  = (1 << 1),
-		SND_PROBE      = (1 << 2),
-		SND_STATE      = (1 << 3),
-		SND_RESET      = (1 << 4),
-		SND_ACTIVATE   = (1 << 5),
-		SND_BCAST_SYNC = (1 << 6)
-	} actions = 0;
-
-	if (l->exec_mode == TIPC_LINK_BLOCKED)
-		return rc;
 
 	switch (l->state) {
-	case TIPC_LINK_WORKING:
+	case LINK_RESETTING:
 		switch (evt) {
-		case TRAFFIC_EVT:
-		case ACTIVATE_EVT:
+		case LINK_PEER_RESET_EVT:
+			l->state = LINK_PEER_RESET;
 			break;
-		case SILENCE_EVT:
-			l->state = TIPC_LINK_PROBING;
-			actions |= SND_PROBE;
+		case LINK_RESET_EVT:
+			l->state = LINK_RESET;
+			break;
+		case LINK_FAILURE_EVT:
+		case LINK_FAILOVER_BEGIN_EVT:
+		case LINK_ESTABLISH_EVT:
+		case LINK_FAILOVER_END_EVT:
+		case LINK_SYNCH_BEGIN_EVT:
+		case LINK_SYNCH_END_EVT:
+		default:
+			goto illegal_evt;
+		}
+		break;
+	case LINK_RESET:
+		switch (evt) {
+		case LINK_PEER_RESET_EVT:
+			l->state = LINK_ESTABLISHING;
 			break;
-		case PEER_RESET_EVT:
-			actions |= LINK_RESET | SND_ACTIVATE;
+		case LINK_FAILOVER_BEGIN_EVT:
+			l->state = LINK_FAILINGOVER;
+		case LINK_FAILURE_EVT:
+		case LINK_RESET_EVT:
+		case LINK_ESTABLISH_EVT:
+		case LINK_FAILOVER_END_EVT:
 			break;
+		case LINK_SYNCH_BEGIN_EVT:
+		case LINK_SYNCH_END_EVT:
 		default:
-			pr_debug("%s%u WORKING\n", link_unk_evt, evt);
+			goto illegal_evt;
 		}
 		break;
-	case TIPC_LINK_PROBING:
+	case LINK_PEER_RESET:
 		switch (evt) {
-		case TRAFFIC_EVT:
-		case ACTIVATE_EVT:
-			l->state = TIPC_LINK_WORKING;
+		case LINK_RESET_EVT:
+			l->state = LINK_ESTABLISHING;
 			break;
-		case PEER_RESET_EVT:
-			actions |= LINK_RESET | SND_ACTIVATE;
+		case LINK_PEER_RESET_EVT:
+		case LINK_ESTABLISH_EVT:
+		case LINK_FAILURE_EVT:
 			break;
-		case SILENCE_EVT:
-			if (l->silent_intv_cnt <= l->abort_limit) {
-				actions |= SND_PROBE;
-				break;
-			}
-			actions |= LINK_RESET | SND_RESET;
+		case LINK_SYNCH_BEGIN_EVT:
+		case LINK_SYNCH_END_EVT:
+		case LINK_FAILOVER_BEGIN_EVT:
+		case LINK_FAILOVER_END_EVT:
+		default:
+			goto illegal_evt;
+		}
+		break;
+	case LINK_FAILINGOVER:
+		switch (evt) {
+		case LINK_FAILOVER_END_EVT:
+			l->state = LINK_RESET;
 			break;
+		case LINK_PEER_RESET_EVT:
+		case LINK_RESET_EVT:
+		case LINK_ESTABLISH_EVT:
+		case LINK_FAILURE_EVT:
+			break;
+		case LINK_FAILOVER_BEGIN_EVT:
+		case LINK_SYNCH_BEGIN_EVT:
+		case LINK_SYNCH_END_EVT:
 		default:
-			pr_err("%s%u PROBING\n", link_unk_evt, evt);
+			goto illegal_evt;
 		}
 		break;
-	case TIPC_LINK_RESETTING:
+	case LINK_ESTABLISHING:
 		switch (evt) {
-		case TRAFFIC_EVT:
+		case LINK_ESTABLISH_EVT:
+			l->state = LINK_ESTABLISHED;
+			rc |= TIPC_LINK_UP_EVT;
 			break;
-		case ACTIVATE_EVT:
-			pl = node_active_link(l->owner, 0);
-			if (pl && link_probing(pl))
-				break;
-			l->state = TIPC_LINK_WORKING;
-			actions |= LINK_ACTIVATE;
-			if (!l->owner->working_links)
-				actions |= SND_BCAST_SYNC;
+		case LINK_FAILOVER_BEGIN_EVT:
+			l->state = LINK_FAILINGOVER;
+			break;
+		case LINK_PEER_RESET_EVT:
+		case LINK_RESET_EVT:
+		case LINK_FAILURE_EVT:
+		case LINK_SYNCH_BEGIN_EVT:
+		case LINK_FAILOVER_END_EVT:
+			break;
+		case LINK_SYNCH_END_EVT:
+		default:
+			goto illegal_evt;
+		}
+		break;
+	case LINK_ESTABLISHED:
+		switch (evt) {
+		case LINK_PEER_RESET_EVT:
+			l->state = LINK_PEER_RESET;
+			rc |= TIPC_LINK_DOWN_EVT;
+			break;
+		case LINK_FAILURE_EVT:
+			l->state = LINK_RESETTING;
+			rc |= TIPC_LINK_DOWN_EVT;
 			break;
-		case PEER_RESET_EVT:
-			l->state = TIPC_LINK_ESTABLISHING;
-			actions |= SND_ACTIVATE;
+		case LINK_RESET_EVT:
+			l->state = LINK_RESET;
 			break;
-		case SILENCE_EVT:
-			actions |= SND_RESET;
+		case LINK_ESTABLISH_EVT:
 			break;
+		case LINK_SYNCH_BEGIN_EVT:
+			l->state = LINK_SYNCHING;
+			break;
+		case LINK_SYNCH_END_EVT:
+		case LINK_FAILOVER_BEGIN_EVT:
+		case LINK_FAILOVER_END_EVT:
 		default:
-			pr_err("%s%u in RESETTING\n", link_unk_evt, evt);
+			goto illegal_evt;
 		}
 		break;
-	case TIPC_LINK_ESTABLISHING:
+	case LINK_SYNCHING:
 		switch (evt) {
-		case TRAFFIC_EVT:
-		case ACTIVATE_EVT:
-			pl = node_active_link(l->owner, 0);
-			if (pl && link_probing(pl))
-				break;
-			l->state = TIPC_LINK_WORKING;
-			actions |= LINK_ACTIVATE;
-			if (!l->owner->working_links)
-				actions |= SND_BCAST_SYNC;
+		case LINK_PEER_RESET_EVT:
+			l->state = LINK_PEER_RESET;
+			rc |= TIPC_LINK_DOWN_EVT;
+			break;
+		case LINK_FAILURE_EVT:
+			l->state = LINK_RESETTING;
+			rc |= TIPC_LINK_DOWN_EVT;
 			break;
-		case PEER_RESET_EVT:
+		case LINK_RESET_EVT:
+			l->state = LINK_RESET;
 			break;
-		case SILENCE_EVT:
-			actions |= SND_ACTIVATE;
+		case LINK_ESTABLISH_EVT:
+		case LINK_SYNCH_BEGIN_EVT:
 			break;
+		case LINK_SYNCH_END_EVT:
+			l->state = LINK_ESTABLISHED;
+			break;
+		case LINK_FAILOVER_BEGIN_EVT:
+		case LINK_FAILOVER_END_EVT:
 		default:
-			pr_err("%s%u ESTABLISHING\n", link_unk_evt, evt);
+			goto illegal_evt;
 		}
 		break;
 	default:
-		pr_err("Unknown link state %u/%u\n", l->state, evt);
-	}
-
-	/* Perform actions as decided by FSM */
-	if (actions & LINK_RESET) {
-		l->exec_mode = TIPC_LINK_BLOCKED;
-		rc = TIPC_LINK_DOWN_EVT;
+		pr_err("Unknown FSM state %x in %s\n", l->state, l->name);
 	}
-	if (actions & LINK_ACTIVATE)
-		rc = TIPC_LINK_UP_EVT;
-
+	return rc;
+illegal_evt:
+	pr_err("Illegal FSM event %x in state %x on link %s\n",
+	       evt, l->state, l->name);
 	return rc;
 }
 
@@ -432,12 +465,11 @@ int tipc_link_timeout(struct tipc_link *l, struct sk_buff_head *xmitq)
 	bool xmit = false;
 	bool prb = false;
 
-	if (l->exec_mode == TIPC_LINK_BLOCKED)
-		return rc;
-
 	link_profile_stats(l);
 
-	if (l->state == TIPC_LINK_WORKING) {
+	switch (l->state) {
+	case LINK_ESTABLISHED:
+	case LINK_SYNCHING:
 		if (!l->silent_intv_cnt) {
 			if (tipc_bclink_acks_missing(l->owner))
 				xmit = true;
@@ -445,17 +477,26 @@ int tipc_link_timeout(struct tipc_link *l, struct sk_buff_head *xmitq)
 			xmit = true;
 			prb = true;
 		} else {
-			l->exec_mode = TIPC_LINK_BLOCKED;
-			rc |= TIPC_LINK_DOWN_EVT;
+			rc |= tipc_link_fsm_evt(l, LINK_FAILURE_EVT);
 		}
 		l->silent_intv_cnt++;
-	} else if (l->state == TIPC_LINK_RESETTING) {
+		break;
+	case LINK_RESET:
 		xmit = true;
 		mtyp = RESET_MSG;
-	} else if (l->state == TIPC_LINK_ESTABLISHING) {
+		break;
+	case LINK_ESTABLISHING:
 		xmit = true;
 		mtyp = ACTIVATE_MSG;
+		break;
+	case LINK_RESETTING:
+	case LINK_PEER_RESET:
+	case LINK_FAILINGOVER:
+		break;
+	default:
+		break;
 	}
+
 	if (xmit)
 		tipc_link_build_proto_msg(l, mtyp, prb, 0, 0, 0, xmitq);
 
@@ -559,7 +600,7 @@ void tipc_link_reset(struct tipc_link *l)
 {
 	struct tipc_node *owner = l->owner;
 
-	l->state = TIPC_LINK_RESETTING;
+	tipc_link_fsm_evt(l, LINK_RESET_EVT);
 
 	/* Link is down, accept any session */
 	l->peer_session = WILDCARD_SESSION;
@@ -902,8 +943,7 @@ static int tipc_link_retransm(struct tipc_link *l, int retransm,
 		l->stale_count = 1;
 	} else if (++l->stale_count > 100) {
 		link_retransmit_failure(l, skb);
-		l->exec_mode = TIPC_LINK_BLOCKED;
-		return TIPC_LINK_DOWN_EVT;
+		return tipc_link_fsm_evt(l, LINK_FAILURE_EVT);
 	}
 	skb_queue_walk(&l->transmq, skb) {
 		if (!retransm)
@@ -1002,25 +1042,23 @@ static int tipc_link_input(struct tipc_link *l, struct sk_buff *skb)
 		l->stats.recv_bundled += msg_msgcnt(hdr);
 		while (tipc_msg_extract(skb, &iskb, &pos))
 			tipc_data_input(l, iskb);
-		return rc;
+		return 0;
 	} else if (usr == MSG_FRAGMENTER) {
 		l->stats.recv_fragments++;
 		if (tipc_buf_append(reasm_skb, &skb)) {
 			l->stats.recv_fragmented++;
 			tipc_data_input(l, skb);
 		} else if (!*reasm_skb) {
-			l->exec_mode = TIPC_LINK_BLOCKED;
-			l->state = TIPC_LINK_RESETTING;
-			rc = TIPC_LINK_DOWN_EVT;
+			return tipc_link_fsm_evt(l, LINK_FAILURE_EVT);
 		}
-		return rc;
+		return 0;
 	} else if (usr == BCAST_PROTOCOL) {
 		tipc_link_sync_rcv(node, skb);
-		return rc;
+		return 0;
 	}
 drop:
 	kfree_skb(skb);
-	return rc;
+	return 0;
 }
 
 static bool tipc_link_release_pkts(struct tipc_link *l, u16 acked)
@@ -1068,9 +1106,9 @@ int tipc_link_rcv(struct tipc_link *l, struct sk_buff *skb,
 			continue;
 		}
 
-		if (unlikely(!link_working(l))) {
-			rc = tipc_link_fsm_evt(l, TRAFFIC_EVT, xmitq);
-			if (!link_working(l)) {
+		if (unlikely(!link_is_up(l))) {
+			rc = tipc_link_fsm_evt(l, LINK_ESTABLISH_EVT);
+			if (!link_is_up(l)) {
 				kfree_skb(__skb_dequeue(arrvq));
 				return rc;
 			}
@@ -1192,7 +1230,7 @@ static void tipc_link_build_proto_msg(struct tipc_link *l, int mtyp, bool probe,
 	int node_up = l->owner->bclink.recv_permitted;
 
 	/* Don't send protocol message during reset or link failover */
-	if (l->exec_mode == TIPC_LINK_BLOCKED)
+	if (tipc_link_is_blocked(l))
 		return;
 
 	msg_set_type(hdr, mtyp);
@@ -1302,7 +1340,6 @@ tnl:
 		tnl->drop_point = l->rcv_nxt;
 		tnl->failover_reasm_skb = l->reasm_buf;
 		l->reasm_buf = NULL;
-		l->exec_mode = TIPC_LINK_BLOCKED;
 	}
 }
 
@@ -1323,7 +1360,7 @@ static int tipc_link_proto_rcv(struct tipc_link *l, struct sk_buff *skb,
 	char *if_name;
 	int rc = 0;
 
-	if (l->exec_mode == TIPC_LINK_BLOCKED)
+	if (tipc_link_is_blocked(l))
 		goto exit;
 
 	if (link_own_addr(l) > msg_prevnode(hdr))
@@ -1337,6 +1374,7 @@ static int tipc_link_proto_rcv(struct tipc_link *l, struct sk_buff *skb,
 		    (l->peer_session != WILDCARD_SESSION))
 			break;
 		/* fall thru' */
+
 	case ACTIVATE_MSG:
 
 		/* Complete own link name with peer's interface name */
@@ -1355,13 +1393,20 @@ static int tipc_link_proto_rcv(struct tipc_link *l, struct sk_buff *skb,
 		if (in_range(peers_prio, l->priority + 1, TIPC_MAX_LINK_PRI))
 			l->priority = peers_prio;
 
+		if (msg_type(hdr) == RESET_MSG) {
+			rc |= tipc_link_fsm_evt(l, LINK_PEER_RESET_EVT);
+		} else if (!link_is_up(l)) {
+			tipc_link_fsm_evt(l, LINK_PEER_RESET_EVT);
+			rc |= tipc_link_fsm_evt(l, LINK_ESTABLISH_EVT);
+		}
 		l->peer_session = msg_session(hdr);
 		l->peer_bearer_id = msg_bearer_id(hdr);
-		rc = tipc_link_fsm_evt(l, msg_type(hdr), xmitq);
 		if (l->mtu > msg_max_pkt(hdr))
 			l->mtu = msg_max_pkt(hdr);
 		break;
+
 	case STATE_MSG:
+
 		/* Update own tolerance if peer indicates a non-zero value */
 		if (in_range(peers_tol, TIPC_MIN_LINK_TOL, TIPC_MAX_LINK_TOL))
 			l->tolerance = peers_tol;
@@ -1370,11 +1415,11 @@ static int tipc_link_proto_rcv(struct tipc_link *l, struct sk_buff *skb,
 		l->stats.recv_states++;
 		if (msg_probe(hdr))
 			l->stats.recv_probes++;
-		rc = tipc_link_fsm_evt(l, TRAFFIC_EVT, xmitq);
-		if (!tipc_link_is_up(l))
+		rc = tipc_link_fsm_evt(l, LINK_ESTABLISH_EVT);
+		if (!link_is_up(l))
 			break;
 
-		/* Has peer sent packets we haven't received yet ? */
+		/* Send NACK if peer has sent pkts we haven't received yet */
 		if (more(peers_snd_nxt, l->rcv_nxt))
 			rcvgap = peers_snd_nxt - l->rcv_nxt;
 		if (rcvgap || (msg_probe(hdr)))
@@ -1387,6 +1432,7 @@ static int tipc_link_proto_rcv(struct tipc_link *l, struct sk_buff *skb,
 			rc = tipc_link_retransm(l, nacked_gap, xmitq);
 			l->stats.recv_nacks++;
 		}
+
 		tipc_link_advance_backlog(l, xmitq);
 		if (unlikely(!skb_queue_empty(&l->wakeupq)))
 			link_prepare_wakeup(l);
@@ -1463,19 +1509,7 @@ static void link_print(struct tipc_link *l, const char *str)
 	u16 head = hskb ? msg_seqno(buf_msg(hskb)) : l->snd_nxt;
 	u16 tail = l->snd_nxt - 1;
 
-	pr_info("%s Link <%s>:", str, l->name);
-
-	if (link_probing(l))
-		pr_cont(":P\n");
-	else if (link_establishing(l))
-		pr_cont(":E\n");
-	else if (link_resetting(l))
-		pr_cont(":R\n");
-	else if (link_working(l))
-		pr_cont(":W\n");
-	else
-		pr_cont("\n");
-
+	pr_info("%s Link <%s> state %x\n", str, l->name, l->state);
 	pr_info("XMTQ: %u [%u-%u], BKLGQ: %u, SNDNX: %u, RCVNX: %u\n",
 		skb_queue_len(&l->transmq), head, tail,
 		skb_queue_len(&l->backlogq), l->snd_nxt, l->rcv_nxt);
-- 
cgit v1.2.3


From 598411d70f85dcf5b5c6c2369cc48637c251b656 Mon Sep 17 00:00:00 2001
From: Jon Paul Maloy <jon.maloy@ericsson.com>
Date: Thu, 30 Jul 2015 18:24:23 -0400
Subject: tipc: make resetting of links non-atomic

In order to facilitate future improvements to the locking structure, we
want to make resetting and establishing of links non-atomic. I.e., the
functions tipc_node_link_up() and tipc_node_link_down() should be called
from outside the node lock context, and grab/release the node lock
themselves. This requires that we can freeze the link state from the
moment it is set to RESETTING or PEER_RESET in one lock context until
it is set to RESET or ESTABLISHING in a later context. The recently
introduced link FSM makes this possible, so we are now ready to introduce
the above change.

This commit implements this.

Tested-by: Ying Xue <ying.xue@windriver.com>
Signed-off-by: Jon Maloy <jon.maloy@ericsson.com>
Signed-off-by: David S. Miller <davem@davemloft.net>
---
 net/tipc/link.c | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

(limited to 'net/tipc/link.c')

diff --git a/net/tipc/link.c b/net/tipc/link.c
index 9840b03348e1..3a92924711a1 100644
--- a/net/tipc/link.c
+++ b/net/tipc/link.c
@@ -489,8 +489,8 @@ int tipc_link_timeout(struct tipc_link *l, struct sk_buff_head *xmitq)
 		xmit = true;
 		mtyp = ACTIVATE_MSG;
 		break;
-	case LINK_RESETTING:
 	case LINK_PEER_RESET:
+	case LINK_RESETTING:
 	case LINK_FAILINGOVER:
 		break;
 	default:
-- 
cgit v1.2.3


From 23d8335d786472021b5c733f228c7074208dcfa0 Mon Sep 17 00:00:00 2001
From: Jon Paul Maloy <jon.maloy@ericsson.com>
Date: Thu, 30 Jul 2015 18:24:24 -0400
Subject: tipc: remove implicit message delivery in node_unlock()

After the most recent changes, all access calls to a link which
may entail addition of messages to the link's input queue are
postpended by an explicit call to tipc_sk_rcv(), using a reference
to the correct queue.

This means that the potentially hazardous implicit delivery, using
tipc_node_unlock() in combination with a binary flag and a cached
queue pointer, now has become redundant.

This commit removes this implicit delivery mechanism both for regular
data messages and for binding table update messages.

Tested-by: Ying Xue <ying.xue@windriver.com>
Signed-off-by: Jon Maloy <jon.maloy@ericsson.com>
Signed-off-by: David S. Miller <davem@davemloft.net>
---
 net/tipc/link.c | 21 +++------------------
 1 file changed, 3 insertions(+), 18 deletions(-)

(limited to 'net/tipc/link.c')

diff --git a/net/tipc/link.c b/net/tipc/link.c
index 3a92924711a1..2aa19de715f6 100644
--- a/net/tipc/link.c
+++ b/net/tipc/link.c
@@ -559,8 +559,6 @@ void link_prepare_wakeup(struct tipc_link *l)
 			break;
 		skb_unlink(skb, &l->wakeupq);
 		skb_queue_tail(l->inputq, skb);
-		l->owner->inputq = l->inputq;
-		l->owner->action_flags |= TIPC_MSG_EVT;
 	}
 }
 
@@ -598,8 +596,6 @@ void tipc_link_purge_queues(struct tipc_link *l_ptr)
 
 void tipc_link_reset(struct tipc_link *l)
 {
-	struct tipc_node *owner = l->owner;
-
 	tipc_link_fsm_evt(l, LINK_RESET_EVT);
 
 	/* Link is down, accept any session */
@@ -611,14 +607,10 @@ void tipc_link_reset(struct tipc_link *l)
 	/* Prepare for renewed mtu size negotiation */
 	l->mtu = l->advertised_mtu;
 
-	/* Clean up all queues, except inputq: */
+	/* Clean up all queues: */
 	__skb_queue_purge(&l->transmq);
 	__skb_queue_purge(&l->deferdq);
-	if (!owner->inputq)
-		owner->inputq = l->inputq;
-	skb_queue_splice_init(&l->wakeupq, owner->inputq);
-	if (!skb_queue_empty(owner->inputq))
-		owner->action_flags |= TIPC_MSG_EVT;
+	skb_queue_splice_init(&l->wakeupq, l->inputq);
 
 	tipc_link_purge_backlog(l);
 	kfree_skb(l->reasm_buf);
@@ -972,7 +964,6 @@ static bool tipc_data_input(struct tipc_link *link, struct sk_buff *skb)
 {
 	struct tipc_node *node = link->owner;
 	struct tipc_msg *msg = buf_msg(skb);
-	u32 dport = msg_destport(msg);
 
 	switch (msg_user(msg)) {
 	case TIPC_LOW_IMPORTANCE:
@@ -980,17 +971,11 @@ static bool tipc_data_input(struct tipc_link *link, struct sk_buff *skb)
 	case TIPC_HIGH_IMPORTANCE:
 	case TIPC_CRITICAL_IMPORTANCE:
 	case CONN_MANAGER:
-		if (tipc_skb_queue_tail(link->inputq, skb, dport)) {
-			node->inputq = link->inputq;
-			node->action_flags |= TIPC_MSG_EVT;
-		}
+		skb_queue_tail(link->inputq, skb);
 		return true;
 	case NAME_DISTRIBUTOR:
 		node->bclink.recv_permitted = true;
-		node->namedq = link->namedq;
 		skb_queue_tail(link->namedq, skb);
-		if (skb_queue_len(link->namedq) == 1)
-			node->action_flags |= TIPC_NAMED_MSG_EVT;
 		return true;
 	case MSG_BUNDLER:
 	case TUNNEL_PROTOCOL:
-- 
cgit v1.2.3


From 9073fb8be3ee6f89492b8ea8f6d3902913a9fc91 Mon Sep 17 00:00:00 2001
From: Jon Paul Maloy <jon.maloy@ericsson.com>
Date: Thu, 30 Jul 2015 18:24:25 -0400
Subject: tipc: use temporary, non-protected skb queue for bundle reception

Currently, when we extract small messages from a message bundle, or
when many messages have accumulated in the link arrival queue, those
messages are added one by one to the lock protected link input queue.
This may increase contention with the reader of that queue, in
the function tipc_sk_rcv().

This commit introduces a temporary, unprotected input queue in
tipc_link_rcv() for such cases. Only when the arrival queue has been
emptied, and the function is ready to return, does it splice the whole
temporary queue into the real input queue.

Tested-by: Ying Xue <ying.xue@windriver.com>
Signed-off-by: Jon Maloy <jon.maloy@ericsson.com>
Signed-off-by: David S. Miller <davem@davemloft.net>
---
 net/tipc/link.c | 34 +++++++++++++++++++---------------
 1 file changed, 19 insertions(+), 15 deletions(-)

(limited to 'net/tipc/link.c')

diff --git a/net/tipc/link.c b/net/tipc/link.c
index 2aa19de715f6..d683fe9f68c8 100644
--- a/net/tipc/link.c
+++ b/net/tipc/link.c
@@ -111,8 +111,6 @@ static void tipc_link_build_proto_msg(struct tipc_link *l, int mtyp, bool probe,
 static void link_reset_statistics(struct tipc_link *l_ptr);
 static void link_print(struct tipc_link *l_ptr, const char *str);
 static void tipc_link_sync_rcv(struct tipc_node *n, struct sk_buff *buf);
-static int tipc_link_input(struct tipc_link *l, struct sk_buff *skb);
-static bool tipc_data_input(struct tipc_link *l, struct sk_buff *skb);
 
 /*
  *  Simple non-static link routines (i.e. referenced outside this file)
@@ -960,18 +958,18 @@ static int tipc_link_retransm(struct tipc_link *l, int retransm,
  * Consumes buffer if message is of right type
  * Node lock must be held
  */
-static bool tipc_data_input(struct tipc_link *link, struct sk_buff *skb)
+static bool tipc_data_input(struct tipc_link *link, struct sk_buff *skb,
+			    struct sk_buff_head *inputq)
 {
 	struct tipc_node *node = link->owner;
-	struct tipc_msg *msg = buf_msg(skb);
 
-	switch (msg_user(msg)) {
+	switch (msg_user(buf_msg(skb))) {
 	case TIPC_LOW_IMPORTANCE:
 	case TIPC_MEDIUM_IMPORTANCE:
 	case TIPC_HIGH_IMPORTANCE:
 	case TIPC_CRITICAL_IMPORTANCE:
 	case CONN_MANAGER:
-		skb_queue_tail(link->inputq, skb);
+		__skb_queue_tail(inputq, skb);
 		return true;
 	case NAME_DISTRIBUTOR:
 		node->bclink.recv_permitted = true;
@@ -993,7 +991,8 @@ static bool tipc_data_input(struct tipc_link *link, struct sk_buff *skb)
  *
  * Consumes buffer
  */
-static int tipc_link_input(struct tipc_link *l, struct sk_buff *skb)
+static int tipc_link_input(struct tipc_link *l, struct sk_buff *skb,
+			   struct sk_buff_head *inputq)
 {
 	struct tipc_node *node = l->owner;
 	struct tipc_msg *hdr = buf_msg(skb);
@@ -1016,7 +1015,7 @@ static int tipc_link_input(struct tipc_link *l, struct sk_buff *skb)
 		hdr = buf_msg(skb);
 		if (less(msg_seqno(hdr), l->drop_point))
 			goto drop;
-		if (tipc_data_input(l, skb))
+		if (tipc_data_input(l, skb, inputq))
 			return rc;
 		usr = msg_user(hdr);
 		reasm_skb = &l->failover_reasm_skb;
@@ -1026,13 +1025,13 @@ static int tipc_link_input(struct tipc_link *l, struct sk_buff *skb)
 		l->stats.recv_bundles++;
 		l->stats.recv_bundled += msg_msgcnt(hdr);
 		while (tipc_msg_extract(skb, &iskb, &pos))
-			tipc_data_input(l, iskb);
+			tipc_data_input(l, iskb, inputq);
 		return 0;
 	} else if (usr == MSG_FRAGMENTER) {
 		l->stats.recv_fragments++;
 		if (tipc_buf_append(reasm_skb, &skb)) {
 			l->stats.recv_fragmented++;
-			tipc_data_input(l, skb);
+			tipc_data_input(l, skb, inputq);
 		} else if (!*reasm_skb) {
 			return tipc_link_fsm_evt(l, LINK_FAILURE_EVT);
 		}
@@ -1070,10 +1069,13 @@ int tipc_link_rcv(struct tipc_link *l, struct sk_buff *skb,
 		  struct sk_buff_head *xmitq)
 {
 	struct sk_buff_head *arrvq = &l->deferdq;
+	struct sk_buff_head tmpq;
 	struct tipc_msg *hdr;
 	u16 seqno, rcv_nxt;
 	int rc = 0;
 
+	__skb_queue_head_init(&tmpq);
+
 	if (unlikely(!__tipc_skb_queue_sorted(arrvq, skb))) {
 		if (!(skb_queue_len(arrvq) % TIPC_NACK_INTV))
 			tipc_link_build_proto_msg(l, STATE_MSG, 0,
@@ -1095,7 +1097,7 @@ int tipc_link_rcv(struct tipc_link *l, struct sk_buff *skb,
 			rc = tipc_link_fsm_evt(l, LINK_ESTABLISH_EVT);
 			if (!link_is_up(l)) {
 				kfree_skb(__skb_dequeue(arrvq));
-				return rc;
+				goto exit;
 			}
 		}
 
@@ -1113,7 +1115,7 @@ int tipc_link_rcv(struct tipc_link *l, struct sk_buff *skb,
 		rcv_nxt = l->rcv_nxt;
 		if (unlikely(less(rcv_nxt, seqno))) {
 			l->stats.deferred_recv++;
-			return rc;
+			goto exit;
 		}
 
 		__skb_dequeue(arrvq);
@@ -1122,14 +1124,14 @@ int tipc_link_rcv(struct tipc_link *l, struct sk_buff *skb,
 		if (unlikely(more(rcv_nxt, seqno))) {
 			l->stats.duplicates++;
 			kfree_skb(skb);
-			return rc;
+			goto exit;
 		}
 
 		/* Packet can be delivered */
 		l->rcv_nxt++;
 		l->stats.recv_info++;
-		if (unlikely(!tipc_data_input(l, skb)))
-			rc = tipc_link_input(l, skb);
+		if (unlikely(!tipc_data_input(l, skb, &tmpq)))
+			rc = tipc_link_input(l, skb, &tmpq);
 
 		/* Ack at regular intervals */
 		if (unlikely(++l->rcv_unacked >= TIPC_MIN_LINK_WIN)) {
@@ -1139,6 +1141,8 @@ int tipc_link_rcv(struct tipc_link *l, struct sk_buff *skb,
 						  0, 0, 0, 0, xmitq);
 		}
 	}
+exit:
+	tipc_skb_queue_splice_tail(&tmpq, l->inputq);
 	return rc;
 }
 
-- 
cgit v1.2.3


From 440d8963cd590ec9387d76a36e60c02da9ed944d Mon Sep 17 00:00:00 2001
From: Jon Paul Maloy <jon.maloy@ericsson.com>
Date: Thu, 30 Jul 2015 18:24:26 -0400
Subject: tipc: clean up link creation

We simplify the link creation function tipc_link_create() and the way
the link struct it is connected to the node struct. In particular, we
remove the duplicate initialization of some fields which are anyway set
in tipc_link_reset().

Tested-by: Ying Xue <ying.xue@windriver.com>
Signed-off-by: Jon Maloy <jon.maloy@ericsson.com>
Signed-off-by: David S. Miller <davem@davemloft.net>
---
 net/tipc/link.c | 136 +++++++++++++++++++++++++-------------------------------
 1 file changed, 60 insertions(+), 76 deletions(-)

(limited to 'net/tipc/link.c')

diff --git a/net/tipc/link.c b/net/tipc/link.c
index d683fe9f68c8..f067e5425560 100644
--- a/net/tipc/link.c
+++ b/net/tipc/link.c
@@ -147,87 +147,71 @@ int tipc_link_is_active(struct tipc_link *l)
 	return (node_active_link(n, 0) == l) || (node_active_link(n, 1) == l);
 }
 
+static u32 link_own_addr(struct tipc_link *l)
+{
+	return msg_prevnode(l->pmsg);
+}
+
 /**
  * tipc_link_create - create a new link
- * @n_ptr: pointer to associated node
- * @b_ptr: pointer to associated bearer
- * @media_addr: media address to use when sending messages over link
+ * @n: pointer to associated node
+ * @b: pointer to associated bearer
+ * @ownnode: identity of own node
+ * @peer: identity of peer node
+ * @maddr: media address to be used
+ * @inputq: queue to put messages ready for delivery
+ * @namedq: queue to put binding table update messages ready for delivery
+ * @link: return value, pointer to put the created link
  *
- * Returns pointer to link.
+ * Returns true if link was created, otherwise false
  */
-struct tipc_link *tipc_link_create(struct tipc_node *n_ptr,
-				   struct tipc_bearer *b_ptr,
-				   const struct tipc_media_addr *media_addr,
-				   struct sk_buff_head *inputq,
-				   struct sk_buff_head *namedq)
+bool tipc_link_create(struct tipc_node *n, struct tipc_bearer *b, u32 session,
+		      u32 ownnode, u32 peer, struct tipc_media_addr *maddr,
+		      struct sk_buff_head *inputq, struct sk_buff_head *namedq,
+		      struct tipc_link **link)
 {
-	struct tipc_net *tn = net_generic(n_ptr->net, tipc_net_id);
-	struct tipc_link *l_ptr;
-	struct tipc_msg *msg;
+	struct tipc_link *l;
+	struct tipc_msg *hdr;
 	char *if_name;
-	char addr_string[16];
-	u32 peer = n_ptr->addr;
 
-	if (n_ptr->link_cnt >= MAX_BEARERS) {
-		tipc_addr_string_fill(addr_string, n_ptr->addr);
-		pr_err("Cannot establish %uth link to %s. Max %u allowed.\n",
-		       n_ptr->link_cnt, addr_string, MAX_BEARERS);
-		return NULL;
-	}
+	l = kzalloc(sizeof(*l), GFP_ATOMIC);
+	if (!l)
+		return false;
+	*link = l;
 
-	if (n_ptr->links[b_ptr->identity].link) {
-		tipc_addr_string_fill(addr_string, n_ptr->addr);
-		pr_err("Attempt to establish second link on <%s> to %s\n",
-		       b_ptr->name, addr_string);
-		return NULL;
-	}
+	/* Note: peer i/f name is completed by reset/activate message */
+	if_name = strchr(b->name, ':') + 1;
+	sprintf(l->name, "%u.%u.%u:%s-%u.%u.%u:unknown",
+		tipc_zone(ownnode), tipc_cluster(ownnode), tipc_node(ownnode),
+		if_name, tipc_zone(peer), tipc_cluster(peer), tipc_node(peer));
 
-	l_ptr = kzalloc(sizeof(*l_ptr), GFP_ATOMIC);
-	if (!l_ptr) {
-		pr_warn("Link creation failed, no memory\n");
-		return NULL;
-	}
-	l_ptr->addr = peer;
-	if_name = strchr(b_ptr->name, ':') + 1;
-	sprintf(l_ptr->name, "%u.%u.%u:%s-%u.%u.%u:unknown",
-		tipc_zone(tn->own_addr), tipc_cluster(tn->own_addr),
-		tipc_node(tn->own_addr),
-		if_name,
-		tipc_zone(peer), tipc_cluster(peer), tipc_node(peer));
-		/* note: peer i/f name is updated by reset/activate message */
-	memcpy(&l_ptr->media_addr, media_addr, sizeof(*media_addr));
-	l_ptr->owner = n_ptr;
-	l_ptr->peer_session = WILDCARD_SESSION;
-	l_ptr->bearer_id = b_ptr->identity;
-	l_ptr->tolerance = b_ptr->tolerance;
-	l_ptr->snd_nxt = 1;
-	l_ptr->rcv_nxt = 1;
-	l_ptr->state = LINK_RESET;
-
-	l_ptr->pmsg = (struct tipc_msg *)&l_ptr->proto_msg;
-	msg = l_ptr->pmsg;
-	tipc_msg_init(tn->own_addr, msg, LINK_PROTOCOL, RESET_MSG, INT_H_SIZE,
-		      l_ptr->addr);
-	msg_set_size(msg, sizeof(l_ptr->proto_msg));
-	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->snd_nxt = 1;
-	__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);
-	l_ptr->inputq = inputq;
-	l_ptr->namedq = namedq;
-	skb_queue_head_init(l_ptr->inputq);
-	link_reset_statistics(l_ptr);
-	tipc_node_attach_link(n_ptr, l_ptr);
-	return l_ptr;
+	l->addr = peer;
+	l->media_addr = maddr;
+	l->owner = n;
+	l->peer_session = WILDCARD_SESSION;
+	l->bearer_id = b->identity;
+	l->tolerance = b->tolerance;
+	l->net_plane = b->net_plane;
+	l->advertised_mtu = b->mtu;
+	l->mtu = b->mtu;
+	l->priority = b->priority;
+	tipc_link_set_queue_limits(l, b->window);
+	l->inputq = inputq;
+	l->namedq = namedq;
+	l->state = LINK_RESETTING;
+	l->pmsg = (struct tipc_msg *)&l->proto_msg;
+	hdr = l->pmsg;
+	tipc_msg_init(ownnode, hdr, LINK_PROTOCOL, RESET_MSG, INT_H_SIZE, peer);
+	msg_set_size(hdr, sizeof(l->proto_msg));
+	msg_set_session(hdr, session);
+	msg_set_bearer_id(hdr, l->bearer_id);
+	strcpy((char *)msg_data(hdr), if_name);
+	__skb_queue_head_init(&l->transmq);
+	__skb_queue_head_init(&l->backlogq);
+	__skb_queue_head_init(&l->deferdq);
+	skb_queue_head_init(&l->wakeupq);
+	skb_queue_head_init(l->inputq);
+	return true;
 }
 
 /* tipc_link_build_bcast_sync_msg() - synchronize broadcast link endpoints.
@@ -643,7 +627,7 @@ int __tipc_link_xmit(struct net *net, struct tipc_link *link,
 	u16 ack = mod(link->rcv_nxt - 1);
 	u16 seqno = link->snd_nxt;
 	u16 bc_last_in = link->owner->bclink.last_in;
-	struct tipc_media_addr *addr = &link->media_addr;
+	struct tipc_media_addr *addr = link->media_addr;
 	struct sk_buff_head *transmq = &link->transmq;
 	struct sk_buff_head *backlogq = &link->backlogq;
 	struct sk_buff *skb, *bskb;
@@ -809,7 +793,7 @@ void tipc_link_push_packets(struct tipc_link *link)
 		link->rcv_unacked = 0;
 		__skb_queue_tail(&link->transmq, skb);
 		tipc_bearer_send(link->owner->net, link->bearer_id,
-				 skb, &link->media_addr);
+				 skb, link->media_addr);
 	}
 	link->snd_nxt = seqno;
 }
@@ -912,7 +896,7 @@ void tipc_link_retransmit(struct tipc_link *l_ptr, struct sk_buff *skb,
 		msg_set_ack(msg, mod(l_ptr->rcv_nxt - 1));
 		msg_set_bcast_ack(msg, l_ptr->owner->bclink.last_in);
 		tipc_bearer_send(l_ptr->owner->net, l_ptr->bearer_id, skb,
-				 &l_ptr->media_addr);
+				 l_ptr->media_addr);
 		retransmits--;
 		l_ptr->stats.retransmitted++;
 	}
@@ -1200,7 +1184,7 @@ void tipc_link_proto_xmit(struct tipc_link *l, u32 msg_typ, int probe_msg,
 	skb = __skb_dequeue(&xmitq);
 	if (!skb)
 		return;
-	tipc_bearer_send(l->owner->net, l->bearer_id, skb, &l->media_addr);
+	tipc_bearer_send(l->owner->net, l->bearer_id, skb, l->media_addr);
 	l->rcv_unacked = 0;
 	kfree_skb(skb);
 }
-- 
cgit v1.2.3


From 5ae2f8e6857968d6dddbd3879ed0a32b860e02d1 Mon Sep 17 00:00:00 2001
From: Jon Paul Maloy <jon.maloy@ericsson.com>
Date: Thu, 20 Aug 2015 02:12:55 -0400
Subject: tipc: interrupt link synchronization when a link goes down

When we introduced the new link failover/synch mechanism
in commit 6e498158a827fd515b514842e9a06bdf0f75ab86
("tipc: move link synch and failover to link aggregation level"),
we missed the case when the non-tunnel link goes down during the link
synchronization period. In this case the tunnel link will remain in
state LINK_SYNCHING, something leading to unpredictable behavior when
the failover procedure is initiated.

In this commit, we ensure that the node and remaining link goes
back to regular communication state (SELF_UP_PEER_UP/LINK_ESTABLISHED)
when one of the parallel links goes down. We also ensure that we don't
re-enter synch mode if subsequent SYNCH packets arrive on the remaining
link.

Reviewed-by: Ying Xue <ying.xue@windriver.com>
Signed-off-by: Jon Maloy <jon.maloy@ericsson.com>
Signed-off-by: David S. Miller <davem@davemloft.net>
---
 net/tipc/link.c | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

(limited to 'net/tipc/link.c')

diff --git a/net/tipc/link.c b/net/tipc/link.c
index f067e5425560..7058c86f5e48 100644
--- a/net/tipc/link.c
+++ b/net/tipc/link.c
@@ -351,11 +351,11 @@ int tipc_link_fsm_evt(struct tipc_link *l, int evt)
 			l->state = LINK_RESET;
 			break;
 		case LINK_ESTABLISH_EVT:
+		case LINK_SYNCH_END_EVT:
 			break;
 		case LINK_SYNCH_BEGIN_EVT:
 			l->state = LINK_SYNCHING;
 			break;
-		case LINK_SYNCH_END_EVT:
 		case LINK_FAILOVER_BEGIN_EVT:
 		case LINK_FAILOVER_END_EVT:
 		default:
-- 
cgit v1.2.3


From 2be80c2d87de789550982e74a11e9f9ff5940845 Mon Sep 17 00:00:00 2001
From: Jon Paul Maloy <jon.maloy@ericsson.com>
Date: Thu, 20 Aug 2015 02:12:56 -0400
Subject: tipc: fix stale link problem during synchronization

Recent changes to the link synchronization means that we can now just
drop packets arriving on the synchronizing link before the synch point
is reached. This has lead to significant simplifications to the
implementation, but also turns out to have a flip side that we need
to consider.

Under unlucky circumstances, the two endpoints may end up
repeatedly dropping each other's packets, while immediately
asking for retransmission of the same packets, just to drop
them once more. This pattern will eventually be broken when
the synch point is reached on the other link, but before that,
the endpoints may have arrived at the retransmission limit
(stale counter) that indicates that the link should be broken.
We see this happen at rare occasions.

The fix for this is to not ask for retransmissions when a link is in
state LINK_SYNCHING. The fact that the link has reached this state
means that it has already received the first SYNCH packet, and that it
knows the synch point. Hence, it doesn't need any more packets until the
other link has reached the synch point, whereafter it can go ahead and
ask for the missing packets.

However, because of the reduced traffic on the synching link that
follows this change, it may now take longer to discover that the
synch point has been reached. We compensate for this by letting all
packets, on any of the links, trig a check for synchronization
termination. This is possible because the packets themselves don't
contain any information that is needed for discovering this condition.

Reviewed-by: Ying Xue <ying.xue@windriver.com>
Signed-off-by: Jon Maloy <jon.maloy@ericsson.com>
Signed-off-by: David S. Miller <davem@davemloft.net>
---
 net/tipc/link.c | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

(limited to 'net/tipc/link.c')

diff --git a/net/tipc/link.c b/net/tipc/link.c
index 7058c86f5e48..75db07c78a69 100644
--- a/net/tipc/link.c
+++ b/net/tipc/link.c
@@ -1330,6 +1330,7 @@ static int tipc_link_proto_rcv(struct tipc_link *l, struct sk_buff *skb,
 	u16 peers_snd_nxt =  msg_next_sent(hdr);
 	u16 peers_tol = msg_link_tolerance(hdr);
 	u16 peers_prio = msg_linkprio(hdr);
+	u16 rcv_nxt = l->rcv_nxt;
 	char *if_name;
 	int rc = 0;
 
@@ -1393,7 +1394,7 @@ static int tipc_link_proto_rcv(struct tipc_link *l, struct sk_buff *skb,
 			break;
 
 		/* Send NACK if peer has sent pkts we haven't received yet */
-		if (more(peers_snd_nxt, l->rcv_nxt))
+		if (more(peers_snd_nxt, rcv_nxt) && !tipc_link_is_synching(l))
 			rcvgap = peers_snd_nxt - l->rcv_nxt;
 		if (rcvgap || (msg_probe(hdr)))
 			tipc_link_build_proto_msg(l, STATE_MSG, 0, rcvgap,
-- 
cgit v1.2.3