Finish recovery after we sure that all forwarded transactions are applied up to some lsn

Author: kelvich

Cluster.pm

@@ -159,7 +159,7 @@ sub configure
 			multimaster.max_nodes = 6
 			# XXX try without ignore_tables_without_pk
 			multimaster.ignore_tables_without_pk = true
-			multimaster.volkswagen_mode = 1
+			# multimaster.volkswagen_mode = 1
 			log_line_prefix = '%t [%p]: '
 		));
 

src/include/multimaster.h

@@ -162,6 +162,7 @@ extern bool MtmBackgroundWorker;
 extern char* MtmRefereeConnStr;
 
 extern LWLock *MtmCommitBarrier;
+extern LWLock *MtmReceiverBarrier;
 
 extern void MtmXactCallback2(XactEvent event, void *arg);
 extern bool MtmIsUserTransaction(void);
@@ -194,5 +195,6 @@ extern void MtmUpdateControlFile(void);
 extern void MtmCheckSlots(void);
 
 extern TimestampTz MtmGetIncreasingTimestamp(void);
+extern bool MtmAllApplyWorkersFinished(void);
 
 #endif

src/include/pglogical_output.h

@@ -43,8 +43,8 @@ typedef struct HookFuncName
 
 typedef struct MtmDecoderPrivate
 {
-	int		magic; // XXX
 	bool	is_recovery;
+	bool	recovery_done;
 	int64	session_id;
 } MtmDecoderPrivate;
 

src/multimaster.c

@@ -124,6 +124,7 @@ static TransactionManager MtmTM =
 };
 
 LWLock *MtmCommitBarrier;
+LWLock *MtmReceiverBarrier;
 
 bool  MtmDoReplication;
 char* MtmDatabaseName;
@@ -601,6 +602,7 @@ static void MtmInitialize()
 	RegisterXactCallback(MtmXactCallback2, NULL);
 
 	MtmCommitBarrier = &(GetNamedLWLockTranche(MULTIMASTER_NAME)[MtmMaxNodes*2+1].lock);
+	MtmReceiverBarrier = &(GetNamedLWLockTranche(MULTIMASTER_NAME)[MtmMaxNodes*2+2].lock);
 
 	MtmDoReplication = true;
 	TM = &MtmTM;
@@ -1188,7 +1190,7 @@ _PG_init(void)
 	 * resources in mtm_shmem_startup().
 	 */
 	RequestAddinShmemSpace(MTM_SHMEM_SIZE + MtmMaxNodes*MtmQueueSize + sizeof(MtmTime));
-	RequestNamedLWLockTranche(MULTIMASTER_NAME, 1 + MtmMaxNodes*2 + 1);
+	RequestNamedLWLockTranche(MULTIMASTER_NAME, 1 + MtmMaxNodes*2 + 2);
 
 	MtmMonitorInitialize();
 
@@ -1837,3 +1839,28 @@ MtmWaitForExtensionCreation(void)
 		MtmSleep(USECS_PER_SEC);
 	}
 }
+
+bool
+MtmAllApplyWorkersFinished()
+{
+	int i;
+
+	for (i = 0; i < Mtm->nAllNodes; i++)
+	{
+		volatile int ntasks;
+
+		if (i == MtmNodeId - 1)
+			continue;
+
+		SpinLockAcquire(&Mtm->nodes[i].pool.lock);
+		ntasks = Mtm->nodes[i].pool.active + Mtm->nodes[i].pool.pending;
+		SpinLockRelease(&Mtm->nodes[i].pool.lock);
+
+		if (ntasks != 0)
+			false;
+	}
+
+	return true;
+}
+
+

src/pglogical_apply.c

@@ -502,7 +502,7 @@ process_remote_message(StringInfo s, MtmReceiverContext *receiver_ctx)
 	char const* messageBody = pq_getmsgbytes(s, messageSize);
 	bool standalone = false;
 
