Perform drop node under lock

Author: knizhnik

multimaster.c

@@ -371,6 +371,11 @@ void MtmLockNode(int nodeId, LWLockMode mode)
 	LWLockAcquire((LWLockId)&Mtm->locks[nodeId], mode);
 }
 
+bool MtmTryLockNode(int nodeId, LWLockMode mode)
+{
+	return LWLockConditionalAcquire((LWLockId)&Mtm->locks[nodeId], mode);
+}
+
 void MtmUnlockNode(int nodeId)
 {
 	Assert(nodeId > 0 && nodeId <= MtmMaxNodes*2);
@@ -1684,7 +1689,14 @@ static void MtmStartRecovery()
 
 static void MtmDropSlot(int nodeId)
 {
-	ReplicationSlotDrop(psprintf(MULTIMASTER_SLOT_PATTERN, nodeId));
+	if (MtmTryLockNode(nodeId, LW_EXCLUSIVE))
+	{
+		MTM_ELOG(INFO, "Drop replication slot for node %d", nodeId);
+		ReplicationSlotDrop(psprintf(MULTIMASTER_SLOT_PATTERN, nodeId));
+		MtmUnlockNode(nodeId);
+	} else {
+		MTM_ELOG(WARNING, "Failed to drop replication slot for node %d", nodeId);
+	}
 	MtmLock(LW_EXCLUSIVE);
 	BIT_SET(Mtm->stalledNodeMask, nodeId-1);
 	BIT_SET(Mtm->stoppedNodeMask, nodeId-1); /* stalled node can not be automatically recovered */
@@ -2766,7 +2778,7 @@ void MtmUpdateNodeConnectionInfo(MtmConnectionInfo* conn, char const* connStr)
 	} else {
 		conn->arbiterPort = MULTIMASTER_DEFAULT_ARBITER_PORT;
 	}
-	MTM_ELOG(WARNING, "Using arbiter port: %d", conn->arbiterPort);
+	MTM_ELOG(INFO, "Using arbiter port: %d", conn->arbiterPort);
 
 	port = strstr(connStr, " port=");
 	if (port == NULL && strncmp(connStr, "port=", 5) == 0) {
@@ -3422,7 +3434,7 @@ _PG_init(void)
 	PreviousProcessUtilityHook = ProcessUtility_hook;
 	ProcessUtility_hook = MtmProcessUtility;
 
-	PreviousSeqNextvalHook = SeqNextvalHook; 
+	PreviousSeqNextvalHook = SeqNextvalHook;
 	SeqNextvalHook = MtmSeqNextvalHook;
 }
 
@@ -5380,7 +5392,7 @@ static void MtmSeqNextvalHook(Oid seqid, int64 next)
 		MtmSeqPosition pos;
 		pos.seqid = seqid;
 		pos.next = next;
-		LogLogicalMessage("N", (char*)&pos, sizeof(pos), true);		
+		LogLogicalMessage("N", (char*)&pos, sizeof(pos), true);
 	}
 }
 
@@ -5567,4 +5579,3 @@ Datum mtm_referee_poll(PG_FUNCTION_ARGS)
 
 	PG_RETURN_INT64(recoveredNodeMask);
 }
-

multimaster.h

@@ -23,25 +23,25 @@
 #define MTM_ERRMSG(fmt,...)     errmsg(MTM_TAG fmt, ## __VA_ARGS__)
 
 #if DEBUG_LEVEL == 0
-#define MTM_LOG1(fmt, ...) elog(LOG, "[MTM] " fmt, ## __VA_ARGS__) 
-#define MTM_LOG2(fmt, ...) 
-#define MTM_LOG3(fmt, ...) 
-#define MTM_LOG4(fmt, ...) 
+#define MTM_LOG1(fmt, ...) elog(LOG, "[MTM] " fmt, ## __VA_ARGS__)
+#define MTM_LOG2(fmt, ...)
+#define MTM_LOG3(fmt, ...)
+#define MTM_LOG4(fmt, ...)
 #elif  DEBUG_LEVEL == 1
-#define MTM_LOG1(fmt, ...) fprintf(stderr, fmt "\n", ## __VA_ARGS__) 
-#define MTM_LOG2(fmt, ...) fprintf(stderr, fmt "\n", ## __VA_ARGS__) 
-#define MTM_LOG3(fmt, ...) 
-#define MTM_LOG4(fmt, ...) 
+#define MTM_LOG1(fmt, ...) fprintf(stderr, fmt "\n", ## __VA_ARGS__)
+#define MTM_LOG2(fmt, ...) fprintf(stderr, fmt "\n", ## __VA_ARGS__)
+#define MTM_LOG3(fmt, ...)
+#define MTM_LOG4(fmt, ...)
 #elif  DEBUG_LEVEL == 2
