sharpetronics/PostgreSQL
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2025-05-30 00:02:11 -04:00
Code Issues Projects Releases Wiki Activity

libpqwalreceiver: Convert to libpq-be-fe-helpers.h

Browse Source 
In contrast to the changes to dblink and postgres_fdw, this does not fix a
bug, as libpqwalreceiver did already process interrupts.

Besides reducing code duplication, the conversion leads to libpqwalreceiver
now using reserving file descriptors for libpq connections. While not strictly
required for the use in walreceiver, we are also using libpqwalreceiver for
logical replication, where it does seem more important.

Even if we eventually decide to backpatch the prior commits, there'd be no
need to backpatch this commit, due to not fixing an active bug.

Reviewed-by: Thomas Munro <thomas.munro@gmail.com>
Discussion: https://postgr.es/m/20220925232237.p6uskba2dw6fnwj2@awork3.anarazel.de
This commit is contained in:
Andres Freund 2023-01-23 19:25:23 -08:00
parent e4602483e9
commit 728f86fec6
1 changed files with 7 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

53
src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
View File

@ -24,6 +24,7 @@
#include "common/connect.h"
#include "funcapi.h"
#include "libpq-fe.h"
#include "libpq/libpq-be-fe-helpers.h"
#include "mb/pg_wchar.h"
#include "miscadmin.h"
#include "pgstat.h"
@ -125,7 +126,6 @@ libpqrcv_connect(const char *conninfo, bool logical, const char *appname,
char **err)
{
WalReceiverConn *conn;
PostgresPollingStatusType status;
const char *keys[6];
const char *vals[6];
int i = 0;
@ -172,49 +172,10 @@ libpqrcv_connect(const char *conninfo, bool logical, const char *appname,
Assert(i < sizeof(keys));
conn = palloc0(sizeof(WalReceiverConn));
conn->streamConn = PQconnectStartParams(keys, vals,
/* expand_dbname = */ true);
if (PQstatus(conn->streamConn) == CONNECTION_BAD)
goto bad_connection_errmsg;
/*
* Poll connection until we have OK or FAILED status.
*
* Per spec for PQconnectPoll, first wait till socket is write-ready.
*/
status = PGRES_POLLING_WRITING;
do
{
int io_flag;
int rc;
if (status == PGRES_POLLING_READING)
io_flag = WL_SOCKET_READABLE;
#ifdef WIN32
/* Windows needs a different test while waiting for connection-made */
else if (PQstatus(conn->streamConn) == CONNECTION_STARTED)
io_flag = WL_SOCKET_CONNECTED;
#endif
else
io_flag = WL_SOCKET_WRITEABLE;
rc = WaitLatchOrSocket(MyLatch,
WL_EXIT_ON_PM_DEATH | WL_LATCH_SET | io_flag,
PQsocket(conn->streamConn),
0,
WAIT_EVENT_LIBPQWALRECEIVER_CONNECT);
/* Interrupted? */
if (rc & WL_LATCH_SET)
{
ResetLatch(MyLatch);
ProcessWalRcvInterrupts();
}
/* If socket is ready, advance the libpq state machine */
if (rc & io_flag)
status = PQconnectPoll(conn->streamConn);
} while (status != PGRES_POLLING_OK && status != PGRES_POLLING_FAILED);
conn->streamConn =
libpqsrv_connect_params(keys, vals,
/* expand_dbname = */ true,
WAIT_EVENT_LIBPQWALRECEIVER_CONNECT);
if (PQstatus(conn->streamConn) != CONNECTION_OK)
goto bad_connection_errmsg;
@ -245,7 +206,7 @@ bad_connection_errmsg:
/* error path, error already set */
bad_connection:
PQfinish(conn->streamConn);
libpqsrv_disconnect(conn->streamConn);
pfree(conn);
return NULL;
}
@ -744,7 +705,7 @@ libpqrcv_PQgetResult(PGconn *streamConn)
static void
libpqrcv_disconnect(WalReceiverConn *conn)
{
PQfinish(conn->streamConn);
libpqsrv_disconnect(conn->streamConn);
PQfreemem(conn->recvBuf);
pfree(conn);
}