Sparse nodemasks

Author: kelvich

src/commit.c

@@ -100,6 +100,8 @@ MtmXactCallback2(XactEvent event, void *arg)
 void
 MtmBeginTransaction()
 {
+	MtmNodeStatus	node_status;
+
 	// XXX: clean MtmTx on commit and check on begin that it is clean.
 	// That should unveil probable issues with subxacts.
 
@@ -138,8 +140,10 @@ MtmBeginTransaction()
 
 	MtmDDLResetStatement();
 
+	node_status = MtmGetCurrentStatus();
+
 	/* Application name can be changed using PGAPPNAME environment variable */
-	if (Mtm->status != MTM_ONLINE
+	if (node_status != MTM_ONLINE
 		&& strcmp(application_name, MULTIMASTER_ADMIN) != 0
 		&& strcmp(application_name, MULTIMASTER_BROADCAST_SERVICE) != 0)
 	{
@@ -150,13 +154,13 @@ MtmBeginTransaction()
 		{
 			mtm_log(ERROR,
 					"Multimaster node is not online: current status %s",
-					MtmNodeStatusMnem[Mtm->status]);
+					MtmNodeStatusMnem[node_status]);
 		}
 		else
 		{
 			mtm_log(FATAL,
 					"Multimaster node is not online: current status %s",
-					MtmNodeStatusMnem[Mtm->status]);
+					MtmNodeStatusMnem[node_status]);
 		}
 	}
 }
@@ -216,12 +220,8 @@ MtmTwoPhaseCommit()
 
 	LWLockAcquire(MtmCommitBarrier, LW_SHARED);
 
-	MtmLock(LW_SHARED);
-	participantsMask = ~Mtm->disabledNodeMask &
+	participantsMask = MtmGetEnabledNodeMask() &
 						~((nodemask_t)1 << (mtm_cfg->my_node_id-1));
-	if (Mtm->status != MTM_ONLINE)
-		mtm_log(ERROR, "This node became offline during current transaction");
-	MtmUnlock();
 
 	ret = PrepareTransactionBlock(gid);
 	if (!ret)
@@ -305,24 +305,13 @@ GatherPrepares(TransactionId xid, nodemask_t participantsMask, int *failed_at)
 			 * disconnected. Let's wait when it became disabled as we can
 			 * became offline by this time.
 			 */
-			MtmLock(LW_SHARED);
-			if (BIT_CHECK(Mtm->disabledNodeMask, sender_to_node[sender_id] - 1))
+			if (!BIT_CHECK(MtmGetEnabledNodeMask(), sender_to_node[sender_id] - 1))
 			{
-				if (Mtm->status != MTM_ONLINE)
-				{
-					elog(ERROR, "our node was disabled during transaction commit");
-				}
-				else
-				{
-					BIT_CLEAR(participantsMask, sender_to_node[sender_id] - 1);
-					mtm_log(MtmTxTrace,
-						"GatherPrepares: dropping node%d from participants of tx" XID_FMT,
-						sender_to_node[sender_id], xid);
-					prepared = false;
-					*failed_at = sender_to_node[sender_id];
-				}
+				BIT_CLEAR(participantsMask, sender_to_node[sender_id] - 1);
+				mtm_log(MtmTxTrace,
+					"GatherPrepares: dropping node%d from participants of tx" XID_FMT,
+					sender_to_node[sender_id], xid);
 			}
-			MtmUnlock();
 		}
 	}
 
@@ -365,22 +354,13 @@ GatherPrecommits(TransactionId xid, nodemask_t participantsMask, MtmMessageCode
 			 * disconnected. Let's wait when it became disabled as we can
 			 * became offline by this time.
 			 */
-			MtmLock(LW_SHARED);
-			if (BIT_CHECK(Mtm->disabledNodeMask, sender_to_node[sender_id] - 1))
+			if (!BIT_CHECK(MtmGetEnabledNodeMask(), sender_to_node[sender_id] - 1))
 			{
-				if (Mtm->status != MTM_ONLINE)
-				{
-					elog(ERROR, "our node was disabled during transaction commit");
-				}
-				else
-				{
-					BIT_CLEAR(participantsMask, sender_to_node[sender_id] - 1);
-					mtm_log(MtmTxTrace,
-						"GatherPrecommit: dropping node%d from participants of tx" XID_FMT,
-						sender_to_node[sender_id], xid);
-				}
+				BIT_CLEAR(participantsMask, sender_to_node[sender_id] - 1);
+				mtm_log(MtmTxTrace,
+					"GatherPrecommit: dropping node%d from participants of tx" XID_FMT,
+					sender_to_node[sender_id], xid);
 			}
-			MtmUnlock();
 		}
 	}
 

src/ddd.c

