Update XTM documentation

Author: knizhnik

arbiter.c

@@ -83,6 +83,7 @@ typedef struct
 	TransactionId  dxid; /* Transaction ID at destination node */
 	TransactionId  sxid; /* Transaction IO at sender node */  
 	csn_t          csn;  /* local CSN in case of sending data from replica to master, global CSN master->replica */
+	int64          disabledNodeMask; /* bitmask of disabled nodes at the sender of message */
 } MtmArbiterMessage;
 
 typedef struct 
@@ -109,7 +110,8 @@ static char const* const messageText[] =
 	"ABORT",
 	"PREPARED",
 	"COMMITTED",
-	"ABORTED"
+	"ABORTED",
+	"STATUS"
 };
 
 
@@ -276,11 +278,28 @@ static int MtmConnectSocket(char const* host, int port, int max_attempts)
 			msg.dxid = HANDSHAKE_MAGIC;
 			msg.sxid = ShmemVariableCache->nextXid;
 			msg.csn  = MtmGetCurrentTime();
+			msg.disabledNodeMask = ds->disabledNodeMask;
 			if (!MtmWriteSocket(sd, &msg, sizeof msg)) { 
 				elog(WARNING, "Arbiter failed to send handshake message to %s:%d: %d", host, port, errno);
 				close(sd);
 				goto Retry;
 			}
+			if (MtmReadSocket(sd, &msg, sizeof msg) != sizeof(msg)) { 
+				elog(WARNING, "Arbiter failed to receive response for handshake message from %s:%d: %d", host, port, errno);
+				close(sd);
+				goto Retry;
+			}
+			if (msg.code != MSG_STATUS || msg.dxid != HANDSHAKE_MAGIC) {
+				elog(WARNING, "Arbiter get unexpected response %d for handshake message from %s:%d: %d", msg.code, host, port, errno);
+				close(sd);
+				goto Retry;
+			}
+				
+			if (BIT_CHECK(msg.disabledNodeMask, MtmNodeId-1)) { 
+				elog(WARNING, "Node is switched to recovery mode");
+				ds->status = MTM_RECOVERY;
+			}
+			ds->disabledNodeMask = msg.disabledNodeMask;
 			return sd;
 		}
     }
@@ -315,11 +334,16 @@ static void MtmOpenConnections()
 			sockets[i] = MtmConnectSocket(host, MtmArbiterPort + i + 1, MtmConnectAttempts);
 			if (sockets[i] < 0) { 
 				MtmDropNode(i+1, false);
-			}
+			} 
 		} else {
 			sockets[i] = -1;
 		}
 	}
+	if (ds->nNodes < MtmNodes/2+1) { /* no quorum */
+		ds->status = MTM_OFFLINE;
+	} else if (ds->status == MTM_INITIALIZATION) { 
+		ds->status = MTM_CONNECTED;
+	}
 }
 
 
