Fix wait socket function

Author: knizhnik

arbiter.c

@@ -220,18 +220,20 @@ static int MtmWaitSocket(int sd, bool forWrite, timestamp_t timeoutMsec)
 	fd_set set;
 	int rc;
 	timestamp_t deadline = MtmGetSystemTime() + MSEC_TO_USEC(timeoutMsec);
-    FD_ZERO(&set); 
-    FD_SET(sd, &set); 
+
 	do { 
 		timestamp_t now;
 		MtmCheckHeartbeat();
-		now = MtmGetSystemTime();
-		if (now > deadline) { 
+        now = MtmGetSystemTime();
+        if (now > deadline) { 
 			now = deadline;
 		}
+		FD_ZERO(&set); 
+		FD_SET(sd, &set); 
 		tv.tv_sec = (deadline - now)/USECS_PER_SEC; 
 		tv.tv_usec = (deadline - now)%USECS_PER_SEC;
 	} while ((rc = select(sd+1, forWrite ? NULL : &set, forWrite ? &set : NULL, NULL, &tv)) < 0 && errno == EINTR);
+
 	return rc;
 }
 
@@ -384,7 +386,7 @@ static void MtmSendHeartbeat()
 					|| !BIT_CHECK(Mtm->disabledNodeMask, i)
 					|| BIT_CHECK(Mtm->reconnectMask, i)))
 			{ 
-				if (!MtmSendToNode(i, &msg, sizeof(msg), MtmHeartbeatSendTimeout)) { 
+				if (!MtmSendToNode(i, &msg, sizeof(msg), MtmHeartbeatRecvTimeout)) { 
 					elog(LOG, "Arbiter failed to send heartbeat to node %d", i+1);
 				} else {
 					if (last_heartbeat_to_node[i] + MSEC_TO_USEC(MtmHeartbeatSendTimeout)*2 < now) { 
@@ -402,7 +404,7 @@ static void MtmSendHeartbeat()
 					MTM_LOG4("Send heartbeat to node %d with timestamp %lld", i+1, now);    
 				}
 			} else { 
-				MTM_LOG2("Do not send heartbeat to node %d, busy mask %lld, status %s", i+1, (long long) busy_mask, MtmNodeStatusMnem[Mtm->status]);
+				MTM_LOG2("Do not send heartbeat to node %d, busy mask %lld, status %s", i+1, busy_mask, MtmNodeStatusMnem[Mtm->status]);
 			}
 		}
 	}
@@ -828,7 +830,7 @@ static void MtmMonitor(Datum arg)
 	BackgroundWorkerInitializeConnection(MtmDatabaseName, NULL);
 
 	while (!stop) {
-		int rc = WaitLatch(&MyProc->procLatch, WL_TIMEOUT | WL_POSTMASTER_DEATH, MtmHeartbeatSendTimeout);
+		int rc = WaitLatch(&MyProc->procLatch, WL_TIMEOUT | WL_POSTMASTER_DEATH, MtmHeartbeatRecvTimeout);
 		if (rc & WL_POSTMASTER_DEATH) { 
 			break;
 		}
@@ -938,7 +940,7 @@ static void MtmReceiver(Datum arg)
 					Assert(node > 0 && node <= nNodes && node != MtmNodeId);
 
 					if (Mtm->nodes[node-1].connectivityMask != msg->connectivityMask) { 
-						elog(LOG, "Node %d changes it connectivity mask from %llx to %llx", node, (long long)Mtm->nodes[node-1].connectivityMask, (long long)msg->connectivityMask);
+						elog(LOG, "Node %d changes it connectivity mask from %llx to %llx", node, Mtm->nodes[node-1].connectivityMask, msg->connectivityMask);
 					}
 
 					Mtm->nodes[node-1].oldestSnapshot = msg->oldestSnapshot;
@@ -1002,11 +1004,11 @@ static void MtmReceiver(Datum arg)
 										replorigin_session_origin = InvalidRepOriginId;
 									} else { 
 										MTM_LOG1("Receive response for transaction %s -> %s, participants=%llx, voted=%llx", 
-												 msg->gid, MtmTxnStatusMnem[msg->status], (long long)ts->participantsMask, (long long)ts->votedMask);		
+												 msg->gid, MtmTxnStatusMnem[msg->status], ts->participantsMask, ts->votedMask);		
 									}
 								} else {
 									elog(LOG, "Receive response %s for transaction %s for node %d, votedMask %llx, participantsMask %llx",
-										 MtmTxnStatusMnem[msg->status], msg->gid, node, (long long)ts->votedMask, (long long)(ts->participantsMask & ~Mtm->disabledNodeMask));
+										 MtmTxnStatusMnem[msg->status], msg->gid, node, ts->votedMask, ts->participantsMask & ~Mtm->disabledNodeMask);
 									continue;
 								}
 							} else if (ts->status == TRANSACTION_STATUS_ABORTED && msg->status == TRANSACTION_STATUS_COMMITTED) {
@@ -1015,7 +1017,7 @@ static void MtmReceiver(Datum arg)
 								elog(WARNING, "Transaction %s is committed at node %d but aborted at node %d", msg->gid, MtmNodeId, node);
 							} else { 
 								elog(LOG, "Receive response %s for transaction %s status %s for node %d, votedMask %llx, participantsMask %llx",
-									 MtmTxnStatusMnem[msg->status], msg->gid, MtmTxnStatusMnem[ts->status], node, (long long)ts->votedMask, (long long)(ts->participantsMask & ~Mtm->disabledNodeMask) );
+									 MtmTxnStatusMnem[msg->status], msg->gid, MtmTxnStatusMnem[ts->status], node, ts->votedMask, ts->participantsMask & ~Mtm->disabledNodeMask);
 							}
 						}
 						continue;

