Add Windows build support: shared memory wiring, PGDLLEXPORT, subscriber tool fixes

pglogical.c

@@ -105,7 +105,7 @@ static bool subowner_missing = false;
 shmem_request_hook_type prev_shmem_request_hook = NULL;
 #endif
 
-void _PG_init(void);
+void PGDLLEXPORT _PG_init(void);
 void PGDLLEXPORT pglogical_supervisor_main(Datum main_arg);
 char *pglogical_extra_connection_options;
 
@@ -701,6 +701,15 @@ start_manager_workers(void)
 void
 pglogical_supervisor_main(Datum main_arg)
 {
+#if defined(WIN32) && PG_VERSION_NUM >= 150000
+	/* Don't overwrite if already set */
+	if (shmem_startup_hook != pglogical_worker_shmem_startup)
+	{
+		prev_shmem_startup_hook = shmem_startup_hook;
+		shmem_startup_hook = pglogical_worker_shmem_startup;
+	}
+#endif
+
 	/* Establish signal handlers. */
 	pqsignal(SIGTERM, handle_sigterm);
 	BackgroundWorkerUnblockSignals();
@@ -793,7 +802,7 @@ pglogical_temp_directory_assing_hook(const char *newval, void *extra)
 					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
 					 errmsg("could not locate temporary directory: %s\n",
 							!ret ? strerror(errno) : "")));
-			return false;
+			return;
 		}
 #endif
 
@@ -912,6 +921,14 @@ _PG_init(void)
 #if PG_VERSION_NUM >= 150000
 	prev_shmem_request_hook = shmem_request_hook;
 	shmem_request_hook = pglogical_worker_shmem_init;
+#ifdef WIN32
+	/* Don't overwrite if already set */
+	if (shmem_startup_hook != pglogical_worker_shmem_startup)
+	{
+		prev_shmem_startup_hook = shmem_startup_hook;
+		shmem_startup_hook = pglogical_worker_shmem_startup;
+	}
+#endif
 #else
 	pglogical_worker_shmem_init();
 #endif

pglogical_create_subscriber.c

@@ -24,7 +24,11 @@
 #include <time.h>
 #include <sys/types.h>
 #include <sys/wait.h>
+
+#ifndef WIN32
 #include <sys/stat.h>
+#endif
+
 #include <unistd.h>
 #include <stdlib.h>
 
@@ -485,7 +489,7 @@ main(int argc, char **argv)
 	 * Start subscriber node with pglogical disabled, and wait until it starts
 	 * accepting connections which means it has caught up to the restore point.
 	 */
-	pg_ctl_ret = run_pg_ctl("start -l \"pglogical_create_subscriber_postgres.log\" -o \"-c shared_preload_libraries=''\"");
+	pg_ctl_ret = run_pg_ctl("start -l \"pglogical_create_subscriber_postgres.log\" -o \"-c shared_preload_libraries=\"");
 	if (pg_ctl_ret != 0)
 		die(_("Postgres startup for restore point catchup failed with %d. See pglogical_create_subscriber_postgres.log."), pg_ctl_ret);
 
@@ -1856,6 +1860,10 @@ static char *
 generate_restore_point_name(void)
 {
 	char *rpn = malloc(NAMEDATALEN);
+#ifdef WIN32
+	snprintf(rpn, NAMEDATALEN-1, "pglogical_create_subscriber_%x", rand());
+#else
 	snprintf(rpn, NAMEDATALEN-1, "pglogical_create_subscriber_%lx", random());
+#endif
 	return rpn;
 }

pglogical_output.c

@@ -21,7 +21,7 @@
 
 PG_MODULE_MAGIC;
 
-extern void		_PG_output_plugin_init(OutputPluginCallbacks *cb);
+extern void		PGDLLEXPORT _PG_output_plugin_init(OutputPluginCallbacks *cb);
 
 void
 _PG_output_plugin_init(OutputPluginCallbacks *cb)

pglogical_output_plugin.c

@@ -44,7 +44,7 @@
 #include "replication/origin.h"
 #endif
 
-extern void		_PG_output_plugin_init(OutputPluginCallbacks *cb);
+extern void		PGDLLEXPORT _PG_output_plugin_init(OutputPluginCallbacks *cb);
 
 static void pg_decode_startup(LogicalDecodingContext * ctx,
 							  OutputPluginOptions *opt, bool is_init);

