Location via proxy:   [ UP ]  
[Report a bug]   [Manage cookies]                
Skip to content

Commit 808969d

Browse files
committed
Add a message type header to the CopyData messages sent from primary
to standby in streaming replication. While we only have one message type at the moment, adding a message type header makes this easier to extend.
1 parent 47c5b8f commit 808969d

File tree

5 files changed

+117
-35
lines changed

5 files changed

+117
-35
lines changed

doc/src/sgml/protocol.sgml

+57-4
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
<!-- $PostgreSQL: pgsql/doc/src/sgml/protocol.sgml,v 1.77 2010/01/15 09:18:59 heikki Exp $ -->
1+
<!-- $PostgreSQL: pgsql/doc/src/sgml/protocol.sgml,v 1.78 2010/02/03 09:47:19 heikki Exp $ -->
22

33
<chapter id="protocol">
44
<title>Frontend/Backend Protocol</title>
@@ -4179,12 +4179,65 @@ The commands accepted in walsender mode are:
41794179
already been recycled. On success, server responds with a
41804180
CopyOutResponse message, and backend starts to stream WAL as CopyData
41814181
messages.
4182+
The payload in CopyData message consists of the following format.
41824183
</para>
41834184

41844185
<para>
4185-
The payload in each CopyData message consists of an XLogRecPtr,
4186-
indicating the starting point of the WAL in the message, immediately
4187-
followed by the WAL data itself.
4186+
<variablelist>
4187+
<varlistentry>
4188+
<term>
4189+
XLogData (B)
4190+
</term>
4191+
<listitem>
4192+
<para>
4193+
<variablelist>
4194+
<varlistentry>
4195+
<term>
4196+
Byte1('w')
4197+
</term>
4198+
<listitem>
4199+
<para>
4200+
Identifies the message as WAL data.
4201+
</para>
4202+
</listitem>
4203+
</varlistentry>
4204+
<varlistentry>
4205+
<term>
4206+
Int32
4207+
</term>
4208+
<listitem>
4209+
<para>
4210+
The log file number of the LSN, indicating the starting point of
4211+
the WAL in the message.
4212+
</para>
4213+
</listitem>
4214+
</varlistentry>
4215+
<varlistentry>
4216+
<term>
4217+
Int32
4218+
</term>
4219+
<listitem>
4220+
<para>
4221+
The byte offset of the LSN, indicating the starting point of
4222+
the WAL in the message.
4223+
</para>
4224+
</listitem>
4225+
</varlistentry>
4226+
<varlistentry>
4227+
<term>
4228+
Byte<replaceable>n</replaceable>
4229+
</term>
4230+
<listitem>
4231+
<para>
4232+
Data that forms part of WAL data stream.
4233+
</para>
4234+
</listitem>
4235+
</varlistentry>
4236+
</variablelist>
4237+
</para>
4238+
</listitem>
4239+
</varlistentry>
4240+
</variablelist>
41884241
</para>
41894242
<para>
41904243
A single WAL record is never split across two CopyData messages. When

src/backend/replication/libpqwalreceiver/libpqwalreceiver.c

+15-19
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@
1010
*
1111
*
1212
* IDENTIFICATION
13-
* $PostgreSQL: pgsql/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c,v 1.2 2010/01/20 11:58:44 heikki Exp $
13+
* $PostgreSQL: pgsql/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c,v 1.3 2010/02/03 09:47:19 heikki Exp $
1414
*
1515
*-------------------------------------------------------------------------
1616
*/
@@ -48,8 +48,8 @@ static char *recvBuf = NULL;
4848

4949
/* Prototypes for interface functions */
5050
static bool libpqrcv_connect(char *conninfo, XLogRecPtr startpoint);
51-
static bool libpqrcv_receive(int timeout, XLogRecPtr *recptr, char **buffer,
52-
int *len);
51+
static bool libpqrcv_receive(int timeout, unsigned char *type,
52+
char **buffer, int *len);
5353
static void libpqrcv_disconnect(void);
5454

5555
/* Prototypes for private functions */
@@ -236,13 +236,13 @@ libpqrcv_disconnect(void)
236236
}
237237

