Merge branch 'tipc-next'

Jon Maloy says:

====================
tipc: separate link aggregation from link layer

We continue the work on separating the roles of the link aggregation and
link layers, as well as making code cleanups in general.

This second commit batch focuses on moving the orchestration of link
failover and synchronization to the node level, as well as preparing the
node lock structure for further future impovements. We also make some
changes to message delivery between link and socket layer, in order to
make this mechanism safer and less obscure.
====================

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

8 files changed, 1009 insertions, 1024 deletions

diff --git a/net/tipc/bearer.c b/net/tipc/bearer.c
index eae58a6b121c..ce9f7bfc0b92 100644
--- a/net/tipc/bearer.c
+++ b/net/tipc/bearer.c
@@ -343,7 +343,7 @@ restart:
 static int tipc_reset_bearer(struct net *net, struct tipc_bearer *b_ptr)
 {
 	pr_info("Resetting bearer <%s>\n", b_ptr->name);
-	tipc_link_delete_list(net, b_ptr->identity);
+	tipc_node_delete_links(net, b_ptr->identity);
 	tipc_disc_reset(net, b_ptr);
 	return 0;
 }
@@ -361,7 +361,7 @@ static void bearer_disable(struct net *net, struct tipc_bearer *b_ptr)
 	pr_info("Disabling bearer <%s>\n", b_ptr->name);
 	b_ptr->media->disable_media(b_ptr);
 
-	tipc_link_delete_list(net, b_ptr->identity);
+	tipc_node_delete_links(net, b_ptr->identity);
 	if (b_ptr->link_req)
 		tipc_disc_delete(b_ptr->link_req);
 
diff --git a/net/tipc/core.h b/net/tipc/core.h
index f4ed67778c54..b96b41eabf12 100644
--- a/net/tipc/core.h
+++ b/net/tipc/core.h
@@ -109,6 +109,11 @@ struct tipc_net {
 	atomic_t subscription_count;
 };
 
