v1-0006-pgbench-Use-pg_threads.h.patch

From ab6b0db829a7e56138fd18ba868efcffa11cd55c Mon Sep 17 00:00:00 2001
From: Thomas Munro <thomas.munro@gmail.com>
Date: Thu, 25 Jun 2026 14:42:31 +1200
Subject: [PATCH v1 06/14] pgbench: Use pg_threads.h.

* adopt pg_threads.h for threads, mutexes, barriers
* remove local threading API abstraction
* remove replacement pthread_barrier_t implementation

Discussion:
Reviewed-by:
---
 configure                       | 17 +-------
 configure.ac                    |  3 +-
 src/bin/pgbench/pgbench.c       | 67 ++++++----------------------
 src/include/port/pg_pthread.h   | 41 ------------------
 src/port/meson.build            |  4 --
 src/port/pthread_barrier_wait.c | 77 ---------------------------------
 6 files changed, 15 insertions(+), 194 deletions(-)
 delete mode 100644 src/include/port/pg_pthread.h
 delete mode 100644 src/port/pthread_barrier_wait.c

diff --git a/configure b/configure
index 35b0b72f0a7..b029844a5f4 100755
--- a/configure
+++ b/configure
@@ -15888,7 +15888,7 @@ fi
 LIBS_including_readline="$LIBS"
 LIBS=`echo "$LIBS" | sed -e 's/-ledit//g' -e 's/-lreadline//g'`
 
-for ac_func in backtrace_symbols copyfile copy_file_range elf_aux_info explicit_memset getauxval getifaddrs getpeerucred inet_pton kqueue localeconv_l mbstowcs_l memset_explicit posix_fallocate ppoll pthread_is_threaded_np setproctitle setproctitle_fast strsignal syncfs sync_file_range uselocale wcstombs_l
+for ac_func in backtrace_symbols copyfile copy_file_range elf_aux_info explicit_memset getauxval getifaddrs getpeerucred inet_pton kqueue localeconv_l mbstowcs_l memset_explicit posix_fallocate ppoll pthread_barrier_wait pthread_is_threaded_np setproctitle setproctitle_fast strsignal syncfs sync_file_range uselocale wcstombs_l
 do :
   as_ac_var=`$as_echo "ac_cv_func_$ac_func" | $as_tr_sh`
 ac_fn_c_check_func "$LINENO" "$ac_func" "$as_ac_var"
@@ -16563,21 +16563,6 @@ fi
 
 
 
-ac_fn_c_check_func "$LINENO" "pthread_barrier_wait" "ac_cv_func_pthread_barrier_wait"
-if test "x$ac_cv_func_pthread_barrier_wait" = xyes; then :
-  $as_echo "#define HAVE_PTHREAD_BARRIER_WAIT 1" >>confdefs.h
-
-else
-  case " $LIBOBJS " in
-  *" pthread_barrier_wait.$ac_objext "* ) ;;
-  *) LIBOBJS="$LIBOBJS pthread_barrier_wait.$ac_objext"
- ;;
-esac
-
-fi
-
-
-
 if test "$PORTNAME" = "win32" -o "$PORTNAME" = "cygwin"; then
 	# Cygwin and (apparently, based on test results) Mingw both
 	# have a broken strtof(), so substitute its implementation.
