WIP: Implement on-reload hooks to reload configuration on SIGHUP.

The configuration is written to file and can be edited with the new command
pgcopydb config set; now this command also signals the running process with
SIGHUP and the config file can be read and applied on-the-fly.

Author: dimitri

src/bin/pgcopydb/cli_clone_follow.c

@@ -7,6 +7,7 @@
 #include <errno.h>
 #include <getopt.h>
 #include <inttypes.h>
+#include <signal.h>
 #include <sys/wait.h>
 #include <unistd.h>
 
@@ -20,6 +21,7 @@
 #include "parsing_utils.h"
 #include "pgsql.h"
 #include "progress.h"
+#include "signals.h"
 #include "string_utils.h"
 #include "summary.h"
 
@@ -171,7 +173,7 @@ cli_clone(int argc, char **argv)
 	}
 
 	/* make sure all sub-processes are now finished */
-	success = success && copydb_wait_for_subprocesses(copySpecs.failFast);
+	success = success && copydb_wait_for_subprocesses(copySpecs.failFast, NULL);
 
 	if (!success)
 	{
@@ -307,7 +309,7 @@ clone_and_follow(CopyDataSpec *copySpecs)
 	}
 
 	/* make sure all sub-processes are now finished */
-	success = success && copydb_wait_for_subprocesses(copySpecs->failFast);
+	success = success && copydb_wait_for_subprocesses(copySpecs->failFast, NULL);
 
 	if (!success)
 	{
@@ -672,8 +674,8 @@ start_follow_process(CopyDataSpec *copySpecs, StreamSpecs *streamSpecs,
 
 
 /*
- * cli_clone_follow_wait_subprocesses waits until both sub-processes are
- * finished.
+ * cli_clone_follow_wait_subprocesses waits until given sub-processes has
+ * terminated, and return success when its return code was zero.
  */
 static bool
 cli_clone_follow_wait_subprocess(const char *name, pid_t pid)
@@ -689,6 +691,21 @@ cli_clone_follow_wait_subprocess(const char *name, pid_t pid)
 
 	while (!exited)
 	{
+		if (asked_to_reload)
+		{
+			log_warn("SIGHUP!");
+
+			/*
+			 * Forward the signal to all the sub-processes.
+			 *
+			 * Ignore errors and continue looping anyway, it just means we
+			 * failed to answer to the SIGHUP signal this once time.
+			 */
+			(void) signal_process_group(SIGHUP);
+
+			asked_to_reload = 0;
+		}
+
 		if (!follow_wait_pid(pid, &exited, &returnCode))
 		{
 			/* errors have already been logged */

src/bin/pgcopydb/cli_config.c

@@ -7,6 +7,7 @@
 #include <errno.h>
 #include <getopt.h>
 #include <inttypes.h>
+#include <signal.h>
 
 #include "cli_common.h"
 #include "cli_root.h"
@@ -15,6 +16,7 @@
 #include "commandline.h"
 #include "env_utils.h"
 #include "log.h"
+#include "pidfile.h"
 #include "parsing_utils.h"
 #include "string_utils.h"
 
@@ -368,14 +370,40 @@ cli_config_set(int argc, char **argv)
 		}
 
 		/* now read the value from just written file */
-		if (config_get_setting(&config, cfname, argv[0], value, BUFSIZE))
+		if (!config_get_setting(&config, cfname, argv[0], value, BUFSIZE))
 		{
-			fformat(stdout, "%s\n", value);
+			log_error("Failed to lookup option %s", argv[0]);
+			exit(EXIT_CODE_BAD_ARGS);
+		}
+
+		/* now signal the pgcopydb process to reload config (SIGHUP) */
+		if (file_exists(copySpecs.cfPaths.pidfile))
+		{
+			pid_t pid;
+
+			log_debug("Reading pidfile \"%s\"", copySpecs.cfPaths.pidfile);
+
+			if (!read_pidfile(copySpecs.cfPaths.pidfile, &pid))
+			{
+				exit(EXIT_CODE_INTERNAL_ERROR);
+			}
+
+			log_notice("Signaling process %d with SIGHUP", pid);
+
+			if (kill(pid, SIGHUP) != 0)
+			{
+				log_error("Failed to send SIGHUP signal to process %d: %m", pid);
+				exit(EXIT_CODE_INTERNAL_ERROR);
+			}
 		}
 		else
 		{
-			log_error("Failed to lookup option %s", argv[0]);
-			exit(EXIT_CODE_BAD_ARGS);
+			log_error("Failed to send SIGHUP: pidfile does not exists: \"%s\"",
+					  copySpecs.cfPaths.pidfile);
+			exit(EXIT_CODE_INTERNAL_ERROR);
 		}
+
+		/* finally output the value we just set and signaled */
+		fformat(stdout, "%s\n", value);
 	}
 }

src/bin/pgcopydb/cli_ping.c

@@ -285,7 +285,7 @@ cli_ping(int argc, char **argv)
 	 */
 	bool failFast = false;
 
-	if (!copydb_wait_for_subprocesses(failFast))
+	if (!copydb_wait_for_subprocesses(failFast, NULL))
 	{
 		/* errors have already been logged */
 		exit(EXIT_CODE_INTERNAL_ERROR);

src/bin/pgcopydb/copydb.c

@@ -968,7 +968,7 @@ copydb_fatal_exit()
 	 * non-zero return code.
 	 */
 	bool failFast = false;
-	return copydb_wait_for_subprocesses(failFast);
+	return copydb_wait_for_subprocesses(failFast, NULL);
 }
 
 
@@ -978,18 +978,30 @@ copydb_fatal_exit()
  * returns true only when all the subprocesses have returned zero (success).
  */
 bool
-copydb_wait_for_subprocesses(bool failFast)
+copydb_wait_for_subprocesses(bool failFast, OnReloadHook *onReloadHook)
 {
 	bool allReturnCodeAreZero = true;
 	log_debug("Waiting for sub-processes to finish");
 
 	for (;;)
 	{
-		int status;
-
 		/* ignore errors */
+		int status;
 		pid_t pid = waitpid(-1, &status, WNOHANG);
 
+		if (asked_to_reload)
+		{
+			log_warn("SIGHUP!");
+
+			/* call the reload hook when defined */
+			if (onReloadHook != NULL)
+			{
+				(void) (onReloadHook->fun)(onReloadHook->specs);
+			}
+
+			asked_to_reload = 0;
+		}
+
 		switch (pid)
 		{
 			case -1:

src/bin/pgcopydb/copydb.h

@@ -281,6 +281,15 @@ typedef enum
 } PostgresDumpSection;
 
 
+/* on-reload hooks */
+typedef void (*reloadHook) (CopyDataSpec *specs);
+
+typedef struct OnReloadHook
+{
+	CopyDataSpec *specs;
+	reloadHook fun;
+} OnReloadHook;
+
 extern GUC srcSettings95[];
 extern GUC srcSettings[];
 extern GUC dstSettings[];
@@ -329,7 +338,7 @@ bool copydb_init_tablepaths_for_part(CopyFilePaths *cfPaths,
 bool copydb_export_snapshot(TransactionSnapshot *snapshot);
 
 bool copydb_fatal_exit(void);
-bool copydb_wait_for_subprocesses(bool failFast);
+bool copydb_wait_for_subprocesses(bool failFast, OnReloadHook *onReloadHook);
 
 bool copydb_register_sysv_semaphore(SysVResArray *array, Semaphore *semaphore);
 bool copydb_register_sysv_queue(SysVResArray *array, Queue *queue);

src/bin/pgcopydb/follow.c

@@ -736,6 +736,12 @@ follow_wait_subprocesses(StreamSpecs *specs)
 			}
 		}
 
+		if (asked_to_reload)
+		{
+			log_warn("SIGHUP!");
+			asked_to_reload = 0;
+		}
+
 		/* re-init stillRunning at each iteration */
 		stillRunning = count;
 

src/bin/pgcopydb/indexes.c

@@ -591,7 +591,7 @@ copydb_start_index_processes(CopyDataSpec *specs,
 		}
 	}
 
-	bool success = copydb_wait_for_subprocesses(specs->failFast);
+	bool success = copydb_wait_for_subprocesses(specs->failFast, NULL);
 
 	/* and write that we successfully finished copying all tables */
 	if (!write_file("", 0, specs->cfPaths.done.indexes))

src/bin/pgcopydb/main.c

@@ -53,7 +53,11 @@ main(int argc, char **argv)
 	 * setpgrp and it has no effect when the calling process is a session
 	 * leader.
 	 */
-	setpgrp();
+	if (setpgid(0, 0) != 0)
+	{
+		log_fatal("Failed to set process group: %m");
+		exit(EXIT_CODE_INTERNAL_ERROR);
+	}
 
 	CommandLine command = root;
 

src/bin/pgcopydb/pidfile.c

@@ -39,8 +39,7 @@ create_pidfile(const char *pidfile, pid_t pid)
 {
 	PQExpBuffer content = createPQExpBuffer();
 
-
-	log_trace("create_pidfile(%d): \"%s\"", pid, pidfile);
+	log_notice("Creating pidfile for %d at \"%s\"", pid, pidfile);
 
 	if (content == NULL)
 	{

src/bin/pgcopydb/signals.c

@@ -268,3 +268,30 @@ signal_to_string(int signal)
 			return "unknown signal";
 	}
 }
+
+
+/*
+ * signal_forward_to_pgroup forwards given signal to our process group.
+ */
+bool
+signal_process_group(int sig)
+{
+	pid_t pid = getpid();
+	pid_t pgrp = getpgid(pid);
+
+	if (pgrp == -1)
+	{
+		/* continue looping anyway, ignore SIGHUP this time */
+		log_error("Failed to retrieve process group: %m");
+		log_error("Failed to process SIGHUP, see above for details");
+		return false;
+	}
+
+	if (killpg(pgrp, SIGHUP) != 0)
+	{
+		log_error("Failed to send SIGHUP signal to process group: %m");
+		return false;
+	}
+
+	return true;
+}