dmq integration: resolver fixes, extend TxFinish logging with keyword to grep for

Author: kelvich

commit.c

@@ -106,6 +106,7 @@ MtmTwoPhaseCommit(MtmCurrentTrans* x)
 	char	stream[DMQ_NAME_MAXLEN];
 	pgid_t  gid;
 
+	/* XXX: avoid Mtm->extension_created */
 	if (!x->isDistributed || !x->containsDML || !Mtm->extension_created)
 		return false;
 
@@ -134,6 +135,7 @@ MtmTwoPhaseCommit(MtmCurrentTrans* x)
 	xid = GetCurrentTransactionId();
 	sprintf(stream, "xid" XID_FMT, xid);
 	dmq_stream_subscribe(stream);
+	mtm_log(MtmTxTrace, "%s subscribed for %s", gid, stream);
 	x->xid = xid;
 
 	MtmGenerateGid(gid, xid);
@@ -144,7 +146,7 @@ MtmTwoPhaseCommit(MtmCurrentTrans* x)
 	ret = PrepareTransactionBlock(gid);
 	if (!ret)
 	{
-		mtm_log(MtmTxFinish, "%s prepared", gid);
+		mtm_log(MtmTxFinish, "TXFINISH: %s prepared", gid);
 		mtm_log(WARNING, "Failed to prepare transaction %s", gid);
 		return false;
 	}
@@ -155,20 +157,21 @@ MtmTwoPhaseCommit(MtmCurrentTrans* x)
 	{
 		dmq_stream_unsubscribe(stream);
 		FinishPreparedTransaction(gid, false, false);
-		mtm_log(MtmTxFinish, "%s aborted", gid);
+		mtm_log(MtmTxFinish, "TXFINISH: %s aborted", gid);
 		mtm_log(ERROR, "Failed to prepare transaction %s at node %d",
 						gid, failed_at);
 	}
 
 	SetPreparedTransactionState(gid, MULTIMASTER_PRECOMMITTED);
-	mtm_log(MtmTxFinish, "%s precommittted", gid);
+	mtm_log(MtmTxFinish, "TXFINISH: %s precommittted", gid);
 	GatherPrecommits(x, participantsMask);
 
 	StartTransactionCommand();
 	FinishPreparedTransaction(gid, true, false);
-	mtm_log(MtmTxFinish, "%s committted", gid);
+	mtm_log(MtmTxFinish, "TXFINISH: %s committted", gid);
 
 	dmq_stream_unsubscribe(stream);
+	mtm_log(MtmTxTrace, "%s unsubscribed for %s", gid, stream);
 
 	return true;
 }

multimaster--1.0.sql

@@ -1,14 +1,14 @@
 -- complain if script is sourced in psql, rather than via CREATE EXTENSION
 \echo Use "CREATE EXTENSION multimaster" to load this file. \quit
 
--- -- check that multimaster shared library is really loaded
--- DO $$
--- BEGIN
---     IF strpos(current_setting('shared_preload_libraries'), 'multimaster') = 0 THEN
---         RAISE EXCEPTION 'Multimaster must be loaded via shared_preload_libraries. Refusing to proceed.';
---     END IF;
--- END
--- $$;
+-- check that multimaster shared library is really loaded
+DO $$
+BEGIN
+    IF strpos(current_setting('shared_preload_libraries'), 'multimaster') = 0 THEN
+        RAISE EXCEPTION 'Multimaster must be loaded via shared_preload_libraries. Refusing to proceed.';
+    END IF;
+END
+$$;
 
 
 CREATE FUNCTION mtm.dmq_receiver_loop(sender_name text) RETURNS void

pglogical_apply.c

@@ -59,6 +59,7 @@
 #include "pglogical_relid_map.h"
 #include "spill.h"
 #include "state.h"
+#include "logger.h"
 
 typedef struct TupleData
 {
@@ -835,6 +836,7 @@ process_remote_commit(StringInfo in, GlobalTransactionId *current_gtid)
 			} else {
 				SetPreparedTransactionState(gid, MULTIMASTER_PRECOMMITTED);
 			}
