diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c index e451c55..5e759b6 100644 --- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c +++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c @@ -54,6 +54,7 @@ static void libpqrcv_disconnect(void); /* Prototypes for private functions */ static bool libpq_select(int timeout_ms); +static PGresult *libpqrcv_PQexec(const char *query); /* * Module load callback @@ -97,10 +98,11 @@ libpqrcv_connect(char *conninfo, XLogRecPtr startpoint) * Get the system identifier and timeline ID as a DataRow message from the * primary server. */ - res = PQexec(streamConn, "IDENTIFY_SYSTEM"); - if (PQresultStatus(res) != PGRES_TUPLES_OK) + res = libpqrcv_PQexec("IDENTIFY_SYSTEM"); + if (res == NULL || PQresultStatus(res) != PGRES_TUPLES_OK) { - PQclear(res); + if (res) + PQclear(res); ereport(ERROR, (errmsg("could not receive database system identifier and timeline ID from " "the primary server: %s", @@ -149,11 +151,15 @@ libpqrcv_connect(char *conninfo, XLogRecPtr startpoint) /* Start streaming from the point requested by startup process */ snprintf(cmd, sizeof(cmd), "START_REPLICATION %X/%X", startpoint.xlogid, startpoint.xrecoff); - res = PQexec(streamConn, cmd); - if (PQresultStatus(res) != PGRES_COPY_OUT) + res = libpqrcv_PQexec(cmd); + if (res == NULL || PQresultStatus(res) != PGRES_COPY_OUT) + { + if (res) + PQclear(res); ereport(ERROR, (errmsg("could not start WAL streaming: %s", PQerrorMessage(streamConn)))); + } PQclear(res); justconnected = true; @@ -225,6 +231,66 @@ libpq_select(int timeout_ms) } /* + * Send a query and wait for the results by using the asynchronous libpq + * functions and the backend version of select(). + * + * We must not use the regular blocking libpq functions like PQexec() + * since they are uninterruptible by signals on some platforms, such as + * Windows. + * + * We must also not use vanilla select() here since it cannot handle the + * signal emulation layer on Windows. + * + * The function is modeled on PQexec() in libpq, but only implements + * those parts that are in use in the walreceiver. + * + * Queries are always executed on the connection in streamConn. + */ +static PGresult * +libpqrcv_PQexec(const char *query) +{ + /* + * PQexec() silently discards any prior query results on the + * connection. This is not required for walreceiver since it's + * expected that walsender won't generate any such junk results. + */ + + /* + * Submit a query. Since we don't use non-blocking mode, this also + * can block. But the risk is relatively small, so we ignore that + * for now. + */ + if (!PQsendQuery(streamConn, query)) + return NULL; + + /* + * Receive data until PQgetResult is ready to get the result + * without blocking. + */ + while (PQisBusy(streamConn)) + { + /* + * We don't need to break down the sleep into smaller increments, + * and check for interrupts after each nap, since we can just + * elog(FATAL) within SIGTERM signal handler if the signal + * arrives in the middle of establishment of replication connection. + */ + if (!libpq_select(-1)) + continue; /* interrupted */ + if (PQconsumeInput(streamConn) == 0) + return NULL; /* trouble */ + } + + /* + * Don't emulate the PQexec()'s behavior of returning the last + * result when there are many, since walreceiver never sends a + * query returning multiple results. Just return the first result + * available on the connection. + */ + return PQgetResult(streamConn); +} + +/* * Disconnect connection to primary, if any. */ static void diff --git a/src/backend/replication/walreceiver.c b/src/backend/replication/walreceiver.c index ff5f293..5ab8f68 100644 --- a/src/backend/replication/walreceiver.c +++ b/src/backend/replication/walreceiver.c @@ -86,8 +86,8 @@ static void DisableWalRcvImmediateExit(void); * We can't just exit(1) within SIGTERM signal handler, because the signal * might arrive in the middle of some critical operation, like while we're * holding a spinlock. We also can't just set a flag in signal handler and - * check it in the main loop, because we perform some blocking libpq - * operations like PQexec(), which can take a long time to finish. + * check it in the main loop, because we perform some blocking operations + * like libpqrcv_PQexec(), which can take a long time to finish. * * We use a combined approach: When WalRcvImmediateInterruptOK is true, it's * safe for the signal handler to elog(FATAL) immediately. Otherwise it just