Fix signal handling in logical replication workers

The logical replication worker processes now use the normal die()
handler for SIGTERM and CHECK_FOR_INTERRUPTS() instead of custom code.
One problem before was that the apply worker would not exit promptly
when a subscription was dropped, which could lead to deadlocks.

Author: Petr Jelinek <petr.jelinek@2ndquadrant.com>
Reported-by: Masahiko Sawada <sawada.mshk@gmail.com>

diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 63d903ac0217281816074803b17d03f83f071959..345a415212331ee2ea4f739df76ec23c75b7ca7a 100644 (file)
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -80,8 +80,8 @@ static void logicalrep_worker_detach(void);
 static void logicalrep_worker_cleanup(LogicalRepWorker *worker);
 
 /* Flags set by signal handlers */
-volatile sig_atomic_t got_SIGHUP = false;
-volatile sig_atomic_t got_SIGTERM = false;
+static volatile sig_atomic_t got_SIGHUP = false;
+static volatile sig_atomic_t got_SIGTERM = false;
 
 static bool on_commit_launcher_wakeup = false;
 
@@ -624,8 +624,8 @@ logicalrep_worker_onexit(int code, Datum arg)
 }
 
 /* SIGTERM: set flag to exit at next convenient time */
-void
-logicalrep_worker_sigterm(SIGNAL_ARGS)
+static void
+logicalrep_launcher_sigterm(SIGNAL_ARGS)
 {
        int                     save_errno = errno;
 
@@ -638,8 +638,8 @@ logicalrep_worker_sigterm(SIGNAL_ARGS)
 }
 
 /* SIGHUP: set flag to reload configuration at next convenient time */
-void
-logicalrep_worker_sighup(SIGNAL_ARGS)
+static void
+logicalrep_launcher_sighup(SIGNAL_ARGS)
 {
        int                     save_errno = errno;
 
@@ -799,8 +799,8 @@ ApplyLauncherMain(Datum main_arg)
        before_shmem_exit(logicalrep_launcher_onexit, (Datum) 0);
 
        /* Establish signal handlers. */
-       pqsignal(SIGHUP, logicalrep_worker_sighup);
-       pqsignal(SIGTERM, logicalrep_worker_sigterm);
+       pqsignal(SIGHUP, logicalrep_launcher_sighup);
+       pqsignal(SIGTERM, logicalrep_launcher_sigterm);
        BackgroundWorkerUnblockSignals();
 
        /* Make it easy to identify our processes. */

diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 515724e1026f05f9097f7d60f7788f077d4dafca..85e480db4bdd6c95eceafac6f42a9ce0b8112d19 100644 (file)
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -154,10 +154,12 @@ wait_for_sync_status_change(Oid relid, char origstate)
        int                     rc;
        char            state = origstate;
 
-       while (!got_SIGTERM)
+       for (;;)
        {
                LogicalRepWorker *worker;
 
+               CHECK_FOR_INTERRUPTS();
+
                LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
                worker = logicalrep_worker_find(MyLogicalRepWorker->subid,
                                                                                relid, false);
@@ -525,7 +527,7 @@ copy_read_data(void *outbuf, int minread, int maxread)
                bytesread += avail;
        }
 
-       while (!got_SIGTERM && maxread > 0 && bytesread < minread)
+       while (maxread > 0 && bytesread < minread)
        {
                pgsocket        fd = PGINVALID_SOCKET;
                int                     rc;
@@ -579,10 +581,6 @@ copy_read_data(void *outbuf, int minread, int maxread)
                ResetLatch(&MyProc->procLatch);
        }
 
-       /* Check for exit condition. */
-       if (got_SIGTERM)
-               proc_exit(0);
-
        return bytesread;
 }
 

diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index ea3ba1d5b4858bb5ca8827eb2f360d1b1792e319..e31551340c9266f601d4ad934c15c14c30737ebf 100644 (file)
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -72,6 +72,8 @@
 #include "storage/proc.h"
 #include "storage/procarray.h"
 
+#include "tcop/tcopprot.h"
+
 #include "utils/builtins.h"
 #include "utils/catcache.h"
 #include "utils/datum.h"
@@ -118,6 +120,9 @@ static void store_flush_position(XLogRecPtr remote_lsn);
 
 static void reread_subscription(void);
 
+/* Flags set by signal handlers */
+static volatile sig_atomic_t got_SIGHUP = false;
+
 /*
  * Should this worker apply changes for given relation.
  *
@@ -1005,7 +1010,7 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
        /* mark as idle, before starting to loop */
        pgstat_report_activity(STATE_IDLE, NULL);
 
