Add wait 2pc timeout

Author: knizhnik

contrib/mmts/arbiter.c

@@ -720,14 +720,24 @@ static void MtmTransReceiver(Datum arg)
 								MtmAbortTransaction(ts);
 							}
 
+							if (!MtmUseDtm && msg->csn > ts->csn) {
+								ts->csn = msg->csn;
+								MtmSyncClock(ts->csn);
+							}
+
 							if (++ts->nVotes == Mtm->nNodes) { 
 								/* All nodes are finished their transactions */
-								if (ts->status == TRANSACTION_STATUS_IN_PROGRESS) {
+								if (ts->status == TRANSACTION_STATUS_ABORTED) { 
+									MtmWakeUpBackend(ts);								
+								} else if (MtmUseDtm) { 
+									Assert(ts->status == TRANSACTION_STATUS_IN_PROGRESS);
 									ts->nVotes = 1; /* I voted myself */
 									MtmSendNotificationMessage(ts, MSG_PREPARE);									  
 								} else { 
-									Assert(ts->status == TRANSACTION_STATUS_ABORTED);
-									MtmWakeUpBackend(ts);								
+									Assert(ts->status == TRANSACTION_STATUS_IN_PROGRESS);
+									ts->csn = MtmAssignCSN();
+									ts->status = TRANSACTION_STATUS_UNKNOWN;
+									MtmWakeUpBackend(ts);
 								}
 							}
 							break;						   

contrib/mmts/multimaster.c

@@ -191,6 +191,8 @@ static int   MtmWorkers;
 static int   MtmVacuumDelay;
 static int   MtmMinRecoveryLag;
 static int   MtmMaxRecoveryLag;
+static int   Mtm2PCPrepareRatio;
+static int   Mtm2PCMinTimeout;
 static bool  MtmIgnoreTablesWithoutPk;
 
 static ExecutorFinish_hook_type PreviousExecutorFinishHook;
@@ -766,8 +768,6 @@ MtmPrePrepareTransaction(MtmCurrentTrans* x)
 
 }
 