+			mtm_log(MtmTxFinish, "TXFINISH: %s precommittted", gid);
 
 			// MtmPrecommitTransaction(gid);
 
@@ -876,6 +878,7 @@ process_remote_commit(StringInfo in, GlobalTransactionId *current_gtid)
 
 			/* PREPARE itself */
 			res = PrepareTransactionBlock(gid);
+			mtm_log(MtmTxFinish, "TXFINISH: %s prepared", gid);
 			CommitTransactionCommand();
 
 			MtmDeadlockDetectorRemoveXact(xid);
@@ -944,6 +947,7 @@ process_remote_commit(StringInfo in, GlobalTransactionId *current_gtid)
 			MtmSetCurrentTransactionGID(gid, origin_node);
 			TXFINISH("%s COMMIT, PGLOGICAL_COMMIT_PREPARED csn=%lld", gid, csn);
 			FinishPreparedTransaction(gid, true, false);
+			mtm_log(MtmTxFinish, "TXFINISH: %s committed", gid);
 			MTM_LOG2("Distributed transaction %s is committed", gid);
 			CommitTransactionCommand();
 			Assert(!MtmTransIsActive());
@@ -959,6 +963,7 @@ process_remote_commit(StringInfo in, GlobalTransactionId *current_gtid)
 			StartTransactionCommand();
 			FinishPreparedTransaction(gid, false, true);
 			CommitTransactionCommand();
+			mtm_log(MtmTxFinish, "TXFINISH: %s aborted", gid);
 			break;
 		}
 		default:

resolver.c

@@ -255,19 +255,21 @@ resolve_tx(const char *gid, int node_id, MtmTxState state)
 
 	tx->state[node_id-1] = state;
 
+	/* XXX: missing ok because we call this concurrently with logrep recovery */
+
 	if (exists(tx, MtmTxAborted | MtmTxNotFound))
 	{
-		FinishPreparedTransaction(gid, false, false);
-		mtm_log(ResolverTxFinish, "%s aborted", gid);
+		FinishPreparedTransaction(gid, false, true);
+		mtm_log(ResolverTxFinish, "TXFINISH: %s aborted", gid);
 		hash_search(gid2tx, gid, HASH_REMOVE, &found);
 		Assert(found);
 		return;
 	}
 
 	if (exists(tx, MtmTxCommited))
 	{
-		FinishPreparedTransaction(gid, true, false);
-		mtm_log(ResolverTxFinish, "%s committed", gid);
+		FinishPreparedTransaction(gid, true, true);
+		mtm_log(ResolverTxFinish, "TXFINISH: %s committed", gid);
 		hash_search(gid2tx, gid, HASH_REMOVE, &found);
 		Assert(found);
 		return;
@@ -279,8 +281,8 @@ resolve_tx(const char *gid, int node_id, MtmTxState state)
 		// XXX: do that through PreCommit
 		// SetPreparedTransactionState(gid, MULTIMASTER_PRECOMMITTED);
 		// tx->state[MtmNodeId-1] = MtmTxPreCommited;
-		FinishPreparedTransaction(gid, true, false);
-		mtm_log(ResolverTxFinish, "%s committed", gid);
+		FinishPreparedTransaction(gid, true, true);
+		mtm_log(ResolverTxFinish, "TXFINISH: %s committed", gid);
 		hash_search(gid2tx, gid, HASH_REMOVE, &found);
 		Assert(found);
 		return;
@@ -291,8 +293,8 @@ resolve_tx(const char *gid, int node_id, MtmTxState state)
 		// XXX: do that through PreAbort
 		// SetPreparedTransactionState(gid, MULTIMASTER_PREABORTED);
 		// tx->state[MtmNodeId-1] = MtmTxPreAborted;
-		FinishPreparedTransaction(gid, false, false);
-		mtm_log(ResolverTxFinish, "%s aborted", gid);
+		FinishPreparedTransaction(gid, false, true);
+		mtm_log(ResolverTxFinish, "TXFINISH: %s aborted", gid);
 		hash_search(gid2tx, gid, HASH_REMOVE, &found);
 		Assert(found);
 		return;