v1-0008-pg_basebackup-Use-pg_threads.h.patch

From d1ecd166adf82227771d57756e17706172f32de0 Mon Sep 17 00:00:00 2001
From: Thomas Munro <thomas.munro@gmail.com>
Date: Wed, 1 Jul 2026 19:14:52 +1200
Subject: [PATCH v1 08/14] pg_basebackup: Use pg_threads.h.

Previously, Windows systems used a thread to stream WAL, and POSIX
systems used a subprocess.  Now all systems use a thread.

Discussion:
Reviewed-by:
---
 src/bin/pg_basebackup/pg_basebackup.c        | 252 ++++---------------
 src/bin/pg_basebackup/t/010_pg_basebackup.pl |   4 +-
 2 files changed, 49 insertions(+), 207 deletions(-)

diff --git a/src/bin/pg_basebackup/pg_basebackup.c b/src/bin/pg_basebackup/pg_basebackup.c
index c82852e6be0..c5074226931 100644
--- a/src/bin/pg_basebackup/pg_basebackup.c
+++ b/src/bin/pg_basebackup/pg_basebackup.c
@@ -36,6 +36,7 @@
 #include "fe_utils/recovery_gen.h"
 #include "getopt_long.h"
 #include "libpq/protocol.h"
+#include "port/pg_threads.h"
 #include "receivelog.h"
 #include "streamutil.h"
 
@@ -170,26 +171,15 @@ static uint64 totaldone;
 static int	tablespacecount;
 static char *progress_filename = NULL;
 
-/* Pipe to communicate with background wal receiver process */
-#ifndef WIN32
-static int	bgpipe[2] = {-1, -1};
-#endif
-
-/* Handle to child process */
-static pid_t bgchild = -1;
-static bool in_log_streamer = false;
+/* WAL streamer thread */
+static pg_thrd_t wal_streamer_thread;
 
-/* Flag to indicate if child process exited unexpectedly */
-static volatile sig_atomic_t bgchild_exited = false;
+/* Flag set by WAL streamer thread if it exits early. */
+static volatile bool wal_streamer_thread_exited = false;
 
 /* End position for xlog streaming, empty string if unknown yet */
 static XLogRecPtr xlogendptr;
-
-#ifndef WIN32
-static int	has_xlogendptr = 0;
-#else
-static volatile LONG has_xlogendptr = 0;
-#endif
+static pg_mtx_t xlogendptr_lock = PG_MTX_INIT;
 
 /* Contents of configuration file to be generated */
 static PQExpBuffer recoveryconfcontents = NULL;
@@ -236,7 +226,7 @@ static void tablespace_list_append(const char *arg);
 static void
 cleanup_directories_atexit(void)
 {
-	if (success || in_log_streamer)
+	if (success)
 		return;
 
 	if (!noclean && !checksum_failure)
@@ -287,32 +277,6 @@ disconnect_atexit(void)
 		PQfinish(conn);
 }
 
