Avoid hanging on exit in pglogical_receiver

knizhnik · knizhnik · commit af5de434dc6d · 2016-08-12T16:43:41.000+03:00
diff --git a/contrib/mmts/multimaster.c b/contrib/mmts/multimaster.c
@@ -2322,10 +2322,16 @@ void MtmReceiverStarted(int nodeId)
  * Druing recovery we need to open only one replication slot from which node should receive all transactions.
  * Slots at other nodes should be removed 
  */
-MtmReplicationMode MtmGetReplicationMode(int nodeId)
+MtmReplicationMode MtmGetReplicationMode(int nodeId, sig_atomic_t volatile* shutdown)
 {
 	bool recovery = false;
-	if (Mtm->status != MTM_CONNECTED && Mtm->status != MTM_ONLINE) { 		
+
+	while (Mtm->status != MTM_CONNECTED && Mtm->status != MTM_ONLINE) 
+	{ 	
+		if (*shutdown) 
+		{ 
+			return REPLMODE_EXIT;
+		}
 		MTM_LOG2("%d: receiver slot mode %s", MyProcPid, MtmNodeStatusMnem[Mtm->status]);
 		if (Mtm->status == MTM_RECOVERY) { 
 			recovery = true;
@@ -2342,7 +2348,6 @@ MtmReplicationMode MtmGetReplicationMode(int nodeId)
 		}
 		/* delay opening of other slots until recovery is completed */
 		MtmSleep(STATUS_POLL_DELAY);
-		return REPLMODE_UNKNOWN;
 	}
 	if (recovery) { 
 		MTM_LOG1("Recreate replication slot for node %d after end of recovery", nodeId);
diff --git a/contrib/mmts/multimaster.h b/contrib/mmts/multimaster.h
@@ -117,7 +117,7 @@ typedef enum
 
 typedef enum
 {
-	REPLMODE_UNKNOWN,    /* receiver should wait */
+	REPLMODE_EXIT,       /* receiver should exit */
 	REPLMODE_RECOVERED,  /* recovery of node is completed so drop old slot and restart replication from the current position in WAL */
 	REPLMODE_RECOVERY,   /* perform recorvery of the node by applying all data from the slot from specified point */
 	REPLMODE_NORMAL      /* normal mode: use existed slot or create new one and start receiving data from it from the specified position */
@@ -245,7 +245,7 @@ extern csn_t MtmAssignCSN(void);
 extern csn_t MtmSyncClock(csn_t csn);
 extern void  MtmJoinTransaction(GlobalTransactionId* gtid, csn_t snapshot);
 extern void  MtmReceiverStarted(int nodeId);
-extern MtmReplicationMode MtmGetReplicationMode(int nodeId);
+extern MtmReplicationMode MtmGetReplicationMode(int nodeId, sig_atomic_t volatile* shutdown);
 extern void  MtmExecute(void* work, int size);
 extern void  MtmExecutor(int id, void* work, size_t size);
 extern void  MtmSendNotificationMessage(MtmTransState* ts, MtmMessageCode cmd);
diff --git a/contrib/mmts/pglogical_receiver.c b/contrib/mmts/pglogical_receiver.c
@@ -193,6 +193,7 @@ feTimestampDifference(int64 start_time, int64 stop_time,
 
 static char const* const MtmReplicationModeName[] = 
 {
+	"exit", 
 	"recovered", /* recovery of node is completed so drop old slot and restart replication from the current position in WAL */
 	"recovery",  /* perform recorvery of the node by applying all data from theslot from specified point */
 	"normal"     /* normal mode: use existed slot or create new one and start receiving data from it from the specified position */
@@ -251,9 +252,10 @@ pglogical_receiver_main(Datum main_arg)
 		 * Druing recovery we need to open only one replication slot from which node should receive all transactions.
 		 * Slots at other nodes should be removed 
 		 */
-		mode = MtmGetReplicationMode(nodeId);	
-		if (mode == REPLMODE_UNKNOWN) { 
-			continue;
+		mode = MtmGetReplicationMode(nodeId, &got_sigterm);	
+		if (mode == REPLMODE_EXIT) 
+		{ 
+			break;
 		}
 		count = Mtm->recoveryCount;
 		
