From a92a90f6de0d51d27aa791286af717ae8bf03aff Mon Sep 17 00:00:00 2001 From: Hannu Krosing Date: Sun, 4 Oct 2026 11:24:37 +0000 Subject: [PATCH v4 03/12] Refactor pgbench: extract socket polling abstraction to poller.c/h Move the multi-platform socket polling macros, `struct socket_set` definitions (for both POLL_USING_PPOLL and POLL_USING_SELECT), and the six socket_set functions out of pgbench.c into poller.c and poller.h. No function bodies, parameter lists, expressions, or comments are modified in this commit. Why function linkage changes are strictly necessary: 1. Removal of `static` on the 6 socket_set functions in poller.c (12 definitions across the `#ifdef POLL_USING_PPOLL` and `#ifdef POLL_USING_SELECT` compile-time branches): All 6 functions are defined in poller.c and called by threadRun() in pgbench.c (later engine.c), requiring external linkage and `extern` prototypes in poller.h: - alloc_socket_set() - free_socket_set() - clear_socket_set() - add_socket_to_set() - wait_on_socket_set() - socket_has_input() (Encapsulating `struct socket_set` as an opaque type and introducing the `PgBenchPoller` wrapper macros is deferred to commit 0010.) --- src/bin/pgbench/Makefile | 1 + src/bin/pgbench/meson.build | 1 + src/bin/pgbench/pgbench.c | 223 +----------------------------------- src/bin/pgbench/poller.c | 202 ++++++++++++++++++++++++++++++++ src/bin/pgbench/poller.h | 61 ++++++++++ 5 files changed, 266 insertions(+), 222 deletions(-) create mode 100644 src/bin/pgbench/poller.c create mode 100644 src/bin/pgbench/poller.h diff --git a/src/bin/pgbench/Makefile b/src/bin/pgbench/Makefile index d951b6d128a..1fd343e65b7 100644 --- a/src/bin/pgbench/Makefile +++ b/src/bin/pgbench/Makefile @@ -11,6 +11,7 @@ OBJS = \ $(WIN32RES) \ exprparse.o \ exprscan.o \ + poller.o \ stats.o \ variable.o \ pgbench.o diff --git a/src/bin/pgbench/meson.build b/src/bin/pgbench/meson.build index ef93805df30..770d580671f 100644 --- a/src/bin/pgbench/meson.build +++ b/src/bin/pgbench/meson.build @@ -2,6 +2,7 @@ pgbench_sources = files( 'pgbench.c', + 'poller.c', 'stats.c', 'variable.c', ) diff --git a/src/bin/pgbench/pgbench.c b/src/bin/pgbench/pgbench.c index d115bfd8bd5..0b6a42d60e5 100644 --- a/src/bin/pgbench/pgbench.c +++ b/src/bin/pgbench/pgbench.c @@ -42,16 +42,6 @@ #include #include /* for getrlimit */ -/* For testing, PGBENCH_USE_SELECT can be defined to force use of that code */ -#if defined(HAVE_PPOLL) && !defined(PGBENCH_USE_SELECT) -#define POLL_USING_PPOLL -#ifdef HAVE_POLL_H -#include -#endif -#else /* no ppoll(), so use select() */ -#define POLL_USING_SELECT -#include -#endif #include "catalog/pg_class_d.h" #include "common/int.h" @@ -66,6 +56,7 @@ #include "getopt_long.h" #include "libpq-fe.h" #include "pgbench.h" +#include "poller.h" #include "stats.h" #include "port/pg_bitutils.h" @@ -87,32 +78,6 @@ #define MM2_MUL_TIMES_8 UINT64CONST(0x35253c9ade8f4ca8) #define MM2_ROT 47 -/* - * Multi-platform socket set implementations - */ - -#ifdef POLL_USING_PPOLL -#define SOCKET_WAIT_METHOD "ppoll" - -typedef struct socket_set -{ - int maxfds; /* allocated length of pollfds[] array */ - int curfds; /* number currently in use */ - struct pollfd pollfds[FLEXIBLE_ARRAY_MEMBER]; -} socket_set; - -#endif /* POLL_USING_PPOLL */ - -#ifdef POLL_USING_SELECT -#define SOCKET_WAIT_METHOD "select" - -typedef struct socket_set -{ - int maxfd; /* largest FD currently set in fds */ - fd_set fds; -} socket_set; - -#endif /* POLL_USING_SELECT */ /* * Multi-platform thread implementations @@ -684,12 +649,6 @@ static void addScript(const ParsedScript *script); static THREAD_FUNC_RETURN_TYPE THREAD_FUNC_CC threadRun(void *arg); static void finishCon(CState *st); static void setalarm(int seconds); -static socket_set *alloc_socket_set(int count); -static void free_socket_set(socket_set *sa); -static void clear_socket_set(socket_set *sa); -static void add_socket_to_set(socket_set *sa, int fd, int idx); -static int wait_on_socket_set(socket_set *sa, int64 usecs); -static bool socket_has_input(socket_set *sa, int fd, int idx); /* callback used to build rows for COPY during data loading */ typedef void (*initRowMethod) (PQExpBufferData *sql, int64 curr); @@ -6945,183 +6904,3 @@ setalarm(int seconds) } #endif /* WIN32 */ - - -/* - * These functions provide an abstraction layer that hides the syscall - * we use to wait for input on a set of sockets. - * - * Currently there are two implementations, based on ppoll(2) and select(2). - * ppoll() is preferred where available due to its typically higher ceiling - * on the number of usable sockets. We do not use the more-widely-available - * poll(2) because it only offers millisecond timeout resolution, which could - * be problematic with high --rate settings. - * - * Function APIs: - * - * alloc_socket_set: allocate an empty socket set with room for up to - * "count" sockets. - * - * free_socket_set: deallocate a socket set. - * - * clear_socket_set: reset a socket set to empty. - * - * add_socket_to_set: add socket with indicated FD to slot "idx" in the - * socket set. Slots must be filled in order, starting with 0. - * - * wait_on_socket_set: wait for input on any socket in set, or for timeout - * to expire. timeout is measured in microseconds; 0 means wait forever. - * Returns result code of underlying syscall (>=0 if OK, else see errno). - * - * socket_has_input: after waiting, call this to see if given socket has - * input. fd and idx parameters should match some previous call to - * add_socket_to_set. - * - * Note that wait_on_socket_set destructively modifies the state of the - * socket set. After checking for input, caller must apply clear_socket_set - * and add_socket_to_set again before waiting again. - */ - -#ifdef POLL_USING_PPOLL - -static socket_set * -alloc_socket_set(int count) -{ - socket_set *sa; - - sa = (socket_set *) pg_malloc0(offsetof(socket_set, pollfds) + - sizeof(struct pollfd) * count); - sa->maxfds = count; - sa->curfds = 0; - return sa; -} - -static void -free_socket_set(socket_set *sa) -{ - pg_free(sa); -} - -static void -clear_socket_set(socket_set *sa) -{ - sa->curfds = 0; -} - -static void -add_socket_to_set(socket_set *sa, int fd, int idx) -{ - Assert(idx < sa->maxfds && idx == sa->curfds); - sa->pollfds[idx].fd = fd; - sa->pollfds[idx].events = POLLIN; - sa->pollfds[idx].revents = 0; - sa->curfds++; -} - -static int -wait_on_socket_set(socket_set *sa, int64 usecs) -{ - if (usecs > 0) - { - struct timespec timeout; - - timeout.tv_sec = usecs / 1000000; - timeout.tv_nsec = (usecs % 1000000) * 1000; - return ppoll(sa->pollfds, sa->curfds, &timeout, NULL); - } - else - { - return ppoll(sa->pollfds, sa->curfds, NULL, NULL); - } -} - -static bool -socket_has_input(socket_set *sa, int fd, int idx) -{ - /* - * In some cases, threadRun will apply clear_socket_set and then try to - * apply socket_has_input anyway with arguments that it used before that, - * or might've used before that except that it exited its setup loop - * early. Hence, if the socket set is empty, silently return false - * regardless of the parameters. If it's not empty, we can Assert that - * the parameters match a previous call. - */ - if (sa->curfds == 0) - return false; - - Assert(idx < sa->curfds && sa->pollfds[idx].fd == fd); - return (sa->pollfds[idx].revents & POLLIN) != 0; -} - -#endif /* POLL_USING_PPOLL */ - -#ifdef POLL_USING_SELECT - -static socket_set * -alloc_socket_set(int count) -{ - return pg_malloc0_object(socket_set); -} - -static void -free_socket_set(socket_set *sa) -{ - pg_free(sa); -} - -static void -clear_socket_set(socket_set *sa) -{ - FD_ZERO(&sa->fds); - sa->maxfd = -1; -} - -static void -add_socket_to_set(socket_set *sa, int fd, int idx) -{ - /* See connect_slot() for background on this code. */ -#ifdef WIN32 - if (sa->fds.fd_count + 1 >= FD_SETSIZE) - { - pg_log_error("too many concurrent database clients for this platform: %d", - sa->fds.fd_count + 1); - exit(1); - } -#else - if (fd < 0 || fd >= FD_SETSIZE) - { - pg_log_error("socket file descriptor out of range for select(): %d", - fd); - pg_log_error_hint("Try fewer concurrent database clients."); - exit(1); - } -#endif - FD_SET(fd, &sa->fds); - if (fd > sa->maxfd) - sa->maxfd = fd; -} - -static int -wait_on_socket_set(socket_set *sa, int64 usecs) -{ - if (usecs > 0) - { - struct timeval timeout; - - timeout.tv_sec = usecs / 1000000; - timeout.tv_usec = usecs % 1000000; - return select(sa->maxfd + 1, &sa->fds, NULL, NULL, &timeout); - } - else - { - return select(sa->maxfd + 1, &sa->fds, NULL, NULL, NULL); - } -} - -static bool -socket_has_input(socket_set *sa, int fd, int idx) -{ - return (FD_ISSET(fd, &sa->fds) != 0); -} - -#endif /* POLL_USING_SELECT */ diff --git a/src/bin/pgbench/poller.c b/src/bin/pgbench/poller.c new file mode 100644 index 00000000000..2bfc519d886 --- /dev/null +++ b/src/bin/pgbench/poller.c @@ -0,0 +1,202 @@ +/*------------------------------------------------------------------------- + * + * poller.c + * Multi-platform socket set implementations for pgbench + * + * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group + * Portions Copyright (c) 1994, Regents of the University of California + * + * src/bin/pgbench/poller.c + * + *------------------------------------------------------------------------- + */ +#if defined(WIN32) && FD_SETSIZE < 1024 +#error FD_SETSIZE needs to have been increased +#endif + +#include "postgres_fe.h" + +#include +#include + +#include "common/logging.h" +#include "poller.h" + +/* + * These functions provide an abstraction layer that hides the syscall + * we use to wait for input on a set of sockets. + * + * Currently there are two implementations, based on ppoll(2) and select(2). + * ppoll() is preferred where available due to its typically higher ceiling + * on the number of usable sockets. We do not use the more-widely-available + * poll(2) because it only offers millisecond timeout resolution, which could + * be problematic with high --rate settings. + * + * Function APIs: + * + * alloc_socket_set: allocate an empty socket set with room for up to + * "count" sockets. + * + * free_socket_set: deallocate a socket set. + * + * clear_socket_set: reset a socket set to empty. + * + * add_socket_to_set: add socket with indicated FD to slot "idx" in the + * socket set. Slots must be filled in order, starting with 0. + * + * wait_on_socket_set: wait for input on any socket in set, or for timeout + * to expire. timeout is measured in microseconds; 0 means wait forever. + * Returns result code of underlying syscall (>=0 if OK, else see errno). + * + * socket_has_input: after waiting, call this to see if given socket has + * input. fd and idx parameters should match some previous call to + * add_socket_to_set. + * + * Note that wait_on_socket_set destructively modifies the state of the + * socket set. After checking for input, caller must apply clear_socket_set + * and add_socket_to_set again before waiting again. + */ + +#ifdef POLL_USING_PPOLL + +socket_set * +alloc_socket_set(int count) +{ + socket_set *sa; + + sa = (socket_set *) pg_malloc0(offsetof(socket_set, pollfds) + + sizeof(struct pollfd) * count); + sa->maxfds = count; + sa->curfds = 0; + return sa; +} + +void +free_socket_set(socket_set *sa) +{ + pg_free(sa); +} + +void +clear_socket_set(socket_set *sa) +{ + sa->curfds = 0; +} + +void +add_socket_to_set(socket_set *sa, int fd, int idx) +{ + Assert(idx < sa->maxfds && idx == sa->curfds); + sa->pollfds[idx].fd = fd; + sa->pollfds[idx].events = POLLIN; + sa->pollfds[idx].revents = 0; + sa->curfds++; +} + +int +wait_on_socket_set(socket_set *sa, int64 usecs) +{ + if (usecs > 0) + { + struct timespec timeout; + + timeout.tv_sec = usecs / 1000000; + timeout.tv_nsec = (usecs % 1000000) * 1000; + return ppoll(sa->pollfds, sa->curfds, &timeout, NULL); + } + else + { + return ppoll(sa->pollfds, sa->curfds, NULL, NULL); + } +} + +bool +socket_has_input(socket_set *sa, int fd, int idx) +{ + /* + * In some cases, threadRun will apply clear_socket_set and then try to + * apply socket_has_input anyway with arguments that it used before that, + * or might've used before that except that it exited its setup loop + * early. Hence, if the socket set is empty, silently return false + * regardless of the parameters. If it's not empty, we can Assert that + * the parameters match a previous call. + */ + if (sa->curfds == 0) + return false; + + Assert(idx < sa->curfds && sa->pollfds[idx].fd == fd); + return (sa->pollfds[idx].revents & POLLIN) != 0; +} + +#endif /* POLL_USING_PPOLL */ + +#ifdef POLL_USING_SELECT + +socket_set * +alloc_socket_set(int count) +{ + return pg_malloc0_object(socket_set); +} + +void +free_socket_set(socket_set *sa) +{ + pg_free(sa); +} + +void +clear_socket_set(socket_set *sa) +{ + FD_ZERO(&sa->fds); + sa->maxfd = -1; +} + +void +add_socket_to_set(socket_set *sa, int fd, int idx) +{ + /* See connect_slot() for background on this code. */ +#ifdef WIN32 + if (sa->fds.fd_count + 1 >= FD_SETSIZE) + { + pg_log_error("too many concurrent database clients for this platform: %d", + sa->fds.fd_count + 1); + exit(1); + } +#else + if (fd < 0 || fd >= FD_SETSIZE) + { + pg_log_error("socket file descriptor out of range for select(): %d", + fd); + pg_log_error_hint("Try fewer concurrent database clients."); + exit(1); + } +#endif + FD_SET(fd, &sa->fds); + if (fd > sa->maxfd) + sa->maxfd = fd; +} + +int +wait_on_socket_set(socket_set *sa, int64 usecs) +{ + if (usecs > 0) + { + struct timeval timeout; + + timeout.tv_sec = usecs / 1000000; + timeout.tv_usec = usecs % 1000000; + return select(sa->maxfd + 1, &sa->fds, NULL, NULL, &timeout); + } + else + { + return select(sa->maxfd + 1, &sa->fds, NULL, NULL, NULL); + } +} + +bool +socket_has_input(socket_set *sa, int fd, int idx) +{ + return (FD_ISSET(fd, &sa->fds) != 0); +} + +#endif /* POLL_USING_SELECT */ diff --git a/src/bin/pgbench/poller.h b/src/bin/pgbench/poller.h new file mode 100644 index 00000000000..275e77d2137 --- /dev/null +++ b/src/bin/pgbench/poller.h @@ -0,0 +1,61 @@ +/*------------------------------------------------------------------------- + * + * poller.h + * Socket set polling abstraction for pgbench + * + * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group + * Portions Copyright (c) 1994, Regents of the University of California + * + * src/bin/pgbench/poller.h + * + *------------------------------------------------------------------------- + */ +#ifndef POLLER_H +#define POLLER_H + +/* For testing, PGBENCH_USE_SELECT can be defined to force use of that code */ +#if defined(HAVE_PPOLL) && !defined(PGBENCH_USE_SELECT) +#define POLL_USING_PPOLL +#ifdef HAVE_POLL_H +#include +#endif +#else /* no ppoll(), so use select() */ +#define POLL_USING_SELECT +#include +#endif + +/* + * Multi-platform socket set implementations + */ + +#ifdef POLL_USING_PPOLL +#define SOCKET_WAIT_METHOD "ppoll" + +typedef struct socket_set +{ + int maxfds; /* allocated length of pollfds[] array */ + int curfds; /* number currently in use */ + struct pollfd pollfds[FLEXIBLE_ARRAY_MEMBER]; +} socket_set; + +#endif /* POLL_USING_PPOLL */ + +#ifdef POLL_USING_SELECT +#define SOCKET_WAIT_METHOD "select" + +typedef struct socket_set +{ + int maxfd; /* largest FD currently set in fds */ + fd_set fds; +} socket_set; + +#endif /* POLL_USING_SELECT */ + +extern socket_set *alloc_socket_set(int count); +extern void free_socket_set(socket_set *sa); +extern void clear_socket_set(socket_set *sa); +extern void add_socket_to_set(socket_set *sa, int fd, int idx); +extern int wait_on_socket_set(socket_set *sa, int64 usecs); +extern bool socket_has_input(socket_set *sa, int fd, int idx); + +#endif /* POLLER_H */ -- 2.56.0.385.gd3acb90ef8-goog