-#ifndef WIN32
-/*
- * If the bgchild exits prematurely and raises a SIGCHLD signal, we can abort
- * processing rather than wait until the backup has finished and error out at
- * that time. On Windows, we use a background thread which can communicate
- * without the need for a signal handler.
- */
-static void
-sigchld_handler(SIGNAL_ARGS)
-{
-	bgchild_exited = true;
-}
-
-/*
- * On windows, our background thread dies along with the process. But on
- * Unix, if we have started a subprocess, we want to kill it off so it
- * doesn't remain running trying to stream data.
- */
-static void
-kill_bgchild_atexit(void)
-{
-	if (bgchild > 0 && !bgchild_exited)
-		kill(bgchild, SIGTERM);
-}
-#endif
-
 /*
  * Split argument into old_dir and new_dir and append to tablespace mapping
  * list.
@@ -452,69 +416,27 @@ usage(void)
 
 
 /*
- * Called in the background process every time data is received.
- * On Unix, we check to see if there is any data on our pipe
- * (which would mean we have a stop position), and if it is, check if
- * it is time to stop.
- * On Windows, we are in a single process, so we can just check if it's
- * time to stop.
+ * Called in the background thread every time data is received.
  */
 static bool
 reached_end_position(XLogRecPtr segendpos, uint32 timeline,
 					 bool segment_finished)
 {
-	if (!has_xlogendptr)
-	{
-#ifndef WIN32
-		fd_set		fds;
-		struct timeval tv = {0};
-		int			r;
+	static bool have_xlogendptr = false;
 
-		/*
-		 * Don't have the end pointer yet - check our pipe to see if it has
-		 * been sent yet.
-		 */
-		FD_ZERO(&fds);
-		FD_SET(bgpipe[0], &fds);
-
-		r = select(bgpipe[0] + 1, &fds, NULL, NULL, &tv);
-		if (r == 1)
-		{
-			char		xlogend[64] = {0};
-			uint32		hi,
-						lo;
-
-			r = read(bgpipe[0], xlogend, sizeof(xlogend) - 1);
-			if (r < 0)
-				pg_fatal("could not read from ready pipe: %m");
-
-			if (sscanf(xlogend, "%X/%08X", &hi, &lo) != 2)
-				pg_fatal("could not parse write-ahead log location \"%s\"",
-						 xlogend);
-			xlogendptr = ((uint64) hi) << 32 | lo;
-			has_xlogendptr = 1;
+	/*
+	 * Wait until xlogendptr has been set by the main thread.  After it has
+	 * been set, it is safe to read it without a lock.
+	 */
+	if (!have_xlogendptr)
+	{
+		pg_mtx_lock(&xlogendptr_lock);
+		have_xlogendptr = xlogendptr != 0;
+		pg_mtx_unlock(&xlogendptr_lock);
 
-			/*
-			 * Fall through to check if we've reached the point further
-			 * already.
-			 */
-		}
-		else
-		{
-			/*
-			 * No data received on the pipe means we don't know the end
-			 * position yet - so just say it's not time to stop yet.
-			 */
+		/* If it's not set yet, we just go back and wait until it shows up. */
+		if (!have_xlogendptr)
 			return false;
-		}
-#else
-
-		/*
-		 * On win32, has_xlogendptr is set by the main thread, so if it's not
-		 * set here, we just go back and wait until it shows up.
-		 */
-		return false;
-#endif
 	}
 
 	/*
@@ -543,21 +465,16 @@ typedef struct
 } logstreamer_param;
 
 static int
-LogStreamerMain(logstreamer_param *param)
+LogStreamerMain(void *argument)
 {
+	logstreamer_param *param = argument;
 	StreamCtl	stream = {0};
 
-	in_log_streamer = true;
-
 	stream.startpos = param->startptr;
 	stream.timeline = param->timeline;
 	stream.sysidentifier = param->sysidentifier;
 	stream.stream_stop = reached_end_position;
-#ifndef WIN32
-	stream.stop_socket = bgpipe[0];
-#else
 	stream.stop_socket = PGINVALID_SOCKET;
-#endif
 	stream.standby_message_timeout = standby_message_timeout;
 	stream.synchronous = false;
 	/* fsync happens at the end of pg_basebackup for all data */
@@ -582,22 +499,14 @@ LogStreamerMain(logstreamer_param *param)
 		 * but we need to tell the parent that we didn't shutdown in a nice
 		 * way.
 		 */
-#ifdef WIN32
-		/*
-		 * In order to signal the main thread of an ungraceful exit we set the
-		 * same flag that we use on Unix to signal SIGCHLD.
-		 */
-		bgchild_exited = true;
-#endif
+		wal_streamer_thread_exited = true;
 		return 1;
 	}
 
 	if (!stream.walmethod->ops->finish(stream.walmethod))
 	{
 		pg_log_error("could not finish writing WAL files: %m");
-#ifdef WIN32
-		bgchild_exited = true;
-#endif
+		wal_streamer_thread_exited = true;
 		return 1;
 	}
 
@@ -637,12 +546,6 @@ StartLogStreamer(char *startpos, uint32 timeline, char *sysidentifier,
 	/* Round off to even segment position */
 	param->startptr -= XLogSegmentOffset(param->startptr, WalSegSz);
 
-#ifndef WIN32
-	/* Create our background pipe */
-	if (pipe(bgpipe) < 0)
-		pg_fatal("could not create pipe for background process: %m");
-#endif
-
 	/* Get a second connection */
 	param->bgconn = GetConnection();
 	if (!param->bgconn)
@@ -715,29 +618,11 @@ StartLogStreamer(char *startpos, uint32 timeline, char *sysidentifier,
 		}
 	}
 
-	/*
-	 * Start a child process and tell it to start streaming. On Unix, this is
-	 * a fork(). On Windows, we create a thread.
-	 */
-#ifndef WIN32
-	bgchild = fork();
-	if (bgchild == 0)
-	{
-		/* in child process */
-		exit(LogStreamerMain(param));
-	}
-	else if (bgchild < 0)
-		pg_fatal("could not create background process: %m");
-
-	/*
-	 * Else we are in the parent process and all is well.
-	 */
-	atexit(kill_bgchild_atexit);
-#else							/* WIN32 */
-	bgchild = _beginthreadex(NULL, 0, (void *) LogStreamerMain, param, 0, NULL);
-	if (bgchild == 0)
-		pg_fatal("could not create background thread: %m");
-#endif
+	/* Start a WAL streaming thread. */
+	if (pg_thrd_create(&wal_streamer_thread,
+					   LogStreamerMain,
+					   param) != pg_thrd_success)
+		pg_fatal("could not create background thread");
 }
 
 /*
@@ -1041,8 +926,9 @@ ReceiveCopyData(PGconn *conn, WriteDataCallback callback,
 			pg_fatal("could not read COPY data: %s",
 					 PQerrorMessage(conn));
 
-		if (bgchild_exited)
-			pg_fatal("background process terminated unexpectedly");
+		/* Periodic check for early exit of WAL streamer thread. */
+		if (wal_streamer_thread_exited)
+			pg_fatal("background thread terminated unexpectedly");
 
 		(*callback) (r, copybuf, callback_data);
 
