From: Andrew Dunstan <andrew.dunstan@2ndquadrant.com>
Subject: Re: PATCH: pgbench - option to build using ppoll() for larger
 connection counts
To: Thomas Munro <thomas.munro@enterprisedb.com>,
 "Rady, Doug" <radydoug@amazon.com>
Cc: Fabien COELHO <coelho@cri.ensmp.fr>, Andres Freund <andres@anarazel.de>,
 "pgsql-hackers@postgresql.org" <pgsql-hackers@postgresql.org>,
 Robert Haas <robertmhaas@gmail.com>
References: <23D017C9-81B7-484D-8490-FD94DEC4DF59@amazon.com>
 <alpine.DEB.2.20.1801240847010.2307@lancre>
 <16744C85-E44E-4354-9E3D-45A7C9BDB19B@amazon.com>
 <alpine.DEB.2.20.1801252339520.26762@lancre>
 <81936381-841A-459D-9FE5-18659680355C@amazon.com>
 <alpine.DEB.2.20.1801262348420.17596@lancre>
 <654A1247-8E1E-4E9C-937D-9119B1731FF2@amazon.com>
 <20180301083404.z2sbnjdaw7gxvz6x@alap3.anarazel.de>
 <alpine.DEB.2.20.1803011121300.7903@lancre>
 <20180304001409.tajjmefj7yq6xs77@alap3.anarazel.de>
 <2879AF01-915B-4D07-9F42-04F7CA71908D@amazon.com>
 <alpine.DEB.2.20.1803251252390.8109@lancre>
 <22D7FB26-058A-4022-A94C-FFDAD9F97764@amazon.com>
 <CAEepm=38nA8ujyrd6AJt-XXjugcGYY5Tx8fgS1TvfKfR9U8hww@mail.gmail.com>
 <db6995e0-a08c-74bd-7157-d27a4e6b5cf8@2ndQuadrant.com>
Message-ID: <cfbd89e1-85f0-3290-9ec7-0ce80f85b8f3@2ndQuadrant.com>
Date: Thu, 9 Aug 2018 12:45:09 -0400
User-Agent: Mozilla/5.0 (X11; Linux x86_64; rv:52.0) Gecko/20100101
 Thunderbird/52.4.0
MIME-Version: 1.0
In-Reply-To: <db6995e0-a08c-74bd-7157-d27a4e6b5cf8@2ndQuadrant.com>
Content-Type: multipart/mixed;
 boundary="------------3B101BA481C791BFF03D2900"
Content-Language: en-MW
Precedence: bulk

This is a multi-part message in MIME format.
--------------3B101BA481C791BFF03D2900
Content-Type: text/plain; charset=utf-8; format=flowed
Content-Transfer-Encoding: 8bit


On 07/03/2018 07:52 PM, Andrew Dunstan wrote:
>
>
> On 05/17/2018 01:23 AM, Thomas Munro wrote:
>> On Tue, Mar 27, 2018 at 9:23 AM, Rady, Doug <radydoug@amazon.com> wrote:
>>> pgbench11-ppoll-v12.patch
>> Hi Doug,
>>
>> FYI this patch is trying and failing to use ppoll() on Windows:
>>
>> https://ci.appveyor.com/project/postgresql-cfbot/postgresql/build/1.0.30
>>
>
>
> It's still failing -  see 
> <https://ci.appveyor.com/project/postgresql-cfbot/postgresql/build/1.0.4098>
>
> I'm setting this back to "Waiting on Author" until that's fixed.
>


The author hasn't replied, but the attached seems to have cured the 
bitrot so that it at least applies. Let's see what the cfbot makes of it 
and then possibly fix any Windows issues.


cheers

andrew

-- 
Andrew Dunstan                https://www.2ndQuadrant.com
PostgreSQL Development, 24x7 Support, Remote DBA, Training & Services


--------------3B101BA481C791BFF03D2900
Content-Type: text/x-patch;
 name="pgbench-ppoll-v13.patch"
Content-Transfer-Encoding: 7bit
Content-Disposition: attachment;
 filename="pgbench-ppoll-v13.patch"

diff --git a/configure b/configure
index 2665213..4579003 100755
--- a/configure
+++ b/configure
@@ -14916,7 +14916,7 @@ fi
 LIBS_including_readline="$LIBS"
 LIBS=`echo "$LIBS" | sed -e 's/-ledit//g' -e 's/-lreadline//g'`
 
-for ac_func in cbrt clock_gettime dlopen fdatasync getifaddrs getpeerucred getrlimit mbstowcs_l memmove poll posix_fallocate pstat pthread_is_threaded_np readlink setproctitle setproctitle_fast setsid shm_open symlink sync_file_range utime utimes wcstombs_l
+for ac_func in cbrt clock_gettime dlopen fdatasync getifaddrs getpeerucred getrlimit mbstowcs_l memmove poll posix_fallocate pstat ppoll pthread_is_threaded_np readlink setproctitle setproctitle_fast setsid shm_open symlink sync_file_range utime utimes wcstombs_l
 do :
   as_ac_var=`$as_echo "ac_cv_func_$ac_func" | $as_tr_sh`
 ac_fn_c_check_func "$LINENO" "$ac_func" "$as_ac_var"
