From ab649c89ca924c7aae39b56a3706d9f934e4e1ef Mon Sep 17 00:00:00 2001 From: Thomas Munro Date: Sat, 10 Jun 2023 09:14:07 +1200 Subject: [PATCH v3] Add port/pg_threads.h for a common threading API. Loosely based on a subset of C11's , but with pg_ prefixes, and some small additions: * static initializers for mutexes * read/write locks * barriers Clean up several several places that had to cope with POSIX and Windows threads, by adopting pg_threads.h: * pgbench had a lot of macros as a local abstraction * ecpg had a sort of pthreads emulation for Windows * libpq had a another pthread emulation for Windows One place in ecpg can be the first to use the new pg_thread_local storage class, which looks like a plain variable, because it has no destructor. Other places can use pg_tss_XXX(), which now runs destructors, even on Windows (which previously leaked memory when every thread exited). XXX To complete this, we need to make port/atomics.h available in frontend code, for the memory barrier macros. Reviewed-by: Andres Freund (earlier versions) Reviewed-by: Heikki Linnakangas Discussion: https://postgr.es/m/CA%2BhUKGLtmexrpMtxBRLCVePqV_dtWG-ZsEbyPrYc%2BNBB2TkNsw%40mail.gmail.com --- configure | 17 +- configure.ac | 3 +- src/bin/pgbench/pgbench.c | 67 +-- src/include/port/pg_pthread.h | 41 -- src/include/port/pg_threads.h | 542 ++++++++++++++++++ src/interfaces/ecpg/ecpglib/connect.c | 53 +- src/interfaces/ecpg/ecpglib/descriptor.c | 14 +- src/interfaces/ecpg/ecpglib/ecpglib_extern.h | 2 +- src/interfaces/ecpg/ecpglib/memory.c | 14 +- src/interfaces/ecpg/ecpglib/misc.c | 87 +-- src/interfaces/ecpg/ecpglib/sqlda.c | 1 - .../ecpg/include/ecpg-pthread-win32.h | 49 -- src/interfaces/ecpg/include/ecpg_config.h.in | 3 + src/interfaces/ecpg/include/meson.build | 1 + .../ecpg/test/expected/thread-alloc.c | 73 +-- .../ecpg/test/expected/thread-descriptor.c | 52 +- .../ecpg/test/expected/thread-prep.c | 99 ++-- .../ecpg/test/expected/thread-thread.c | 92 ++- .../test/expected/thread-thread_implicit.c | 92 ++- src/interfaces/ecpg/test/thread/alloc.pgc | 39 +- .../ecpg/test/thread/descriptor.pgc | 38 +- src/interfaces/ecpg/test/thread/prep.pgc | 37 +- src/interfaces/ecpg/test/thread/thread.pgc | 38 +- .../ecpg/test/thread/thread_implicit.pgc | 38 +- src/interfaces/libpq/Makefile | 1 - src/interfaces/libpq/fe-connect.c | 13 +- src/interfaces/libpq/fe-secure-openssl.c | 45 +- src/interfaces/libpq/fe-secure.c | 4 +- src/interfaces/libpq/libpq-int.h | 5 - src/interfaces/libpq/meson.build | 5 +- src/interfaces/libpq/pthread-win32.c | 66 --- src/port/Makefile | 1 + src/port/meson.build | 5 +- src/port/pg_threads.c | 436 ++++++++++++++ src/port/pthread-win32.h | 31 - src/port/pthread_barrier_wait.c | 77 --- src/tools/pginclude/headerscheck | 1 - src/tools/pgindent/typedefs.list | 9 + 38 files changed, 1255 insertions(+), 936 deletions(-) delete mode 100644 src/include/port/pg_pthread.h create mode 100644 src/include/port/pg_threads.h delete mode 100644 src/interfaces/ecpg/include/ecpg-pthread-win32.h delete mode 100644 src/interfaces/libpq/pthread-win32.c create mode 100644 src/port/pg_threads.c delete mode 100644 src/port/pthread-win32.h delete mode 100644 src/port/pthread_barrier_wait.c diff --git a/configure b/configure index 2abbeb2794..8cd797a3cb 100755 --- a/configure +++ b/configure @@ -15232,7 +15232,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 getifaddrs getpeerucred inet_pton kqueue mbstowcs_l memset_s posix_fallocate ppoll pthread_is_threaded_np setproctitle setproctitle_fast strchrnul strsignal syncfs sync_file_range uselocale wcstombs_l +for ac_func in backtrace_symbols copyfile copy_file_range getifaddrs getpeerucred inet_pton kqueue mbstowcs_l memset_s posix_fallocate ppoll pthread_barrier_wait pthread_is_threaded_np setproctitle setproctitle_fast strchrnul 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" @@ -15920,21 +15920,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 c46ed2c591..59518a6335 100644 --- a/configure.ac +++ b/configure.ac @@ -1740,6 +1740,7 @@ AC_CHECK_FUNCS(m4_normalize([ memset_s posix_fallocate ppoll + pthread_barrier_wait pthread_is_threaded_np setproctitle setproctitle_fast @@ -1799,8 +1800,6 @@ AC_REPLACE_FUNCS(m4_normalize([ strsep ])) -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 61618f2e18..a47a34f651 100644 --- a/src/bin/pgbench/pgbench.c +++ b/src/bin/pgbench/pgbench.c @@ -66,6 +66,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 */ @@ -113,49 +114,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 */ @@ -478,7 +436,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. @@ -646,7 +604,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[] */ @@ -830,7 +788,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); @@ -7311,7 +7269,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"); @@ -7321,7 +7279,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"); @@ -7344,7 +7302,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) @@ -7384,7 +7342,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."); @@ -7392,7 +7350,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; @@ -7429,7 +7387,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; @@ -7453,7 +7411,7 @@ threadRun(void *arg) } /* GO */ - THREAD_BARRIER_WAIT(&barrier); + pg_barrier_wait(&barrier); start = pg_time_now(); thread->bench_start = start; @@ -7689,7 +7647,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 d102ce9d6f..0000000000 --- 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 - -#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/include/port/pg_threads.h b/src/include/port/pg_threads.h new file mode 100644 index 0000000000..2425d5cb1f --- /dev/null +++ b/src/include/port/pg_threads.h @@ -0,0 +1,542 @@ +/*------------------------------------------------------------------------- + * + * pg_threads.h + * Portable multi-threading API. + * + * A multi-threading API abstraction loosely based on a subset C11 + * standard's header. The identifiers have a pg_ prefix. + * + * We have some extensions of our own, not present in C11: + * + * - pg_rwlock_t for read/write locks + * - pg_mtx_t static initializer PG_MTX_STATIC_INIT + * - pg_barrier_t + * + * Portions Copyright (c) 1996-2024, PostgreSQL Global Development Group + * + * IDENTIFICATION + * src/port/pg_threads.c + * + *------------------------------------------------------------------------- + */ + +#ifndef PG_THREADS_H +#define PG_THREADS_H + +#ifdef WIN32 +/* + * We use the macro PG_THREADS_WIN32 rather than WIN32 directly, because we + * might want to use the C11 APIs in Visual Studio 2022+ at some point. + * While using Windows native APIs, need an in-house implementation of TSS + * destructors, which we also gate separately so that it can be + * tested/maintained on other OSes too. + */ +#define PG_THREADS_WIN32 +#define PG_THREADS_NEED_DESTRUCTOR_TABLE +#endif + +/* + * To test our own destructor mechanism on POSIX systems, for the + * benefit of developers maintaining it, define this macro. + */ +/* #define PG_THREADS_NEED_DESTRUCTOR_TABLE */ + +#if defined(PG_THREADS_WIN32) +#include +#else +#include +#endif + + +/*------------------------------------------------------------------------- + * + * Thread-local storage class. This is a C11 language feature, not a + * library feature. We don't require C11, but we expect compilers to + * provide some way to request thread-local storage. (See also + * pg_tss_t, which is similar but uses explicit set/get functions and + * supports destructor function that are called at thread exit.) + * + *------------------------------------------------------------------------- + */ + +#if defined(_MSC_VER) +/* MSVC */ +#define pg_thread_local __declspec(thread) +#elif defined(__GNUC__) || defined(__INTEL_COMPILER) || defined(__SUNPRO_C) +/* GCC, Clang, Intel C, XLC, Solaris Studio */ +#define pg_thread_local __thread +#else +#error "no known thread_local storage class for this compiler" +#endif + + +/*------------------------------------------------------------------------- + * + * Return values. + * + *------------------------------------------------------------------------- + */ + +typedef enum pg_thrd_error_t +{ + pg_thrd_success = 0, + pg_thrd_nomem = 1, + pg_thrd_timedout = 2, + pg_thrd_busy = 3, + pg_thrd_error = 4, + + /* Not from C11. Needed by our pg_barrier_wait(). */ + pg_thrd_success_last = 5, +} pg_thrd_error_t; + +static inline int +pg_thrd_maperror(int error) +{ +#ifdef PG_THREADS_WIN32 + return error ? pg_thrd_success : pg_thrd_error; +#else + return error == 0 ? pg_thrd_success : pg_thrd_error; +#endif +} + + +/*------------------------------------------------------------------------- + * + * Threads. + * + *------------------------------------------------------------------------- + */ + +#ifdef PG_THREADS_WIN32 +typedef HANDLE pg_thrd_t; +#else +typedef pthread_t pg_thrd_t; +#endif + +typedef int (*pg_thrd_start_t) (void *); + +extern int pg_thrd_create(pg_thrd_t *thread, pg_thrd_start_t function, void *argument); +extern int pg_thrd_join(pg_thrd_t thread, int *result); +extern void pg_thrd_exit(int result); + +static inline pg_thrd_t +pg_thrd_current(void) +{ +#ifdef PG_THREADS_WIN32 + return GetCurrentThreadId(); +#else + return pthread_self(); +#endif +} + +static inline int +pg_thrd_equal(pg_thrd_t lhs, pg_thrd_t rhs) +{ +#ifdef PG_THREADS_WIN32 + return lhs == rhs; +#else + return pthread_equal(lhs, rhs); +#endif +} + + +/*------------------------------------------------------------------------- + * + * Initialization functions. + * + *------------------------------------------------------------------------- + */ + +#ifdef PG_THREADS_WIN32 +typedef INIT_ONCE pg_once_flag; +#define PG_ONCE_FLAG_INIT INIT_ONCE_STATIC_INIT +#else +typedef pthread_once_t pg_once_flag; +#define PG_ONCE_FLAG_INIT PTHREAD_ONCE_INIT +#endif + +typedef void (*pg_call_once_function_t) (void); + +#ifdef PG_THREADS_WIN32 +extern BOOL CALLBACK pg_call_once_trampoline(pg_once_flag *flag, + void *parameter, + void **context); +#endif + +static inline void +pg_call_once(pg_once_flag *flag, pg_call_once_function_t function) +{ +#ifdef PG_THREADS_WIN32 + InitOnceExecuteOnce(flag, pg_call_once_trampoline, (void *) function, NULL); +#else + pthread_once(flag, function); +#endif +} + + +/*------------------------------------------------------------------------- + * + * Thread-specific storage. This mechanism is an alternative to using + * the pg_thread_local storage class, which should be preferred where + * possible. The only advantage is that the TSS interface allows a + * destructor functions to be run for non-NULL values when each thread + * exits. + * + *------------------------------------------------------------------------- + */ + +#ifdef PG_THREADS_WIN32 +typedef DWORD pg_tss_t; +#else +typedef pthread_key_t pg_tss_t; +#endif + +typedef void (*pg_tss_dtor_t) (void *); + +/* + * How long before we give up trying to call all the registered + * destructors, if the destructors themselves are calling pg_tss_set() + * to befuddle us by storing new non-NULL values? + */ +#ifdef PG_THREADS_NEED_DESTRUCTOR_TABLE +#define PG_TSS_DTOR_ITERATIONS 8 +#else +#define PG_TSS_DTOR_ITERATIONS PTHREAD_DESTRUCTOR_ITERATIONS +#endif + +extern int pg_tss_create(pg_tss_t *tss_id, pg_tss_dtor_t destructor); +extern void pg_tss_dtor_delete(pg_tss_t tss_id); +#ifdef PG_THREADS_NEED_DESTRUCTOR_TABLE +extern void pg_tss_ensure_destructors_in_this_thread(void); +#endif + +static inline void * +pg_tss_get(pg_tss_t key) +{ +#ifdef PG_THREADS_WIN32 + return TlsGetValue(key); +#else + return pthread_getspecific(key); +#endif +} + +static inline int +pg_tss_set(pg_tss_t tss_id, void *value) +{ +#ifdef PG_THREADS_NEED_DESTRUCTOR_TABLE + if (value) + pg_tss_ensure_destructors_in_this_thread(); +#endif + +#ifdef PG_THREADS_WIN32 + return pg_thrd_maperror(TlsSetValue(tss_id, value)); +#else + return pg_thrd_maperror(pthread_setspecific(tss_id, value)); +#endif +} + + +/*------------------------------------------------------------------------- + * + * Read/write locks. Not in C11. + * + * Unfortunately Windows makes you say whether you're unlocking a read lock or + * a write lock, so we have to expose that here too. POSIX already knows. + * + *------------------------------------------------------------------------- + */ + +#ifdef PG_THREADS_WIN32 +typedef SRWLOCK pg_rwlock_t; +#define PG_RWLOCK_STATIC_INIT SRWLOCK_INIT +#else +typedef pthread_rwlock_t pg_rwlock_t; +#define PG_RWLOCK_STATIC_INIT PTHREAD_RWLOCK_INITIALIZER +#endif + +static inline int +pg_rwlock_init(pg_rwlock_t * lock) +{ +#ifdef PG_THREADS_WIN32 + InitializeSRWLock(lock); + return pg_thrd_success; +#else + return pg_thrd_maperror(pthread_rwlock_init(lock, NULL)); +#endif +} + +static inline int +pg_rwlock_rdlock(pg_rwlock_t * lock) +{ +#ifdef PG_THREADS_WIN32 + AcquireSRWLockShared(lock); + return pg_thrd_success; +#else + return pg_thrd_maperror(pthread_rwlock_rdlock(lock)); +#endif +} + +static inline int +pg_rwlock_wrlock(pg_rwlock_t * lock) +{ +#ifdef PG_THREADS_WIN32 + AcquireSRWLockExclusive(lock); + return pg_thrd_success; +#else + return pg_thrd_maperror(pthread_rwlock_wrlock(lock)); +#endif +} + +static inline int +pg_wrlock_unlock(pg_rwlock_t * lock) +{ +#ifdef PG_THREADS_WIN32 + ReleaseSRWLockExclusive(lock); + return pg_thrd_success; +#else + return pg_thrd_maperror(pthread_rwlock_unlock(lock)); +#endif +} + +static inline int +pg_rdlock_unlock(pg_rwlock_t * lock) +{ +#ifdef PG_THREADS_WIN32 + ReleaseSRWLockShared(lock); + return pg_thrd_success; +#else + return pg_thrd_maperror(pthread_rwlock_unlock(lock)); +#endif +} + + +/*------------------------------------------------------------------------- + * + * Simple mutexes. + * + *------------------------------------------------------------------------- + */ + +#ifdef PG_THREADS_WIN32 +/* + * CRITICAL_SECTION might be the most obvious Windows mechanism for + * pg_mtx_t, but SRWLock is reported to be at least as fast when used + * only in exclusive mode, and has the advantage of a static + * initializer (CRITICAL_SECTION must be initialized and destroyed + * explicitly because it allocates resources other than the space it + * occupies.) C11 doesn't define a static initializer (possibly + * because CRITICAL_SECTION doesn't?), but we want one anyway. So + * we'll just point pg_mtx_t to pg_rwlock_t. + */ +typedef pg_rwlock_t pg_mtx_t; +#define PG_MTX_STATIC_INIT PG_RWLOCK_STATIC_INIT +#else +typedef pthread_mutex_t pg_mtx_t; +#define PG_MTX_STATIC_INIT PTHREAD_MUTEX_INITIALIZER +#endif + +typedef enum pg_mtx_type_t +{ + pg_mtx_plain = 0 +} pg_mtx_type_t; + + +static inline int +pg_mtx_init(pg_mtx_t *mutex, int type) +{ +#ifdef PG_THREADS_WIN32 + return pg_rwlock_init(mutex); +#else + return pg_thrd_maperror(pthread_mutex_init(mutex, NULL)); +#endif +} + +static inline int +pg_mtx_lock(pg_mtx_t *mutex) +{ +#ifdef PG_THREADS_WIN32 + return pg_rwlock_wrlock(mutex); +#else + return pg_thrd_maperror(pthread_mutex_lock(mutex)); +#endif +} + +static inline int +pg_mtx_unlock(pg_mtx_t *mutex) +{ +#ifdef PG_THREADS_WIN32 + return pg_wrlock_unlock(mutex); +#else + return pg_thrd_maperror(pthread_mutex_unlock(mutex)); +#endif +} + +static inline int +pg_mtx_destroy(pg_mtx_t *mutex) +{ +#ifdef PG_THREADS_WIN32 + return pg_thrd_success; +#else + return pg_thrd_maperror(pthread_mutex_destroy(mutex)); +#endif +} + + +/*------------------------------------------------------------------------- + * + * Condition variables. + * + *------------------------------------------------------------------------- + */ + +#ifdef PG_THREADS_WIN32 +typedef CONDITION_VARIABLE pg_cnd_t; +#else +typedef pthread_cond_t pg_cnd_t; +#endif + +static inline int +pg_cnd_init(pg_cnd_t *condvar) +{ +#ifdef PG_THREADS_WIN32 + InitializeConditionVariable(condvar); + return pg_thrd_success; +#else + return pg_thrd_maperror(pthread_cond_init(condvar, NULL)); +#endif +} + +static inline int +pg_cnd_broadcast(pg_cnd_t *condvar) +{ +#ifdef PG_THREADS_WIN32 + WakeAllConditionVariable(condvar); + return pg_thrd_success; +#else + return pg_thrd_maperror(pthread_cond_broadcast(condvar)); +#endif +} + +static inline int +pg_cnd_wait(pg_cnd_t *condvar, pg_mtx_t *mutex) +{ +#ifdef PG_THREADS_WIN32 + SleepConditionVariableSRW(condvar, mutex, INFINITE, 0); + return pg_thrd_success; +#else + return pg_thrd_maperror(pthread_cond_wait(condvar, mutex)); +#endif +} + +static inline int +pg_cnd_destroy(pg_cnd_t *condvar) +{ +#ifdef PG_THREADS_WIN32 + return pg_thrd_success; +#else + return pg_thrd_maperror(pthread_cond_destroy(condvar)); +#endif +} + + +/*------------------------------------------------------------------------- + * + * Barriers. Not in C11. Apple currently lacks the POSIX version. + * We assume that the OS might know a better way to implement it that + * we do, so we only provide our own if we have to. + * + *------------------------------------------------------------------------- + */ + +#ifdef PG_THREADS_WIN32 +typedef SYNCHRONIZATION_BARRIER pg_barrier_t; +#elif defined(HAVE_PTHREAD_BARRIER) +typedef pthread_barrier_t pg_barrier_t; +#else +typedef struct pg_barrier_t +{ + bool sense; + int expected; + int arrived; + pg_mtx_t mutex; + pg_cnd_t cond; +} pg_barrier_t; +#endif + +static inline int +pg_barrier_init(pg_barrier_t *barrier, int count) +{ +#ifdef PG_THREADS_WIN32 + return pg_thrd_maperror(InitializeSynchronizationBarrier(barrier, count, 0)); +#elif defined(HAVE_PTHREAD_BARRIER) + return pg_thrd_maperror(pthread_barrier_init(barrier, NULL, count)); +#else + barrier->sense = false; + barrier->expected = count; + barrier->arrived = 0; + if (pg_cnd_init(&barrier->cond) != pg_thrd_success) + return pg_thrd_error; + if (pg_mtx_init(&barrier->mutex, pg_mtx_plain) != pg_thrd_success) + { + pg_cnd_destroy(&barrier->cond); + return pg_thrd_error; + } + return pg_thrd_success; +#endif +} + +static inline int +pg_barrier_wait(pg_barrier_t *barrier) +{ +#ifdef PG_THREADS_WIN32 + if (EnterSynchronizationBarrier(barrier, SYNCHRONIZATION_BARRIER_FLAGS_BLOCK_ONLY)) + return pg_thrd_success_last; + else + return pg_thrd_success; +#elif defined(HAVE_PTHREAD_BARRIER) + int error = pthread_barrier_wait(barrier); + + if (error == 0) + return pg_thrd_success; + else if (error == PTHREAD_BARRIER_SERIAL_THREAD) + return pg_thrd_success_last; + else + return pg_thrd_error; +#else + bool initial_sense; + + pg_mtx_lock(&barrier->mutex); + barrier->arrived++; + if (barrier->arrived == barrier->expected) + { + barrier->arrived = 0; + barrier->sense = !barrier->sense; + pg_mtx_unlock(&barrier->mutex); + pg_cnd_broadcast(&barrier->cond); + return pg_thrd_success_last; + } + initial_sense = barrier->sense; + do + { + pg_cnd_wait(&barrier->cond, &barrier->mutex); + } while (barrier->sense == initial_sense); + pg_mtx_unlock(&barrier->mutex); + return pg_thrd_success; +#endif +} + +static inline int +pg_barrier_destroy(pg_barrier_t *barrier) +{ +#ifdef PG_THREADS_WIN32 + return pg_thrd_success; +#elif defined(HAVE_PTHREAD_BARRIER) + return pg_thrd_maperror(pthread_barrier_destroy(barrier)); +#else + pg_mtx_destroy(&barrier->mutex); + pg_cnd_destroy(&barrier->cond); + return pg_thrd_success; +#endif +} + +#endif diff --git a/src/interfaces/ecpg/ecpglib/connect.c b/src/interfaces/ecpg/ecpglib/connect.c index 8afb1f0a26..073e069a21 100644 --- a/src/interfaces/ecpg/ecpglib/connect.c +++ b/src/interfaces/ecpg/ecpglib/connect.c @@ -2,8 +2,8 @@ #define POSTGRES_ECPG_INTERNAL #include "postgres_fe.h" +#include "port/pg_threads.h" -#include "ecpg-pthread-win32.h" #include "ecpgerrno.h" #include "ecpglib.h" #include "ecpglib_extern.h" @@ -14,24 +14,11 @@ locale_t ecpg_clocale = (locale_t) 0; #endif -static pthread_mutex_t connections_mutex = PTHREAD_MUTEX_INITIALIZER; -static pthread_key_t actual_connection_key; -static pthread_once_t actual_connection_key_once = PTHREAD_ONCE_INIT; +static pg_mtx_t connections_mutex = PG_MTX_STATIC_INIT; +static pg_thread_local struct connection *actual_connection_this_thread; static struct connection *actual_connection = NULL; static struct connection *all_connections = NULL; -static void -ecpg_actual_connection_init(void) -{ - pthread_key_create(&actual_connection_key, NULL); -} - -void -ecpg_pthreads_init(void) -{ - pthread_once(&actual_connection_key_once, ecpg_actual_connection_init); -} - static struct connection * ecpg_get_connection_nr(const char *connection_name) { @@ -39,9 +26,7 @@ ecpg_get_connection_nr(const char *connection_name) if ((connection_name == NULL) || (strcmp(connection_name, "CURRENT") == 0)) { - ecpg_pthreads_init(); /* ensure actual_connection_key is valid */ - - ret = pthread_getspecific(actual_connection_key); + ret = actual_connection_this_thread; /* * if no connection in TSD for this thread, get the global default @@ -74,9 +59,7 @@ ecpg_get_connection(const char *connection_name) if ((connection_name == NULL) || (strcmp(connection_name, "CURRENT") == 0)) { - ecpg_pthreads_init(); /* ensure actual_connection_key is valid */ - - ret = pthread_getspecific(actual_connection_key); + ret = actual_connection_this_thread; /* * if no connection in TSD for this thread, get the global default @@ -89,11 +72,11 @@ ecpg_get_connection(const char *connection_name) } else { - pthread_mutex_lock(&connections_mutex); + pg_mtx_lock(&connections_mutex); ret = ecpg_get_connection_nr(connection_name); - pthread_mutex_unlock(&connections_mutex); + pg_mtx_unlock(&connections_mutex); } return ret; @@ -127,8 +110,8 @@ ecpg_finish(struct connection *act) con->next = act->next; } - if (pthread_getspecific(actual_connection_key) == act) - pthread_setspecific(actual_connection_key, all_connections); + if (actual_connection_this_thread == act) + actual_connection_this_thread = all_connections; if (actual_connection == act) actual_connection = all_connections; @@ -194,7 +177,7 @@ ECPGsetconn(int lineno, const char *connection_name) if (!ecpg_init(con, connection_name, lineno)) return false; - pthread_setspecific(actual_connection_key, con); + actual_connection_this_thread = con; return true; } @@ -481,7 +464,7 @@ ECPGconnect(int lineno, int c, const char *name, const char *user, const char *p } /* add connection to our list */ - pthread_mutex_lock(&connections_mutex); + pg_mtx_lock(&connections_mutex); /* * ... but first, make certain we have created ecpg_clocale. Rely on @@ -493,7 +476,7 @@ ECPGconnect(int lineno, int c, const char *name, const char *user, const char *p ecpg_clocale = newlocale(LC_NUMERIC_MASK, "C", (locale_t) 0); if (!ecpg_clocale) { - pthread_mutex_unlock(&connections_mutex); + pg_mtx_unlock(&connections_mutex); ecpg_raise(lineno, ECPG_OUT_OF_MEMORY, ECPG_SQLSTATE_ECPG_OUT_OF_MEMORY, NULL); if (host) @@ -530,7 +513,7 @@ ECPGconnect(int lineno, int c, const char *name, const char *user, const char *p this->next = all_connections; all_connections = this; - pthread_setspecific(actual_connection_key, all_connections); + actual_connection_this_thread = all_connections; actual_connection = all_connections; ecpg_log("ECPGconnect: opening database %s on %s port %s %s%s %s%s\n", @@ -648,7 +631,7 @@ ECPGconnect(int lineno, int c, const char *name, const char *user, const char *p ecpg_log("ECPGconnect: %s", errmsg); ecpg_finish(this); - pthread_mutex_unlock(&connections_mutex); + pg_mtx_unlock(&connections_mutex); ecpg_raise(lineno, ECPG_CONNECT, ECPG_SQLSTATE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION, db); if (realname) @@ -660,7 +643,7 @@ ECPGconnect(int lineno, int c, const char *name, const char *user, const char *p if (realname) ecpg_free(realname); - pthread_mutex_unlock(&connections_mutex); + pg_mtx_unlock(&connections_mutex); this->autocommit = autocommit; @@ -682,7 +665,7 @@ ECPGdisconnect(int lineno, const char *connection_name) return false; } - pthread_mutex_lock(&connections_mutex); + pg_mtx_lock(&connections_mutex); if (strcmp(connection_name, "ALL") == 0) { @@ -701,14 +684,14 @@ ECPGdisconnect(int lineno, const char *connection_name) if (!ecpg_init(con, connection_name, lineno)) { - pthread_mutex_unlock(&connections_mutex); + pg_mtx_unlock(&connections_mutex); return false; } else ecpg_finish(con); } - pthread_mutex_unlock(&connections_mutex); + pg_mtx_unlock(&connections_mutex); return true; } diff --git a/src/interfaces/ecpg/ecpglib/descriptor.c b/src/interfaces/ecpg/ecpglib/descriptor.c index ad279e245c..b57e065672 100644 --- a/src/interfaces/ecpg/ecpglib/descriptor.c +++ b/src/interfaces/ecpg/ecpglib/descriptor.c @@ -7,11 +7,11 @@ #include "postgres_fe.h" #include "catalog/pg_type_d.h" -#include "ecpg-pthread-win32.h" #include "ecpgerrno.h" #include "ecpglib.h" #include "ecpglib_extern.h" #include "ecpgtype.h" +#include "port/pg_threads.h" #include "sql3types.h" #include "sqlca.h" #include "sqlda.h" @@ -19,8 +19,8 @@ static void descriptor_free(struct descriptor *desc); /* We manage descriptors separately for each thread. */ -static pthread_key_t descriptor_key; -static pthread_once_t descriptor_once = PTHREAD_ONCE_INIT; +static pg_tss_t descriptor_key; +static pg_once_flag descriptor_once = PG_ONCE_FLAG_INIT; static void descriptor_deallocate_all(struct descriptor *list); @@ -33,20 +33,20 @@ descriptor_destructor(void *arg) static void descriptor_key_init(void) { - pthread_key_create(&descriptor_key, descriptor_destructor); + pg_tss_create(&descriptor_key, descriptor_destructor); } static struct descriptor * get_descriptors(void) { - pthread_once(&descriptor_once, descriptor_key_init); - return (struct descriptor *) pthread_getspecific(descriptor_key); + pg_call_once(&descriptor_once, descriptor_key_init); + return (struct descriptor *) pg_tss_get(descriptor_key); } static void set_descriptors(struct descriptor *value) { - pthread_setspecific(descriptor_key, value); + pg_tss_set(descriptor_key, value); } /* old internal convenience function that might go away later */ diff --git a/src/interfaces/ecpg/ecpglib/ecpglib_extern.h b/src/interfaces/ecpg/ecpglib/ecpglib_extern.h index 01b4309a71..d8416b19e3 100644 --- a/src/interfaces/ecpg/ecpglib/ecpglib_extern.h +++ b/src/interfaces/ecpg/ecpglib/ecpglib_extern.h @@ -169,7 +169,7 @@ bool ecpg_get_data(const PGresult *, int, int, int, enum ECPGttype type, enum ECPGttype, char *, char *, long, long, long, enum ARRAY_TYPE, enum COMPAT_MODE, bool); -void ecpg_pthreads_init(void); +#define ecpg_pthreads_init() struct connection *ecpg_get_connection(const char *connection_name); char *ecpg_alloc(long size, int lineno); char *ecpg_auto_alloc(long size, int lineno); diff --git a/src/interfaces/ecpg/ecpglib/memory.c b/src/interfaces/ecpg/ecpglib/memory.c index a83637ac75..985f05087d 100644 --- a/src/interfaces/ecpg/ecpglib/memory.c +++ b/src/interfaces/ecpg/ecpglib/memory.c @@ -3,11 +3,11 @@ #define POSTGRES_ECPG_INTERNAL #include "postgres_fe.h" -#include "ecpg-pthread-win32.h" #include "ecpgerrno.h" #include "ecpglib.h" #include "ecpglib_extern.h" #include "ecpgtype.h" +#include "port/pg_threads.h" void ecpg_free(void *ptr) @@ -68,8 +68,8 @@ struct auto_mem struct auto_mem *next; }; -static pthread_key_t auto_mem_key; -static pthread_once_t auto_mem_once = PTHREAD_ONCE_INIT; +static pg_tss_t auto_mem_key; +static pg_once_flag auto_mem_once = PG_ONCE_FLAG_INIT; static void auto_mem_destructor(void *arg) @@ -81,20 +81,20 @@ auto_mem_destructor(void *arg) static void auto_mem_key_init(void) { - pthread_key_create(&auto_mem_key, auto_mem_destructor); + pg_tss_create(&auto_mem_key, auto_mem_destructor); } static struct auto_mem * get_auto_allocs(void) { - pthread_once(&auto_mem_once, auto_mem_key_init); - return (struct auto_mem *) pthread_getspecific(auto_mem_key); + pg_call_once(&auto_mem_once, auto_mem_key_init); + return (struct auto_mem *) pg_tss_get(auto_mem_key); } static void set_auto_allocs(struct auto_mem *am) { - pthread_setspecific(auto_mem_key, am); + pg_tss_set(auto_mem_key, am); } char * diff --git a/src/interfaces/ecpg/ecpglib/misc.c b/src/interfaces/ecpg/ecpglib/misc.c index 2ae989e3e5..6fa50e9513 100644 --- a/src/interfaces/ecpg/ecpglib/misc.c +++ b/src/interfaces/ecpg/ecpglib/misc.c @@ -6,7 +6,6 @@ #include #include -#include "ecpg-pthread-win32.h" #include "ecpgerrno.h" #include "ecpglib.h" #include "ecpglib_extern.h" @@ -16,6 +15,7 @@ #include "pgtypes_interval.h" #include "pgtypes_numeric.h" #include "pgtypes_timestamp.h" +#include "port/pg_threads.h" #include "sqlca.h" #ifndef LONG_LONG_MIN @@ -55,11 +55,10 @@ static struct sqlca_t sqlca_init = } }; -static pthread_key_t sqlca_key; -static pthread_once_t sqlca_key_once = PTHREAD_ONCE_INIT; - -static pthread_mutex_t debug_mutex = PTHREAD_MUTEX_INITIALIZER; -static pthread_mutex_t debug_init_mutex = PTHREAD_MUTEX_INITIALIZER; +static pg_tss_t sqlca_key; +static pg_once_flag ecpg_once = PG_ONCE_FLAG_INIT; +static pg_mtx_t debug_mutex = PG_MTX_STATIC_INIT; +static pg_mtx_t debug_init_mutex = PG_MTX_STATIC_INIT; static volatile int simple_debug = 0; static FILE *debugstream = NULL; @@ -99,9 +98,9 @@ ecpg_sqlca_key_destructor(void *arg) } static void -ecpg_sqlca_key_init(void) +ecpg_init_once(void) { - pthread_key_create(&sqlca_key, ecpg_sqlca_key_destructor); + pg_tss_create(&sqlca_key, ecpg_sqlca_key_destructor); } struct sqlca_t * @@ -109,16 +108,16 @@ ECPGget_sqlca(void) { struct sqlca_t *sqlca; - pthread_once(&sqlca_key_once, ecpg_sqlca_key_init); + pg_call_once(&ecpg_once, ecpg_init_once); - sqlca = pthread_getspecific(sqlca_key); + sqlca = pg_tss_get(sqlca_key); if (sqlca == NULL) { sqlca = malloc(sizeof(struct sqlca_t)); if (sqlca == NULL) return NULL; ecpg_init_sqlca(sqlca); - pthread_setspecific(sqlca_key, sqlca); + pg_tss_set(sqlca_key, sqlca); } return sqlca; } @@ -204,10 +203,10 @@ void ECPGdebug(int n, FILE *dbgs) { /* Interlock against concurrent executions of ECPGdebug() */ - pthread_mutex_lock(&debug_init_mutex); + pg_mtx_lock(&debug_init_mutex); /* Prevent ecpg_log() from printing while we change settings */ - pthread_mutex_lock(&debug_mutex); + pg_mtx_lock(&debug_mutex); if (n > 100) { @@ -220,12 +219,12 @@ ECPGdebug(int n, FILE *dbgs) debugstream = dbgs; /* We must release debug_mutex before invoking ecpg_log() ... */ - pthread_mutex_unlock(&debug_mutex); + pg_mtx_unlock(&debug_mutex); /* ... but keep holding debug_init_mutex to avoid racy printout */ ecpg_log("ECPGdebug: set to %d\n", simple_debug); - pthread_mutex_unlock(&debug_init_mutex); + pg_mtx_unlock(&debug_init_mutex); } void @@ -262,7 +261,7 @@ ecpg_log(const char *format,...) else snprintf(fmt, bufsize, "[%d]: %s", (int) getpid(), intl_format); - pthread_mutex_lock(&debug_mutex); + pg_mtx_lock(&debug_mutex); /* Now that we hold the mutex, recheck simple_debug */ if (simple_debug) @@ -281,7 +280,7 @@ ecpg_log(const char *format,...) fflush(debugstream); } - pthread_mutex_unlock(&debug_mutex); + pg_mtx_unlock(&debug_mutex); free(fmt); } @@ -422,60 +421,6 @@ ECPGis_noind_null(enum ECPGttype type, const void *ptr) return false; } -#ifdef WIN32 - -int -pthread_mutex_init(pthread_mutex_t *mp, void *attr) -{ - mp->initstate = 0; - return 0; -} - -int -pthread_mutex_lock(pthread_mutex_t *mp) -{ - /* Initialize the csection if not already done */ - if (mp->initstate != 1) - { - LONG istate; - - while ((istate = InterlockedExchange(&mp->initstate, 2)) == 2) - Sleep(0); /* wait, another thread is doing this */ - if (istate != 1) - InitializeCriticalSection(&mp->csection); - InterlockedExchange(&mp->initstate, 1); - } - EnterCriticalSection(&mp->csection); - return 0; -} - -int -pthread_mutex_unlock(pthread_mutex_t *mp) -{ - if (mp->initstate != 1) - return EINVAL; - LeaveCriticalSection(&mp->csection); - return 0; -} - -static pthread_mutex_t win32_pthread_once_lock = PTHREAD_MUTEX_INITIALIZER; - -void -win32_pthread_once(volatile pthread_once_t *once, void (*fn) (void)) -{ - if (!*once) - { - pthread_mutex_lock(&win32_pthread_once_lock); - if (!*once) - { - fn(); - *once = true; - } - pthread_mutex_unlock(&win32_pthread_once_lock); - } -} -#endif /* WIN32 */ - #ifdef ENABLE_NLS char * diff --git a/src/interfaces/ecpg/ecpglib/sqlda.c b/src/interfaces/ecpg/ecpglib/sqlda.c index 081e32666f..7231cc4a6f 100644 --- a/src/interfaces/ecpg/ecpglib/sqlda.c +++ b/src/interfaces/ecpg/ecpglib/sqlda.c @@ -11,7 +11,6 @@ #include "catalog/pg_type_d.h" #include "decimal.h" -#include "ecpg-pthread-win32.h" #include "ecpgerrno.h" #include "ecpglib.h" #include "ecpglib_extern.h" diff --git a/src/interfaces/ecpg/include/ecpg-pthread-win32.h b/src/interfaces/ecpg/include/ecpg-pthread-win32.h deleted file mode 100644 index 7b6ba46b34..0000000000 --- a/src/interfaces/ecpg/include/ecpg-pthread-win32.h +++ /dev/null @@ -1,49 +0,0 @@ -/* src/interfaces/ecpg/include/ecpg-pthread-win32.h */ -/* - * pthread mapping macros for win32 native thread implementation - */ -#ifndef _ECPG_PTHREAD_WIN32_H -#define _ECPG_PTHREAD_WIN32_H - -#ifndef WIN32 - -#include -#else - -typedef struct pthread_mutex_t -{ - /* initstate = 0: not initialized; 1: init done; 2: init in progress */ - LONG initstate; - CRITICAL_SECTION csection; -} pthread_mutex_t; - -typedef DWORD pthread_key_t; -typedef bool pthread_once_t; - -#define PTHREAD_MUTEX_INITIALIZER { 0 } -#define PTHREAD_ONCE_INIT false - -int pthread_mutex_init(pthread_mutex_t *, void *attr); -int pthread_mutex_lock(pthread_mutex_t *); -int pthread_mutex_unlock(pthread_mutex_t *); - -void win32_pthread_once(volatile pthread_once_t *once, void (*fn) (void)); - -#define pthread_getspecific(key) \ - TlsGetValue((key)) - -#define pthread_setspecific(key, value) \ - TlsSetValue((key), (value)) - -/* FIXME: destructor is never called in Win32. */ -#define pthread_key_create(key, destructor) \ - do { *(key) = TlsAlloc(); ((void)(destructor)); } while(0) - -#define pthread_once(once, fn) \ - do { \ - if (!*(once)) \ - win32_pthread_once((once), (fn)); \ - } while(0) -#endif /* WIN32 */ - -#endif /* _ECPG_PTHREAD_WIN32_H */ diff --git a/src/interfaces/ecpg/include/ecpg_config.h.in b/src/interfaces/ecpg/include/ecpg_config.h.in index 824617b917..04629f937e 100644 --- a/src/interfaces/ecpg/include/ecpg_config.h.in +++ b/src/interfaces/ecpg/include/ecpg_config.h.in @@ -10,5 +10,8 @@ /* Define to 1 if `long long int' works and is 64 bits. */ #undef HAVE_LONG_LONG_INT_64 +/* Define to 1 if you have the `pthread_barrier_wait' function. */ +#undef HAVE_PTHREAD_BARRIER_WAIT + /* Define to 1 to use to define type bool. */ #undef PG_USE_STDBOOL diff --git a/src/interfaces/ecpg/include/meson.build b/src/interfaces/ecpg/include/meson.build index 31610fef58..a4023f3684 100644 --- a/src/interfaces/ecpg/include/meson.build +++ b/src/interfaces/ecpg/include/meson.build @@ -6,6 +6,7 @@ ecpg_conf_keys = [ 'HAVE_INT64', 'HAVE_LONG_INT_64', 'HAVE_LONG_LONG_INT_64', + 'HAVE_PTHREAD_BARRIER_WAIT', 'PG_USE_STDBOOL', ] diff --git a/src/interfaces/ecpg/test/expected/thread-alloc.c b/src/interfaces/ecpg/test/expected/thread-alloc.c index 3b31d27fd3..97ed928ef1 100644 --- a/src/interfaces/ecpg/test/expected/thread-alloc.c +++ b/src/interfaces/ecpg/test/expected/thread-alloc.c @@ -10,16 +10,7 @@ #include #include #include "ecpg_config.h" - -#ifdef WIN32 -#define WIN32_LEAN_AND_MEAN -#include -#include -#include -#else -#include -#endif -#include +#include "port/pg_threads.h" #define THREADS 16 #define REPEATS 50 @@ -93,7 +84,7 @@ struct sqlca_t *ECPGget_sqlca(void); #endif -#line 18 "alloc.pgc" +#line 9 "alloc.pgc" #line 1 "regression.h" @@ -103,21 +94,17 @@ struct sqlca_t *ECPGget_sqlca(void); -#line 19 "alloc.pgc" +#line 10 "alloc.pgc" /* exec sql whenever sqlerror sqlprint ; */ -#line 21 "alloc.pgc" +#line 12 "alloc.pgc" /* exec sql whenever not found sqlprint ; */ -#line 22 "alloc.pgc" +#line 13 "alloc.pgc" -#ifdef WIN32 -static unsigned __stdcall fn(void* arg) -#else -static void* fn(void* arg) -#endif +static int fn(void* arg) { int i; @@ -126,54 +113,54 @@ static void* fn(void* arg) -#line 33 "alloc.pgc" +#line 20 "alloc.pgc" int value ; -#line 34 "alloc.pgc" +#line 21 "alloc.pgc" char name [ 100 ] ; -#line 35 "alloc.pgc" +#line 22 "alloc.pgc" char ** r = NULL ; /* exec sql end declare section */ -#line 36 "alloc.pgc" +#line 23 "alloc.pgc" value = (intptr_t) arg; sprintf(name, "Connection: %d", value); { ECPGconnect(__LINE__, 0, "ecpg1_regression" , NULL, NULL , name, 0); -#line 41 "alloc.pgc" +#line 28 "alloc.pgc" if (sqlca.sqlcode < 0) sqlprint();} -#line 41 "alloc.pgc" +#line 28 "alloc.pgc" { ECPGsetcommit(__LINE__, "on", NULL); -#line 42 "alloc.pgc" +#line 29 "alloc.pgc" if (sqlca.sqlcode < 0) sqlprint();} -#line 42 "alloc.pgc" +#line 29 "alloc.pgc" for (i = 1; i <= REPEATS; ++i) { { ECPGdo(__LINE__, 0, 1, NULL, 0, ECPGst_normal, "select relname from pg_class where relname = 'pg_class'", ECPGt_EOIT, ECPGt_char,&(r),(long)0,(long)0,(1)*sizeof(char), ECPGt_NO_INDICATOR, NULL , 0L, 0L, 0L, ECPGt_EORT); -#line 45 "alloc.pgc" +#line 32 "alloc.pgc" if (sqlca.sqlcode == ECPG_NOT_FOUND) sqlprint(); -#line 45 "alloc.pgc" +#line 32 "alloc.pgc" if (sqlca.sqlcode < 0) sqlprint();} -#line 45 "alloc.pgc" +#line 32 "alloc.pgc" free(r); r = NULL; } { ECPGdisconnect(__LINE__, name); -#line 49 "alloc.pgc" +#line 36 "alloc.pgc" if (sqlca.sqlcode < 0) sqlprint();} -#line 49 "alloc.pgc" +#line 36 "alloc.pgc" return 0; @@ -182,28 +169,12 @@ if (sqlca.sqlcode < 0) sqlprint();} int main () { intptr_t i; -#ifdef WIN32 - HANDLE threads[THREADS]; -#else - pthread_t threads[THREADS]; -#endif - -#ifdef WIN32 - for (i = 0; i < THREADS; ++i) - { - unsigned id; - threads[i] = (HANDLE)_beginthreadex(NULL, 0, fn, (void*)i, 0, &id); - } + pg_thrd_t threads[THREADS]; - WaitForMultipleObjects(THREADS, threads, TRUE, INFINITE); for (i = 0; i < THREADS; ++i) - CloseHandle(threads[i]); -#else + pg_thrd_create(&threads[i], fn, (void *) i); for (i = 0; i < THREADS; ++i) - pthread_create(&threads[i], NULL, fn, (void *) i); - for (i = 0; i < THREADS; ++i) - pthread_join(threads[i], NULL); -#endif + pg_thrd_join(threads[i], NULL); return 0; } diff --git a/src/interfaces/ecpg/test/expected/thread-descriptor.c b/src/interfaces/ecpg/test/expected/thread-descriptor.c index e34f4708d1..50cdbf1b2f 100644 --- a/src/interfaces/ecpg/test/expected/thread-descriptor.c +++ b/src/interfaces/ecpg/test/expected/thread-descriptor.c @@ -7,15 +7,7 @@ #define ECPGdebug(X,Y) ECPGdebug((X)+100,(Y)) #line 1 "descriptor.pgc" -#ifdef WIN32 -#define WIN32_LEAN_AND_MEAN -#include -#include -#include -#else -#include -#endif -#include +#include "port/pg_threads.h" #define THREADS 16 #define REPEATS 50000 @@ -89,36 +81,32 @@ struct sqlca_t *ECPGget_sqlca(void); #endif -#line 14 "descriptor.pgc" +#line 6 "descriptor.pgc" /* exec sql whenever sqlerror sqlprint ; */ -#line 15 "descriptor.pgc" +#line 7 "descriptor.pgc" /* exec sql whenever not found sqlprint ; */ -#line 16 "descriptor.pgc" +#line 8 "descriptor.pgc" -#if defined(WIN32) -static unsigned __stdcall fn(void* arg) -#else -static void* fn(void* arg) -#endif +static int fn(void* arg) { int i; for (i = 1; i <= REPEATS; ++i) { ECPGallocate_desc(__LINE__, "mydesc"); -#line 28 "descriptor.pgc" +#line 16 "descriptor.pgc" if (sqlca.sqlcode < 0) sqlprint(); -#line 28 "descriptor.pgc" +#line 16 "descriptor.pgc" ECPGdeallocate_desc(__LINE__, "mydesc"); -#line 29 "descriptor.pgc" +#line 17 "descriptor.pgc" if (sqlca.sqlcode < 0) sqlprint(); -#line 29 "descriptor.pgc" +#line 17 "descriptor.pgc" } @@ -128,28 +116,12 @@ if (sqlca.sqlcode < 0) sqlprint(); int main () { int i; -#ifdef WIN32 - HANDLE threads[THREADS]; -#else - pthread_t threads[THREADS]; -#endif - -#ifdef WIN32 - for (i = 0; i < THREADS; ++i) - { - unsigned id; - threads[i] = (HANDLE)_beginthreadex(NULL, 0, fn, NULL, 0, &id); - } + pg_thrd_t threads[THREADS]; - WaitForMultipleObjects(THREADS, threads, TRUE, INFINITE); - for (i = 0; i < THREADS; ++i) - CloseHandle(threads[i]); -#else for (i = 0; i < THREADS; ++i) - pthread_create(&threads[i], NULL, fn, NULL); + pg_thrd_create(&threads[i], fn, NULL); for (i = 0; i < THREADS; ++i) - pthread_join(threads[i], NULL); -#endif + pg_thrd_join(threads[i], NULL); return 0; } diff --git a/src/interfaces/ecpg/test/expected/thread-prep.c b/src/interfaces/ecpg/test/expected/thread-prep.c index 052e27b634..c5b60d18cb 100644 --- a/src/interfaces/ecpg/test/expected/thread-prep.c +++ b/src/interfaces/ecpg/test/expected/thread-prep.c @@ -10,15 +10,8 @@ #include #include #include "ecpg_config.h" +#include "port/pg_threads.h" -#ifdef WIN32 -#define WIN32_LEAN_AND_MEAN -#include -#include -#include -#else -#include -#endif #include #define THREADS 16 @@ -93,7 +86,7 @@ struct sqlca_t *ECPGget_sqlca(void); #endif -#line 18 "prep.pgc" +#line 11 "prep.pgc" #line 1 "regression.h" @@ -103,21 +96,17 @@ struct sqlca_t *ECPGget_sqlca(void); -#line 19 "prep.pgc" +#line 12 "prep.pgc" /* exec sql whenever sqlerror sqlprint ; */ -#line 21 "prep.pgc" +#line 14 "prep.pgc" /* exec sql whenever not found sqlprint ; */ -#line 22 "prep.pgc" +#line 15 "prep.pgc" -#ifdef WIN32 -static unsigned __stdcall fn(void* arg) -#else -static void* fn(void* arg) -#endif +static int fn(void* arg) { int i; @@ -126,64 +115,64 @@ static void* fn(void* arg) -#line 33 "prep.pgc" +#line 22 "prep.pgc" int value ; -#line 34 "prep.pgc" +#line 23 "prep.pgc" char name [ 100 ] ; -#line 35 "prep.pgc" +#line 24 "prep.pgc" char query [ 256 ] = "INSERT INTO T VALUES ( ? )" ; /* exec sql end declare section */ -#line 36 "prep.pgc" +#line 25 "prep.pgc" value = (intptr_t) arg; sprintf(name, "Connection: %d", value); { ECPGconnect(__LINE__, 0, "ecpg1_regression" , NULL, NULL , name, 0); -#line 41 "prep.pgc" +#line 30 "prep.pgc" if (sqlca.sqlcode < 0) sqlprint();} -#line 41 "prep.pgc" +#line 30 "prep.pgc" { ECPGsetcommit(__LINE__, "on", NULL); -#line 42 "prep.pgc" +#line 31 "prep.pgc" if (sqlca.sqlcode < 0) sqlprint();} -#line 42 "prep.pgc" +#line 31 "prep.pgc" for (i = 1; i <= REPEATS; ++i) { { ECPGprepare(__LINE__, NULL, 0, "i", query); -#line 45 "prep.pgc" +#line 34 "prep.pgc" if (sqlca.sqlcode < 0) sqlprint();} -#line 45 "prep.pgc" +#line 34 "prep.pgc" { ECPGdo(__LINE__, 0, 1, NULL, 0, ECPGst_execute, "i", ECPGt_int,&(value),(long)1,(long)1,sizeof(int), ECPGt_NO_INDICATOR, NULL , 0L, 0L, 0L, ECPGt_EOIT, ECPGt_EORT); -#line 46 "prep.pgc" +#line 35 "prep.pgc" if (sqlca.sqlcode == ECPG_NOT_FOUND) sqlprint(); -#line 46 "prep.pgc" +#line 35 "prep.pgc" if (sqlca.sqlcode < 0) sqlprint();} -#line 46 "prep.pgc" +#line 35 "prep.pgc" } { ECPGdeallocate(__LINE__, 0, NULL, "i"); -#line 48 "prep.pgc" +#line 37 "prep.pgc" if (sqlca.sqlcode < 0) sqlprint();} -#line 48 "prep.pgc" +#line 37 "prep.pgc" { ECPGdisconnect(__LINE__, name); -#line 49 "prep.pgc" +#line 38 "prep.pgc" if (sqlca.sqlcode < 0) sqlprint();} -#line 49 "prep.pgc" +#line 38 "prep.pgc" return 0; @@ -192,59 +181,43 @@ if (sqlca.sqlcode < 0) sqlprint();} int main () { intptr_t i; -#ifdef WIN32 - HANDLE threads[THREADS]; -#else - pthread_t threads[THREADS]; -#endif + pg_thrd_t threads[THREADS]; { ECPGconnect(__LINE__, 0, "ecpg1_regression" , NULL, NULL , NULL, 0); -#line 63 "prep.pgc" +#line 48 "prep.pgc" if (sqlca.sqlcode < 0) sqlprint();} -#line 63 "prep.pgc" +#line 48 "prep.pgc" { ECPGsetcommit(__LINE__, "on", NULL); -#line 64 "prep.pgc" +#line 49 "prep.pgc" if (sqlca.sqlcode < 0) sqlprint();} -#line 64 "prep.pgc" +#line 49 "prep.pgc" { ECPGdo(__LINE__, 0, 1, NULL, 0, ECPGst_normal, "drop table if exists T", ECPGt_EOIT, ECPGt_EORT); -#line 65 "prep.pgc" +#line 50 "prep.pgc" if (sqlca.sqlcode < 0) sqlprint();} -#line 65 "prep.pgc" +#line 50 "prep.pgc" { ECPGdo(__LINE__, 0, 1, NULL, 0, ECPGst_normal, "create table T ( i int )", ECPGt_EOIT, ECPGt_EORT); -#line 66 "prep.pgc" +#line 51 "prep.pgc" if (sqlca.sqlcode < 0) sqlprint();} -#line 66 "prep.pgc" +#line 51 "prep.pgc" { ECPGdisconnect(__LINE__, "CURRENT"); -#line 67 "prep.pgc" +#line 52 "prep.pgc" if (sqlca.sqlcode < 0) sqlprint();} -#line 67 "prep.pgc" +#line 52 "prep.pgc" -#ifdef WIN32 for (i = 0; i < THREADS; ++i) - { - unsigned id; - threads[i] = (HANDLE)_beginthreadex(NULL, 0, fn, (void*)i, 0, &id); - } - - WaitForMultipleObjects(THREADS, threads, TRUE, INFINITE); + pg_thrd_create(&threads[i], fn, (void *) i); for (i = 0; i < THREADS; ++i) - CloseHandle(threads[i]); -#else - for (i = 0; i < THREADS; ++i) - pthread_create(&threads[i], NULL, fn, (void *) i); - for (i = 0; i < THREADS; ++i) - pthread_join(threads[i], NULL); -#endif + pg_thrd_join(threads[i], NULL); return 0; } diff --git a/src/interfaces/ecpg/test/expected/thread-thread.c b/src/interfaces/ecpg/test/expected/thread-thread.c index 95faa223c2..03c7169995 100644 --- a/src/interfaces/ecpg/test/expected/thread-thread.c +++ b/src/interfaces/ecpg/test/expected/thread-thread.c @@ -14,13 +14,7 @@ #include #include #include "ecpg_config.h" - -#ifndef WIN32 -#include -#else -#include -#include -#endif +#include "port/pg_threads.h" #line 1 "regression.h" @@ -30,29 +24,25 @@ -#line 16 "thread.pgc" +#line 10 "thread.pgc" -void *test_thread(void *arg); +int test_thread(void *arg); int nthreads = 10; int iterations = 20; int main() { -#ifndef WIN32 - pthread_t *threads; -#else - HANDLE *threads; -#endif + pg_thrd_t *threads; intptr_t n; /* exec sql begin declare section */ -#line 32 "thread.pgc" +#line 22 "thread.pgc" int l_rows ; /* exec sql end declare section */ -#line 33 "thread.pgc" +#line 23 "thread.pgc" /* Do not switch on debug output for regression tests. The threads get executed in @@ -61,22 +51,22 @@ int main() /* setup test_thread table */ { ECPGconnect(__LINE__, 0, "ecpg1_regression" , NULL, NULL , NULL, 0); } -#line 40 "thread.pgc" +#line 30 "thread.pgc" { ECPGdo(__LINE__, 0, 1, NULL, 0, ECPGst_normal, "drop table test_thread", ECPGt_EOIT, ECPGt_EORT);} -#line 41 "thread.pgc" +#line 31 "thread.pgc" /* DROP might fail */ { ECPGtrans(__LINE__, NULL, "commit");} -#line 42 "thread.pgc" +#line 32 "thread.pgc" { ECPGdo(__LINE__, 0, 1, NULL, 0, ECPGst_normal, "create table test_thread ( tstamp timestamp not null default cast ( timeofday ( ) as timestamp ) , thread text not null , iteration integer not null , primary key ( thread , iteration ) )", ECPGt_EOIT, ECPGt_EORT);} -#line 47 "thread.pgc" +#line 37 "thread.pgc" { ECPGtrans(__LINE__, NULL, "commit");} -#line 48 "thread.pgc" +#line 38 "thread.pgc" { ECPGdisconnect(__LINE__, "CURRENT");} -#line 49 "thread.pgc" +#line 39 "thread.pgc" /* create, and start, threads */ @@ -87,39 +77,27 @@ int main() return 1; } for( n = 0; n < nthreads; n++ ) - { -#ifndef WIN32 - pthread_create(&threads[n], NULL, test_thread, (void *) (n + 1)); -#else - threads[n] = CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE) (void (*) (void)) test_thread, (void *) (n + 1), 0, NULL); -#endif - } + pg_thrd_create(&threads[n], test_thread, (void *) (n + 1)); /* wait for thread completion */ -#ifndef WIN32 for( n = 0; n < nthreads; n++ ) - { - pthread_join(threads[n], NULL); - } -#else - WaitForMultipleObjects(nthreads, threads, TRUE, INFINITE); -#endif + pg_thrd_join(threads[n], NULL); free(threads); /* and check results */ { ECPGconnect(__LINE__, 0, "ecpg1_regression" , NULL, NULL , NULL, 0); } -#line 79 "thread.pgc" +#line 57 "thread.pgc" { ECPGdo(__LINE__, 0, 1, NULL, 0, ECPGst_normal, "select count ( * ) from test_thread", ECPGt_EOIT, ECPGt_int,&(l_rows),(long)1,(long)1,sizeof(int), ECPGt_NO_INDICATOR, NULL , 0L, 0L, 0L, ECPGt_EORT);} -#line 80 "thread.pgc" +#line 58 "thread.pgc" { ECPGtrans(__LINE__, NULL, "commit");} -#line 81 "thread.pgc" +#line 59 "thread.pgc" { ECPGdisconnect(__LINE__, "CURRENT");} -#line 82 "thread.pgc" +#line 60 "thread.pgc" if( l_rows == (nthreads * iterations) ) printf("Success.\n"); @@ -129,7 +107,7 @@ int main() return 0; } -void *test_thread(void *arg) +int test_thread(void *arg) { long threadnum = (intptr_t) arg; @@ -137,13 +115,13 @@ void *test_thread(void *arg) -#line 96 "thread.pgc" +#line 74 "thread.pgc" int l_i ; -#line 97 "thread.pgc" +#line 75 "thread.pgc" char l_connection [ 128 ] ; /* exec sql end declare section */ -#line 98 "thread.pgc" +#line 76 "thread.pgc" /* build up connection name, and connect to database */ @@ -153,24 +131,24 @@ void *test_thread(void *arg) _snprintf(l_connection, sizeof(l_connection), "thread_%03ld", threadnum); #endif /* exec sql whenever sqlerror sqlprint ; */ -#line 106 "thread.pgc" +#line 84 "thread.pgc" { ECPGconnect(__LINE__, 0, "ecpg1_regression" , NULL, NULL , l_connection, 0); -#line 107 "thread.pgc" +#line 85 "thread.pgc" if (sqlca.sqlcode < 0) sqlprint();} -#line 107 "thread.pgc" +#line 85 "thread.pgc" if( sqlca.sqlcode != 0 ) { printf("%s: ERROR: cannot connect to database!\n", l_connection); - return NULL; + return 0; } { ECPGtrans(__LINE__, l_connection, "begin"); -#line 113 "thread.pgc" +#line 91 "thread.pgc" if (sqlca.sqlcode < 0) sqlprint();} -#line 113 "thread.pgc" +#line 91 "thread.pgc" /* insert into test_thread table */ @@ -181,10 +159,10 @@ if (sqlca.sqlcode < 0) sqlprint();} ECPGt_NO_INDICATOR, NULL , 0L, 0L, 0L, ECPGt_int,&(l_i),(long)1,(long)1,sizeof(int), ECPGt_NO_INDICATOR, NULL , 0L, 0L, 0L, ECPGt_EOIT, ECPGt_EORT); -#line 118 "thread.pgc" +#line 96 "thread.pgc" if (sqlca.sqlcode < 0) sqlprint();} -#line 118 "thread.pgc" +#line 96 "thread.pgc" if( sqlca.sqlcode != 0 ) printf("%s: ERROR: insert failed!\n", l_connection); @@ -192,16 +170,16 @@ if (sqlca.sqlcode < 0) sqlprint();} /* all done */ { ECPGtrans(__LINE__, l_connection, "commit"); -#line 124 "thread.pgc" +#line 102 "thread.pgc" if (sqlca.sqlcode < 0) sqlprint();} -#line 124 "thread.pgc" +#line 102 "thread.pgc" { ECPGdisconnect(__LINE__, l_connection); -#line 125 "thread.pgc" +#line 103 "thread.pgc" if (sqlca.sqlcode < 0) sqlprint();} -#line 125 "thread.pgc" +#line 103 "thread.pgc" - return NULL; + return 0; } diff --git a/src/interfaces/ecpg/test/expected/thread-thread_implicit.c b/src/interfaces/ecpg/test/expected/thread-thread_implicit.c index 7ac0297a23..c859012f50 100644 --- a/src/interfaces/ecpg/test/expected/thread-thread_implicit.c +++ b/src/interfaces/ecpg/test/expected/thread-thread_implicit.c @@ -14,13 +14,7 @@ #include #include #include "ecpg_config.h" - -#ifndef WIN32 -#include -#else -#include -#include -#endif +#include "port/pg_threads.h" #line 1 "regression.h" @@ -30,29 +24,25 @@ -#line 16 "thread_implicit.pgc" +#line 10 "thread_implicit.pgc" -void *test_thread(void *arg); +int test_thread(void *arg); int nthreads = 10; int iterations = 20; int main() { -#ifndef WIN32 - pthread_t *threads; -#else - HANDLE *threads; -#endif + pg_thrd_t *threads; intptr_t n; /* exec sql begin declare section */ -#line 32 "thread_implicit.pgc" +#line 22 "thread_implicit.pgc" int l_rows ; /* exec sql end declare section */ -#line 33 "thread_implicit.pgc" +#line 23 "thread_implicit.pgc" /* Do not switch on debug output for regression tests. The threads get executed in @@ -61,22 +51,22 @@ int main() /* setup test_thread table */ { ECPGconnect(__LINE__, 0, "ecpg1_regression" , NULL, NULL , NULL, 0); } -#line 40 "thread_implicit.pgc" +#line 30 "thread_implicit.pgc" { ECPGdo(__LINE__, 0, 1, NULL, 0, ECPGst_normal, "drop table test_thread", ECPGt_EOIT, ECPGt_EORT);} -#line 41 "thread_implicit.pgc" +#line 31 "thread_implicit.pgc" /* DROP might fail */ { ECPGtrans(__LINE__, NULL, "commit");} -#line 42 "thread_implicit.pgc" +#line 32 "thread_implicit.pgc" { ECPGdo(__LINE__, 0, 1, NULL, 0, ECPGst_normal, "create table test_thread ( tstamp timestamp not null default cast ( timeofday ( ) as timestamp ) , thread text not null , iteration integer not null , primary key ( thread , iteration ) )", ECPGt_EOIT, ECPGt_EORT);} -#line 47 "thread_implicit.pgc" +#line 37 "thread_implicit.pgc" { ECPGtrans(__LINE__, NULL, "commit");} -#line 48 "thread_implicit.pgc" +#line 38 "thread_implicit.pgc" { ECPGdisconnect(__LINE__, "CURRENT");} -#line 49 "thread_implicit.pgc" +#line 39 "thread_implicit.pgc" /* create, and start, threads */ @@ -87,39 +77,27 @@ int main() return 1; } for( n = 0; n < nthreads; n++ ) - { -#ifndef WIN32 - pthread_create(&threads[n], NULL, test_thread, (void *) (n + 1)); -#else - threads[n] = CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE) (void (*) (void)) test_thread, (void *) (n+1), 0, NULL); -#endif - } + pg_thrd_create(&threads[n], test_thread, (void *) (n + 1)); /* wait for thread completion */ -#ifndef WIN32 for( n = 0; n < nthreads; n++ ) - { - pthread_join(threads[n], NULL); - } -#else - WaitForMultipleObjects(nthreads, threads, TRUE, INFINITE); -#endif + pg_thrd_join(threads[n], NULL); free(threads); /* and check results */ { ECPGconnect(__LINE__, 0, "ecpg1_regression" , NULL, NULL , NULL, 0); } -#line 79 "thread_implicit.pgc" +#line 57 "thread_implicit.pgc" { ECPGdo(__LINE__, 0, 1, NULL, 0, ECPGst_normal, "select count ( * ) from test_thread", ECPGt_EOIT, ECPGt_int,&(l_rows),(long)1,(long)1,sizeof(int), ECPGt_NO_INDICATOR, NULL , 0L, 0L, 0L, ECPGt_EORT);} -#line 80 "thread_implicit.pgc" +#line 58 "thread_implicit.pgc" { ECPGtrans(__LINE__, NULL, "commit");} -#line 81 "thread_implicit.pgc" +#line 59 "thread_implicit.pgc" { ECPGdisconnect(__LINE__, "CURRENT");} -#line 82 "thread_implicit.pgc" +#line 60 "thread_implicit.pgc" if( l_rows == (nthreads * iterations) ) printf("Success.\n"); @@ -129,7 +107,7 @@ int main() return 0; } -void *test_thread(void *arg) +int test_thread(void *arg) { long threadnum = (intptr_t) arg; @@ -137,13 +115,13 @@ void *test_thread(void *arg) -#line 96 "thread_implicit.pgc" +#line 74 "thread_implicit.pgc" int l_i ; -#line 97 "thread_implicit.pgc" +#line 75 "thread_implicit.pgc" char l_connection [ 128 ] ; /* exec sql end declare section */ -#line 98 "thread_implicit.pgc" +#line 76 "thread_implicit.pgc" /* build up connection name, and connect to database */ @@ -153,24 +131,24 @@ void *test_thread(void *arg) _snprintf(l_connection, sizeof(l_connection), "thread_%03ld", threadnum); #endif /* exec sql whenever sqlerror sqlprint ; */ -#line 106 "thread_implicit.pgc" +#line 84 "thread_implicit.pgc" { ECPGconnect(__LINE__, 0, "ecpg1_regression" , NULL, NULL , l_connection, 0); -#line 107 "thread_implicit.pgc" +#line 85 "thread_implicit.pgc" if (sqlca.sqlcode < 0) sqlprint();} -#line 107 "thread_implicit.pgc" +#line 85 "thread_implicit.pgc" if( sqlca.sqlcode != 0 ) { printf("%s: ERROR: cannot connect to database!\n", l_connection); - return NULL; + return 0; } { ECPGtrans(__LINE__, NULL, "begin"); -#line 113 "thread_implicit.pgc" +#line 91 "thread_implicit.pgc" if (sqlca.sqlcode < 0) sqlprint();} -#line 113 "thread_implicit.pgc" +#line 91 "thread_implicit.pgc" /* insert into test_thread table */ @@ -181,10 +159,10 @@ if (sqlca.sqlcode < 0) sqlprint();} ECPGt_NO_INDICATOR, NULL , 0L, 0L, 0L, ECPGt_int,&(l_i),(long)1,(long)1,sizeof(int), ECPGt_NO_INDICATOR, NULL , 0L, 0L, 0L, ECPGt_EOIT, ECPGt_EORT); -#line 118 "thread_implicit.pgc" +#line 96 "thread_implicit.pgc" if (sqlca.sqlcode < 0) sqlprint();} -#line 118 "thread_implicit.pgc" +#line 96 "thread_implicit.pgc" if( sqlca.sqlcode != 0 ) printf("%s: ERROR: insert failed!\n", l_connection); @@ -192,16 +170,16 @@ if (sqlca.sqlcode < 0) sqlprint();} /* all done */ { ECPGtrans(__LINE__, NULL, "commit"); -#line 124 "thread_implicit.pgc" +#line 102 "thread_implicit.pgc" if (sqlca.sqlcode < 0) sqlprint();} -#line 124 "thread_implicit.pgc" +#line 102 "thread_implicit.pgc" { ECPGdisconnect(__LINE__, l_connection); -#line 125 "thread_implicit.pgc" +#line 103 "thread_implicit.pgc" if (sqlca.sqlcode < 0) sqlprint();} -#line 125 "thread_implicit.pgc" +#line 103 "thread_implicit.pgc" - return NULL; + return 0; } diff --git a/src/interfaces/ecpg/test/thread/alloc.pgc b/src/interfaces/ecpg/test/thread/alloc.pgc index d3d35493bf..248a29447c 100644 --- a/src/interfaces/ecpg/test/thread/alloc.pgc +++ b/src/interfaces/ecpg/test/thread/alloc.pgc @@ -1,16 +1,7 @@ #include #include #include "ecpg_config.h" - -#ifdef WIN32 -#define WIN32_LEAN_AND_MEAN -#include -#include -#include -#else -#include -#endif -#include +#include "port/pg_threads.h" #define THREADS 16 #define REPEATS 50 @@ -21,11 +12,7 @@ exec sql include ../regression; exec sql whenever sqlerror sqlprint; exec sql whenever not found sqlprint; -#ifdef WIN32 -static unsigned __stdcall fn(void* arg) -#else -static void* fn(void* arg) -#endif +static int fn(void* arg) { int i; @@ -54,28 +41,12 @@ static void* fn(void* arg) int main () { intptr_t i; -#ifdef WIN32 - HANDLE threads[THREADS]; -#else - pthread_t threads[THREADS]; -#endif - -#ifdef WIN32 - for (i = 0; i < THREADS; ++i) - { - unsigned id; - threads[i] = (HANDLE)_beginthreadex(NULL, 0, fn, (void*)i, 0, &id); - } + pg_thrd_t threads[THREADS]; - WaitForMultipleObjects(THREADS, threads, TRUE, INFINITE); - for (i = 0; i < THREADS; ++i) - CloseHandle(threads[i]); -#else for (i = 0; i < THREADS; ++i) - pthread_create(&threads[i], NULL, fn, (void *) i); + pg_thrd_create(&threads[i], fn, (void *) i); for (i = 0; i < THREADS; ++i) - pthread_join(threads[i], NULL); -#endif + pg_thrd_join(threads[i], NULL); return 0; } diff --git a/src/interfaces/ecpg/test/thread/descriptor.pgc b/src/interfaces/ecpg/test/thread/descriptor.pgc index 30bce7c87b..faa9a0a656 100644 --- a/src/interfaces/ecpg/test/thread/descriptor.pgc +++ b/src/interfaces/ecpg/test/thread/descriptor.pgc @@ -1,12 +1,4 @@ -#ifdef WIN32 -#define WIN32_LEAN_AND_MEAN -#include -#include -#include -#else -#include -#endif -#include +#include "port/pg_threads.h" #define THREADS 16 #define REPEATS 50000 @@ -15,11 +7,7 @@ EXEC SQL include sqlca; EXEC SQL whenever sqlerror sqlprint; EXEC SQL whenever not found sqlprint; -#if defined(WIN32) -static unsigned __stdcall fn(void* arg) -#else -static void* fn(void* arg) -#endif +static int fn(void* arg) { int i; @@ -35,28 +23,12 @@ static void* fn(void* arg) int main () { int i; -#ifdef WIN32 - HANDLE threads[THREADS]; -#else - pthread_t threads[THREADS]; -#endif + pg_thrd_t threads[THREADS]; -#ifdef WIN32 for (i = 0; i < THREADS; ++i) - { - unsigned id; - threads[i] = (HANDLE)_beginthreadex(NULL, 0, fn, NULL, 0, &id); - } - - WaitForMultipleObjects(THREADS, threads, TRUE, INFINITE); - for (i = 0; i < THREADS; ++i) - CloseHandle(threads[i]); -#else - for (i = 0; i < THREADS; ++i) - pthread_create(&threads[i], NULL, fn, NULL); + pg_thrd_create(&threads[i], fn, NULL); for (i = 0; i < THREADS; ++i) - pthread_join(threads[i], NULL); -#endif + pg_thrd_join(threads[i], NULL); return 0; } diff --git a/src/interfaces/ecpg/test/thread/prep.pgc b/src/interfaces/ecpg/test/thread/prep.pgc index f61b31ce10..f5a875cc28 100644 --- a/src/interfaces/ecpg/test/thread/prep.pgc +++ b/src/interfaces/ecpg/test/thread/prep.pgc @@ -1,15 +1,8 @@ #include #include #include "ecpg_config.h" +#include "port/pg_threads.h" -#ifdef WIN32 -#define WIN32_LEAN_AND_MEAN -#include -#include -#include -#else -#include -#endif #include #define THREADS 16 @@ -21,11 +14,7 @@ exec sql include ../regression; exec sql whenever sqlerror sqlprint; exec sql whenever not found sqlprint; -#ifdef WIN32 -static unsigned __stdcall fn(void* arg) -#else -static void* fn(void* arg) -#endif +static int fn(void* arg) { int i; @@ -54,11 +43,7 @@ static void* fn(void* arg) int main () { intptr_t i; -#ifdef WIN32 - HANDLE threads[THREADS]; -#else - pthread_t threads[THREADS]; -#endif + pg_thrd_t threads[THREADS]; EXEC SQL CONNECT TO REGRESSDB1; EXEC SQL SET AUTOCOMMIT TO ON; @@ -66,22 +51,10 @@ int main () EXEC SQL CREATE TABLE T ( i int ); EXEC SQL DISCONNECT; -#ifdef WIN32 for (i = 0; i < THREADS; ++i) - { - unsigned id; - threads[i] = (HANDLE)_beginthreadex(NULL, 0, fn, (void*)i, 0, &id); - } - - WaitForMultipleObjects(THREADS, threads, TRUE, INFINITE); - for (i = 0; i < THREADS; ++i) - CloseHandle(threads[i]); -#else - for (i = 0; i < THREADS; ++i) - pthread_create(&threads[i], NULL, fn, (void *) i); + pg_thrd_create(&threads[i], fn, (void *) i); for (i = 0; i < THREADS; ++i) - pthread_join(threads[i], NULL); -#endif + pg_thrd_join(threads[i], NULL); return 0; } diff --git a/src/interfaces/ecpg/test/thread/thread.pgc b/src/interfaces/ecpg/test/thread/thread.pgc index b9b9ebb441..bc967b9da8 100644 --- a/src/interfaces/ecpg/test/thread/thread.pgc +++ b/src/interfaces/ecpg/test/thread/thread.pgc @@ -5,28 +5,18 @@ #include #include #include "ecpg_config.h" - -#ifndef WIN32 -#include -#else -#include -#include -#endif +#include "port/pg_threads.h" exec sql include ../regression; -void *test_thread(void *arg); +int test_thread(void *arg); int nthreads = 10; int iterations = 20; int main() { -#ifndef WIN32 - pthread_t *threads; -#else - HANDLE *threads; -#endif + pg_thrd_t *threads; intptr_t n; EXEC SQL BEGIN DECLARE SECTION; int l_rows; @@ -56,23 +46,11 @@ int main() return 1; } for( n = 0; n < nthreads; n++ ) - { -#ifndef WIN32 - pthread_create(&threads[n], NULL, test_thread, (void *) (n + 1)); -#else - threads[n] = CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE) (void (*) (void)) test_thread, (void *) (n + 1), 0, NULL); -#endif - } + pg_thrd_create(&threads[n], test_thread, (void *) (n + 1)); /* wait for thread completion */ -#ifndef WIN32 for( n = 0; n < nthreads; n++ ) - { - pthread_join(threads[n], NULL); - } -#else - WaitForMultipleObjects(nthreads, threads, TRUE, INFINITE); -#endif + pg_thrd_join(threads[n], NULL); free(threads); /* and check results */ @@ -88,7 +66,7 @@ int main() return 0; } -void *test_thread(void *arg) +int test_thread(void *arg) { long threadnum = (intptr_t) arg; @@ -108,7 +86,7 @@ void *test_thread(void *arg) if( sqlca.sqlcode != 0 ) { printf("%s: ERROR: cannot connect to database!\n", l_connection); - return NULL; + return 0; } EXEC SQL AT :l_connection BEGIN; @@ -123,5 +101,5 @@ void *test_thread(void *arg) /* all done */ EXEC SQL AT :l_connection COMMIT; EXEC SQL DISCONNECT :l_connection; - return NULL; + return 0; } diff --git a/src/interfaces/ecpg/test/thread/thread_implicit.pgc b/src/interfaces/ecpg/test/thread/thread_implicit.pgc index ff9b12a943..7a793dc7f8 100644 --- a/src/interfaces/ecpg/test/thread/thread_implicit.pgc +++ b/src/interfaces/ecpg/test/thread/thread_implicit.pgc @@ -5,28 +5,18 @@ #include #include #include "ecpg_config.h" - -#ifndef WIN32 -#include -#else -#include -#include -#endif +#include "port/pg_threads.h" exec sql include ../regression; -void *test_thread(void *arg); +int test_thread(void *arg); int nthreads = 10; int iterations = 20; int main() { -#ifndef WIN32 - pthread_t *threads; -#else - HANDLE *threads; -#endif + pg_thrd_t *threads; intptr_t n; EXEC SQL BEGIN DECLARE SECTION; int l_rows; @@ -56,23 +46,11 @@ int main() return 1; } for( n = 0; n < nthreads; n++ ) - { -#ifndef WIN32 - pthread_create(&threads[n], NULL, test_thread, (void *) (n + 1)); -#else - threads[n] = CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE) (void (*) (void)) test_thread, (void *) (n+1), 0, NULL); -#endif - } + pg_thrd_create(&threads[n], test_thread, (void *) (n + 1)); /* wait for thread completion */ -#ifndef WIN32 for( n = 0; n < nthreads; n++ ) - { - pthread_join(threads[n], NULL); - } -#else - WaitForMultipleObjects(nthreads, threads, TRUE, INFINITE); -#endif + pg_thrd_join(threads[n], NULL); free(threads); /* and check results */ @@ -88,7 +66,7 @@ int main() return 0; } -void *test_thread(void *arg) +int test_thread(void *arg) { long threadnum = (intptr_t) arg; @@ -108,7 +86,7 @@ void *test_thread(void *arg) if( sqlca.sqlcode != 0 ) { printf("%s: ERROR: cannot connect to database!\n", l_connection); - return NULL; + return 0; } EXEC SQL BEGIN; @@ -123,5 +101,5 @@ void *test_thread(void *arg) /* all done */ EXEC SQL COMMIT; EXEC SQL DISCONNECT :l_connection; - return NULL; + return 0; } diff --git a/src/interfaces/libpq/Makefile b/src/interfaces/libpq/Makefile index 27f8499d8a..b8440a081c 100644 --- a/src/interfaces/libpq/Makefile +++ b/src/interfaces/libpq/Makefile @@ -69,7 +69,6 @@ endif ifeq ($(PORTNAME), win32) OBJS += \ - pthread-win32.o \ win32.o endif diff --git a/src/interfaces/libpq/fe-connect.c b/src/interfaces/libpq/fe-connect.c index ab308a0580..2397865a6f 100644 --- a/src/interfaces/libpq/fe-connect.c +++ b/src/interfaces/libpq/fe-connect.c @@ -32,6 +32,7 @@ #include "mb/pg_wchar.h" #include "pg_config_paths.h" #include "port/pg_bswap.h" +#include "port/pg_threads.h" #ifdef WIN32 #include "win32.h" @@ -52,12 +53,6 @@ #include #endif -#ifdef WIN32 -#include "pthread-win32.h" -#else -#include -#endif - #ifdef USE_LDAP #ifdef WIN32 #include @@ -7776,16 +7771,16 @@ error: static void default_threadlock(int acquire) { - static pthread_mutex_t singlethread_lock = PTHREAD_MUTEX_INITIALIZER; + static pg_mtx_t singlethread_lock = PG_MTX_STATIC_INIT; if (acquire) { - if (pthread_mutex_lock(&singlethread_lock)) + if (pg_mtx_lock(&singlethread_lock) != pg_thrd_success) Assert(false); } else { - if (pthread_mutex_unlock(&singlethread_lock)) + if (pg_mtx_unlock(&singlethread_lock) != pg_thrd_success) Assert(false); } } diff --git a/src/interfaces/libpq/fe-secure-openssl.c b/src/interfaces/libpq/fe-secure-openssl.c index b6fffd7b9b..e03a444b07 100644 --- a/src/interfaces/libpq/fe-secure-openssl.c +++ b/src/interfaces/libpq/fe-secure-openssl.c @@ -30,6 +30,7 @@ #include "fe-auth.h" #include "fe-secure-common.h" #include "libpq-int.h" +#include "port/pg_threads.h" #ifdef WIN32 #include "win32.h" @@ -44,12 +45,6 @@ #include -#ifdef WIN32 -#include "pthread-win32.h" -#else -#include -#endif - /* * These SSL-related #includes must come after all system-provided headers. * This ensures that OpenSSL can take care of conflicts with Windows' @@ -91,7 +86,7 @@ static bool ssl_lib_initialized = false; static long crypto_open_connections = 0; -static pthread_mutex_t ssl_config_mutex = PTHREAD_MUTEX_INITIALIZER; +static pg_mtx_t ssl_config_mutex = PG_MTX_STATIC_INIT; static PQsslKeyPassHook_OpenSSL_type PQsslKeyPassHook = NULL; static int ssl_protocol_version_to_openssl(const char *protocol); @@ -725,14 +720,14 @@ static unsigned long pq_threadidcallback(void) { /* - * This is not standards-compliant. pthread_self() returns pthread_t, and - * shouldn't be cast to unsigned long, but CRYPTO_set_id_callback requires - * it, so we have to do it. + * This is not standards-compliant. pg_thrd_current() returns pg_thrd_t, + * and shouldn't be cast to unsigned long, but CRYPTO_set_id_callback + * requires it, so we have to do it. */ - return (unsigned long) pthread_self(); + return (unsigned long) pg_thrd_current(); } -static pthread_mutex_t *pq_lockarray; +static pg_mtx_t *pq_lockarray; static void pq_lockingcallback(int mode, int n, const char *file, int line) @@ -744,12 +739,12 @@ pq_lockingcallback(int mode, int n, const char *file, int line) */ if (mode & CRYPTO_LOCK) { - if (pthread_mutex_lock(&pq_lockarray[n])) + if (pg_mtx_lock(&pq_lockarray[n]) != pg_thrd_success) Assert(false); } else { - if (pthread_mutex_unlock(&pq_lockarray[n])) + if (pg_mtx_unlock(&pq_lockarray[n]) != pg_thrd_success) Assert(false); } } @@ -768,7 +763,7 @@ pq_lockingcallback(int mode, int n, const char *file, int line) int pgtls_init(PGconn *conn, bool do_ssl, bool do_crypto) { - if (pthread_mutex_lock(&ssl_config_mutex)) + if (pg_mtx_lock(&ssl_config_mutex) != pg_thrd_success) return -1; #ifdef HAVE_CRYPTO_LOCK @@ -782,19 +777,19 @@ pgtls_init(PGconn *conn, bool do_ssl, bool do_crypto) { int i; - pq_lockarray = malloc(sizeof(pthread_mutex_t) * CRYPTO_num_locks()); + pq_lockarray = malloc(sizeof(pg_mtx_t) * CRYPTO_num_locks()); if (!pq_lockarray) { - pthread_mutex_unlock(&ssl_config_mutex); + pg_mtx_unlock(&ssl_config_mutex); return -1; } for (i = 0; i < CRYPTO_num_locks(); i++) { - if (pthread_mutex_init(&pq_lockarray[i], NULL)) + if (pg_mtx_init(&pq_lockarray[i], NULL) != pg_thrd_success) { free(pq_lockarray); pq_lockarray = NULL; - pthread_mutex_unlock(&ssl_config_mutex); + pg_mtx_unlock(&ssl_config_mutex); return -1; } } @@ -835,7 +830,7 @@ pgtls_init(PGconn *conn, bool do_ssl, bool do_crypto) ssl_lib_initialized = true; } - pthread_mutex_unlock(&ssl_config_mutex); + pg_mtx_unlock(&ssl_config_mutex); return 0; } @@ -855,7 +850,7 @@ static void destroy_ssl_system(void) { #if defined(HAVE_CRYPTO_LOCK) - if (pthread_mutex_lock(&ssl_config_mutex)) + if (pg_mtx_lock(&ssl_config_mutex) != pg_thrd_success) return; if (pq_init_crypto_lib && crypto_open_connections > 0) @@ -881,7 +876,7 @@ destroy_ssl_system(void) */ } - pthread_mutex_unlock(&ssl_config_mutex); + pg_mtx_unlock(&ssl_config_mutex); #endif } @@ -1973,7 +1968,7 @@ my_BIO_s_socket(void) { BIO_METHOD *res; - if (pthread_mutex_lock(&ssl_config_mutex)) + if (pg_mtx_lock(&ssl_config_mutex) != pg_thrd_success) return NULL; res = my_bio_methods; @@ -2018,7 +2013,7 @@ my_BIO_s_socket(void) } my_bio_methods = res; - pthread_mutex_unlock(&ssl_config_mutex); + pg_mtx_unlock(&ssl_config_mutex); return res; err: @@ -2029,7 +2024,7 @@ err: if (res) free(res); #endif - pthread_mutex_unlock(&ssl_config_mutex); + pg_mtx_unlock(&ssl_config_mutex); return NULL; } diff --git a/src/interfaces/libpq/fe-secure.c b/src/interfaces/libpq/fe-secure.c index f628082337..ea4b1e6288 100644 --- a/src/interfaces/libpq/fe-secure.c +++ b/src/interfaces/libpq/fe-secure.c @@ -35,9 +35,7 @@ #include -#ifdef WIN32 -#include "pthread-win32.h" -#else +#ifndef WIN32 #include #endif diff --git a/src/interfaces/libpq/libpq-int.h b/src/interfaces/libpq/libpq-int.h index 8ed1b28fcc..ce78898b48 100644 --- a/src/interfaces/libpq/libpq-int.h +++ b/src/interfaces/libpq/libpq-int.h @@ -31,11 +31,6 @@ #include #endif -#ifdef WIN32 -#include "pthread-win32.h" -#else -#include -#endif #include /* include stuff common to fe and be */ diff --git a/src/interfaces/libpq/meson.build b/src/interfaces/libpq/meson.build index 7623aeadab..89e39340d9 100644 --- a/src/interfaces/libpq/meson.build +++ b/src/interfaces/libpq/meson.build @@ -22,7 +22,7 @@ libpq_sources = files( libpq_so_sources = [] # for shared lib, in addition to the above if host_system == 'windows' - libpq_sources += files('pthread-win32.c', 'win32.c') + libpq_sources += files('win32.c') libpq_so_sources += rc_lib_gen.process(win32ver_rc, extra_args: [ '--NAME', 'libpq', '--FILEDESC', 'PostgreSQL Access Library',]) @@ -44,8 +44,7 @@ export_file = custom_target('libpq.exports', kwargs: gen_export_kwargs, ) -# port needs to be in include path due to pthread-win32.h -libpq_inc = include_directories('.', '../../port') +libpq_inc = include_directories('.') libpq_c_args = ['-DSO_MAJOR_VERSION=5'] # Not using both_libraries() here as diff --git a/src/interfaces/libpq/pthread-win32.c b/src/interfaces/libpq/pthread-win32.c deleted file mode 100644 index b40872898d..0000000000 --- a/src/interfaces/libpq/pthread-win32.c +++ /dev/null @@ -1,66 +0,0 @@ -/*------------------------------------------------------------------------- -* -* pthread-win32.c -* partial pthread implementation for win32 -* -* Copyright (c) 2004-2024, PostgreSQL Global Development Group -* IDENTIFICATION -* src/interfaces/libpq/pthread-win32.c -* -*------------------------------------------------------------------------- -*/ - -#include "postgres_fe.h" - -#include "pthread-win32.h" - -DWORD -pthread_self(void) -{ - return GetCurrentThreadId(); -} - -void -pthread_setspecific(pthread_key_t key, void *val) -{ -} - -void * -pthread_getspecific(pthread_key_t key) -{ - return NULL; -} - -int -pthread_mutex_init(pthread_mutex_t *mp, void *attr) -{ - mp->initstate = 0; - return 0; -} - -int -pthread_mutex_lock(pthread_mutex_t *mp) -{ - /* Initialize the csection if not already done */ - if (mp->initstate != 1) - { - LONG istate; - - while ((istate = InterlockedExchange(&mp->initstate, 2)) == 2) - Sleep(0); /* wait, another thread is doing this */ - if (istate != 1) - InitializeCriticalSection(&mp->csection); - InterlockedExchange(&mp->initstate, 1); - } - EnterCriticalSection(&mp->csection); - return 0; -} - -int -pthread_mutex_unlock(pthread_mutex_t *mp) -{ - if (mp->initstate != 1) - return EINVAL; - LeaveCriticalSection(&mp->csection); - return 0; -} diff --git a/src/port/Makefile b/src/port/Makefile index db7c02117b..617d7e16d7 100644 --- a/src/port/Makefile +++ b/src/port/Makefile @@ -46,6 +46,7 @@ OBJS = \ path.o \ pg_bitutils.o \ pg_strong_random.o \ + pg_threads.o \ pgcheckdir.o \ pgmkdirp.o \ pgsleep.o \ diff --git a/src/port/meson.build b/src/port/meson.build index ff54b7b53e..3df9b1c9b5 100644 --- a/src/port/meson.build +++ b/src/port/meson.build @@ -8,6 +8,7 @@ pgport_sources = [ 'path.c', 'pg_bitutils.c', 'pg_strong_random.c', + 'pg_threads.c', 'pgcheckdir.c', 'pgmkdirp.c', 'pgsleep.c', @@ -73,10 +74,6 @@ replace_funcs_neg = [ ['strsep'], ] -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/pg_threads.c b/src/port/pg_threads.c new file mode 100644 index 0000000000..a40910715b --- /dev/null +++ b/src/port/pg_threads.c @@ -0,0 +1,436 @@ +/*------------------------------------------------------------------------- + * + * pg_threads.c + * Out-of-line parts of portable multi-threading API. + * + * Portions Copyright (c) 1996-2024, PostgreSQL Global Development Group + * + * IDENTIFICATION + * src/port/pg_threads.c + * + *------------------------------------------------------------------------- + */ +#include "c.h" +#include "port/pg_threads.h" + +#include +#include +#include + + +/* XXX TODO: make atomics avialable in frontend so we can use these! */ +#define pg_read_barrier() +#define pg_write_barrier() + + +/*------------------------------------------------------------------------- + * + * Threads. + * + * There are small differences between the function types in C11, + * POSIX (return type) and Windows (return type signedness, calling + * convention). The int return value will survive casting to/from + * void * and DWORD respectively, but we still need a small trampoline + * function to deal with the different function pointer type. + * + *------------------------------------------------------------------------- + */ + +typedef struct pg_thrd_thunk +{ + pg_thrd_start_t function; + void *argument; +} pg_thrd_thunk; + +/* + * A trampoline function, to handle calling convention and parameter + * variations in the native APIs. + */ +#ifdef PG_THREADS_WIN32 +static DWORD __stdcall +pg_thrd_body(void *vthunk) +#else +static void * +pg_thrd_body(void *vthunk) +#endif +{ + pg_thrd_thunk *thunk = (pg_thrd_thunk *) vthunk; + void *argument = thunk->argument; + pg_thrd_start_t function = thunk->function; + int result; + + free(vthunk); + + result = function(argument); + +#ifdef PG_THREADS_WIN32 + return (DWORD) result; +#else + return (void *) (intptr_t) result; +#endif +} + +int +pg_thrd_create(pg_thrd_t *thread, pg_thrd_start_t function, void *argument) +{ + pg_thrd_thunk *thunk; + + thunk = malloc(sizeof(*thunk)); + if (thunk == NULL) + return pg_thrd_nomem; + thunk->function = function; + thunk->argument = argument; + +#ifdef WIN32 + *thread = CreateThread(NULL, 0, pg_thrd_body, thunk, 0, 0); + if (*thread != NULL) + return pg_thrd_success; +#else + if (pthread_create(thread, NULL, pg_thrd_body, thunk) == 0) + return pg_thrd_success; +#endif + + free(thunk); + return pg_thrd_error; +} + +int +pg_thrd_join(pg_thrd_t thread, int *result) +{ +#ifdef WIN32 + DWORD dword_result; + + if (WaitForSingleObject(thread, INFINITE) == WAIT_OBJECT_0) + { + if (result) + { + if (!GetExitCodeThread(thread, &dword_result)) + return pg_thrd_error; + *result = (int) dword_result; + } + CloseHandle(thread); + return pg_thrd_success; + } +#else + void *void_star_result; + + if (pthread_join(thread, &void_star_result) == 0) + { + if (result) + *result = (int) (intptr_t) void_star_result; + return pg_thrd_success; + } +#endif + return pg_thrd_error; +} + +void +pg_thrd_exit(int result) +{ +#ifdef WIN32 + ExitThread((DWORD) result); +#else + pthread_exit((void *) (intptr_t) result); +#endif +} + + +/*------------------------------------------------------------------------- + * + * Initialization functions. + * + *------------------------------------------------------------------------- + */ + +#ifdef WIN32 +BOOL CALLBACK +pg_call_once_trampoline(pg_once_flag *flag, void *parameter, void **context) +{ + pg_call_once_function_t function = (pg_call_once_function_t) parameter; + + function(); + return TRUE; +} +#endif + + +/*------------------------------------------------------------------------- + * + * Thread-specific storage. + * + * This extra support code is only needed when we can't use the native + * support for thread-local storage destructors. Normally that is + * Windows, due to incompatible calling conventions, but this code + * path can be activated on POSIX systems too for testing. + * + *------------------------------------------------------------------------- + */ + +#ifdef PG_THREADS_NEED_DESTRUCTOR_TABLE + +struct dtor_table_entry +{ + pg_tss_t tss_id; + pg_tss_dtor_t function; +}; + +static pg_rwlock_t dtor_table_lock = PG_RWLOCK_STATIC_INIT; +static size_t dtor_table_count = 0; +static size_t dtor_table_capacity = 0; +static struct dtor_table_entry *dtor_table; + +/* One native TLS key with a native destructor, which drives all others. */ +static pg_tss_t pg_tss_destructor_hook; +static bool pg_tss_run_destructors_installed; + +/* + * Helper function for recording the destructor for a given tss_id. + * Returns true on success, or false if the table is full. + */ +static bool +pg_tss_dtor_set(pg_tss_t tss_id, pg_tss_dtor_t destructor) +{ + bool have_space = true; + + pg_rwlock_wrlock(&dtor_table_lock); + + /* Make sure we have space, or fail. */ + if (dtor_table_count == dtor_table_capacity) + { + struct dtor_table_entry *new_dtor_table; + size_t new_dtor_table_capacity; + + new_dtor_table_capacity = Max(1, dtor_table_capacity * 2); + new_dtor_table = malloc(sizeof(dtor_table[0]) * new_dtor_table_capacity); + if (new_dtor_table == NULL) + { + /* Out of memory. */ + have_space = false; + } + else + { + if (dtor_table_count > 0) + { + memcpy(new_dtor_table, + dtor_table, + sizeof(dtor_table[0]) * dtor_table_count); + free(dtor_table); + } + dtor_table = new_dtor_table; + dtor_table_capacity = new_dtor_table_capacity; + } + } + +#ifdef USE_ASSERTION_CHECKING + /* We don't expect to see this ID already in the table. */ + for (size_t i = 0; i < dtor_table_count; i++) + Assert(dtor_table[i].tss_id != tss_id); +#endif + + /* Store it. */ + if (have_space) + { + Assert(dtor_table_count < dtor_table_capacity); + dtor_table[dtor_table_count].tss_id = tss_id; + dtor_table[dtor_table_count].function = destructor; + + dtor_table_count++; + } + + pg_wrlock_unlock(&dtor_table_lock); + + return have_space; +} + +/* + * The destructor installed for the single special FLS value that will + * be called by Windows (or POSIX if we are using the special test + * mode). This must have CALLBACK calling convention on Windows, + * which is the reason we can't just use its FlsAlloc() destructors + * directly for pg_tss_create(). + */ +static void +#ifdef WIN32 + CALLBACK +#endif +pg_tss_run_destructors(void *data) +{ + pg_rwlock_rdlock(&dtor_table_lock); + + for (int i = 0; i < PG_TSS_DTOR_ITERATIONS; ++i) + { + bool seen_non_null_value = false; + + for (size_t slot = 0; slot < dtor_table_count; ++slot) + { + pg_tss_t tss_id = dtor_table[slot].tss_id; + void *value = pg_tss_get(tss_id); + + if (value) + { + pg_tss_dtor_t function = dtor_table[slot].function; + + Assert(function); + + /* Clear value. */ + pg_tss_set(tss_id, NULL); + + /* + * We'll need to go around again to make sure that a + * destructor called in this iteration didn't set something. + */ + seen_non_null_value = true; + + /* Unlock while running the destructor. */ + pg_rdlock_unlock(&dtor_table_lock); + function(value); + pg_rwlock_rdlock(&dtor_table_lock); + } + } + + /* If we didn't see any values, we're finished. */ + if (!seen_non_null_value) + break; + } + pg_rdlock_unlock(&dtor_table_lock); +} + +static void +pg_tss_install_run_destructors(void) +{ + /* + * We need a way to make sure our TSS destructors run at thread exit, even + * if the thread exits via native calls instead of our own pg_thrd_exit() + * or trampoline function. So we register one real native 'hook' + * destructor that will then call all the destructors in our own + * destructor table. + * + * On Windows, we use FlsAlloc(), not TlsAlloc(), because that supports + * destructors. Unforunately they have the wrong calling convention, or + * we could simply use them directly instead of doing all this extra work. + */ +#ifdef PG_THREADS_WIN32 + pg_tss_destructor_hook = FlsAlloc(pg_tss_run_destructors); + if (pg_tss_destructor_hook == FLS_OUT_OF_INDEXES) + return; +#else + if (pthread_key_create(&pg_tss_destructor_hook, pg_tss_run_destructors) != 0) + return; +#endif + pg_write_barrier(); + pg_tss_run_destructors_installed = true; + + /* + * Make sure that any thread that receives a pg_tss_t and might store a + * value can see that there is now potentially a registered destructor. + */ + pg_write_barrier(); +} + +/* + * Called every time pg_tss_set() installs a non-NULL value. + */ +void +pg_tss_ensure_destructors_in_this_thread(void) +{ + /* + * Pairs with pg_tss_install_run_destructors(), called by pg_tss_create(). + * This makes sure that we know if the tss_id being set could possibly + * have a destructor. We don't want to pay the cost of checking, but we + * can check with a simple load if *any* tss_id has a destructor. If so, + * we make sure that pg_tss_destructor_hook has a non-NULL value in *this* + * thread, because both Windows and POSIX will only call a destructor for + * a non-NULL value. + */ + pg_read_barrier(); + if (pg_tss_run_destructors_installed) + { +#ifdef PG_THREADS_WIN32 + if (FlsGetValue(pg_tss_destructor_hook) == NULL) + FlsSetValue(pg_tss_destructor_hook, (void *) 1); +#else + if (pthread_getspecific(pg_tss_destructor_hook) == NULL) + pthread_setspecific(pg_tss_destructor_hook, (void *) 1); +#endif + } +} +#endif + +int +pg_tss_create(pg_tss_t *tss_id, pg_tss_dtor_t destructor) +{ +#ifdef PG_THREADS_NEED_DESTRUCTOR_TABLE + static pg_once_flag destructor_cleanup_once; + + /* + * Make sure our destructor hook is registered with the operating system + * in this process. This happens only once in the whole process. Making + * sure it will run actually in each thread happens in + * pg_tss_ensure_destructors_will_run(). + */ + pg_call_once(&destructor_cleanup_once, pg_tss_install_run_destructors); + if (!pg_tss_run_destructors_installed) + return pg_thrd_error; +#endif + +#ifdef PG_THREADS_WIN32 + /* Windows native TSL, our own destructors machinery. */ + *tss_id = TlsAlloc(); + if (*tss_id == TLS_OUT_OF_INDEXES) + return pg_thrd_error; +#elif defined(PG_THREADS_NEED_DESTRUCTOR_TABLE) + /* POSIX, but testing our own destructor machinery. */ + if (pthread_key_create(tss_id, NULL) != 0) + return pg_thrd_error; +#else + /* POSIX handles destructors. */ + return pg_thrd_maperror(pthread_key_create(tss_id, destructor)); +#endif + +#ifdef PG_THREADS_NEED_DESTRUCTOR_TABLE + /* Allocate destructor table entry, or fail and clean up. */ + if (destructor &&!pg_tss_dtor_set(*tss_id, destructor)) + { +#ifdef PG_THREADS_WIN32 + TlsFree(*tss_id); +#else + pthread_key_delete(*tss_id); +#endif + return pg_thrd_error; + } +#endif + + return pg_thrd_success; +} + +void +pg_tss_dtor_delete(pg_tss_t tss_id) +{ +#ifdef PG_THREADS_NEED_DESTRUCTOR_TABLE + /* + * We have to search the destructor table linearly, but deleting IDs is + * probably very rare so that's OK. + */ + pg_rwlock_wrlock(&dtor_table_lock); + for (size_t i = 0; i < dtor_table_count; ++i) + { + if (dtor_table[i].tss_id == tss_id) + { + /* Move the other values to compact the table. */ + if (i < dtor_table_count - 1) + memmove(&dtor_table[i], + &dtor_table[i + 1], + sizeof(dtor_table[i]) * (dtor_table_count - i - 1)); + dtor_table_count--; + break; + } + } + pg_wrlock_unlock(&dtor_table_lock); +#endif + +#ifdef PG_THREADS_WIN32 + TlsFree(tss_id); +#else + pthread_key_delete(tss_id); +#endif +} diff --git a/src/port/pthread-win32.h b/src/port/pthread-win32.h deleted file mode 100644 index 5f33269057..0000000000 --- a/src/port/pthread-win32.h +++ /dev/null @@ -1,31 +0,0 @@ -/* - * src/port/pthread-win32.h - */ -#ifndef __PTHREAD_H -#define __PTHREAD_H - -typedef ULONG pthread_key_t; - -typedef struct pthread_mutex_t -{ - /* initstate = 0: not initialized; 1: init done; 2: init in progress */ - LONG initstate; - CRITICAL_SECTION csection; -} pthread_mutex_t; - -#define PTHREAD_MUTEX_INITIALIZER { 0 } - -typedef int pthread_once_t; - -DWORD pthread_self(void); - -void pthread_setspecific(pthread_key_t, void *); -void *pthread_getspecific(pthread_key_t); - -int pthread_mutex_init(pthread_mutex_t *, void *attr); -int pthread_mutex_lock(pthread_mutex_t *); - -/* blocking */ -int pthread_mutex_unlock(pthread_mutex_t *); - -#endif diff --git a/src/port/pthread_barrier_wait.c b/src/port/pthread_barrier_wait.c deleted file mode 100644 index 835dbf1c7a..0000000000 --- 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-2024, 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; -} diff --git a/src/tools/pginclude/headerscheck b/src/tools/pginclude/headerscheck index 436e2b92a3..03faaefb77 100755 --- a/src/tools/pginclude/headerscheck +++ b/src/tools/pginclude/headerscheck @@ -102,7 +102,6 @@ do test "$f" = src/include/port/win32_msvc/dirent.h && continue test "$f" = src/include/port/win32_msvc/utime.h && continue test "$f" = src/include/port/win32ntdll.h && continue - test "$f" = src/port/pthread-win32.h && continue # Likewise, these files are platform-specific, and the one # relevant to our platform will be included by atomics.h. diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 547d14b3e7..b4e5356468 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -3676,12 +3676,14 @@ pgParameterStatus pg_atomic_flag pg_atomic_uint32 pg_atomic_uint64 +pg_barrier_t pg_be_sasl_mech pg_case_map pg_category_range pg_checksum_context pg_checksum_raw_context pg_checksum_type +pg_cnd_t pg_compress_algorithm pg_compress_specification pg_conn_host @@ -3706,7 +3708,10 @@ pg_local_to_utf_combined pg_locale_t pg_mb_radix_tree pg_md5_ctx +pg_mtx_t +pg_mtx_type_t pg_on_exit_callback +pg_once_flag pg_prng_state pg_re_flags pg_regex_t @@ -3720,8 +3725,12 @@ pg_sha384_ctx pg_sha512_ctx pg_snapshot pg_stack_base_t +pg_thrd_error_t +pg_thrd_t +pg_thrd_thunk pg_time_t pg_time_usec_t +pg_tss_t pg_tz pg_tz_cache pg_tzenum -- 2.39.2