clusterer: Fix node pinging regression in 7d74f3d

While commit 7d74f3d fixed links going down during pinging, it
introduced a bug where node links could flip/flop between Up/Down state
intermittently on some setups.  This patch addresses the new issue.

(cherry picked from commit f98189a)

Author: liviuchircu

modules/clusterer/clusterer.c

@@ -350,10 +350,7 @@ static int msg_send_retry(bin_packet_t *packet, node_info_t *dest,
 	gettimeofday(&now, NULL);
 
 	/* sent a TCP BIN packet directly to @dest -> delay next ping */
-	lock_get(chosen_dest->lock);
-	if (chosen_dest->link_state == LS_UP)
-		chosen_dest->last_ping = now;
-	lock_release(chosen_dest->lock);
+	chosen_dest->last_sent = now;
 
 	return 0;
 }
@@ -1081,9 +1078,7 @@ void bin_rcv_cl_extra_packets(bin_packet_t *packet, int packet_type,
 	}
 
 	lock_get(node->lock);
-
-	/* bump "last pong" ts, since we fully read a valid TCP BIN packet */
-	node->last_pong = now;
+	node->last_recv = now;
 
 	if (!(node->flags & NODE_STATE_ENABLED)) {
 		lock_release(node->lock);
@@ -1217,9 +1212,8 @@ void bin_rcv_cl_packets(bin_packet_t *packet, int packet_type,
 		}
 
 		lock_get(node->lock);
-
-		/* bump "last pong" ts, since we fully read a valid TCP BIN packet */
-		node->last_pong = now;
+		if (packet_type != CLUSTERER_PONG)
+			node->last_recv = now;
 
 		if (!(node->flags & NODE_STATE_ENABLED)) {
 			lock_release(node->lock);
@@ -1365,9 +1359,7 @@ static void bin_rcv_mod_packets(bin_packet_t *packet, int packet_type,
 	}
 
 	lock_get(node->lock);
-
-	/* bump "last pong" ts, since we fully read a valid TCP BIN packet */
-	node->last_pong = now;
+	node->last_recv = now;
 
 	if (!(node->flags & NODE_STATE_ENABLED)) {
 		lock_release(node->lock);
@@ -1885,4 +1877,4 @@ unsigned long clusterer_get_num_nodes(int state)
 	lock_stop_read(cl_list_lock);
 
 	return nodecount;
-}
+}

modules/clusterer/clusterer_mod.c

@@ -70,6 +70,9 @@ void *cl_srg=NULL;
 
 str node_st_sr_ident = str_init("node_states");
 
+long ping_timeout_us;
+long ping_interval_us;
+
 /* module interface functions */
 static int mod_init(void);
 static int child_init(int rank);
@@ -411,6 +414,9 @@ static int mod_init(void)
 		LM_WARN("Invalid ping_timeout parameter, using default value\n");
 		ping_timeout = DEFAULT_PING_TIMEOUT;
 	}
