Correctly handle connect timeouts

Author: knizhnik

arbiter.c

@@ -366,14 +366,16 @@ static void MtmCheckHeartbeat()
 }
 
 
-static int MtmConnectSocket(char const* host, int port, int max_attempts)
+static int MtmConnectSocket(char const* host, int port, int timeout)
 {
     struct sockaddr_in sock_inet;
     unsigned addrs[MAX_ROUTES];
     unsigned i, n_addrs = sizeof(addrs) / sizeof(addrs[0]);
 	MtmHandshakeMessage req;
 	MtmArbiterMessage   resp;
 	int sd;
+	timestamp_t start = MtmGetSystemTime();
+	
 
     sock_inet.sin_family = AF_INET;
 	sock_inet.sin_port = htons(port);
@@ -390,7 +392,10 @@ static int MtmConnectSocket(char const* host, int port, int max_attempts)
 		if (sd < 0) {
 			elog(ERROR, "Arbiter failed to create socket: %d", errno);
 		}
-		fcntl(sd, F_SETFL, O_NONBLOCK);
+		rc = fcntl(sd, F_SETFL, O_NONBLOCK);
+		if (rc < 0) {
+			elog(ERROR, "Arbiter failed to switch socket to non-blocking mode: %d", errno);
+		}
 		busy_socket = sd;
 		for (i = 0; i < n_addrs; ++i) {
 			memcpy(&sock_inet.sin_addr, &addrs[i], sizeof sock_inet.sin_addr);
@@ -405,17 +410,19 @@ static int MtmConnectSocket(char const* host, int port, int max_attempts)
 		if (rc == 0) {
 			break;
 		}
-		if (errno != EINPROGRESS || max_attempts == 0) {
+		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);
 			return -1;
 		} else {
-			rc = MtmWaitSocket(sd, true, MtmConnectTimeout);
+			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);
 					return -1;
 				}
 				if (rc == 0) { 
@@ -426,8 +433,8 @@ static int MtmConnectSocket(char const* host, int port, int max_attempts)
 			} else { 
 				elog(WARNING, "Arbiter waiting socket to %s:%d: rc=%d, error=%d", host, port, rc, errno);
 			}
-			max_attempts -= 1;
-			MtmSleep(MSEC_TO_USEC(MtmConnectTimeout));
+			close(sd);
+			MtmSleep(MSEC_TO_USEC(MtmHeartbeatSendTimeout));
 		}
 	}
 	MtmSetSocketOptions(sd);
@@ -479,7 +486,7 @@ static void MtmOpenConnections()
 	}
 	for (i = 0; i < nNodes; i++) {
 		if (i+1 != MtmNodeId && i < Mtm->nAllNodes) { 
-			sockets[i] = MtmConnectSocket(Mtm->nodes[i].con.hostName, MtmArbiterPort + i + 1, MtmConnectAttempts);
+			sockets[i] = MtmConnectSocket(Mtm->nodes[i].con.hostName, MtmArbiterPort + i + 1, MtmConnectTimeout);
 			if (sockets[i] < 0) { 
 				MtmOnNodeDisconnect(i+1);
 			} 
@@ -511,7 +518,7 @@ static bool MtmSendToNode(int node, void const* buf, int size)
 				close(sockets[node]);
 				sockets[node] = -1;
 			}
-			sockets[node] = MtmConnectSocket(Mtm->nodes[node].con.hostName, MtmArbiterPort + node + 1, MtmReconnectAttempts);
+			sockets[node] = MtmConnectSocket(Mtm->nodes[node].con.hostName, MtmArbiterPort + node + 1, MtmReconnectTimeout);
 			if (sockets[node] < 0) { 
 				MtmOnNodeDisconnect(node+1);
 				return false;

multimaster.c

@@ -193,10 +193,9 @@ int   MtmNodes;
 int   MtmNodeId;
 int   MtmReplicationNodeId;
 int   MtmArbiterPort;
-int   MtmConnectAttempts;
 int   MtmConnectTimeout;
+int   MtmReconnectTimeout;
 int   MtmRaftPollDelay;
-int   MtmReconnectAttempts;
 int   MtmNodeDisableDelay;
 int   MtmTransSpillThreshold;
 int   MtmMaxNodes;
@@ -2030,9 +2029,9 @@ _PG_init(void)
 	DefineCustomIntVariable(
 		"multimaster.connect_timeout",
 		"Multimaster nodes connect timeout",
-		"Interval in milliseconds between connection attempts",
+		"Interval in milliseconds for establishing connection with cluster node",
 		&MtmConnectTimeout,
-		1000,
+		10000, /* 10 seconds */
 		1,
 		INT_MAX,
 		PGC_BACKEND,
@@ -2043,11 +2042,11 @@ _PG_init(void)
 	);
 
 	DefineCustomIntVariable(
-		"multimaster.raft_poll_delay",
-		"Multimaster delay of polling cluster state from Raftable after updating local node status",
-		"Timeout in milliseconds before polling state of nodes",
-		&MtmRaftPollDelay,
-		1000,
+		"multimaster.reconnect_timeout",
+		"Multimaster nodes reconnect timeout",
+		"Interval in milliseconds for establishing connection with cluster node",
+		&MtmReconnectTimeout,
+		5000, /* 5 seconds */
 		1,
 		INT_MAX,
 		PGC_BACKEND,
@@ -2058,11 +2057,11 @@ _PG_init(void)
 	);
 
 	DefineCustomIntVariable(
-		"multimaster.connect_attempts",
-		"Multimaster number of connect attemts",
-		"Maximal number of attempt to establish connection with other node after which multimaster is give up",
-		&MtmConnectAttempts,
-		10,
+		"multimaster.raft_poll_delay",
+		"Multimaster delay of polling cluster state from Raftable after updating local node status",
+		"Timeout in milliseconds before polling state of nodes",
+		&MtmRaftPollDelay,
+		1000,
 		1,
 		INT_MAX,
 		PGC_BACKEND,
@@ -2072,20 +2071,6 @@ _PG_init(void)
 		NULL
 	);
 
-	DefineCustomIntVariable(
-		"multimaster.reconnect_attempts",
-		"Multimaster number of reconnect attemts",
-		"Maximal number of attempt to reestablish connection with other node after which node is considered to be offline",
-		&MtmReconnectAttempts,
-		10,
-		1,
-		INT_MAX,
-		PGC_BACKEND,
-		0,
-		NULL,
-		NULL,
-		NULL
-	);
 
 	MtmSplitConnStrs();
     MtmStartReceivers();

multimaster.h

@@ -215,9 +215,8 @@ extern int   MtmReplicationNodeId;
 extern int   MtmNodes;
 extern int   MtmArbiterPort;
 extern char* MtmDatabaseName;
-extern int   MtmConnectAttempts;
 extern int   MtmConnectTimeout;
-extern int   MtmReconnectAttempts;
+extern int   MtmReconnectTimeout;
 extern int   MtmRaftPollDelay;
 extern int   MtmNodeDisableDelay;
 extern int   MtmTransSpillThreshold;