MMTS bug fixes

Author: knizhnik

arbiter.c

@@ -1,7 +1,7 @@
 /*
- * multimaster.c
+ * arbiter.c
  *
- * Multimaster based on logical replication
+ * Coordinate global transaction commit
  *
  */
 
@@ -99,15 +99,15 @@ static void DtmTransReceiver(Datum arg);
 static BackgroundWorker DtmSender = {
 	"mm-sender",
 	BGWORKER_SHMEM_ACCESS |  BGWORKER_BACKEND_DATABASE_CONNECTION, /* do not need connection to the database */
-	BgWorkerStart_PostmasterStart,
+	BgWorkerStart_ConsistentState,
 	1, /* restrart in one second (is it possible to restort immediately?) */
 	DtmTransSender
 };
 
 static BackgroundWorker DtmRecevier = {
 	"mm-receiver",
 	BGWORKER_SHMEM_ACCESS |  BGWORKER_BACKEND_DATABASE_CONNECTION, /* do not need connection to the database */
-	BgWorkerStart_PostmasterStart,
+	BgWorkerStart_ConsistentState,
 	1, /* restrart in one second (is it possible to restort immediately?) */
 	DtmTransReceiver
 };
@@ -216,7 +216,7 @@ static int connectSocket(char const* host, int port)
 
 static void openConnections()
 {
-	int nNodes = ds->nNodes;
+	int nNodes = MMNodes;
 	int i;
 	char* connStr = pstrdup(MMConnStrs);
 
@@ -228,9 +228,14 @@ static void openConnections()
 		if (host == NULL) {
 			elog(ERROR, "Invalid connection string: '%s'", MMConnStrs);
 		}
-		for (end = host+5; *end != ' ' && *end != ',' && end != '\0'; end++);
-		*end = '\0';
-		connStr = end + 1;
+		host += 5;
+		for (end = host; *end != ' ' && *end != ',' && *end != '\0'; end++);
+		if (*end != '\0') { 
+			*end = '\0';
+			connStr = end + 1;
+		} else { 
+			connStr = end;
+		}
 		sockets[i] = i+1 != MMNodeId ? connectSocket(host, MMArbiterPort + i) : -1;
 	}
 }
@@ -241,7 +246,7 @@ static void acceptConnections()
 	int i;
 	int sd;
     int on = 1;
-	int nNodes = ds->nNodes-1;
+	int nNodes = MMNodes-1;
 
 	sockets = (int*)palloc(sizeof(int)*nNodes);
 
@@ -359,7 +364,6 @@ static void DtmTransReceiver(Datum arg)
 {
 	int nNodes = MMNodes-1;
 	int i, j, rc;
-	int rxBufPos = 0;
 	DtmBuffer* rxBuffer = (DtmBuffer*)palloc(sizeof(DtmBuffer)*nNodes);
 	HTAB* xid2state;
 
@@ -408,7 +412,7 @@ static void DtmTransReceiver(Datum arg)
 #endif
 			{
 				int nResponses;
-				rxBuffer[i].used += readSocket(sockets[i], (char*)rxBuffer[i].data + rxBuffer[i].used, BUFFER_SIZE-rxBufPos);
+				rxBuffer[i].used += readSocket(sockets[i], (char*)rxBuffer[i].data + rxBuffer[i].used, BUFFER_SIZE-rxBuffer[i].used);
 				nResponses = rxBuffer[i].used/sizeof(DtmCommitMessage);
 
 				LWLockAcquire(ds->hashLock, LW_SHARED);						
@@ -426,6 +430,8 @@ static void DtmTransReceiver(Datum arg)
 						SetLatch(&ProcGlobal->allProcs[ts->procno].procLatch);
 					}
 				}
