mirror of
https://git.postgresql.org/git/postgresql.git
synced 2024-09-27 22:41:57 +02:00
318 lines
7.9 KiB
C
318 lines
7.9 KiB
C
|
/*-------------------------------------------------------------------------
|
||
|
*
|
||
|
* libpqwalreceiver.c
|
||
|
*
|
||
|
* This file contains the libpq-specific parts of walreceiver. It's
|
||
|
* loaded as a dynamic module to avoid linking the main server binary with
|
||
|
* libpq.
|
||
|
*
|
||
|
* Portions Copyright (c) 2010-2010, PostgreSQL Global Development Group
|
||
|
*
|
||
|
*
|
||
|
* IDENTIFICATION
|
||
|
* $PostgreSQL: pgsql/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c,v 1.1 2010/01/20 09:16:24 heikki Exp $
|
||
|
*
|
||
|
*-------------------------------------------------------------------------
|
||
|
*/
|
||
|
#include "postgres.h"
|
||
|
|
||
|
#include <unistd.h>
|
||
|
#include <sys/time.h>
|
||
|
|
||
|
#include "libpq-fe.h"
|
||
|
#include "access/xlog.h"
|
||
|
#include "miscadmin.h"
|
||
|
#include "replication/walreceiver.h"
|
||
|
#include "utils/builtins.h"
|
||
|
|
||
|
#ifdef HAVE_POLL_H
|
||
|
#include <poll.h>
|
||
|
#endif
|
||
|
#ifdef HAVE_SYS_POLL_H
|
||
|
#include <sys/poll.h>
|
||
|
#endif
|
||
|
#ifdef HAVE_SYS_SELECT_H
|
||
|
#include <sys/select.h>
|
||
|
#endif
|
||
|
|
||
|
PG_MODULE_MAGIC;
|
||
|
|
||
|
void _PG_init(void);
|
||
|
|
||
|
/* Current connection to the primary, if any */
|
||
|
static PGconn *streamConn = NULL;
|
||
|
static bool justconnected = false;
|
||
|
|
||
|
/* Buffer for currently read records */
|
||
|
static char *recvBuf = NULL;
|
||
|
|
||
|
/* Prototypes for interface functions */
|
||
|
static bool libpqrcv_connect(char *conninfo, XLogRecPtr startpoint);
|
||
|
static bool libpqrcv_receive(int timeout, XLogRecPtr *recptr, char **buffer,
|
||
|
int *len);
|
||
|
static void libpqrcv_disconnect(void);
|
||
|
|
||
|
/* Prototypes for private functions */
|
||
|
static bool libpq_select(int timeout_ms);
|
||
|
|
||
|
/*
|
||
|
* Module load callback
|
||
|
*/
|
||
|
void
|
||
|
_PG_init(void)
|
||
|
{
|
||
|
/* Tell walreceiver how to reach us */
|
||
|
if (walrcv_connect != NULL || walrcv_receive != NULL || walrcv_disconnect)
|
||
|
elog(ERROR, "libpqwalreceiver already loaded");
|
||
|
walrcv_connect = libpqrcv_connect;
|
||
|
walrcv_receive = libpqrcv_receive;
|
||
|
walrcv_disconnect = libpqrcv_disconnect;
|
||
|
}
|
||
|
|
||
|
/*
|
||
|
* Establish the connection to the primary server for XLOG streaming
|
||
|
*/
|
||
|
static bool
|
||
|
libpqrcv_connect(char *conninfo, XLogRecPtr startpoint)
|
||
|
{
|
||
|
char conninfo_repl[MAXCONNINFO + 14];
|
||
|
char *primary_sysid;
|
||
|
char standby_sysid[32];
|
||
|
TimeLineID primary_tli;
|
||
|
TimeLineID standby_tli;
|
||
|
PGresult *res;
|
||
|
char cmd[64];
|
||
|
|
||
|
Assert(startpoint.xlogid != 0 || startpoint.xrecoff != 0);
|
||
|
|
||
|
/* Connect */
|
||
|
snprintf(conninfo_repl, sizeof(conninfo_repl), "%s replication=true", conninfo);
|
||
|
|
||
|
streamConn = PQconnectdb(conninfo_repl);
|
||
|
if (PQstatus(streamConn) != CONNECTION_OK)
|
||
|
ereport(ERROR,
|
||
|
(errmsg("could not connect to the primary server : %s",
|
||
|
PQerrorMessage(streamConn))));
|
||
|
|
||
|
/*
|
||
|
* 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);
|
||
|
ereport(ERROR,
|
||
|
(errmsg("could not receive the SYSID and timeline ID from "
|
||
|
"the primary server: %s",
|
||
|
PQerrorMessage(streamConn))));
|
||
|
}
|
||
|
if (PQnfields(res) != 2 || PQntuples(res) != 1)
|
||
|
{
|
||
|
int ntuples = PQntuples(res);
|
||
|
int nfields = PQnfields(res);
|
||
|
PQclear(res);
|
||
|
ereport(ERROR,
|
||
|
(errmsg("invalid response from primary server"),
|
||
|
errdetail("expected 1 tuple with 2 fields, got %d tuples with %d fields",
|
||
|
ntuples, nfields)));
|
||
|
}
|
||
|
primary_sysid = PQgetvalue(res, 0, 0);
|
||
|
primary_tli = pg_atoi(PQgetvalue(res, 0, 1), 4, 0);
|
||
|
|
||
|
/*
|
||
|
* Confirm that the system identifier of the primary is the same
|
||
|
* as ours.
|
||
|
*/
|
||
|
snprintf(standby_sysid, sizeof(standby_sysid), UINT64_FORMAT,
|
||
|
GetSystemIdentifier());
|
||
|
if (strcmp(primary_sysid, standby_sysid) != 0)
|
||
|
{
|
||
|
PQclear(res);
|
||
|
ereport(ERROR,
|
||
|
(errmsg("system differs between the primary and standby"),
|
||
|
errdetail("the primary SYSID is %s, standby SYSID is %s",
|
||
|
primary_sysid, standby_sysid)));
|
||
|
}
|
||
|
|
||
|
/*
|
||
|
* Confirm that the current timeline of the primary is the same
|
||
|
* as the recovery target timeline.
|
||
|
*/
|
||
|
standby_tli = GetRecoveryTargetTLI();
|
||
|
PQclear(res);
|
||
|
if (primary_tli != standby_tli)
|
||
|
ereport(ERROR,
|
||
|
(errmsg("timeline %u of the primary does not match recovery target timeline %u",
|
||
|
primary_tli, standby_tli)));
|
||
|
ThisTimeLineID = primary_tli;
|
||
|
|
||
|
/* 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 XLOG streaming: %s",
|
||
|
PQerrorMessage(streamConn))));
|
||
|
PQclear(res);
|
||
|
|
||
|
justconnected = true;
|
||
|
|
||
|
return true;
|
||
|
}
|
||
|
|
||
|
/*
|
||
|
* Wait until we can read WAL stream, or timeout.
|
||
|
*
|
||
|
* Returns true if data has become available for reading, false if timed out
|
||
|
* or interrupted by signal.
|
||
|
*
|
||
|
* This is based on pqSocketCheck.
|
||
|
*/
|
||
|
static bool
|
||
|
libpq_select(int timeout_ms)
|
||
|
{
|
||
|
int ret;
|
||
|
|
||
|
Assert(streamConn != NULL);
|
||
|
if (PQsocket(streamConn) < 0)
|
||
|
ereport(ERROR,
|
||
|
(errcode_for_socket_access(),
|
||
|
errmsg("socket not open")));
|
||
|
|
||
|
/* We use poll(2) if available, otherwise select(2) */
|
||
|
{
|
||
|
#ifdef HAVE_POLL
|
||
|
struct pollfd input_fd;
|
||
|
|
||
|
input_fd.fd = PQsocket(streamConn);
|
||
|
input_fd.events = POLLIN | POLLERR;
|
||
|
input_fd.revents = 0;
|
||
|
|
||
|
ret = poll(&input_fd, 1, timeout_ms);
|
||
|
#else /* !HAVE_POLL */
|
||
|
|
||
|
fd_set input_mask;
|
||
|
struct timeval timeout;
|
||
|
struct timeval *ptr_timeout;
|
||
|
|
||
|
FD_ZERO(&input_mask);
|
||
|
FD_SET(PQsocket(streamConn), &input_mask);
|
||
|
|
||
|
if (timeout_ms < 0)
|
||
|
ptr_timeout = NULL;
|
||
|
else
|
||
|
{
|
||
|
timeout.tv_sec = timeout_ms / 1000;
|
||
|
timeout.tv_usec = (timeout_ms % 1000) * 1000;
|
||
|
ptr_timeout = &timeout;
|
||
|
}
|
||
|
|
||
|
ret = select(PQsocket(streamConn) + 1, &input_mask,
|
||
|
NULL, NULL, ptr_timeout);
|
||
|
#endif /* HAVE_POLL */
|
||
|
}
|
||
|
|
||
|
if (ret == 0 || (ret < 0 && errno == EINTR))
|
||
|
return false;
|
||
|
if (ret < 0)
|
||
|
ereport(ERROR,
|
||
|
(errcode_for_socket_access(),
|
||
|
errmsg("select() failed: %m")));
|
||
|
return true;
|
||
|
}
|
||
|
|
||
|
/*
|
||
|
* Disconnect connection to primary, if any.
|
||
|
*/
|
||
|
static void
|
||
|
libpqrcv_disconnect(void)
|
||
|
{
|
||
|
PQfinish(streamConn);
|
||
|
streamConn = NULL;
|
||
|
justconnected = false;
|
||
|
}
|
||
|
|
||
|
/*
|
||
|
* Receive any WAL records available from XLOG stream, blocking for
|
||
|
* maximum of 'timeout' ms.
|
||
|
*
|
||
|
* Returns:
|
||
|
*
|
||
|
* True if data was received. *recptr, *buffer and *len are set to
|
||
|
* the WAL location of the received data, buffer holding it, and length,
|
||
|
* respectively.
|
||
|
*
|
||
|
* False if no data was available within timeout, or wait was interrupted
|
||
|
* by signal.
|
||
|
*
|
||
|
* The buffer returned is only valid until the next call of this function or
|
||
|
* libpq_connect/disconnect.
|
||
|
*
|
||
|
* ereports on error.
|
||
|
*/
|
||
|
static bool
|
||
|
libpqrcv_receive(int timeout, XLogRecPtr *recptr, char **buffer, int *len)
|
||
|
{
|
||
|
int rawlen;
|
||
|
|
||
|
if (recvBuf != NULL)
|
||
|
PQfreemem(recvBuf);
|
||
|
recvBuf = NULL;
|
||
|
|
||
|
/*
|
||
|
* If the caller requested to block, wait for data to arrive. But if
|
||
|
* this is the first call after connecting, don't wait, because
|
||
|
* there might already be some data in libpq buffer that we haven't
|
||
|
* returned to caller.
|
||
|
*/
|
||
|
if (timeout > 0 && !justconnected)
|
||
|
{
|
||
|
if (!libpq_select(timeout))
|
||
|
return false;
|
||
|
|
||
|
if (PQconsumeInput(streamConn) == 0)
|
||
|
ereport(ERROR,
|
||
|
(errmsg("could not read xlog records: %s",
|
||
|
PQerrorMessage(streamConn))));
|
||
|
}
|
||
|
justconnected = false;
|
||
|
|
||
|
/* Receive CopyData message */
|
||
|
rawlen = PQgetCopyData(streamConn, &recvBuf, 1);
|
||
|
if (rawlen == 0) /* no records available yet, then return */
|
||
|
return false;
|
||
|
if (rawlen == -1) /* end-of-streaming or error */
|
||
|
{
|
||
|
PGresult *res;
|
||
|
|
||
|
res = PQgetResult(streamConn);
|
||
|
if (PQresultStatus(res) == PGRES_COMMAND_OK)
|
||
|
{
|
||
|
PQclear(res);
|
||
|
ereport(ERROR,
|
||
|
(errmsg("replication terminated by primary server")));
|
||
|
}
|
||
|
PQclear(res);
|
||
|
ereport(ERROR,
|
||
|
(errmsg("could not read xlog records: %s",
|
||
|
PQerrorMessage(streamConn))));
|
||
|
}
|
||
|
if (rawlen < -1)
|
||
|
ereport(ERROR,
|
||
|
(errmsg("could not read xlog records: %s",
|
||
|
PQerrorMessage(streamConn))));
|
||
|
|
||
|
if (rawlen < sizeof(XLogRecPtr))
|
||
|
ereport(ERROR,
|
||
|
(errmsg("invalid WAL message received from primary")));
|
||
|
|
||
|
/* Return received WAL records to caller */
|
||
|
*recptr = *((XLogRecPtr *) recvBuf);
|
||
|
*buffer = recvBuf + sizeof(XLogRecPtr);
|
||
|
*len = rawlen - sizeof(XLogRecPtr);
|
||
|
|
||
|
return true;
|
||
|
}
|