238238
/*
239-
* Receive any WAL records available from XLOG stream, blocking for
239+
* Receive a message available from XLOG stream, blocking for
240240
* maximum of 'timeout' ms.
241241
*
242242
* Returns:
243243
*
244-
* True if data was received. *recptr, *buffer and *len are set to
245-
* the WAL location of the received data, buffer holding it, and length,
244+
* True if data was received. *type, *buffer and *len are set to
245+
* the type of the received data, buffer holding it, and length,
246246
* respectively.
247247
*
248248
* False if no data was available within timeout, or wait was interrupted
@@ -254,7 +254,7 @@ libpqrcv_disconnect(void)
254254
* ereports on error.
255255
*/
256256
static bool
257-
libpqrcv_receive(int timeout, XLogRecPtr *recptr, char **buffer, int *len)
257+
libpqrcv_receive(int timeout, unsigned char *type, char **buffer, int *len)
258258
{
259259
int rawlen;
260260

@@ -275,14 +275,14 @@ libpqrcv_receive(int timeout, XLogRecPtr *recptr, char **buffer, int *len)
275275

276276
if (PQconsumeInput(streamConn) == 0)
277277
ereport(ERROR,
278-
(errmsg("could not read xlog records: %s",
278+
(errmsg("could not receive data from XLOG stream: %s",
279279
PQerrorMessage(streamConn))));
280280
}
281281
justconnected = false;
282282

283283
/* Receive CopyData message */
284284
rawlen = PQgetCopyData(streamConn, &recvBuf, 1);
285-
if (rawlen == 0) /* no records available yet, then return */
285+
if (rawlen == 0) /* no data available yet, then return */
286286
return false;
287287
if (rawlen == -1) /* end-of-streaming or error */
288288
{
@@ -297,22 +297,18 @@ libpqrcv_receive(int timeout, XLogRecPtr *recptr, char **buffer, int *len)
297297
}
298298
PQclear(res);
299299
ereport(ERROR,
300-
(errmsg("could not read xlog records: %s",
300+
(errmsg("could not receive data from XLOG stream: %s",
301301
PQerrorMessage(streamConn))));
302302
}
303303
if (rawlen < -1)
304304
ereport(ERROR,
305-
(errmsg("could not read xlog records: %s",
305+
(errmsg("could not receive data from XLOG stream: %s",
306306
PQerrorMessage(streamConn))));
307307

308-
if (rawlen < sizeof(XLogRecPtr))
309-
ereport(ERROR,
310-
(errmsg("invalid WAL message received from primary")));
311-
312-
/* Return received WAL records to caller */
313-
*recptr = *((XLogRecPtr *) recvBuf);
314-
*buffer = recvBuf + sizeof(XLogRecPtr);
315-
*len = rawlen - sizeof(XLogRecPtr);
308+
/* Return received messages to caller */
309+
*type = *((unsigned char *) recvBuf);
310+
*buffer = recvBuf + sizeof(*type);
311+
*len = rawlen - sizeof(*type);
316312

317313
return true;
318314
}

src/backend/replication/walreceiver.c

+40-9
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,7 @@
2929
*
3030
*
3131
* IDENTIFICATION
32-
* $PostgreSQL: pgsql/src/backend/replication/walreceiver.c,v 1.2 2010/01/27 15:27:51 heikki Exp $
32+
* $PostgreSQL: pgsql/src/backend/replication/walreceiver.c,v 1.3 2010/02/03 09:47:19 heikki Exp $
3333
*
3434
*-------------------------------------------------------------------------
3535
*/
@@ -135,6 +135,7 @@ static void WalRcvQuickDieHandler(SIGNAL_ARGS);
135135

136136
/* Prototypes for private functions */
137137
static void WalRcvDie(int code, Datum arg);
138+
static void XLogWalRcvProcessMsg(unsigned char type, char *buf, Size len);
138139
static void XLogWalRcvWrite(char *buf, Size nbytes, XLogRecPtr recptr);
139140
static void XLogWalRcvFlush(void);
140141

@@ -258,7 +259,7 @@ WalReceiverMain(void)
258259
/* Loop until end-of-streaming or error */
259260
for (;;)
260261
{
261-
XLogRecPtr recptr;
262+
unsigned char type;
262263
char *buf;
263264
int len;
264265

@@ -287,17 +288,17 @@ WalReceiverMain(void)
287288
}
288289

289290
/* Wait a while for data to arrive */
290-
if (walrcv_receive(NAPTIME_PER_CYCLE, &recptr, &buf, &len))
291+
if (walrcv_receive(NAPTIME_PER_CYCLE, &type, &buf, &len))
291292
{
292-
/* Write received WAL records to disk */
293-
XLogWalRcvWrite(buf, len, recptr);
293+
/* Accept the received data, and process it */
294+
XLogWalRcvProcessMsg(type, buf, len);
294295

295-
/* Receive any more WAL records we can without sleeping */
296-
while(walrcv_receive(0, &recptr, &buf, &len))
297-
XLogWalRcvWrite(buf, len, recptr);
296+
/* Receive any more data we can without sleeping */
297+
while(walrcv_receive(0, &type, &buf, &len))
298+
XLogWalRcvProcessMsg(type, buf, len);
298299

299300
/*
300-
* Now that we've written some records, flush them to disk and
301+
* If we've written some records, flush them to disk and
301302
* let the startup process know about them.
302303
*/
303304
XLogWalRcvFlush();
@@ -375,6 +376,36 @@ WalRcvQuickDieHandler(SIGNAL_ARGS)
375376
exit(2);
376377
}
377378

379+
/*
380+
* Accept the message from XLOG stream, and process it.
381+
*/
382+
static void
383+
XLogWalRcvProcessMsg(unsigned char type, char *buf, Size len)
384+
{
385+
switch (type)
386+
{
387+
case 'w': /* WAL records */
388+
{
389+
XLogRecPtr recptr;
390+
391+
if (len < sizeof(XLogRecPtr))
392+
ereport(ERROR,
393+
(errmsg("invalid WAL message received from primary")));
394+
395+
recptr = *((XLogRecPtr *) buf);
396+
buf += sizeof(XLogRecPtr);
397+
len -= sizeof(XLogRecPtr);
398+
XLogWalRcvWrite(buf, len, recptr);
399+
break;
400+
}
401+
default:
402+
ereport(ERROR,
403+
(errcode(ERRCODE_PROTOCOL_VIOLATION),
404+
errmsg("invalid replication message type %d",
405+
type)));
406+
}
407+
}
408+
378409
/*
379410
* Write XLOG data to disk.
380411
*/

src/backend/replication/walsender.c

+2-1
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,7 @@
3030
*
3131
*
3232
* IDENTIFICATION
33-
* $PostgreSQL: pgsql/src/backend/replication/walsender.c,v 1.4 2010/01/27 16:41:09 heikki Exp $
33+
* $PostgreSQL: pgsql/src/backend/replication/walsender.c,v 1.5 2010/02/03 09:47:19 heikki Exp $
3434
*
3535
*-------------------------------------------------------------------------
3636
*/
@@ -659,6 +659,7 @@ XLogSend(StringInfo outMsg)
659659
* have the same byte order. If they have different byte order, we
660660
* don't reach here.
661661
*/
662+
pq_sendbyte(outMsg, 'w');
662663
pq_sendbytes(outMsg, (char *) &startptr, sizeof(startptr));
663664

664665
if (endptr.xlogid != startptr.xlogid)

src/include/replication/walreceiver.h

+3-2
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@
55
*
66
* Portions Copyright (c) 2010-2010, PostgreSQL Global Development Group
77
*
8-
* $PostgreSQL: pgsql/src/include/replication/walreceiver.h,v 1.5 2010/01/27 15:27:51 heikki Exp $
8+
* $PostgreSQL: pgsql/src/include/replication/walreceiver.h,v 1.6 2010/02/03 09:47:19 heikki Exp $
99
*
1010
*-------------------------------------------------------------------------
1111
*/
@@ -66,7 +66,8 @@ extern WalRcvData *WalRcv;
6666
typedef bool (*walrcv_connect_type) (char *conninfo, XLogRecPtr startpoint);
6767
extern PGDLLIMPORT walrcv_connect_type walrcv_connect;
6868

69-
typedef bool (*walrcv_receive_type) (int timeout, XLogRecPtr *recptr, char **buffer, int *len);
69+
typedef bool (*walrcv_receive_type) (int timeout, unsigned char *type,
70+
char **buffer, int *len);
7071
extern PGDLLIMPORT walrcv_receive_type walrcv_receive;
7172

7273
typedef void (*walrcv_disconnect_type) (void);

0 commit comments

Comments
 (0)