multimaster--1.0.sql

@@ -43,7 +43,7 @@ AS 'MODULE_PATHNAME','mtm_get_nodes_state'
 LANGUAGE C;
 
 CREATE TYPE mtm.cluster_state AS ("status" text, "disabledNodeMask" bigint, "disconnectedNodeMask" bigint, "catchUpNodeMask" bigint, "liveNodes" integer, "allNodes" integer, "nActiveQueries" integer, "nPendingQueries" integer, "queueSize" bigint, "transCount" bigint, "timeShift" bigint, "recoverySlot" integer,
-"xidHashSize" bigint, "gidHashSize" bigint, "oldestXid" bigint, "configChanges" integer, "stalledNodeMask" bigint, "stoppedNodeMask" bigint);
+"xidHashSize" bigint, "gidHashSize" bigint, "oldestXid" bigint, "configChanges" integer, "stalledNodeMask" bigint, "stoppedNodeMask" bigint, "lastStatusChange" timestamp);
 
 CREATE TYPE mtm.trans_state AS ("status" text, "gid" text, "xid" bigint, "coordinator" integer, "gxid" bigint, "csn" timestamp, "snapshot" timestamp, "local" boolean, "prepared" boolean, "active" boolean, "twophase" boolean, "votingCompleted" boolean, "participants" bigint, "voted" bigint, "configChanges" integer);
 

multimaster.c

@@ -2128,7 +2128,7 @@ void MtmRefreshClusterStatus()
 		 * connectivity graph is stabilized.
 		 */
 		oldClique = newClique;		
-		MtmSleep(MSEC_TO_USEC(MtmHeartbeatSendTimeout)*2); /* double timeout to condider worst case when heartbeat send interval is added with refresh cluster status interval */
+		MtmSleep(MSEC_TO_USEC(MtmHeartbeatRecvTimeout)*2); /* double timeout to consider the worst case when heartbeat receive interval is added with refresh cluster status interval */
 		MtmBuildConnectivityMatrix(matrix);
 		newClique = MtmFindMaxClique(matrix, Mtm->nAllNodes, &cliqueSize);
 	} while (newClique != oldClique);
@@ -2232,7 +2232,7 @@ void MtmOnNodeDisconnect(int nodeId)
 	BIT_SET(SELF_CONNECTIVITY_MASK, nodeId-1);
 	BIT_SET(Mtm->reconnectMask, nodeId-1);
 	elog(LOG, "Disconnect node %d connectivity mask %llx", 
-		 nodeId, (long long)SELF_CONNECTIVITY_MASK);
+		 nodeId, SELF_CONNECTIVITY_MASK);
 	MtmUnlock();
 }
 
