Merge branch 'master' of github.com:postgrespro/postgres_cluster

Author: kvap

contrib/mmts/arbiter.c

@@ -386,7 +386,7 @@ static void MtmOpenConnections()
 static bool MtmSendToNode(int node, void const* buf, int size)
 {
 	while (sockets[node] < 0 || !MtmWriteSocket(sockets[node], buf, size)) { 
-		elog(WARNING, "Arbiter failed to write socket: %d", errno);
+		elog(WARNING, "Arbiter failed to write to node %d: %d", node+1, errno);
 		if (sockets[node] >= 0) { 
 			close(sockets[node]);
 		}
@@ -395,6 +395,7 @@ static bool MtmSendToNode(int node, void const* buf, int size)
 			MtmOnNodeDisconnect(node+1);
 			return false;
 		}
+		elog(NOTICE, "Arbiter restablish connection with node %d", node+1);
 	}
 	return true;
 }
@@ -688,6 +689,8 @@ static void MtmTransReceiver(Datum arg)
 							if ((~msg->disabledNodeMask & Mtm->disabledNodeMask) != 0) { 
 								/* Coordinator's disabled mask is wider than of this node: so reject such transaction to avoid 
 								   commit on smaller subset of nodes */
+								elog(WARNING, "Coordinator of distributed transaction see less nodes than node %d: %lx instead of %lx",
+									 msg->node, Mtm->disabledNodeMask, msg->disabledNodeMask);
 								ts->status = TRANSACTION_STATUS_ABORTED;
 								MtmAdjustSubtransactions(ts);
 							}

contrib/mmts/multimaster.c

@@ -637,7 +637,7 @@ MtmBeginTransaction(MtmCurrentTrans* x)
 		x->isPrepared = false;
 		x->isTransactionBlock = IsTransactionBlock();
 		/* Application name can be cahnged usnig PGAPPNAME environment variable */
-		if (x->isDistributed && Mtm->status != MTM_ONLINE && strcmp(application_name, MULTIMASTER_ADMIN) != 0) { 
+		if (!IsBackgroundWorker && x->isDistributed && Mtm->status != MTM_ONLINE && strcmp(application_name, MULTIMASTER_ADMIN) != 0) { 
 			/* reject all user's transactions at offline cluster */
 			MtmUnlock();			
 			elog(ERROR, "Multimaster node is not online: current status %s", MtmNodeStatusMnem[Mtm->status]);
@@ -673,7 +673,7 @@ MtmPrePrepareTransaction(MtmCurrentTrans* x)
 
 	if (Mtm->disabledNodeMask != 0) { 
 		MtmRefreshClusterStatus(true);
-		if (Mtm->status != MTM_ONLINE) { 
+		if (!IsBackgroundWorker && Mtm->status != MTM_ONLINE) { 
 			elog(ERROR, "Abort current transaction because this cluster node is not online");			
 		}
 	}
@@ -683,7 +683,9 @@ MtmPrePrepareTransaction(MtmCurrentTrans* x)
 	/*
 	 * Check if there is global multimaster lock preventing new transaction from commit to make a chance to wal-senders to catch-up
 	 */
-	MtmCheckClusterLock();
+	if (!x->isReplicated) { 
+		MtmCheckClusterLock();
+	}
 
 	ts = hash_search(MtmXid2State, &x->xid, HASH_ENTER, NULL);
 	ts->status = TRANSACTION_STATUS_IN_PROGRESS;
@@ -998,16 +1000,15 @@ bool MtmRecoveryCaughtUp(int nodeId, XLogRecPtr slotLSN)
 		if (slotLSN == walLSN) {
 			if (BIT_CHECK(Mtm->nodeLockerMask, nodeId-1)) { 
 				elog(WARNING,"Node %d is caught-up", nodeId);	
-				BIT_CLEAR(Mtm->disabledNodeMask, nodeId-1);
 				BIT_CLEAR(Mtm->walSenderLockerMask, MyWalSnd - WalSndCtl->walsnds);
 				BIT_CLEAR(Mtm->nodeLockerMask, nodeId-1);
 				Mtm->nLockers -= 1;
 			} else { 
 				elog(WARNING,"Node %d is caugth-up without locking cluster", nodeId);	
 				/* We are lucky: caugth-up without locking cluster! */
-				Mtm->nNodes += 1;
-				BIT_CLEAR(Mtm->disabledNodeMask, nodeId-1);
 			}
+			BIT_CLEAR(Mtm->disabledNodeMask, nodeId-1);
+			Mtm->nNodes += 1;
 			caughtUp = true;
 		} else if (!BIT_CHECK(Mtm->nodeLockerMask, nodeId-1)
 				   && slotLSN + MtmMinRecoveryLag > walLSN) 
@@ -2250,7 +2251,12 @@ MtmExecutorFinish(QueryDesc *queryDesc)
 
 void MtmExecute(void* work, int size)
 {
-    BgwPoolExecute(&Mtm->pool, work, size);
+	if (Mtm->status == MTM_RECOVERY) { 
+		/* During recovery apply changes sequentially to preserve commit order */
+		MtmExecutor(0, work, size);
+	} else { 
+		BgwPoolExecute(&Mtm->pool, work, size);
+	}
 }
 
 static BgwPool* 

contrib/mmts/multimaster.h

@@ -8,7 +8,7 @@
 #include "pglogical_output/hooks.h"
 
 #define MTM_TUPLE_TRACE(fmt, ...)
-#if 0
+#if 1
 #define MTM_INFO(fmt, ...) fprintf(stderr, fmt, ## __VA_ARGS__) 
 #define MTM_TRACE(fmt, ...) fprintf(stderr, fmt, ## __VA_ARGS__) 
 #else

contrib/mmts/pglogical_proto.c

@@ -147,7 +147,6 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
 			return;
 		}
 	}
-
     pq_sendbyte(out, 'C');		/* sending COMMIT */
 
 	MTM_INFO("PGLOGICAL_SEND commit: event=%d, gid=%s, commit_lsn=%lx, txn->end_lsn=%lx, xlog=%lx\n", flags, txn->gid, commit_lsn, txn->end_lsn, GetXLogInsertRecPtr());

src/backend/replication/logical/decode.c

@@ -488,7 +488,7 @@ DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 {
 	XLogRecPtr	origin_lsn = InvalidXLogRecPtr;
 	TimestampTz	commit_time = parsed->xact_time;
-	XLogRecPtr	origin_id = XLogRecGetOrigin(buf->record);
+	RepOriginId	origin_id = XLogRecGetOrigin(buf->record);
 	int			i;
 
 	if (parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
@@ -541,6 +541,9 @@ DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 		(parsed->dbId != InvalidOid && parsed->dbId != ctx->slot->data.database) ||
 		FilterByOrigin(ctx, origin_id))
 	{
+		elog(WARNING, "%d: WAL-SENDER ignore record %lx with origin %d: SnapBuildXactNeedsSkip=%d, FilterByOrigin=%d", 
+			 getpid(), buf->origptr, origin_id, 
+			 SnapBuildXactNeedsSkip(ctx->snapshot_builder, buf->origptr), FilterByOrigin(ctx, origin_id));
 		for (i = 0; i < parsed->nsubxacts; i++)
 		{
 			ReorderBufferForget(ctx->reorder, parsed->subxacts[i], buf->origptr);