+				LWLockRelease(ds->hashLock);
+				
 				rxBuffer[i].used -= nResponses*sizeof(DtmCommitMessage);
 				if (rxBuffer[i].used != 0) { 
 					memmove(rxBuffer[i].data, (char*)rxBuffer[i].data + nResponses*sizeof(DtmCommitMessage), rxBuffer[i].used);

multimaster.c

@@ -453,7 +453,7 @@ static void DtmPrepareTransaction(DtmCurrentTrans* x)
 	ts->status = TRANSACTION_STATUS_UNKNOWN;
 	ts->csn = dtm_get_csn();	
 	ts->procno = MyProc->pgprocno;
-	ts->nVotes = 1; /* I voted myself */
+	ts->nVotes = 1; /* My own voice */
 	for (i = 0; i < MMNodes; i++) { 
 		ts->xids[i] = InvalidTransactionId;
 	}
@@ -479,7 +479,8 @@ static XidStatus
 DtmCommitTransaction(TransactionId xid, int nsubxids, TransactionId *subxids)
 {
 	DtmTransState* ts;
-	csn_t csn;
+	csn_t localCSN;
+	csn_t globalCSN;
 	int i;
 	XidStatus status;
 
@@ -489,44 +490,45 @@ DtmCommitTransaction(TransactionId xid, int nsubxids, TransactionId *subxids)
 
 	/* now transaction is in doubt state */
 	ts->status = TRANSACTION_STATUS_IN_PROGRESS;
-	csn = dtm_get_csn();	
-	if (csn > ts->csn) {
-		ts->csn = csn;
-	}
+	localCSN = dtm_get_csn();	
+	ts->csn = localCSN;
 	DtmTransactionListAppend(ts);
 	DtmAddSubtransactions(ts, subxids, nsubxids);
 
 	MMVoteForTransaction(ts); /* wait until transaction at all nodes are prepared */
-	csn = ts->csn;
-	if (csn != INVALID_CSN) { 
-		dtm_sync(csn);
+	globalCSN = ts->csn;
+	Assert(globalCSN >= localCSN);
+
+	if (globalCSN != INVALID_CSN) { 
+		dtm_sync(globalCSN);
 		status = TRANSACTION_STATUS_COMMITTED;
 	} else { 
+		ts->csn = globalCSN = localCSN;
 		status = TRANSACTION_STATUS_ABORTED;
 	}
 	ts->status = status;
 	for (i = 0; i < nsubxids; i++) { 	
 		ts = ts->next;
 		ts->status = status;
-		ts->csn = csn;
+		ts->csn = globalCSN;
 	}        
 	LWLockRelease(dtm->hashLock);
 	return status;
 }
 
 static void 
-DtmAbortTransaction(TransactionId xid, int nsubxids, TransactionId *subxids)
+DtmFinishTransaction(TransactionId xid, int nsubxids, TransactionId *subxids, XidStatus status)
 {
 	int i;
 	DtmTransState* ts;
 
 	LWLockAcquire(dtm->hashLock, LW_EXCLUSIVE);
 	ts = hash_search(xid2state, &xid, HASH_FIND, NULL);
 	Assert(ts != NULL); /* should be created by DtmPrepareTransaction */
-	ts->status = TRANSACTION_STATUS_ABORTED;	
+	ts->status = status;
 	for (i = 0; i < nsubxids; i++) { 	
 		ts = ts->next;
-		ts->status = TRANSACTION_STATUS_ABORTED;
+		ts->status = status;
 	}        
 	LWLockRelease(dtm->hashLock);
 }	
@@ -539,9 +541,10 @@ DtmSetTransactionStatus(TransactionId xid, int nsubxids, TransactionId *subxids,
 	DTM_INFO("%d: DtmSetTransactionStatus %u = %u\n", getpid(), xid, status);
 	if (dtmTx.isDistributed)
 	{
+		Assert(xid == dtmTx.xid);
 		if (status == TRANSACTION_STATUS_ABORTED || !dtmTx.containsDML)
 		{
-			DtmAbortTransaction(xid, nsubxids, subxids);	
+			DtmFinishTransaction(xid, nsubxids, subxids, status);	
 			DTM_INFO("Abort transaction %d\n", xid);
 		}
 		else
@@ -990,12 +993,18 @@ MMPoolConstructor(void)
 static void 
 SendCommitMessage(DtmTransState* ts)
 {
+	DtmTransState* votingList;
+
 	SpinLockAcquire(&dtm->votingSpinlock);
-	ts->nextVoting = dtm->votingTransactions;
+	votingList = dtm->votingTransactions;
+	ts->nextVoting = votingList;
 	dtm->votingTransactions = ts;
 	SpinLockRelease(&dtm->votingSpinlock);
 
-	PGSemaphoreUnlock(&dtm->votingSemaphore);
+	if (votingList == NULL) { 
+		/* singal semaphreo only once for the whole list */
+		PGSemaphoreUnlock(&dtm->votingSemaphore);
+	}
 }
 
 static void 
@@ -1011,7 +1020,7 @@ MMVoteForTransaction(DtmTransState* ts)
 		/* ... and then send notifications to replicas */
 		SendCommitMessage(ts);
 	} else {
-		/* I am replica: first notify master... */
+		/* I am replica: first notify coordinator... */
 		ts->nVotes = dtm->nNodes-1; /* I just need one confirmation from coordinator */
 		SendCommitMessage(ts);
 		/* ... and wait response from it */