pglogical_sync.c

@@ -2087,7 +2087,7 @@ QuoteWindowsArgvElement(StringInfo cmdline, const char *arg, bool force)
  * (This should arguably be part of libpq_fe.c, but I didn't want to expand our
  * abuse of PqExpBuffer.)
  */
-static void
+void
 QuoteWindowsArgv(StringInfo cmdline, const char * argv[])
 {
 	/* argv0 is required */
@@ -2126,7 +2126,7 @@ static int
 exec_cmd_win32(const char *cmd, char *cmdargv[])
 {
 	BOOL					ret;
-	int						exitcode = -1;
+	DWORD					exitcode = -1;
 	PROCESS_INFORMATION 	pi;
 
 	elog(DEBUG1, "trying to launch \"%s\"", cmd);
@@ -2139,7 +2139,7 @@ exec_cmd_win32(const char *cmd, char *cmdargv[])
 
 		/* Deal with insane windows command line quoting */
 		initStringInfo(&cmdline);
-		QuoteWindowsArgv(&cmdline, cmdargv);
+		QuoteWindowsArgv(&cmdline, (const char **)cmdargv);
 
 		/* CreateProcess may scribble on the cmd string */
 		cmd_tmp = pstrdup(cmd);

pglogical_worker.c

@@ -45,15 +45,19 @@ static	List *signal_workers = NIL;
 
 volatile sig_atomic_t	got_SIGTERM = false;
 
-PGLogicalContext	   *PGLogicalCtx = NULL;
+PGDLLEXPORT PGLogicalContext	   *PGLogicalCtx = NULL;
 PGLogicalWorker		   *MyPGLogicalWorker = NULL;
 static uint16			MyPGLogicalWorkerGeneration;
 
 static bool xacthook_signal_workers = false;
 static bool xact_cb_installed = false;
 
 
+#ifdef WIN32
+shmem_startup_hook_type prev_shmem_startup_hook = NULL;
+#else
 static shmem_startup_hook_type prev_shmem_startup_hook = NULL;
+#endif
 
 static void pglogical_worker_detach(bool crash);
 static void wait_for_worker_startup(PGLogicalWorker *worker,
@@ -680,14 +684,31 @@ worker_shmem_size(int nworkers)
 /*
  * Init shmem needed for workers.
  */
+#ifdef WIN32
+void
+#else
 static void
+#endif
 pglogical_worker_shmem_startup(void)
 {
 	bool        found;
 	int			nworkers;
+#ifdef WIN32
+	static bool already_attached = false;
+#endif
 
+#ifdef WIN32
+	/* Chained hook must not recurse into our own hook on re-registration. */
+	if (prev_shmem_startup_hook && prev_shmem_startup_hook != pglogical_worker_shmem_startup)
+		prev_shmem_startup_hook();
+	/* Windows children re-initialize shared memory; init the context once. */
+	if (already_attached)
+		return;
+	already_attached = true;
+#else
 	if (prev_shmem_startup_hook != NULL)
 		prev_shmem_startup_hook();
+#endif
 
 	/*
 	 * This is kludge for Windows (Postgres does not define the GUC variable
@@ -700,6 +721,11 @@ pglogical_worker_shmem_startup(void)
 	PGLogicalCtx = ShmemInitStruct("pglogical_context",
 								   worker_shmem_size(nworkers), &found);
 
+#ifdef WIN32
+	if (!PGLogicalCtx)
+		return;
+#endif
+
 	if (!found)
 	{
 		PGLogicalCtx->lock = &(GetNamedLWLockTranche("pglogical"))->lock;

pglogical_worker.h

@@ -17,6 +17,13 @@
 
 #include "pglogical.h"
 
+#ifdef WIN32
+#include "storage/ipc.h"	/* defines shmem_startup_hook_type */
+
+extern void PGDLLEXPORT pglogical_worker_shmem_startup(void);
+extern PGDLLEXPORT shmem_startup_hook_type prev_shmem_startup_hook;
+#endif
+
 typedef enum {
 	PGLOGICAL_WORKER_NONE,		/* Unused slot. */
 	PGLOGICAL_WORKER_MANAGER,	/* Manager. */