Fix sending hearbeats

Author: knizhnik

arbiter.c

@@ -107,7 +107,7 @@ static int         gateway;
 static bool        send_heartbeat;
 static timestamp_t last_sent_heartbeat;
 static TimeoutId   heartbeat_timer;
-static int         busy_socket;
+static nodemask_t  busy_mask;
 
 static void MtmTransSender(Datum arg);
 static void MtmTransReceiver(Datum arg);
@@ -243,23 +243,19 @@ static int MtmWaitSocket(int sd, bool forWrite, time_t timeoutMsec)
 static bool MtmWriteSocket(int sd, void const* buf, int size)
 {
     char* src = (char*)buf;
-	busy_socket = sd;
     while (size != 0) {
 		int rc = MtmWaitSocket(sd, true, MtmHeartbeatSendTimeout);
 		if (rc == 1) { 
 			while ((rc = send(sd, src, size, 0)) < 0 && errno == EINTR);			
 			if (rc < 0) {
-				busy_socket = -1;
 				return false;
 			}
 			size -= rc;
 			src += rc;
 		} else if (rc < 0) { 
-			busy_socket = -1;
 			return false;
 		}
     }
-	busy_socket = -1;
 	return true;
 }
 
@@ -271,8 +267,6 @@ static int MtmReadSocket(int sd, void* buf, int buf_size)
 		rc = MtmWaitSocket(sd, false, MtmHeartbeatSendTimeout);
 		if (rc == 1) { 
 			while ((rc = recv(sd, buf, buf_size, 0)) < 0 && errno == EINTR);			
-		} else { 
-			return 0;
 		}
 	}
 	return rc;
