Fix memory leak in MMTS

Author: knizhnik

decoder_raw.c

@@ -108,11 +108,11 @@ decoder_raw_begin_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn)
     Assert(lastXid != txn->xid);
     lastXid = txn->xid;
     if (MMIsLocalTransaction(txn->xid)) {
-        MTM_INFO("Skip local transaction %u\n", txn->xid);
+        MTM_TRACE("Skip local transaction %u\n", txn->xid);
         data->isLocal = true;
     } else { 
         OutputPluginPrepareWrite(ctx, true);
-        MTM_INFO("Send transaction %u to replica\n", txn->xid);
+        MTM_TRACE("Send transaction %u to replica\n", txn->xid);
         appendStringInfo(ctx->out, "BEGIN %u;", txn->xid);
         OutputPluginWrite(ctx, true);
         data->isLocal = false;
@@ -126,12 +126,12 @@ decoder_raw_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 {
 	DecoderRawData *data = ctx->output_plugin_private;
     if (!data->isLocal) { 
-        MTM_INFO("Send commit of transaction %u to replica\n", txn->xid);
+        MTM_TRACE("Send commit of transaction %u to replica\n", txn->xid);
         OutputPluginPrepareWrite(ctx, true);
         appendStringInfoString(ctx->out, "COMMIT;");
         OutputPluginWrite(ctx, true);
     } else { 
-        MTM_INFO("Skip commit of transaction %u\n", txn->xid);
+        MTM_TRACE("Skip commit of transaction %u\n", txn->xid);
     }
 }
 
@@ -483,10 +483,10 @@ decoder_raw_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 
 	data = ctx->output_plugin_private;
     if (data->isLocal) { 
-        MTM_INFO("Skip action %d in transaction %u\n", change->action, txn->xid);
+        MTM_TRACE("Skip action %d in transaction %u\n", change->action, txn->xid);
         return;
     }
-    MTM_INFO("Send action %d in transaction %u to replica\n", change->action, txn->xid);
+    MTM_TRACE("Send action %d in transaction %u to replica\n", change->action, txn->xid);
 
  	/* Avoid leaking memory by using and resetting our own context */
 	old = MemoryContextSwitchTo(data->context);

multimaster.c

@@ -527,6 +527,9 @@ MtmResetTransaction(MtmCurrentTrans* x)
 	x->xid = InvalidTransactionId;
 	x->gtid.xid = InvalidTransactionId;
 	x->isDistributed = false;
+	x->isPrepared = false;
+	x->isPrepared = false;
+	x->status = TRANSACTION_STATUS_UNKNOWN;
 }
 
 static void 
