*** a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c --- b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c *************** *** 54,59 **** static void libpqrcv_disconnect(void); --- 54,60 ---- /* Prototypes for private functions */ static bool libpq_select(int timeout_ms); + static PGresult *libpqrcv_PQexec(const char *query); /* * Module load callback *************** *** 97,103 **** 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) { PQclear(res); --- 98,104 ---- * Get the system identifier and timeline ID as a DataRow message from the * primary server. */ ! res = libpqrcv_PQexec("IDENTIFY_SYSTEM"); if (PQresultStatus(res) != PGRES_TUPLES_OK) { PQclear(res); *************** *** 149,155 **** 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) ereport(ERROR, (errmsg("could not start WAL streaming: %s", --- 150,156 ---- /* Start streaming from the point requested by startup process */ snprintf(cmd, sizeof(cmd), "START_REPLICATION %X/%X", startpoint.xlogid, startpoint.xrecoff); ! res = libpqrcv_PQexec(cmd); if (PQresultStatus(res) != PGRES_COPY_OUT) ereport(ERROR, (errmsg("could not start WAL streaming: %s", *************** *** 225,230 **** libpq_select(int timeout_ms) --- 226,302 ---- } /* + * 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 blocking libpq functions like PQexec() for that + * purpose because they are uninterruptible by signals on some platforms. + * Similarly, we must not use the vanilla select() here because it cannot + * handle the signals emulated for Windows. The signal emulation layer + * compatible select() must be called instead. + */ + static PGresult * + libpqrcv_PQexec(const char *query) + { + PGresult *res = NULL; + + /* + * PQexec() silently discards any prior query results at first. + * But this preparation is not required for walreceiver because + * it's expected that walsender doesn't generate such junk results. + */ + + /* + * Submit a query. Since we don't use non-blocking mode, this also + * can block. But its risk is relatively small, so we ignore that + * for now. + */ + if (!PQsendQuery(streamConn, query)) + return NULL; + + for (;;) + { + PGresult *next; + + /* + * Receive data until PQgetResult has been 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. Because we can just + * elog(FATAL) within SIGTERM signal handler when 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. + */ + if ((next = PQgetResult(streamConn)) == NULL) + break; /* query is complete */ + if (PQresultStatus(next) == PGRES_FATAL_ERROR) + return next; + PQclear(res); + res = next; + if (PQresultStatus(res) == PGRES_COPY_IN || + PQresultStatus(res) == PGRES_COPY_OUT || + PQstatus(streamConn) == CONNECTION_BAD) + break; + } + + return res; + } + + /* * Disconnect connection to primary, if any. */ static void *** a/src/backend/replication/walreceiver.c --- b/src/backend/replication/walreceiver.c *************** *** 86,93 **** 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. * * We use a combined approach: When WalRcvImmediateInterruptOK is true, it's * safe for the signal handler to elog(FATAL) immediately. Otherwise it just --- 86,93 ---- * 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 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