+static inline struct tipc_net *tipc_net(struct net *net)
+{
+	return net_generic(net, tipc_net_id);
+}
+
 static inline u16 mod(u16 x)
 {
 	return x & 0xffffu;
diff --git a/net/tipc/discover.c b/net/tipc/discover.c
index 164d08907d6f..d14e0a4aa9af 100644
--- a/net/tipc/discover.c
+++ b/net/tipc/discover.c
@@ -120,29 +120,24 @@ static void disc_dupl_alert(struct tipc_bearer *b_ptr, u32 node_addr,
  * @buf: buffer containing message
  * @bearer: bearer that message arrived on
  */
-void tipc_disc_rcv(struct net *net, struct sk_buff *buf,
+void tipc_disc_rcv(struct net *net, struct sk_buff *skb,
 		   struct tipc_bearer *bearer)
 {
 	struct tipc_net *tn = net_generic(net, tipc_net_id);
-	struct tipc_node *node;
 	struct tipc_media_addr maddr;
-	struct sk_buff *rbuf;
-	struct tipc_msg *msg = buf_msg(buf);
-	u32 ddom = msg_dest_domain(msg);
-	u32 onode = msg_prevnode(msg);
-	u32 net_id = msg_bc_netid(msg);
-	u32 mtyp = msg_type(msg);
-	u32 signature = msg_node_sig(msg);
-	u16 caps = msg_node_capabilities(msg);
-	bool addr_match = false;
-	bool sign_match = false;
-	bool link_up = false;
-	bool accept_addr = false;
-	bool accept_sign = false;
+	struct sk_buff *rskb;
+	struct tipc_msg *hdr = buf_msg(skb);
+	u32 ddom = msg_dest_domain(hdr);
+	u32 onode = msg_prevnode(hdr);
+	u32 net_id = msg_bc_netid(hdr);
+	u32 mtyp = msg_type(hdr);
+	u32 signature = msg_node_sig(hdr);
+	u16 caps = msg_node_capabilities(hdr);
 	bool respond = false;
+	bool dupl_addr = false;
 
-	bearer->media->msg2addr(bearer, &maddr, msg_media_addr(msg));
-	kfree_skb(buf);
+	bearer->media->msg2addr(bearer, &maddr, msg_media_addr(hdr));
+	kfree_skb(skb);
 
 	/* Ensure message from node is valid and communication is permitted */
 	if (net_id != tn->net_id)
@@ -164,91 +159,20 @@ void tipc_disc_rcv(struct net *net, struct sk_buff *buf,
 	if (!tipc_in_scope(bearer->domain, onode))
 		return;
 
-	node = tipc_node_create(net, onode);
-	if (!node)
-		return;
-	tipc_node_lock(node);
-	node->capabilities = caps;
-
-	/* Prepare to validate requesting node's signature and media address */
-	sign_match = (signature == node->signature);
-	tipc_node_check_dest(node, bearer, &link_up, &addr_match, &maddr);
-
-	/* These three flags give us eight permutations: */
-
-	if (sign_match && addr_match && link_up) {
-		/* All is fine. Do nothing. */
-	} else if (sign_match && addr_match && !link_up) {
-		/* Respond. The link will come up in due time */
-		respond = true;
-	} else if (sign_match && !addr_match && link_up) {
-		/* Peer has changed i/f address without rebooting.
-		 * If so, the link will reset soon, and the next
-		 * discovery will be accepted. So we can ignore it.
-		 * It may also be an cloned or malicious peer having
-		 * chosen the same node address and signature as an
-		 * existing one.
-		 * Ignore requests until the link goes down, if ever.
-		 */
+	tipc_node_check_dest(net, onode, bearer, caps, signature,
+			     &maddr, &respond, &dupl_addr);
+	if (dupl_addr)
 		disc_dupl_alert(bearer, onode, &maddr);
-	} else if (sign_match && !addr_match && !link_up) {
-		/* Peer link has changed i/f address without rebooting.
-		 * It may also be a cloned or malicious peer; we can't
-		 * distinguish between the two.
-		 * The signature is correct, so we must accept.
-		 */
-		accept_addr = true;
-		respond = true;
-	} else if (!sign_match && addr_match && link_up) {
-		/* Peer node rebooted. Two possibilities:
-		 *  - Delayed re-discovery; this link endpoint has already
-		 *    reset and re-established contact with the peer, before
-		 *    receiving a discovery message from that node.
-		 *    (The peer happened to receive one from this node first).
-		 *  - The peer came back so fast that our side has not
-		 *    discovered it yet. Probing from this side will soon
-		 *    reset the link, since there can be no working link
-		 *    endpoint at the peer end, and the link will re-establish.
-		 *  Accept the signature, since it comes from a known peer.
-		 */
-		accept_sign = true;
-	} else if (!sign_match && addr_match && !link_up) {
-		/*  The peer node has rebooted.
-		 *  Accept signature, since it is a known peer.
-		 */
-		accept_sign = true;
-		respond = true;
-	} else if (!sign_match && !addr_match && link_up) {
-		/* Peer rebooted with new address, or a new/duplicate peer.
-		 * Ignore until the link goes down, if ever.
-		 */
-		disc_dupl_alert(bearer, onode, &maddr);
-	} else if (!sign_match && !addr_match && !link_up) {
-		/* Peer rebooted with new address, or it is a new peer.
-		 * Accept signature and address.
-		*/
-		accept_sign = true;
-		accept_addr = true;
-		respond = true;
-	}
-
-	if (accept_sign)
-		node->signature = signature;
-
-	if (accept_addr && !tipc_node_update_dest(node, bearer, &maddr))
-		respond = false;
 
 	/* Send response, if necessary */
 	if (respond && (mtyp == DSC_REQ_MSG)) {
-		rbuf = tipc_buf_acquire(MAX_H_SIZE);
-		if (rbuf) {
-			tipc_disc_init_msg(net, rbuf, DSC_RESP_MSG, bearer);
-			tipc_bearer_send(net, bearer->identity, rbuf, &maddr);
-			kfree_skb(rbuf);
+		rskb = tipc_buf_acquire(MAX_H_SIZE);
+		if (rskb) {
+			tipc_disc_init_msg(net, rskb, DSC_RESP_MSG, bearer);
+			tipc_bearer_send(net, bearer->identity, rskb, &maddr);
+			kfree_skb(rskb);
 		}
 	}
-	tipc_node_unlock(node);
-	tipc_node_put(node);
 }
 
 /**
diff --git a/net/tipc/link.c b/net/tipc/link.c
index cc40aa6eb66c..f067e5425560 100644
--- a/net/tipc/link.c
+++ b/net/tipc/link.c
@@ -48,9 +48,8 @@
 /*
  * 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 ";
 
 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)
+static int link_is_up(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)
-{
-	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,
@@ -134,38 +110,34 @@ 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 void tipc_link_input(struct tipc_link *l, struct sk_buff *skb);
-static bool tipc_data_input(struct tipc_link *l, struct sk_buff *skb);
-static bool tipc_link_failover_rcv(struct tipc_link *l, struct sk_buff **skb);
 
 /*
- *  Simple link routines
+ *  Simple non-static link routines (i.e. referenced outside this file)
  */
-static unsigned int align(unsigned int i)
+bool tipc_link_is_up(struct tipc_link *l)
 {
-	return (i + 3) & ~3u;
+	return link_is_up(l);
 }
 
-static struct tipc_link *tipc_parallel_link(struct tipc_link *l)
+bool tipc_link_is_reset(struct tipc_link *l)
 {
-	struct tipc_node *n = l->owner;
+	return l->state & (LINK_RESET | LINK_FAILINGOVER | LINK_ESTABLISHING);
+}
 
-	if (node_active_link(n, 0) != l)
-		return node_active_link(n, 0);
-	return node_active_link(n, 1);
+bool tipc_link_is_synching(struct tipc_link *l)
+{
+	return l->state == LINK_SYNCHING;
 }
 
-/*
- *  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_failingover(struct tipc_link *l)
 {
-	if (!l_ptr)
-		return 0;
-	return link_working(l_ptr) || link_probing(l_ptr);
+	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)
@@ -175,113 +147,71 @@ int tipc_link_is_active(struct tipc_link *l)
 	return (node_active_link(n, 0) == l) || (node_active_link(n, 1) == l);
 }
 
-/**
- * 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
- *
- * Returns pointer to link.
- */
-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)
+static u32 link_own_addr(struct tipc_link *l)
 {
-	struct tipc_net *tn = net_generic(n_ptr->net, tipc_net_id);
-	struct tipc_link *l_ptr;
-	struct tipc_msg *msg;
-	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;
-	}
-
-	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;
-	}
-
-	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->state = TIPC_LINK_RESETTING;
-
-	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;
+	return msg_prevnode(l->pmsg);
 }
 
 /**
- * tipc_link_delete - Delete a link
- * @l: link to be deleted
+ * tipc_link_create - create a new 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 true if link was created, otherwise false
  */
-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)
+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(net, tipc_net_id);
-	struct tipc_link *link;
-	struct tipc_node *node;
+	struct tipc_link *l;
+	struct tipc_msg *hdr;
+	char *if_name;
 
-	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();
+	l = kzalloc(sizeof(*l), GFP_ATOMIC);
+	if (!l)
+		return false;
+	*link = l;
+
+	/* 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->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.
@@ -289,8 +219,8 @@ void tipc_link_delete_list(struct net *net, unsigned int bearer_id)
  * 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;
@@ -311,130 +241,159 @@ static 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 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),
-		SND_BCAST_SYNC = (1 << 6)
-	} actions = 0;
-
-	if (l->exec_mode == TIPC_LINK_BLOCKED)
-		return rc;
+	int rc = 0;
 
 	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 LINK_RESET_EVT:
+			l->state = LINK_RESET;
 			break;
-		case SILENCE_EVT:
-			l->state = TIPC_LINK_PROBING;
-			actions |= SND_PROBE;
+		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;
-			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 LINK_RESET_EVT:
+			l->state = LINK_RESET;
 			break;
-		case PEER_RESET_EVT:
-			l->state = TIPC_LINK_ESTABLISHING;
-			actions |= SND_ACTIVATE;
+		case LINK_ESTABLISH_EVT:
 			break;
-		case SILENCE_EVT:
-			actions |= SND_RESET;
+		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;
-			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;
-	}
-	if (actions & LINK_ACTIVATE) {
-		l->exec_mode = TIPC_LINK_OPEN;
-		rc |= TIPC_LINK_UP_EVT;
+		pr_err("Unknown FSM state %x in %s\n", l->state, l->name);
 	}
-	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;
+illegal_evt:
+	pr_err("Illegal FSM event %x in state %x on link %s\n",
+	       evt, l->state, l->name);
 	return rc;
 }
 
@@ -484,13 +443,45 @@ 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;
 
 	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++;
+
+	switch (l->state) {
+	case LINK_ESTABLISHED:
+	case LINK_SYNCHING:
+		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 {
+			rc |= tipc_link_fsm_evt(l, LINK_FAILURE_EVT);
+		}
+		l->silent_intv_cnt++;
+		break;
+	case LINK_RESET:
+		xmit = true;
+		mtyp = RESET_MSG;
+		break;
+	case LINK_ESTABLISHING:
+		xmit = true;
+		mtyp = ACTIVATE_MSG;
+		break;
+	case LINK_PEER_RESET:
+	case LINK_RESETTING:
+	case LINK_FAILINGOVER:
+		break;
+	default:
+		break;
+	}
+
+	if (xmit)
+		tipc_link_build_proto_msg(l, mtyp, prb, 0, 0, 0, xmitq);
+
 	return rc;
 }
 
@@ -550,8 +541,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;
 	}
 }
 
@@ -587,69 +576,36 @@ 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;
-	int was_active_link = tipc_link_is_active(l_ptr);
-	struct tipc_node *owner = l_ptr->owner;
-	struct tipc_link *pl = tipc_parallel_link(l_ptr);
-
-	msg_set_session(l_ptr->pmsg, ((msg_session(l_ptr->pmsg) + 1) & 0xffff));
+	tipc_link_fsm_evt(l, LINK_RESET_EVT);
 
 	/* Link is down, accept any session */
-	l_ptr->peer_session = WILDCARD_SESSION;
-
-	/* Prepare for renewed mtu size negotiation */
-	l_ptr->mtu = l_ptr->advertised_mtu;
-
-	l_ptr->state = TIPC_LINK_RESETTING;
-
-	if ((prev_state == TIPC_LINK_RESETTING) ||
-	    (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)) {
-		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);
-	if (!owner->inputq)
-		owner->inputq = l_ptr->inputq;
-	skb_queue_splice_init(&l_ptr->wakeupq, owner->inputq);
-	if (!skb_queue_empty(owner->inputq))
-		owner->action_flags |= TIPC_MSG_EVT;
-	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);
-}
+	l->peer_session = WILDCARD_SESSION;
 
