From ab6b0db829a7e56138fd18ba868efcffa11cd55c Mon Sep 17 00:00:00 2001 From: Thomas Munro 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 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 -#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 to be included by every translation - * unit. - * - *------------------------------------------------------------------------- - */ - -#ifndef PG_PTHREAD_H -#define PG_PTHREAD_H - -#include /* 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