On Sun, Apr 14, 2024 at 3:16 PM Thomas Munro <thomas.mu...@gmail.com> wrote: > So I'll go ahead and add the storage class to the next version, and > contemplate a couple of different options for the tss stuff, including > perhaps leaving it out if that seems doable.
Here is a new attempt at pg_threads.h. Still work in progress, but passing tests, with storage class and working TSS, showing various users. I eventually understood first that my TSS destructor problems on Windows came from mismatched calling conventions, and that you can't really trampoline your way around that, at least not without doing some pretty unreasonable things, and that is why nobody can emulate either tss_create() or pthread_key_create() directly with Windows' FlsAlloc(), so everybody who tries finishes up building their own infrastructure to track destructors, or in ECPG's case just leaks all the memory instead. Here's the simplest implementation I could come up with so far. I don't have Windows so I made it possible to use emulated TSS destructors on my local machine with a special macro for testing, and then argued with CI for a while until the other machines agreed. Otherwise, it's all a fairly thin wrapper and hopefully not suprising. In one place, an ECPG thread-local variable has no destructor, so we can use it as the first example of the new pg_thread_local storage class. One thing this would need to be complete, at least the way I've implemented it, is memory barriers, for non-TSO hardware, which would require lifting the ban on atomics.h in frontend code, or at least parts of it. Only 64 bit emulation is actually tied to the backend now (because it calls spinlock stuff, that itself is backend-only, but also it doesn't actually need to be either). Or maybe I can figure out a different scheme that doesn't need that. Or something... WIP patch attached.
From ab649c89ca924c7aae39b56a3706d9f934e4e1ef Mon Sep 17 00:00:00 2001 From: Thomas Munro <thomas.mu...@gmail.com> 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 <threads.h>, 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 <and...@anarazel.de> (earlier versions) Reviewed-by: Heikki Linnakangas <heikki.linnakan...@iki.fi> 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 <math.h> 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 <windows.h> -#define GETERRNO() (_dosmaperr(GetLastError()), errno) -#define THREAD_T HANDLE -#define THREAD_FUNC_RETURN_TYPE unsigned -#define THREAD_FUNC_RETURN return 0 -#define THREAD_FUNC_CC __stdcall -#define THREAD_CREATE(handle, function, arg) \ - ((*(handle) = (HANDLE) _beginthreadex(NULL, 0, (function), (arg), 0, NULL)) == 0 ? errno : 0) -#define THREAD_JOIN(handle) \ - (WaitForSingleObject(handle, INFINITE) != WAIT_OBJECT_0 ? \ - GETERRNO() : CloseHandle(handle) ? 0 : GETERRNO()) -#define THREAD_BARRIER_T SYNCHRONIZATION_BARRIER -#define THREAD_BARRIER_INIT(barrier, n) \ - (InitializeSynchronizationBarrier((barrier), (n), 0) ? 0 : GETERRNO()) -#define THREAD_BARRIER_WAIT(barrier) \ - EnterSynchronizationBarrier((barrier), \ - SYNCHRONIZATION_BARRIER_FLAGS_BLOCK_ONLY) -#define THREAD_BARRIER_DESTROY(barrier) -#else -/* Use POSIX threads */ -#include "port/pg_pthread.h" -#define THREAD_T pthread_t -#define THREAD_FUNC_RETURN_TYPE void * -#define THREAD_FUNC_RETURN return NULL -#define THREAD_FUNC_CC -#define THREAD_CREATE(handle, function, arg) \ - pthread_create((handle), NULL, (function), (arg)) -#define THREAD_JOIN(handle) \ - pthread_join((handle), NULL) -#define THREAD_BARRIER_T pthread_barrier_t -#define THREAD_BARRIER_INIT(barrier, n) \ - pthread_barrier_init((barrier), NULL, (n)) -#define THREAD_BARRIER_WAIT(barrier) pthread_barrier_wait((barrier)) -#define THREAD_BARRIER_DESTROY(barrier) pthread_barrier_destroy((barrier)) -#endif - - /******************************************************************** * some configurable parameters */ @@ -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 <pthread.h> to be included by every translation - * unit. - * - *------------------------------------------------------------------------- - */ - -#ifndef PG_PTHREAD_H -#define PG_PTHREAD_H - -#include <pthread.h> - -#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 <threads.h> 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 <windows.h> +#else +#include <pthread.h> +#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 <limits.h> #include <unistd.h> -#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 <pthread.h> -#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 <stdbool.h> 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 <stdint.h> #include <stdlib.h> #include "ecpg_config.h" - -#ifdef WIN32 -#define WIN32_LEAN_AND_MEAN -#include <windows.h> -#include <process.h> -#include <locale.h> -#else -#include <pthread.h> -#endif -#include <stdio.h> +#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 <windows.h> -#include <process.h> -#include <locale.h> -#else -#include <pthread.h> -#endif -#include <stdio.h> +#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 <stdint.h> #include <stdlib.h> #include "ecpg_config.h" +#include "port/pg_threads.h" -#ifdef WIN32 -#define WIN32_LEAN_AND_MEAN -#include <windows.h> -#include <process.h> -#include <locale.h> -#else -#include <pthread.h> -#endif #include <stdio.h> #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 <stdint.h> #include <stdlib.h> #include "ecpg_config.h" - -#ifndef WIN32 -#include <pthread.h> -#else -#include <windows.h> -#include <locale.h> -#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 <stdint.h> #include <stdlib.h> #include "ecpg_config.h" - -#ifndef WIN32 -#include <pthread.h> -#else -#include <windows.h> -#include <locale.h> -#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 <stdint.h> #include <stdlib.h> #include "ecpg_config.h" - -#ifdef WIN32 -#define WIN32_LEAN_AND_MEAN -#include <windows.h> -#include <process.h> -#include <locale.h> -#else -#include <pthread.h> -#endif -#include <stdio.h> +#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 <windows.h> -#include <process.h> -#include <locale.h> -#else -#include <pthread.h> -#endif -#include <stdio.h> +#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 <stdint.h> #include <stdlib.h> #include "ecpg_config.h" +#include "port/pg_threads.h" -#ifdef WIN32 -#define WIN32_LEAN_AND_MEAN -#include <windows.h> -#include <process.h> -#include <locale.h> -#else -#include <pthread.h> -#endif #include <stdio.h> #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 <stdint.h> #include <stdlib.h> #include "ecpg_config.h" - -#ifndef WIN32 -#include <pthread.h> -#else -#include <windows.h> -#include <locale.h> -#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 <stdint.h> #include <stdlib.h> #include "ecpg_config.h" - -#ifndef WIN32 -#include <pthread.h> -#else -#include <windows.h> -#include <locale.h> -#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 <netinet/tcp.h> #endif -#ifdef WIN32 -#include "pthread-win32.h" -#else -#include <pthread.h> -#endif - #ifdef USE_LDAP #ifdef WIN32 #include <winldap.h> @@ -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 <sys/stat.h> -#ifdef WIN32 -#include "pthread-win32.h" -#else -#include <pthread.h> -#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 <sys/stat.h> -#ifdef WIN32 -#include "pthread-win32.h" -#else +#ifndef WIN32 #include <pthread.h> #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 <sys/time.h> #endif -#ifdef WIN32 -#include "pthread-win32.h" -#else -#include <pthread.h> -#endif #include <signal.h> /* 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 <errno.h> +#include <stdlib.h> +#include <string.h> + + +/* 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