@@ -362,8 +386,13 @@ static void MtmAcceptOneConnection()
 			close(fd);
 		} else{ 			
 			Assert(msg.node > 0 && msg.node <= MtmNodes && msg.node != MtmNodeId);
-			if (BIT_SET(ds->disabledNodeMask, msg.node-1)) { 
-				elog(WARNING, "Reject attempt to reconnect from disabled node %d", msg.node); 
+			msg.code = MSG_STATUS;
+			msg.disabledNodeMask = ds->disabledNodeMask;
+			msg.dxid = HANDSHAKE_MAGIC;
+			msg.sxid = ShmemVariableCache->nextXid;
+			msg.csn  = MtmGetCurrentTime();
+			if (!MtmWriteSocket(fd, &msg, sizeof msg)) { 
+				elog(WARNING, "Arbiter failed to write response for handshake message from node %d", msg.node);
 				close(fd);
 			} else { 
 				elog(NOTICE, "Arbiter established connection with node %d", msg.node); 
@@ -427,6 +456,7 @@ static void MtmAppendBuffer(MtmBuffer* txBuffer, TransactionId xid, int node, Mt
 	buf->data[buf->used].sxid = ts->xid;
 	buf->data[buf->used].csn =  ts->csn;
 	buf->data[buf->used].node = MtmNodeId;
+	buf->data[buf->used].disabledNodeMask = ds->disabledNodeMask;
 	buf->used += 1;
 }
 
@@ -659,12 +689,20 @@ static void MtmTransReceiver(Datum arg)
 						switch (msg->code) { 
 						case MSG_PREPARE:
  					        Assert(ts->status == TRANSACTION_STATUS_IN_PROGRESS); 
-							ts->status = TRANSACTION_STATUS_UNKNOWN;
-							ts->csn = MtmAssignCSN();
-							ts->cmd = MSG_PREPARED;
+							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 */
+								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;
-							break;
 						case MSG_COMMIT:
 							Assert(ts->status == TRANSACTION_STATUS_UNKNOWN);
 							Assert(ts->csn < msg->csn);

multimaster.c

@@ -70,6 +70,7 @@ typedef struct {
 #define USEC 1000000
 #define MIN_WAIT_TIMEOUT 1000
 #define MAX_WAIT_TIMEOUT 100000
+#define STATUS_POLL_DELAY USEC
 
 void _PG_init(void);
 void _PG_fini(void);
@@ -147,7 +148,7 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 void MtmLock(LWLockMode mode)
 {
 #ifdef USE_SPINLOCK
-	SpinLockAcquire(&dtm->hashSpinlock);
+	SpinLockAcquire(&dtm->spinlock);
 #else
 	LWLockAcquire(dtm->hashLock, mode);
 #endif
@@ -156,7 +157,7 @@ void MtmLock(LWLockMode mode)
 void MtmUnlock(void)
 {
 #ifdef USE_SPINLOCK
-	SpinLockRelease(&dtm->hashSpinlock);
+	SpinLockRelease(&dtm->spinlock);
 #else
 	LWLockRelease(dtm->hashLock);
 #endif
@@ -409,20 +410,22 @@ static void MtmInitialize()
 	dtm = (MtmState*)ShmemInitStruct(MULTIMASTER_NAME, sizeof(MtmState), &found);
 	if (!found)
 	{
+		dtm->status = MTM_INITIALIZATION;
+		dtm->recoverySlot = 0;
 		dtm->hashLock = (LWLock*)GetNamedLWLockTranche(MULTIMASTER_NAME);
 		dtm->csn = MtmGetCurrentTime();
 		dtm->oldestXid = FirstNormalTransactionId;
         dtm->nNodes = MtmNodes;
 		dtm->disabledNodeMask = 0;
+		dtm->pglogicalNodeMask = 0;
 		dtm->votingTransactions = NULL;
         dtm->transListHead = NULL;
-        dtm->transListTail = &dtm->transListHead;
-        pg_atomic_write_u32(&dtm->nReceivers, 0);
+        dtm->transListTail = &dtm->transListHead;		
+        dtm->nReceivers = 0;
 		dtm->timeShift = 0;
-        dtm->initialized = false;
 		PGSemaphoreCreate(&dtm->votingSemaphore);
 		PGSemaphoreReset(&dtm->votingSemaphore);
-		SpinLockInit(&dtm->hashSpinlock);
+		SpinLockInit(&dtm->spinlock);
         BgwPoolInit(&dtm->pool, MtmExecutor, MtmDatabaseName, MtmQueueSize);
 		RegisterXactCallback(MtmXactCallback, NULL);
 		dtmTx.snapshot = INVALID_CSN;
@@ -463,7 +466,7 @@ MtmBeginTransaction(MtmCurrentTrans* x)
 		MtmLock(LW_EXCLUSIVE);
 		x->xid = GetCurrentTransactionIdIfAny();
         x->isReplicated = false;
-        x->isDistributed = IsNormalProcessingMode() && dtm->initialized && MtmDoReplication && !am_walsender && !IsBackgroundWorker && !IsAutoVacuumWorkerProcess();
+        x->isDistributed = IsNormalProcessingMode() && dtm->status == MTM_ONLINE && MtmDoReplication && !am_walsender && !IsBackgroundWorker && !IsAutoVacuumWorkerProcess();
 		x->containsDML = false;
         x->snapshot = MtmAssignCSN();	
 		x->gtid.xid = InvalidTransactionId;
@@ -575,8 +578,6 @@ MtmFinishTransaction(TransactionId xid, int nsubxids, TransactionId *subxids, Xi
 		XidStatus prevStatus = TRANSACTION_STATUS_UNKNOWN;
 		bool found;
 
-		Assert(status == TRANSACTION_STATUS_ABORTED);
-
 		MtmLock(LW_EXCLUSIVE);
 		ts = hash_search(xid2state, &xid, HASH_ENTER, &found);
 		if (!found) {
@@ -590,7 +591,7 @@ MtmFinishTransaction(TransactionId xid, int nsubxids, TransactionId *subxids, Xi
 		ts->status = status;
 		MtmAdjustSubtransactions(ts);
 
-		if (prevStatus != TRANSACTION_STATUS_ABORTED) {
+		if (dtm->status != MTM_RECOVERY && prevStatus != TRANSACTION_STATUS_ABORTED) {
 			ts->cmd = MSG_ABORTED;
 			MtmSendNotificationMessage(ts);
 		}
@@ -607,7 +608,7 @@ MtmSetTransactionStatus(TransactionId xid, int nsubxids, TransactionId *subxids,
 	MTM_TRACE("%d: MtmSetTransactionStatus %u(%u) = %u, isDistributed=%d\n", getpid(), xid, dtmTx.xid, status, dtmTx.isDistributed);
 	if (xid == dtmTx.xid && dtmTx.isDistributed)
 	{
-		if (status == TRANSACTION_STATUS_ABORTED || !dtmTx.containsDML)
+		if (status == TRANSACTION_STATUS_ABORTED || !dtmTx.containsDML || dtm->status == MTM_RECOVERY)
 		{
 			MtmFinishTransaction(xid, nsubxids, subxids, status);	
 			MTM_TRACE("Finish transaction %d, status=%d, DML=%d\n", xid, status, dtmTx.containsDML);
@@ -863,11 +864,17 @@ void MtmJoinTransaction(GlobalTransactionId* gtid, csn_t globalSnapshot)
 	dtmTx.containsDML = true;
 }
 
-void MtmReceiverStarted()
+void MtmReceiverStarted(int nodeId)
 {
-     if (pg_atomic_fetch_add_u32(&dtm->nReceivers, 1) == dtm->nNodes-2) {
-         dtm->initialized = true;
+	SpinLockAcquire(&dtm->spinlock);	
+	if (!BIT_CHECK(dtm->pglogicalNodeMask, nodeId-1)) { 
+		dtm->pglogicalNodeMask |= (int64)1 << (nodeId-1);
+		if (++dtm->nReceivers == dtm->nNodes-1) {
+			Assert(dtm->status == MTM_CONNECTED);
+			dtm->status = MTM_ONLINE;
+		}
      }
+	SpinLockRelease(&dtm->spinlock);	
 }
 
 csn_t MtmTransactionSnapshot(TransactionId xid)
@@ -885,10 +892,23 @@ csn_t MtmTransactionSnapshot(TransactionId xid)
     return snapshot;
 }
 
-
+MtmSlotMode MtmReceiverSlotMode(int nodeId)
+{
+	while (dtm->status != MTM_CONNECTED && dtm->status != MTM_ONLINE) { 		
+		if (dtm->status == MTM_RECOVERY) { 
+			if (dtm->recoverySlot == 0 || dtm->recoverySlot == nodeId) { 
+				dtm->recoverySlot = nodeId;
+				return SLOT_OPEN_EXISTED;
+			}
+		}
+		MtmSleep(STATUS_POLL_DELAY);
+	}
+	return dtm->recoverySlot ? SLOT_CREATE_NEW : SLOT_OPEN_ALWAYS;
+}
+			
 void MtmDropNode(int nodeId, bool dropSlot)
 {
-	if (!BIT_SET(dtm->disabledNodeMask, nodeId-1))
+	if (!BIT_CHECK(dtm->disabledNodeMask, nodeId-1))
 	{
 		if (nodeId <= 0 || nodeId > dtm->nNodes) 
 		{ 
@@ -969,7 +989,7 @@ static void MtmBroadcastUtilityStmt(char const* sql, bool ignoreError)
 			p = conn_str_end;
 		}
 		*p = '\0';
-		if (!BIT_SET(disabledNodeMask, i)) 
+		if (!BIT_CHECK(disabledNodeMask, i)) 
 		{
 			conns[i] = PQconnectdb(conn_str);
 			if (PQstatus(conns[i]) != CONNECTION_OK)

multimaster.h

@@ -12,7 +12,7 @@
 #define MTM_TUPLE_TRACE(fmt, ...) fprintf(stderr, fmt, ## __VA_ARGS__) 
 */
 
-#define BIT_SET(mask, bit) ((mask) & ((int64)1 << (bit)))
+#define BIT_CHECK(mask, bit) ((mask) & ((int64)1 << (bit)))
 
 #define MULTIMASTER_NAME           "mtm"
 #define MULTIMASTER_SCHEMA_NAME    "mtm"
@@ -46,9 +46,25 @@ typedef enum
 	MSG_ABORT,
 	MSG_PREPARED,
 	MSG_COMMITTED,
-	MSG_ABORTED
+	MSG_ABORTED,
+	MSG_STATUS
 } MtmMessageCode;
 
+typedef enum
+{
+	MTM_INITIALIZATION, /* Initial status */
+	MTM_OFFLINE,        /* Node is out of quorum */
+	MTM_CONNECTED,      /* Arbiter is established connections with other nodes */
+	MTM_ONLINE,         /* Ready to receive client's queries */
+	MTM_RECOVERY        /* Node is in recovery process */
+} MtmNodeStatus;
+
+typedef enum
+{
+	SLOT_CREATE_NEW,   /* create new slot (drop existed) */
+	SLOT_OPEN_EXISTED, /* open existed slot */
+	SLOT_OPEN_ALWAYS,  /* open existed slot or create new if noty exists */
+} MtmSlotMode;
 
 typedef struct MtmTransState
 {
@@ -71,16 +87,18 @@ typedef struct MtmTransState
 
 typedef struct
 {
-	volatile slock_t hashSpinlock;     /* spinlock used to protect access to hash table */
+	MtmNodeStatus status;              /* Status of this node */
+	int recoverySlot;                  /* NodeId of recovery slot or 0 if none */
+	volatile slock_t spinlock;         /* spinlock used to protect access to hash table */
 	PGSemaphoreData votingSemaphore;   /* semaphore used to notify mtm-sender about new responses to coordinator */
 	LWLockId hashLock;                 /* lock to synchronize access to hash table */
 	TransactionId oldestXid;           /* XID of oldest transaction visible by any active transaction (local or global) */
-	int64  disabledNodeMask;           /* bitmask of disable nodes (so no more than 64 nodes in multimaster:) */
+	int64  disabledNodeMask;           /* bitmask of disabled nodes (so no more than 64 nodes in multimaster:) */
+	int64  pglogicalNodeMask;          /* bitmask of started pglogic receviers */
     int    nNodes;                     /* number of active nodes */
-    pg_atomic_uint32 nReceivers;       /* number of initialized logical receivers (used to determine moment when Mtm intialization is completed */
-	long timeShift;                    /* local time correction */
-    bool initialized;             
-	csn_t csn;                         /* last obtained CSN: used to provide unique acending CSNs based on system time */
+    int    nReceivers;                 /* number of initialized logical receivers (used to determine moment when Mtm intialization is completed */
+	long   timeShift;                  /* local time correction */
+	csn_t  csn;                        /* last obtained CSN: used to provide unique acending CSNs based on system time */
 	MtmTransState* votingTransactions; /* L1-list of replicated transactions sendings notifications to coordinator.
 									 	 This list is used to pass information to mtm-sender BGW */
     MtmTransState* transListHead;      /* L1 list of all finished transactions present in xid2state hash.
@@ -107,7 +125,8 @@ extern csn_t MtmTransactionSnapshot(TransactionId xid);
 extern csn_t MtmAssignCSN(void);
 extern csn_t MtmSyncClock(csn_t csn);
 extern void  MtmJoinTransaction(GlobalTransactionId* gtid, csn_t snapshot);
-extern void  MtmReceiverStarted(void);
+extern void  MtmReceiverStarted(int nodeId);
+extern MtmSlotMode MtmReceiverSlotMode(int nodeId);
 extern void  MtmExecute(void* work, int size);
 extern void  MtmExecutor(int id, void* work, size_t size);
 extern HTAB* MtmCreateHash(void);

pglogical_proto.c

@@ -108,7 +108,7 @@ pglogical_write_begin(StringInfo out, PGLogicalOutputData *data,
     PGLogicalProtoMM* mm = (PGLogicalProtoMM*)data->api;
 	csn_t csn = MtmTransactionSnapshot(txn->xid);
 	MTM_TRACE("pglogical_write_begin %d CSN=%ld\n", txn->xid, csn);
-    if (csn == INVALID_CSN || BIT_SET(mm->state->disabledNodeMask, mm->nodeId-1)) {
+    if (csn == INVALID_CSN || BIT_CHECK(mm->state->disabledNodeMask, mm->nodeId-1)) {
         mm->isLocal = true;
     } else { 
         mm->isLocal = false;