database/postgres
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2025-07-02 09:02:37 +03:00
Code Activity

Make the streaming replication protocol messages architecture-independent.

Browse Source 
We used to send structs wrapped in CopyData messages, which works as long as
the client and server agree on things like endianess, timestamp format and
alignment. That's good enough for running a standby server, which has to run
on the same platform anyway, but it's useful for tools like pg_receivexlog
to work across platforms.

This breaks protocol compatibility of streaming replication, but we never
promised that to be compatible across versions, anyway.
This commit is contained in:
Heikki Linnakangas
2012-11-07 18:59:12 +02:00
parent ed5699dd1b
commit add6c3179a
7 changed files with 384 additions and 367 deletions
Download Patch File Download Diff File Expand all files Collapse all files

135
src/backend/replication/walreceiver.c
View File

@ -39,9 +39,9 @@
#include <unistd.h>
#include "access/xlog_internal.h"
#include "libpq/pqformat.h"
#include "libpq/pqsignal.h"
#include "miscadmin.h"
#include "replication/walprotocol.h"
#include "replication/walreceiver.h"
#include "replication/walsender.h"
#include "storage/ipc.h"
@ -93,8 +93,8 @@ static struct
XLogRecPtr Flush; /* last byte + 1 flushed in the standby */
} LogstreamResult;
static StandbyReplyMessage reply_message;
static StandbyHSFeedbackMessage feedback_message;
static StringInfoData reply_message;
static StringInfoData incoming_message;
/*
* About SIGTERM handling:
@ -279,10 +279,10 @@ WalReceiverMain(void)
walrcv_connect(conninfo, startpoint);
DisableWalRcvImmediateExit();
/* Initialize LogstreamResult, reply_message and feedback_message */
/* Initialize LogstreamResult and buffers for processing messages */
LogstreamResult.Write = LogstreamResult.Flush = GetXLogReplayRecPtr(NULL);
MemSet(&reply_message, 0, sizeof(reply_message));
MemSet(&feedback_message, 0, sizeof(feedback_message));
initStringInfo(&reply_message);
initStringInfo(&incoming_message);
/* Initialize the last recv timestamp */
last_recv_timestamp = GetCurrentTimestamp();
@ -480,41 +480,58 @@ WalRcvQuickDieHandler(SIGNAL_ARGS)
static void
XLogWalRcvProcessMsg(unsigned char type, char *buf, Size len)
{
int hdrlen;
XLogRecPtr dataStart;
XLogRecPtr walEnd;
TimestampTz sendTime;
bool replyRequested;
resetStringInfo(&incoming_message);
switch (type)
{
case 'w': /* WAL records */
{
WalDataMessageHeader msghdr;
if (len < sizeof(WalDataMessageHeader))
/* copy message to StringInfo */
hdrlen = sizeof(int64) + sizeof(int64) + sizeof(int64);
if (len < hdrlen)
ereport(ERROR,
(errcode(ERRCODE_PROTOCOL_VIOLATION),
errmsg_internal("invalid WAL message received from primary")));
/* memcpy is required here for alignment reasons */
memcpy(&msghdr, buf, sizeof(WalDataMessageHeader));
appendBinaryStringInfo(&incoming_message, buf, hdrlen);
ProcessWalSndrMessage(msghdr.walEnd, msghdr.sendTime);
/* read the fields */
dataStart = pq_getmsgint64(&incoming_message);
walEnd = pq_getmsgint64(&incoming_message);
sendTime = IntegerTimestampToTimestampTz(
pq_getmsgint64(&incoming_message));
ProcessWalSndrMessage(walEnd, sendTime);
buf += sizeof(WalDataMessageHeader);
len -= sizeof(WalDataMessageHeader);
XLogWalRcvWrite(buf, len, msghdr.dataStart);
buf += hdrlen;
len -= hdrlen;
XLogWalRcvWrite(buf, len, dataStart);
break;
}
case 'k': /* Keepalive */
{
PrimaryKeepaliveMessage keepalive;
if (len != sizeof(PrimaryKeepaliveMessage))
/* copy message to StringInfo */
hdrlen = sizeof(int64) + sizeof(int64) + sizeof(char);
if (len != hdrlen)
ereport(ERROR,
(errcode(ERRCODE_PROTOCOL_VIOLATION),
errmsg_internal("invalid keepalive message received from primary")));
/* memcpy is required here for alignment reasons */
memcpy(&keepalive, buf, sizeof(PrimaryKeepaliveMessage));
appendBinaryStringInfo(&incoming_message, buf, hdrlen);
ProcessWalSndrMessage(keepalive.walEnd, keepalive.sendTime);
/* read the fields */
walEnd = pq_getmsgint64(&incoming_message);
sendTime = IntegerTimestampToTimestampTz(
pq_getmsgint64(&incoming_message));
replyRequested = pq_getmsgbyte(&incoming_message);
ProcessWalSndrMessage(walEnd, sendTime);
/* If the primary requested a reply, send one immediately */
if (keepalive.replyRequested)
if (replyRequested)
XLogWalRcvSendReply(true, false);
break;
}
@ -685,7 +702,10 @@ XLogWalRcvFlush(bool dying)
static void
XLogWalRcvSendReply(bool force, bool requestReply)
{
char buf[sizeof(StandbyReplyMessage) + 1];
static XLogRecPtr writePtr = 0;
static XLogRecPtr flushPtr = 0;
XLogRecPtr applyPtr;
static TimestampTz sendTime = 0;
TimestampTz now;
/*
@ -708,28 +728,34 @@ XLogWalRcvSendReply(bool force, bool requestReply)
* probably OK.
*/
if (!force
&& XLByteEQ(reply_message.write, LogstreamResult.Write)
&& XLByteEQ(reply_message.flush, LogstreamResult.Flush)
&& !TimestampDifferenceExceeds(reply_message.sendTime, now,
&& XLByteEQ(writePtr, LogstreamResult.Write)
&& XLByteEQ(flushPtr, LogstreamResult.Flush)
&& !TimestampDifferenceExceeds(sendTime, now,
wal_receiver_status_interval * 1000))
return;
sendTime = now;
/* Construct a new message */
reply_message.write = LogstreamResult.Write;
reply_message.flush = LogstreamResult.Flush;
reply_message.apply = GetXLogReplayRecPtr(NULL);
reply_message.sendTime = now;
reply_message.replyRequested = requestReply;
writePtr = LogstreamResult.Write;
flushPtr = LogstreamResult.Flush;
applyPtr = GetXLogReplayRecPtr(NULL);
elog(DEBUG2, "sending write %X/%X flush %X/%X apply %X/%X",
(uint32) (reply_message.write >> 32), (uint32) reply_message.write,
(uint32) (reply_message.flush >> 32), (uint32) reply_message.flush,
(uint32) (reply_message.apply >> 32), (uint32) reply_message.apply);
resetStringInfo(&reply_message);
pq_sendbyte(&reply_message, 'r');
pq_sendint64(&reply_message, writePtr);
pq_sendint64(&reply_message, flushPtr);
pq_sendint64(&reply_message, applyPtr);
pq_sendint64(&reply_message, GetCurrentIntegerTimestamp());
pq_sendbyte(&reply_message, requestReply ? 1 : 0);
/* Prepend with the message type and send it. */
buf[0] = 'r';
memcpy(&buf[1], &reply_message, sizeof(StandbyReplyMessage));
walrcv_send(buf, sizeof(StandbyReplyMessage) + 1);
/* Send it */
elog(DEBUG2, "sending write %X/%X flush %X/%X apply %X/%X%s",
(uint32) (writePtr >> 32), (uint32) writePtr,
(uint32) (flushPtr >> 32), (uint32) flushPtr,
(uint32) (applyPtr >> 32), (uint32) applyPtr,
requestReply ? " (reply requested)" : "");
walrcv_send(reply_message.data, reply_message.len);
}
/*
@ -739,11 +765,11 @@ XLogWalRcvSendReply(bool force, bool requestReply)
static void
XLogWalRcvSendHSFeedback(void)
{
char buf[sizeof(StandbyHSFeedbackMessage) + 1];
TimestampTz now;
TransactionId nextXid;
uint32 nextEpoch;
TransactionId xmin;
static TimestampTz sendTime = 0;
/*
* If the user doesn't want status to be reported to the master, be sure
@ -758,9 +784,10 @@ XLogWalRcvSendHSFeedback(void)
/*
* Send feedback at most once per wal_receiver_status_interval.
*/
if (!TimestampDifferenceExceeds(feedback_message.sendTime, now,
if (!TimestampDifferenceExceeds(sendTime, now,
wal_receiver_status_interval * 1000))
return;
sendTime = now;
/*
* If Hot Standby is not yet active there is nothing to send. Check this
@ -783,25 +810,23 @@ XLogWalRcvSendHSFeedback(void)
if (nextXid < xmin)
nextEpoch--;
/*
* Always send feedback message.
*/
feedback_message.sendTime = now;
feedback_message.xmin = xmin;
feedback_message.epoch = nextEpoch;
elog(DEBUG2, "sending hot standby feedback xmin %u epoch %u",
feedback_message.xmin,
feedback_message.epoch);
xmin, nextEpoch);
/* Prepend with the message type and send it. */
buf[0] = 'h';
memcpy(&buf[1], &feedback_message, sizeof(StandbyHSFeedbackMessage));
walrcv_send(buf, sizeof(StandbyHSFeedbackMessage) + 1);
/* Construct the the message and send it. */
resetStringInfo(&reply_message);
pq_sendbyte(&reply_message, 'h');
pq_sendint64(&reply_message, GetCurrentIntegerTimestamp());
pq_sendint(&reply_message, xmin, 4);
pq_sendint(&reply_message, nextEpoch, 4);
walrcv_send(reply_message.data, reply_message.len);
}
/*
* Keep track of important messages from primary.
* Update shared memory status upon receiving a message from primary.
*
* 'walEnd' and 'sendTime' are the end-of-WAL and timestamp of the latest
* message, reported by primary.
*/
static void
ProcessWalSndrMessage(XLogRecPtr walEnd, TimestampTz sendTime)

142
src/backend/replication/walsender.c
View File

@ -48,7 +48,6 @@
#include "nodes/replnodes.h"
#include "replication/basebackup.h"
#include "replication/syncrep.h"
#include "replication/walprotocol.h"
#include "replication/walreceiver.h"
#include "replication/walsender.h"
#include "replication/walsender_private.h"
@ -66,6 +65,16 @@
#include "utils/timeout.h"
#include "utils/timestamp.h"
/*
* Maximum data payload in a WAL data message. Must be >= XLOG_BLCKSZ.
*
* We don't have a good idea of what a good value would be; there's some
* overhead per message in both walsender and walreceiver, but on the other
* hand sending large batches makes walsender less responsive to signals
* because signals are checked only between messages. 128kB (with
* default 8k blocks) seems like a reasonable guess for now.
*/
#define MAX_SEND_SIZE (XLOG_BLCKSZ * 16)
/* Array of WalSnds in shared memory */
WalSndCtlData *WalSndCtl = NULL;
@ -103,13 +112,10 @@ static uint32 sendOff = 0;
*/
static XLogRecPtr sentPtr = 0;
/* Buffer for processing reply messages. */
/* Buffers for constructing outgoing messages and processing reply messages. */
static StringInfoData output_message;
static StringInfoData reply_message;
/*
* Buffer for constructing outgoing messages.
* (1 + sizeof(WalDataMessageHeader) + MAX_SEND_SIZE bytes)
*/
static char *output_message;
static StringInfoData tmpbuf;
/*
* Timestamp of the last receipt of the reply from the standby.
@ -526,17 +532,26 @@ ProcessStandbyMessage(void)
static void
ProcessStandbyReplyMessage(void)
{
StandbyReplyMessage reply;
XLogRecPtr writePtr,
flushPtr,
applyPtr;
bool replyRequested;
pq_copymsgbytes(&reply_message, (char *) &reply, sizeof(StandbyReplyMessage));
/* the caller already consumed the msgtype byte */
writePtr = pq_getmsgint64(&reply_message);
flushPtr = pq_getmsgint64(&reply_message);
applyPtr = pq_getmsgint64(&reply_message);
(void) pq_getmsgint64(&reply_message); /* sendTime; not used ATM */
replyRequested = pq_getmsgbyte(&reply_message);
elog(DEBUG2, "write %X/%X flush %X/%X apply %X/%X",
(uint32) (reply.write >> 32), (uint32) reply.write,
(uint32) (reply.flush >> 32), (uint32) reply.flush,
(uint32) (reply.apply >> 32), (uint32) reply.apply);
elog(DEBUG2, "write %X/%X flush %X/%X apply %X/%X%s",
(uint32) (writePtr >> 32), (uint32) writePtr,
(uint32) (flushPtr >> 32), (uint32) flushPtr,
(uint32) (applyPtr >> 32), (uint32) applyPtr,
replyRequested ? " (reply requested)" : "");
/* Send a reply if the standby requested one. */
if (reply.replyRequested)
if (replyRequested)
WalSndKeepalive(false);
/*
@ -548,9 +563,9 @@ ProcessStandbyReplyMessage(void)
volatile WalSnd *walsnd = MyWalSnd;
SpinLockAcquire(&walsnd->mutex);
walsnd->write = reply.write;
walsnd->flush = reply.flush;
walsnd->apply = reply.apply;
walsnd->write = writePtr;
walsnd->flush = flushPtr;
walsnd->apply = applyPtr;
SpinLockRelease(&walsnd->mutex);
}
@ -564,20 +579,25 @@ ProcessStandbyReplyMessage(void)
static void
ProcessStandbyHSFeedbackMessage(void)
{
StandbyHSFeedbackMessage reply;
TransactionId nextXid;
uint32 nextEpoch;
TransactionId feedbackXmin;
uint32 feedbackEpoch;
/* Decipher the reply message */
pq_copymsgbytes(&reply_message, (char *) &reply,
sizeof(StandbyHSFeedbackMessage));
/*
* Decipher the reply message. The caller already consumed the msgtype
* byte.
*/
(void) pq_getmsgint64(&reply_message); /* sendTime; not used ATM */
feedbackXmin = pq_getmsgint(&reply_message, 4);
feedbackEpoch = pq_getmsgint(&reply_message, 4);
elog(DEBUG2, "hot standby feedback xmin %u epoch %u",
reply.xmin,
reply.epoch);
feedbackXmin,
feedbackEpoch);
/* Ignore invalid xmin (can't actually happen with current walreceiver) */
if (!TransactionIdIsNormal(reply.xmin))
if (!TransactionIdIsNormal(feedbackXmin))
return;
/*
@ -589,18 +609,18 @@ ProcessStandbyHSFeedbackMessage(void)
*/
GetNextXidAndEpoch(&nextXid, &nextEpoch);
if (reply.xmin <= nextXid)
if (feedbackXmin <= nextXid)
{
if (reply.epoch != nextEpoch)
if (feedbackEpoch != nextEpoch)
return;
}
else
{
if (reply.epoch + 1 != nextEpoch)
if (feedbackEpoch + 1 != nextEpoch)
return;
}
if (!TransactionIdPrecedesOrEquals(reply.xmin, nextXid))
if (!TransactionIdPrecedesOrEquals(feedbackXmin, nextXid))
return; /* epoch OK, but it's wrapped around */
/*
@ -610,9 +630,9 @@ ProcessStandbyHSFeedbackMessage(void)
* cleanup conflicts on the standby server.
*
* There is a small window for a race condition here: although we just
* checked that reply.xmin precedes nextXid, the nextXid could have gotten
* checked that feedbackXmin precedes nextXid, the nextXid could have gotten
* advanced between our fetching it and applying the xmin below, perhaps
* far enough to make reply.xmin wrap around. In that case the xmin we
* far enough to make feedbackXmin wrap around. In that case the xmin we
* set here would be "in the future" and have no effect. No point in
* worrying about this since it's too late to save the desired data
* anyway. Assuming that the standby sends us an increasing sequence of
@ -625,7 +645,7 @@ ProcessStandbyHSFeedbackMessage(void)
* safe, and if we're moving it backwards, well, the data is at risk
* already since a VACUUM could have just finished calling GetOldestXmin.)
*/
MyPgXact->xmin = reply.xmin;
MyPgXact->xmin = feedbackXmin;
}
/* Main loop of walsender process that streams the WAL over Copy messages. */
@ -635,17 +655,12 @@ WalSndLoop(void)
bool caughtup = false;
/*
* Allocate buffer that will be used for each output message. We do this
* just once to reduce palloc overhead. The buffer must be made large
* enough for maximum-sized messages.
*/
output_message = palloc(1 + sizeof(WalDataMessageHeader) + MAX_SEND_SIZE);
/*
* Allocate buffer that will be used for processing reply messages. As
* above, do this just once to reduce palloc overhead.
* Allocate buffers that will be used for each outgoing and incoming
* message. We do this just once to reduce palloc overhead.
*/
initStringInfo(&output_message);
initStringInfo(&reply_message);
initStringInfo(&tmpbuf);
/* Initialize the last reply timestamp */
last_reply_timestamp = GetCurrentTimestamp();
@ -1048,7 +1063,6 @@ XLogSend(bool *caughtup)
XLogRecPtr startptr;
XLogRecPtr endptr;
Size nbytes;
WalDataMessageHeader msghdr;
/*
* Attempt to send all data that's already been written out and fsync'd to
@ -1125,25 +1139,31 @@ XLogSend(bool *caughtup)
/*
* OK to read and send the slice.
*/
output_message[0] = 'w';
resetStringInfo(&output_message);
pq_sendbyte(&output_message, 'w');
pq_sendint64(&output_message, startptr); /* dataStart */
pq_sendint64(&output_message, SendRqstPtr); /* walEnd */
pq_sendint64(&output_message, 0); /* sendtime, filled in last */
/*
* Read the log directly into the output buffer to avoid extra memcpy
* calls.
*/
XLogRead(output_message + 1 + sizeof(WalDataMessageHeader), startptr, nbytes);
enlargeStringInfo(&output_message, nbytes);
XLogRead(&output_message.data[output_message.len], startptr, nbytes);
output_message.len += nbytes;
output_message.data[output_message.len] = '\0';
/*
* We fill the message header last so that the send timestamp is taken as
* late as possible.
* Fill the send timestamp last, so that it is taken as late as possible.
*/
msghdr.dataStart = startptr;
msghdr.walEnd = SendRqstPtr;
msghdr.sendTime = GetCurrentTimestamp();
resetStringInfo(&tmpbuf);
pq_sendint64(&tmpbuf, GetCurrentIntegerTimestamp());
memcpy(&output_message.data[1 + sizeof(int64) + sizeof(int64)],
tmpbuf.data, sizeof(int64));
memcpy(output_message + 1, &msghdr, sizeof(WalDataMessageHeader));
pq_putmessage_noblock('d', output_message, 1 + sizeof(WalDataMessageHeader) + nbytes);
pq_putmessage_noblock('d', output_message.data, output_message.len);
sentPtr = endptr;
@ -1518,19 +1538,17 @@ pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
static void
WalSndKeepalive(bool requestReply)
{
PrimaryKeepaliveMessage keepalive_message;
/* Construct a new message */
keepalive_message.walEnd = sentPtr;
keepalive_message.sendTime = GetCurrentTimestamp();
keepalive_message.replyRequested = requestReply;
elog(DEBUG2, "sending replication keepalive");
/* Prepend with the message type and send it. */
output_message[0] = 'k';
memcpy(output_message + 1, &keepalive_message, sizeof(PrimaryKeepaliveMessage));
pq_putmessage_noblock('d', output_message, sizeof(PrimaryKeepaliveMessage) + 1);
/* construct the message... */
resetStringInfo(&output_message);
pq_sendbyte(&output_message, 'k');
pq_sendint64(&output_message, sentPtr);
pq_sendint64(&output_message, GetCurrentIntegerTimestamp());
pq_sendbyte(&output_message, requestReply ? 1 : 0);
/* ... and send it wrapped in CopyData */
pq_putmessage_noblock('d', output_message.data, output_message.len);
}
/*