-	MtmBeginSession(MtmReplicationNodeId);
+	MtmBeginSession(receiver_ctx->node_id);
 
 	switch (action)
 	{

src/pglogical_output.c

@@ -470,16 +470,31 @@ pg_decode_caughtup(LogicalDecodingContext *ctx)
 	 */
 	if (data->api && hooks_data->is_recovery)
 	{
-		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.
+		 * Create safe point in our WAL so that recovered node will not lose
+		 * any transactions during switch to normal mode.
 		 */
-		proc_exit(0);
+		if (!hooks_data->recovery_done)
+		{
+			LWLockAcquire(MtmReceiverBarrier, LW_EXCLUSIVE);
+			hooks_data->recovery_done = true;
+		}
+		else if (MtmAllApplyWorkersFinished())
+		{
+			MtmOutputPluginPrepareWrite(ctx, true, true);
+			data->api->write_caughtup(ctx->out, data, ctx->reader->EndRecPtr);
+			MtmOutputPluginWrite(ctx, true, true);
+
+			LWLockRelease(MtmReceiverBarrier);
+
+			/*
+			* 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

@@ -708,6 +708,8 @@ MtmReplicationStartupHook(struct PGLogicalStartupHookArgs* args)
 	hooks_data = (MtmDecoderPrivate *) palloc0(sizeof(MtmDecoderPrivate));
 	args->private_data = hooks_data;
 	hooks_data->session_id = 0;
+	hooks_data->recovery_done = false;
+	hooks_data->is_recovery = false;
 
 	Mtm->nodes[MtmReplicationNodeId-1].senderPid = MyProcPid;
 

src/pglogical_receiver.c

@@ -252,23 +252,24 @@ MtmUpdateLsnMapping(int node_id, lsn_t end_lsn)
 	lsn_t local_flush = GetFlushRecPtr();
 	MemoryContext old_context = MemoryContextSwitchTo(TopMemoryContext);
 
-	if (end_lsn != INVALID_LSN) {
+	if (end_lsn != INVALID_LSN)
+	{
 		/* Track commit lsn */
 		flushpos = (MtmFlushPosition *) palloc(sizeof(MtmFlushPosition));
 		flushpos->node_id = node_id;
 		flushpos->local_end = XactLastCommitEnd;
 		flushpos->remote_end = end_lsn;
 		dlist_push_tail(&MtmLsnMapping, &flushpos->node);
 	}
+
 	MtmLock(LW_EXCLUSIVE);
 	dlist_foreach_modify(iter, &MtmLsnMapping)
 	{
 		flushpos = dlist_container(MtmFlushPosition, node, iter.cur);
 		if (flushpos->local_end <= local_flush)
 		{
-			if (Mtm->nodes[node_id-1].flushPos < flushpos->remote_end) {
+			if (Mtm->nodes[node_id-1].flushPos < flushpos->remote_end)
 				Mtm->nodes[node_id-1].flushPos = flushpos->remote_end;
-			}
 			dlist_delete(iter.cur);
 			pfree(flushpos);
 		} else {
@@ -280,15 +281,23 @@ MtmUpdateLsnMapping(int node_id, lsn_t end_lsn)
 }
 
 static void
-MtmExecute(void* work, int size, MtmReceiverContext *receiver_ctx)
+MtmExecute(void* work, int size, MtmReceiverContext *receiver_ctx, bool cddl)
 {
 	/* parallel_allowed should never be set during recovery */
 	Assert( !(receiver_ctx->is_recovery && receiver_ctx->parallel_allowed) );
 
-	if (receiver_ctx->is_recovery || !receiver_ctx->parallel_allowed)
+	// end_lsn
+
+	LWLockAcquire(MtmReceiverBarrier, LW_SHARED);
+
+	if (receiver_ctx->is_recovery || !receiver_ctx->parallel_allowed || cddl)
 		MtmExecutor(work, size, receiver_ctx);
 	else
 		BgwPoolExecute(&Mtm->nodes[MtmReplicationNodeId-1].pool, work, size, receiver_ctx);
+
+	/* Our error handler can release lock */
+	if (LWLockHeldByMe(MtmReceiverBarrier))
+		LWLockRelease(MtmReceiverBarrier);
 }
 
 /*
@@ -766,10 +775,15 @@ pglogical_receiver_main(Datum main_arg)
 						ByteBufferReset(&buf);
 					}
 					if (stmt[0] == 'Z' || (stmt[0] == 'M' && (stmt[1] == 'L' || stmt[1] == 'P' || stmt[1] == 'C'))) {
-						if (stmt[0] == 'M' && stmt[1] == 'C') { /* concurrent DDL should be executed by parallel workers */
-							MtmExecute(stmt, msg_len, &receiver_ctx);
-						} else {
-							MtmExecutor(stmt, msg_len, &receiver_ctx); /* all other messages can be processed by receiver itself */
+						if (stmt[0] == 'M' && stmt[1] == 'C')
+						{
+							/* concurrent DDL should be executed by parallel workers */
+							MtmExecute(stmt, msg_len, &receiver_ctx, true);
+						}
+						else
+						{
+							/* all other messages should be processed by receiver itself */
+							MtmExecute(stmt, msg_len, &receiver_ctx, false);
 						}
 					} else {
 						ByteBufferAppend(&buf, stmt, msg_len);
@@ -783,23 +797,13 @@ pglogical_receiver_main(Datum main_arg)
 									pq_sendint(&spill_info, buf.used, 4);
 									MtmSpillToFile(spill_file, buf.data, buf.used);
 									MtmCloseSpillFile(spill_file);
-									MtmExecute(spill_info.data, spill_info.len, &receiver_ctx);
+									MtmExecute(spill_info.data, spill_info.len, &receiver_ctx, false);
 									spill_file = -1;
 									resetStringInfo(&spill_info);
-								} else {
-									if (MtmPreserveCommitOrder && buf.used == msg_len) {
-										/* Perform commit-prepared and rollback-prepared requested directly in receiver */
-										// timestamp_t stop, start = MtmGetSystemTime();
-										MtmExecutor(buf.data, buf.used, &receiver_ctx);
-										// stop = MtmGetSystemTime();
-										// if (stop - start > USECS_PER_SEC) {
-										// 	elog(WARNING, "Commit of prepared transaction takes %lld usec, flags=%x", stop - start, stmt[1]);
-										// }
-									} else {
-										/* all other commits should be applied in place */
-										// Assert(stmt[1] == PGLOGICAL_PREPARE || stmt[1] == PGLOGICAL_COMMIT || stmt[1] == PGLOGICAL_PRECOMMIT_PREPARED);
-										MtmExecute(buf.data, buf.used, &receiver_ctx);
-									}
+								}
+								else
+								{
+									MtmExecute(buf.data, buf.used, &receiver_ctx, false);
 								}
 							} else if (spill_file >= 0) {
 								MtmCloseSpillFile(spill_file);

tests/lib/bank_client.py

@@ -180,6 +180,16 @@ def no_prepared_tx(self):
         print("n_prepared = %d" % (n_prepared))
         return (n_prepared)
 
+    def list_prepared(self, node_id):
+        con = psycopg2.connect(self.dsns[node_id] + " application_name=mtm_admin")
+        cur = con.cursor()
+        cur.execute('select * from pg_prepared_xacts order by prepared;')
+        for pxact in cur.fetchall():
+            for i, col in enumerate(["transaction", "gid", "prepared", "owner", "database", "state3pc"]):
+                print(pxact[i], end="\t")
+            print("\n")
+        print("----\n")
+
     def insert_counts(self):
         counts = []
 
@@ -308,13 +318,12 @@ def total_tx(self, conn, cur, agg, conn_i):
             print(datetime.datetime.utcnow(), 'Isolation error, total ', self.total, ' -> ', total[0], ', node ', conn_i+1)
             self.total = total[0]
             print(self.oops)
-            # yield from cur.execute('select * from mtm.get_nodes_state()')
-            # nodes_state = yield from cur.fetchall()
-            # for i, col in enumerate(self.nodes_state_fields):
-            #     print("%17s" % col, end="\t")
-            #     for j in range(3):
-            #          print("%19s" % nodes_state[j][i], end="\t")
-            #     print("\n")
+            yield from cur.execute('select * from pg_prepared_xacts order by prepared;')
+            pxacts = yield from cur.fetchall()
+            for pxact in pxacts:
+                for i, col in enumerate(["transaction", "gid", "prepared", "owner", "database", "state3pc"]):
+                    print(pxact[i], end="\t")
+                print("\n")
 
     def run(self):
         # asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

tests/lib/test_helper.py

@@ -44,6 +44,7 @@ def awaitCommit(self, node_id):
             if ('commit' in aggs[node_id]['transfer']['finish'] and
                     aggs[node_id]['transfer']['finish']['commit'] > 10):
                 break
+            self.client.list_prepared(node_id)
             time.sleep(5)
             total_sleep += 5