Reorganize logical replication worker disconnect code

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 <petr.jelinek@2ndquadrant.com>

diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index b956052014b2e83eb3a1b4bfd75e0712073a7150..63d903ac0217281816074803b17d03f83f071959 100644 (file)
--- 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
index fe45fb88203fdb270881c3572446562ef0c2632c..515724e1026f05f9097f7d60f7788f077d4dafca 100644 (file)
--- 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
index c67720bd2fdc76a397fbd430cfeb69793e6b8099..ea3ba1d5b4858bb5ca8827eb2f360d1b1792e319 100644 (file)
--- 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);
 }