-       while (!got_SIGTERM)
+       for (;;)
        {
                pgsocket        fd = PGINVALID_SOCKET;
                int                     rc;
@@ -1015,6 +1020,8 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
                TimestampTz last_recv_timestamp = GetCurrentTimestamp();
                bool            ping_sent = false;
 
+               CHECK_FOR_INTERRUPTS();
+
                MemoryContextSwitchTo(ApplyMessageContext);
 
                len = walrcv_receive(wrconn, &buf, &fd);
@@ -1437,6 +1444,19 @@ subscription_change_cb(Datum arg, int cacheid, uint32 hashvalue)
        MySubscriptionValid = false;
 }
 
+/* SIGHUP: set flag to reload configuration at next convenient time */
+static void
+logicalrep_worker_sighup(SIGNAL_ARGS)
+{
+       int                     save_errno = errno;
+
+       got_SIGHUP = true;
+
+       /* Waken anything waiting on the process latch */
+       SetLatch(MyLatch);
+
+       errno = save_errno;
+}
 
 /* Logical Replication Apply worker entry point */
 void
@@ -1454,7 +1474,7 @@ ApplyWorkerMain(Datum main_arg)
 
        /* Setup signal handling */
        pqsignal(SIGHUP, logicalrep_worker_sighup);
-       pqsignal(SIGTERM, logicalrep_worker_sigterm);
+       pqsignal(SIGTERM, die);
        BackgroundWorkerUnblockSignals();
 
        /* Initialise stats to a sanish value */
@@ -1604,6 +1624,14 @@ ApplyWorkerMain(Datum main_arg)
        /* Run the main loop. */
        LogicalRepApplyLoop(origin_startpos);
 
-       /* We should only get here if we received SIGTERM */
        proc_exit(0);
 }
+
+/*
+ * Is current process a logical replication worker?
+ */
+bool
+IsLogicalWorker(void)
+{
+       return MyLogicalRepWorker != NULL;
+}

diff --git a/src/backend/tcop/postgres.c b/src/backend/tcop/postgres.c
index 75c2d9a61d0dc067e9844f986cf23e23f724e565..13577691505b3654132ddd7177685f008cc0805b 100644 (file)
--- a/src/backend/tcop/postgres.c
+++ b/src/backend/tcop/postgres.c
@@ -55,6 +55,7 @@
 #include "pg_getopt.h"
 #include "postmaster/autovacuum.h"
 #include "postmaster/postmaster.h"
+#include "replication/logicalworker.h"
 #include "replication/slot.h"
 #include "replication/walsender.h"
 #include "rewrite/rewriteHandler.h"
@@ -2845,6 +2846,10 @@ ProcessInterrupts(void)
                        ereport(FATAL,
                                        (errcode(ERRCODE_ADMIN_SHUTDOWN),
                                         errmsg("terminating autovacuum process due to administrator command")));
+               else if (IsLogicalWorker())
+                       ereport(FATAL,
+                                       (errcode(ERRCODE_ADMIN_SHUTDOWN),
+                                        errmsg("terminating logical replication worker due to administrator command")));
                else if (RecoveryConflictPending && RecoveryConflictRetryable)
                {
                        pgstat_report_recovery_conflict(RecoveryConflictReason);

diff --git a/src/include/replication/logicalworker.h b/src/include/replication/logicalworker.h
index 3e0affa190b728b01adf0e90fe4ec062fd37c861..5877a930f68ca635c68c25fa031a28e09b824b80 100644 (file)
--- a/src/include/replication/logicalworker.h
+++ b/src/include/replication/logicalworker.h
@@ -14,4 +14,6 @@
 
 extern void ApplyWorkerMain(Datum main_arg);
 
+extern bool IsLogicalWorker(void);
+
 #endif   /* LOGICALWORKER_H */

diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 0654461305b33cb420f62d871b71eb88e2394254..2bfff5c1205dc499f845f2b725db56473773ab51 100644 (file)
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -67,8 +67,6 @@ extern Subscription *MySubscription;
 extern LogicalRepWorker *MyLogicalRepWorker;
 
 extern bool in_remote_transaction;
-extern volatile sig_atomic_t got_SIGHUP;
-extern volatile sig_atomic_t got_SIGTERM;
 
 extern void logicalrep_worker_attach(int slot);
 extern LogicalRepWorker *logicalrep_worker_find(Oid subid, Oid relid,
@@ -81,8 +79,6 @@ extern void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker);
 
 extern int     logicalrep_sync_worker_count(Oid subid);
 
-extern void logicalrep_worker_sighup(SIGNAL_ARGS);
-extern void logicalrep_worker_sigterm(SIGNAL_ARGS);
 extern char *LogicalRepSyncTableStart(XLogRecPtr *origin_startpos);
 void           process_syncing_tables(XLogRecPtr current_lsn);
 void invalidate_syncing_table_states(Datum arg, int cacheid,