Fix unreleased lock in GetPreparedTransactionState

Author: knizhnik

contrib/mmts/multimaster.c

@@ -1778,6 +1778,10 @@ void MtmRecoveryCompleted(void)
 	MTM_LOG1("Recovery of node %d is completed, disabled mask=%llx, connectivity mask=%llx, endLSN=%lx, live nodes=%d",
 			 MtmNodeId, (long long) Mtm->disabledNodeMask, 
 			 (long long)SELF_CONNECTIVITY_MASK, GetXLogInsertRecPtr(), Mtm->nLiveNodes);
+	if (Mtm->nAllNodes >= 3) { 
+		elog(WARNING, "restartLSNs at the end of recovery: {%lx, %lx, %lx}", 
+			 Mtm->nodes[0].restartLSN, Mtm->nodes[1].restartLSN, Mtm->nodes[2].restartLSN);
+	}
 	MtmLock(LW_EXCLUSIVE);
 	Mtm->recoverySlot = 0;
 	Mtm->recoveredLSN = GetXLogInsertRecPtr();
@@ -3249,7 +3253,12 @@ MtmReplicationMode MtmGetReplicationMode(int nodeId, sig_atomic_t volatile* shut
 				|| Mtm->recoverySlot == nodeId) 
 			{ 
 				/* Choose for recovery first available slot or slot of donor node (if any) */
-				elog(WARNING, "Process %d starts recovery from node %d", MyProcPid, nodeId);
+				if (Mtm->nAllNodes >= 3) { 
+					elog(WARNING, "Process %d starts recovery from node %d restartLSNs={%lx, %lx, %lx}", 
+						 MyProcPid, nodeId, Mtm->nodes[0].restartLSN, Mtm->nodes[1].restartLSN, Mtm->nodes[2].restartLSN);
+				} else {
+					elog(WARNING, "Process %d starts recovery from node %d", MyProcPid, nodeId);
+				}
 				Mtm->recoverySlot = nodeId;
 				Mtm->nReceivers = 0;
 				Mtm->nSenders = 0;
@@ -3388,7 +3397,7 @@ MtmReplicationStartupHook(struct PGLogicalStartupHookArgs* args)
 				sscanf(strVal(elem->arg), "%lx", &recoveredLSN);
 				MTM_LOG1("Recovered position of node %d is %lx", MtmReplicationNodeId, recoveredLSN); 
 				if (Mtm->nodes[MtmReplicationNodeId-1].restartLSN < recoveredLSN) { 
-					MTM_LOG2("[restartlsn] node %d: %lx -> %lx (MtmReplicationStartupHook)", MtmReplicationNodeId, Mtm->nodes[MtmReplicationNodeId-1].restartLSN, recoveredLSN);
+					MTM_LOG1("Advance restartLSN for node %d from %lx to %lx (MtmReplicationStartupHook)", MtmReplicationNodeId, Mtm->nodes[MtmReplicationNodeId-1].restartLSN, recoveredLSN);
 					Assert(Mtm->nodes[MtmReplicationNodeId-1].restartLSN == InvalidXLogRecPtr 
 						   || recoveredLSN < Mtm->nodes[MtmReplicationNodeId-1].restartLSN + MtmMaxRecoveryLag);
 					Mtm->nodes[MtmReplicationNodeId-1].restartLSN = recoveredLSN;
@@ -3592,9 +3601,13 @@ bool MtmFilterTransaction(char* record, int size)
 	origin_node = pq_getmsgbyte(&s);
 	origin_lsn = pq_getmsgint64(&s);
 
-	Assert(replication_node == MtmReplicationNodeId && 
-		   origin_node != 0 &&
-		   (Mtm->status == MTM_RECOVERY || origin_node == replication_node));	
+	Assert(replication_node == MtmReplicationNodeId);
+	if (!(origin_node != 0 &&
+		  (Mtm->status == MTM_RECOVERY || origin_node == replication_node)))
+	{
+		elog(WARNING, "Receive redirected commit event %d from node %d origin node %d origin LSN %lx in %s mode",
+			 event, replication_node, origin_node, origin_lsn, MtmNodeStatusMnem[Mtm->status]);
+	}
 
     switch (event)
     {
@@ -3621,8 +3634,8 @@ bool MtmFilterTransaction(char* record, int size)
 	}
 
 	if (duplicate) {
-		MTM_LOG1("Ignore transaction %s from node %d event=%x because our LSN position %lx for origin node %d is greater or equal than LSN %lx of this transaction (end_lsn=%lx, origin_lsn=%lx)", 
-				 gid, replication_node, event, Mtm->nodes[origin_node-1].restartLSN, origin_node, restart_lsn, end_lsn, origin_lsn);
+		MTM_LOG1("Ignore transaction %s from node %d event=%x because our LSN position %lx for origin node %d is greater or equal than LSN %lx of this transaction (end_lsn=%lx, origin_lsn=%lx) mode %s", 
+				 gid, replication_node, event, Mtm->nodes[origin_node-1].restartLSN, origin_node, restart_lsn, end_lsn, origin_lsn, MtmNodeStatusMnem[Mtm->status]);
 	} else {
 		MTM_LOG2("Apply transaction %s from node %d lsn %lx, event=%x, origin node %d, original lsn=%lx, current lsn=%lx", 
 				 gid, replication_node, end_lsn, event, origin_node, origin_lsn, restart_lsn);

contrib/mmts/pglogical_receiver.c

@@ -342,11 +342,11 @@ pglogical_receiver_main(Datum main_arg)
 				 * Them are either empty, either new node is synchronized using base_backup.
 				 * So we assume that LSNs are the same for local and remote node
 				 */
-				originStartPos = Mtm->status == MTM_RECOVERY && Mtm->donorNodeId == nodeId ? GetXLogInsertRecPtr() : InvalidXLogRecPtr;
+				originStartPos = (Mtm->status == MTM_RECOVERY && Mtm->donorNodeId == nodeId) ? GetXLogInsertRecPtr() : InvalidXLogRecPtr;
 				MTM_LOG1("Start logical receiver at position %lx from node %d", originStartPos, nodeId);
 			} else { 
 				if (Mtm->nodes[nodeId-1].restartLSN < originStartPos) { 
-					MTM_LOG2("[restartlsn] node %d: %lx -> %lx (pglogical_receiver_mains)", nodeId, Mtm->nodes[nodeId-1].restartLSN, originStartPos);
+					MTM_LOG1("Advance restartLSN for node %d: from %lx to %lx (pglogical_receiver_main)", nodeId, Mtm->nodes[nodeId-1].restartLSN, originStartPos);
 					Mtm->nodes[nodeId-1].restartLSN = originStartPos;
 				}
 				MTM_LOG1("Restart logical receiver at position %lx with origin=%d from node %d", originStartPos, originId, nodeId);
@@ -545,16 +545,17 @@ pglogical_receiver_main(Datum main_arg)
 					}
 					if (stmt[0] == 'Z' || (stmt[0] == 'M' && (stmt[1] == 'L' || stmt[1] == 'A' || stmt[1] == 'C'))) {
 						MTM_LOG3("Process '%c' message from %d", stmt[1], nodeId);
-						if ( stmt[1] == 'C') { /* concurrent DDL */
+						if (stmt[0] == 'M' && stmt[1] == 'C') { /* concurrent DDL should be executed by parallel workers */
 							MtmExecute(stmt, rc - hdr_len);
 						} else {
-							MtmExecutor(stmt, rc - hdr_len);
+							MtmExecutor(stmt, rc - hdr_len); /* all other messages can be processed by receiver itself */
 						}
 					} else { 
 						ByteBufferAppend(&buf, stmt, rc - hdr_len);
 						if (stmt[0] == 'C') /* commit */
 						{
-							if (!MtmFilterTransaction(stmt, rc - hdr_len)) { 
+							if (!MtmFilterTransaction(stmt, rc - hdr_len)) 
+							{ 
 								if (spill_file >= 0) { 
 									ByteBufferAppend(&buf, ")", 1);
 									pq_sendbyte(&spill_info, '(');
@@ -574,6 +575,7 @@ pglogical_receiver_main(Datum main_arg)
 											elog(WARNING, "Commit of prepared transaction takes %ld usec, flags=%x", stop - start, stmt[1]);
 										}
 									} else {
+										Assert(stmt[1] == PGLOGICAL_PREPARE || stmt[1] == PGLOGICAL_COMMIT); /* all other commits should be applied in place */
 										MtmExecute(buf.data, buf.used);
 									}
 								}

src/backend/access/transam/twophase.c

@@ -475,6 +475,7 @@ MarkAsPreparing(TransactionId xid, const char *gid,
 	{
 		if (strcmp(gxact->gid, gid) == 0)
 		{
+			LWLockRelease(TwoPhaseStateLock);
 			ereport(ERROR,
 					(errcode(ERRCODE_DUPLICATE_OBJECT),
 					 errmsg("transaction identifier \"%s\" is already in use",
@@ -484,11 +485,14 @@ MarkAsPreparing(TransactionId xid, const char *gid,
 
 	/* Get a free gxact from the freelist */
 	if (TwoPhaseState->freeGXacts == NULL)
+	{
+		LWLockRelease(TwoPhaseStateLock);
 		ereport(ERROR,
 				(errcode(ERRCODE_OUT_OF_MEMORY),
 				 errmsg("maximum number of prepared transactions reached"),
 				 errhint("Increase max_prepared_transactions (currently %d).",
 						 max_prepared_xacts)));
+	}
 	gxact = TwoPhaseState->freeGXacts;
 	TwoPhaseState->freeGXacts = gxact->next;
 
@@ -793,6 +797,7 @@ bool GetPreparedTransactionState(char const* gid, char* state)
 {
 	int i;
 	GlobalTransaction gxact;
+	bool result = false;
 
 	LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
 	i = string_hash(gid, 0)  % max_prepared_xacts;
@@ -801,11 +806,12 @@ bool GetPreparedTransactionState(char const* gid, char* state)
 		if (strcmp(gxact->gid, gid) == 0)
 		{
 			strcpy(state, gxact->state_3pc);
-			return true;
+			result = true;
+			break;
 		}
 	}
 	LWLockRelease(TwoPhaseStateLock);
-	return false;
+	return result;
 }
 
 
@@ -845,6 +851,9 @@ void SetPreparedTransactionState(char const* gid, char const* state)
 	START_CRIT_SECTION();
 	MyPgXact->delayChkpt = true;
 
+	hdr->xl_origin.origin_lsn = replorigin_session_origin_lsn;
+	hdr->xl_origin.origin_timestamp = replorigin_session_origin_timestamp;
+
 	XLogBeginInsert();
 	XLogRegisterData(buf, hdr->total_len - sizeof(pg_crc32c));
 	XLogIncludeOrigin();

src/bin/pg_xlogdump/pg_xlogdump.c

@@ -18,10 +18,12 @@
 #include "access/xlogreader.h"
 #include "access/xlogrecord.h"
 #include "access/xlog_internal.h"
+#include "access/xact.h"
 #include "access/transam.h"
 #include "common/fe_memutils.h"
 #include "getopt_long.h"
 #include "rmgrdesc.h"
+#include "replication/origin.h"
 
 
 static const char *progname;
@@ -42,6 +44,7 @@ typedef struct XLogDumpConfig
 	int			stop_after_records;
 	int			already_displayed_records;
 	bool		follow;
+	bool		dump_origin;
 	bool		stats;
 	bool		stats_per_record;
 
@@ -439,6 +442,35 @@ XLogDumpDisplayRecord(XLogDumpConfig *config, XLogReaderState *record)
 		   XLogRecGetXid(record),
 		   (uint32) (record->ReadRecPtr >> 32), (uint32) record->ReadRecPtr,
 		   (uint32) (xl_prev >> 32), (uint32) xl_prev);
+
+	if (config->dump_origin && XLogRecGetOrigin(record) != InvalidRepOriginId) { 
+		switch (info & XLOG_XACT_OPMASK) { 
+		case XLOG_XACT_COMMIT:
+		case XLOG_XACT_COMMIT_PREPARED:
+			{
+				xl_xact_commit *xlrec;
+				xl_xact_parsed_commit parsed;
+
+				xlrec = (xl_xact_commit *) XLogRecGetData(record);
+				ParseCommitRecord(info, xlrec, &parsed);
+				printf("origin_id=%d, origin_lsn=%llx, ", XLogRecGetOrigin(record), (long long)parsed.origin_lsn);
+				break;
+			}
+		case XLOG_XACT_ABORT:
+		case XLOG_XACT_ABORT_PREPARED:
+			{
+				xl_xact_abort *xlrec;
+				xl_xact_parsed_abort parsed;
+
+				xlrec = (xl_xact_abort *) XLogRecGetData(record);
+				ParseAbortRecord(info, xlrec, &parsed);
+
+				printf("origin_id=%d, origin_lsn=%llx, ", XLogRecGetOrigin(record), (long long)parsed.origin_lsn);
+				break;
+			}
+		}
+	}		
+
 	printf("desc: %s ", id);
 
 	/* the desc routine will printf the description directly to stdout */
@@ -678,6 +710,7 @@ usage(void)
 	printf("  -b, --bkp-details      output detailed information about backup blocks\n");
 	printf("  -e, --end=RECPTR       stop reading at log position RECPTR\n");
 	printf("  -f, --follow           keep retrying after reaching end of WAL\n");
+	printf("  -o, --origin           dump origins\n");
 	printf("  -n, --limit=N          number of records to display\n");
 	printf("  -p, --path=PATH        directory in which to find log segment files\n");
 	printf("                         (default: ./pg_xlog)\n");
@@ -710,6 +743,7 @@ main(int argc, char **argv)
 		{"bkp-details", no_argument, NULL, 'b'},
 		{"end", required_argument, NULL, 'e'},
 		{"follow", no_argument, NULL, 'f'},
+		{"origin", no_argument, NULL, 'o'},
 		{"help", no_argument, NULL, '?'},
 		{"limit", required_argument, NULL, 'n'},
 		{"path", required_argument, NULL, 'p'},
@@ -740,6 +774,7 @@ main(int argc, char **argv)
 	config.stop_after_records = -1;
 	config.already_displayed_records = 0;
 	config.follow = false;
+	config.dump_origin = false;
 	config.filter_by_rmgr = -1;
 	config.filter_by_xid = InvalidTransactionId;
 	config.filter_by_xid_enabled = false;
@@ -752,7 +787,7 @@ main(int argc, char **argv)
 		goto bad_argument;
 	}
 
-	while ((option = getopt_long(argc, argv, "be:?fn:p:r:s:t:Vx:z",
+	while ((option = getopt_long(argc, argv, "be:?f?on:p:r:s:t:Vx:z",
 								 long_options, &optindex)) != -1)
 	{
 		switch (option)
@@ -772,6 +807,9 @@ main(int argc, char **argv)
 			case 'f':
 				config.follow = true;
 				break;
+		    case 'o':
+			    config.dump_origin = true;
+				break;
 			case '?':
 				usage();
 				exit(EXIT_SUCCESS);