Start rewriting multimaster to 2PC

Author: knizhnik

arbiter.c

@@ -458,7 +458,7 @@ static void MtmAppendBuffer(MtmBuffer* txBuffer, TransactionId xid, int node, Mt
 	MTM_TRACE("Send message %s CSN=%ld to node %d from node %d for global transaction %d/local transaction %d\n", 
 			  messageText[ts->cmd], ts->csn, node+1, MtmNodeId, ts->gtid.xid, ts->xid);
 	Assert(ts->cmd != MSG_INVALID);
-	buf->data[buf->used].code = ts->cmd;
+	buf->data[buf->used].code = ts->status == TRANSACTION_STATUS_ABORTED ? MSG_ABORTED : MSG_PREPARED;
 	buf->data[buf->used].dxid = xid;
 	buf->data[buf->used].sxid = ts->xid;
 	buf->data[buf->used].csn =  ts->csn;
@@ -467,21 +467,6 @@ static void MtmAppendBuffer(MtmBuffer* txBuffer, TransactionId xid, int node, Mt
 	buf->used += 1;
 }
 
-static void MtmBroadcastMessage(MtmBuffer* txBuffer, MtmTransState* ts)
-{
-	int i;
-	int n = 1;
-	for (i = 0; i < MtmNodes; i++)
-	{
-		if (TransactionIdIsValid(ts->xids[i])) { 
-			Assert(i+1 != MtmNodeId);
-			MtmAppendBuffer(txBuffer, ts->xids[i], i, ts);
-			n += 1;
-		}
-	}
-	Assert(n == ds->nNodes);
-}
-
 static void MtmTransSender(Datum arg)
 {
 	int nNodes = MtmNodes;
@@ -508,12 +493,7 @@ static void MtmTransSender(Datum arg)
 		MtmLock(LW_SHARED); 
 
 		for (ts = ds->votingTransactions; ts != NULL; ts = ts->nextVoting) {
-			if (MtmIsCoordinator(ts)) { 
-				MtmBroadcastMessage(txBuffer, ts);
-			} else { 
-				MtmAppendBuffer(txBuffer, ts->gtid.xid, ts->gtid.node-1, ts);
-			}
-			ts->cmd = MSG_INVALID;
+			MtmAppendBuffer(txBuffer, ts->gtid.xid, ts->gtid.node-1, ts);
 		}
 		ds->votingTransactions = NULL;
 		MtmUnlock();
@@ -634,109 +614,31 @@ static void MtmTransReceiver(Datum arg)
 					MtmArbiterMessage* msg = &rxBuffer[i].data[j];
 					MtmTransState* ts = (MtmTransState*)hash_search(xid2state, &msg->dxid, HASH_FIND, NULL);
 					Assert(ts != NULL);
-					Assert(ts->cmd == MSG_INVALID);
 					Assert(msg->node > 0 && msg->node <= nNodes && msg->node != MtmNodeId);
-					ts->xids[msg->node-1] = msg->sxid;
-
-					if (MtmIsCoordinator(ts)) { 
-						switch (msg->code) { 
-						case MSG_READY:
-							Assert(ts->status == TRANSACTION_STATUS_ABORTED || ts->status == TRANSACTION_STATUS_IN_PROGRESS);
-							Assert(ts->nVotes < ds->nNodes);	
-							ds->nodeTransDelay[msg->node-1] += MtmGetCurrentTime() - ts->csn;
-							if (++ts->nVotes == ds->nNodes) { 
-								/* All nodes are finished their transactions */
-								if (ts->status == TRANSACTION_STATUS_IN_PROGRESS) {
-									ts->nVotes = 1; /* I voted myself */
-									ts->cmd = MSG_PREPARE;
-								} else { 
-									ts->status = TRANSACTION_STATUS_ABORTED;
-									ts->cmd = MSG_ABORT;
-									MtmAdjustSubtransactions(ts);
-									MtmWakeUpBackend(ts);								
-								}
-								MtmSendNotificationMessage(ts);									  
-							}
-							break;
+					Assert (MtmIsCoordinator(ts));
+					switch (msg->code) { 
 						case MSG_PREPARED:
- 						    Assert(ts->status == TRANSACTION_STATUS_IN_PROGRESS);
-							Assert(ts->nVotes < ds->nNodes);
-							if (msg->csn > ts->csn) {
-								ts->csn = msg->csn;
-								MtmSyncClock(ts->csn);
-							}
-							if (++ts->nVotes == ds->nNodes) { 
-								/* ts->csn is maximum of CSNs at all nodes */
-								ts->nVotes = 1; /* I voted myself */
-								ts->cmd = MSG_COMMIT;
-								ts->csn = MtmAssignCSN();
-								ts->status = TRANSACTION_STATUS_UNKNOWN;
-								MtmAdjustSubtransactions(ts);
-								MtmSendNotificationMessage(ts);
-							}
-							break;
-						case MSG_COMMITTED:
-							Assert(ts->status == TRANSACTION_STATUS_UNKNOWN);
-							Assert(ts->nVotes < ds->nNodes);
-							if (++ts->nVotes == ds->nNodes) { 									
-								/* All nodes have the same CSN */
-								MtmWakeUpBackend(ts);
+							if (ts->status == TRANSACTION_STATUS_IN_PROGRESS) { 
+								if (msg->csn > ts->csn) {
+									ts->csn = msg->csn;
+									MtmSyncClock(ts->csn);
+								}
+								if (++ts->nVotes == ds->nNodes) { 
+									MtmWakeUpBackend(ts);
+								}
 							}
 							break;
 						case MSG_ABORTED:
-							Assert(ts->status == TRANSACTION_STATUS_ABORTED || ts->status == TRANSACTION_STATUS_IN_PROGRESS);
-							Assert(ts->nVotes < ds->nNodes);
-							ts->status = TRANSACTION_STATUS_ABORTED;									
-							if (++ts->nVotes == ds->nNodes) { 
-								ts->cmd = MSG_ABORT;
-								MtmAdjustSubtransactions(ts);
-								MtmSendNotificationMessage(ts);		
-								MtmWakeUpBackend(ts);								
-							}
-							break;
-						default:
-							Assert(false);
-						}
-					} else { /* replica */
-						switch (msg->code) { 
-						case MSG_PREPARE:
- 					        Assert(ts->status == TRANSACTION_STATUS_IN_PROGRESS); 
-							if ((msg->disabledNodeMask & ~ds->disabledNodeMask) != 0) { 
-								/* Coordinator's disabled mask is wider than my: so reject such transaction to avoid 
-								   commit on smaller subset of nodes */
+							if (ts->status == TRANSACTION_STATUS_IN_PROGRESS) { 
 								ts->status = TRANSACTION_STATUS_ABORTED;
-								ts->cmd = MSG_ABORT;
-								MtmAdjustSubtransactions(ts);
-								MtmWakeUpBackend(ts);
-							} else { 
-								ts->status = TRANSACTION_STATUS_UNKNOWN;
-								ts->csn = MtmAssignCSN();
-								ts->cmd = MSG_PREPARED;
-							}
-							MtmSendNotificationMessage(ts);
-							break;
-						case MSG_COMMIT:
-							Assert(ts->status == TRANSACTION_STATUS_UNKNOWN);
-							Assert(ts->csn < msg->csn);
-							ts->csn = msg->csn;
-							MtmSyncClock(ts->csn);
-							ts->cmd = MSG_COMMITTED;							
-							MtmAdjustSubtransactions(ts);
-							MtmSendNotificationMessage(ts);
-							MtmWakeUpBackend(ts);
-							break;
-						case MSG_ABORT:
-							if (ts->status != TRANSACTION_STATUS_ABORTED) {
-								Assert(ts->status == TRANSACTION_STATUS_UNKNOWN || ts->status == TRANSACTION_STATUS_IN_PROGRESS);
-								ts->status = TRANSACTION_STATUS_ABORTED;								
 								MtmAdjustSubtransactions(ts);
 								MtmWakeUpBackend(ts);
 							}
 							break;
 						default:
 							Assert(false);
 						}
-					}
+					} 
 				}
 				MtmUnlock();
 

multimaster.c

@@ -64,7 +64,6 @@ typedef struct {
 	bool  isReplicated;   /* transaction on replica */
 	bool  isDistributed;  /* transaction performed INSERT/UPDATE/DELETE and has to be replicated to other nodes */
 	bool  containsDML;    /* transaction contains DML statements */
-	bool  isPrepared;     /* transaction is prepared as part of 2PC */
     csn_t snapshot;       /* transaction snaphsot */
 } MtmCurrentTrans;
 
@@ -100,8 +99,8 @@ static void MtmSetTransactionStatus(TransactionId xid, int nsubxids, Transaction
 static void MtmInitialize(void);
 static void MtmXactCallback(XactEvent event, void *arg);
 static void MtmBeginTransaction(MtmCurrentTrans* x);
-static void MtmPrecommitTransaction(MtmCurrentTrans* x);
 static bool MtmCommitTransaction(TransactionId xid, int nsubxids, TransactionId *subxids);
+static void MtmPrePrepareTransaction(MtmCurrentTrans* x);
 static void MtmPrepareTransaction(MtmCurrentTrans* x);
 static void MtmCommitPreparedTransaction(MtmCurrentTrans* x);
 static void MtmEndTransaction(MtmCurrentTrans* x, bool commit);
@@ -347,15 +346,6 @@ bool MtmXidInMVCCSnapshot(TransactionId xid, Snapshot snapshot)
 	return PgXidInMVCCSnapshot(xid, snapshot);
 }    
 
-static uint32 MtmXidHashFunc(const void *key, Size keysize)
-{
-	return (uint32)*(TransactionId*)key;
-}
-
-static int MtmXidMatchFunc(const void *key1, const void *key2, Size keysize)
-{
-	return *(TransactionId*)key1 - *(TransactionId*)key2;
-}
 
 static void MtmTransactionListAppend(MtmTransState* ts)
 {
@@ -489,8 +479,8 @@ MtmXactCallback(XactEvent event, void *arg)
 	  case XACT_EVENT_START: 
 	    MtmBeginTransaction(&dtmTx);
         break;
-	  case XACT_EVENT_PRE_COMMIT:
-		MtmPrecommitTransaction(&dtmTx);
+	  case XACT_EVENT_PRE_PREPARE:
+		MtmPrePrepareTransaction(&dtmTx);
 		break;
 	  case XACT_EVENT_PREPARE:
 		MtmPrepareTransaction(&dtmTx);
@@ -598,16 +588,14 @@ MtmCheckClusterLock()
 }	
 
 /* 
- * This functions is called as pre-commit callback.
- * We need to pass snapshot to WAL-sender, so create record in transaction status hash table 
- * before commit
+ * Prepare transaction for two-phase commit
  */
-static void MtmPrecommitTransaction(MtmCurrentTrans* x)
+MtmPrePrepareTransaction(MtmCurrentTrans* x)
 { 
 	MtmTransState* ts;
 	int i;
 
-	if (!x->isDistributed || x->isPrepared) {
+	if (!x->isDistributed) {
 		return;
 	}
 
@@ -632,7 +620,6 @@ static void MtmPrecommitTransaction(MtmCurrentTrans* x)
 	ts->snapshot = x->isReplicated || !x->containsDML ? INVALID_CSN : x->snapshot;
 	ts->csn = MtmAssignCSN();	
 	ts->gtid = x->gtid;
-	ts->cmd = MSG_INVALID;
 	ts->procno = MyProc->pgprocno;
 	ts->nVotes = 0; 
 	ts->voteCompleted = false;
@@ -644,24 +631,41 @@ static void MtmPrecommitTransaction(MtmCurrentTrans* x)
 		ts->gtid.xid = x->xid;
 		ts->gtid.node = MtmNodeId;
 	}
-	for (i = 0; i < MtmNodes; i++) { 
-		ts->xids[i] = InvalidTransactionId;
-	}
 	MtmTransactionListAppend(ts);
 
 	MtmUnlock();
 
 	MTM_TRACE("%d: MtmPrepareTransaction prepare commit of %d CSN=%ld\n", MyProcPid, x->xid, ts->csn);
 }
 
-static void 
 MtmPrepareTransaction(MtmCurrentTrans* x)
-{	
-	MtmPrecommitTransaction(x);
-	MTM_TRACE("Prepare transaction %d", x->xid);
-	x->isPrepared = true;	
+{ 
+	MtmLock(LW_EXCLUSIVE);
+	if (ts->status = TRANSACTION_STATUS_IN_PROGRESS) { 
+		ts->status = TRANSACTION_STATUS_UNKNOWN;	
+		MtmAdjustSubtransactions(ts);
+	}
+
+	if (!MtmIsCoordinator(ts)) {
+		MtmSendNotificationMessage(ts); /* send notification to coordinator */
+		MtmUnlock();
+	} else { 
+		/* wait N commits or just one ABORT */
+		ts->nVotes += 1;
+		while (ts->nVotes != dtm->nNodes && ts->status == TRANSACTION_STATUS_PROGRESS) { 
+			MtmUnlock();
+			WaitLatch(&MyProc->procLatch, WL_LATCH_SET, -1);
+			ResetLatch(&MyProc->procLatch);			
+			MtmLock(LW_SHARED);
+		}
+		MtmUnlock();
+		if (ts->status == TRANSACTION_STATUS_ABORTED) { 
+			elog(ERROR, "Distributed transaction %d is rejected by DTM", x->xid);
+		}
+	}
 }
 
+
 static void 
 MtmCommitPreparedTransaction(MtmCurrentTrans* x)
 {
@@ -1679,14 +1683,12 @@ HTAB* MtmCreateHash(void)
 	Assert(MtmNodes > 0);
 	memset(&info, 0, sizeof(info));
 	info.keysize = sizeof(TransactionId);
-	info.entrysize = sizeof(MtmTransState) + (MtmNodes-1)*sizeof(TransactionId);
-	info.hash = MtmXidHashFunc;
-	info.match = MtmXidMatchFunc;
+	info.entrysize = sizeof(MtmTransState);
 	htab = ShmemInitHash(
 		"xid2state",
 		MTM_HASH_SIZE, MTM_HASH_SIZE,
 		&info,
-		HASH_ELEM | HASH_FUNCTION | HASH_COMPARE
+		HASH_ELEM
 	);
 	return htab;
 }

multimaster.h

@@ -19,6 +19,7 @@
 #define MULTIMASTER_SLOT_PATTERN        "mtm_slot_%d"
 #define MULTIMASTER_MIN_PROTO_VERSION   1
 #define MULTIMASTER_MAX_PROTO_VERSION   1
+#define MULTIMASTER_MAX_GID             32
 
 #define USEC 1000000
 
@@ -51,12 +52,7 @@ typedef enum
 { 
 	MSG_INVALID,
 	MSG_HANDSHAKE,
-	MSG_READY,
-	MSG_PREPARE,
-	MSG_COMMIT,
-	MSG_ABORT,
 	MSG_PREPARED,
-	MSG_COMMITTED,
 	MSG_ABORTED,
 	MSG_STATUS
 } MtmMessageCode;
@@ -88,11 +84,10 @@ typedef struct MtmTransState
 							              finally should be nNodes-1 */
 	int            procno;             /* pgprocno of transaction coordinator waiting for responses from replicas, 
 							              used to notify coordinator by arbiter */
-	MtmMessageCode cmd;                /* Notification message code to be sent */
+	bool           voteCompleted;      /* Responses necessary to make a decision are received by coordinator of transaction */
 	int            nSubxids;           /* Number of subtransanctions */
 	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 */
-	bool voteCompleted;                /* Responses necessary to make a decision are received by coordinator of transaction */
 	TransactionId xids[1];             /* transaction ID at replicas: varying size MtmNodes */
 } MtmTransState;