Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 4 additions & 3 deletions src/backend/libpq/pqcomm.c
Original file line number Diff line number Diff line change
Expand Up @@ -245,7 +245,7 @@ pq_configure(Port *port)
{
int compression_level = ZPQ_DEFAULT_COMPRESSION_LEVEL;
int impl = -1;
char **server_compression_algorithms = zpq_get_supported_algorithms();
char **server_compression_algorithms = zs_get_supported_algorithms();
int index = -1;
char *protocol_extension = strchr(client_compression_algorithms, ';');

Expand Down Expand Up @@ -289,7 +289,8 @@ pq_configure(Port *port)

if (index >= 0) /* Use compression */
{
PqStream = zpq_create(impl, compression_level, impl, write_compressed, read_compressed, MyProcPort, NULL, 0);
PqStream = zpq_create(impl, compression_level, impl, write_compressed, read_compressed, MyProcPort,
NULL, 0);
if (!PqStream)
{
ereport(LOG,
Expand Down Expand Up @@ -1080,7 +1081,7 @@ pq_recvbuf(bool nowait)

if (r < 0)
{
if (r == ZPQ_DECOMPRESS_ERROR)
if (r == ZS_DECOMPRESS_ERROR)
{
char const *msg = zpq_decompress_error(PqStream);

Expand Down
17 changes: 13 additions & 4 deletions src/bin/pgbench/pgbench.c
Original file line number Diff line number Diff line change
Expand Up @@ -6299,6 +6299,9 @@ threadRun(void *arg)
int nsocks; /* number of sockets to be waited for */
int64 min_usec;
int64 now_usec = 0; /* set this only if needed */
bool buffered_rx = false; /* true if some of the clients has
* data left in SSL/ZPQ read
* buffers */

/*
* identify which client sockets should be checked for input, and
Expand Down Expand Up @@ -6339,6 +6342,9 @@ threadRun(void *arg)
*/
int sock = PQsocket(st->con);

/* check if conn has buffered SSL / ZPQ read data */
buffered_rx = buffered_rx || PQreadPending(st->con);

if (sock < 0)
{
pg_log_error("invalid socket: %s", PQerrorMessage(st->con));
Expand Down Expand Up @@ -6389,7 +6395,7 @@ threadRun(void *arg)
{
if (nsocks > 0)
{
rc = wait_on_socket_set(sockets, min_usec);
rc = buffered_rx ? 1 : wait_on_socket_set(sockets, min_usec);
}
else /* nothing active, simple sleep */
{
Expand All @@ -6398,7 +6404,7 @@ threadRun(void *arg)
}
else /* no explicit delay, wait without timeout */
{
rc = wait_on_socket_set(sockets, 0);
rc = buffered_rx ? 1 : wait_on_socket_set(sockets, 0);
}

if (rc < 0)
Expand Down Expand Up @@ -6437,8 +6443,11 @@ threadRun(void *arg)
pg_log_error("invalid socket: %s", PQerrorMessage(st->con));
goto done;
}

if (!socket_has_input(sockets, sock, nsocks++))
if (PQreadPending(st->con))
{
nsocks++;
}
else if (!socket_has_input(sockets, sock, nsocks++))
continue;
}
else if (st->state == CSTATE_FINISHED ||
Expand Down
1 change: 1 addition & 0 deletions src/common/Makefile
Original file line number Diff line number Diff line change
Expand Up @@ -78,6 +78,7 @@ OBJS_COMMON = \
username.o \
wait_error.o \
wchar.o \
z_stream.o \
zpq_stream.o

ifeq ($(with_openssl),yes)
Expand Down
Loading