@@ -17,6 +17,7 @@
 #include "ddd.h"
 #include "bytebuf.h"
 #include "mm.h"
+#include "state.h"
 #include "logger.h"
 
 #define LOCK_BY_INDEX(i) ((LWLockId)&ddd_shared->locks[(i)])
@@ -383,7 +384,7 @@ MtmDetectGlobalDeadLockForXid(TransactionId xid)
 	MtmGraphAdd(&graph, (GlobalTransactionId*)buf.data, buf.used/sizeof(GlobalTransactionId));
 	ByteBufferFree(&buf);
 	for (i = 0; i < ddd_shared->n_nodes; i++) {
-		if (i+1 != Mtm->my_node_id && !BIT_CHECK(Mtm->disabledNodeMask, i)) {
+		if (i+1 != Mtm->my_node_id && BIT_CHECK(MtmGetEnabledNodeMask(), i)) {
 			size_t lockGraphSize;
 			void* lockGraphData;
 
@@ -439,7 +440,7 @@ MtmDetectGlobalDeadLock(PGPROC* proc)
 
 	mtm_log(DeadlockCheck, "Detect global deadlock for " XID_FMT " by backend %d", pgxact->xid, MyProcPid);
 
-	if (Mtm->status != MTM_ONLINE || !TransactionIdIsValid(pgxact->xid))
+	if (!TransactionIdIsValid(pgxact->xid))
 		return false;
 
 	return MtmDetectGlobalDeadLockForXid(pgxact->xid);

src/include/multimaster.h

@@ -133,16 +133,6 @@ typedef enum
 	MSG_POLL_STATUS
 } MtmMessageCode;
 
-typedef enum
-{
-	MTM_DISABLED,				/* Node disabled */
-	MTM_RECOVERY,				/* Node is in recovery process */
-	MTM_RECOVERED,				/* Node is recovered by is not yet switched to
-								 * ONLINE because not all sender/receivers are
-								 * restarted */
-	MTM_ONLINE					/* Ready to receive client's queries */
-} MtmNodeStatus;
-
 typedef struct
 {
 	MtmMessageCode code;		/* Message code: MSG_PREPARE, MSG_PRECOMMIT,
@@ -184,54 +174,28 @@ typedef struct
 {
 	int			my_node_id;
 	bool		stop_new_commits;
-	bool		recovered;
 	XLogRecPtr	latestSyncpoint;
-	MtmNodeStatus status;		/* Status of this node */
-	char	   *statusReason;	/* A human-readable description of why the
-								 * current status was set */
-	int			recoverySlot;	/* NodeId of recovery slot or 0 if none */
-	LWLockPadded *locks;		/* multimaster lock tranche */
-	nodemask_t	selfConnectivityMask;
-	nodemask_t	disabledNodeMask;	/* Bitmask of disabled nodes */
-	nodemask_t	clique;			/* Bitmask of nodes that are connected and we
-								 * allowed to connect/send wal/receive wal
-								 * with them */
-	bool		refereeGrant;	/* Referee allowed us to work with half of the
-								 * nodes */
-	int			refereeWinnerId;	/* Node that won referee contest */
-	nodemask_t	stalledNodeMask;	/* Bitmask of stalled nodes (node with
-									 * dropped replication slot which makes it
-									 * not possible automatic recovery of such
-									 * node) */
-	nodemask_t	stoppedNodeMask;	/* Bitmask of stopped (permanently
-									 * disabled nodes) */
-	nodemask_t	pglogicalReceiverMask;	/* bitmask of started pglogic
-										 * receivers */
-	nodemask_t	pglogicalSenderMask;	/* bitmask of started pglogic senders */
 	bool		localTablesHashLoaded;	/* Whether data from local_tables
 										 * table is loaded in shared memory
 										 * hash table */
-	int			nAllNodes;		/* Total number of nodes */
-	int			recoveryCount;	/* Number of completed recoveries */
-	int			donorNodeId;	/* Cluster node from which this node was
-								 * populated */
 	struct {
 		MtmReplicationMode	receiver_mode;
 		pid_t				sender_pid;
 		pid_t				receiver_pid;
 		int					dmq_dest_id;
 	} peers[MTM_MAX_NODES];
 	BgwPool		pools[FLEXIBLE_ARRAY_MEMBER];		/* [Mtm->nAllNodes]: per-node data */
-} MtmState;
+} MtmShared;
 
-extern MtmState *Mtm;
+extern MtmShared *Mtm;
 
 /* XXX: to delete */
 extern int	MtmReplicationNodeId;
 extern MtmCurrentTrans MtmTx;
 extern MemoryContext MtmApplyContext;
 
 /* Locks */
+extern LWLock *MtmLock;
 extern LWLock *MtmCommitBarrier;
 extern LWLock *MtmReceiverBarrier;
 extern LWLock *MtmSyncpointLock;
@@ -252,10 +216,6 @@ extern int	MtmMaxWorkers;
 extern int	MtmMaxNodes;
 extern bool	MtmBreakConnection;
 
-/*  XXX! need rename: that's actually a disconnectivity mask */
-#define SELF_CONNECTIVITY_MASK  (Mtm->selfConnectivityMask)
-#define EFFECTIVE_CONNECTIVITY_MASK  ( SELF_CONNECTIVITY_MASK | Mtm->stoppedNodeMask | ~Mtm->clique )
-
 extern bool MtmTwoPhaseCommit(void);
 extern void MtmBeginTransaction(void);
 
@@ -278,12 +238,6 @@ extern MtmNode *MtmNodeById(MtmConfig *cfg, int node_id);
 
 extern void MtmStateFill(MtmConfig *cfg);
 
-extern void MtmLock(LWLockMode mode);
-extern void MtmUnlock(void);
-extern void MtmLockNode(int nodeId, LWLockMode mode);
-extern bool MtmTryLockNode(int nodeId, LWLockMode mode);
-extern void MtmUnlockNode(int nodeId);
-
 extern void MtmInitMessage(MtmArbiterMessage *msg, MtmMessageCode code);
 
 extern bool MtmIsEnabled(void);

src/include/resolver.h

@@ -20,8 +20,8 @@ typedef int MtmTxStateMask;
 extern void ResolverMain(Datum main_arg);
 extern void ResolverInit(void);
 extern BackgroundWorkerHandle *ResolverStart(Oid db_id, Oid user_id);
-extern void ResolveTransactionsForNode(int node_id);
-extern void ResolveAllTransactions(void);
+extern void ResolveTransactionsForNode(int node_id, int n_all_nodes);
+extern void ResolveAllTransactions(int n_all_nodes);
 extern char *MtmTxStateMnem(MtmTxState state);
 
 #endif  /* RESOLVER_H */

src/include/state.h

@@ -1,11 +1,15 @@
+#ifndef STATE_H
+#define STATE_H
 
 typedef enum
 {
 	MTM_NEIGHBOR_CLIQUE_DISABLE,
 	MTM_NEIGHBOR_WAL_RECEIVER_START,
+	MTM_NEIGHBOR_WAL_RECEIVER_ERROR,
 	MTM_NEIGHBOR_WAL_SENDER_START_RECOVERY,
 	MTM_NEIGHBOR_WAL_SENDER_START_RECOVERED,
-	MTM_NEIGHBOR_RECOVERY_CAUGHTUP
+	MTM_NEIGHBOR_RECOVERY_CAUGHTUP,
+	MTM_NEIGHBOR_WAL_SENDER_STOP
 } MtmNeighborEvent;
 
 typedef enum
@@ -21,24 +25,40 @@ typedef enum
 	MTM_NONRECOVERABLE_ERROR
 } MtmEvent;
 
