Skip to content
Snippets Groups Projects
Commit a5a02a74 authored by Heikki Linnakangas's avatar Heikki Linnakangas
Browse files

Fix the logic in libpqrcv_receive() to determine if there's any incoming data

that can be read without blocking. It used to conclude that there isn't, even
though there was data in the socket receive buffer. That lead walreceiver to
flush the WAL after every received chunk, potentially causing big performance
issues.

Backpatch to 9.0, because the performance impact can be very significant.
parent c667cc24
No related branches found
No related tags found
No related merge requests found
...@@ -41,7 +41,6 @@ void _PG_init(void); ...@@ -41,7 +41,6 @@ void _PG_init(void);
/* Current connection to the primary, if any */ /* Current connection to the primary, if any */
static PGconn *streamConn = NULL; static PGconn *streamConn = NULL;
static bool justconnected = false;
/* Buffer for currently read records */ /* Buffer for currently read records */
static char *recvBuf = NULL; static char *recvBuf = NULL;
...@@ -168,7 +167,6 @@ libpqrcv_connect(char *conninfo, XLogRecPtr startpoint) ...@@ -168,7 +167,6 @@ libpqrcv_connect(char *conninfo, XLogRecPtr startpoint)
} }
PQclear(res); PQclear(res);
justconnected = true;
ereport(LOG, ereport(LOG,
(errmsg("streaming replication successfully connected to primary"))); (errmsg("streaming replication successfully connected to primary")));
...@@ -321,7 +319,6 @@ libpqrcv_disconnect(void) ...@@ -321,7 +319,6 @@ libpqrcv_disconnect(void)
{ {
PQfinish(streamConn); PQfinish(streamConn);
streamConn = NULL; streamConn = NULL;
justconnected = false;
} }
/* /*
...@@ -351,28 +348,30 @@ libpqrcv_receive(int timeout, unsigned char *type, char **buffer, int *len) ...@@ -351,28 +348,30 @@ libpqrcv_receive(int timeout, unsigned char *type, char **buffer, int *len)
PQfreemem(recvBuf); PQfreemem(recvBuf);
recvBuf = NULL; recvBuf = NULL;
/* Try to receive a CopyData message */
rawlen = PQgetCopyData(streamConn, &recvBuf, 1);
if (rawlen == 0)
{
/* /*
* If the caller requested to block, wait for data to arrive. But if this * No data available yet. If the caller requested to block, wait for
* is the first call after connecting, don't wait, because there might * more data to arrive.
* already be some data in libpq buffer that we haven't returned to
* caller.
*/ */
if (timeout > 0 && !justconnected) if (timeout > 0)
{ {
if (!libpq_select(timeout)) if (!libpq_select(timeout))
return false; return false;
}
if (PQconsumeInput(streamConn) == 0) if (PQconsumeInput(streamConn) == 0)
ereport(ERROR, ereport(ERROR,
(errmsg("could not receive data from WAL stream: %s", (errmsg("could not receive data from WAL stream: %s",
PQerrorMessage(streamConn)))); PQerrorMessage(streamConn))));
}
justconnected = false;
/* Receive CopyData message */ /* Now that we've consumed some input, try again */
rawlen = PQgetCopyData(streamConn, &recvBuf, 1); rawlen = PQgetCopyData(streamConn, &recvBuf, 1);
if (rawlen == 0) /* no data available yet, then return */ if (rawlen == 0)
return false; return false;
}
if (rawlen == -1) /* end-of-streaming or error */ if (rawlen == -1) /* end-of-streaming or error */
{ {
PGresult *res; PGresult *res;
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment