Allow parallel apply at receiver only for txes that include receiver

node as a participant. Otherwise receiver may reorder prepares and
corresponding commits as coordinator doesn't wait for confirmation
after prepare from this receiver. As a consequence we don't need to
rely on Mtm->state in MtmExecute.

Author: kelvich

src/bgwpool.c

@@ -93,7 +93,7 @@ static void BgwPoolMainLoop(BgwPool* pool)
 			pool->lastPeakTime = 0;
         }
         SpinLockRelease(&pool->lock);
-        pool->executor(work, size);
+		pool->executor(work, size, NULL);
         pfree(work);
         SpinLockAcquire(&pool->lock);
         pool->active -= 1;
@@ -209,7 +209,7 @@ void BgwPoolExecute(BgwPool* pool, void* work, size_t size)
 		 * Size of work is larger than size of shared buffer: 
 		 * run it immediately
 		 */
-		pool->executor(work, size);
+		pool->executor(work, size, NULL);
 		return;
 	}
 

src/commit.c

@@ -150,6 +150,21 @@ MtmTwoPhaseCommit(MtmCurrentTrans* x)
 
 	x->xid = xid;
 
+	/*
+	 * This lock is taken for a quite a long period of time but normally
+	 * all callers lock it in shared mode, so it shouldn't be noticeable
+	 * performance-wise.
+	 *
+	 * It is only used during startup of WalSender(node_id) in recovered mode
+	 * to create a barrier after which all transactions doing our 3PC are
+	 * guaranted to have seen participantsMask with node_id enabled, so the
+	 * receiver can apply them in parallel and be sure that precommit will
+	 * not happens before node_id applies prepare.
+	 *
+	 * See also comments at the end of MtmReplicationStartupHook().
+	 */
+	LWLockAcquire(MtmCommitBarrier, LW_SHARED);
+
 	MtmLock(LW_SHARED);
 	participantsMask = (((nodemask_t)1 << Mtm->nAllNodes) - 1) &
 								  ~Mtm->disabledNodeMask &
@@ -187,6 +202,8 @@ MtmTwoPhaseCommit(MtmCurrentTrans* x)
 	mtm_log(MtmTxFinish, "TXFINISH: %s committed", gid);
 	GatherPrecommits(x, participantsMask, MSG_COMMITTED);
 
+	LWLockRelease(MtmCommitBarrier);
+
 	dmq_stream_unsubscribe(stream);
 	mtm_log(MtmTxTrace, "%s unsubscribed for %s", gid, stream);
 

src/dmq.c

@@ -856,6 +856,8 @@ _pq_getmessage_if_avalable(StringInfo s)
 		recv_bytes += rc;
 		Assert(recv_bytes >= read_bytes && recv_bytes <= DMQ_RECV_BUFFER);
 
+		mtm_log(DmqTraceIncoming, "dmq: got %d bytes", rc);
+
 		/*
 		 * Here we need to re-check for full message again, so the caller will know
 		 * whether he should wait for event on socket.
@@ -881,7 +883,7 @@ _pq_getbyte_if_available(unsigned char *c)
 	int rc;
 
 	/*
-	 * That is why we re-implementing this function: byte ccan be already in
+	 * That is why we re-implementing this function: byte can be already in
 	 * our recv buffer, so pqcomm version will miss it.
 	 */
 	if (recv_bytes > read_bytes)

src/include/bgwpool.h

@@ -4,9 +4,11 @@
 #include "storage/s_lock.h"
 #include "storage/spin.h"
 #include "storage/pg_sema.h"
-#include "bkb.h"
+#include "bkb.h" // XXX
 
-typedef void(*BgwPoolExecutor)(void* work, size_t size);
+#include "mm.h"
+
+typedef void(*BgwPoolExecutor)(void* work, size_t size, MtmReceiverContext *ctx);
 
 typedef long timestamp_t;
 

src/include/mm.h

@@ -45,6 +45,13 @@ typedef struct MtmSeqPosition
 	int64	next;
 } MtmSeqPosition;
 
+typedef struct
+{
+	int node_id;
+	bool is_recovery;
+	bool parallel_allowed;
+} MtmReceiverContext;
+
 /* XXX: drop that */
 typedef long long long64; /* we are not using int64 here because we want to use %lld format for this type */
 typedef unsigned long long ulong64; /* we are not using uint64 here because we want to use %lld format for this type */

src/include/multimaster.h

@@ -160,6 +160,8 @@ extern bool MtmMajorNode;
 extern bool MtmBackgroundWorker;
 extern char* MtmRefereeConnStr;
 
+extern LWLock *MtmCommitBarrier;
+
 extern void MtmXactCallback2(XactEvent event, void *arg);
 extern bool MtmIsUserTransaction(void);
 extern void MtmGenerateGid(char *gid, TransactionId xid);

src/include/pglogical_output.h

@@ -43,8 +43,9 @@ typedef struct HookFuncName
 
 typedef struct MtmDecoderPrivate
 {
-	int magic;
-	bool is_recovery;
+	int		magic; // XXX
+	bool	is_recovery;
+	int		node_id;
 } MtmDecoderPrivate;
 
 typedef struct PGLogicalOutputData

