Support configuration reload in logical replication workers

Author: Michael Paquier <michael.paquier@gmail.com>
Reviewed-by: Petr Jelinek <petr.jelinek@2ndquadrant.com>
Reported-by: Fujii Masao <masao.fujii@gmail.com>

diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index fecff936c07c8ecf0bf9359d3667cb9c52c2a2f9..7ba239c02c19777a1728a23c536915be5e2f05ab 100644 (file)
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -75,7 +75,10 @@ LogicalRepCtxStruct *LogicalRepCtx;
 static void logicalrep_worker_onexit(int code, Datum arg);
 static void logicalrep_worker_detach(void);
 
-bool           got_SIGTERM = false;
+/* Flags set by signal handlers */
+volatile sig_atomic_t got_SIGHUP = false;
+volatile sig_atomic_t got_SIGTERM = false;
+
 static bool    on_commit_launcher_wakeup = false;
 
 Datum pg_stat_get_subscription(PG_FUNCTION_ARGS);
@@ -495,10 +498,28 @@ logicalrep_worker_onexit(int code, Datum arg)
 void
 logicalrep_worker_sigterm(SIGNAL_ARGS)
 {
+       int save_errno = errno;
+
        got_SIGTERM = true;
 
        /* Waken anything waiting on the process latch */
        SetLatch(MyLatch);
+
+       errno = save_errno;
+}
+
+/* SIGHUP: set flag to reload configuration at next convenient time */
+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;
 }
 
 /*
@@ -637,6 +658,7 @@ ApplyLauncherMain(Datum main_arg)
                        (errmsg("logical replication launcher started")));
 
        /* Establish signal handlers. */
+       pqsignal(SIGHUP, logicalrep_worker_sighup);
        pqsignal(SIGTERM, logicalrep_worker_sigterm);
        BackgroundWorkerUnblockSignals();
 
@@ -728,6 +750,12 @@ ApplyLauncherMain(Datum main_arg)
                if (rc & WL_POSTMASTER_DEATH)
                        proc_exit(1);
 
+               if (got_SIGHUP)
+               {
+                       got_SIGHUP = false;
+                       ProcessConfigFile(PGC_SIGHUP);
+               }
+
                ResetLatch(&MyProc->procLatch);
        }
 

diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 8a984a4150827ab7a0226c475832e8b498a53029..3313448e7b9a408d8902e20c1938dea94ed0db16 100644 (file)
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -1138,6 +1138,12 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
                if (rc & WL_POSTMASTER_DEATH)
                        proc_exit(1);
 
+               if (got_SIGHUP)
+               {
+                       got_SIGHUP = false;
+                       ProcessConfigFile(PGC_SIGHUP);
+               }
+
                if (rc & WL_TIMEOUT)
                {
                        /*
@@ -1441,6 +1447,7 @@ ApplyWorkerMain(Datum main_arg)
        logicalrep_worker_attach(worker_slot);
 
        /* Setup signal handling */
+       pqsignal(SIGHUP, logicalrep_worker_sighup);
        pqsignal(SIGTERM, logicalrep_worker_sigterm);
        BackgroundWorkerUnblockSignals();
 

diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 5bebca34ef86148294d0ea7a158724f61196c2e9..035467ce2464e2f9c32e98ed474eef7b49004baf 100644 (file)
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -56,7 +56,8 @@ extern Subscription              *MySubscription;
 extern LogicalRepWorker           *MyLogicalRepWorker;
 
 extern bool    in_remote_transaction;
-extern bool    got_SIGTERM;
+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,
@@ -69,6 +70,7 @@ 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);