@@ -625,16 +628,15 @@ static void
 MtmPostPrepareTransaction(MtmCurrentTrans* x)
 { 
 	MtmTransState* ts;
-	MtmTransMap* tm;
 
 	MtmLock(LW_EXCLUSIVE);
 	ts = hash_search(MtmXid2State, &x->xid, HASH_FIND, NULL);
 	Assert(ts != NULL);
-	tm = (MtmTransMap*)hash_search(MtmGid2State, x->gid, HASH_ENTER, NULL);
-	Assert(x->gid[0]);
-	tm->state = ts;
 
 	if (!MtmIsCoordinator(ts)) {
+		MtmTransMap* tm = (MtmTransMap*)hash_search(MtmGid2State, x->gid, HASH_ENTER, NULL);
+		Assert(x->gid[0]);
+		tm->state = ts;
 		MtmSendNotificationMessage(ts, MSG_READY); /* send notification to coordinator */
 		MtmUnlock();
 		MtmResetTransaction(x);
@@ -658,21 +660,23 @@ MtmAbortPreparedTransaction(MtmCurrentTrans* x)
 {
 	MtmTransMap* tm;
 
-	MtmLock(LW_EXCLUSIVE);
-	tm = (MtmTransMap*)hash_search(MtmGid2State, x->gid, HASH_REMOVE, NULL);
-	Assert(tm != NULL);
-	tm->state->status = TRANSACTION_STATUS_ABORTED;
-	MtmAdjustSubtransactions(tm->state);
-	MtmUnlock();
-	MtmResetTransaction(x);		
+	if (x->status != TRANSACTION_STATUS_ABORTED) { 
+		MtmLock(LW_EXCLUSIVE);
+		tm = (MtmTransMap*)hash_search(MtmGid2State, x->gid, HASH_REMOVE, NULL);
+		Assert(tm != NULL);
+		tm->state->status = TRANSACTION_STATUS_ABORTED;
+		MtmAdjustSubtransactions(tm->state);
+		MtmUnlock();
+		x->status = TRANSACTION_STATUS_ABORTED;
+	}
 }
 
 static void 
 MtmEndTransaction(MtmCurrentTrans* x, bool commit)
 {
 	MTM_TRACE("%d: End transaction %d, prepared=%d, replicated=%d, distributed=%d, gid=%s -> %s\n", 
 			  MyProcPid, x->xid, x->isPrepared, x->isReplicated, x->isDistributed, x->gid, commit ? "commit" : "abort");
-	if (x->isDistributed && (x->isPrepared || x->isReplicated)) {
+	if (x->status != TRANSACTION_STATUS_ABORTED && x->isDistributed && (x->isPrepared || x->isReplicated)) {
 		MtmTransState* ts = NULL;
 		MtmLock(LW_EXCLUSIVE);
 		if (x->isPrepared) { 
@@ -690,6 +694,10 @@ MtmEndTransaction(MtmCurrentTrans* x, bool commit)
 			if (commit) {
 				Assert(ts->status == TRANSACTION_STATUS_UNKNOWN);
 				ts->status = TRANSACTION_STATUS_COMMITTED;
+				if (x->csn > ts->csn) {
+					ts->csn = x->csn;
+					MtmSyncClock(ts->csn);
+				}
 			} else { 
 				ts->status = TRANSACTION_STATUS_ABORTED;
 			}
@@ -799,6 +807,26 @@ XidStatus MtmGetGlobalTransactionStatus(char const* gid)
 	return status;
 }
 
+void  MtmSetCurrentTransactionCSN(csn_t csn)
+{
+	MTM_TRACE("Set current transaction CSN %ld\n", csn);
+	dtmTx.csn = csn;
+	dtmTx.isDistributed = true;
+	dtmTx.isReplicated = true;
+}
+
+
+csn_t MtmGetTransactionCSN(TransactionId xid)
+{
+	MtmTransState* ts;
+	csn_t csn;
+	MtmLock(LW_SHARED);
+	ts = (MtmTransState*)hash_search(MtmXid2State, &xid, HASH_FIND, NULL);
+	Assert(ts != NULL);
+	csn = ts->csn;
+	MtmUnlock();
+	return csn;
+}
 
 /*
  * -------------------------------------------

multimaster.h

@@ -10,8 +10,8 @@
 #define MTM_INFO(fmt, ...) fprintf(stderr, fmt, ## __VA_ARGS__) 
 #define MTM_TRACE(fmt, ...) fprintf(stderr, fmt, ## __VA_ARGS__) 
 */
+#define MTM_INFO(fmt, ...) fprintf(stderr, fmt, ## __VA_ARGS__) 
 #define MTM_TRACE(fmt, ...) 
-#define MTM_TUPLE_TRACE(fmt, ...) 
 /* */
 
 #define MULTIMASTER_NAME                "multimaster"
@@ -162,6 +162,8 @@ extern MtmState* MtmGetState(void);
 extern timestamp_t MtmGetCurrentTime(void);
 extern void  MtmSleep(timestamp_t interval);
 extern void  MtmSetCurrentTransactionGID(char const* gid);
+extern csn_t MtmGetTransactionCSN(TransactionId xid);
+extern void  MtmSetCurrentTransactionCSN(csn_t csn);
 extern TransactionId MtmGetCurrentTransactionId(void);
 extern XidStatus MtmGetCurrentTransactionStatus(void);
 extern XidStatus MtmGetGlobalTransactionStatus(char const* gid);

pglogical_apply.c

@@ -465,20 +465,37 @@ read_rel(StringInfo s, LOCKMODE mode)
 }
 
 static void
-MtmSetCurrentSession(int nodeId)
+MtmBeginSession(int nodeId)
 {
+#if 0
 	char slot_name[MULTIMASTER_MAX_SLOT_NAME_SIZE];
 	sprintf(slot_name, MULTIMASTER_SLOT_PATTERN, nodeId);
+	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);
 	replorigin_session_setup(replorigin_session_origin);
+	MTM_INFO("%d: End setup replorigin session: %d\n", MyProcPid, replorigin_session_origin);
+#endif
+}
+
+static void 
+MtmEndSession(void)
+{
+	if (replorigin_session_origin != InvalidRepOriginId) { 
+		MTM_INFO("%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);
+	}
 }
 
 static void
 process_remote_commit(StringInfo in)
 {
-	uint8 			flags;
-	uint8 			nodeId;
-	const char	   *gid = NULL;
+	uint8 		flags;
+	uint8 		nodeId;
+	csn_t       csn;
+	const char *gid = NULL;	
 
 	/* read flags */
 	flags = pq_getmsgbyte(in);
@@ -489,14 +506,16 @@ process_remote_commit(StringInfo in)
 	pq_getmsgint64(in); /* end_lsn */
 	replorigin_session_origin_timestamp = pq_getmsgint64(in); /* commit_time */
 
+	Assert(replorigin_session_origin == InvalidRepOriginId);
+
 	switch(PGLOGICAL_XACT_EVENT(flags))
 	{
 		case PGLOGICAL_COMMIT:
 		{
 			MTM_TRACE("%d: PGLOGICAL_COMMIT commit\n", MyProcPid);
 			if (IsTransactionState()) {
 				Assert(TransactionIdIsValid(MtmGetCurrentTransactionId()));
-				MtmSetCurrentSession(nodeId);
+				MtmBeginSession(nodeId);
 				CommitTransactionCommand();
 			}
 			break;
@@ -510,7 +529,7 @@ process_remote_commit(StringInfo in)
 			BeginTransactionBlock();
 			CommitTransactionCommand();
 			StartTransactionCommand();
-			MtmSetCurrentSession(nodeId);
+			MtmBeginSession(nodeId);
 			/* PREPARE itself */
 			MtmSetCurrentTransactionGID(gid);
 			PrepareTransactionBlock(gid);
@@ -520,10 +539,12 @@ process_remote_commit(StringInfo in)
 		case PGLOGICAL_COMMIT_PREPARED:
 		{
 			Assert(!TransactionIdIsValid(MtmGetCurrentTransactionId()));
+			csn = pq_getmsgint64(in); 
 			gid = pq_getmsgstring(in);
 			MTM_TRACE("%d: PGLOGICAL_COMMIT_PREPARED commit: csn=%ld, gid=%s\n", MyProcPid, csn, gid);
 			StartTransactionCommand();
-			MtmSetCurrentSession(nodeId);
+			MtmBeginSession(nodeId);
+			MtmSetCurrentTransactionCSN(csn);
 			MtmSetCurrentTransactionGID(gid);
 			FinishPreparedTransaction(gid, true);
 			CommitTransactionCommand();
@@ -545,8 +566,7 @@ process_remote_commit(StringInfo in)
 		default:
 			Assert(false);
 	}
-	replorigin_session_reset();
-	replorigin_session_origin = InvalidRepOriginId;
+	MtmEndSession();
 }
 
 static void
@@ -859,10 +879,10 @@ void MtmExecutor(int id, void* work, size_t size)
 {
     StringInfoData s;
     Relation rel = NULL;
-    initStringInfo(&s);
     s.data = work;
     s.len = size;
     s.maxlen = -1;
+	s.cursor = 0;
 
     if (ApplyContext == NULL) {
         ApplyContext = AllocSetContextCreate(TopMemoryContext,
@@ -910,12 +930,10 @@ void MtmExecutor(int id, void* work, size_t size)
     }
     PG_CATCH();
     {
-		if (replorigin_session_origin != InvalidRepOriginId) { 
-			replorigin_session_reset();
-		}
 		EmitErrorReport();
         FlushErrorState();
 		MTM_TRACE("%d: REMOTE begin abort transaction %d\n", MyProcPid, MtmGetCurrentTransactionId());
+		MtmEndSession();
         AbortCurrentTransaction();
 		MTM_TRACE("%d: REMOTE end abort transaction %d\n", MyProcPid, MtmGetCurrentTransactionId());
     }

pglogical_proto.c

@@ -169,6 +169,9 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
     pq_sendint64(out, txn->end_lsn);
     pq_sendint64(out, txn->commit_time);
 
+	if (flags == PGLOGICAL_COMMIT_PREPARED) { 
+		pq_sendint64(out, MtmGetTransactionCSN(txn->xid));
+	}
     if (flags != PGLOGICAL_COMMIT) { 
     	pq_sendstring(out, txn->gid);
 	}

pglogical_receiver.c

@@ -214,6 +214,8 @@ pglogical_receiver_main(Datum main_arg)
 	XLogRecPtr originStartPos = 0;
 	RepOriginId originId;
 	char* originName;
+	/* Buffer for COPY data */
+	char	*copybuf = NULL;
 
 	/* Register functions for SIGTERM/SIGHUP management */
 	pqsignal(SIGHUP, receiver_raw_sighup);
@@ -314,8 +316,6 @@ pglogical_receiver_main(Datum main_arg)
 	while (!got_sigterm)
 	{
 		int rc, hdr_len;
-		/* Buffer for COPY data */
-		char	*copybuf = NULL;
 		/* Wait necessary amount of time */
 		rc = WaitLatch(&MyProc->procLatch,
 					   WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH,
@@ -347,13 +347,6 @@ pglogical_receiver_main(Datum main_arg)
 		}
 
 
-		/* Some cleanup */
-		if (copybuf != NULL)
-		{
-			PQfreemem(copybuf);
-			copybuf = NULL;
-		}
-
 		/*
 		 * Receive data.
 		 */
@@ -362,6 +355,13 @@ pglogical_receiver_main(Datum main_arg)
 			XLogRecPtr  walEnd;
             char* stmt;
 
+			/* Some cleanup */
+			if (copybuf != NULL)
+			{
+				PQfreemem(copybuf);
+				copybuf = NULL;
+			}
+
 			rc = PQgetCopyData(conn, &copybuf, 1);
 			if (rc <= 0) {
 				break;