src/include/receiver.h

@@ -1,11 +1,16 @@
+#ifndef MTM_RECEIVER_H
+#define MTM_RECEIVER_H
+
 #include "multimaster.h"
 #include "mm.h"
 
 extern void MtmStartReceivers(void);
 extern void MtmStartReceiver(int nodeId, bool dynamic);
 
-extern void MtmExecutor(void* work, size_t size);
+extern void MtmExecutor(void* work, size_t size, MtmReceiverContext *rctx);
 extern void MtmUpdateLsnMapping(int node_id, lsn_t end_lsn);
 
 extern void MtmBeginSession(int nodeId);
 extern void MtmEndSession(int nodeId, bool unlock);
+
+#endif

src/multimaster.c

@@ -114,6 +114,8 @@ static TransactionManager MtmTM =
 	MtmResumeTransaction
 };
 
+LWLock *MtmCommitBarrier;
+
 bool  MtmDoReplication;
 char* MtmDatabaseName;
 char* MtmDatabaseUser;
@@ -530,7 +532,6 @@ static void MtmInitialize()
 		Mtm->status = MTM_DISABLED; //MTM_INITIALIZATION;
 		Mtm->recoverySlot = 0;
 		Mtm->locks = GetNamedLWLockTranche(MULTIMASTER_NAME);
-
 		Mtm->nAllNodes = MtmNodes;
 		Mtm->disabledNodeMask =  (((nodemask_t)1 << MtmNodes) - 1);
 		Mtm->clique = (((nodemask_t)1 << Mtm->nAllNodes) - 1); //0;
@@ -563,6 +564,8 @@ static void MtmInitialize()
 
 	RegisterXactCallback(MtmXactCallback2, NULL);
 
+	MtmCommitBarrier = &(GetNamedLWLockTranche(MULTIMASTER_NAME)[MtmMaxNodes*2+1].lock);
+
 	MtmDoReplication = true;
 	TM = &MtmTM;
 	LWLockRelease(AddinShmemInitLock);
@@ -1149,7 +1152,7 @@ _PG_init(void)
 	 * resources in mtm_shmem_startup().
 	 */
 	RequestAddinShmemSpace(MTM_SHMEM_SIZE + MtmMaxNodes*MtmQueueSize);
-	RequestNamedLWLockTranche(MULTIMASTER_NAME, 1 + MtmMaxNodes*2);
+	RequestNamedLWLockTranche(MULTIMASTER_NAME, 1 + MtmMaxNodes*2 + 1);
 
 	MtmMonitorInitialize();
 

src/pglogical_apply.c

@@ -85,7 +85,7 @@ static void UserTableUpdateOpenIndexes(EState *estate, TupleTableSlot *slot);
 static void UserTableUpdateIndexes(EState *estate, TupleTableSlot *slot);
 
 static bool process_remote_begin(StringInfo s, GlobalTransactionId *gtid);
-static bool process_remote_message(StringInfo s);
+static bool process_remote_message(StringInfo s, MtmReceiverContext *receiver_ctx);
 static void process_remote_commit(StringInfo s, GlobalTransactionId *current_gtid);
 static void process_remote_insert(StringInfo s, Relation rel);
 static void process_remote_update(StringInfo s, Relation rel);
@@ -495,7 +495,7 @@ process_remote_begin(StringInfo s, GlobalTransactionId *gtid)
 }
 
 static bool
-process_remote_message(StringInfo s)
+process_remote_message(StringInfo s, MtmReceiverContext *receiver_ctx)
 {
 	char action = pq_getmsgbyte(s);
 	int messageSize = pq_getmsgint(s, 4);
@@ -528,6 +528,23 @@ process_remote_message(StringInfo s)
 			standalone = true;
 			break;
 		}
+		case 'P':
+		{
+			int node_id = -1;
+
+			sscanf(messageBody, "%d", &node_id);
+
+			Assert(node_id > 0);
+			Assert(receiver_ctx != NULL);
+			// XXX assert that it is receiver itself
+
+			if (!receiver_ctx->is_recovery && node_id == MtmNodeId)
+			{
+				Assert(!receiver_ctx->parallel_allowed);
+				receiver_ctx->parallel_allowed = true;
+			}
+			break;
+		}
 		default:
 			Assert(false);
  	}
@@ -1229,7 +1246,8 @@ process_remote_delete(StringInfo s, Relation rel)
 	CommandCounterIncrement();
 }
 