-#define MTM_LOG1(fmt, ...) fprintf(stderr, fmt "\n", ## __VA_ARGS__) 
-#define MTM_LOG2(fmt, ...) fprintf(stderr, fmt "\n", ## __VA_ARGS__) 
-#define MTM_LOG3(fmt, ...) fprintf(stderr, fmt "\n", ## __VA_ARGS__) 
-#define MTM_LOG4(fmt, ...) 
+#define MTM_LOG1(fmt, ...) fprintf(stderr, fmt "\n", ## __VA_ARGS__)
+#define MTM_LOG2(fmt, ...) fprintf(stderr, fmt "\n", ## __VA_ARGS__)
+#define MTM_LOG3(fmt, ...) fprintf(stderr, fmt "\n", ## __VA_ARGS__)
+#define MTM_LOG4(fmt, ...)
 #elif  DEBUG_LEVEL >= 3
-#define MTM_LOG1(fmt, ...) fprintf(stderr, fmt "\n", ## __VA_ARGS__) 
-#define MTM_LOG2(fmt, ...) fprintf(stderr, fmt "\n", ## __VA_ARGS__) 
-#define MTM_LOG3(fmt, ...) fprintf(stderr, fmt "\n", ## __VA_ARGS__) 
-#define MTM_LOG4(fmt, ...) fprintf(stderr, fmt "\n", ## __VA_ARGS__) 
+#define MTM_LOG1(fmt, ...) fprintf(stderr, fmt "\n", ## __VA_ARGS__)
+#define MTM_LOG2(fmt, ...) fprintf(stderr, fmt "\n", ## __VA_ARGS__)
+#define MTM_LOG3(fmt, ...) fprintf(stderr, fmt "\n", ## __VA_ARGS__)
+#define MTM_LOG4(fmt, ...) fprintf(stderr, fmt "\n", ## __VA_ARGS__)
 #endif
 
 #if MTM_TRACE == 0
@@ -98,7 +98,7 @@ typedef char pgid_t[MULTIMASTER_MAX_GID_SIZE];
 #define SELF_CONNECTIVITY_MASK  (Mtm->nodes[MtmNodeId-1].connectivityMask)
 
 typedef enum
