Update Paxos interface

Author: knizhnik

arbiter.c

@@ -297,7 +297,7 @@ static int MtmConnectSocket(char const* host, int port, int max_attempts)
 
 			/* Some node considered that I am dead, so switch to recovery mode */
 			if (BIT_CHECK(msg.disabledNodeMask, MtmNodeId-1)) { 
-				MtmClusterSwitchMode(MTM_RECOVERY);
+				MtmSwitchClusterMode(MTM_RECOVERY);
 			}
 			/* Combine disable masks from all node. Is it actually correct or we should better check availability of nodes ourselves? */
 			ds->disabledNodeMask |= msg.disabledNodeMask;
@@ -344,7 +344,7 @@ static void MtmOpenConnections()
 		elog(WARNING, "Node is out of quorum: only %d nodes from %d are accssible", ds->nNodes, MtmNodes);
 		ds->status = MTM_OFFLINE;
 	} else if (ds->status == MTM_INITIALIZATION) { 
-		MtmClusterSwitchMode(MTM_CONNECTED);
+		MtmSwitchClusterMode(MTM_CONNECTED);
 	}
 }
 
@@ -745,7 +745,7 @@ static void MtmTransReceiver(Datum arg)
 		}
 		if (n == 0 && ds->disabledNodeMask != 0) { 
 			/* If timeout is expired and there are didabled nodes, then recheck cluster's state */
-			MtmUpdateClusterStatus();
+			MtmRefreshClusterStatus(false);
 		}
 	}
 }

multimaster.c

@@ -131,7 +131,7 @@ static TransactionManager MtmTM = {
 	MtmGetName 
 };
 