-extern char const* const MtmNodeStatusMnem[];
+typedef enum
+{
+	MTM_DISABLED,				/* Node disabled */
+	MTM_RECOVERY,				/* Node is in recovery process */
+	MTM_RECOVERED,				/* Node is recovered by is not yet switched to
+								 * ONLINE because not all sender/receivers are
+								 * restarted */
+	MTM_ONLINE					/* Ready to receive client's queries */
+} MtmNodeStatus;
+
+extern char const *const MtmNodeStatusMnem[];
 
+extern void MtmStateInit(void);
+extern void MtmStateShmemStartup(void);
 extern void MtmStateLoad(MtmConfig *cfg);
 
 extern void MtmMonitorStart(Oid db_id, Oid user_id);
 
 extern void MtmStateProcessNeighborEvent(int node_id, MtmNeighborEvent ev, bool locked);
 extern void MtmStateProcessEvent(MtmEvent ev, bool locked);
-extern void MtmDisableNode(int nodeId);
-extern void MtmEnableNode(int nodeId);
 
 extern void MtmOnNodeDisconnect(char *node_name);
 extern void MtmOnNodeConnect(char *node_name);
-extern void MtmReconnectNode(int nodeId);
 
 extern void MtmRefreshClusterStatus(void);
 
 extern int countZeroBits(nodemask_t mask, int nNodes);
 
+extern MtmReplicationMode MtmGetReplicationMode(int nodeId);
+
 extern MtmNodeStatus MtmGetCurrentStatus(void);
-extern nodemask_t MtmGetDisabledNodeMask(void);
+extern nodemask_t MtmGetDisabledNodeMask(void);
+extern nodemask_t MtmGetConnectedNodeMask(void);
+extern nodemask_t MtmGetEnabledNodeMask(void);
+extern int MtmGetRecoveryCount(void);
+
+#endif