diff --git a/configure.in b/configure.in
index 397f6bc..bd08068 100644
--- a/configure.in
+++ b/configure.in
@@ -1540,7 +1540,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 setproctitle_fast 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 pstat ppoll pthread_is_threaded_np readlink setproctitle setproctitle_fast 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 41b756c..e23ba1a 100644
--- a/src/bin/pgbench/pgbench.c
+++ b/src/bin/pgbench/pgbench.c
@@ -45,9 +45,18 @@
 #include <signal.h>
 #include <time.h>
 #include <sys/time.h>
+#ifndef PGBENCH_USE_SELECT			/* force use of select(2)? */
+#ifdef HAVE_PPOLL
+#define POLL_USING_PPOLL
+#include <poll.h>
+#endif
+#endif
+#ifndef POLL_USING_PPOLL
 #ifdef HAVE_SYS_SELECT_H
+#define POLL_USING_SELECT
 #include <sys/select.h>
 #endif
+#endif
 
 #ifdef HAVE_SYS_RESOURCE_H
 #include <sys/resource.h>		/* for getrlimit */
@@ -92,13 +101,19 @@ static int	pthread_join(pthread_t th, void **thread_return);
 
 /********************************************************************
  * some configurable parameters */
-
-/* max number of clients allowed */
+#ifdef POLL_USING_SELECT	/* using select(2) */
+#define SOCKET_WAIT_METHOD "select"
+typedef fd_set socket_set;
 #ifdef FD_SETSIZE
-#define MAXCLIENTS	(FD_SETSIZE - 10)
+#define MAXCLIENTS	(FD_SETSIZE - 10) /* system limited max number of clients allowed */
 #else
-#define MAXCLIENTS	1024
+#define MAXCLIENTS	1024		/* max number of clients allowed */
 #endif
+#else	/* using ppoll(2) */
+#define SOCKET_WAIT_METHOD "ppoll"
+typedef struct pollfd socket_set;
+#define MAXCLIENTS	-1		/* unlimited number of clients */
+#endif /* POLL_USING_SELECT */
 
 #define DEFAULT_INIT_STEPS "dtgvp"	/* default -I setting */
 
@@ -525,6 +540,13 @@ static void addScript(ParsedScript script);
 static void *threadRun(void *arg);
 static void setalarm(int seconds);
 static void finishCon(CState *st);
+static socket_set *alloc_socket_set(int count);
+static bool error_on_socket(socket_set *sa, int idx, PGconn *con);
+static void free_socket_set(socket_set *sa);
+static bool ignore_socket(socket_set *sa, int idx, PGconn *con);
+static void clear_socket_set(socket_set *sa, int count);
+static void set_socket(socket_set *sa, int fd, int idx);
+static int wait_on_socket_set(socket_set *sa, int nstate, int maxsock, int64 usec);
 
 
 /* callback functions for our flex lexer */
@@ -1143,6 +1165,7 @@ doConnect(void)
 			!have_password)
 		{
 			PQfinish(conn);
+			conn = NULL;
 			simple_prompt("Password: ", password, sizeof(password), false);
 			have_password = true;
 			new_pass = true;
@@ -4903,7 +4926,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);
@@ -5614,6 +5637,7 @@ threadRun(void *arg)
 	int64		next_report = last_report + (int64) progress * 1000000;
 	StatsData	last,
 				aggs;
+	socket_set	*sockets = alloc_socket_set(nstate);
 
 	/*
 	 * Initialize throttling rate target for all of the thread's clients.  It
@@ -5657,6 +5681,7 @@ threadRun(void *arg)
 		{
 			if ((state[i].con = doConnect()) == NULL)
 				goto done;
+			set_socket(sockets, PQsocket(state[i].con), i);
 		}
 	}
 
@@ -5673,13 +5698,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);
+		clear_socket_set(sockets, nstate);
 		maxsock = -1;
 		min_usec = PG_INT64_MAX;
 		for (i = 0; i < nstate; i++)
@@ -5728,7 +5752,7 @@ threadRun(void *arg)
 					goto done;
 				}
 
-				FD_SET(sock, &input_mask);
+				set_socket(sockets, sock, i);
 				if (maxsock < sock)
 					maxsock = sock;
 			}
@@ -5765,7 +5789,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)
@@ -5776,11 +5800,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_set(sockets, nstate, maxsock, min_usec);
 				}
 				else			/* nothing active, simple sleep */
 				{
@@ -5789,7 +5809,7 @@ threadRun(void *arg)
 			}
 			else				/* no explicit delay, select without timeout */
 			{
-				nsocks = select(maxsock + 1, &input_mask, NULL, NULL, NULL);
+				nsocks = wait_on_socket_set(sockets, nstate, maxsock, 0);
 			}
 
 			if (nsocks < 0)
