handle exit of wal-sender

knizhnik · knizhnik · commit 463c1bc6c521 · 2016-04-05T18:02:40.000+03:00
diff --git a/contrib/mmts/multimaster.c b/contrib/mmts/multimaster.c
@@ -865,10 +865,13 @@ void MtmSendNotificationMessage(MtmTransState* ts, MtmMessageCode cmd)
 
 void MtmJoinTransaction(GlobalTransactionId* gtid, csn_t globalSnapshot)
 {
-	MtmLock(LW_EXCLUSIVE);
-	MtmSyncClock(globalSnapshot);	
-	MtmUnlock();
-	
+	if (globalSnapshot != INVALID_CSN) {
+		MtmLock(LW_EXCLUSIVE);
+		MtmSyncClock(globalSnapshot);	
+		MtmUnlock();
+	} else { 
+		globalSnapshot = MtmTx.snapshot;
+	}
 	if (!TransactionIdIsValid(gtid->xid)) { 
 		/* In case of recovery InvalidTransactionId is passed */
 		Assert(Mtm->status == MTM_RECOVERY);
@@ -1877,6 +1880,14 @@ void MtmDropNode(int nodeId, bool dropSlot)
 		}		
 	}
 }
+static void
+MtmOnProcExit(int code, Datum arg)
+{
+	if (MtmReplicationNodeId >= 0) { 
+		elog(WARNING, "WAL-sender to %d is terminated", MtmReplicationNodeId); 
+		MtmOnNodeDisconnect(MtmReplicationNodeId);
+	}
+}
 
 static void 
 MtmReplicationStartupHook(struct PGLogicalStartupHookArgs* args)
@@ -1923,13 +1934,17 @@ MtmReplicationStartupHook(struct PGLogicalStartupHookArgs* args)
 		elog(NOTICE, "Node %d start logical replication to node %d in normal mode", MtmNodeId, MtmReplicationNodeId); 
 	}
 	MtmUnlock();
+	on_proc_exit(MtmOnProcExit, 0);
 }
 
 static void 
 MtmReplicationShutdownHook(struct PGLogicalShutdownHookArgs* args)
 {
-	elog(WARNING, "Logical replication to node %d is stopped", MtmReplicationNodeId); 
-	MtmOnNodeDisconnect(MtmReplicationNodeId);
+	if (MtmReplicationNodeId >= 0) { 
+		elog(WARNING, "Logical replication to node %d is stopped", MtmReplicationNodeId); 
+		MtmOnNodeDisconnect(MtmReplicationNodeId);
+		MtmReplicationNodeId = -1; /* defuse on_proc_exit hook */
+	}
 }
 
 static bool 
@@ -2167,7 +2182,8 @@ static bool MtmRunUtilityStmt(PGconn* conn, char const* sql, char **errmsg)
 	return ret;
 }
 
-void MtmNoticeReceiver(void *i, const PGresult *res)
+static void 
+MtmNoticeReceiver(void *i, const PGresult *res)
 {
 	char *notice = PQresultErrorMessage(res);
 	char *stripped_notice;
diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c
@@ -541,9 +541,6 @@ 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);

Original file line number	Diff line number	Diff line change
`@@ -541,9 +541,6 @@ DecodeCommit(LogicalDecodingContext ctx, XLogRecordBuffer buf,`
`541`	`541`	`(parsed->dbId != InvalidOid && parsed->dbId != ctx->slot->data.database) \|\|`
`542`	`542`	`FilterByOrigin(ctx, origin_id))`
`543`	`543`	`{`
`544`		`- elog(WARNING, "%d: WAL-SENDER ignore record %lx with origin %d: SnapBuildXactNeedsSkip=%d, FilterByOrigin=%d",`
`545`		`- getpid(), buf->origptr, origin_id,`
`546`		`- SnapBuildXactNeedsSkip(ctx->snapshot_builder, buf->origptr), FilterByOrigin(ctx, origin_id));`
`547`	`544`	`for (i = 0; i < parsed->nsubxacts; i++)`
`548`	`545`	`{`
`549`	`546`	`ReorderBufferForget(ctx->reorder, parsed->subxacts[i], buf->origptr);`