origin patch

Author: knizhnik

multimaster.c

@@ -80,8 +80,7 @@ typedef struct {
 
 typedef enum 
 {
-	MTM_STATE_LOCK_ID,
-	N_LOCKS
+	MTM_STATE_LOCK_ID
 } MtmLockIds;
 
 #define MTM_SHMEM_SIZE (64*1024*1024)
@@ -208,6 +207,17 @@ void MtmUnlock(void)
 #endif
 }
 
+void MtmLockNode(int nodeId)
+{
+	Assert(nodeId > 0 && nodeId <= MtmNodes);
+	LWLockAcquire((LWLockId)&dtm->locks[nodeId], LW_EXCLUSIVE);
+}
+
+void MtmUnlockNode(int nodeId)
+{
+	Assert(nodeId > 0 && nodeId <= MtmNodes);
+	LWLockRelease((LWLockId)&dtm->locks[nodeId]);	
+}
 
 /*
  * -------------------------------------------
@@ -1370,15 +1380,6 @@ _PG_init(void)
 		NULL
 	);
 
-
-	/*
-	 * Request additional shared resources.  (These are no-ops if we're not in
-	 * the postmaster process.)  We'll allocate or attach to the shared
-	 * resources in mtm_shmem_startup().
-	 */
-	RequestAddinShmemSpace(MTM_SHMEM_SIZE + MtmQueueSize);
-	RequestNamedLWLockTranche(MULTIMASTER_NAME, N_LOCKS);
-
     MtmNodes = MtmStartReceivers(MtmConnStrs, MtmNodeId);
     if (MtmNodes < 2) { 
         elog(ERROR, "Multimaster should have at least two nodes");
@@ -1387,6 +1388,14 @@ _PG_init(void)
         elog(ERROR, "Multimaster with mor than %d nodes is not currently supported", MAX_NODES);
 	}		
 
+	/*
+	 * Request additional shared resources.  (These are no-ops if we're not in
+	 * the postmaster process.)  We'll allocate or attach to the shared
+	 * resources in mtm_shmem_startup().
+	 */
+	RequestAddinShmemSpace(MTM_SHMEM_SIZE + MtmQueueSize);
+	RequestNamedLWLockTranche(MULTIMASTER_NAME, 1 + MtmNodes);
+
     BgwPoolStart(MtmWorkers, MtmPoolConstructor);
 
 	MtmArbiterInitialize();

multimaster.h

@@ -154,6 +154,8 @@ extern void  MtmSendNotificationMessage(MtmTransState* ts, MtmMessageCode cmd);
 extern void  MtmAdjustSubtransactions(MtmTransState* ts);
 extern void  MtmLock(LWLockMode mode);
 extern void  MtmUnlock(void);
+extern void  MtmLockNode(int nodeId);
+extern void  MtmUnlockNode(int nodeId);
 extern void  MtmDropNode(int nodeId, bool dropSlot);
 extern void  MtmRecoverNode(int nodeId);
 extern void  MtmOnNodeDisconnect(int nodeId);

pglogical_apply.c

@@ -73,6 +73,8 @@ static void process_remote_insert(StringInfo s, Relation rel);
 static void process_remote_update(StringInfo s, Relation rel);
 static void process_remote_delete(StringInfo s, Relation rel);
 
+static int MtmReplicationNode;
+
 /*
  * Search the index 'idxrel' for a tuple identified by 'skey' in 'rel'.
  *
@@ -465,41 +467,40 @@ read_rel(StringInfo s, LOCKMODE mode)
 }
 
 static void
-MtmBeginSession(int nodeId)
+MtmBeginSession(void)
 {
-#if 0
 	char slot_name[MULTIMASTER_MAX_SLOT_NAME_SIZE];
-	sprintf(slot_name, MULTIMASTER_SLOT_PATTERN, nodeId);
+	MtmLockNode(MtmReplicationNode);
+	sprintf(slot_name, MULTIMASTER_SLOT_PATTERN, MtmReplicationNode);
 	Assert(replorigin_session_origin == InvalidRepOriginId);
 	replorigin_session_origin = replorigin_by_name(slot_name, false); 
-	MTM_INFO("%d: Begin setup replorigin session: %d\n", MyProcPid, replorigin_session_origin);
+	MTM_TRACE("%d: Begin setup replorigin session: %d\n", MyProcPid, replorigin_session_origin);
 	replorigin_session_setup(replorigin_session_origin);
-	MTM_INFO("%d: End setup replorigin session: %d\n", MyProcPid, replorigin_session_origin);
-#endif
+	MTM_TRACE("%d: End setup replorigin session: %d\n", MyProcPid, replorigin_session_origin);
 }
 
 static void 
 MtmEndSession(void)
 {
 	if (replorigin_session_origin != InvalidRepOriginId) { 
-		MTM_INFO("%d: Begin reset replorigin session: %d\n", MyProcPid, replorigin_session_origin);
+		MTM_TRACE("%d: Begin reset replorigin session: %d\n", MyProcPid, replorigin_session_origin);
 		replorigin_session_origin = InvalidRepOriginId;
 		replorigin_session_reset();
-		MTM_INFO("%d: End reset replorigin session: %d\n", MyProcPid, replorigin_session_origin);
+		MtmUnlockNode(MtmReplicationNode);
+		MTM_TRACE("%d: End reset replorigin session: %d\n", MyProcPid, replorigin_session_origin);
 	}
 }
 
 static void
 process_remote_commit(StringInfo in)
 {
 	uint8 		flags;
-	uint8 		nodeId;
 	csn_t       csn;
 	const char *gid = NULL;	
 
 	/* read flags */
 	flags = pq_getmsgbyte(in);
-	nodeId = pq_getmsgbyte(in);
+	MtmReplicationNode = pq_getmsgbyte(in);
 
 	/* read fields */
 	replorigin_session_origin_lsn = pq_getmsgint64(in); /* commit_lsn */
@@ -515,7 +516,7 @@ process_remote_commit(StringInfo in)
 			MTM_TRACE("%d: PGLOGICAL_COMMIT commit\n", MyProcPid);
 			if (IsTransactionState()) {
 				Assert(TransactionIdIsValid(MtmGetCurrentTransactionId()));
-				MtmBeginSession(nodeId);
+				MtmBeginSession();
 				CommitTransactionCommand();
 			}
 			break;
@@ -529,7 +530,8 @@ process_remote_commit(StringInfo in)
 			BeginTransactionBlock();
 			CommitTransactionCommand();
 			StartTransactionCommand();
-			MtmBeginSession(nodeId);
+			
+			MtmBeginSession();
 			/* PREPARE itself */
 			MtmSetCurrentTransactionGID(gid);
 			PrepareTransactionBlock(gid);
@@ -543,7 +545,7 @@ process_remote_commit(StringInfo in)
 			gid = pq_getmsgstring(in);
 			MTM_TRACE("%d: PGLOGICAL_COMMIT_PREPARED commit: csn=%ld, gid=%s\n", MyProcPid, csn, gid);
 			StartTransactionCommand();
-			MtmBeginSession(nodeId);
+			MtmBeginSession();
 			MtmSetCurrentTransactionCSN(csn);
 			MtmSetCurrentTransactionGID(gid);
 			FinishPreparedTransaction(gid, true);