-static char const* const MtmNodeStatusMnem[] = 
+char const* const MtmNodeStatusMnem[] = 
 { 
 	"Intialization", 
 	"Offline", 
@@ -602,7 +602,7 @@ static void MtmPrecommitTransaction(MtmCurrentTrans* x)
 	x->xid = GetCurrentTransactionId();
 
 	if (dtm->disabledNodeMask != 0) { 
-		MtmUpdateClusterStatus();
+		MtmRefreshClusterStatus(true);
 		if (dtm->status != MTM_ONLINE) { 
 			elog(ERROR, "Abort current transaction because this cluster node is not online");			
 		}
@@ -1096,7 +1096,7 @@ _PG_fini(void)
  */
 
 
-void MtmClusterSwitchMode(MtmNodeStatus mode)
+void MtmSwitchClusterMode(MtmNodeStatus mode)
 {
 	dtm->status = mode;
 	elog(WARNING, "Switch to %s mode", MtmNodeStatusMnem[mode]);
@@ -1121,7 +1121,8 @@ void MtmJoinTransaction(GlobalTransactionId* gtid, csn_t globalSnapshot)
 		Assert(dtm->status == MTM_RECOVERY);
 	} else if (dtm->status == MTM_RECOVERY) { 
 		/* When recovery is completed we get normal transaction ID and switch to normal mode */
-		MtmClusterSwitchMode(MTM_ONLINE);
+		dtm->recoverySlot = 0;
+		MtmSwitchClusterMode(MTM_ONLINE);
 	}
 	dtmTx.gtid = *gtid;
 	dtmTx.xid = GetCurrentTransactionId();
@@ -1137,9 +1138,8 @@ void MtmReceiverStarted(int nodeId)
 	if (!BIT_CHECK(dtm->pglogicalNodeMask, nodeId-1)) { 
 		BIT_SET(dtm->pglogicalNodeMask, nodeId-1);
 		if (++dtm->nReceivers == dtm->nNodes-1) {
-			elog(WARNING, "All receivers are started, switch to normal mode");
 			Assert(dtm->status == MTM_CONNECTED);
-			dtm->status = MTM_ONLINE;
+			MtmSwitchClusterMode(MTM_OFFLINE);
 		}
      }
 	SpinLockRelease(&dtm->spinlock);	
@@ -1632,14 +1632,14 @@ MtmDetectGlobalDeadLock(PGPROC* proc)
 
         ByteBufferAlloc(&buf);
         EnumerateLocks(MtmSerializeLock, &buf);
-		PaxosSet(psprintf("lock-graph-%d", MtmNodeId), buf.data, buf.used);
+		PaxosSet(psprintf("lock-graph-%d", MtmNodeId), buf.data, buf.used, true);
 		MtmGraphInit(&graph);
 		MtmGraphAdd(&graph, (GlobalTransactionId*)buf.data, buf.used/sizeof(GlobalTransactionId));
         ByteBufferFree(&buf);
 		for (i = 0; i < MtmNodes; i++) { 
 			if (i+1 != MtmNodeId && !BIT_CHECK(dtm->disabledNodeMask, i)) { 
 				int size;
-				void* data = PaxosGet(psprintf("lock-graph-%d", i+1), &size, NULL);
+				void* data = PaxosGet(psprintf("lock-graph-%d", i+1), &size, NULL, true);
 				if (data == NULL) { 
 					hasDeadlock = true; /* Just temporary hack until no Paxos */
 				} else { 
@@ -1655,12 +1655,12 @@ MtmDetectGlobalDeadLock(PGPROC* proc)
 }
 
 static void 
-MtmBuildConnectivityMatrix(nodemask_t* matrix)
+MtmBuildConnectivityMatrix(nodemask_t* matrix, bool nowait)
 {
 	int i, j, n = MtmNodes;
 	for (i = 0; i < n; i++) { 
 		if (i+1 != MtmNodeId) { 
-			void* data = PaxosGet(psprintf("node-mask-%d", i+1), NULL, NULL);
+			void* data = PaxosGet(psprintf("node-mask-%d", i+1), NULL, NULL, nowait);
 			matrix[i] = *(nodemask_t*)data;
 		} else { 
 			matrix[i] = dtm->connectivityMask;
@@ -1678,14 +1678,14 @@ MtmBuildConnectivityMatrix(nodemask_t* matrix)
  * Build connectivity graph, find clique in it and extend disabledNodeMask by nodes not included in clique.
  * This function returns false if current node is excluded from cluster, true otherwise
  */
-void MtmUpdateClusterStatus(void)
+void MtmRefreshClusterStatus(bool nowait)
 {
 	nodemask_t mask, clique;
 	nodemask_t matrix[MAX_NODES];
 	int clique_size;
 	int i;
 
-	MtmBuildConnectivityMatrix(matrix);
+	MtmBuildConnectivityMatrix(matrix, nowait);
 
 	clique = MtmFindMaxClique(matrix, MtmNodes, &clique_size);
 	if (clique_size >= MtmNodes/2+1) { /* have quorum */
@@ -1708,11 +1708,11 @@ void MtmUpdateClusterStatus(void)
 		if (BIT_CHECK(dtm->disabledNodeMask, MtmNodeId-1)) { 
 			if (dtm->status == MTM_ONLINE) {
 				/* I was excluded from cluster:( */
-				MtmClusterSwitchMode(MTM_OFFLINE);
+				MtmSwitchClusterMode(MTM_OFFLINE);
 			}
 		} else if (dtm->status == MTM_OFFLINE) {
 			/* Should we somehow restart logical receivers? */ 
-			MtmClusterSwitchMode(MTM_RECOVERY);
+			MtmSwitchClusterMode(MTM_RECOVERY);
 		}
 	} else { 
 		elog(WARNING, "Clique %lx has no quorum", clique);
@@ -1722,30 +1722,30 @@ void MtmUpdateClusterStatus(void)
 void MtmOnNodeDisconnect(int nodeId)
 {
 	BIT_SET(dtm->connectivityMask, nodeId-1);
-	PaxosSet(psprintf("node-mask-%d", MtmNodeId), &dtm->connectivityMask, sizeof dtm->connectivityMask);
+	PaxosSet(psprintf("node-mask-%d", MtmNodeId), &dtm->connectivityMask, sizeof dtm->connectivityMask, false);
 
 	/* Wait more than socket KEEPALIVE timeout to let other nodes update their statuses */
 	MtmSleep(MtmKeepaliveTimeout);
 
-	MtmUpdateClusterStatus();
+	MtmRefreshClusterStatus(false);
 }
 
 void MtmOnNodeConnect(int nodeId)
 {
 	BIT_CLEAR(dtm->connectivityMask, nodeId-1);
-	PaxosSet(psprintf("node-mask-%d", MtmNodeId), &dtm->connectivityMask, sizeof dtm->connectivityMask);
+	PaxosSet(psprintf("node-mask-%d", MtmNodeId), &dtm->connectivityMask, sizeof dtm->connectivityMask, false);
 }
 
 /*
  * Paxos function stubs (until them are miplemented)
  */
-void* PaxosGet(char const* key, int* size, PaxosTimestamp* ts)
+void* PaxosGet(char const* key, int* size, PaxosTimestamp* ts, bool nowait)
 {
 	if (size != NULL) { 
 		*size = 0;
 	}
 	return NULL;
 }
 
-void  PaxosSet(char const* key, void const* value, int size)
+void  PaxosSet(char const* key, void const* value, int size, bool nowait)
 {}

multimaster.h

@@ -118,6 +118,8 @@ typedef struct
 
 #define MtmIsCoordinator(ts) (ts->gtid.node == MtmNodeId)
 
+extern char const* const MtmNodeStatusMnem[];
+
 extern char* MtmConnStrs;
 extern int   MtmNodeId;
 extern int   MtmNodes;
@@ -150,6 +152,6 @@ extern MtmState* MtmGetState(void);
 extern timestamp_t MtmGetCurrentTime(void);
 extern void  MtmSleep(timestamp_t interval);
 extern bool  MtmIsRecoveredNode(int nodeId);
-extern void  MtmUpdateClusterStatus(void);
-extern void  MtmClusterSwitchMode(MtmNodeStatus mode);
+extern void  MtmRefreshClusterStatus(bool nowait);
+extern void  MtmSwitchClusterMode(MtmNodeStatus mode);
 #endif

paxos.h

@@ -7,7 +7,26 @@ typedef struct PaxosTimestamp {
 	uint32 psn;    /* PAXOS serial number */
 } PaxosTimestamp;
 
-extern void* PaxosGet(char const* key, int* size, PaxosTimestamp* ts);
-extern void  PaxosSet(char const* key, void const* value, int size);
+/*
+ * Get key value.
+ * Returns NULL if key doesn't exist.
+ * Value should be copied into caller's private memory using palloc.
+ * If "size" is not NULL, then it is assigned size of value.
+ * If "ts" is not NULL, then it is assigned timestamp of last update of this value
+ * If RAFT master is not accessible, then depending non value of "nowait" parameter, this funciton should either block until RAFT quorum is reached, either report error.
+ */
+extern void* PaxosGet(char const* key, int* size, PaxosTimestamp* ts, bool nowait);
 
+/*
+ * Set new value for the specified key. IF value is NULL, then key should be deleted.
+ * If RAFT master is not accessible, then depending non value of "nowait" parameter, this funciton should either block until RAFT quorum is reached, either report error.
+ */
+extern void  PaxosSet(char const* key, void const* value, int size, bool nowait);
+
+/* 
+ * If key doesn't exists or its value is not equal to the specified value then store this value and return true.
+ * Otherwise do nothing and return false.
+ * If RAFT master is not accessible, then depending non value of "nowait" parameter, this funciton should either block until RAFT quorum is reached, either report error.
+ */
+extern bool  PaxosCAS(char const* key, char const* value, bool nowait);
 #endif

pglogical_receiver.c

@@ -332,6 +332,12 @@ pglogical_receiver_main(Datum main_arg)
 		if (rc & WL_POSTMASTER_DEATH)
 			proc_exit(1);
 
+		if (ds->status != MTM_ONLINE && (ds->status != MTM_RECOVERY || ds->recoverySlot != args->remote_node)) {
+			ereport(LOG, (errmsg("%s: terminating WAL receiver because node is switched to %s mode", worker_proc, MtmNodeStatusMnem[ds->status])));
+			proc_exit(0);
+		}
+			
+
 		/* Some cleanup */
 		if (copybuf != NULL)
 		{