-void MtmExecutor(void* work, size_t size)
+void
+MtmExecutor(void* work, size_t size, MtmReceiverContext *receiver_ctx)
 {
     StringInfoData s;
     Relation rel = NULL;
@@ -1342,7 +1360,7 @@ void MtmExecutor(void* work, size_t size)
 			{
   			    close_rel(rel);
 				rel = NULL;
-				inside_transaction = !process_remote_message(&s);
+				inside_transaction = !process_remote_message(&s, receiver_ctx);
 				break;
 			}
 			case 'Z':

src/pglogical_output.c

@@ -36,6 +36,8 @@
 #include "replication/message.h"
 #include "replication/origin.h"
 
+#include "storage/ipc.h"
+
 #include "utils/builtins.h"
 #include "utils/catcache.h"
 #include "utils/guc.h"
@@ -471,6 +473,13 @@ pg_decode_caughtup(LogicalDecodingContext *ctx)
 		MtmOutputPluginPrepareWrite(ctx, true, true);
 		data->api->write_caughtup(ctx->out, data, ctx->reader->EndRecPtr);
 		MtmOutputPluginWrite(ctx, true, true);
+
+		/*
+		 * This hook can be called mupltiple times when there is concurrent
+		 * load, so exit right after we wrote recovery message first time during
+		 * current recovery session.
+		 */
+		proc_exit(0);
 	}
 }
 

src/pglogical_proto.c

@@ -36,6 +36,8 @@
 #include "utils/typcache.h"
 #include "utils/snapmgr.h"
 
+#include "replication/message.h"
+
 #include "pglogical_relid_map.h"
 
 #include "multimaster.h"
@@ -272,7 +274,9 @@ void pglogical_write_caughtup(StringInfo out, PGLogicalOutputData *data,
 	MtmDecoderPrivate *hooks_data = (MtmDecoderPrivate *) data->hooks.hooks_private_data;
 
 	Assert(hooks_data->is_recovery);
-	pq_sendbyte(out, 'Z');		/* sending CAUGHT-UP */
+	/* sending CAUGHT-UP */
+	pq_sendbyte(out, 'Z');
+	MtmStateProcessNeighborEvent(MtmReplicationNodeId, MTM_NEIGHBOR_RECOVERY_CAUGHTUP, false);
 }
 
 /*
@@ -736,15 +740,14 @@ MtmReplicationStartupHook(struct PGLogicalStartupHookArgs* args)
 		}
 	}
 
-	MtmLock(LW_EXCLUSIVE);
-
 	/*
 	 * Set proper originId mappings.
 	 *
 	 * This is copypasted from receiver. Better to have normal init method
 	 * to setup all stuff in shared memory. But seems that there is no such
 	 * callback in vanilla pg and adding one will require some carefull thoughts.
 	 */
+	MtmLock(LW_EXCLUSIVE);
 	for (i = 0; i < Mtm->nAllNodes; i++)
 	{
 		char	   *originName;
@@ -777,6 +780,7 @@ MtmReplicationStartupHook(struct PGLogicalStartupHookArgs* args)
 		mtm_log(ERROR, "Out-of-clique node %d tries to connect",
 				 MtmReplicationNodeId);
 	}
+	MtmUnlock();
 
 	if (hooks_data->is_recovery)
 	{
@@ -785,17 +789,40 @@ MtmReplicationStartupHook(struct PGLogicalStartupHookArgs* args)
 				MtmReplicationNodeId);
 
 		Assert(MyReplicationSlot != NULL);
-		MtmStateProcessNeighborEvent(MtmReplicationNodeId, MTM_NEIGHBOR_WAL_SENDER_START_RECOVERY, true);
+		MtmStateProcessNeighborEvent(MtmReplicationNodeId, MTM_NEIGHBOR_WAL_SENDER_START_RECOVERY, false);
 	}
 	else
 	{
 		mtm_log(ProtoTraceMode,
 				"Walsender starts in recovered mode to node %d",
 				MtmReplicationNodeId);
 
-		MtmStateProcessNeighborEvent(MtmReplicationNodeId, MTM_NEIGHBOR_WAL_SENDER_START_RECOVERED, true);
+		/*
+		 * Indicate receiver that after this point in wal it is safe to send
+		 * transaction to the pool of workers. Before this point in wal (or in other
+		 * words before we processed MTM_NEIGHBOR_WAL_SENDER_START_RECOVERY event
+		 * and enabled this node in our disabledNodeMask) our backends are not waiting
+		 * for prepare confirmations from this node, so receiver can get precommit
+		 * before it will finish prepare. This will leave receiver with lots of
+		 * prepared transactions that will never be commited as precommit and commit
+		 * already happend before prepare.
+		 *
+		 * To ensure that all transactions ended after this message had seen right
+		 * disabledNodeMask we took MtmCommitBarrier in exclusive mode to await
+		 * finish of all transactions with potentially old disabledNodeMask.
+		 */
+		if (!hooks_data->is_recovery)
+		{
+			char *dest_id = psprintf("%d", MtmReplicationNodeId);
+
+			LWLockAcquire(MtmCommitBarrier, LW_EXCLUSIVE);
+			MtmStateProcessNeighborEvent(MtmReplicationNodeId, MTM_NEIGHBOR_WAL_SENDER_START_RECOVERED, false);
+			LWLockRelease(MtmCommitBarrier);
+
+			XLogFlush(LogLogicalMessage("P", dest_id, strlen(dest_id), false));
+		}
 	}
-	MtmUnlock();
+
 
 }