pg.ddx.io pgsql-hackers@postgresql.org mailing list archive help / color / mirror / Atom feed From: Rady, Doug <radydoug@amazon.com>
To: pgsql-hackers@postgresql.org <pgsql-hackers@postgresql.org>
Cc: Robert Haas <robertmhaas@gmail.com>
Subject: PATCH: pgbench - option to build using ppoll() for larger connection counts
Date: Mon, 22 Jan 2018 08:34:06 +0000
Message-ID: <23D017C9-81B7-484D-8490-FD94DEC4DF59@amazon.com> (raw )
This version of the patch attempts to address the feedback for the previous
submission on 28-Nov-2017
This patch enables building pgbench to use ppoll() instead of select()
to allow for more than (FD_SETSIZE - 10) connections. As implemented,
when using ppoll(), the only connection limitation is system resources.
“… ppoll() is to poll() as pselect() is to select(). Since the
existing code uses select(), why not convert to poll() rather than
ppoll()?”
Time in pgbench is measured in microseconds.
The select() uses microseconds.
The ppoll() and pselect() call use nanoseconds
The poll() call uses milliseconds.
In order to not downgrade time resolution, ppoll() is used instead of poll().
Without this patch, one is limited to '(FD_SETSIZE - 10)’ number of connections.
Example of something that fails without this patch but works with the patch:
Without the patch:
$ pgbench -j 1500 -c 1500
invalid number of clients: "1500"
With the patch:
$ pgbench -j 1500 -c 1500
starting vacuum...end.
progress: 12.0 s, 782.3 tps, lat 917.507 ms stddev 846.929
transaction type: <builtin: TPC-B (sort of)>
scaling factor: 2000
query mode: simple
number of clients: 1500
number of threads: 1500
number of transactions per client: 10
number of transactions actually processed: 15000/15000
latency average = 1180.216 ms
latency stddev = 855.126 ms
tps = 658.674816 (including connections establishing)
tps = 765.289160 (excluding connections establishing)
--
Doug Rady
Amazon Aurora, RDS PostgreSQL
radydoug@amazon.com
Attachments:
[application/octet-stream] pgbench11-ppoll-v6.patch (10.7K, ../23D017C9-81B7-484D-8490-FD94DEC4DF59@amazon.com/3-pgbench11-ppoll-v6.patch)
download | inline diff:
diff --git a/configure.in b/configure.in
index 4d26034579..7b2b531675 100644
--- a/configure.in
+++ b/configure.in
@@ -1425,7 +1425,7 @@ PGAC_FUNC_WCSTOMBS_L
LIBS_including_readline="$LIBS"
LIBS=`echo "$LIBS" | sed -e 's/-ledit//g' -e 's/-lreadline//g'`
-AC_CHECK_FUNCS([cbrt clock_gettime dlopen fdatasync getifaddrs getpeerucred getrlimit mbstowcs_l memmove poll posix_fallocate pstat pthread_is_threaded_np readlink setproctitle setsid shm_open symlink sync_file_range utime utimes wcstombs_l])
+AC_CHECK_FUNCS([cbrt clock_gettime dlopen fdatasync getifaddrs getpeerucred getrlimit mbstowcs_l memmove poll posix_fallocate ppoll pstat pthread_is_threaded_np readlink setproctitle setsid shm_open symlink sync_file_range utime utimes wcstombs_l])
AC_REPLACE_FUNCS(fseeko)
case $host_os in
diff --git a/src/bin/pgbench/pgbench.c b/src/bin/pgbench/pgbench.c
index 31ea6ca06e..05773d8bb1 100644
--- a/src/bin/pgbench/pgbench.c
+++ b/src/bin/pgbench/pgbench.c
@@ -47,6 +47,9 @@
#ifdef HAVE_SYS_SELECT_H
#include <sys/select.h>
#endif
+#ifdef HAVE_POLL_H
+#include <poll.h>
+#endif
#ifdef HAVE_SYS_RESOURCE_H
#include <sys/resource.h> /* for getrlimit */
@@ -83,12 +86,28 @@ static int pthread_join(pthread_t th, void **thread_return);
/********************************************************************
* some configurable parameters */
-/* max number of clients allowed */
+#if defined(HAVE_PPOLL)
+#define USE_PPOLL 1
+#endif
+
+#if defined(USE_PPOLL)
+/* using ppoll(2) */
+#define SOCKET_WAIT_METHOD "ppoll"
+typedef struct pollfd socket_array;
+/* unlimited number of clients */
+#define MAXCLIENTS -1
+#else
+/* using select(2) */
+#define SOCKET_WAIT_METHOD "select"
+typedef fd_set socket_array;
#ifdef FD_SETSIZE
+/* system limited max number of clients allowed */
#define MAXCLIENTS (FD_SETSIZE - 10)
#else
+/* max number of clients allowed */
#define MAXCLIENTS 1024
#endif
+#endif
#define DEFAULT_INIT_STEPS "dtgvp" /* default -I setting */
@@ -502,6 +521,15 @@ static void pgbench_error(const char *fmt,...) pg_attribute_printf(1, 2);
static void addScript(ParsedScript script);
static void *threadRun(void *arg);
static void setalarm(int seconds);
+static void finishCon(CState *st);
+static socket_array *alloc_socket_array(int count);
+static void clear_socket(socket_array *sa, int fd, int idx);
+static int error_on_socket(socket_array *sa, int idx, PGconn *con);
+static void free_socket_array(socket_array *sa);
+static int ignore_socket(socket_array *sa, int idx, PGconn *con);
+static void init_socket_array(socket_array *sa, int count);
+static void set_socket(socket_array *sa, int fd, int idx);
+static int wait_on_socket_array(socket_array *sa, int nstate, int maxsock, int64 usec);
/* callback functions for our flex lexer */
@@ -2704,6 +2732,7 @@ doCustom(TState *thread, CState *st, StatsData *agg)
if (INSTR_TIME_IS_ZERO(now))
INSTR_TIME_SET_CURRENT(now);
start = now;
+
if ((st->con = doConnect()) == NULL)
{
fprintf(stderr, "client %d aborted while establishing connection\n",
@@ -2981,8 +3010,7 @@ doCustom(TState *thread, CState *st, StatsData *agg)
if (is_connect)
{
- PQfinish(st->con);
- st->con = NULL;
+ finishCon(st);
INSTR_TIME_SET_ZERO(now);
}
@@ -3019,11 +3047,7 @@ doCustom(TState *thread, CState *st, StatsData *agg)
*/
case CSTATE_ABORTED:
case CSTATE_FINISHED:
- if (st->con != NULL)
- {
- PQfinish(st->con);
- st->con = NULL;
- }
+ finishCon(st);
return;
}
}
@@ -3171,13 +3195,7 @@ disconnect_all(CState *state, int length)
int i;
for (i = 0; i < length; i++)
- {
- if (state[i].con)
- {
- PQfinish(state[i].con);
- state[i].con = NULL;
- }
- }
+ finishCon(&state[i]);
}
/*
@@ -4507,7 +4525,7 @@ main(int argc, char **argv)
case 'c':
benchmarking_option_set = true;
nclients = atoi(optarg);
- if (nclients <= 0 || nclients > MAXCLIENTS)
+ if (nclients <= 0 || (MAXCLIENTS != -1 && nclients > MAXCLIENTS))
{
fprintf(stderr, "invalid number of clients: \"%s\"\n",
optarg);
@@ -5189,6 +5207,7 @@ threadRun(void *arg)
int64 next_report = last_report + (int64) progress * 1000000;
StatsData last,
aggs;
+ socket_array *sockets = alloc_socket_array(nstate);
/*
* Initialize throttling rate target for all of the thread's clients. It
@@ -5232,6 +5251,7 @@ threadRun(void *arg)
{
if ((state[i].con = doConnect()) == NULL)
goto done;
+ set_socket(sockets, PQsocket(state[i].con), i);
}
}
@@ -5248,13 +5268,12 @@ threadRun(void *arg)
/* loop till all clients have terminated */
while (remains > 0)
{
- fd_set input_mask;
int maxsock; /* max socket number to be waited for */
int64 min_usec;
int64 now_usec = 0; /* set this only if needed */
- /* identify which client sockets should be checked for input */
- FD_ZERO(&input_mask);
+ init_socket_array(sockets, nstate);
+
maxsock = -1;
min_usec = PG_INT64_MAX;
for (i = 0; i < nstate; i++)
@@ -5265,8 +5284,7 @@ threadRun(void *arg)
{
/* interrupt client that has not started a transaction */
st->state = CSTATE_FINISHED;
- PQfinish(st->con);
- st->con = NULL;
+ finishCon(st);
remains--;
}
else if (st->state == CSTATE_SLEEP || st->state == CSTATE_THROTTLE)
@@ -5304,7 +5322,7 @@ threadRun(void *arg)
goto done;
}
- FD_SET(sock, &input_mask);
+ set_socket(sockets, sock, i);
if (maxsock < sock)
maxsock = sock;
}
@@ -5341,7 +5359,7 @@ threadRun(void *arg)
/*
* If no clients are ready to execute actions, sleep until we receive
* data from the server, or a nap-time specified in the script ends,
- * or it's time to print a progress report. Update input_mask to show
+ * or it's time to print a progress report. Update sockets to show
* which client(s) received data.
*/
if (min_usec > 0)
@@ -5352,11 +5370,7 @@ threadRun(void *arg)
{
if (maxsock != -1)
{
- struct timeval timeout;
-
- timeout.tv_sec = min_usec / 1000000;
- timeout.tv_usec = min_usec % 1000000;
- nsocks = select(maxsock + 1, &input_mask, NULL, NULL, &timeout);
+ nsocks = wait_on_socket_array(sockets, nstate, maxsock, min_usec);
}
else /* nothing active, simple sleep */
{
@@ -5365,7 +5379,7 @@ threadRun(void *arg)
}
else /* no explicit delay, select without timeout */
{
- nsocks = select(maxsock + 1, &input_mask, NULL, NULL, NULL);
+ nsocks = wait_on_socket_array(sockets, nstate, maxsock, 0);
}
if (nsocks < 0)
@@ -5376,7 +5390,7 @@ threadRun(void *arg)
continue;
}
/* must be something wrong */
- fprintf(stderr, "select() failed: %s\n", strerror(errno));
+ fprintf(stderr, "%s() failed: %s\n", SOCKET_WAIT_METHOD, strerror(errno));
goto done;
}
}
@@ -5385,7 +5399,7 @@ threadRun(void *arg)
/* min_usec == 0, i.e. something needs to be executed */
/* If we didn't call select(), don't try to read any data */
- FD_ZERO(&input_mask);
+ init_socket_array(sockets, nstate);
}
/* ok, advance the state machine of each connection */
@@ -5396,16 +5410,11 @@ threadRun(void *arg)
if (st->state == CSTATE_WAIT_RESULT)
{
/* don't call doCustom unless data is available */
- int sock = PQsocket(st->con);
- if (sock < 0)
- {
- fprintf(stderr, "invalid socket: %s",
- PQerrorMessage(st->con));
+ if (error_on_socket(sockets, i, st->con))
goto done;
- }
- if (!FD_ISSET(sock, &input_mask))
+ if (ignore_socket(sockets, i, st->con))
continue;
}
else if (st->state == CSTATE_FINISHED ||
@@ -5543,9 +5552,137 @@ done:
fclose(thread->logfile);
thread->logfile = NULL;
}
+ free_socket_array(sockets);
+ sockets = NULL;
return NULL;
}
+static void
+finishCon(CState *st)
+{
+ if (st->con)
+ {
+ PQfinish(st->con);
+ st->con = NULL;
+ }
+}
+
+static socket_array *
+alloc_socket_array(int count)
+{
+ return (socket_array *) pg_malloc0(sizeof(socket_array) * ((MAXCLIENTS > 0) ? 1 : count));
+}
+
+static void
+free_socket_array(socket_array *sa)
+{
+ pg_free(sa);
+}
+
+#if defined(USE_PPOLL)
+#ifdef POLLRDHUP
+#define POLL_EVENTS (POLLRDHUP|POLLIN|POLLPRI)
+#define POLL_ISSUES (POLLRDHUP|POLLERR|POLLHUP|POLLNVAL|POLLRDHUP)
+#else
+#define POLL_EVENTS (POLLIN|POLLPRI)
+#define POLL_ISSUES (POLLERR|POLLHUP|POLLNVAL)
+#endif
+
+static void
+clear_socket(socket_array *sa, int fd, int idx)
+{
+ set_socket(sa, -1, idx);
+}
+
+static int
+error_on_socket(socket_array *sa, int idx, PGconn *con)
+{
+ if (sa[idx].fd == -1 || (PQsocket(con) >= 0 && !(sa[idx].revents & POLL_ISSUES)))
+ return 0;
+ fprintf(stderr, "ppoll() fail - errno: %d, idx: %d, events: %x, %s\n",
+ errno, idx, (sa[idx].revents & POLL_ISSUES),
+ PQerrorMessage(con));
+ return 1;
+}
+
+static int
+ignore_socket(socket_array *sa, int idx, PGconn *con)
+{
+ return (sa[idx].fd != -1 && !sa[idx].revents) ? 1 : 0;
+}
+
+static void
+init_socket_array(socket_array *sa, int count)
+{
+ int i = 0;
+ for (i = 0; i < count; i++)
+ clear_socket(sa, -1, i);
+}
+
+static void
+set_socket(socket_array *sa, int fd, int idx)
+{
+ sa[idx].fd = fd;
+ sa[idx].events = POLL_EVENTS;
+ sa[idx].revents = 0;
+}
+
+static int
+wait_on_socket_array(socket_array *sa, int nstate, int maxsock, int64 usec)
+{
+ struct timespec timeout;
+
+ if (usec)
+ {
+ timeout.tv_sec = usec / 1000000;
+ timeout.tv_nsec = usec % 1000000000;
+ }
+ return ppoll(sa, nstate, usec ? &timeout : NULL, NULL);
+}
+#else
+static void
+clear_socket(socket_array *sa, int fd, int idx)
+{
+ FD_CLR(fd, sa);
+}
+
+static int
+error_on_socket(socket_array *sa, int idx, PGconn *con)
+{
+ if (PQsocket(con) >= 0) return 0;
+ fprintf(stderr, "invalid socket: %s", PQerrorMessage(con));
+ return 0;
+}
+
+static int
+ignore_socket(socket_array *sa, int idx, PGconn *con)
+{
+ return (!FD_ISSET(PQsocket(con), sa)) ? 1 : 0;
+}
+
+static void
+init_socket_array(socket_array *sa, int count)
+{
+ FD_ZERO(sa);
+}
+
+static void
+set_socket(socket_array *sa, int fd, int idx)
+{
+ FD_SET(fd, sa);
+}
+
+static int
+wait_on_socket_array(socket_array *sa, int nstate, int maxsock, int64 usec)
+{
+ struct timeval timeout;
+
+ timeout.tv_sec = usec / 1000000;
+ timeout.tv_usec = usec % 1000000;
+ return select(maxsock + 1, sa, NULL, NULL, usec ? &timeout : NULL);
+}
+#endif
+
/*
* Support for duration option: set timer_exceeded after so many seconds.
*/
diff --git a/src/include/pg_config.h.in b/src/include/pg_config.h.in
index f98f773ff0..63207e13fd 100644
--- a/src/include/pg_config.h.in
+++ b/src/include/pg_config.h.in
@@ -409,6 +409,9 @@
/* Define to 1 if the assembler supports PPC's LWARX mutex hint bit. */
#undef HAVE_PPC_LWARX_MUTEX_HINT
+/* Define to 1 if you have the `ppoll' function. */
+#undef HAVE_PPOLL
+
/* Define to 1 if you have the `pstat' function. */
#undef HAVE_PSTAT
view thread (51+ messages) latest in thread
Message-ID: <23D017C9-81B7-484D-8490-FD94DEC4DF59@amazon.com>
Permalink: ../23D017C9-81B7-484D-8490-FD94DEC4DF59@amazon.com/
Also on: postgresql.org/message-id/23D017C9-81B7-484D-8490-FD94DEC4DF59@amazon.com
copy link · copy postgr.es · copy git am
reply Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: pgsql-hackers@postgresql.org
Cc: radydoug@amazon.com, robertmhaas@gmail.com
Subject: Re: PATCH: pgbench - option to build using ppoll() for larger connection counts
In-Reply-To: <23D017C9-81B7-484D-8490-FD94DEC4DF59@amazon.com>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by DDX for PostgreSQL; see mirroring instructions
for how to clone and mirror all data and code used for this inbox