-void tipc_link_activate(struct tipc_link *link)
-{
-	struct tipc_node *node = link->owner;
+	/* If peer is up, it only accepts an incremented session number */
+	msg_set_session(l->pmsg, msg_session(l->pmsg) + 1);
 
-	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);
+	/* Prepare for renewed mtu size negotiation */
+	l->mtu = l->advertised_mtu;
+
+	/* Clean up all queues: */
+	__skb_queue_purge(&l->transmq);
+	__skb_queue_purge(&l->deferdq);
+	skb_queue_splice_init(&l->wakeupq, l->inputq);
+
+	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);
 }
 
 /**
@@ -671,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;
@@ -792,20 +748,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
@@ -851,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;
 }
@@ -884,26 +826,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)
 {
@@ -920,7 +842,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;
@@ -975,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++;
 	}
@@ -996,7 +917,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);
-		return TIPC_LINK_DOWN_EVT;
+		return tipc_link_fsm_evt(l, LINK_FAILURE_EVT);
 	}
 	skb_queue_walk(&l->transmq, skb) {
 		if (!retransm)
@@ -1016,60 +937,27 @@ 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
  * 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);
-	u32 dport = msg_destport(msg);
 
-	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:
-		if (tipc_skb_queue_tail(link->inputq, skb, dport)) {
-			node->inputq = link->inputq;
-			node->