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

@@ -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 @@ static bool tipc_link_failover_rcv(struct tipc_link *link,
 	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]) {

net/tipc/link.h

@@ -146,9 +146,7 @@ struct tipc_link {
 	u32 addr;
 	char name[TIPC_MAX_LINK_NAME];
 	struct tipc_media_addr media_addr;
-	struct timer_list timer;
 	struct tipc_node *owner;
-	struct kref ref;
 
 	/* Management and link supervision data */
 	u32 peer_session;

net/tipc/node.c

@@ -44,6 +44,7 @@
 static void node_lost_contact(struct tipc_node *n_ptr);
 static void node_established_contact(struct tipc_node *n_ptr);
 static void tipc_node_delete(struct tipc_node *node);
+static void tipc_node_timeout(unsigned long data);
 
 struct tipc_sock_conn {
 	u32 port;
@@ -145,11 +146,27 @@ struct tipc_node *tipc_node_create(struct net *net, u32 addr)
 	n_ptr->active_links[0] = INVALID_BEARER_ID;
 	n_ptr->active_links[1] = INVALID_BEARER_ID;
 	tipc_node_get(n_ptr);
+	setup_timer(&n_ptr->timer, tipc_node_timeout, (unsigned long)n_ptr);
+	n_ptr->keepalive_intv = U32_MAX;
 exit:
 	spin_unlock_bh(&tn->node_list_lock);
 	return n_ptr;
 }
 
+static void tipc_node_calculate_timer(struct tipc_node *n, struct tipc_link *l)
+{
+	unsigned long tol = l->tolerance;
+	unsigned long intv = ((tol / 4) > 500) ? 500 : tol / 4;
+	unsigned long keepalive_intv = msecs_to_jiffies(intv);
+
+	/* Link with lowest tolerance determines timer interval */
+	if (keepalive_intv < n->keepalive_intv)
+		n->keepalive_intv = keepalive_intv;
+
+	/* Ensure link's abort limit corresponds to current interval */
+	l->abort_limit = l->tolerance / jiffies_to_msecs(n->keepalive_intv);
+}
+
 static void tipc_node_delete(struct tipc_node *node)
 {
 	list_del_rcu(&node->list);
@@ -163,8 +180,11 @@ void tipc_node_stop(struct net *net)
 	struct tipc_node *node, *t_node;
 
 	spin_lock_bh(&tn->node_list_lock);
-	list_for_each_entry_safe(node, t_node, &tn->node_list, list)
+	list_for_each_entry_safe(node, t_node, &tn->node_list, list) {
+		if (del_timer(&node->timer))
+			tipc_node_put(node);
 		tipc_node_put(node);
+	}
 	spin_unlock_bh(&tn->node_list_lock);
 }
 
@@ -222,6 +242,38 @@ void tipc_node_remove_conn(struct net *net, u32 dnode, u32 port)
 	tipc_node_put(node);
 }
 
+/* tipc_node_timeout - handle expiration of node timer
+ */
+static void tipc_node_timeout(unsigned long data)
+{
+	struct tipc_node *n = (struct tipc_node *)data;
+	struct sk_buff_head xmitq;
+	struct tipc_link *l;
+	struct tipc_media_addr *maddr;
+	int bearer_id;
+	int rc = 0;
+
+	__skb_queue_head_init(&xmitq);
+
+	for (bearer_id = 0; bearer_id < MAX_BEARERS; bearer_id++) {
+		tipc_node_lock(n);
+		l = n->links[bearer_id].link;
+		if (l) {
+			/* Link tolerance may change asynchronously: */
+			tipc_node_calculate_timer(n, l);
+			rc = tipc_link_timeout(l, &xmitq);
+			if (rc & TIPC_LINK_DOWN_EVT)
+				tipc_link_reset(l);
+		}
+		tipc_node_unlock(n);
+		maddr = &n->links[bearer_id].maddr;
+		tipc_bearer_xmit(n->net, bearer_id, &xmitq, maddr);
+	}
+	if (!mod_timer(&n->timer, jiffies + n->keepalive_intv))
+		tipc_node_get(n);
+	tipc_node_put(n);
+}
+
 /**
  * tipc_node_link_up - handle addition of link
  *
@@ -335,10 +387,16 @@ bool tipc_node_update_dest(struct tipc_node *n,  struct tipc_bearer *b,
 	struct tipc_media_addr *curr = &n->links[b->identity].maddr;
 	struct sk_buff_head *inputq = &n->links[b->identity].inputq;
 
-	if (!l)
+	if (!l) {
 		l = tipc_link_create(n, b, maddr, inputq, &n->bclink.namedq);
 		if (!l)
 			return false;
+		tipc_node_calculate_timer(n, l);
+		if (n->link_cnt == 1) {
+			if (!mod_timer(&n->timer, jiffies + n->keepalive_intv))
+				tipc_node_get(n);
+		}
+	}
 	memcpy(&l->media_addr, maddr, sizeof(*maddr));
 	memcpy(curr, maddr, sizeof(*maddr));
 	tipc_link_reset(l);

net/tipc/node.h

@@ -140,6 +140,8 @@ struct tipc_node {
 	u32 link_id;
 	struct list_head publ_list;
 	struct list_head conn_sks;
+	unsigned long keepalive_intv;
+	struct timer_list timer;
 	struct rcu_head rcu;
 };