diff --git a/configure b/configure index c6a44a9078..0456064456 100755 --- a/configure +++ b/configure @@ -15060,7 +15060,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 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 fdatasync getifaddrs getpeerucred getrlimit mbstowcs_l memmove poll posix_fallocate ppoll pstat 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 3ada48b5f9..fceba79023 100644 --- a/configure.in +++ b/configure.in @@ -1544,7 +1544,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 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 fdatasync getifaddrs getpeerucred getrlimit mbstowcs_l memmove poll posix_fallocate ppoll pstat 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 41b756c089..3d378db714 100644 --- a/src/bin/pgbench/pgbench.c +++ b/src/bin/pgbench/pgbench.c @@ -45,9 +45,18 @@ #include #include #include +#ifndef PGBENCH_USE_SELECT /* force use of select(2)? */ +#ifdef HAVE_PPOLL +#define POLL_USING_PPOLL +#include +#endif +#endif +#ifndef POLL_USING_PPOLL +#define POLL_USING_SELECT #ifdef HAVE_SYS_SELECT_H #include #endif +#endif #ifdef HAVE_SYS_RESOURCE_H #include /* 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 4094e22776..5d4079609c 100644 --- a/src/include/pg_config.h.in +++ b/src/include/pg_config.h.in @@ -443,6 +443,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