-{ 
+{
 	PGLOGICAL_COMMIT,
 	PGLOGICAL_PREPARE,
 	PGLOGICAL_COMMIT_PREPARED,
@@ -107,7 +107,7 @@ typedef enum
 } PGLOGICAL_EVENT;
 
 /* Identifier of global transaction */
-typedef struct 
+typedef struct
 {
 	int node;          /* Zero based index of node initiating transaction */
 	TransactionId xid; /* Transaction ID at this node */
@@ -116,7 +116,7 @@ typedef struct
 #define EQUAL_GTID(x,y) ((x).node == (y).node && (x).xid == (y).xid)
 
 typedef enum
-{ 
+{
 	MSG_INVALID,
 	MSG_HANDSHAKE,
 	MSG_PREPARED,
@@ -153,12 +153,12 @@ typedef enum
 typedef struct
 {
 	MtmMessageCode code;   /* Message code: MSG_PREPARE, MSG_PRECOMMIT, MSG_COMMIT, MSG_ABORT,... */
-    int            node;   /* Sender node ID */	
+    int            node;   /* Sender node ID */
 	bool           lockReq;/* Whether sender node needs to lock cluster to let wal-sender caught-up and complete recovery */
 	bool           locked; /* Whether sender node is locked */
 	TransactionId  dxid;   /* Transaction ID at destination node */
-	TransactionId  sxid;   /* Transaction ID at sender node */  
-    XidStatus      status; /* Transaction status */	
+	TransactionId  sxid;   /* Transaction ID at sender node */
+    XidStatus      status; /* Transaction status */
 	csn_t          csn;    /* Local CSN in case of sending data from replica to master, global CSN master->replica */
 	csn_t          oldestSnapshot; /* Oldest snapshot used by active transactions at this node */
 	nodemask_t     disabledNodeMask; /* Bitmask of disabled nodes at the sender of message */
@@ -185,13 +185,13 @@ typedef struct MtmMessageQueue
 	struct MtmMessageQueue* next;
 } MtmMessageQueue;
 
-typedef struct 
+typedef struct
 {
 	MtmArbiterMessage hdr;
 	char connStr[MULTIMASTER_MAX_CONN_STR_SIZE];
 } MtmHandshakeMessage;
 
-typedef struct 
+typedef struct
 {
 	int used;
 	int size;
@@ -227,7 +227,7 @@ typedef struct
 	int         senderPid;
 	int         receiverPid;
 	lsn_t       flushPos;
-	csn_t       oldestSnapshot;        /* Oldest snapshot used by active transactions at this node */	
+	csn_t       oldestSnapshot;        /* Oldest snapshot used by active transactions at this node */
 	lsn_t       restartLSN;
 	RepOriginId originId;
 	int         timeline;
@@ -246,12 +246,12 @@ typedef struct MtmL2List
 typedef struct MtmTransState
 {
     TransactionId  xid;
-    XidStatus      status; 
+    XidStatus      status;
 	pgid_t         gid;                /* Global transaction ID (used for 2PC) */
 	GlobalTransactionId gtid;          /* Transaction id at coordinator */
     csn_t          csn;                /* commit serial number */
     csn_t          snapshot;           /* transaction snapshot, or INVALID_CSN for local transactions */
-	int            procno;             /* pgprocno of transaction coordinator waiting for responses from replicas, 
+	int            procno;             /* pgprocno of transaction coordinator waiting for responses from replicas,
 							              used to notify coordinator by arbiter */
 	int            nSubxids;           /* Number of subtransanctions */
     struct MtmTransState* next;        /* Next element in L1 list of all finished transaction present in xid2state hash */
@@ -293,7 +293,7 @@ typedef struct
 	nodemask_t pglogicalSenderMask;    /* bitmask of started pglogic senders */
 	nodemask_t currentLockNodeMask;    /* Mask of nodes IDs which are locking the cluster */
 	nodemask_t inducedLockNodeMask;    /* Mask of node IDs which requested cluster-wide lock */
-	nodemask_t originLockNodeMask;     /* Mask of node IDs which WAL-senders are locking the cluster. 
+	nodemask_t originLockNodeMask;     /* Mask of node IDs which WAL-senders are locking the cluster.
 										* MtmNodeId bit is used by recovered node to complete recovery and by MtmLockCluster method */
 	nodemask_t reconnectMask; 	       /* Mask of nodes connection to which has to be reestablished by sender */
 	int        lastLockHolder;         /* PID of process last obtaining the node lock */
@@ -319,13 +319,13 @@ typedef struct
     MtmTransState** transListTail;     /* Tail of L1 list of all finished transactions, used to append new elements.
 								  		  This list is expected to be in CSN ascending order, by strict order may be violated */
 	MtmL2List activeTransList;         /* List of active transactions */
-	ulong64 transCount;                /* Counter of transactions performed by this node */	
+	ulong64 transCount;                /* Counter of transactions performed by this node */
 	ulong64 gcCount;                   /* Number of global transactions performed since last GC */
 	MtmMessageQueue* sendQueue;        /* Messages to be sent by arbiter sender */
 	MtmMessageQueue* freeQueue;        /* Free messages */
 	lsn_t recoveredLSN;           /* LSN at the moment of recovery completion */
 	BgwPool pool;                      /* Pool of background workers for applying logical replication patches */
-	MtmNodeInfo nodes[1];              /* [Mtm->nAllNodes]: per-node data */ 
+	MtmNodeInfo nodes[1];              /* [Mtm->nAllNodes]: per-node data */
 } MtmState;
 
 typedef struct MtmFlushPosition
@@ -342,7 +342,7 @@ typedef struct MtmSeqPosition
 	Oid   seqid;
 	int64 next;
 } MtmSeqPosition;
- 
+
 #define MtmIsCoordinator(ts) (ts->gtid.node == MtmNodeId)
 
 extern char const* const MtmNodeStatusMnem[];
@@ -394,14 +394,15 @@ extern void  MtmAdjustSubtransactions(MtmTransState* ts);
 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  MtmStopNode(int nodeId, bool dropSlot);
 extern void  MtmReconnectNode(int nodeId);
 extern void  MtmRecoverNode(int nodeId);
 extern void  MtmOnNodeDisconnect(int nodeId);
 extern void  MtmOnNodeConnect(int nodeId);
 extern void  MtmWakeUpBackend(MtmTransState* ts);
-extern void  MtmSleep(timestamp_t interval); 
+extern void  MtmSleep(timestamp_t interval);
 extern void  MtmAbortTransaction(MtmTransState* ts);
 extern void  MtmSetCurrentTransactionGID(char const* gid);
 extern csn_t MtmGetTransactionCSN(TransactionId xid);