Add functions for getting transaction state

Author: knizhnik

arbiter.c

@@ -382,6 +382,7 @@ static void MtmSendHeartbeat()
 					last_heartbeat_to_node[i] = now;
 					/* Connectivity mask can be cleared by MtmWatchdog: in this case sockets[i] >= 0 */
 					if (BIT_CHECK(Mtm->connectivityMask, i)) { 
+						MTM_LOG1("Force reconnect to node %d", i+1);    
 						close(sockets[i]);
 						sockets[i] = -1;
 						MtmReconnectNode(i+1); /* set reconnect mask to force node reconnent */
@@ -484,6 +485,7 @@ static int MtmConnectSocket(int node, int port, int timeout)
 				elog(WARNING, "Arbiter waiting socket to %s:%d: rc=%d, error=%d", host, port, rc, errno);
 			}
 			close(sd);
+			MtmCheckHeartbeat();
 			MtmSleep(MSEC_TO_USEC(MtmHeartbeatSendTimeout));
 		}
 	}
@@ -827,6 +829,7 @@ static void MtmReceiver(Datum arg)
 	MtmBuffer* rxBuffer = (MtmBuffer*)palloc0(sizeof(MtmBuffer)*nNodes);
 	timestamp_t lastHeartbeatCheck = MtmGetSystemTime();
 	timestamp_t now;
+	timestamp_t selectTimeout = MtmHeartbeatRecvTimeout;
 
 #if USE_EPOLL
 	struct epoll_event* events = (struct epoll_event*)palloc(sizeof(struct epoll_event)*nNodes);
