Mark node as online only when it totally completed recovery. Also

pass ReceiverContext to bgworkers to avoid unnecessary replies
during non parallel-safe replication.

Author: kelvich

src/bgwpool.c

@@ -39,10 +39,13 @@ BgwShutdownHandler(int sig)
 	die(sig);
 }
 
-static void BgwPoolMainLoop(BgwPool* pool)
+static void
+BgwPoolMainLoop(BgwPool* pool)
 {
-    int size;
-    void* work;
+	int size;
+	void* work;
+	size_t payload = sizeof(MtmReceiverContext) + sizeof(size_t);
+	MtmReceiverContext ctx;
 	static PortalData fakePortal;
 
 	mtm_log(BgwPoolEvent, "Start background worker %d, shutdown=%d", MyProcPid, pool->shutdown);
@@ -56,61 +59,72 @@ static void BgwPoolMainLoop(BgwPool* pool)
 	pqsignal(SIGTERM, BgwShutdownHandler);
 	pqsignal(SIGHUP, PostgresSigHupHandler);
 
-    BackgroundWorkerUnblockSignals();
+	BackgroundWorkerUnblockSignals();
 	BackgroundWorkerInitializeConnection(pool->dbname, pool->dbuser, 0);
 	ActivePortal = &fakePortal;
 	ActivePortal->status = PORTAL_ACTIVE;
 	ActivePortal->sourceText = "";
 
-	while (true) {
+	while (true)
+	{
 		if (ConfigReloadPending)
 		{
 			ConfigReloadPending = false;
 			ProcessConfigFile(PGC_SIGHUP);
 		}
 
 		PGSemaphoreLock(pool->available);
-        SpinLockAcquire(&pool->lock);
-		if (pool->shutdown) { 	
+		SpinLockAcquire(&pool->lock);
+		if (pool->shutdown)
+		{
 			PGSemaphoreUnlock(pool->available);
 			break;
 		}
-        size = *(int*)&pool->queue[pool->head];
-        Assert(size < pool->size);
-        work = palloc(size);
-        pool->pending -= 1;
-        pool->active += 1;
-		if (pool->lastPeakTime == 0 && pool->active == pool->nWorkers && pool->pending != 0) {
+		size = * (int *) &pool->queue[pool->head];
+		ctx  = * (MtmReceiverContext *) &pool->queue[pool->head + sizeof(size_t)];
+
+		Assert(size < pool->size);
+		work = palloc(size);
+		pool->pending -= 1;
+		pool->active += 1;
+		if (pool->lastPeakTime == 0 && pool->active == pool->nWorkers && pool->pending != 0)
 			pool->lastPeakTime = MtmGetSystemTime();
+
+		if (pool->head + size + payload > pool->size)
+		{
+			memcpy(work, pool->queue, size);
+			pool->head = INTALIGN(size);
 		}
-        if (pool->head + size + 4 > pool->size) { 
-            memcpy(work, pool->queue, size);
-            pool->head = INTALIGN(size);
-        } else { 
-            memcpy(work, &pool->queue[pool->head+4], size);
-            pool->head += 4 + INTALIGN(size);
-        }
-        if (pool->size == pool->head) { 
-            pool->head = 0;
-        }
-        if (pool->producerBlocked) {
-            pool->producerBlocked = false;
+		else
+		{
+			memcpy(work, &pool->queue[pool->head + payload], size);
+			pool->head += payload + INTALIGN(size);
+		}
+
+		if (pool->size == pool->head)
+			pool->head = 0;
+
+		if (pool->producerBlocked)
+		{
+			pool->producerBlocked = false;
 			PGSemaphoreUnlock(pool->overflow);
 			pool->lastPeakTime = 0;
-        }
-        SpinLockRelease(&pool->lock);
+		}
+
+		SpinLockRelease(&pool->lock);
 
 		/* Ignore cancel that arrived before we started current command */
 		QueryCancelPending = false;
 
-		pool->executor(work, size, NULL);
+		pool->executor(work, size, &ctx);
 		pfree(work);
 
-        SpinLockAcquire(&pool->lock);
-        pool->active -= 1;
+		SpinLockAcquire(&pool->lock);
+		pool->active -= 1;
 		pool->lastPeakTime = 0;
-        SpinLockRelease(&pool->lock);
-    }
+		SpinLockRelease(&pool->lock);
+	}
+
 	SpinLockRelease(&pool->lock);
 	mtm_log(BgwPoolEvent, "Shutdown background worker %d", MyProcPid);
 }
@@ -211,9 +225,13 @@ static void BgwStartExtraWorker(BgwPool* pool)
 	}
 }
 
