On Fri, Jul 31, 2026 at 4:19 AM Thomas Munro <[email protected]> wrote:
>
> [..v3..]
Hi Thomas,
I've wanted to somehow help the threads initiative for long time so I've
started playing with patchset, mainly with using multi-threaded pgbench
and pgbasebackup to get a basic feeling, and yay it works! I haven't catched
any problems so far with it, but I had some ideas when dealing with this:
so all of this is mostly about v3 0002+0007+0009:
1. Couldn't we have another (optional?) arg for pg_thrd_create() to issue
pthread_setname_np() from day 1? It would be nice to have something to see
which thread does what (similiar to setproctiltle()) or should we directly
embed something like pg_thrd_setname() that uses pthread_setname_np()?
E.g. @@ -7493,6 +7493,9 @@ threadRun(void *arg)
+ char thrid[20];
+ snprintf(thrid, sizeof(thrid), "pgbench thr%d", thread->tid);
+ pthread_setname_np(pg_thrd_current(), thrid);
Then we could use something like "ps -aeL -o tid,comm,args | grep bench"
or in GDB to see those threads. So problem seems to be that comm is not
often displayed and sometimes thread 0 (tid=pid) is the process itself
because e.g. in pgbench.c case main calls threadRun() directly too to
make it thread#0. Alternative we could make it conditional there in
threadRun() to bypass that if tid == pid...
BTW: I have found portable way of doing this in MySQL code, see [1]
2. In src/include/port/pg_threads.h shouldn't the enum be like below?
enum
{
pg_mtx_plain = pg_mtx_plain_impl,
- pg_mtx_recursive = pg_mtx_plain_impl,
+ pg_mtx_recursive = pg_mtx_recursive_impl,
?
3. Could we have PTHREAD_MUTEX_ERRORCHECK enabled by default in at least
assert builds? I'm not sure, but if we have stuff like that
pthread_mutexattr_t attr;
pthread_mutexattr_init(&attr);
pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_ERRORCHECK);
pthread_mutex_init(&mutex, &attr);
pthread_mutex_unlock(&mutex); // returns EPERM (unlocking unlocked)
pthread_mutex_lock(&mutex);
pthread_mutex_lock(&mutex); // returns EDEADLK (instead of deadlock)
Please see attached sample nanopatch, but this bring me to the next
problem:
3b.The problem with above is that e.g. double pg_mtx_lock() on same mutex in
with the patchset didn't abort with above add-on for #2 (ERRORCHECK),
because the remapping pg_threads layer does not trigger any Assert() or
we are not checking for any non-zero retcode at all, so which way it should
be ? (should check errors on every pg_mtx_* on every call site?
or should be that part of API to have code reuse?)
4. BTW: the v03-0009/pg_basebackup didn't want to apply due to the
introduction of g_parse_lsn() there in f31d6fbc31d3. Attached is simple
fixup patch. I was kind of interested in that to see how threads there
could be used in far future to unlock even more performance (but that would
have to occur probably after [2] with some paralell-backup redesign, but
that got me thinking on how we are going to be compatible with all this
stuff: liburing, pthreads one day) - frankley I couldn't think of any
issues. Anyway, thinking of basebackups, I've reminded myself that zstd
can already uses pthreads both on client (relevant to this $thread) and
server too, e.g:
pg_basebackup -c fast -v -Ft -D /tmp/full.tar -Z client-zstd:workers=2
but when thinking through all of this of my only worry would be, that
in such case with this patch applied and in far future I would be having
>100% CPU PID with multiple threads and without fix from #1 it would be
impossible to tell what is being bottlenecked? (saturated ZSTD threads or
now the the thread fetching the)
5. This is something that sent me to the land of doubt: when reviewing that
0009 for basebackup there's this change:
-static volatile sig_atomic_t bgchild_exited = false;
+static volatile bool wal_streamer_thread_exited = false;
It works here (x86_64), but is it safe/platform compatible? I've read a lot
about _Atomic / atomic_int / sig_atomic_t / pg_atomic_flag / stdatomic.h
patch of Your's in [3] and Greg even mention pg_atomic_bool by Heikki [4]
there, but the more I read the more confused I am, so any gudiance and
help please? :) (and could we maybe put some README to nearby API
implementaion to mention that for such usecase what should be used going
forward as solid point of reference? I would almost by defintion use
"sig_atomic_t" there for such case.
(it's similiar to size_t vs pgoff_t vs Size vs ...). The only thing I
believe right now that atomic_int store really disassembly down to xchgl'
instruction (sounds like it is safer?)
6. I was wondering shouldn't we have some stub (for now) to initialize the
whole thing just before first use, something like: pg_thrd_init().
Over time we
could place pthread_attr_setstacksize() there if necessary or some (frontend
for now?) stuff like even fprintf() to show some debug info.
-J.
[1] -
https://github.com/MariaDB/server/blob/bfabe0e53042d6a954c84b58a3c9eade794b9e90/mysys/my_thread_name.cc#L73
[2] -
https://www.postgresql.org/message-id/flat/cakzirmwww-hdc3b6erjb+pax7rnsbcqlheq1kdstf42cgur...@mail.gmail.com
[3] -
https://www.postgresql.org/message-id/CA%2BhUKGKfNuXYVKT7WPpKTNYTgPduzu0%3DG5yFEMju_4kbW0ybOQ%40mail.gmail.com
[4] -
https://www.postgresql.org/message-id/bb0ba423-816c-4e21-a40f-b1be13b54c5f%40iki.fi
From a4f3ca35712934a5a668ead82d6cd67e58e61d3c Mon Sep 17 00:00:00 2001
From: Thomas Munro <[email protected]>
Date: Wed, 1 Jul 2026 19:14:52 +1200
Subject: [PATCH vfixupv3] 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 | 259 ++++---------------
src/bin/pg_basebackup/t/010_pg_basebackup.pl | 4 +-
2 files changed, 57 insertions(+), 206 deletions(-)
diff --git a/src/bin/pg_basebackup/pg_basebackup.c b/src/bin/pg_basebackup/pg_basebackup.c
index c3b87a19e76..6c506fdfa95 100644
--- a/src/bin/pg_basebackup/pg_basebackup.c
+++ b/src/bin/pg_basebackup/pg_basebackup.c
@@ -37,6 +37,7 @@
#include "fe_utils/recovery_gen.h"
#include "getopt_long.h"
#include "libpq/protocol.h"
+#include "port/pg_threads_ext.h"
#include "receivelog.h"
#include "streamutil.h"
@@ -171,26 +172,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_PLAIN_INIT;
/* Contents of configuration file to be generated */
static PQExpBuffer recoveryconfcontents = NULL;
@@ -237,7 +227,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)
@@ -288,32 +278,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.
@@ -453,67 +417,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)
- {
- ssize_t nread;
- char xlogend[64] = {0};
-
- nread = read(bgpipe[0], xlogend, sizeof(xlogend) - 1);
- if (nread < 0)
- pg_fatal("could not read from ready pipe: %m");
-
- if (!pg_parse_lsn(xlogend, &xlogendptr))
- pg_fatal("could not parse write-ahead log location \"%s\"",
- xlogend);
- 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
}
/*
@@ -542,21 +466,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 */
@@ -581,22 +500,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;
}
@@ -619,6 +530,7 @@ StartLogStreamer(char *startpos, uint32 timeline, char *sysidentifier,
{
logstreamer_param *param;
char statusdir[MAXPGPATH];
+ int error;
param = pg_malloc0_object(logstreamer_param);
param->timeline = timeline;
@@ -633,12 +545,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)
@@ -711,29 +617,12 @@ 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 ((error = pg_thrd_create(&wal_streamer_thread,
+ LogStreamerMain,
+ param)) != pg_thrd_success)
+ pg_fatal("could not create background thread: %s",
+ pg_thrd_error_string_with_detail(error));
}
/*
@@ -1037,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);
@@ -2198,66 +2088,40 @@ BaseBackup(char *compression_algorithm, char *compression_detail,
exit(1);
}
- if (bgchild > 0)
+ if (includewal == STREAM_WAL)
{
-#ifndef WIN32
+ int error;
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;
-#endif
+ XLogRecPtr endptr;
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 (!pg_parse_lsn(xlogend, &xlogendptr))
+ if (!pg_parse_lsn(xlogend, &endptr))
pg_fatal("could not parse write-ahead log location \"%s\"",
xlogend);
- InterlockedIncrement(&has_xlogendptr);
+
+ /*
+ * XXX This could be done with atomics, once we can use those in
+ * frontend code.
+ */
+ pg_mtx_lock(&xlogendptr_lock);
+ xlogendptr = endptr;
+ 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 ((error = pg_thrd_join(wal_streamer_thread,
+ &status)) != pg_thrd_success)
+ pg_fatal("could not wait for child thread: %s",
+ pg_thrd_error_string_with_detail(error));
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 */
@@ -2794,19 +2658,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 2442131e179..62b94848e90 100644
--- a/src/bin/pg_basebackup/t/010_pg_basebackup.pl
+++ b/src/bin/pg_basebackup/t/010_pg_basebackup.pl
@@ -1062,8 +1062,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.43.0
From 74db4c3270928389b58e6d960958d2f594a0e29d Mon Sep 17 00:00:00 2001
From: Jakub Wartak <[email protected]>
Date: Tue, 8 Sep 2026 12:44:55 +0200
Subject: [PATCH videa1] partially working USE_ASSERT_CHECKING for pthreads
mutexes
---
src/include/port/pg_threads/map_pthread_ext.h | 5 +++++
src/port/pg_threads.c | 7 +++++++
2 files changed, 12 insertions(+)
diff --git a/src/include/port/pg_threads/map_pthread_ext.h b/src/include/port/pg_threads/map_pthread_ext.h
index 3c591f15b55..38b74832211 100644
--- a/src/include/port/pg_threads/map_pthread_ext.h
+++ b/src/include/port/pg_threads/map_pthread_ext.h
@@ -45,8 +45,13 @@ enum
#ifndef PG_THREADS_USE_THREADS_H
#define PG_CND_INIT_IMPL PTHREAD_COND_INITIALIZER
+#if defined(USE_ASSERT_CHECKING) && defined (PTHREAD_ERRORCHECK_MUTEX_INITIALIZER_NP)
+/* XXX: this seems to be glibc only */
+#define PG_MTX_PLAIN_INIT_IMPL PTHREAD_ERRORCHECK_MUTEX_INITIALIZER_NP
+#else
#define PG_MTX_PLAIN_INIT_IMPL PTHREAD_MUTEX_INITIALIZER
#endif
+#endif
#define PG_RWLOCK_INIT_IMPL PTHREAD_RWLOCK_INITIALIZER
/* pg_rwlock_t */
diff --git a/src/port/pg_threads.c b/src/port/pg_threads.c
index 07dc7fe0dd2..14ab6399d9d 100644
--- a/src/port/pg_threads.c
+++ b/src/port/pg_threads.c
@@ -797,6 +797,13 @@ pg_mtx_init_impl(pg_mtx_impl *mtx, int type)
pthread_mutexattr_init(&attr);
if (type & pg_mtx_recursive_impl)
pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_RECURSIVE);
+#ifdef USE_ASSERT_CHECKING
+ else
+ {
+ /* Requires mutex initialized with PTHREAD_ERRORCHECK_MUTEX_INITIALIZER */
+ pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_ERRORCHECK);
+ }
+#endif
result = pg_thrd_map(pthread_mutex_init(mtx, &attr));
pthread_mutexattr_destroy(&attr);
--
2.43.0
