On Mon, Oct 18, 2021 at 10:07 AM Thomas Munro <thomas.mu...@gmail.com> wrote:
> Note: The best kind would be *unnamed* POSIX semas, where we get to
> control their placement in existing memory; that's what we do on Linux
> and FreeBSD. They weren't supported on OpenBSD last time we checked:
> it rejects requests for shared ones. I wonder if someone could
> implement them with just a few lines of user space code, using atomic
> counters and futex() for waiting.
I meant that it'd be cool if OpenBSD implemented shared memory unnamed
semas that way (as other OSes do), but just for fun I tried
implementing that in PostgreSQL. I already had a patch to provide a
wrapper API for futexes on a bunch of OSes including OpenBSD (because
I've been looking into ways to rewrite lwlock.c to use futexes
directly and skip all the per-backend semaphore stuff). That made it
easy to write a quick-and-dirty clone of sem_{init,wait,post}() using
atomics and futexes.
Sadly, although the attached proof-of-concept patch allows a
PREFERRED_SEMAPHORES=FUTEX build to pass tests on macOS (which also
lacks native unnamed semas), FreeBSD and Linux (which don't need this
but are interesting to test), and it also works on OpenBSD with
shared_memory_type=sysv, it doesn't work on OpenBSD with
shared_memory_type=mmap (the default). I suspect OpenBSD's futex(2)
has a bug: inherited anonymous shared mmap memory seems to confuse it
so that wakeups are lost. Arrrgh!
From 38bb61aeeefc24cdc772474b0fb4f1802d3b7577 Mon Sep 17 00:00:00 2001
From: Thomas Munro <thomas.mu...@gmail.com>
Date: Sun, 24 Oct 2021 21:48:26 +1300
Subject: [PATCH 1/2] A basic API for futexes.
A thin wrapper for basic 32 bit futex wait and wake. Currently, it maps
to native support on Linux, FreeBSD, OpenBSD and macOS, with detection
via configure. More operating systems and more operations are possible.
---
configure | 4 +-
configure.ac | 4 ++
src/include/pg_config.h.in | 12 ++++
src/include/port/pg_futex.h | 139 ++++++++++++++++++++++++++++++++++++
4 files changed, 157 insertions(+), 2 deletions(-)
create mode 100644 src/include/port/pg_futex.h
diff --git a/configure b/configure
index 4ffefe4655..23f1cbe9d0 100755
--- a/configure
+++ b/configure
@@ -13381,7 +13381,7 @@ $as_echo "#define HAVE_STDBOOL_H 1" >>confdefs.h
fi
-for ac_header in atomic.h copyfile.h execinfo.h getopt.h ifaddrs.h langinfo.h mbarrier.h poll.h sys/epoll.h sys/event.h sys/ipc.h sys/prctl.h sys/procctl.h sys/pstat.h sys/resource.h sys/select.h sys/sem.h sys/shm.h sys/sockio.h sys/tas.h sys/uio.h sys/un.h termios.h ucred.h wctype.h
+for ac_header in atomic.h copyfile.h execinfo.h getopt.h ifaddrs.h langinfo.h linux/futex.h mbarrier.h poll.h sys/epoll.h sys/event.h sys/futex.h sys/ipc.h sys/prctl.h sys/procctl.h sys/pstat.h sys/resource.h sys/select.h sys/sem.h sys/shm.h sys/sockio.h sys/tas.h sys/uio.h sys/umtx.h sys/un.h termios.h ucred.h wctype.h
do :
as_ac_Header=`$as_echo "ac_cv_header_$ac_header" | $as_tr_sh`
ac_fn_c_check_header_mongrel "$LINENO" "$ac_header" "$as_ac_Header" "$ac_includes_default"
@@ -15492,7 +15492,7 @@ fi
LIBS_including_readline="$LIBS"
LIBS=`echo "$LIBS" | sed -e 's/-ledit//g' -e 's/-lreadline//g'`
-for ac_func in backtrace_symbols clock_gettime copyfile fdatasync getifaddrs getpeerucred getrlimit kqueue mbstowcs_l memset_s poll posix_fallocate ppoll pstat pthread_is_threaded_np readlink readv setproctitle setproctitle_fast setsid shm_open strchrnul strsignal symlink syncfs sync_file_range uselocale wcstombs_l writev
+for ac_func in __ulock_wait backtrace_symbols clock_gettime copyfile fdatasync getifaddrs getpeerucred getrlimit kqueue mbstowcs_l memset_s poll posix_fallocate ppoll pstat pthread_is_threaded_np readlink readv setproctitle setproctitle_fast setsid shm_open strchrnul strsignal symlink syncfs sync_file_range uselocale wcstombs_l writev
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"
diff --git a/configure.ac b/configure.ac
index 44ee3ebe2f..542b46437e 100644
--- a/configure.ac
+++ b/configure.ac
@@ -1363,10 +1363,12 @@ AC_CHECK_HEADERS(m4_normalize([
getopt.h
ifaddrs.h
langinfo.h
+ linux/futex.h
mbarrier.h
poll.h
sys/epoll.h
sys/event.h
+ sys/futex.h
sys/ipc.h
sys/prctl.h
sys/procctl.h
@@ -1378,6 +1380,7 @@ AC_CHECK_HEADERS(m4_normalize([
sys/sockio.h
sys/tas.h
sys/uio.h
+ sys/umtx.h
sys/un.h
termios.h
ucred.h
@@ -1698,6 +1701,7 @@ LIBS_including_readline="$LIBS"
LIBS=`echo "$LIBS" | sed -e 's/-ledit//g' -e 's/-lreadline//g'`
AC_CHECK_FUNCS(m4_normalize([
+ __ulock_wait
backtrace_symbols
clock_gettime
copyfile
diff --git a/src/include/pg_config.h.in b/src/include/pg_config.h.in
index 15ffdd895a..6bd2f7b5d8 100644
--- a/src/include/pg_config.h.in
+++ b/src/include/pg_config.h.in
@@ -367,6 +367,9 @@
/* Define to 1 if you have the `link' function. */
#undef HAVE_LINK
+/* Define to 1 if you have the <linux/futex.h> header file. */
+#undef HAVE_LINUX_FUTEX_H
+
/* Define to 1 if the system has the type `locale_t'. */
#undef HAVE_LOCALE_T
@@ -623,6 +626,9 @@
/* Define to 1 if you have the <sys/event.h> header file. */
#undef HAVE_SYS_EVENT_H
+/* Define to 1 if you have the <sys/futex.h> header file. */
+#undef HAVE_SYS_FUTEX_H
+
/* Define to 1 if you have the <sys/ipc.h> header file. */
#undef HAVE_SYS_IPC_H
@@ -665,6 +671,9 @@
/* Define to 1 if you have the <sys/uio.h> header file. */
#undef HAVE_SYS_UIO_H
+/* Define to 1 if you have the <sys/umtx.h> header file. */
+#undef HAVE_SYS_UMTX_H
+
/* Define to 1 if you have the <sys/un.h> header file. */
#undef HAVE_SYS_UN_H
@@ -779,6 +788,9 @@
/* Define to 1 if you have the `__strtoull' function. */
#undef HAVE___STRTOULL
+/* Define to 1 if you have the `__ulock_wait' function. */
+#undef HAVE___ULOCK_WAIT
+
/* Define to the appropriate printf length modifier for 64-bit ints. */
#undef INT64_MODIFIER
diff --git a/src/include/port/pg_futex.h b/src/include/port/pg_futex.h
new file mode 100644
index 0000000000..bf254a18b0
--- /dev/null
+++ b/src/include/port/pg_futex.h
@@ -0,0 +1,139 @@
+/*
+ * Minimal wrapper over futex APIs.
+ */
+
+#ifndef PG_FUTEX_H
+#define PG_FUTEX_H
+
+#if defined(HAVE_LINUX_FUTEX_H)
+
+/* https://man7.org/linux/man-pages/man2/futex.2.html */
+
+#include <linux/futex.h>
+#include <sys/syscall.h>
+
+#elif defined(HAVE_SYS_FUTEX_H)
+
+/* https://man.openbsd.org/futex, since OpenBSD 6.2. */
+
+#include <sys/time.h>
+#include <sys/futex.h>
+
+#elif defined(HAVE_SYS_UMTX_H)
+
+/* https://www.freebsd.org/cgi/man.cgi?query=_umtx_op */
+
+#include <sys/types.h>
+#include <sys/umtx.h>
+
+#elif defined(HAVE___ULOCK_WAIT)
+
+/*
+ * This interface is undocumented, but provided by libSystem.dylib since
+ * xnu-3789.1.32 (macOS 10.12, 2016) and is used by eg libc++.
+ *
+ * https://github.com/apple/darwin-xnu/blob/main/bsd/kern/sys_ulock.c
+ * https://github.com/apple/darwin-xnu/blob/main/bsd/sys/ulock.h
+ */
+
+#include <stdint.h>
+
+#define UL_COMPARE_AND_WAIT_SHARED 3
+#define ULF_WAKE_ALL 0x00000100
+extern int __ulock_wait(uint32_t operation,
+ void *addr,
+ uint64_t value,
+ uint32_t timeout);
+extern int __ulock_wake(uint32_t operation,
+ void *addr,
+ uint64_t wake_value);
+
+#endif
+
+/*
+ * Wait for someone to call pg_futex_wake() for the same address, with an
+ * initial check that the value pointed to by 'fut' matches 'value' and an
+ * optional timeout. Returns 0 when woken, and otherwise -1, with errno set to
+ * EAGAIN if the initial value check fails, and otherwise errors including
+ * EINTR, ETIMEDOUT and EFAULT.
+ */
+static int
+pg_futex_wait_u32(volatile void *fut,
+ uint32 value,
+ struct timespec *timeout)
+{
+#if defined(HAVE_LINUX_FUTEX_H)
+ if (syscall(SYS_futex, fut, FUTEX_WAIT, value, timeout, 0, 0) == 0)
+ return 0;
+#elif defined(HAVE_SYS_FUTEX_H)
+ if ((errno = futex(fut, FUTEX_WAIT, (int) value, timeout, NULL)) == 0)
+ return 0;
+ if (errno == ECANCELED)
+ errno = EINTR;
+#elif defined(HAVE_SYS_UMTX_H)
+ if (_umtx_op(fut, UMTX_OP_WAIT_UINT, value, 0, timeout) == 0)
+ return 0;
+#elif defined (HAVE___ULOCK_WAIT)
+ if (__ulock_wait(UL_COMPARE_AND_WAIT_SHARED,
+ (void *) fut,
+ value,
+ timeout ? timeout->tv_sec * 1000000 + timeout->tv_nsec / 1000 : 0) >= 0)
+ return 0;
+#else
+ /*
+ * If we wanted to simulate futexes on systems that don't have them, here
+ * we could add a link from our PGPROC struct to a shared memory hash
+ * table using "fut" (ie address) as the key, then compare *fut == value.
+ * If false, remove link and fail with EAGAIN. If so, sleep on proc latch.
+ * The main complication is that it wouldn't work for DSM segments; for
+ * those, we could have variants that take a dsm_segment and pointer and
+ * convert that to segment key + offset.
+ */
+ errno = ENOSYS;
+#endif
+
+ Assert(errno != 0);
+
+ return -1;
+}
+
+/*
+ * Wake up to nwaiters waiters that currently wait on the same address as
+ * 'fut'. Returns 0 on success, and -1 on failure, with errno set. Though
+ * some of these interfaces can tell us how many were woken, they can't all do
+ * that, so we'll hide that information.
+ */
+static int
+pg_futex_wake(volatile void *fut, int nwaiters)
+{
+#if defined(HAVE_LINUX_FUTEX_H)
+ if (syscall(SYS_futex, fut, FUTEX_WAKE, nwaiters, NULL, 0, 0) >= 0)
+ return 0;
+#elif defined(HAVE_SYS_FUTEX_H)
+ if (futex(fut, FUTEX_WAKE, nwaiters, NULL, NULL) >= 0)
+ return 0;
+#elif defined(HAVE_SYS_UMTX_H)
+ if (_umtx_op(fut, UMTX_OP_WAKE, nwaiters, 0, 0) == 0)
+ return 0;
+#elif defined (HAVE___ULOCK_WAIT)
+ if (__ulock_wake(UL_COMPARE_AND_WAIT_SHARED | (nwaiters > 1 ? ULF_WAKE_ALL : 0),
+ (void *) fut,
+ 0) >= 0)
+ return 0;
+ if (errno == ENOENT)
+ return 0;
+#else
+ /*
+ * If we wanted to simulate futexes on systems that don't have them, here
+ * we could look up "fut" in a shmem hash table set and set latches for
+ * any matches.
+ */
+ errno = ENOSYS;
+#endif
+
+ Assert(errno != 0);
+
+ return -1;
+}
+
+#endif
--
2.30.2
From bfe33b34dd6368c5c874d2777c045899272e7ff8 Mon Sep 17 00:00:00 2001
From: Thomas Munro <thomas.mu...@gmail.com>
Date: Sun, 24 Oct 2021 21:48:26 +1300
Subject: [PATCH 2/2] Add futex-based semaphore replacement.
Provide a drop-in replacement for POSIX unnamed semaphores using
futexes. Select with PREFERRED_SEMAPHORES=FUTEX. Perhaps useful on
OSes that lack shmem unnamed semaphores but have a futex facility.
XXX POC code only!
---
configure | 16 ++++-
configure.ac | 16 ++++-
src/backend/port/posix_sema.c | 106 +++++++++++++++++++++++++++++++++-
src/include/pg_config.h.in | 3 +
4 files changed, 134 insertions(+), 7 deletions(-)
diff --git a/configure b/configure
index 23f1cbe9d0..4e57c6436b 100755
--- a/configure
+++ b/configure
@@ -18486,6 +18486,10 @@ if test "$ac_res" != no; then :
fi
fi
+ if test x"$PREFERRED_SEMAPHORES" = x"FUTEX" ; then
+ # Need futex implementation for this
+ USE_FUTEX_SEMAPHORES=1
+ fi
{ $as_echo "$as_me:${as_lineno-$LINENO}: checking which semaphore API to use" >&5
$as_echo_n "checking which semaphore API to use... " >&6; }
if test x"$USE_NAMED_POSIX_SEMAPHORES" = x"1" ; then
@@ -18502,11 +18506,19 @@ $as_echo "#define USE_UNNAMED_POSIX_SEMAPHORES 1" >>confdefs.h
SEMA_IMPLEMENTATION="src/backend/port/posix_sema.c"
sematype="unnamed POSIX"
else
+ if test x"$USE_FUTEX_SEMAPHORES" = x"1" ; then
+
+$as_echo "#define USE_FUTEX_SEMAPHORES 1" >>confdefs.h
+
+ SEMA_IMPLEMENTATION="src/backend/port/posix_sema.c"
+ sematype="futex"
+ else
$as_echo "#define USE_SYSV_SEMAPHORES 1" >>confdefs.h
- SEMA_IMPLEMENTATION="src/backend/port/sysv_sema.c"
- sematype="System V"
+ SEMA_IMPLEMENTATION="src/backend/port/sysv_sema.c"
+ sematype="System V"
+ fi
fi
fi
{ $as_echo "$as_me:${as_lineno-$LINENO}: result: $sematype" >&5
diff --git a/configure.ac b/configure.ac
index 542b46437e..149ee401de 100644
--- a/configure.ac
+++ b/configure.ac
@@ -2217,6 +2217,10 @@ if test "$PORTNAME" != "win32"; then
# Need sem_init for this
AC_SEARCH_LIBS(sem_init, [rt pthread], [USE_UNNAMED_POSIX_SEMAPHORES=1])
fi
+ if test x"$PREFERRED_SEMAPHORES" = x"FUTEX" ; then
+ # Need futex implementation for this
+ USE_FUTEX_SEMAPHORES=1
+ fi
AC_MSG_CHECKING([which semaphore API to use])
if test x"$USE_NAMED_POSIX_SEMAPHORES" = x"1" ; then
AC_DEFINE(USE_NAMED_POSIX_SEMAPHORES, 1, [Define to select named POSIX semaphores.])
@@ -2228,9 +2232,15 @@ if test "$PORTNAME" != "win32"; then
SEMA_IMPLEMENTATION="src/backend/port/posix_sema.c"
sematype="unnamed POSIX"
else
- AC_DEFINE(USE_SYSV_SEMAPHORES, 1, [Define to select SysV-style semaphores.])
- SEMA_IMPLEMENTATION="src/backend/port/sysv_sema.c"
- sematype="System V"
+ if test x"$USE_FUTEX_SEMAPHORES" = x"1" ; then
+ AC_DEFINE(USE_FUTEX_SEMAPHORES, 1, [Define to select futex semaphores.])
+ SEMA_IMPLEMENTATION="src/backend/port/posix_sema.c"
+ sematype="futex"
+ else
+ AC_DEFINE(USE_SYSV_SEMAPHORES, 1, [Define to select SysV-style semaphores.])
+ SEMA_IMPLEMENTATION="src/backend/port/sysv_sema.c"
+ sematype="System V"
+ fi
fi
fi
AC_MSG_RESULT([$sematype])
diff --git a/src/backend/port/posix_sema.c b/src/backend/port/posix_sema.c
index 114da3b30c..b0c65d8716 100644
--- a/src/backend/port/posix_sema.c
+++ b/src/backend/port/posix_sema.c
@@ -36,6 +36,10 @@
#include "storage/pg_sema.h"
#include "storage/shmem.h"
+#if defined(USE_FUTEX_SEMAPHORES)
+#include "port/atomics.h"
+#include "port/pg_futex.h"
+#endif
/* see file header comment */
#if defined(USE_NAMED_POSIX_SEMAPHORES) && defined(EXEC_BACKEND)
@@ -45,6 +49,9 @@
typedef union SemTPadded
{
sem_t pgsem;
+#if defined(USE_FUTEX_SEMAPHORES)
+ pg_atomic_uint32 futexsem;
+#endif
char pad[PG_CACHE_LINE_SIZE];
} SemTPadded;
@@ -70,6 +77,78 @@ static int nextSemKey; /* next name to try */
static void ReleaseSemaphores(int status, Datum arg);
+#ifdef USE_FUTEX_SEMAPHORES
+
+/*
+ * An implementation of POSIX unnamed semaphores in shared memory, for OSes
+ * that lack them but have futexes.
+ */
+
+static void
+pg_futex_sem_init(pg_atomic_uint32 *fut, uint32 value)
+{
+ pg_atomic_init_u32(fut, value);
+}
+
+static int
+pg_futex_sem_wait(pg_atomic_uint32 *fut)
+{
+ /*
+ * This will only work if our atomic types are wrappers around raw
+ * integers, so that the kernel can test the value at that address.
+ */
+ StaticAssertStmt(sizeof(*fut) == sizeof(uint32),
+ "cannot use emulated atomics with futexes");
+
+ for (;;)
+ {
+ uint32 old_value = pg_atomic_read_u32(fut);
+ if (old_value == 0)
+ {
+ /* Wait for someone else to move it above 0. */
+ if (pg_futex_wait_u32(fut, 0, NULL) < 0)
+ {
+ if (errno != EAGAIN)
+ return -1;
+ /* The value changed under our feet. Try again. */
+ }
+ }
+ else
+ {
+ /* Try to decrement it. */
+ if (pg_atomic_compare_exchange_u32(fut, &old_value, old_value - 1))
+ return 0; /* success */
+ }
+ }
+
+ pg_unreachable();
+}
+
+static int
+pg_futex_sem_post(pg_atomic_uint32 *fut)
+{
+ for (;;)
+ {
+ uint32 old_value = pg_atomic_read_u32(fut);
+
+ if (pg_atomic_compare_exchange_u32(fut, &old_value, old_value + 1))
+ {
+ /*
+ * XXX TODO: encode nwaiters into value, so we can suppress useless
+ * wake calls.
+ */
+ if (pg_futex_wake(fut, INT_MAX) < 0)
+ {
+ /* Undo value change? */
+ return -1;
+ }
+ break;
+ }
+ }
+ return 0;
+}
+
+#endif
#ifdef USE_NAMED_POSIX_SEMAPHORES
@@ -124,7 +203,7 @@ PosixSemaphoreCreate(void)
return mySem;
}
-#else /* !USE_NAMED_POSIX_SEMAPHORES */
+#elif defined(USE_UNNAMED_POSIX_SEMAPHORES)
/*
* PosixSemaphoreCreate
@@ -139,6 +218,7 @@ PosixSemaphoreCreate(sem_t *sem)
}
#endif /* USE_NAMED_POSIX_SEMAPHORES */
+#ifndef USE_FUTEX_SEMAPHORES
/*
* PosixSemaphoreKill - removes a semaphore
@@ -156,6 +236,7 @@ PosixSemaphoreKill(sem_t *sem)
elog(LOG, "sem_destroy failed: %m");
#endif
}
+#endif
/*
@@ -239,18 +320,22 @@ PGReserveSemaphores(int maxSemas)
static void
ReleaseSemaphores(int status, Datum arg)
{
+#ifdef USE_NAMED_POSIX_SEMAPHORES
int i;
-#ifdef USE_NAMED_POSIX_SEMAPHORES
for (i = 0; i < numSems; i++)
PosixSemaphoreKill(mySemPointers[i]);
free(mySemPointers);
#endif
#ifdef USE_UNNAMED_POSIX_SEMAPHORES
+ int i;
+
for (i = 0; i < numSems; i++)
PosixSemaphoreKill(PG_SEM_REF(sharedSemas + i));
#endif
+
+ /* Futex-based semaphores have no kernel resource to clean up. */
}
/*
@@ -262,7 +347,9 @@ PGSemaphore
PGSemaphoreCreate(void)
{
PGSemaphore sema;
+#ifndef USE_FUTEX_SEMAPHORES
sem_t *newsem;
+#endif
/* Can't do this in a backend, because static state is postmaster's */
Assert(!IsUnderPostmaster);
@@ -275,6 +362,9 @@ PGSemaphoreCreate(void)
/* Remember new sema for ReleaseSemaphores */
mySemPointers[numSems] = newsem;
sema = (PGSemaphore) newsem;
+#elif defined(USE_FUTEX_SEMAPHORES)
+ sema = &sharedSemas[numSems];
+ pg_futex_sem_init(&sema->sem_padded.futexsem, 1);
#else
sema = &sharedSemas[numSems];
newsem = PG_SEM_REF(sema);
@@ -294,6 +384,9 @@ PGSemaphoreCreate(void)
void
PGSemaphoreReset(PGSemaphore sema)
{
+#ifdef USE_FUTEX_SEMAPHORES
+ pg_atomic_write_u32(&sema->sem_padded.futexsem, 0);
+#else
/*
* There's no direct API for this in POSIX, so we have to ratchet the
* semaphore down to 0 with repeated trywait's.
@@ -309,6 +402,7 @@ PGSemaphoreReset(PGSemaphore sema)
elog(FATAL, "sem_trywait failed: %m");
}
}
+#endif
}
/*
@@ -324,7 +418,11 @@ PGSemaphoreLock(PGSemaphore sema)
/* See notes in sysv_sema.c's implementation of PGSemaphoreLock. */
do
{
+#if defined(USE_FUTEX_SEMAPHORES)
+ errStatus = pg_futex_sem_wait(&sema->sem_padded.futexsem);
+#else
errStatus = sem_wait(PG_SEM_REF(sema));
+#endif
} while (errStatus < 0 && errno == EINTR);
if (errStatus < 0)
@@ -349,7 +447,11 @@ PGSemaphoreUnlock(PGSemaphore sema)
*/
do
{
+#if defined(USE_FUTEX_SEMAPHORES)
+ errStatus = pg_futex_sem_post(&sema->sem_padded.futexsem);
+#else
errStatus = sem_post(PG_SEM_REF(sema));
+#endif
} while (errStatus < 0 && errno == EINTR);
if (errStatus < 0)
diff --git a/src/include/pg_config.h.in b/src/include/pg_config.h.in
index 6bd2f7b5d8..b365d6ced3 100644
--- a/src/include/pg_config.h.in
+++ b/src/include/pg_config.h.in
@@ -916,6 +916,9 @@
/* Define to 1 to build with BSD Authentication support. (--with-bsd-auth) */
#undef USE_BSD_AUTH
+/* Define to select futex semaphores. */
+#undef USE_FUTEX_SEMAPHORES
+
/* Define to build with ICU support. (--with-icu) */
#undef USE_ICU
--
2.30.2