-void BgwPoolExecute(BgwPool* pool, void* work, size_t size)
+void
+BgwPoolExecute(BgwPool* pool, void* work, size_t size, MtmReceiverContext *ctx)
 {
-    if (size+4 > pool->size) {
+	size_t payload = sizeof(MtmReceiverContext) + sizeof(size_t);
+
+	if (size + payload > pool->size)
+	{
 		/* 
 		 * Size of work is larger than size of shared buffer: 
 		 * run it immediately
@@ -222,42 +240,52 @@ void BgwPoolExecute(BgwPool* pool, void* work, size_t size)
 		return;
 	}
 
-    SpinLockAcquire(&pool->lock);
-    while (!pool->shutdown) { 
-        if ((pool->head <= pool->tail && pool->size - pool->tail < size + 4 && pool->head < size) 
-            || (pool->head > pool->tail && pool->head - pool->tail < size + 4))
-        {
-            if (pool->lastPeakTime == 0) {
+	SpinLockAcquire(&pool->lock);
+	while (!pool->shutdown)
+	{
+		if ((pool->head <= pool->tail && pool->size - pool->tail < size + payload && pool->head < size)
+			|| (pool->head > pool->tail && pool->head - pool->tail < size + payload))
+		{
+			if (pool->lastPeakTime == 0)
 				pool->lastPeakTime = MtmGetSystemTime();
-			}
+
 			pool->producerBlocked = true;
-            SpinLockRelease(&pool->lock);
+			SpinLockRelease(&pool->lock);
 			PGSemaphoreLock(pool->overflow);
-            SpinLockAcquire(&pool->lock);
-        } else {
-            pool->pending += 1;
-			if (pool->active + pool->pending > pool->nWorkers) { 
-				BgwStartExtraWorker(pool);				
-			}
-			if (pool->lastPeakTime == 0 && pool->active == pool->nWorkers && pool->pending != 0) {
+			SpinLockAcquire(&pool->lock);
+		}
+		else
+		{
+			pool->pending += 1;
+
+			if (pool->active + pool->pending > pool->nWorkers)
+				BgwStartExtraWorker(pool);
+
+			if (pool->lastPeakTime == 0 && pool->active == pool->nWorkers && pool->pending != 0)
 				pool->lastPeakTime = MtmGetSystemTime();
+
+			*(int *)&pool->queue[pool->tail] = size;
+			*(MtmReceiverContext *)&pool->queue[pool->tail + sizeof(size_t)] = *ctx;
+
+			if (pool->size - pool->tail >= size + payload)
+			{
+				memcpy(&pool->queue[pool->tail + payload], work, size);
+				pool->tail += payload + INTALIGN(size);
+			}
+			else
+			{
+				memcpy(pool->queue, work, size);
+				pool->tail = INTALIGN(size);
 			}
-            *(int*)&pool->queue[pool->tail] = size;
-            if (pool->size - pool->tail >= size + 4) { 
-                memcpy(&pool->queue[pool->tail+4], work, size);
-                pool->tail += 4 + INTALIGN(size);
-            } else { 
-                memcpy(pool->queue, work, size);
-                pool->tail = INTALIGN(size);
-            }
-            if (pool->tail == pool->size) {
-                pool->tail = 0;
-            }
+
+			if (pool->tail == pool->size)
+				pool->tail = 0;
+
 			PGSemaphoreUnlock(pool->available);
-            break;
-        }
-    }
-    SpinLockRelease(&pool->lock);            
+			break;
+		}
+	}
+	SpinLockRelease(&pool->lock);
 }
 
 void BgwPoolStop(BgwPool* pool)

src/dmq.c

@@ -1072,8 +1072,8 @@ dmq_receiver_loop(PG_FUNCTION_ARGS)
 
 			if (rc > 0)
 			{
-				last_message_at = dmq_now();
 				dmq_handle_message(&s, mq_handles, seg);
+				last_message_at = dmq_now();
 				reader_state = NeedByte;
 			}
 		}

src/include/bgwpool.h

@@ -54,7 +54,7 @@ extern void BgwPoolStart(BgwPool* pool, char *poolName);
 
 extern void BgwPoolInit(BgwPool* pool, BgwPoolExecutor executor, char const* dbname, char const* dbuser, size_t queueSize, size_t nWorkers);
 
-extern void BgwPoolExecute(BgwPool* pool, void* work, size_t size);
+extern void BgwPoolExecute(BgwPool* pool, void* work, size_t size, MtmReceiverContext *ctx);
 
 extern size_t BgwPoolGetQueueSize(BgwPool* pool);
 

src/include/logger.h

@@ -18,8 +18,8 @@
 typedef enum MtmLogTag
 {
 	/* general */
-	MtmTxTrace				= DEBUG1,
-	MtmTxFinish				= DEBUG1,
+	MtmTxTrace				= LOG,
+	MtmTxFinish				= LOG,
 
 	/* dmq */
 	DmqStateIntermediate	= DEBUG1,

src/include/multimaster.h

@@ -126,6 +126,7 @@ typedef struct
 typedef struct
 {
 	bool extension_created;
+	bool recovered;
 	MtmNodeStatus status;              /* Status of this node */
 	char *statusReason;                /* A human-readable description of why the current status was set */
 	int recoverySlot;                  /* NodeId of recovery slot or 0 if none */

src/multimaster.c

@@ -529,6 +529,7 @@ static void MtmInitialize()
 	{
 		MemSet(Mtm, 0, sizeof(MtmState) + sizeof(MtmNodeInfo)*(MtmMaxNodes-1));
 		Mtm->extension_created = false;
+		Mtm->recovered = false;
 		Mtm->status = MTM_DISABLED; //MTM_INITIALIZATION;
 		Mtm->recoverySlot = 0;
 		Mtm->locks = GetNamedLWLockTranche(MULTIMASTER_NAME);

src/pglogical_apply.c

@@ -86,7 +86,7 @@ static void UserTableUpdateIndexes(EState *estate, TupleTableSlot *slot);
 
 static bool process_remote_begin(StringInfo s, GlobalTransactionId *gtid);
 static bool process_remote_message(StringInfo s, MtmReceiverContext *receiver_ctx);
-static void process_remote_commit(StringInfo s, GlobalTransactionId *current_gtid);
+static void process_remote_commit(StringInfo s, GlobalTransactionId *current_gtid, MtmReceiverContext *receiver_ctx);
 static void process_remote_insert(StringInfo s, Relation rel);
 static void process_remote_update(StringInfo s, Relation rel);
 static void process_remote_delete(StringInfo s, Relation rel);
@@ -535,13 +535,17 @@ process_remote_message(StringInfo s, MtmReceiverContext *receiver_ctx)
 			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);
+				Assert(receiver_ctx->node_id > 0);
+				Assert(receiver_ctx->node_id == MtmReplicationNodeId);
+				Assert(receiver_ctx->node_id != node_id);
+
 				receiver_ctx->parallel_allowed = true;
+				MtmStateProcessNeighborEvent(MtmReplicationNodeId, MTM_NEIGHBOR_WAL_RECEIVER_START, false);
 			}
 			standalone = true;
 			break;
@@ -724,7 +728,7 @@ mtm_send_reply(TransactionId xid, int node_id, MtmMessageCode msg_code)
 }
 
 static void
-process_remote_commit(StringInfo in, GlobalTransactionId *current_gtid)
+process_remote_commit(StringInfo in, GlobalTransactionId *current_gtid, MtmReceiverContext *receiver_ctx)
 {
 	uint8 		event;
 	lsn_t       end_lsn;
@@ -767,7 +771,7 @@ process_remote_commit(StringInfo in, GlobalTransactionId *current_gtid)
 			}
 			mtm_log(MtmTxFinish, "TXFINISH: %s precommitted", gid);
 
-			if (Mtm->status != MTM_RECOVERY)
+			if (receiver_ctx->parallel_allowed)
 			{
 				TransactionId origin_xid = MtmGidParseXid(gid);
 				mtm_send_reply(origin_xid, origin_node, MSG_PRECOMMITTED);
@@ -813,8 +817,7 @@ process_remote_commit(StringInfo in, GlobalTransactionId *current_gtid)
 
 			MtmDeadlockDetectorRemoveXact(xid);
 
-			/* XXX: we can move that to callbacks to keep apply clean */
-			if (Mtm->status != MTM_RECOVERY)
+			if (receiver_ctx->parallel_allowed)
 			{
 				TransactionId origin_xid = MtmGidParseXid(gid);
 				mtm_send_reply(origin_xid, origin_node, res ? MSG_PREPARED : MSG_ABORTED);
@@ -842,7 +845,7 @@ process_remote_commit(StringInfo in, GlobalTransactionId *current_gtid)
 			mtm_log(MtmTxFinish, "TXFINISH: %s committed", gid);
 			CommitTransactionCommand();
 
-			if (Mtm->status != MTM_RECOVERY)
+			if (receiver_ctx->parallel_allowed)
 			{
 				TransactionId origin_xid = MtmGidParseXid(gid);
 				mtm_send_reply(origin_xid, origin_node, MSG_COMMITTED);
@@ -868,7 +871,7 @@ process_remote_commit(StringInfo in, GlobalTransactionId *current_gtid)
 			Assert(false);
 	}
 
-	if (Mtm->status == MTM_RECOVERY)
+	if (!receiver_ctx->parallel_allowed)
 	{
 		// XXX
 		elog(LOG, "Recover transaction %s event=%d", gid, event);
@@ -1295,7 +1298,7 @@ MtmExecutor(void* work, size_t size, MtmReceiverContext *receiver_ctx)
                 /* COMMIT */
             case 'C':
   			    close_rel(rel);
-				process_remote_commit(&s, &current_gtid);
+				process_remote_commit(&s, &current_gtid, receiver_ctx);
 				inside_transaction = false;
                 break;
                 /* INSERT */
@@ -1379,7 +1382,6 @@ MtmExecutor(void* work, size_t size, MtmReceiverContext *receiver_ctx)
     }
     PG_CATCH();
     {
-		// XXX: change all this to re-throw
 
 		old_context = MemoryContextSwitchTo(MtmApplyContext);
 		MtmHandleApplyError();
@@ -1397,7 +1399,7 @@ MtmExecutor(void* work, size_t size, MtmReceiverContext *receiver_ctx)
 		if (TransactionIdIsValid(current_gtid.xid))
 		{
 			MtmDeadlockDetectorRemoveXact(GetCurrentTransactionIdIfAny());
-			if (Mtm->status != MTM_RECOVERY)
+			if (receiver_ctx->parallel_allowed)
 				mtm_send_reply(current_gtid.xid, current_gtid.node, MSG_ABORTED);
 		}