diff --git a/configure.ac b/configure.ac
index 0e624fe36b9..376652301e4 100644
--- a/configure.ac
+++ b/configure.ac
@@ -1865,6 +1865,7 @@ AC_CHECK_FUNCS(m4_normalize([
 	memset_explicit
 	posix_fallocate
 	ppoll
+	pthread_barrier_wait
 	pthread_is_threaded_np
 	setproctitle
 	setproctitle_fast
@@ -1925,8 +1926,6 @@ AC_REPLACE_FUNCS(m4_normalize([
 	timingsafe_bcmp
 ]))
 
-AC_REPLACE_FUNCS(pthread_barrier_wait)
-
 if test "$PORTNAME" = "win32" -o "$PORTNAME" = "cygwin"; then
 	# Cygwin and (apparently, based on test results) Mingw both
 	# have a broken strtof(), so substitute its implementation.
diff --git a/src/bin/pgbench/pgbench.c b/src/bin/pgbench/pgbench.c
index 8ab35a8c83e..1eb4c05057c 100644
--- a/src/bin/pgbench/pgbench.c
+++ b/src/bin/pgbench/pgbench.c
@@ -67,6 +67,7 @@
 #include "libpq-fe.h"
 #include "pgbench.h"
 #include "port/pg_bitutils.h"
+#include "port/pg_threads.h"
 #include "portability/instr_time.h"
 
 /* X/Open (XSI) requires <math.h> to provide M_PI, but core POSIX does not */
@@ -114,49 +115,6 @@ typedef struct socket_set
 
 #endif							/* POLL_USING_SELECT */
 
-/*
- * Multi-platform thread implementations
- */
-
-#ifdef WIN32
-/* Use Windows threads */
-#include <windows.h>
-#define GETERRNO() (_dosmaperr(GetLastError()), errno)
-#define THREAD_T HANDLE
-#define THREAD_FUNC_RETURN_TYPE unsigned
-#define THREAD_FUNC_RETURN return 0
-#define THREAD_FUNC_CC __stdcall
-#define THREAD_CREATE(handle, function, arg) \
-	((*(handle) = (HANDLE) _beginthreadex(NULL, 0, (function), (arg), 0, NULL)) == 0 ? errno : 0)
-#define THREAD_JOIN(handle) \
-	(WaitForSingleObject(handle, INFINITE) != WAIT_OBJECT_0 ? \
-	GETERRNO() : CloseHandle(handle) ? 0 : GETERRNO())
-#define THREAD_BARRIER_T SYNCHRONIZATION_BARRIER
-#define THREAD_BARRIER_INIT(barrier, n) \
-	(InitializeSynchronizationBarrier((barrier), (n), 0) ? 0 : GETERRNO())
-#define THREAD_BARRIER_WAIT(barrier) \
-	EnterSynchronizationBarrier((barrier), \
-								SYNCHRONIZATION_BARRIER_FLAGS_BLOCK_ONLY)
-#define THREAD_BARRIER_DESTROY(barrier)
-#else
-/* Use POSIX threads */
-#include "port/pg_pthread.h"
-#define THREAD_T pthread_t
-#define THREAD_FUNC_RETURN_TYPE void *
-#define THREAD_FUNC_RETURN return NULL
-#define THREAD_FUNC_CC
-#define THREAD_CREATE(handle, function, arg) \
-	pthread_create((handle), NULL, (function), (arg))
-#define THREAD_JOIN(handle) \
-	pthread_join((handle), NULL)
-#define THREAD_BARRIER_T pthread_barrier_t
-#define THREAD_BARRIER_INIT(barrier, n) \
-	pthread_barrier_init((barrier), NULL, (n))
-#define THREAD_BARRIER_WAIT(barrier) pthread_barrier_wait((barrier))
-#define THREAD_BARRIER_DESTROY(barrier) pthread_barrier_destroy((barrier))
-#endif
-
-
 /********************************************************************
  * some configurable parameters */
 
@@ -486,7 +444,7 @@ typedef enum TStatus
 static pg_prng_state base_random_sequence;
 
 /* Synchronization barrier for start and connection */
-static THREAD_BARRIER_T barrier;
+static pg_barrier_t barrier;
 
 /*
  * Connection state machine states.
@@ -654,7 +612,7 @@ typedef struct
 typedef struct
 {
 	int			tid;			/* thread id */
-	THREAD_T	thread;			/* thread handle */
+	pg_thrd_t	thread;			/* thread handle */
 	CState	   *state;			/* array of CState */
 	int			nstate;			/* length of state[] */
 
@@ -839,7 +797,7 @@ static void doLog(TState *thread, CState *st,
 static void processXactStats(TState *thread, CState *st, pg_time_usec_t *now,
 							 bool skipped, StatsData *agg);
 static void addScript(const ParsedScript *script);
-static THREAD_FUNC_RETURN_TYPE THREAD_FUNC_CC threadRun(void *arg);
+static int	threadRun(void *arg);
 static void finishCon(CState *st);
 static void setalarm(int seconds);
 static socket_set *alloc_socket_set(int count);
@@ -7445,7 +7403,7 @@ main(int argc, char **argv)
 	if (duration > 0)
 		setalarm(duration);
 
-	errno = THREAD_BARRIER_INIT(&barrier, nthreads);
+	errno = pg_barrier_init(&barrier, nthreads);
 	if (errno != 0)
 		pg_fatal("could not initialize barrier: %m");
 
@@ -7455,7 +7413,7 @@ main(int argc, char **argv)
 		TState	   *thread = &threads[i];
 
 		thread->create_time = pg_time_now();
-		errno = THREAD_CREATE(&thread->thread, threadRun, thread);
+		errno = pg_thrd_create(&thread->thread, threadRun, thread);
 
 		if (errno != 0)
 			pg_fatal("could not create thread: %m");
@@ -7478,7 +7436,7 @@ main(int argc, char **argv)
 		TState	   *thread = &threads[i];
 
 		if (i > 0)
-			THREAD_JOIN(thread->thread);
+			pg_thrd_join(thread->thread, NULL);
 
 		for (int j = 0; j < thread->nstate; j++)
 			if (thread->state[j].state != CSTATE_FINISHED)
@@ -7519,7 +7477,7 @@ main(int argc, char **argv)
 	printResults(&stats, pg_time_now() - bench_start, conn_total_duration,
 				 bench_start - start_time, latency_late);
 
-	THREAD_BARRIER_DESTROY(&barrier);
+	pg_barrier_destroy(&barrier);
 
 	if (exit_code != 0)
 		pg_log_error("Run was aborted; the above results are incomplete.");
@@ -7527,7 +7485,7 @@ main(int argc, char **argv)
 	return exit_code;
 }
 
-static THREAD_FUNC_RETURN_TYPE THREAD_FUNC_CC
+static int
 threadRun(void *arg)
 {
 	TState	   *thread = (TState *) arg;
@@ -7564,7 +7522,7 @@ threadRun(void *arg)
 		state[i].state = CSTATE_CHOOSE_SCRIPT;
 
 	/* READY */
-	THREAD_BARRIER_WAIT(&barrier);
+	pg_barrier_wait(&barrier);
 
 	thread_start = pg_time_now();
 	thread->started_time = thread_start;
@@ -7588,7 +7546,7 @@ threadRun(void *arg)
 	}
 
 	/* GO */
-	THREAD_BARRIER_WAIT(&barrier);
+	pg_barrier_wait(&barrier);
 
 	start = pg_time_now();
 	thread->bench_start = start;
@@ -7824,7 +7782,8 @@ done:
 		thread->logfile = NULL;
 	}
 	free_socket_set(sockets);
-	THREAD_FUNC_RETURN;
+
+	return 0;
 }
 
 static void
diff --git a/src/include/port/pg_pthread.h b/src/include/port/pg_pthread.h
deleted file mode 100644
index 4dbb3c8a52f..00000000000
--- a/src/include/port/pg_pthread.h
+++ /dev/null
@@ -1,41 +0,0 @@
-/*-------------------------------------------------------------------------
- *
- * Declarations for missing POSIX thread components.
- *
- *	  Currently this supplies an implementation of pthread_barrier_t for the
- *	  benefit of macOS, which lacks it.  These declarations are not in port.h,
- *	  because that'd require <pthread.h> to be included by every translation
- *	  unit.
- *
- *-------------------------------------------------------------------------
- */
-
-#ifndef PG_PTHREAD_H
-#define PG_PTHREAD_H
-
-#include <pthread.h>			/* IWYU pragma: export */
-
-#ifndef HAVE_PTHREAD_BARRIER_WAIT
-
-#ifndef PTHREAD_BARRIER_SERIAL_THREAD
-#define PTHREAD_BARRIER_SERIAL_THREAD (-1)
-#endif
-
-typedef struct pg_pthread_barrier
-{
-	bool		sense;			/* we only need a one bit phase */
-	int			count;			/* number of threads expected */
-	int			arrived;		/* number of threads that have arrived */
-	pthread_mutex_t mutex;
-	pthread_cond_t cond;
-} pthread_barrier_t;
-
-extern int	pthread_barrier_init(pthread_barrier_t *barrier,
-								 const void *attr,
-								 int count);
-extern int	pthread_barrier_wait(pthread_barrier_t *barrier);
-extern int	pthread_barrier_destroy(pthread_barrier_t *barrier);
-
-#endif
-
-#endif
diff --git a/src/port/meson.build b/src/port/meson.build
index a6ea79d3421..d8ca8af072d 100644
--- a/src/port/meson.build
+++ b/src/port/meson.build
@@ -79,10 +79,6 @@ replace_funcs_neg = [
   ['timingsafe_bcmp'],
 ]
 
-if host_system != 'windows'
-  replace_funcs_neg += [['pthread_barrier_wait']]
-endif
-
 # Replacement functionality to be built if corresponding configure symbol
 # is true
 replace_funcs_pos = [
diff --git a/src/port/pthread_barrier_wait.c b/src/port/pthread_barrier_wait.c
deleted file mode 100644
index caa1b12aaf3..00000000000
--- a/src/port/pthread_barrier_wait.c
+++ /dev/null
@@ -1,77 +0,0 @@
-/*-------------------------------------------------------------------------
- *
- * pthread_barrier_wait.c
- *    Implementation of pthread_barrier_t support for platforms lacking it.
- *
- * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
- *
- * IDENTIFICATION
- *    src/port/pthread_barrier_wait.c
- *
- *-------------------------------------------------------------------------
- */
-
-#include "c.h"
-
-#include "port/pg_pthread.h"
-
-int
-pthread_barrier_init(pthread_barrier_t *barrier, const void *attr, int count)
-{
-	int			error;
-
-	barrier->sense = false;
-	barrier->count = count;
-	barrier->arrived = 0;
-	if ((error = pthread_cond_init(&barrier->cond, NULL)) != 0)
-		return error;
-	if ((error = pthread_mutex_init(&barrier->mutex, NULL)) != 0)
-	{
-		pthread_cond_destroy(&barrier->cond);
-		return error;
-	}
-
-	return 0;
-}
-
-int
-pthread_barrier_wait(pthread_barrier_t *barrier)
-{
-	bool		initial_sense;
-
-	pthread_mutex_lock(&barrier->mutex);
-
-	/* We have arrived at the barrier. */
-	barrier->arrived++;
-	Assert(barrier->arrived <= barrier->count);
-
-	/* If we were the last to arrive, release the others and return. */
-	if (barrier->arrived == barrier->count)
-	{
-		barrier->arrived = 0;
-		barrier->sense = !barrier->sense;
-		pthread_mutex_unlock(&barrier->mutex);
-		pthread_cond_broadcast(&barrier->cond);
-
-		return PTHREAD_BARRIER_SERIAL_THREAD;
-	}
-
-	/* Wait for someone else to flip the sense. */
-	initial_sense = barrier->sense;
-	do
-	{
-		pthread_cond_wait(&barrier->cond, &barrier->mutex);
-	} while (barrier->sense == initial_sense);
-
-	pthread_mutex_unlock(&barrier->mutex);
-
-	return 0;
-}
-
-int
-pthread_barrier_destroy(pthread_barrier_t *barrier)
-{
-	pthread_cond_destroy(&barrier->cond);
-	pthread_mutex_destroy(&barrier->mutex);
-	return 0;
-}
-- 
2.47.3