@@ -2242,7 +2242,7 @@ void MtmOnNodeDisconnect(int nodeId)
 void MtmOnNodeConnect(int nodeId)
 {
 	MtmLock(LW_EXCLUSIVE);	
-	elog(LOG, "Connect node %d connectivity mask %llx", nodeId, (long long)SELF_CONNECTIVITY_MASK);
+	elog(LOG, "Connect node %d connectivity mask %llx", nodeId, SELF_CONNECTIVITY_MASK);
 	BIT_CLEAR(SELF_CONNECTIVITY_MASK, nodeId-1);
 	BIT_SET(Mtm->reconnectMask, nodeId-1); /* force sender to reestablish connection and send heartbeat */
 	MtmUnlock();
@@ -2254,6 +2254,7 @@ void MtmOnNodeConnect(int nodeId)
 void MtmReconnectNode(int nodeId)
 {
 	MtmLock(LW_EXCLUSIVE);	
+	elog(LOG, "Reconnect node %d connectivity mask %llx", nodeId, SELF_CONNECTIVITY_MASK);
 	BIT_SET(Mtm->reconnectMask, nodeId-1);
 	MtmUnlock();
 }
@@ -3289,7 +3290,9 @@ void MtmFinishPreparedTransaction(MtmTransState* ts, bool commit)
 	MtmSetCurrentTransactionGID(ts->gid);
 	MtmTx.isActive = true;
 	FinishPreparedTransaction(ts->gid, commit);
-	
+	if (commit) { 
+		MTM_LOG1("Distributed transaction %s is committed", ts->gid);
+	}
 	if (!insideTransaction) { 
 		CommitTransactionCommand();
 		Assert(!MtmTx.isActive);
@@ -3326,15 +3329,21 @@ MtmReplicationMode MtmGetReplicationMode(int nodeId, sig_atomic_t volatile* shut
 			MtmUnlock();
 			return REPLMODE_EXIT;
 		}
-		/* We are not interested in receiving any deteriorated logical messages from recovered node, do recreate slot */
+		/* We are not interested in receiving any deteriorated logical messages from recovered node, so recreate slot */
 		if (BIT_CHECK(Mtm->disabledNodeMask, nodeId-1)) { 
 			mode = REPLMODE_CREATE_NEW;
 		}
 		MTM_LOG2("%d: receiver slot mode %s", MyProcPid, MtmNodeStatusMnem[Mtm->status]);
 		if (Mtm->status == MTM_RECOVERY) { 
 			mode = REPLMODE_RECOVERED;
-			if ((Mtm->recoverySlot == 0 && (Mtm->donorNodeId == MtmNodeId || Mtm->donorNodeId == nodeId))
-				|| Mtm->recoverySlot == nodeId) 
+			/* Choose node for recovery if 
+			 * 1. It is not chosen yet or the same node was chosen before
+			 * 2. It is donor node or there is no donor node
+			 * 3. Connections with all other live nodes were established
+			 */
+			if ((Mtm->recoverySlot == 0 || Mtm->recoverySlot == nodeId)
+				&& (Mtm->donorNodeId == MtmNodeId || Mtm->donorNodeId == nodeId)
+				&& (SELF_CONNECTIVITY_MASK & ~Mtm->disabledNodeMask) == 0)
 			{ 
 				/* Choose for recovery first available slot or slot of donor node (if any) */
 				if (Mtm->nAllNodes >= 3) { 
@@ -3354,6 +3363,8 @@ MtmReplicationMode MtmGetReplicationMode(int nodeId, sig_atomic_t volatile* shut
 				return REPLMODE_RECOVERY;
 			}
 		}
+		MTM_LOG1("Replication to node %d is pending: recovery node=%d, donor node=%d, connectivity mask=%llx, disabled mask=%llx", 
+				 nodeId, Mtm->recoverySlot, Mtm->donorNodeId, SELF_CONNECTIVITY_MASK, Mtm->disabledNodeMask);
 		MtmUnlock();
 		/* delay opening of other slots until recovery is completed */
 		MtmSleep(STATUS_POLL_DELAY);
@@ -3492,6 +3503,7 @@ MtmReplicationStartupHook(struct PGLogicalStartupHookArgs* args)
 			}
 		}
 	}
+	MTM_LOG1("Startup of logical replication to node %d", MtmReplicationNodeId);
 	MtmLock(LW_EXCLUSIVE);
 	if (BIT_CHECK(Mtm->stoppedNodeMask, MtmReplicationNodeId-1)) {		
 		elog(WARNING, "Stopped node %d tries to initiate recovery", MtmReplicationNodeId);
@@ -4074,6 +4086,7 @@ mtm_get_cluster_state(PG_FUNCTION_ARGS)
 	values[15] = Int32GetDatum(Mtm->nConfigChanges);
 	values[16] = Int64GetDatum(Mtm->stalledNodeMask);
 	values[17] = Int64GetDatum(Mtm->stoppedNodeMask);
+	values[18] = TimestampTzGetDatum(time_t_to_timestamptz(Mtm->nodes[MtmNodeId-1].lastStatusChangeTime/USECS_PER_SEC));
 
 	PG_RETURN_DATUM(HeapTupleGetDatum(heap_form_tuple(desc, values, nulls)));
 }
@@ -4456,6 +4469,7 @@ static bool MtmTwoPhaseCommit(MtmCurrentTrans* x)
 					elog(ERROR, "Transaction %s (%llu) is aborted by DTM", x->gid, (long64)x->xid);
 				} else {
 					FinishPreparedTransaction(x->gid, true);
+					MTM_LOG1("Distributed transaction %s is committed", x->gid);
 				}
 			}
 		}

multimaster.h

@@ -82,7 +82,7 @@
 
 #define Natts_mtm_trans_state   15
 #define Natts_mtm_nodes_state   16
-#define Natts_mtm_cluster_state 18
+#define Natts_mtm_cluster_state 19
 
 typedef ulong64 csn_t; /* commit serial number */
 #define INVALID_CSN  ((csn_t)-1)

pglogical_apply.c

@@ -703,6 +703,7 @@ process_remote_commit(StringInfo in)
 			MtmSetCurrentTransactionCSN(csn);
 			MtmSetCurrentTransactionGID(gid);
 			FinishPreparedTransaction(gid, true);
+			MTM_LOG1("Distributed transaction %s is committed", gid);
 			CommitTransactionCommand();
 			Assert(!MtmTransIsActive());
 			MtmEndSession(origin_node, true);