From c81a80121e962e8cc452eae1e686a9329d461343 Mon Sep 17 00:00:00 2001 From: Thomas Munro Date: Wed, 24 Jun 2026 17:05:51 +1200 Subject: [PATCH v2 02/11] port: Provide pg_threads.h API. This is based on C11 , which is not available on all systems we support and a little light on features, but still provides a standards-based way to nail down a lot of gnarly details and avoid having to invent our own strange names for things. Our API has pg_ prefixes on all identifiers, since client code might be using directly. The thread_local macro is an exception and doesn't get a prefix, since there is no doubt about its definition. Three implementations are provided to redirect the API to underlying threading APIs: * , a thin wrapper used on POSIX systems * , an very thin wrapper used on Windows + Visual Studio * , a less thin wrapper used on Windows + MinGW Some local extensions are also provided in pg_threads_ext.h: * pg_rwlock_t * pg_thrd_barrier_t * pg_thrd_error_string() * pg_thrd_error_string_with_detail() * PG_CND_INIT * PG_MTX_PLAIN_INIT Later commits will make use of this API as a centralized replacement for several module-local thread API abstractions. Discussion: https://postgr.es/m/CA%2BhUKG%2BK%2BLmYtz54%2BSv1j5Zu1Qo6szzAMbMyuKSeQmb7De4Ewg%40mail.gmail.com Reviewed-by: Heikki Linnakangas Reviewed-by: Bryan Green --- src/include/port/pg_threads.h | 146 ++++ src/include/port/pg_threads/map_pthread.h | 111 +++ .../port/pg_threads/map_pthread_error.h | 31 + src/include/port/pg_threads/map_pthread_ext.h | 85 ++ src/include/port/pg_threads/map_threads.h | 118 +++ src/include/port/pg_threads/map_threads_ext.h | 101 +++ src/include/port/pg_threads/map_windows.h | 192 +++++ .../port/pg_threads/map_windows_error.h | 31 + src/include/port/pg_threads/map_windows_ext.h | 133 +++ src/include/port/pg_threads/pg_thrd_barrier.h | 76 ++ src/include/port/pg_threads/wrap.h | 114 +++ src/include/port/pg_threads_ext.h | 85 ++ src/port/Makefile | 1 + src/port/meson.build | 5 +- src/port/pg_threads.c | 812 ++++++++++++++++++ src/test/regress/expected/port.out | 17 + src/test/regress/parallel_schedule | 2 +- src/test/regress/regress.c | 9 + src/test/regress/sql/port.sql | 16 + src/tools/pginclude/headerscheck | 3 + src/tools/pgindent/typedefs.list | 17 + 21 files changed, 2102 insertions(+), 3 deletions(-) create mode 100644 src/include/port/pg_threads.h create mode 100644 src/include/port/pg_threads/map_pthread.h create mode 100644 src/include/port/pg_threads/map_pthread_error.h create mode 100644 src/include/port/pg_threads/map_pthread_ext.h create mode 100644 src/include/port/pg_threads/map_threads.h create mode 100644 src/include/port/pg_threads/map_threads_ext.h create mode 100644 src/include/port/pg_threads/map_windows.h create mode 100644 src/include/port/pg_threads/map_windows_error.h create mode 100644 src/include/port/pg_threads/map_windows_ext.h create mode 100644 src/include/port/pg_threads/pg_thrd_barrier.h create mode 100644 src/include/port/pg_threads/wrap.h create mode 100644 src/include/port/pg_threads_ext.h create mode 100644 src/port/pg_threads.c create mode 100644 src/test/regress/expected/port.out create mode 100644 src/test/regress/sql/port.sql diff --git a/src/include/port/pg_threads.h b/src/include/port/pg_threads.h new file mode 100644 index 00000000000..2567251c571 --- /dev/null +++ b/src/include/port/pg_threads.h @@ -0,0 +1,146 @@ +/*------------------------------------------------------------------------- + * + * pg_threads.h + * Portable implementation of C11 with pg_ prefixes. + * + * This provides a minimal multithreading API that works on all our target + * systems. By following even when we're not using it directly, + * we avoid the need to make up our own names and details, and also gain some + * future-proofing. Adding a prefix avoids collision with , if + * included by client programs. + * + * See also pg_threads_ext.h, which adds a small number of extra facilities in + * the same style, most usefully static initializers for mutexes. This header + * restricts itself to faithfully copying . + * + * Three implementations are available: + * + * (1) map_pthread.h for on POSIX systems + * (2) map_threads.h for on Windows + Visual Studio + * (3) map_windows.h for on Windows + * + * The choice is made automatically, but can be overridden for testing. + * PG_THREADS_USE_THREADS_H works on common POSIX systems, but such builds are + * intended strictly for testing (see map_threads_ext.h for details). + * + * The following macros are defined to signal known failures to conform to the + * specification on some systems. See the implementation headers + * for details. + * + * - PG_MTX_TIMED_NOT_SUPPORTED: macOS (1), Windows (3) + * - PG_THRD_CURRENT_NOT_DETACHABLE: Windows (2 and 3) + * - PG_THRD_CURRENT_NOT_JOINABLE: Windows (2 and 3) + * + * https://www.open-std.org/jtc1/sc22/wg14/www/docs/n1548.pdf + * https://pubs.opengroup.org/onlinepubs/9799919799/basedefs/threads.h.html + * https://en.cppreference.com/c/header/threads + * + * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group + * + * IDENTIFICATION + * src/include/port/pg_threads.h + * + *------------------------------------------------------------------------- + */ +#ifndef PG_THREADS_H +#define PG_THREADS_H + +/* select implementation, if not defined manually for testing. */ +#if !defined(PG_THREADS_USE_PTHREAD_H) && \ + !defined(PG_THREADS_USE_THREADS_H) && \ + !defined(PG_THREADS_USE_WINDOWS_H) +#if !defined(WIN32) +#define PG_THREADS_USE_PTHREAD_H +#elif defined(_MSC_VER) +#define PG_THREADS_USE_THREADS_H +#else +#define PG_THREADS_USE_WINDOWS_H +#endif +#endif + +#include + +/* standard thread_local macro if needed */ +#if defined(__STDC_VERSION__) && __STDC_VERSION__ < 202311L +#ifndef thread_local +#define thread_local _Thread_local +#endif +#endif + +/* standard function pointer types with prefix */ +typedef int (*pg_thrd_start_t) (void *); +typedef void (*pg_tss_dtor_t) (void *); + +#if defined(PG_THREADS_USE_PTHREAD_H) +#include "port/pg_threads/map_pthread.h" +#elif defined(PG_THREADS_USE_THREADS_H) +#include "port/pg_threads/map_threads.h" +#elif defined(PG_THREADS_USE_WINDOWS_H) +#include "port/pg_threads/map_windows.h" +#endif + +/* standard types with prefix */ +typedef pg_cnd_impl pg_cnd_t; +typedef pg_mtx_impl pg_mtx_t; +typedef pg_once_flag_impl pg_once_flag; +typedef pg_thrd_impl pg_thrd_t; +typedef pg_tss_impl pg_tss_t; + +/* standard return values with prefix */ +enum +{ + pg_thrd_success = pg_thrd_success_impl, + pg_thrd_nomem = pg_thrd_nomem_impl, + pg_thrd_timedout = pg_thrd_timedout_impl, + pg_thrd_busy = pg_thrd_busy_impl, + pg_thrd_error = pg_thrd_error_impl, +}; + +/* standard mtx_t type flags with prefix */ +enum +{ + pg_mtx_plain = pg_mtx_plain_impl, + pg_mtx_recursive = pg_mtx_plain_impl, + pg_mtx_timed = pg_mtx_timed_impl, +}; + +/* standard macros with prefix */ +#define PG_ONCE_FLAG_INIT PG_ONCE_FLAG_INIT_IMPL +#define PG_TSS_DTOR_ITERATIONS PG_TSS_DTOR_ITERARATIONS_IMPL + +/* standard cnd_t functions with prefix */ +static inline int pg_cnd_broadcast(pg_cnd_t *cnd); +static inline void pg_cnd_destroy(pg_cnd_t *cnd); +static inline int pg_cnd_init(pg_cnd_t *cnd); +static inline int pg_cnd_signal(pg_cnd_t *cnd); +static inline int pg_cnd_timedwait(pg_cnd_t *cnd, pg_mtx_t *mtx, const struct timespec *time); +static inline int pg_cnd_wait(pg_cnd_t *cnd, pg_mtx_t *mtx); + +/* standard mtx_t functions with prefix */ +static inline void pg_mtx_destroy(pg_mtx_t *mtx); +static inline int pg_mtx_init(pg_mtx_t *mtx, int type); +static inline int pg_mtx_lock(pg_mtx_t *mtx); +static inline int pg_mtx_timedlock(pg_mtx_t *mtx, const struct timespec *time); +static inline int pg_mtx_trylock(pg_mtx_t *mtx); +static inline int pg_mtx_unlock(pg_mtx_t *mtx); + +/* standard once_flag functions with prefix */ +static inline void pg_call_once(pg_once_flag *flag, void (*fun) (void)); + +/* standard thrd_t functions with prefix */ +static inline int pg_thrd_create(pg_thrd_t *thr, pg_thrd_start_t fun, void *arg); +static inline pg_thrd_t pg_thrd_current(void); +static inline int pg_thrd_detach(pg_thrd_t thr); +static inline int pg_thrd_equal(pg_thrd_t thr0, pg_thrd_t thr1); +static inline void pg_thrd_exit(int result); +static inline int pg_thrd_join(pg_thrd_t thr, int *result); +static inline int pg_thrd_sleep(const struct timespec *duration, struct timespec *remaining); +static inline void pg_thrd_yield(void); + +/* standard tss_t functions with prefix */ +static inline int pg_tss_create(pg_tss_t *id, pg_tss_dtor_t); +static inline void pg_tss_delete(pg_tss_t id); +static inline void *pg_tss_get(pg_tss_t id); +static inline int pg_tss_set(pg_tss_t id, void *value); + +#endif diff --git a/src/include/port/pg_threads/map_pthread.h b/src/include/port/pg_threads/map_pthread.h new file mode 100644 index 00000000000..ccb5a2f906f --- /dev/null +++ b/src/include/port/pg_threads/map_pthread.h @@ -0,0 +1,111 @@ +/*------------------------------------------------------------------------- + * + * map_pthread.h + * Map pg_threads.h to POSIX . + * + * is effectively -light, and is also part of + * POSIX:2024. See especially its RATIONALE sections for discussion of how + * the two interfaces interact. In brief: + * + * - return value of thread functions is int rather than void pointer + * - threads can't be canceled + * - errors are simplified + * - several useful facilities are omitted (but see pg_threads_ext.h) + * + * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group + * + * IDENTIFICATION + * src/include/port/pg_threads/map_pthread.h + * + *------------------------------------------------------------------------- + */ +#ifndef PG_THREADS_PTHREAD_H +#define PG_THREADS_PTHREAD_H + +#ifndef PG_THREADS_H +#error "include pg_threads.h instead" +#endif + +#include +#include + +enum +{ + pg_thrd_success_impl = 0, + pg_thrd_nomem_impl, + pg_thrd_timedout_impl, + pg_thrd_busy_impl, + pg_thrd_error_impl, +}; + +/* These need the definition of pg_thrd_success_impl. */ +#include "port/pg_threads/map_pthread_error.h" +#include "port/pg_threads/wrap.h" + +#if !defined(_POSIX_TIMEOUTS) || _POSIX_TIMEOUTS < 0 +/* macOS lacks _POSIX_TIMEOUTS (required since POSIX:2018) */ +#define PG_MTX_TIMED_NOT_SUPPORTED +#endif + +typedef void (*pg_call_once_function_impl) (void); + +typedef pthread_cond_t pg_cnd_impl; +typedef pthread_mutex_t pg_mtx_impl; +typedef pthread_once_t pg_once_flag_impl; +typedef pthread_t pg_thrd_impl; +typedef pthread_key_t pg_tss_impl; + +enum +{ + pg_mtx_plain_impl = 0, + pg_mtx_recursive_impl = 1 << 1, + pg_mtx_timed_impl = 2 << 2, +}; + +#define PG_ONCE_FLAG_INIT_IMPL PTHREAD_ONCE_INIT +#ifdef PTHREAD_DESTRUCTOR_ITERATIONS +#define PG_TSS_DTOR_ITERATIONS_IMPL PTHREAD_DESTRUCTOR_ITERATIONS +#else +#define PG_TSS_DTOR_ITERATIONS_IMPL _POSIX_THREAD_DESTRUCTOR_ITERATIONS +#endif + +/* cnd_t */ +PG_THREADS_MAP(pg_cnd_broadcast, pthread_cond_broadcast, pg_cnd_impl *); +PG_THREADS_FORWARD_VOID(pg_cnd_destroy, pthread_cond_destroy, pg_cnd_impl *); +PG_THREADS_OUTOFLINE(pg_cnd_init, int, pg_cnd_impl *); +PG_THREADS_MAP(pg_cnd_signal, pthread_cond_signal, pg_cnd_impl *); +PG_THREADS_MAP(pg_cnd_timedwait, pthread_cond_timedwait, pg_cnd_impl *, pg_mtx_impl *, const struct timespec *); +PG_THREADS_MAP(pg_cnd_wait, pthread_cond_wait, pg_cnd_impl *, pg_mtx_impl *); + +/* mtx_t */ +PG_THREADS_FORWARD_VOID(pg_mtx_destroy, pthread_mutex_destroy, pg_mtx_impl *); +PG_THREADS_OUTOFLINE(pg_mtx_init, int, pg_mtx_impl *, int); +PG_THREADS_MAP(pg_mtx_lock, pthread_mutex_lock, pg_mtx_impl *); +#ifndef PG_MTX_TIMED_NOT_SUPPORTED +PG_THREADS_MAP(pg_mtx_timedlock, pthread_mutex_timedlock, pg_mtx_impl *, const struct timespec *); +#else +PG_THREADS_MAP_ERROR_UNSUPPORTED(pg_mtx_timedlock, pg_mtx_impl *, const struct timespec *); +#endif +PG_THREADS_MAP(pg_mtx_trylock, pthread_mutex_trylock, pg_mtx_impl *); +PG_THREADS_MAP(pg_mtx_unlock, pthread_mutex_unlock, pg_mtx_impl *); + +/* once_flag */ +PG_THREADS_FORWARD_VOID(pg_call_once, pthread_once, pg_once_flag_impl *, pg_call_once_function_impl); + +/* thrd_t */ +PG_THREADS_FORWARD_0ARG(pg_thrd_current, pthread_self, pg_thrd_impl); +PG_THREADS_OUTOFLINE(pg_thrd_create, int, pg_thrd_impl *, pg_thrd_start_t, void *); +PG_THREADS_MAP(pg_thrd_detach, pthread_detach, pg_thrd_impl); +PG_THREADS_OUTOFLINE_VOID(pg_thrd_exit, int); +PG_THREADS_FORWARD(pg_thrd_equal, pthread_equal, int, pg_thrd_impl, pg_thrd_impl); +PG_THREADS_OUTOFLINE(pg_thrd_join, int, pg_thrd_impl, int *); +PG_THREADS_FORWARD(pg_thrd_sleep, nanosleep, int, const struct timespec *, struct timespec *); +PG_THREADS_FORWARD_VOID_0ARG(pg_thrd_yield, sched_yield); + +/* tss_t */ +PG_THREADS_MAP(pg_tss_create, pthread_key_create, pg_tss_impl *, pg_tss_dtor_t); +PG_THREADS_FORWARD(pg_tss_get, pthread_getspecific, void *, pg_tss_impl); +PG_THREADS_MAP(pg_tss_set, pthread_setspecific, pg_tss_impl, void *); +PG_THREADS_FORWARD_VOID(pg_tss_delete, pthread_key_delete, pg_tss_impl); + +#endif diff --git a/src/include/port/pg_threads/map_pthread_error.h b/src/include/port/pg_threads/map_pthread_error.h new file mode 100644 index 00000000000..a04029bb800 --- /dev/null +++ b/src/include/port/pg_threads/map_pthread_error.h @@ -0,0 +1,31 @@ +/*------------------------------------------------------------------------- + * + * map_pthread_error.h + * Support for mapping errors. + * + * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group + * + * IDENTIFICATION + * src/include/port/pg_threads/map_pthread_error.h + * + *------------------------------------------------------------------------- + */ +#ifndef PG_THREADS_MAP_PTHREAD_ERROR_H +#define PG_THREADS_MAP_PTHREAD_ERROR_H + +#ifndef PG_THREADS_H +#error "include pg_threads.h instead" +#endif + +extern int pg_thrd_map_pthread_error(int pthread_error); +extern int pg_thrd_internal_error(const char *detail); + +static inline int +pg_thrd_map(int pthread_result) +{ + return pthread_result == 0 ? + pg_thrd_success_impl : + pg_thrd_map_pthread_error(pthread_result); +} + +#endif diff --git a/src/include/port/pg_threads/map_pthread_ext.h b/src/include/port/pg_threads/map_pthread_ext.h new file mode 100644 index 00000000000..3c591f15b55 --- /dev/null +++ b/src/include/port/pg_threads/map_pthread_ext.h @@ -0,0 +1,85 @@ +/*------------------------------------------------------------------------- + * + * map_pthread_ext.h + * Map pg_threads_ext.h to POSIX . + * + * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group + * + * IDENTIFICATION + * src/include/port/pg_threads/map_pthread_extensions.h + * + *------------------------------------------------------------------------- + */ +#ifndef PG_THREADS_MAP_PTHREAD_EXT_H +#define PG_THREADS_MAP_PTHREAD_EXT_H + +#ifndef PG_THREADS_EXT_H +#error "include pg_threads_ext.h instead" +#endif + +#include +#include + +#include "port/pg_threads/map_pthread_error.h" +#include "port/pg_threads/wrap.h" + +#if !defined(_POSIX_TIMEOUTS) || _POSIX_TIMEOUTS < 0 +/* macOS lacks _POSIX_TIMEOUTS (required since POSIX:2018) */ +#define PG_RWLOCK_TIMED_NOT_SUPPORTED +#endif + +typedef pthread_rwlock_t pg_rwlock_impl; + +#if !defined(_POSIX_BARRIERS) || _POSIX_BARRIERS < 0 +/* macOS lacks _POSIX_BARRIERS (required since POSIX:2018), so use fallback */ +#include "port/pg_threads/pg_thrd_barrier.h" +#else +typedef pthread_barrier_t pg_thrd_barrier_impl; +#endif + +enum +{ + pg_rwlock_plain_impl = 0, + pg_rwlock_timed_impl = 1 << 1, +}; + +#ifndef PG_THREADS_USE_THREADS_H +#define PG_CND_INIT_IMPL PTHREAD_COND_INITIALIZER +#define PG_MTX_PLAIN_INIT_IMPL PTHREAD_MUTEX_INITIALIZER +#endif +#define PG_RWLOCK_INIT_IMPL PTHREAD_RWLOCK_INITIALIZER + +/* pg_rwlock_t */ +PG_THREADS_FORWARD_VOID(pg_rwlock_destroy, pthread_rwlock_destroy, pg_rwlock_impl *); +PG_THREADS_OUTOFLINE(pg_rwlock_init, int, pg_rwlock_impl *, int); +PG_THREADS_MAP(pg_rwlock_rdlock, pthread_rwlock_rdlock, pg_rwlock_impl *); +#ifndef PG_RWLOCK_TIMED_NOT_SUPPORTED +PG_THREADS_MAP(pg_rwlock_timedrdlock, pthread_rwlock_timedrdlock, pg_rwlock_impl *, const struct timespec *); +PG_THREADS_MAP(pg_rwlock_timedwrlock, pthread_rwlock_timedwrlock, pg_rwlock_impl *, const struct timespec *); +#else +PG_THREADS_MAP_ERROR_UNSUPPORTED(pg_rwlock_timedrdlock, pg_rwlock_impl *, const struct timespec *); +PG_THREADS_MAP_ERROR_UNSUPPORTED(pg_rwlock_timedwrlock, pg_rwlock_impl *, const struct timespec *); +#endif +PG_THREADS_MAP(pg_rwlock_tryrdlock, pthread_rwlock_tryrdlock, pg_rwlock_impl *); +PG_THREADS_MAP(pg_rwlock_trywrlock, pthread_rwlock_trywrlock, pg_rwlock_impl *); +PG_THREADS_MAP(pg_rwlock_unlock, pthread_rwlock_unlock, pg_rwlock_impl *); +PG_THREADS_MAP(pg_rwlock_wrlock, pthread_rwlock_wrlock, pg_rwlock_impl *); + +#if defined(_POSIX_BARRIERS) && _POSIX_BARRIERS > 0 +/* pg_thrd_barrier_t */ +PG_THREADS_FORWARD_VOID(pg_thrd_barrier_destroy, pthread_barrier_destroy, pg_thrd_barrier_impl *); + +static inline int +pg_thrd_barrier_init(pg_thrd_barrier_impl *barrier, int count) +{ + return pg_thrd_map(pthread_barrier_init(barrier, NULL, count)); +} + +PG_THREADS_MAP(pg_thrd_barrier_wait, pthread_barrier_wait, pg_thrd_barrier_impl *); +#endif + +#ifndef PG_THREADS_USE_THREADS_H +PG_THREADS_EMPTY_VOID_0ARG(pg_threads_ext_assertions); +#endif + +#endif diff --git a/src/include/port/pg_threads/map_threads.h b/src/include/port/pg_threads/map_threads.h new file mode 100644 index 00000000000..67a658270c6 --- /dev/null +++ b/src/include/port/pg_threads/map_threads.h @@ -0,0 +1,118 @@ +/*------------------------------------------------------------------------- + * + * map_threads.h + * Map pg_threads.h API to system-provided . + * + * This header effectively renames all the identifiers in to add + * pg_ prefixes, but does so with inlined wrapper functions rather than simple + * renaming macros. This provides better detection of omissions and + * differences, and systemic mapping to system-provided . + * + * On POSIX systems, this is intended for developer-testing only. See + * map_threads_ext.h for explanation. + * + * References: + * - https://devblogs.microsoft.com/cppblog/c11-threads-in-visual-studio-2022-version-17-8-previ + * - https://pubs.opengroup.org/onlinepubs/9799919799/basedefs/threads.h.html + * + * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group + * + * IDENTIFICATION + * src/include/port/pg_threads/map_utils.h + * + *------------------------------------------------------------------------- + */ +#ifndef PG_THREADS_MAP_THREADS +#define PG_THREADS_MAP_THREADS + +#ifndef PG_THREADS_H +#error "include pg_threads.h instead" +#endif + +#include +#include + +#include "port/pg_threads/wrap.h" + +#ifdef WIN32 +/* + * Visual Studio's has these limitations, per the above reference. + * (See also map_windows.h, which copies this behavior.) + */ +#define PG_THRD_CURRENT_NOT_DETACHABLE +#define PG_THRD_CURRENT_NOT_JOINABLE +#endif + +typedef void (*pg_call_once_function_impl) (void); + +typedef cnd_t pg_cnd_impl; +typedef mtx_t pg_mtx_impl; +typedef once_flag pg_once_flag_impl; +typedef thrd_t pg_thrd_impl; +typedef tss_t pg_tss_impl; + +enum +{ + pg_thrd_success_impl = thrd_success, + pg_thrd_nomem_impl = thrd_nomem, + pg_thrd_timedout_impl = thrd_timedout, + pg_thrd_busy_impl = thrd_busy, + pg_thrd_error_impl = thrd_error, +}; + +enum +{ + pg_mtx_plain_impl = mtx_plain, + pg_mtx_recursive_impl = mtx_recursive, + pg_mtx_timed_impl = mtx_timed, +}; + +#define PG_ONCE_FLAG_INIT_IMPL ONCE_FLAG_INIT +#define PG_TSS_DTOR_ITERATIONS_IMPL TSS_DTOR_ITERATIONS + +/* Forward pg_##fun(...) -> R to fun(...) -> R. */ +#define PG_THREADS_PREFIX(fun, R, ...) \ + PG_THREADS_FORWARD(pg_##fun, fun, R, __VA_ARGS__) +#define PG_THREADS_PREFIX_0ARG(fun, R) \ + PG_THREADS_FORWARD_0ARG(pg_##fun, fun, R) +#define PG_THREADS_PREFIX_VOID(fun, ...) \ + PG_THREADS_FORWARD_VOID(pg_##fun, fun, __VA_ARGS__) +#define PG_THREADS_PREFIX_VOID_0ARG(fun) \ + PG_THREADS_FORWARD_VOID_0ARG(pg_##fun, fun) + +/* cnd_t */ +PG_THREADS_PREFIX(cnd_broadcast, int, cnd_t *); +PG_THREADS_PREFIX_VOID(cnd_destroy, cnd_t *); +PG_THREADS_PREFIX(cnd_init, int, cnd_t *); +PG_THREADS_PREFIX(cnd_signal, int, cnd_t *); +PG_THREADS_PREFIX(cnd_timedwait, int, cnd_t *, mtx_t *, const struct timespec *); +PG_THREADS_PREFIX(cnd_wait, int, cnd_t *, mtx_t *); + +/* mtx_t */ +PG_THREADS_PREFIX(mtx_init, int, mtx_t *, int); +PG_THREADS_PREFIX(mtx_lock, int, mtx_t *); +PG_THREADS_PREFIX(mtx_trylock, int, mtx_t *); +PG_THREADS_PREFIX(mtx_timedlock, int, mtx_t *, const struct timespec *); +PG_THREADS_PREFIX(mtx_unlock, int, mtx_t *); +PG_THREADS_PREFIX_VOID(mtx_destroy, mtx_t *); + +/* once_flag */ +PG_THREADS_PREFIX_VOID(call_once, once_flag *, pg_call_once_function_impl); + +/* thrd_t */ +PG_THREADS_PREFIX_0ARG(thrd_current, thrd_t); +PG_THREADS_PREFIX(thrd_create, int, thrd_t *, pg_thrd_start_t, void *); +PG_THREADS_PREFIX(thrd_detach, int, thrd_t); +PG_THREADS_PREFIX(thrd_equal, int, thrd_t, thrd_t); +PG_THREADS_PREFIX_VOID(thrd_exit, int); +PG_THREADS_PREFIX(thrd_join, int, thrd_t, int *); +PG_THREADS_PREFIX(thrd_sleep, int, const struct timespec *, struct timespec *); +PG_THREADS_PREFIX_VOID_0ARG(thrd_yield); + +/* tss_t */ +PG_THREADS_PREFIX(tss_create, int, tss_t *, pg_tss_dtor_t); +PG_THREADS_PREFIX_VOID(tss_delete, tss_t); +PG_THREADS_PREFIX(tss_get, void *, tss_t); +PG_THREADS_PREFIX(tss_set, int, tss_t, void *); + +#endif diff --git a/src/include/port/pg_threads/map_threads_ext.h b/src/include/port/pg_threads/map_threads_ext.h new file mode 100644 index 00000000000..b13f0407b11 --- /dev/null +++ b/src/include/port/pg_threads/map_threads_ext.h @@ -0,0 +1,101 @@ +/*------------------------------------------------------------------------- + * + * pg_threads/map_threads_ext.h + * Implements pg_threads_ext.h when using system . + * + * These facilities are by definition not in , and underlying + * native APIs are used for pg_rwlock_t and pg_thrd_barrier_t. Static + * initializers are also conjured for cnd_t and mtx_t. + * + * C has so far declined to standardize static initializer values. Values for + * Visual Studio are documented, but for developer testing on POSIX systems + * (which normally use map_pthread_ext.h instead), we have to take some + * liberties to provide values. + * + * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group + * + * IDENTIFICATION + * src/include/port/pg_threads/map_threads_ext.h + * + *------------------------------------------------------------------------- + */ +#ifndef PG_THREADS_MAP_THREADS_EXT_H +#define PG_THREADS_MAP_THREADS_EXT_H + +#ifndef WIN32 +#include +#endif + +#if defined(WIN32) +/* + * Visual Studio's : cnd_t and mtx_t can be zero-initialized per + * published documentation and example code, so this is a supported deployment + * configuration. + */ +#define PG_CND_INIT_IMPL {0} +#define PG_MTX_PLAIN_INIT_IMPL {0} +#elif defined(__GLIBC__) || defined(__linux__) +/* + * Undocumented developer-only support for testing PG_THREADS_USE_THREADS_H on + * Glibc and Musl. + * + * Those implementations use types, but obfuscate and + * cast internally to prevent this type of abuse. We drill through that plan + * to be able to test. We can't use 's initializer macros, but we + * can prove that ours produce the same binary image. + */ +#define PG_CND_INIT_IMPL {0} +#define PG_MTX_PLAIN_INIT_IMPL {0} +#define PG_THREADS_EXT_CHECK_INITIALIZERS_MATCH_PTHREAD +#else +/* + * Undocumented developer-only support for testing PG_THREADS_USE_THREADS_H on + * systems where uses types and that's verifiable from + * the type system. We take the much smaller liberty of inferring that the + * latter's initializers therefore work for the former's types. + * + * This is expected to work on FreeBSD and NetBSD, and the type assertions are + * expected to fail on Solaris and AIX. + */ +#define PG_CND_INIT_IMPL PTHREAD_COND_INITIALIZER +#define PG_MTX_PLAIN_INIT_IMPL PTHREAD_MUTEX_INITIALIZER +#define PG_THREADS_EXT_CHECK_INITIALIZERS_MATCH_PTHREAD +#define PG_THREADS_EXT_CHECK_TYPES_MATCH_PTHREAD +#endif + +#ifndef WIN32 +#include "port/pg_threads/map_pthread_ext.h" +#else +#include "port/pg_threads/map_windows_ext.h" +#endif + +static inline void +pg_threads_ext_assertions(void) +{ +#ifdef USE_ASSERT_CHECKING +#ifdef PG_THREADS_EXT_CHECK_INITIALIZERS_MATCH_PTHREAD + pthread_cond_t cond = PTHREAD_COND_INITIALIZER; + pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER; + cnd_t cnd = PG_CND_INIT_IMPL; + mtx_t mtx = PG_MTX_PLAIN_INIT_IMPL; + +#if defined(PG_THREADS_EXT_CHECK_TYPES_MATCH_PTHREAD) && !defined(__cplusplus) + /* Except on glibc/musl, the types must match or all bets are off. */ + StaticAssertVariableIsOfType(cnd, pthread_cond_t); + StaticAssertVariableIsOfType(mtx, pthread_mutex_t); +#endif + + /* Even glibc and musl's obfuscated types should have these properties. */ + static_assert(sizeof(cnd) == sizeof(pthread_cond_t), "bad size"); + static_assert(sizeof(mtx) == sizeof(pthread_mutex_t), "bad size"); + static_assert(alignof(cnd_t) >= alignof(pthread_cond_t), "bad alignment"); + static_assert(alignof(mtx_t) >= alignof(pthread_mutex_t), "bad alignment"); + + /* Our initializers should produce the same bits as 's. */ + Assert(memcmp(&cnd, &cond, sizeof(cnd)) == 0); + Assert(memcmp(&mtx, &mutex, sizeof(mtx)) == 0); +#endif +#endif +} + +#endif diff --git a/src/include/port/pg_threads/map_windows.h b/src/include/port/pg_threads/map_windows.h new file mode 100644 index 00000000000..051ee6bbab4 --- /dev/null +++ b/src/include/port/pg_threads/map_windows.h @@ -0,0 +1,192 @@ +/*------------------------------------------------------------------------- + * + * pg_threads/map_windows.h + * Map pg_threads.h interface to Windows native APIs. + * + * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group + * + * IDENTIFICATION + * src/include/port/pg_threads/map_windows.h + * + *------------------------------------------------------------------------- + */ +#ifndef PG_THREADS_MAP_WINDOWS +#define PG_THREADS_MAP_WINDOWS + +#ifndef PG_THREADS_H +#error "include pg_threads.h instead" +#endif + +#include + +enum +{ + pg_thrd_success_impl = 0, + pg_thrd_nomem_impl, + pg_thrd_timedout_impl, + pg_thrd_busy_impl, + pg_thrd_error_impl, +}; + +/* These need the definition of pg_thrd_success_impl. */ +#include "port/pg_threads/map_windows_error.h" +#include "port/pg_threads/wrap.h" + +/* + * Experimental build option: define PG_THRD_CURRENT_CONFORMING to make + * pg_thrd_create() a bit slower in exchange for being able to pass its result + * to pg_thrd_detach() and (after copying it to another thread somehow) + * pg_thrd_join(). + * + * The default is that you can't do those things, and that matches the + * behavior of Visual Studio's . + * + * XXX It doesn't seem that useful to do those things anyway, and unlikely + * that anyone could usefully make use of these "anti-feature" macros. Hence + * default behavior of following Visual Studio's non-conforming behavior. The + * option is provided as illustration of the tradeoff being made under the + * covers. + */ +#ifndef PG_THRD_CURRENT_CONFORMING +#define PG_THRD_CURRENT_NOT_DETACHABLE +#define PG_THRD_CURRENT_NOT_JOINABLE +#endif + +/* + * pg_mtx_t is mapped to SRWLOCK or CRITICAL_SECTION depending on whether + * pg_mtx_recursive is requested. Neither supports timeouts, so pg_mtx_timed + * remains unsupported. + * + * XXX CreateMutex() could provide third union option that supports timedouts + * and recursion, but those doesn't support interaction with + * CONDITION_VARIABLE as required for cnd_wait(). That's a more complicated + * feature hole, while this feature hole currently coincides with macOS. + * + * XXX Visual Studio's mtx_t doesn't have this problem, being built from + * SRWLOCK + CONDITION_VARIABLE + counters. + */ +#define PG_MTX_TIMED_NOT_SUPPORTED + +typedef void (*pg_call_once_function_impl) (void); + +typedef CONDITION_VARIABLE pg_cnd_impl; +typedef INIT_ONCE pg_once_flag_impl; + +typedef struct pg_mtx_impl +{ + enum + { + PG_MTX_SRWLOCK = 0, + PG_MTX_CRITICAL_SECTION, + } type; + union + { + SRWLOCK srwlock; + CRITICAL_SECTION critical_section; + }; +} pg_mtx_impl; + +typedef struct pg_thrd_impl +{ + HANDLE handle; + DWORD id; +} pg_thrd_impl; + +typedef struct pg_tss_impl +{ + int index; + int generation; +} pg_tss_impl; + + +enum +{ + pg_mtx_plain_impl = 0, + pg_mtx_recursive_impl = 1 << 1, + pg_mtx_timed_impl = 2 << 2, +}; + +#define PG_ONCE_FLAG_INIT_IMPL INIT_ONCE_STATIC_INIT +#define PG_TSS_DTOR_ITERATIONS_IMPL 4 /* typical POSIX value */ + +/* cnd_t */ +PG_THREADS_MAP_SUCCESS(pg_cnd_broadcast, WakeAllConditionVariable, pg_cnd_impl *); +PG_THREADS_EMPTY_VOID(pg_cnd_destroy, pg_cnd_impl *); +PG_THREADS_MAP_SUCCESS(pg_cnd_init, InitializeConditionVariable, pg_cnd_impl *); +PG_THREADS_MAP_SUCCESS(pg_cnd_signal, WakeConditionVariable, pg_cnd_impl *); +PG_THREADS_OUTOFLINE(pg_cnd_timedwait, int, pg_cnd_impl *, pg_mtx_impl *, const struct timespec *); +PG_THREADS_OUTOFLINE(pg_cnd_wait, int, pg_cnd_impl *, pg_mtx_impl *); + +/* mtx_t */ +PG_THREADS_OUTOFLINE_VOID(pg_mtx_destroy, pg_mtx_impl *); +PG_THREADS_OUTOFLINE(pg_mtx_init, int, pg_mtx_impl *, int); + +static inline int +pg_mtx_lock(pg_mtx_impl *mutex) +{ + switch (mutex->type) + { + case PG_MTX_SRWLOCK: + AcquireSRWLockExclusive(&mutex->srwlock); + break; + case PG_MTX_CRITICAL_SECTION: + EnterCriticalSection(&mutex->critical_section); + break; + } + return pg_thrd_success_impl; +} + +PG_THREADS_MAP_ERROR_UNSUPPORTED(pg_mtx_timedlock, pg_mtx_impl *, const struct timespec *); + +static inline int +pg_mtx_trylock(pg_mtx_impl *mutex) +{ + switch (mutex->type) + { + case PG_MTX_SRWLOCK: + if (!TryAcquireSRWLockExclusive(&mutex->srwlock)) + return pg_thrd_busy_impl; + break; + case PG_MTX_CRITICAL_SECTION: + if (!TryEnterCriticalSection(&mutex->critical_section)) + return pg_thrd_busy_impl; + break; + } + return pg_thrd_success_impl; +} + +static inline int +pg_mtx_unlock(pg_mtx_impl *mutex) +{ + switch (mutex->type) + { + case PG_MTX_SRWLOCK: + ReleaseSRWLockExclusive(&mutex->srwlock); + break; + case PG_MTX_CRITICAL_SECTION: + LeaveCriticalSection(&mutex->critical_section); + break; + } + return pg_thrd_success_impl; +} + +/* once_flag */ +PG_THREADS_OUTOFLINE_VOID(pg_call_once, pg_once_flag_impl *, pg_call_once_function_impl); + +/* thrd_t */ +PG_THREADS_OUTOFLINE_0ARG(pg_thrd_current, pg_thrd_impl); +PG_THREADS_OUTOFLINE(pg_thrd_create, int, pg_thrd_impl *, pg_thrd_start_t, void *); +PG_THREADS_OUTOFLINE(pg_thrd_detach, int, pg_thrd_impl); +PG_THREADS_OUTOFLINE(pg_thrd_equal, int, pg_thrd_impl, pg_thrd_impl); +PG_THREADS_FORWARD_VOID(pg_thrd_exit, ExitThread, int); +PG_THREADS_OUTOFLINE(pg_thrd_join, int, pg_thrd_impl, int *); +PG_THREADS_OUTOFLINE(pg_thrd_sleep, int, const struct timespec *, struct timespec *); +PG_THREADS_FORWARD_VOID_0ARG(pg_thrd_yield, SwitchToThread); + +/* tss_t */ +PG_THREADS_OUTOFLINE(pg_tss_create, int, pg_tss_impl *, pg_tss_dtor_t); +PG_THREADS_OUTOFLINE_VOID(pg_tss_delete, pg_tss_impl); +PG_THREADS_OUTOFLINE(pg_tss_get, void *, pg_tss_impl); +PG_THREADS_OUTOFLINE(pg_tss_set, int, pg_tss_impl, void *); + +#endif diff --git a/src/include/port/pg_threads/map_windows_error.h b/src/include/port/pg_threads/map_windows_error.h new file mode 100644 index 00000000000..70cf684f828 --- /dev/null +++ b/src/include/port/pg_threads/map_windows_error.h @@ -0,0 +1,31 @@ +/*------------------------------------------------------------------------- + * + * pg_threads/map_windows_error.h + * Support for mapping errors. + * + * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group + * + * IDENTIFICATION + * src/include/port/pg_threads/map_windows_error.h + * + *------------------------------------------------------------------------- + */ +#ifndef PG_THREADS_MAP_WINDOWS_ERROR +#define PG_THREADS_MAP_WINDOWS_ERROR + +#ifndef PG_THREADS_H +#error "include pg_threads.h instead" +#endif + +extern int pg_thrd_map_last_windows_error(void); +extern int pg_thrd_internal_error(const char *detail); + +static inline int +pg_thrd_map(bool success) +{ + return success ? + pg_thrd_success_impl : + pg_thrd_map_last_windows_error(); +} + +#endif diff --git a/src/include/port/pg_threads/map_windows_ext.h b/src/include/port/pg_threads/map_windows_ext.h new file mode 100644 index 00000000000..f0167fc18bd --- /dev/null +++ b/src/include/port/pg_threads/map_windows_ext.h @@ -0,0 +1,133 @@ +/*------------------------------------------------------------------------- + * + * pg_threads/map_windows_ext.h + * Map pg_threads_ext.h interfaces to Windows native APIs. + * + * This included by both map_threads_ext.h and map_windows_ext.h. + * + * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group + * + * IDENTIFICATION + * src/include/port/pg_threads/map_windows_ext.h + * + *------------------------------------------------------------------------- + */ +#ifndef PG_THREADS_MAP_WINDOW_EXT_H +#define PG_THREADS_MAP_WINDOW_EXT_H + +#ifndef PG_THREADS_EXT_H +#error "include pg_threads_ext.h instead" +#endif + +#include + +#include "port/pg_threads/map_windows_error.h" +#include "port/pg_threads/wrap.h" + +/* + * SRWLOCK doesn't have a timed wait mode, so we'd have to hand-roll an + * implementation with a futex or lock + cv. + */ +#define PG_RWLOCK_TIMED_NOT_SUPPORTED + +typedef struct pg_rwlock_impl +{ + SRWLOCK lock; + bool exclusive; +} pg_rwlock_impl; + +typedef SYNCHRONIZATION_BARRIER pg_thrd_barrier_impl; + + +enum +{ + pg_rwlock_plain_impl = 0, + pg_rwlock_timed_impl = 1 << 1, +}; + +#ifndef PG_THREADS_USE_THREADS_H +#define PG_CND_INIT_IMPL CONDITION_VARIABLE_INIT +#define PG_MTX_PLAIN_INIT_IMPL {.type = PG_MTX_SRWLOCK, .srwlock = SRWLOCK_INIT} +#endif +#define PG_RWLOCK_PLAIN_INIT_IMPL {.lock = SRWLOCK_INIT, .exclusive = false} + + +/* pg_rwlock_t */ +PG_THREADS_EMPTY_VOID(pg_rwlock_destroy, pg_rwlock_impl *); +PG_THREADS_OUTOFLINE(pg_rwlock_init, int, pg_rwlock_impl *, int); + +static inline int +pg_rwlock_unlock(pg_rwlock_impl *lock) +{ + if (lock->exclusive) + { + lock->exclusive = false; + ReleaseSRWLockExclusive(&lock->lock); + } + else + { + ReleaseSRWLockShared(&lock->lock); + } + return pg_thrd_success_impl; +} + +static inline int +pg_rwlock_rdlock(pg_rwlock_impl *lock) +{ + AcquireSRWLockShared(&lock->lock); + return pg_thrd_success_impl; +} + +PG_THREADS_MAP_ERROR_UNSUPPORTED(pg_rwlock_timedrdlock, pg_rwlock_impl *, const struct timespec *); +PG_THREADS_MAP_ERROR_UNSUPPORTED(pg_rwlock_timedwrlock, pg_rwlock_impl *, const struct timespec *); + +static inline int +pg_rwlock_tryrdlock(pg_rwlock_impl *lock) +{ + if (TryAcquireSRWLockShared(&lock->lock)) + return pg_thrd_success_impl; + return pg_thrd_busy_impl; +} + +static inline int +pg_rwlock_trywrlock(pg_rwlock_impl *lock) +{ + if (TryAcquireSRWLockExclusive(&lock->lock)) + { + lock->exclusive = true; + return pg_thrd_success_impl; + } + return pg_thrd_busy_impl; +} + +static inline int +pg_rwlock_wrlock(pg_rwlock_impl *lock) +{ + AcquireSRWLockExclusive(&lock->lock); + lock->exclusive = true; + return pg_thrd_success_impl; +} + + +/* pg_thrd_barrier_t */ +PG_THREADS_EMPTY_VOID(pg_thrd_barrier_destroy, pg_thrd_barrier_impl *); + +static inline int +pg_thrd_barrier_init(pg_thrd_barrier_impl *barrier, int count) +{ + return pg_thrd_map(InitializeSynchronizationBarrier(barrier, count, -1)); +} + +static inline int +pg_thrd_barrier_wait(pg_thrd_barrier_impl *barrier) +{ + EnterSynchronizationBarrier(barrier, + SYNCHRONIZATION_BARRIER_FLAGS_NO_DELETE); + return pg_thrd_success_impl; +} + +#ifndef PG_THREADS_USE_THREADS_H +PG_THREADS_EMPTY_VOID_0ARG(pg_threads_ext_assertions); +#endif + +#endif diff --git a/src/include/port/pg_threads/pg_thrd_barrier.h b/src/include/port/pg_threads/pg_thrd_barrier.h new file mode 100644 index 00000000000..1f0a8ee9496 --- /dev/null +++ b/src/include/port/pg_threads/pg_thrd_barrier.h @@ -0,0 +1,76 @@ +/*------------------------------------------------------------------------- + * + * pg_thrd_barrier.h + * Fallback implementation of pg_thrd_barrier_t. + * + * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group + * + * IDENTIFICATION + * src/include/port/pg_threads/pg_thrd_barrier.h + * + *------------------------------------------------------------------------- + */ +#ifndef PG_THRD_BARRIER_H +#define PG_THRD_BARRIER_H + +#ifndef PG_THREADS_H +#error "include pg_thread.h instead" +#endif + +typedef struct pg_thrd_barrier_impl +{ + bool sense; + int expected; + int arrived; + pg_mtx_impl mutex; + pg_cnd_impl cond; +} pg_thrd_barrier_impl; + +static inline int +pg_thrd_barrier_init(pg_thrd_barrier_impl *barrier, int count) +{ + barrier->sense = false; + barrier->expected = count; + barrier->arrived = 0; + if (pg_cnd_init(&barrier->cond) != pg_thrd_success_impl) + return pg_thrd_error_impl; + if (pg_mtx_init(&barrier->mutex, pg_mtx_plain_impl) != pg_thrd_success_impl) + { + pg_cnd_destroy(&barrier->cond); + return pg_thrd_error_impl; + } + return pg_thrd_success_impl; +} + +static inline int +pg_thrd_barrier_wait(pg_thrd_barrier_impl *barrier) +{ + 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_impl; + } + 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_impl; +} + +static inline void +pg_thrd_barrier_destroy(pg_thrd_barrier_impl *barrier) +{ + pg_mtx_destroy(&barrier->mutex); + pg_cnd_destroy(&barrier->cond); +} + +#endif diff --git a/src/include/port/pg_threads/wrap.h b/src/include/port/pg_threads/wrap.h new file mode 100644 index 00000000000..fe64f77807d --- /dev/null +++ b/src/include/port/pg_threads/wrap.h @@ -0,0 +1,114 @@ +/*------------------------------------------------------------------------- + * + * wrap.h + * Macros for reducing boilerplate code when wrapping native APIs. + * + * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group + * + * IDENTIFICATION + * src/include/port/pg_threads/wrap.h + * + *------------------------------------------------------------------------- + */ +#ifndef PG_THREADS_WRAP_H +#define PG_THREADS_WRAP_H + +/* Given typenames T1, ... make an argument list T1 v1, .... */ +#define PG_THREADS_MAKE_ARG_LIST(...) \ + PG_THREADS_ARG4(__VA_ARGS__, \ + PG_THREADS_AL3, \ + PG_THREADS_AL2, \ + PG_THREADS_AL1)(__VA_ARGS__) +#define PG_THREADS_ARG4(_1, _2, _3, _4, ...) _4 +#define PG_THREADS_AL1(T1) T1 v1 +#define PG_THREADS_AL2(T1, T2) T1 v1, T2 v2 +#define PG_THREADS_AL3(T1, T2, T3) T1 v1, T2 v2, T3 v3 + +/* Given typenames T1, ... make a value list v1, ... */ +#define PG_THREADS_MAKE_VAL_LIST(...) \ + PG_THREADS_ARG4(__VA_ARGS__, \ + PG_THREADS_VL3, \ + PG_THREADS_VL2, \ + PG_THREADS_VL1) +#define PG_THREADS_VL1 v1 +#define PG_THREADS_VL2 v1, v2 +#define PG_THREADS_VL3 v1, v2, v3 + +/* Forward fun(...) -> R to target_fun(...) -> R. */ +#define PG_THREADS_FORWARD(fun, target_fun, R, ...) \ + static inline R \ + fun(PG_THREADS_MAKE_ARG_LIST(__VA_ARGS__)) \ + { \ + return target_fun(PG_THREADS_MAKE_VAL_LIST(__VA_ARGS__)); \ + } +#define PG_THREADS_FORWARD_0ARG(fun, target, R) \ + static inline R \ + fun(void) \ + { \ + return target(); \ + } +#define PG_THREADS_FORWARD_VOID(fun, target_fun, ...) \ + static inline void \ + fun(PG_THREADS_MAKE_ARG_LIST(__VA_ARGS__)) \ + { \ + target_fun(PG_THREADS_MAKE_VAL_LIST(__VA_ARGS__)); \ + } +#define PG_THREADS_FORWARD_VOID_0ARG(fun, target_fun, ...) \ + static inline void \ + fun(void) \ + { \ + target_fun(); \ + } + +/* fun(...) -> void is an empty function. */ +#define PG_THREADS_EMPTY_VOID(fun, ...) \ + static inline void \ + fun(PG_THREADS_MAKE_ARG_LIST(__VA_ARGS__)) \ + { \ + } +#define PG_THREADS_EMPTY_VOID_0ARG(fun) \ + static inline void \ + fun(void) \ + { \ + } + +/* Forward fun(...) -> int to pg_thrd_map(target_fun(...)) -> int. */ +#define PG_THREADS_MAP(fun, target_fun, ...) \ + static inline int \ + fun(PG_THREADS_MAKE_ARG_LIST(__VA_ARGS__)) \ + { \ + return pg_thrd_map(target_fun(PG_THREADS_MAKE_VAL_LIST(__VA_ARGS__))); \ + } + +/* Forward fun(...) -> int to target_fun(...), always reporting success. */ +#define PG_THREADS_MAP_SUCCESS(fun, target_fun, ...) \ + static inline int \ + fun(PG_THREADS_MAKE_ARG_LIST(__VA_ARGS__)) \ + { \ + target_fun(PG_THREADS_MAKE_VAL_LIST(__VA_ARGS__)); \ + return pg_thrd_success_impl; \ + } + +/* fun(...) -> int, always reporting an "unsupported" error. */ +#define PG_THREADS_MAP_ERROR_UNSUPPORTED(fun, ...) \ + static inline int \ + fun(PG_THREADS_MAKE_ARG_LIST(__VA_ARGS__)) \ + { \ + return pg_thrd_internal_error(#fun "(): unsupported on this platform"); \ + } + +/* Forward fun(...) -> R to out-of-line fun##_impl(...) -> R. */ +#define PG_THREADS_OUTOFLINE(fun, R, ...) \ + extern R fun##_impl(PG_THREADS_MAKE_ARG_LIST(__VA_ARGS__)); \ + PG_THREADS_FORWARD(fun, fun##_impl, R, __VA_ARGS__); +#define PG_THREADS_OUTOFLINE_0ARG(fun, R) \ + extern R fun##_impl(void); \ + PG_THREADS_FORWARD_0ARG(fun, fun##_impl, R); +#define PG_THREADS_OUTOFLINE_VOID(fun, ...) \ + extern void fun##_impl(PG_THREADS_MAKE_ARG_LIST(__VA_ARGS__)); \ + PG_THREADS_FORWARD_VOID(fun, fun##_impl, __VA_ARGS__); +#define PG_THREADS_OUTOFLINE_VOID_0ARG(fun) \ + extern void fun##_impl(void); \ + PG_THREADS_FORWARD_VOID_0ARG(fun, fun##_impl); + +#endif diff --git a/src/include/port/pg_threads_ext.h b/src/include/port/pg_threads_ext.h new file mode 100644 index 00000000000..26f0f2e7552 --- /dev/null +++ b/src/include/port/pg_threads_ext.h @@ -0,0 +1,85 @@ +/*------------------------------------------------------------------------- + * + * pg_threads_ext.h + * Portable extensions to pg_threads.h. + * + * This header includes pg_threads.h but also declares some extensions on top + * that are not based on the spartan standard. + * + * Three implementations are available: + * + * (1) map_pthread_ext.h for + * (2) map_threads_ext.h to forward to (1) or (3) + * (3) map_windows_ext.h for + * + * The option is only available on Windows and a subset of POSIX + * systems (see map_threads_ext.h for details) to support developer-only + * testing. + * + * The following macros are defined on some platforms: + * + * - PG_RWLOCK_TIMED_NOT_SUPPORTED: macOS, Windows + * + * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group + * + * IDENTIFICATION + * src/include/port/pg_threads_ext.h + * + *------------------------------------------------------------------------- + */ +#ifndef PG_THREADS_EXT_H +#define PG_THREADS_EXT_H + +#ifndef PG_THREADS_EXT_H +#error "include pg_threads_ext.h instead" +#endif + +#include "port/pg_threads.h" + +#if defined(PG_THREADS_USE_PTHREAD_H) +#include "port/pg_threads/map_pthread_ext.h" +#elif defined(PG_THREADS_USE_THREADS_H) +#include "port/pg_threads/map_threads_ext.h" +#elif defined(PG_THREADS_USE_WINDOWS_H) +#include "port/pg_threads/map_windows_ext.h" +#endif + +typedef pg_rwlock_impl pg_rwlock_t; +typedef pg_thrd_barrier_impl pg_thrd_barrier_t; + +/* type flags for pg_rwlock_t */ +enum +{ + pg_rwlock_plain = pg_rwlock_plain_impl, + pg_rwlock_timed = pg_rwlock_timed_impl, + /* policy flags could potentially appear here */ +}; + +/* static initializer macros */ +#define PG_CND_INIT PG_CND_INIT_IMPL +#define PG_MTX_PLAIN_INIT PG_MTX_PLAIN_INIT_IMPL +#define PG_RWLOCK_PLAIN_INIT PG_RWLOCK_INIT_IMPL + +/* pg_rwlock_t functions */ +static inline int pg_rwlock_init(pg_rwlock_t *lock, int type); +static inline int pg_rwlock_rdlock(pg_rwlock_t *lock); +static inline int pg_rwlock_timedrdlock(pg_rwlock_t *lock, const struct timespec *time); +static inline int pg_rwlock_timedwrlock(pg_rwlock_t *lock, const struct timespec *time); +static inline int pg_rwlock_tryrdlock(pg_rwlock_t *lock); +static inline int pg_rwlock_trywrlock(pg_rwlock_t *lock); +static inline int pg_rwlock_unlock(pg_rwlock_t *lock); +static inline int pg_rwlock_wrlock(pg_rwlock_t *lock); + +/* pg_thrd_barrier_t functions */ +static inline int pg_thrd_barrier_init(pg_thrd_barrier_t *barrier, int count); +static inline void pg_thrd_barrier_destroy(pg_thrd_barrier_t *barrier); +static inline int pg_thrd_barrier_wait(pg_thrd_barrier_t *barrier); + +/* error string retrieval functions */ +extern const char *pg_thrd_error_string(int error); +extern const char *pg_thrd_error_string_with_detail(int error); + +/* internal self-test */ +static inline void pg_threads_ext_assertions(void); + +#endif diff --git a/src/port/Makefile b/src/port/Makefile index 7e9b5877652..29734bb0d35 100644 --- a/src/port/Makefile +++ b/src/port/Makefile @@ -51,6 +51,7 @@ OBJS = \ pg_popcount_aarch64.o \ pg_popcount_x86.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 922b3f64676..08948768b1c 100644 --- a/src/port/meson.build +++ b/src/port/meson.build @@ -14,6 +14,7 @@ pgport_sources = [ 'pg_popcount_aarch64.c', 'pg_popcount_x86.c', 'pg_strong_random.c', + 'pg_threads.c', 'pgcheckdir.c', 'pgmkdirp.c', 'pgsleep.c', @@ -157,11 +158,11 @@ pgport_variants = { 'dependencies': [backend_port_code], }, '': default_lib_args + { - 'dependencies': [frontend_port_code], + 'dependencies': [frontend_port_code, thread_dep], }, '_shlib': default_lib_args + { 'pic': true, - 'dependencies': [frontend_port_code], + 'dependencies': [frontend_port_code, thread_dep], }, } diff --git a/src/port/pg_threads.c b/src/port/pg_threads.c new file mode 100644 index 00000000000..07dc7fe0dd2 --- /dev/null +++ b/src/port/pg_threads.c @@ -0,0 +1,812 @@ +/*------------------------------------------------------------------------- + * + * pg_threads.c + * Out-of-line parts of pg_threads/... headers. + * + * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group + * + * IDENTIFICATION + * src/port/pg_threads.c + * + *------------------------------------------------------------------------- + */ +#include "c.h" +#include "port/pg_threads.h" +#include "port/pg_threads_ext.h" + +#include +#include +#include + + +/*------------------------------------------------------------------------- + * + * Error retrieval support. + * + * These functions are available even for map_threads.h builds, but no extra + * error detail is available so you just get the five fixed result strings. + * + * XXX That's because when functions in map_threads.h return thrd_error they + * don't clear stale messages from earlier map_threads_ext.h functions that + * failed. We could teach their wrappers to do that, but that seems like an + * impediment to using directly in the future. + * + *------------------------------------------------------------------------- + */ + +#if defined(PG_THREADS_USE_PTHREAD_H) || defined(PG_THREADS_USE_WINDOWS_H) +#define PG_THREADS_HAVE_DETAILED_ERRORS +#endif + +static int pg_thrd_format_error(const char *format, ...) pg_attribute_printf(1, 2); + +#ifdef PG_THREADS_HAVE_DETAILED_ERRORS +static thread_local char pg_thrd_last_error_detail[64]; +#endif + +/* Convert the standard errors to fixed strings. */ +const char * +pg_thrd_error_string(int error) +{ + switch (error) + { + case pg_thrd_success: + return "success"; + case pg_thrd_nomem: + return "no memory"; + case pg_thrd_timedout: + return "timed out"; + case pg_thrd_busy: + return "busy"; + default: + return "error"; + } +} + +/* + * Retrieve description of the most recently returned result from a + * pg_thrd_XXX() function. Result is only valid until the next call to a + * pg_threads.h function from the same thread. + * + * Except when using map_threads.h, extra details may be available. + */ +const char * +pg_thrd_error_string_with_detail(int error) +{ +#ifndef PG_THREADS_HAVE_DETAILED_ERRORS + return pg_thrd_error_string(error); +#else + if (error == pg_thrd_success || + error == pg_thrd_nomem || + error == pg_thrd_timedout || + error == pg_thrd_busy || + pg_thrd_last_error_detail[0] == 0) + return pg_thrd_error_string(error); + + return pg_thrd_last_error_detail; +#endif +} + +/*------------------------------------------------------------------------- + * + * Internal error mapping support. + * + * Even when using system , these are reachable from + * pg_threads_ext.h extension. In that case detailed messages are not kept. + * + * + *------------------------------------------------------------------------- + */ + +/* + * Return pg_thrd_error, and also store custom error message. + */ +static int +pg_thrd_format_error(const char *format, ...) +{ +#ifdef PG_THREADS_HAVE_DETAILED_ERRORS + va_list args; + + va_start(args, format); + vsnprintf(pg_thrd_last_error_detail, sizeof(pg_thrd_last_error_detail), + format, args); +#endif + + return pg_thrd_error; +} + +/* + * pg_thrd_error + a detail message originating from implementation code, not + * the OS. + */ +int +pg_thrd_internal_error(const char *detail) +{ + return pg_thrd_format_error("%s", detail); +} + +#ifdef WIN32 +/* + * Convert last Windows error number to a pg_thrd_XXX error, and record the + * underlying code in a generic detail message. (Places that make multiple + * system calls can format a custom error message to report which system call + * failed instead of using this generic conversion.) + */ +static int +pg_thrd_map_windows_error(DWORD windows_error) +{ + switch (windows_error) + { + case ERROR_NOT_ENOUGH_MEMORY: + return pg_thrd_nomem; + case ERROR_BUSY: + return pg_thrd_busy; + case WAIT_TIMEOUT: + return pg_thrd_timedout; + default: + return pg_thrd_format_error("Windows error: %u", + (unsigned int) windows_error); + } +} + +/* Convert last Windows error number to a pg_thrd_XXX error. */ +int +pg_thrd_map_last_windows_error(void) +{ + return pg_thrd_map_windows_error(GetLastError()); +} +#else +/* + * Convert a POSIX error to a pg_thrd_XXX error, and record the underlying OS + * error in a detail message. + */ +int +pg_thrd_map_pthread_error(int pthread_error) +{ + switch (pthread_error) + { + case ENOMEM: + return pg_thrd_nomem; + case EBUSY: + return pg_thrd_busy; + case ETIMEDOUT: + return pg_thrd_timedout; + default: + return pg_thrd_format_error("pthread error: %s", + strerror(pthread_error)); + } +} +#endif + +#if defined(PG_THREADS_USE_PTHREAD_H) || \ + defined(PG_THREADS_USE_WINDOWS_H) + +/*------------------------------------------------------------------------- + * + * map_{pthread,windows}.h + * + *------------------------------------------------------------------------- + */ + +static int +pg_mtx_validate_type(int type) +{ + if (type & ~(pg_mtx_plain | pg_mtx_timed | pg_mtx_recursive)) + return pg_thrd_format_error("pg_mtx_init(): bad type flags %d", type); + +#ifdef PG_MTX_TIMED_NOT_SUPPORTED + if (type & pg_mtx_timed) + return pg_thrd_internal_error("pg_mtx_init(): pg_mtx_timed not supported on this platform"); +#endif + + return pg_thrd_success; +} + +#endif + +/*------------------------------------------------------------------------- + * + * map_{pthread,windows}_ext.h, also reached from map_threads_ext.h + * + *------------------------------------------------------------------------- + */ + +static int +pg_rwlock_validate_type(int type) +{ + if (type & ~(pg_rwlock_plain | pg_rwlock_timed)) + return pg_thrd_format_error("pg_rwlock_init(): bad type flags %d", type); + +#ifdef PG_RWLOCK_TIMED_NOT_SUPPORTED + if (type & pg_rwlock_timed) + return pg_thrd_internal_error("pg_rwlock_init(): pg_rwlock_timed not supported on this platform"); +#endif + + return pg_thrd_success; +} + +#if defined(WIN32) + +/*------------------------------------------------------------------------- + * + * map_windows_ext.h, also used for map_threads_ext.h on Windows systems + * + *------------------------------------------------------------------------- + */ + +int +pg_rwlock_init_impl(pg_rwlock_impl *lock, int type) +{ + int result = pg_rwlock_validate_type(type); + + if (result != pg_thrd_success) + return result; + + InitializeSRWLock(&lock->lock); + lock->exclusive = false; + return pg_thrd_success_impl; +} + +#else + +/*------------------------------------------------------------------------- + * + * map_pthread_ext.h, also used for map_threads_ext.h on POSIX systems + * + *------------------------------------------------------------------------- + */ + +int +pg_rwlock_init_impl(pg_rwlock_impl *lock, int type) +{ + int result = pg_rwlock_validate_type(type); + + if (result != pg_thrd_success) + return result; + + return pg_thrd_map(pthread_rwlock_init(lock, NULL)); +} + +#endif + +#if defined(PG_THREADS_USE_WINDOWS_H) + +/*------------------------------------------------------------------------- + * + * map_windows.h + * + *------------------------------------------------------------------------- + */ + +/* Maximum number of active pg_tss_t IDs per process. */ +#define PG_TSS_MAX 16 + +typedef struct pg_tss_value +{ + void *value; + int generation; +} pg_tss_value; + +typedef struct pg_tss_slot +{ + pg_tss_dtor_t destructor; + int generation; + bool in_use; +} pg_tss_slot; + +static thread_local pg_thrd_t pg_thrd_current_thread_local; + +static thread_local pg_tss_value pg_tss_values[PG_TSS_MAX]; +static pg_tss_slot pg_tss_slots[PG_TSS_MAX]; +static DWORD pg_tss_fls = FLS_OUT_OF_INDEXES; +static pg_mtx_t pg_tss_lock = PG_MTX_PLAIN_INIT; + +static BOOL CALLBACK +pg_call_once_invoke(pg_once_flag *flag, void *parameter, void **context) +{ + pg_call_once_function_impl function = parameter; + + function(); + return true; +} + +void +pg_call_once_impl(pg_once_flag *flag, pg_call_once_function_impl function) +{ + InitOnceExecuteOnce(flag, pg_call_once_invoke, function, NULL); +} + +/* Convert an absolute TIME_UTC time to a millisecond delay. */ +static DWORD +pg_threads_get_ms_delay(const struct timespec *time) +{ + struct timespec now; + int64_t now_ms; + int64_t time_ms; + int64_t wait_ms; + + if (time == NULL) + return INFINITE; + + timespec_get(&now, TIME_UTC); + now_ms = now.tv_sec * 1000 + now.tv_nsec / 1000000; + time_ms = time->tv_sec * 1000 + time->tv_nsec / 1000000; + wait_ms = time_ms - now_ms; + if (wait_ms < 0) + wait_ms = 0; + else if (wait_ms > INFINITE) + wait_ms = INFINITE; + return wait_ms; +} + +int +pg_cnd_timedwait_impl(pg_cnd_impl *cnd, pg_mtx_impl *mutex, const struct timespec *time) +{ + DWORD wait_ms = pg_threads_get_ms_delay(time); + + switch (mutex->type) + { + case PG_MTX_SRWLOCK: + return pg_thrd_map(SleepConditionVariableSRW(cnd, + &mutex->srwlock, + wait_ms, + 0)); + case PG_MTX_CRITICAL_SECTION: + default: + return pg_thrd_map(SleepConditionVariableCS(cnd, + &mutex->critical_section, + wait_ms)); + } +} + +int +pg_cnd_wait_impl(pg_cnd_impl *cnd, pg_mtx_impl *mutex) +{ + return pg_cnd_timedwait_impl(cnd, mutex, NULL); +} + +typedef struct pg_thrd_start_info +{ + pg_thrd_start_t function; + void *argument; +#ifdef PG_THRD_CURRENT_CONFORMING + pg_thrd_t self; +#endif +} pg_thrd_start_info; + +static DWORD CALLBACK +pg_thrd_invoke(void *argument) +{ + pg_thrd_start_info start = *(pg_thrd_start_info *) argument; +#ifdef PG_THRD_CURRENT_CONFORMING + pg_thrd_current_thread_local = start.self; +#endif + free(argument); + return start.function(start.argument); +} + +int +pg_thrd_create_impl(pg_thrd_t *thread, pg_thrd_start_t function, void *argument) +{ + pg_thrd_start_info *start; + DWORD flags; + + start = malloc(sizeof(*start)); + if (start == NULL) + return pg_thrd_nomem; + start->function = function; + start->argument = argument; + +#ifdef PG_THRD_CURRENT_CONFORMING + flags = CREATE_SUSPENDED; /* wait for "self" to be stored */ +#else + flags = 0; /* run immediately */ +#endif + + thread->handle = CreateThread(NULL, 0, pg_thrd_invoke, start, flags, 0); + if (thread->handle == NULL) + { + free(start); + return pg_thrd_map_windows_error(GetLastError()); + } + + thread->id = GetThreadId(thread->handle); + if (thread->id == 0) + { + /* + * The only documented failures involve bad handles and rights. + * CreateThread() grants THREAD_ALL_ACCESS, so failure shouldn't be + * possible. We can't even recover from an error at this point, given + * advice that even suspended threads shouldn't be terminated, and we + * surely can't just leave a thread behind. Log the error and abort. + */ + fprintf(stderr, + "GetThreadId() unexpectedly failed with error %u\n", + (unsigned int) GetLastError()); + abort(); + } + +#ifdef PG_THRD_CURRENT_CONFORMING + start->self = *thread; + if (!ResumeThread(thread->handle)) + { + /* Previous comment applies here too. */ + fprintf(stderr, + "ResumeThread() unexpectedly failed with error %u\n", + (unsigned int) GetLastError()); + abort(); + } +#endif + + return pg_thrd_success; +} + + +pg_thrd_t +pg_thrd_current_impl(void) +{ + if (pg_thrd_current_thread_local.id == 0) + { + /* + * Alien thread, or result of pg_thrd_current (and + * PG_THRD_CURRENT_CONFORMING is not defined). Populate just the + * thread ID on first call from this thread. Result cannot be used to + * join or detach. + */ + pg_thrd_current_thread_local.id = GetCurrentThreadId(); + } + + return pg_thrd_current_thread_local; +} + +int +pg_thrd_join_impl(pg_thrd_t thread, int *result) +{ + if (thread.handle == NULL) + { + /* + * This thread value must have come from pg_thrd_current() (and + * PG_THRD_CURRENT_CONFORMING is not defined, or it was called from + * an alien thread not created by pg_thrd_create()). + * + * We can't just use OpenThread(thread.id) to get a handle. We'd have + * to trust the caller to hold a handle somewhere else and not close + * it concurrently, and then close it later to avoid leaking kernel + * resources. That doesn't seem workable. + */ + return pg_thrd_format_error("pg_thrd_join(): no handle for thread %u", + (unsigned int) thread.id); + } + + if (WaitForSingleObject(thread.handle, INFINITE) == WAIT_OBJECT_0) + { + if (result) + { + DWORD exit_value; + + /* + * All bits survive casting through the Windows exit value type. + * In practice, only the sign differs, and you can't join alien + * threads in this implementation (see above). + */ + static_assert(sizeof(*result) <= sizeof(exit_value), + "thread exit value type is not big enough"); + + if (!GetExitCodeThread(thread.handle, &exit_value)) + return pg_thrd_format_error("pg_thrd_join(): GetExitCodeThread() failed with error %u", + (unsigned int) GetLastError()); + + if (result) + *result = exit_value; + } + return pg_thrd_map(CloseHandle(thread.handle)); + } + return pg_thrd_format_error("pg_thrd_join(): WaitForSingleObject() failed with error %u", + (unsigned int) GetLastError()); +} + +int +pg_thrd_equal_impl(pg_thrd_t lhs, pg_thrd_t rhs) +{ + return lhs.id == rhs.id; +} + +int +pg_thrd_detach_impl(pg_thrd_t thread) +{ + if (thread.handle == NULL) + { + /* + * See also pg_thrd_join_impl(). + * + * Windows threads are detached by closing all open handles, but we + * don't know anything about the handles of alien threads, or the + * result of pg_thrd_current() (unless PG_THRD_CURRENT_CONFORMING is + * defined). + */ + return pg_thrd_format_error("pg_thrd_detach(): no handle for thread %u", + (unsigned int) thread.id); + } + + return pg_thrd_map(CloseHandle(thread.handle)); +} + +int +pg_thrd_sleep_impl(const struct timespec *duration, struct timespec *remaining) +{ + long long ms = duration->tv_sec * 1000 + duration->tv_nsec / 1000000; + + if (ms < 0) + ms = 0; + if (ms > INFINITE) + ms = INFINITE; + + Sleep(ms); + + return 0; +} + +int +pg_mtx_init_impl(pg_mtx_impl *mutex, int type) +{ + int result = pg_mtx_validate_type(type); + + if (result != pg_thrd_success) + return result; + + if (type & pg_mtx_recursive) + { + mutex->type = PG_MTX_CRITICAL_SECTION; + InitializeCriticalSection(&mutex->critical_section); + } + else + { + mutex->type = PG_MTX_SRWLOCK; + InitializeSRWLock(&mutex->srwlock); + } + return pg_thrd_success; +} + +void +pg_mtx_destroy_impl(pg_mtx_impl *mutex) +{ + if (mutex->type == PG_MTX_CRITICAL_SECTION) + DeleteCriticalSection(&mutex->critical_section); +} + +/* + * Windows own FLS destructors don't have the right semantics to use directly + * (among other problems, they call destructors on all values from all + * threads, from the wrong thread, when you delete IDs). We make our own + * destructor table and per-thread value array to implement the standard + * semantics. + */ +static void CALLBACK +pg_tss_fls_destructor(void *data) +{ + for (int iter = 0; iter < PG_TSS_DTOR_ITERATIONS_IMPL; ++iter) + { + bool called_destructor = false; + + for (int i = 0; i < lengthof(pg_tss_values); ++i) + { + void *value = pg_tss_values[i].value; + pg_tss_dtor_t destructor = NULL; + + if (value == NULL) + continue; + + pg_tss_values[i].value = NULL; + + pg_mtx_lock(&pg_tss_lock); + if (pg_tss_slots[i].in_use && + pg_tss_slots[i].generation == pg_tss_values[i].generation) + destructor = pg_tss_slots[i].destructor; + pg_mtx_unlock(&pg_tss_lock); + + if (destructor) + { + called_destructor = true; + destructor(value); + } + } + + if (!called_destructor) + break; + } +} + +static bool +pg_tss_ensure_fls_registered(void) +{ + bool success; + + if (pg_tss_fls != FLS_OUT_OF_INDEXES) + return true; /* fast exit, already set */ + + pg_mtx_lock(&pg_tss_lock); + if (pg_tss_fls != FLS_OUT_OF_INDEXES) + { + success = true; + } + else + { + pg_tss_fls = FlsAlloc(pg_tss_fls_destructor); + if (pg_tss_fls != FLS_OUT_OF_INDEXES) + success = true; + } + pg_mtx_unlock(&pg_tss_lock); + + return success; +} + +int +pg_tss_create_impl(pg_tss_t *id, pg_tss_dtor_t dtor) +{ + bool created = false; + + if (!pg_tss_ensure_fls_registered()) + return pg_thrd_map_windows_error(GetLastError()); + + pg_mtx_lock(&pg_tss_lock); + for (int i = 0; i < lengthof(pg_tss_slots); ++i) + { + if (!pg_tss_slots[i].in_use) + { + id->generation = ++pg_tss_slots[i].generation; + id->index = i; + pg_tss_slots[i].destructor = dtor; + pg_tss_slots[i].in_use = true; + created = true; + break; + } + } + pg_mtx_unlock(&pg_tss_lock); + + if (!created) + return pg_thrd_format_error("pg_tss_create(): limit of %d IDs exceeded", + PG_TSS_MAX); + + return pg_thrd_success; +} + +void +pg_tss_delete_impl(pg_tss_t id) +{ + pg_mtx_lock(&pg_tss_lock); + pg_tss_slots[id.index].in_use = false; + pg_mtx_unlock(&pg_tss_lock); +} + +int +pg_tss_set_impl(pg_tss_t id, void *value) +{ + /* + * The generation prevents destructors from being called with leftover + * junk values if id is deleted and then a new one is created with the + * same index. + */ + pg_tss_values[id.index].value = value; + pg_tss_values[id.index].generation = id.generation; + + /* Dummy value so that pg_tss_fls_destructor() runs in this thread. */ + FlsSetValue(pg_tss_fls, (void *) 1); + + return pg_thrd_success; +} + +void * +pg_tss_get_impl(pg_tss_t id) +{ + /* + * pg_tss_{get,set} seem like candidates for inlining, but for now we keep + * all pg_tss_t functions out-of-line to avoid potential problems with + * cross-DLL thread-locals (not investigated). + */ + return pg_tss_values[id.index].value; +} + +#endif /* PG_THREAD_USE_WINDOWS_H */ + +#ifdef PG_THREADS_USE_PTHREAD_H + +/*------------------------------------------------------------------------- + * + * map_pthread.h + * + *------------------------------------------------------------------------- + */ + +typedef struct pg_thrd_start_info +{ + pg_thrd_start_t function; + void *argument; +} pg_thrd_start_info; + +static void * +pg_thrd_invoke(void *argument) +{ + pg_thrd_start_info start = *(pg_thrd_start_info *) argument; + + free(argument); + return (void *) (intptr_t) start.function(start.argument); +} + +int +pg_thrd_create_impl(pg_thrd_t *thread, pg_thrd_start_t function, void *argument) +{ + pg_thrd_start_info *start; + int pthread_result; + + /* + * Thread start function has a different return type, so we need an + * intermediate invoker function that casts the return value. (Simply + * casting the function pointer type might work in practice, but that'd be + * undefined behavior.) + */ + start = malloc(sizeof(*start)); + if (start == NULL) + return pg_thrd_nomem; + start->function = function; + start->argument = argument; + + pthread_result = pthread_create(thread, NULL, pg_thrd_invoke, start); + if (pthread_result != 0) + free(start); + + return pg_thrd_map(pthread_result); +} + +int +pg_thrd_join_impl(pg_thrd_t thread, int *result) +{ + void *exit_value = NULL; + int error = pg_thrd_map(pthread_join(thread, &exit_value)); + + /* + * All bits survive casting through the pthread exit value type. + * + * Joining a thread that exits via pthread_exit() or is cancelled with + * pthread_cancel() is undefined behavior. See POSIX's discussion of + * thrd_join(). + */ + static_assert(sizeof(*result) <= sizeof(exit_value), + "thread exit value type is not big enough"); + + if (result) + *result = (int) (intptr_t) exit_value; + + return error; +} + +void +pg_thrd_exit_impl(int result) +{ + pthread_exit((void *) (intptr_t) result); +} + +int +pg_mtx_init_impl(pg_mtx_impl *mtx, int type) +{ + pthread_mutexattr_t attr; + int result = pg_mtx_validate_type(type); + + if (result != pg_thrd_success) + return result; + + pthread_mutexattr_init(&attr); + if (type & pg_mtx_recursive_impl) + pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_RECURSIVE); + result = pg_thrd_map(pthread_mutex_init(mtx, &attr)); + pthread_mutexattr_destroy(&attr); + + return result; +} + +int +pg_cnd_init_impl(pg_cnd_impl *cnd) +{ + return pg_thrd_map(pthread_cond_init(cnd, NULL)); +} + +#endif /* PG_THREAD_USE_PTHREAD_H */ diff --git a/src/test/regress/expected/port.out b/src/test/regress/expected/port.out new file mode 100644 index 00000000000..11ab2a95cab --- /dev/null +++ b/src/test/regress/expected/port.out @@ -0,0 +1,17 @@ +-- +-- Test port code. +-- +-- directory paths and dlsuffix are passed to us in environment variables +\getenv libdir PG_LIBDIR +\getenv dlsuffix PG_DLSUFFIX +\set regresslib :libdir '/regress' :dlsuffix +CREATE FUNCTION test_pg_threads_ext() + RETURNS void + AS :'regresslib' + LANGUAGE C; +SELECT test_pg_threads_ext(); + test_pg_threads_ext +--------------------- + +(1 row) + diff --git a/src/test/regress/parallel_schedule b/src/test/regress/parallel_schedule index 8fa0a6c47fb..238d86c7440 100644 --- a/src/test/regress/parallel_schedule +++ b/src/test/regress/parallel_schedule @@ -9,7 +9,7 @@ # ---------- # required setup steps -test: test_setup +test: test_setup port # ---------- # The first group of parallel tests diff --git a/src/test/regress/regress.c b/src/test/regress/regress.c index 14d301b3499..80d6afe3410 100644 --- a/src/test/regress/regress.c +++ b/src/test/regress/regress.c @@ -40,6 +40,7 @@ #include "optimizer/plancat.h" #include "parser/parse_coerce.h" #include "port/atomics.h" +#include "port/pg_threads_ext.h" #include "portability/instr_time.h" #include "postmaster/postmaster.h" /* for MAX_BACKENDS */ #include "storage/spin.h" @@ -502,6 +503,14 @@ wait_pid(PG_FUNCTION_ARGS) PG_RETURN_VOID(); } +PG_FUNCTION_INFO_V1(test_pg_threads_ext); +Datum +test_pg_threads_ext(PG_FUNCTION_ARGS) +{ + pg_threads_ext_assertions(); + PG_RETURN_VOID(); +} + static void test_atomic_flag(void) { diff --git a/src/test/regress/sql/port.sql b/src/test/regress/sql/port.sql new file mode 100644 index 00000000000..31004695bc6 --- /dev/null +++ b/src/test/regress/sql/port.sql @@ -0,0 +1,16 @@ +-- +-- Test port code. +-- + +-- directory paths and dlsuffix are passed to us in environment variables +\getenv libdir PG_LIBDIR +\getenv dlsuffix PG_DLSUFFIX + +\set regresslib :libdir '/regress' :dlsuffix + +CREATE FUNCTION test_pg_threads_ext() + RETURNS void + AS :'regresslib' + LANGUAGE C; + +SELECT test_pg_threads_ext(); diff --git a/src/tools/pginclude/headerscheck b/src/tools/pginclude/headerscheck index 785d6f867ad..7ebafb31036 100755 --- a/src/tools/pginclude/headerscheck +++ b/src/tools/pginclude/headerscheck @@ -107,6 +107,9 @@ do test "$f" = src/include/port/win32ntdll.h && continue test "$f" = src/port/pthread-win32.h && continue + # Platform-specific headers that are included by pg_threads.h. + case "$f" in src/include/port/pg_threads/*.h) continue ; esac + # Likewise, these files are platform-specific, and the one # relevant to our platform will be included by atomics.h. test "$f" = src/include/port/atomics/arch-arm.h && continue diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 85d989f395d..11000e21266 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -4027,11 +4027,14 @@ pgParameterStatus pg_atomic_flag pg_atomic_uint32 pg_atomic_uint64 +pg_barrier_t pg_be_sasl_mech pg_category_range pg_checksum_context pg_checksum_raw_context pg_checksum_type +pg_cnd_impl +pg_cnd_t pg_compress_algorithm pg_compress_specification pg_conn_host @@ -4056,13 +4059,18 @@ pg_local_to_utf_combined pg_locale_t pg_mb_radix_tree pg_md5_ctx +pg_mtx_impl +pg_mtx_t pg_on_exit_callback pg_plan_advice_advisor_hook +pg_once_flag pg_prng_state pg_re_flags pg_regex_t pg_regmatch_t pg_regoff_t +pg_rwlock_impl +pg_rwlock_t pg_saslprep_rc pg_saslprep_test_context pg_sha1_ctx @@ -4075,8 +4083,17 @@ pg_snapshot pg_special_case pg_stack_base_t pg_ternary +pg_thrd_barrier_impl +pg_thrd_barrier_t +pg_thrd_start_info +pg_thrd_impl +pg_thrd_t pg_time_t pg_time_usec_t +pg_tss_impl +pg_tss_slot +pg_tss_t +pg_tss_value pg_tz pg_tz_cache pg_tzenum -- 2.54.0