@@ -2202,69 +2088,38 @@ BaseBackup(char *compression_algorithm, char *compression_detail,
 		exit(1);
 	}
 
-	if (bgchild > 0)
+	if (includewal == STREAM_WAL)
 	{
-#ifndef WIN32
 		int			status;
-		pid_t		r;
-#else
-		DWORD		status;
-
-		/*
-		 * get a pointer sized version of bgchild to avoid warnings about
-		 * casting to a different size on WIN64.
-		 */
-		intptr_t	bgchild_handle = bgchild;
 		uint32		hi,
 					lo;
-#endif
 
 		if (verbose)
-			pg_log_info("waiting for background process to finish streaming ...");
-
-#ifndef WIN32
-		if (write(bgpipe[1], xlogend, strlen(xlogend)) != strlen(xlogend))
-			pg_fatal("could not send command to background pipe: %m");
-
-		/* Just wait for the background process to exit */
-		r = waitpid(bgchild, &status, 0);
-		if (r == (pid_t) -1)
-			pg_fatal("could not wait for child process: %m");
-		if (r != bgchild)
-			pg_fatal("child %d died, expected %d", (int) r, (int) bgchild);
-		if (status != 0)
-			pg_fatal("%s", wait_result_to_str(status));
-		/* Exited normally, we're happy! */
-#else							/* WIN32 */
+			pg_log_info("waiting for WAL thread to finish streaming ...");
 
 		/*
-		 * On Windows, since we are in the same process, we can just store the
-		 * value directly in the variable, and then set the flag that says
-		 * it's there.
+		 * Since we are in the same process, we can just store the value
+		 * directly in the variable, and then set the flag that says it's
+		 * there.
 		 */
 		if (sscanf(xlogend, "%X/%08X", &hi, &lo) != 2)
 			pg_fatal("could not parse write-ahead log location \"%s\"",
 					 xlogend);
+
+		/*
+		 * XXX This could be done with atomics, once we can use those in
+		 * frontend code.
+		 */
+		pg_mtx_lock(&xlogendptr_lock);
 		xlogendptr = ((uint64) hi) << 32 | lo;
-		InterlockedIncrement(&has_xlogendptr);
+		pg_mtx_unlock(&xlogendptr_lock);
 
 		/* First wait for the thread to exit */
-		if (WaitForSingleObjectEx((HANDLE) bgchild_handle, INFINITE, FALSE) !=
-			WAIT_OBJECT_0)
-		{
-			_dosmaperr(GetLastError());
-			pg_fatal("could not wait for child thread: %m");
-		}
-		if (GetExitCodeThread((HANDLE) bgchild_handle, &status) == 0)
-		{
-			_dosmaperr(GetLastError());
-			pg_fatal("could not get child thread exit status: %m");
-		}
+		if (pg_thrd_join(wal_streamer_thread, &status) != pg_thrd_success)
+			pg_fatal("could not wait for child thread");
 		if (status != 0)
-			pg_fatal("child thread exited with error %u",
-					 (unsigned int) status);
+			pg_fatal("child thread exited with error %d", status);
 		/* Exited normally, we're happy */
-#endif
 	}
 
 	/* Free the configuration file contents */
@@ -2801,19 +2656,6 @@ main(int argc, char **argv)
 	}
 	atexit(disconnect_atexit);
 
-#ifndef WIN32
-
-	/*
-	 * Trap SIGCHLD to be able to handle the WAL stream process exiting. There
-	 * is no SIGCHLD on Windows, there we rely on the background thread
-	 * setting the signal variable on unexpected but graceful exit. If the WAL
-	 * stream thread crashes on Windows it will bring down the entire process
-	 * as it's a thread, so there is nothing to catch should that happen. A
-	 * crash on UNIX will be caught by the signal handler.
-	 */
-	pqsignal(SIGCHLD, sigchld_handler);
-#endif
-
 	/*
 	 * Set umask so that directories/files are created with the same
 	 * permissions as directories/files in the source data directory.
diff --git a/src/bin/pg_basebackup/t/010_pg_basebackup.pl b/src/bin/pg_basebackup/t/010_pg_basebackup.pl
index cfcfdb8b580..eab1fec7053 100644
--- a/src/bin/pg_basebackup/t/010_pg_basebackup.pl
+++ b/src/bin/pg_basebackup/t/010_pg_basebackup.pl
@@ -1048,8 +1048,8 @@ is( $node->poll_query_until(
 
 ok( pump_until(
 		$sigchld_bb, $sigchld_bb_timeout,
-		\$sigchld_bb_stderr, qr/background process terminated unexpectedly/),
-	'background process exit message');
+		\$sigchld_bb_stderr, qr/background thread terminated unexpectedly/),
+	'background thread exit message');
 $sigchld_bb->finish();
 
 # Test that we can back up an in-place tablespace
-- 
2.47.3