-static time_t maxWakeupTime;
-
 static void
 MtmPostPrepareTransaction(MtmCurrentTrans* x)
 { 
@@ -783,25 +783,32 @@ MtmPostPrepareTransaction(MtmCurrentTrans* x)
 		tm->state = ts;	
 		ts->votingCompleted = true;
 		if (Mtm->status != MTM_RECOVERY) { 
-			MtmSendNotificationMessage(ts, MtmUseDtm ? MSG_READY : MSG_PREPARED); /* send notification to coordinator */
+			if (MtmUseDtm) { 
+				MtmSendNotificationMessage(ts, MSG_READY); /* send notification to coordinator */
+			} else { 
+				ts->csn = MtmAssignCSN();
+				MtmSendNotificationMessage(ts, MSG_PREPARED); /* send notification to coordinator */
+				ts->status = TRANSACTION_STATUS_UNKNOWN;
+			}
 		} else {
 			ts->status = TRANSACTION_STATUS_UNKNOWN;
 		}
 		MtmUnlock();
 		MtmResetTransaction(x);
 	} else { 
-		time_t wakeupTime;
+		time_t timeout = Max(Mtm2PCMinTimeout, (ts->csn - ts->snapshot)*Mtm2PCPrepareRatio/100000); /* usec->msec and percents */ 
+		int result = 0;
 		/* wait votes from all nodes */
-		while (!ts->votingCompleted) { 
+		while (!ts->votingCompleted && !(result & WL_TIMEOUT)) {
 			MtmUnlock();
-			WaitLatch(&MyProc->procLatch, WL_LATCH_SET, -1);
+			result = WaitLatch(&MyProc->procLatch, WL_LATCH_SET|WL_TIMEOUT, timeout);
 			ResetLatch(&MyProc->procLatch);			
-			wakeupTime = MtmGetCurrentTime() - ts->wakeupTime;
-			if (wakeupTime > maxWakeupTime) { 
-				maxWakeupTime = wakeupTime;
-			}
 			MtmLock(LW_SHARED);
 		}
+		if (!ts->votingCompleted) {
+			ts->status = TRANSACTION_STATUS_ABORTED;
+			elog(WARNING, "Transaction is aborted because of %d msec timeout expiration", (int)timeout);
+		}
 		x->status = ts->status;
 		MTM_LOG3("%d: Result of vote: %d", MyProcPid, ts->status);
 		MtmUnlock();
@@ -989,11 +996,12 @@ csn_t MtmGetTransactionCSN(TransactionId xid)
 }
 
 void MtmWakeUpBackend(MtmTransState* ts)
-{
-	MTM_LOG3("Wakeup backed procno=%d, pid=%d", ts->procno, ProcGlobal->allProcs[ts->procno].pid);
-	ts->votingCompleted = true;
-	ts->wakeupTime = MtmGetCurrentTime();
-	SetLatch(&ProcGlobal->allProcs[ts->procno].procLatch); 
+{																		
+	if (!ts->votingCompleted) { 
+		MTM_LOG3("Wakeup backed procno=%d, pid=%d", ts->procno, ProcGlobal->allProcs[ts->procno].pid);
+		ts->votingCompleted = true;
+		SetLatch(&ProcGlobal->allProcs[ts->procno].procLatch); 
+	}
 }
 
 void MtmAbortTransaction(MtmTransState* ts)
@@ -1599,6 +1607,38 @@ _PG_init(void)
 	if (!process_shared_preload_libraries_in_progress)
 		return;
 
+	DefineCustomIntVariable(
+		"multimaster.2pc_min_timeout",
+		"Minamal amount of time (milliseconds) to wait 2PC confirmation from all nodes",
+		"Timeout for 2PC is calculated as MAX(prepare_time*2pc_prepare_ratio/100,2pc_min_timeout)",
+		&Mtm2PCMinTimeout,
+		10000,
+		0,
+		INT_MAX,
+		PGC_BACKEND,
+		0,
+		NULL,
+		NULL,
+		NULL
+	);
+
+	DefineCustomIntVariable(
+		"multimaster.2pc_prepare_ratio",
+		"Percent of prepare time for maximal time of second phase of two-pahse commit",
+		"Timeout for 2PC is calculated as MAX(prepare_time*2pc_prepare_ratio/100,2pc_min_timeout)",
+		&Mtm2PCPrepareRatio,
+		100,
+		0,
+		INT_MAX,
+		PGC_BACKEND,
+		0,
+		NULL,
+		NULL,
+		NULL
+	);
+
+
+
 	DefineCustomIntVariable(
 		"multimaster.node_disable_delay",
 		"Minamal amount of time (sec) between node status change",

contrib/mmts/multimaster.h

@@ -135,7 +135,6 @@ typedef struct MtmTransState
 	int            procno;             /* pgprocno of transaction coordinator waiting for responses from replicas, 
 							              used to notify coordinator by arbiter */
 	int            nSubxids;           /* Number of subtransanctions */
-	time_t         wakeupTime;         
 	MtmMessageCode cmd;                /* Notification message to be sent */
   	struct MtmTransState* nextVoting;  /* Next element in L1-list of voting transactions. */
     struct MtmTransState* next;        /* Next element in L1 list of all finished transaction present in xid2state hash */

contrib/mmts/pglogical_receiver.c

@@ -53,15 +53,13 @@ static volatile sig_atomic_t got_sighup = false;
 
 /* GUC variables */
 static int receiver_idle_time = 0;
-static bool receiver_sync_mode = true;
+static bool receiver_sync_mode = false;
 
 /* Worker name */
-char worker_proc[BGW_MAXLEN];
+static char worker_proc[BGW_MAXLEN];
 
 /* Lastly written positions */
 static XLogRecPtr output_written_lsn = InvalidXLogRecPtr;
-static XLogRecPtr output_fsync_lsn = InvalidXLogRecPtr;
-static XLogRecPtr output_applied_lsn = InvalidXLogRecPtr;
 
 /* Stream functions */
 static void fe_sendint64(int64 i, char *buf);
@@ -91,16 +89,17 @@ receiver_raw_sighup(SIGNAL_ARGS)
  * Send a Standby Status Update message to server.
  */
 static bool
-sendFeedback(PGconn *conn, int64 now)
+sendFeedback(PGconn *conn, int64 now, RepOriginId originId)
 {
 	char		replybuf[1 + 8 + 8 + 8 + 8 + 1];
 	int		 len = 0;
+	XLogRecPtr output_applied_lsn = replorigin_get_progress(originId, true);
 
 	replybuf[len] = 'r';
 	len += 1;
 	fe_sendint64(output_written_lsn, &replybuf[len]);   /* write */
 	len += 8;
-	fe_sendint64(output_fsync_lsn, &replybuf[len]);	 /* flush */
+	fe_sendint64(output_applied_lsn, &replybuf[len]);	/* flush */
 	len += 8;
 	fe_sendint64(output_applied_lsn, &replybuf[len]);	/* apply */
 	len += 8;
@@ -409,8 +408,6 @@ pglogical_receiver_main(Datum main_arg)
 
 				/* Update written position */
 				output_written_lsn = Max(walEnd, output_written_lsn);
-				output_fsync_lsn = output_written_lsn;
-				output_applied_lsn = output_written_lsn;
 
 				/*
 				 * If the server requested an immediate reply, send one.
@@ -424,7 +421,7 @@ pglogical_receiver_main(Datum main_arg)
 					int64 now = feGetCurrentTimestamp();
 
 					/* Leave is feedback is not sent properly */
-					if (!sendFeedback(conn, now))
+					if (!sendFeedback(conn, now, originId))
 						proc_exit(1);
 				}
 				continue;
@@ -482,8 +479,6 @@ pglogical_receiver_main(Datum main_arg)
             }
 			/* Update written position */
 			output_written_lsn = Max(walEnd, output_written_lsn);
-			output_fsync_lsn = output_written_lsn;
-			output_applied_lsn = output_written_lsn;
 		}
 
 		/* No data, move to next loop */

contrib/mmts/tests/dtmbench.cpp

@@ -129,7 +129,7 @@ void* reader(void* arg)
         result r = txn.exec("select sum(v) from t");
         int64_t sum = r[0][0].as(int64_t());
         if (sum != prevSum) {
-			r = txn.exec("select mtm_get_snapshot()");			
+			r = txn.exec("select mtm.get_snapshot()");			
             printf("Total=%ld, snapshot=%ld\n", sum, r[0][0].as(int64_t()));
             prevSum = sum;
         }