@@ -365,7 +359,7 @@ static void MtmSendHeartbeat()
 
 	for (i = 0; i < Mtm->nAllNodes; i++)
 	{
-		if (i+1 != MtmNodeId && sockets[i] != busy_socket 
+		if (i+1 != MtmNodeId && !BIT_CHECK(busy_mask, i)
 			&& (Mtm->status != MTM_ONLINE 
 				|| (sockets[i] >= 0 && !BIT_CHECK(Mtm->disabledNodeMask, i) && !BIT_CHECK(Mtm->reconnectMask, i))))
 		{ 
@@ -399,6 +393,7 @@ static int MtmConnectSocket(int node, int port, int timeout)
 	int sd;
 	timestamp_t start = MtmGetSystemTime();
 	char const* host = Mtm->nodes[node].con.hostName;
+	nodemask_t save_mask = busy_mask;
 
     sock_inet.sin_family = AF_INET;
 	sock_inet.sin_port = htons(port);
@@ -407,22 +402,24 @@ static int MtmConnectSocket(int node, int port, int timeout)
 		elog(LOG, "Arbiter failed to resolve host '%s' by name", host);
 		return -1;
 	}
-
+	BIT_SET(busy_mask, node);
+	
   Retry:
     while (1) {
 		int rc = -1;
 
 		sd = socket(AF_INET, SOCK_STREAM, 0);
 		if (sd < 0) {
 			elog(LOG, "Arbiter failed to create socket: %d", errno);
+			busy_mask = save_mask;
 			return -1;
 		}
 		rc = fcntl(sd, F_SETFL, O_NONBLOCK);
 		if (rc < 0) {
 			elog(LOG, "Arbiter failed to switch socket to non-blocking mode: %d", errno);
+			busy_mask = save_mask;
 			return -1;
 		}
-		busy_socket = sd;
 		for (i = 0; i < n_addrs; ++i) {
 			memcpy(&sock_inet.sin_addr, &addrs[i], sizeof sock_inet.sin_addr);
 			do {
@@ -438,17 +435,17 @@ static int MtmConnectSocket(int node, int port, int timeout)
 		}
 		if (errno != EINPROGRESS || start  + MSEC_TO_USEC(timeout) < MtmGetSystemTime()) {
 			elog(WARNING, "Arbiter failed to connect to %s:%d: error=%d", host, port, errno);
-			busy_socket = -1;
 			close(sd);
+			busy_mask = save_mask;
 			return -1;
 		} else {
 			rc = MtmWaitSocket(sd, true, MtmHeartbeatSendTimeout);
 			if (rc == 1) {
 				socklen_t optlen = sizeof(int); 
 				if (getsockopt(sd, SOL_SOCKET, SO_ERROR, (void*)&rc, &optlen) < 0) { 
 					elog(WARNING, "Arbiter failed to getsockopt for %s:%d: error=%d", host, port, errno);
-					busy_socket = -1;
 					close(sd);
+					busy_mask = save_mask;
 					return -1;
 				}
 				if (rc == 0) { 
@@ -491,6 +488,8 @@ static int MtmConnectSocket(int node, int port, int timeout)
 	MtmCheckResponse(&resp);
 	MtmUnlock();
 
+	busy_mask = save_mask;
+
 	return sd;
 }
 
@@ -533,6 +532,9 @@ static void MtmOpenConnections()
 
 static bool MtmSendToNode(int node, void const* buf, int size)
 {	
+	bool result = true;
+	nodemask_t save_mask = busy_mask;
+	BIT_SET(busy_mask, node);
 	while (true) {
 		if (sockets[node] >= 0 && BIT_CHECK(Mtm->reconnectMask, node)) {
 			elog(WARNING, "Arbiter is forced to reconnect to node %d", node+1); 
@@ -553,13 +555,17 @@ static bool MtmSendToNode(int node, void const* buf, int size)
 			sockets[node] = MtmConnectSocket(node, MtmArbiterPort + node + 1, MtmReconnectTimeout);
 			if (sockets[node] < 0) { 
 				MtmOnNodeDisconnect(node+1);
-				return false;
+				result = false;
+				break;
 			}
 			MTM_LOG3("Arbiter restablished connection with node %d", node+1);
 		} else { 
-			return true;
+			result = true;
+			break;
 		}
 	}
+	busy_mask = save_mask;
+	return result;
 }
 
 static int MtmReadFromNode(int node, void* buf, int buf_size)

multimaster.c

@@ -534,7 +534,11 @@ MtmAdjustOldestXid(TransactionId xid)
 				oldestSnapshot = Mtm->nodes[i].oldestSnapshot;
 			}
 		}
-		oldestSnapshot -= MtmVacuumDelay*USECS_PER_SEC;
+		if (oldestSnapshot > MtmVacuumDelay*USECS_PER_SEC) { 
+			oldestSnapshot -= MtmVacuumDelay*USECS_PER_SEC;
+		} else { 
+			oldestSnapshot = 0;
+		}
 
 		for (ts = Mtm->transListHead; 
 			 ts != NULL 
@@ -1570,7 +1574,16 @@ void MtmOnNodeDisconnect(int nodeId)
 	BIT_SET(Mtm->reconnectMask, nodeId-1);
 	MtmUnlock();
 
-	RaftableSet(psprintf("node-mask-%d", MtmNodeId), &Mtm->connectivityMask, sizeof Mtm->connectivityMask, false);
+	if (!RaftableSet(psprintf("node-mask-%d", MtmNodeId), &Mtm->connectivityMask, sizeof Mtm->connectivityMask, false))
+	{
+		elog(WARNING, "Disable node which is in minority according to RAFT");
+		MtmLock(LW_EXCLUSIVE);
+		if (Mtm->status == MTM_ONLINE) { 
+			MtmSwitchClusterMode(MTM_IN_MINORITY);
+		}
+		MtmUnlock();
+		return;
+	}
 
 	MtmSleep(MSEC_TO_USEC(MtmHeartbeatSendTimeout));
 

raftable.c

@@ -22,10 +22,10 @@ void* RaftableGet(char const* key, size_t* size, RaftableTimestamp* ts, bool now
 }
 
 
-void RaftableSet(char const* key, void const* value, size_t size, bool nowait)
+bool RaftableSet(char const* key, void const* value, size_t size, bool nowait)
 {
 	if (MtmUseRaftable) {
-		int tries = 10;
+		int tries = MtmHeartbeatRecvTimeout/MtmHeartbeatSendTimeout;
 		timestamp_t start, stop;
 		start = MtmGetSystemTime();
 		if (nowait) {
@@ -36,8 +36,8 @@ void RaftableSet(char const* key, void const* value, size_t size, bool nowait)
 				MtmCheckHeartbeat();
 				if (tries-- <= 0)
 				{
-					MTM_LOG1("RaftableSet nowait=%d, all attempts failed", nowait);
-					break;
+					elog(WARNING, "Failed to send data to raftable in %d msec", MtmHeartbeatRecvTimeout);
+					return false;
 				}
 			}
 		}
@@ -46,6 +46,7 @@ void RaftableSet(char const* key, void const* value, size_t size, bool nowait)
 			MTM_LOG1("Raftable set nowait=%d takes %ld microseconds", nowait, stop - start);
 		}
 	}		
+	return true;
 }
 
 bool RaftableCAS(char const* key, char const* value, bool nowait)

raftable_wrapper.h

@@ -20,8 +20,9 @@ extern void* RaftableGet(char const* key, size_t* size, RaftableTimestamp* ts, b
 /*
  * Set new value for the specified key. IF value is NULL, then key should be deleted.
  * If RAFT master is not accessible, then depending non value of "nowait" parameter, this funciton should either block until RAFT quorum is reached, either report error.
+ * Returns false if rafttable is in minority
  */
-extern void  RaftableSet(char const* key, void const* value, size_t size, bool nowait);
+extern bool RaftableSet(char const* key, void const* value, size_t size, bool nowait);
 
 /* 
  * If key doesn't exists or its value is not equal to the specified value then store this value and return true.