@@ -5800,7 +5820,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;
 			}
 		}
@@ -5809,7 +5829,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);
+			clear_socket_set(sockets, nstate);
 		}
 
 		/* ok, advance the state machine of each connection */
@@ -5820,16 +5840,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 ||
@@ -5967,6 +5982,8 @@ done:
 		fclose(thread->logfile);
 		thread->logfile = NULL;
 	}
+	free_socket_set(sockets);
+	sockets = NULL;
 	return NULL;
 }
 
@@ -5980,6 +5997,135 @@ finishCon(CState *st)
 	}
 }
 
+#ifdef POLL_USING_SELECT	/* select(2) based socket polling */
+static socket_set *
+alloc_socket_set(int count)
+{
+	return (socket_set *) pg_malloc0(sizeof(socket_set));
+}
+
+static void
+free_socket_set(socket_set *sa)
+{
+	pg_free(sa);
+}
+
+static bool
+error_on_socket(socket_set *sa, int idx, PGconn *con)
+{
+	if (PQsocket(con) >= 0) return false;
+	fprintf(stderr, "invalid socket: %s", PQerrorMessage(con));
+	return true;
+}
+
+static bool
+ignore_socket(socket_set *sa, int idx, PGconn *con)
+{
+	return !(FD_ISSET(PQsocket(con), sa));
+}
+
+static void
+clear_socket_set(socket_set *sa, int count)
+{
+	FD_ZERO(sa);
+}
+
+static void
+set_socket(socket_set *sa, int fd, int idx)
+{
+	FD_SET(fd, sa);
+}
+
+static int
+wait_on_socket_set(socket_set *sa, int nstate, int maxsock, int64 usec)
+{
+	struct timeval timeout;
+
+	if (usec)
+	{
+		timeout.tv_sec = usec / 1000000;
+		timeout.tv_usec = usec % 1000000;
+		return select(maxsock + 1, sa, NULL, NULL, &timeout);
+	}
+	else
+	{
+		return select(maxsock + 1, sa, NULL, NULL, NULL);
+	}
+}
+#else	/* ppoll(2) based socket polling */
+/* ppoll() will block until timeout or one of POLL_EVENTS occurs. */
+#define POLL_EVENTS (POLLRDHUP|POLLIN|POLLPRI)
+/* ppoll() events returned that we do not want/expect to see. */
+#define POLL_UNWANTED (POLLRDHUP|POLLERR|POLLHUP|POLLNVAL)
+
+static socket_set *
+alloc_socket_set(int count)
+{
+	return (socket_set *) pg_malloc0(sizeof(socket_set) * count);
+}
+
+static void
+free_socket_set(socket_set *sa)
+{
+	pg_free(sa);
+}
+
+static bool
+error_on_socket(socket_set *sa, int idx, PGconn *con)
+{
+	/*
+	 * No error if socket not used or non-error status from PQsocket() and none
+	 * of the unwanted ppoll() return events.
+	 */
+	if (sa[idx].fd == -1 || (PQsocket(con) >= 0 && !(sa[idx].revents & POLL_UNWANTED)))
+		return false;
+	fprintf(stderr, "invalid socket: %s", PQerrorMessage(con));
+	if (debug)
+		fprintf(stderr, "ppoll() fail - errno: %d, socket: %d, events: %x\n",
+			errno, sa[idx].fd, (sa[idx].revents & POLL_UNWANTED));
+	return true;
+}
+
+static bool
+ignore_socket(socket_set *sa, int idx, PGconn *con)
+{
+	return (sa[idx].fd != -1 && !sa[idx].revents);
+}
+
+static void
+clear_socket_set(socket_set *sa, int count)
+{
+	int i = 0;
+	for (i = 0; i < count; i++)
+		set_socket(sa, -1, i);
+}
+
+static void
+set_socket(socket_set *sa, int fd, int idx)
+{
+	sa[idx].fd = fd;
+	sa[idx].events = POLL_EVENTS;
+	sa[idx].revents = 0;
+}
+
+static int
+wait_on_socket_set(socket_set *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, &timeout, NULL);
+	}
+	else
+	{
+		return ppoll(sa, nstate, NULL, NULL);
+	}
+}
+#endif	/* PGBENCH_USE_SELECT */
+
 /*
  * 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 b7e4696..5e0e5f3 100644
--- a/src/include/pg_config.h.in
+++ b/src/include/pg_config.h.in
@@ -434,6 +434,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
 

--------------3B101BA481C791BFF03D2900--