@@ -857,7 +860,7 @@ static void MtmReceiver(Datum arg)
 
 	while (!stop) {
 #if USE_EPOLL
-        n = epoll_wait(epollfd, events, nNodes, MtmHeartbeatRecvTimeout);
+        n = epoll_wait(epollfd, events, nNodes, selectTimeout);
 		if (n < 0) { 
 			if (errno == EINTR) { 
 				continue;
@@ -871,16 +874,15 @@ static void MtmReceiver(Datum arg)
 				MtmDisconnect(i);
 			} 
 		}
-		now = MtmGetSystemTime();
 		for (j = 0; j < n; j++) {
 			if (events[j].events & EPOLLIN)  
 #else
         fd_set events;
 		do { 
 			struct timeval tv;
 			events = inset;
-			tv.tv_sec = MtmHeartbeatRecvTimeout/1000;
-			tv.tv_usec = MtmHeartbeatRecvTimeout%1000*1000;
+			tv.tv_sec = selectTimeout/1000;
+			tv.tv_usec = selectTimeout%1000*1000;
 			do { 
 				n = select(max_fd+1, &events, NULL, NULL, &tv);
 			} while (n < 0 && errno == EINTR);
@@ -889,7 +891,6 @@ static void MtmReceiver(Datum arg)
 		if (n < 0) {
 			elog(ERROR, "Arbiter failed to select sockets: %d", errno);
 		}
-		now = MtmGetSystemTime();
 		for (i = 0; i < nNodes; i++) { 
 			if (sockets[i] >= 0 && FD_ISSET(sockets[i], &events)) 
 #endif
@@ -1070,8 +1071,8 @@ static void MtmReceiver(Datum arg)
 							break;						   
 						  case MSG_ABORTED:
 							if (ts->status == TRANSACTION_STATUS_COMMITTED) { 
-								elog(WARNING, "Receive ABORTED response for already committed transaction %d from node %d",
-									 ts->xid, node);
+								elog(WARNING, "Receive ABORTED response for already committed transaction %d (%s) from node %d",
+									 ts->xid, ts->gid, node);
 								continue;
 							}
 							if (ts->status != TRANSACTION_STATUS_ABORTED) { 
@@ -1084,8 +1085,12 @@ static void MtmReceiver(Datum arg)
 							break;
 						  case MSG_PRECOMMITTED:
 							MTM_TXTRACE(ts, "MtmTransReceiver got MSG_PRECOMMITTED");
-							if (ts->status != TRANSACTION_STATUS_ABORTED) { 
-								Assert(ts->status == TRANSACTION_STATUS_IN_PROGRESS);
+                            if (ts->status == TRANSACTION_STATUS_COMMITTED) {
+                                elog(WARNING, "Receive PRECOMMITTED response for already committed transaction %d (%s) from node %d",
+                                     ts->xid, ts->gid, node);
+                                continue;
+                            }
+							if (ts->status == TRANSACTION_STATUS_IN_PROGRESS) {
 								if (msg->csn > ts->csn) {
 									ts->csn = msg->csn;
 									MtmSyncClock(ts->csn);
@@ -1096,7 +1101,9 @@ static void MtmReceiver(Datum arg)
 									MtmWakeUpBackend(ts);
 								}
 							} else { 
-								elog(WARNING, "Receive PRECOMMITTED response for aborted transaction"); // How it can happen? SHould we use assert here?
+								Assert(ts->status == TRANSACTION_STATUS_ABORTED);
+								elog(WARNING, "Receive PRECOMMITTED response for aborted transaction %d (%s) from node %d", 
+									 ts->xid, ts->gid, node); // How it can happen? SHould we use assert here?
 								if ((ts->participantsMask & ~Mtm->disabledNodeMask & ~ts->votedMask) == 0) {
 									MtmWakeUpBackend(ts);
 								}
@@ -1134,21 +1141,34 @@ static void MtmReceiver(Datum arg)
 			}
 		}
 		if (Mtm->status == MTM_ONLINE) { 
-			/* "now" is time of performing select, so that delays in processing should not cause false detection */
-			if (now > lastHeartbeatCheck + MSEC_TO_USEC(MtmHeartbeatRecvTimeout)) { 
-				if (!MtmWatchdog(now)) { 
-					for (i = 0; i < nNodes; i++) { 
-						if (Mtm->nodes[i].lastHeartbeat != 0 && sockets[i] >= 0) {
-							MTM_LOG1("Last heartbeat from node %d received %ld microseconds ago", i+1, now - Mtm->nodes[i].lastHeartbeat);
+			/* Check for hearbeat only in case of timeout expiration: it means that we do not have unproceeded events.
+			 * It helps to avoid false node failure detection because of blocking receiver.
+			 */
+			now = MtmGetSystemTime();
+			if (n == 0) {
+				selectTimeout = MtmHeartbeatRecvTimeout; /* restore select timeout */ 
+				if (now > lastHeartbeatCheck + MSEC_TO_USEC(MtmHeartbeatRecvTimeout)) { 
+					if (!MtmWatchdog(now)) { 
+						for (i = 0; i < nNodes; i++) { 
+							if (Mtm->nodes[i].lastHeartbeat != 0 && sockets[i] >= 0) {
+								MTM_LOG1("Last heartbeat from node %d received %ld microseconds ago", i+1, now - Mtm->nodes[i].lastHeartbeat);
+							}
 						}
 					}
+					lastHeartbeatCheck = now;
+				}
+				if (Mtm->disabledNodeMask != 0) { 
+					/* If timeout is expired and there are disabled nodes, then recheck cluster's state */
+					MtmRefreshClusterStatus(false);
+				}
+			} else {
+				if (now > lastHeartbeatCheck + MSEC_TO_USEC(MtmHeartbeatRecvTimeout)) { 
+					/* Switch to non-blocking mode to proceed all pending requests before doing watchdog check */
+					selectTimeout = 0;
 				}
-				lastHeartbeatCheck = now;
-			}
-			if (n == 0 && Mtm->disabledNodeMask != 0) { 
-				/* If timeout is expired and there are disabled nodes, then recheck cluster's state */
-				MtmRefreshClusterStatus(false);
 			}
+		} else if (n == 0) { 
+			selectTimeout = MtmHeartbeatRecvTimeout; /* restore select timeout */ 
 		}
 	}
 	proc_exit(1); /* force restart of this bgwroker */

multimaster--1.0.sql

@@ -45,6 +45,16 @@ 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" integer, "configChanges" integer);
 
+CREATE TYPE mtm.trans_state AS ("status" text, "gid" text, "xid" integer, "coordinator" integer, "gxid" integer, "csn" timestamp, "snapshot" timestamp, "local" boolean, "prepared" boolean, "active" boolean, "twophase" boolean, "votingCompleted" boolean, "participants" bigint, "voted" bigint);
+
+CREATE FUNCTION mtm.get_trans_by_gid(git text) RETURNS mtm.trans_state
+AS 'MODULE_PATHNAME','mtm_get_trans_by_gid'
+LANGUAGE C;
+
+CREATE FUNCTION mtm.get_trans_by_xid(tid xid) RETURNS mtm.trans_state
+AS 'MODULE_PATHNAME','mtm_get_trans_by_xid'
+LANGUAGE C;
+
 CREATE FUNCTION mtm.get_cluster_state() RETURNS mtm.cluster_state 
 AS 'MODULE_PATHNAME','mtm_get_cluster_state'
 LANGUAGE C;

multimaster.c

@@ -115,6 +115,8 @@ PG_FUNCTION_INFO_V1(mtm_poll_node);
 PG_FUNCTION_INFO_V1(mtm_recover_node);
 PG_FUNCTION_INFO_V1(mtm_get_snapshot);
 PG_FUNCTION_INFO_V1(mtm_get_csn);
+PG_FUNCTION_INFO_V1(mtm_get_trans_by_gid);
+PG_FUNCTION_INFO_V1(mtm_get_trans_by_xid);
 PG_FUNCTION_INFO_V1(mtm_get_last_csn);
 PG_FUNCTION_INFO_V1(mtm_get_nodes_state);
 PG_FUNCTION_INFO_V1(mtm_get_cluster_state);
@@ -3649,7 +3651,89 @@ mtm_get_nodes_state(PG_FUNCTION_ARGS)
 	SRF_RETURN_NEXT(funcctx, HeapTupleGetDatum(heap_form_tuple(usrfctx->desc, usrfctx->values, usrfctx->nulls)));
 }
 
+Datum
+mtm_get_trans_by_gid(PG_FUNCTION_ARGS)
+{
+	TupleDesc desc;
+    Datum     values[Natts_mtm_trans_state];
+    bool      nulls[Natts_mtm_trans_state] = {false};
+	MtmTransState* ts;
+	MtmTransMap* tm;
+	char *gid = text_to_cstring(PG_GETARG_TEXT_PP(0));
+	int i;
+
+	MtmLock(LW_SHARED);
+	tm = (MtmTransMap*)hash_search(MtmGid2State, gid, HASH_FIND, NULL);
+	if (tm == NULL) {
+		MtmUnlock();
+		PG_RETURN_NULL();
+	}
+
+	values[1] = CStringGetTextDatum(gid);
+
+	ts = tm->state;
+	if (ts == NULL) { 
+		values[0] = CStringGetTextDatum(MtmTxnStatusMnem[tm->status]);
+		for (i = 2; i < Natts_mtm_trans_state; i++) { 
+			nulls[i] = true;
+		}
+	} else { 
+		values[0] = CStringGetTextDatum(MtmTxnStatusMnem[ts->status]);		
+		values[2] = Int32GetDatum(ts->xid);
+		values[3] = Int32GetDatum(ts->gtid.node);
+		values[4] = Int32GetDatum(ts->gtid.xid);
+		values[5] = TimestampTzGetDatum(time_t_to_timestamptz(ts->csn/USECS_PER_SEC));  
+		values[6] = TimestampTzGetDatum(time_t_to_timestamptz(ts->snapshot/USECS_PER_SEC));  
+		values[7] = BoolGetDatum(ts->isLocal);
+		values[8] = BoolGetDatum(ts->isPrepared);
+		values[9] = BoolGetDatum(ts->isActive);
+		values[10] = BoolGetDatum(ts->isTwoPhase);
+		values[11] = BoolGetDatum(ts->votingCompleted);
+		values[12] = Int64GetDatum(ts->participantsMask);
+		values[13] = Int64GetDatum(ts->votedMask);
+	}
+	MtmUnlock();
+
+	get_call_result_type(fcinfo, NULL, &desc);
+	PG_RETURN_DATUM(HeapTupleGetDatum(heap_form_tuple(desc, values, nulls)));
+}
+	
+Datum
+mtm_get_trans_by_xid(PG_FUNCTION_ARGS)
+{
+	TupleDesc desc;
+    Datum     values[Natts_mtm_trans_state];
+    bool      nulls[Natts_mtm_trans_state] = {false};
+	TransactionId xid = PG_GETARG_INT32(0);
+	MtmTransState* ts;
 
+	MtmLock(LW_SHARED);
+	ts = (MtmTransState*)hash_search(MtmXid2State, &xid, HASH_FIND, NULL);
+	if (ts == NULL) {
+		MtmUnlock();
+		PG_RETURN_NULL();
+	}
+
+	values[0] = CStringGetTextDatum(MtmTxnStatusMnem[ts->status]);		
+	values[1] = CStringGetTextDatum(ts->gid);	
+	values[2] = Int32GetDatum(ts->xid);
+	values[3] = Int32GetDatum(ts->gtid.node);
+	values[4] = Int32GetDatum(ts->gtid.xid);
+	values[5] = TimestampTzGetDatum(time_t_to_timestamptz(ts->csn/USECS_PER_SEC));  
+	values[6] = TimestampTzGetDatum(time_t_to_timestamptz(ts->snapshot/USECS_PER_SEC));  
+	values[7] = BoolGetDatum(ts->isLocal);
+	values[8] = BoolGetDatum(ts->isPrepared);
+	values[9] = BoolGetDatum(ts->isActive);
+	values[10] = BoolGetDatum(ts->isTwoPhase);
+	values[11] = BoolGetDatum(ts->votingCompleted);
+	values[12] = Int64GetDatum(ts->participantsMask);
+	values[13] = Int64GetDatum(ts->votedMask);
+	MtmUnlock();
+
+	get_call_result_type(fcinfo, NULL, &desc);
+	PG_RETURN_DATUM(HeapTupleGetDatum(heap_form_tuple(desc, values, nulls)));
+}
+	
 Datum
 mtm_get_cluster_state(PG_FUNCTION_ARGS)
 {
@@ -4740,7 +4824,6 @@ MtmDetectGlobalDeadLockForXid(TransactionId xid)
 		Assert(replorigin_session_origin == InvalidRepOriginId);
 		XLogFlush(LogLogicalMessage("L", buf.data, buf.used, false));
 
-		MtmSleep(MSEC_TO_USEC(DeadlockTimeout));
 		MtmGraphInit(&graph);
 		MtmGraphAdd(&graph, (GlobalTransactionId*)buf.data, buf.used/sizeof(GlobalTransactionId));
         ByteBufferFree(&buf);

multimaster.h

@@ -82,6 +82,7 @@
 
 #define Natts_mtm_cluster_state 16
 #define Natts_mtm_nodes_state   13
+#define Natts_mtm_trans_state   14
 
 typedef uint64 csn_t; /* commit serial number */
 #define INVALID_CSN  ((csn_t)-1)

tests2/test_recovery_up.py

@@ -62,7 +62,7 @@ class RecoveryTest(unittest.TestCase, TestHelper):
 
     def setUp(self):
         time.sleep(20)
-        print('Start new test')
+        print('Start new test at ',datetime.datetime.now())
         warnings.simplefilter("ignore", ResourceWarning)
         self.client = MtmClient([
             "dbname=regression user=postgres host=127.0.0.1 port=15432",
@@ -72,7 +72,7 @@ def setUp(self):
         self.client.bgrun()
 
     def tearDown(self):
-        print('tearDown')
+        print('Finish test at ',datetime.datetime.now())
         self.client.stop()