+	ping_timeout_us = ping_timeout * 1000;
+	ping_interval_us = ping_interval * 1000000;
+
 	if (seed_fb_interval < 0) {
 		LM_WARN("Invalid seed_fallback_interval parameter, using default value\n");
 		seed_fb_interval = DEFAULT_SEED_FB_INTERVAL;

modules/clusterer/node_info.h

@@ -87,8 +87,8 @@ struct node_info {
 	/* fields protected by node lock */
 	clusterer_link_state link_state;	/* state of the "link" with this node */
 	int last_ping_state;				/* state(success/error) of the last ping sent to this node */
-	struct timeval last_ping;       	/* last ping sent to this node */
-	struct timeval last_pong;       	/* last pong received from this node */
+	struct timeval last_ping, last_sent;  /* last ping/packet sent to this node */
+	struct timeval last_pong, last_recv;  /* last pong/packet received from this node */
 	struct neighbour *neighbour_list;   /* list of directly reachable neighbours */
 	int ls_seq_no;                      /* sequence number of the last link state update */
 	int top_seq_no;                     /* sequence number of the last topology update message */

modules/clusterer/topology.c

@@ -28,11 +28,11 @@
 
 extern int ping_interval;
 extern int node_timeout;
-extern int ping_timeout;
+extern long ping_timeout_us, ping_interval_us;
 
 #define PING_REPLY_INTERVAL(_node) \
-	((_node)->last_ping.tv_sec*1000000 + (_node)->last_ping.tv_usec \
-	- (_node)->last_pong.tv_sec*1000000 - (_node)->last_pong.tv_usec)
+	((_node)->last_pong.tv_sec*1000000 + (_node)->last_pong.tv_usec \
+	- (_node)->last_ping.tv_sec*1000000 - (_node)->last_ping.tv_usec)
 
 static int send_ping(node_info_t *node, int req_node_list)
 {
@@ -184,7 +184,7 @@ static void do_action_trans_5(node_info_t *node, int *link_state_to_set,
 void heartbeats_timer(void)
 {
 	struct timeval now;
-	time_t last_ping_int, ping_reply_int;
+	time_t last_ping_int, last_sent_int, ping_reply_int;
 	cluster_info_t *clusters_it;
 	node_info_t *node;
 	int ev_actions_required[MAX_NO_CLUSTERS] = {0};
@@ -212,6 +212,7 @@ void heartbeats_timer(void)
 			gettimeofday(&now, NULL);
 			ping_reply_int = PING_REPLY_INTERVAL(node);
 			last_ping_int = TIME_DIFF(node->last_ping, now);
+			last_sent_int = TIME_DIFF(node->last_sent, now);
 
 			prev_ls = -1;
 			new_ls = -1;
@@ -224,16 +225,21 @@ void heartbeats_timer(void)
 				/* restart pinging sequence */
 				do_action_trans_0(node, &new_ls);
 			} else if (node->link_state == LS_RETRY_SEND_FAIL &&
-				last_ping_int >= (utime_t)ping_timeout*1000) {
+				last_ping_int >= (time_t)ping_timeout_us) {
 				CL_DBG("case 1: RETRY_SEND_FAIL and timeout\n");
 				prev_ls = node->link_state;
 				lock_release(node->lock);
 
 				/* failed to send previous ping, retry */
 				do_action_trans_1(node, &new_ls);
 			} else if ((node->link_state == LS_UP || node->link_state == LS_RESTARTED) &&
-				(ping_reply_int >= (time_t)ping_timeout*1000) &&
-				last_ping_int >= (utime_t)ping_timeout*1000) {
+					/* have yet to receive a ping reply, or it was unacceptably slow */
+				(ping_reply_int <= 0 || ping_reply_int >= (time_t)ping_timeout_us) &&
+					/* ... and we're pinging or haven't sent a recent BIN packet */
+				(node->link_state == LS_RESTARTED
+				     || last_sent_int >= (time_t)ping_timeout_us) &&
+					/* ... and a new ping packet is due */
+				last_ping_int >= (time_t)ping_timeout_us) {
 				CL_DBG("case 2: LS_UP and timeout\n");
 				prev_ls = -2;
 				lock_release(node->lock);
@@ -242,24 +248,25 @@ void heartbeats_timer(void)
 				do_action_trans_2(node, &new_ls);
 				ev_actions_required[no_clusters] = 1;
 			} else if (node->link_state == LS_RETRYING &&
-				(ping_reply_int >= (time_t)ping_timeout*1000) &&
-				last_ping_int >= (utime_t)ping_timeout*1000) {
+				(ping_reply_int <= 0 || ping_reply_int >= (time_t)ping_timeout_us) &&
+				last_ping_int >= (time_t)ping_timeout_us) {
 				CL_DBG("case 3: LS_RETRYING and timeout\n");
 				prev_ls = node->link_state;
 				lock_release(node->lock);
 
 				/* previous ping retry not replied, continue to retry */
 				do_action_trans_3(node, &new_ls);
 			} else if (node->link_state == LS_DOWN &&
-				last_ping_int >= (utime_t)node_timeout*1000000) {
+				last_ping_int >= (time_t)node_timeout*1000000) {
 				CL_DBG("case 4: LS_DOWN and timeout\n");
 				prev_ls = node->link_state;
 				lock_release(node->lock);
 
 				/* ping a failed node after node_timeout since last ping */
 				do_action_trans_4(node, &new_ls);
 			} else if (node->link_state == LS_UP &&
-				last_ping_int >= (utime_t)ping_interval*1000000) {
+				    last_sent_int >= (time_t)ping_interval_us &&
+				    last_ping_int >= (time_t)ping_interval_us) {
 				CL_DBG("case 5: LS_UP and timeout\n");
 				prev_ls = node->link_state;
 				lock_release(node->lock);
@@ -1346,29 +1353,39 @@ void handle_ping(bin_packet_t *received, node_info_t *src_node,
 void handle_pong(bin_packet_t *received, node_info_t *src_node,
 	struct timeval rcv_time, int *ev_actions_required)
 {
+	time_t last_recv_int;
 	int node_list[MAX_NO_NODES], i, nr_nodes;
 
 	bin_pop_int(received, &nr_nodes);
 	for (i=0; i<nr_nodes; i++)
 		bin_pop_int(received, &node_list[i]);
 
+	last_recv_int = TIME_DIFF(src_node->last_recv, rcv_time);
+
 	lock_get(src_node->lock);
 
 	src_node->last_pong = rcv_time;
+	src_node->last_recv = rcv_time;
 
 	/* check possible races between setting the appropriate state
 	 * after sending ping and receiving the reply */
 	if ((src_node->link_state == LS_RESTART_PINGING ||
 		src_node->link_state == LS_RETRY_SEND_FAIL ||
 		src_node->link_state == LS_DOWN) &&
 		src_node->last_ping_state == 0 &&
-		TIME_DIFF(src_node->last_ping, rcv_time) < (utime_t)ping_timeout*1000)
+		TIME_DIFF(src_node->last_ping, rcv_time) < (time_t)ping_timeout_us)
 		src_node->link_state = LS_TEMP;
 
 	/* if the node was retried and a reply was expected, it should be UP again */
-	if (src_node->link_state == LS_RESTARTED ||
+	if ((src_node->link_state == LS_RESTARTED ||
 		src_node->link_state == LS_RETRYING ||
-		src_node->link_state == LS_TEMP) {
+               src_node->link_state == LS_TEMP) &&
+			/* if either this PONG wasn't too late or we received
+			 *    *any* other type of BIN packet in the mean time */
+            ((PING_REPLY_INTERVAL(src_node) > 0 &&
+            PING_REPLY_INTERVAL(src_node) < (time_t)ping_timeout_us)
+			 || last_recv_int <= 0 || last_recv_int < (time_t)ping_timeout_us)) {
+
 		lock_release(src_node->lock);
 
 		set_link_w_neigh_up(src_node, nr_nodes, node_list);