Reorganize logical replication worker disconnect code

petere · petere · commit 6812330f1cc9 · 2017-06-01T23:16:20.000-04:00
Move the walrcv_disconnect() calls into the before_shmem_exit handler.
This makes sure the call is always made even during exit by signal, it
saves some duplicate code, and it makes the logic more similar to
walreceiver.c.

Author: Petr Jelinek &lt;petr.jelinek@2ndquadrant.com&gt;
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
@@ -614,7 +614,13 @@ logicalrep_launcher_onexit(int code, Datum arg)
 static void
 logicalrep_worker_onexit(int code, Datum arg)
 {
+	/* Disconnect gracefully from the remote side. */
+	if (wrconn)
+		walrcv_disconnect(wrconn);
+
 	logicalrep_worker_detach();
+
+	ApplyLauncherWakeup();
 }
 
 /* SIGTERM: set flag to exit at next convenient time */
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
@@ -140,7 +140,6 @@ finish_sync_worker(void)
 	CommitTransactionCommand();
 
 	/* Stop gracefully */
-	walrcv_disconnect(wrconn);
 	proc_exit(0);
 }
 
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
@@ -1329,7 +1329,6 @@ reread_subscription(void)
 				   "stop because the subscription was removed",
 				   MySubscription->name)));
 
-		walrcv_disconnect(wrconn);
 		proc_exit(0);
 	}
 
@@ -1344,7 +1343,6 @@ reread_subscription(void)
 				   "stop because the subscription was disabled",
 				   MySubscription->name)));
 
-		walrcv_disconnect(wrconn);
 		proc_exit(0);
 	}
 
@@ -1359,7 +1357,6 @@ reread_subscription(void)
 				   "restart because the connection information was changed",
 				   MySubscription->name)));
 
-		walrcv_disconnect(wrconn);
 		proc_exit(0);
 	}
 
@@ -1374,7 +1371,6 @@ reread_subscription(void)
 				   "restart because subscription was renamed",
 				   MySubscription->name)));
 
-		walrcv_disconnect(wrconn);
 		proc_exit(0);
 	}
 
@@ -1392,7 +1388,6 @@ reread_subscription(void)
 				   "restart because the replication slot name was changed",
 				   MySubscription->name)));
 
-		walrcv_disconnect(wrconn);
 		proc_exit(0);
 	}
 
@@ -1407,7 +1402,6 @@ reread_subscription(void)
 				   "restart because subscription's publications were changed",
 				   MySubscription->name)));
 
-		walrcv_disconnect(wrconn);
 		proc_exit(0);
 	}
 
@@ -1610,8 +1604,6 @@ ApplyWorkerMain(Datum main_arg)
 	/* Run the main loop. */
 	LogicalRepApplyLoop(origin_startpos);
 
-	walrcv_disconnect(wrconn);
-
 	/* We should only get here if we received SIGTERM */
 	proc_exit(0);
 }

Original file line number	Diff line number	Diff line change
`@@ -614,7 +614,13 @@ logicalrep_launcher_onexit(int code, Datum arg)`
`614`	`614`	`static void`
`615`	`615`	`logicalrep_worker_onexit(int code, Datum arg)`
`616`	`616`	`{`
	`617`	`+ /* Disconnect gracefully from the remote side. */`
	`618`	`+ if (wrconn)`
	`619`	`+ walrcv_disconnect(wrconn);`
	`620`	`+`
`617`	`621`	`logicalrep_worker_detach();`
	`622`	`+`
	`623`	`+ ApplyLauncherWakeup();`
`618`	`624`	`}`
`619`	`625`
`620`	`626`	`/* SIGTERM: set flag to exit at next convenient time */`
Original file line number	Diff line number	Diff line change
`@@ -140,7 +140,6 @@ finish_sync_worker(void)`
`140`	`140`	`CommitTransactionCommand();`
`141`	`141`
`142`	`142`	`/* Stop gracefully */`
`143`		`- walrcv_disconnect(wrconn);`
`144`	`143`	`proc_exit(0);`
`145`	`144`	`}`
`146`	`145`
Original file line number	Diff line number	Diff line change
`@@ -1329,7 +1329,6 @@ reread_subscription(void)`
`1329`	`1329`	`"stop because the subscription was removed",`
`1330`	`1330`	`MySubscription->name)));`
`1331`	`1331`
`1332`		`- walrcv_disconnect(wrconn);`
`1333`	`1332`	`proc_exit(0);`
`1334`	`1333`	`}`
`1335`	`1334`
`@@ -1344,7 +1343,6 @@ reread_subscription(void)`
`1344`	`1343`	`"stop because the subscription was disabled",`
`1345`	`1344`	`MySubscription->name)));`
`1346`	`1345`
`1347`		`- walrcv_disconnect(wrconn);`
`1348`	`1346`	`proc_exit(0);`
`1349`	`1347`	`}`
`1350`	`1348`
`@@ -1359,7 +1357,6 @@ reread_subscription(void)`
`1359`	`1357`	`"restart because the connection information was changed",`
`1360`	`1358`	`MySubscription->name)));`
`1361`	`1359`
`1362`		`- walrcv_disconnect(wrconn);`
`1363`	`1360`	`proc_exit(0);`
`1364`	`1361`	`}`
`1365`	`1362`
`@@ -1374,7 +1371,6 @@ reread_subscription(void)`
`1374`	`1371`	`"restart because subscription was renamed",`
`1375`	`1372`	`MySubscription->name)));`
`1376`	`1373`
`1377`		`- walrcv_disconnect(wrconn);`
`1378`	`1374`	`proc_exit(0);`
`1379`	`1375`	`}`
`1380`	`1376`
`@@ -1392,7 +1388,6 @@ reread_subscription(void)`
`1392`	`1388`	`"restart because the replication slot name was changed",`
`1393`	`1389`	`MySubscription->name)));`
`1394`	`1390`
`1395`		`- walrcv_disconnect(wrconn);`
`1396`	`1391`	`proc_exit(0);`
`1397`	`1392`	`}`
`1398`	`1393`
`@@ -1407,7 +1402,6 @@ reread_subscription(void)`
`1407`	`1402`	`"restart because subscription's publications were changed",`
`1408`	`1403`	`MySubscription->name)));`
`1409`	`1404`
`1410`		`- walrcv_disconnect(wrconn);`
`1411`	`1405`	`proc_exit(0);`
`1412`	`1406`	`}`
`1413`	`1407`
`@@ -1610,8 +1604,6 @@ ApplyWorkerMain(Datum main_arg)`
`1610`	`1604`	`/* Run the main loop. */`
`1611`	`1605`	`LogicalRepApplyLoop(origin_startpos);`
`1612`	`1606`
`1613`		`- walrcv_disconnect(wrconn);`
`1614`		`-`
`1615`	`1607`	`/* We should only get here if we received SIGTERM */`
`1616`	`1608`	`proc_exit(0);`
`1617`	`1609`	`}`