Hi,
I'd like to propose cascade replication feature (i.e., allow the
standby to accept
replication connection from another standby) for 9.2. This feature is useful to
reduce the overhead of the master since by using that we can decrease the
number of standbys directly connecting to the master.
I attached the WIP patch, which changes walsender so that it starts replication
even during recovery. Then, the walsender attempts to send all WAL that's
already been fsync'd to the standby's disk (i.e., send WAL up to the bigger
location between the receive location and the replay one). When the standby is
promoted, all walsenders in that standby end because they cannot continue
replication any more in that case because of the timeline mismatch.
The standby must not accept replication connection from that standby itself.
Otherwise, since any new WAL data would not appear in that standby,
replication cannot advance any more. As a safeguard against this, I introduced
new ID to identify each instance. The walsender sends that ID as the fourth
field of the reply of IDENTIFY_SYSTEM, and then walreceiver checks whether
the IDs are the same between two servers. If they are the same, which means
that the standby is just connecting to that standby itself, so walreceiver
emits ERROR.
One remaining problem which I'll have to tackle is that: Even while walreceiver
is not in progress (i.e., the startup process is retrieving WAL file from the
archive), the cascading walsender should continuously send new WAL data.
This means that the walsender should send the WAL file restored from the
archive. The problem is that the name of such a restored WAL file is always
"RECOVERYXLOG". For now, walsender cannot handle the WAL file with such
a name.
To address the above problem, I'm thinking to make the startup process restore
the WAL file with its real name instead of "RECOVERYXLOG". Then, like in the
master, the walsender can read and send the restored WAL file. The required
WAL file can be recycled before being sent. So we might need to enable
wal_keep_segments setting even in the standby.
Comments? Objections?
Regards,
--
Fujii Masao
NIPPON TELEGRAPH AND TELEPHONE CORPORATION
NTT Open Source Software Center
*** a/doc/src/sgml/protocol.sgml
--- b/doc/src/sgml/protocol.sgml
***************
*** 1357,1362 **** The commands accepted in walsender mode are:
--- 1357,1374 ----
</listitem>
</varlistentry>
+ <varlistentry>
+ <term>
+ identificationkey
+ </term>
+ <listitem>
+ <para>
+ Identification key. Also useful to check that the standby is
+ not connecting to that standby itself.
+ </para>
+ </listitem>
+ </varlistentry>
+
</variablelist>
</para>
</listitem>
*** a/src/backend/access/transam/xlog.c
--- b/src/backend/access/transam/xlog.c
***************
*** 9551,9556 **** GetXLogReplayRecPtr(void)
--- 9551,9572 ----
}
/*
+ * Get current standby flush position, ie, the last WAL position
+ * known to be fsync'd to disk in standby.
+ */
+ XLogRecPtr
+ GetStandbyFlushRecPtr(void)
+ {
+ XLogRecPtr recvptr;
+ XLogRecPtr redoptr;
+
+ recvptr = GetWalRcvWriteRecPtr(NULL);
+ redoptr = GetXLogReplayRecPtr();
+
+ return XLByteLT(recvptr, redoptr) ? redoptr : recvptr;
+ }
+
+ /*
* Report the last WAL replay location (same format as pg_start_backup etc)
*
* This is useful for determining how much of WAL is visible to read-only
*** a/src/backend/postmaster/postmaster.c
--- b/src/backend/postmaster/postmaster.c
***************
*** 351,357 **** static void processCancelRequest(Port *port, void *pkt);
static int initMasks(fd_set *rmask);
static void report_fork_failure_to_client(Port *port, int errnum);
static CAC_state canAcceptConnections(void);
- static long PostmasterRandom(void);
static void RandomSalt(char *md5Salt);
static void signal_child(pid_t pid, int signal);
static bool SignalSomeChildren(int signal, int targets);
--- 351,356 ----
***************
*** 2410,2415 **** reaper(SIGNAL_ARGS)
--- 2409,2423 ----
pmState = PM_RUN;
/*
+ * Kill the cascading walsender to urge the cascaded standby to
+ * reread the timeline history file, adjust its timeline and
+ * establish replication connection again. This is required
+ * because the timeline of cascading standby is not consistent
+ * with that of cascaded one just after failover.
+ */
+ SignalSomeChildren(SIGUSR2, BACKEND_TYPE_WALSND);
+
+ /*
* Crank up the background writer, if we didn't do that already
* when we entered consistent recovery state. It doesn't matter
* if this fails, we'll just try again later.
***************
*** 4369,4375 **** RandomSalt(char *md5Salt)
/*
* PostmasterRandom
*/
! static long
PostmasterRandom(void)
{
/*
--- 4377,4383 ----
/*
* PostmasterRandom
*/
! long
PostmasterRandom(void)
{
/*
*** a/src/backend/replication/basebackup.c
--- b/src/backend/replication/basebackup.c
***************
*** 339,344 **** SendBaseBackup(BaseBackupCmd *cmd)
--- 339,349 ----
MemoryContext old_context;
basebackup_options opt;
+ if (cascading_walsender)
+ ereport(FATAL,
+ (errcode(ERRCODE_CANNOT_CONNECT_NOW),
+ errmsg("recovery is still in progress, can't accept WAL streaming connections for backup")));
+
parse_basebackup_options(cmd->options, &opt);
backup_context = AllocSetContextCreate(CurrentMemoryContext,
*** a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
--- b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
***************
*** 83,88 **** libpqrcv_connect(char *conninfo, XLogRecPtr startpoint)
--- 83,90 ----
char standby_sysid[32];
TimeLineID primary_tli;
TimeLineID standby_tli;
+ long primary_key;
+ long standby_key;
PGresult *res;
char cmd[64];
***************
*** 110,120 **** libpqrcv_connect(char *conninfo, XLogRecPtr startpoint)
{
PQclear(res);
ereport(ERROR,
! (errmsg("could not receive database system identifier and timeline ID from "
! "the primary server: %s",
PQerrorMessage(streamConn))));
}
! if (PQnfields(res) != 3 || PQntuples(res) != 1)
{
int ntuples = PQntuples(res);
int nfields = PQnfields(res);
--- 112,122 ----
{
PQclear(res);
ereport(ERROR,
! (errmsg("could not receive database system identifier, timeline ID and "
! "identification key from the primary server: %s",
PQerrorMessage(streamConn))));
}
! if (PQnfields(res) != 4 || PQntuples(res) != 1)
{
int ntuples = PQntuples(res);
int nfields = PQnfields(res);
***************
*** 122,132 **** libpqrcv_connect(char *conninfo, XLogRecPtr startpoint)
PQclear(res);
ereport(ERROR,
(errmsg("invalid response from primary server"),
! errdetail("Expected 1 tuple with 3 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.
--- 124,135 ----
PQclear(res);
ereport(ERROR,
(errmsg("invalid response from primary server"),
! errdetail("Expected 1 tuple with 4 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);
+ primary_key = (long) pg_atoi(PQgetvalue(res, 0, 3), 4, 0);
/*
* Confirm that the system identifier of the primary is the same as ours.
***************
*** 147,159 **** libpqrcv_connect(char *conninfo, XLogRecPtr startpoint)
* 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);
--- 150,174 ----
* recovery target timeline.
*/
standby_tli = GetRecoveryTargetTLI();
if (primary_tli != standby_tli)
+ {
+ PQclear(res);
ereport(ERROR,
(errmsg("timeline %u of the primary does not match recovery target timeline %u",
primary_tli, standby_tli)));
+ }
ThisTimeLineID = primary_tli;
+ /*
+ * Confirm that the walreceiver is not connecting to its own standby.
+ */
+ standby_key = GetWalRcvKey();
+ PQclear(res);
+ if (primary_key == standby_key)
+ ereport(ERROR,
+ (errmsg("the standby is just connecting to that standby itself"),
+ errhint("The standby must connect to the primary or another standby.")));
+
/* Start streaming from the point requested by startup process */
snprintf(cmd, sizeof(cmd), "START_REPLICATION %X/%X",
startpoint.xlogid, startpoint.xrecoff);
*** a/src/backend/replication/walreceiver.c
--- b/src/backend/replication/walreceiver.c
***************
*** 42,49 ****
--- 42,51 ----
#include "access/xlog_internal.h"
#include "libpq/pqsignal.h"
#include "miscadmin.h"
+ #include "postmaster/postmaster.h"
#include "replication/walprotocol.h"
#include "replication/walreceiver.h"
+ #include "replication/walsender.h"
#include "storage/ipc.h"
#include "storage/pmsignal.h"
#include "storage/procarray.h"
***************
*** 171,176 **** WalReceiverMain(void)
--- 173,179 ----
{
char conninfo[MAXCONNINFO];
XLogRecPtr startpoint;
+ long walRcvKey;
/* use volatile pointer to prevent code rearrangement */
volatile WalRcvData *walrcv = WalRcv;
***************
*** 184,189 **** WalReceiverMain(void)
--- 187,199 ----
Assert(walrcv != NULL);
/*
+ * Compute the identification key that will be assigned to this walreceiver.
+ * This key is used to check that this walreceiver is not connecting to its
+ * standby itself.
+ */
+ walRcvKey = PostmasterRandom();
+
+ /*
* Mark walreceiver as running in shared memory.
*
* Do this as early as possible, so that if we fail later on, we'll set
***************
*** 215,220 **** WalReceiverMain(void)
--- 225,231 ----
/* Advertise our PID so that the startup process can kill us */
walrcv->pid = MyProcPid;
walrcv->walRcvState = WALRCV_RUNNING;
+ walrcv->walRcvKey = walRcvKey;
/* Fetch information required to start streaming */
strlcpy(conninfo, (char *) walrcv->conninfo, MAXCONNINFO);
***************
*** 354,359 **** WalRcvDie(int code, Datum arg)
--- 365,371 ----
Assert(walrcv->walRcvState == WALRCV_RUNNING ||
walrcv->walRcvState == WALRCV_STOPPING);
walrcv->walRcvState = WALRCV_STOPPED;
+ walrcv->walRcvKey = 0;
walrcv->pid = 0;
SpinLockRelease(&walrcv->mutex);
***************
*** 564,571 **** XLogWalRcvFlush(bool dying)
}
SpinLockRelease(&walrcv->mutex);
! /* Signal the startup process that new WAL has arrived */
WakeupRecovery();
/* Report XLOG streaming progress in PS display */
if (update_process_title)
--- 576,585 ----
}
SpinLockRelease(&walrcv->mutex);
! /* Signal the startup process and walsender that new WAL has arrived */
WakeupRecovery();
+ if (max_wal_senders > 0)
+ WalSndWakeup();
/* Report XLOG streaming progress in PS display */
if (update_process_title)
*** a/src/backend/replication/walreceiverfuncs.c
--- b/src/backend/replication/walreceiverfuncs.c
***************
*** 238,240 **** GetWalRcvWriteRecPtr(XLogRecPtr *latestChunkStart)
--- 238,257 ----
return recptr;
}
+
+ /*
+ * Returns the identification key for the walreceiver.
+ */
+ long
+ GetWalRcvKey(void)
+ {
+ /* use volatile pointer to prevent code rearrangement */
+ volatile WalRcvData *walrcv = WalRcv;
+ long walRcvKey;
+
+ SpinLockAcquire(&walrcv->mutex);
+ walRcvKey = walrcv->walRcvKey;
+ SpinLockRelease(&walrcv->mutex);
+
+ return walRcvKey;
+ }
*** a/src/backend/replication/walsender.c
--- b/src/backend/replication/walsender.c
***************
*** 48,53 ****
--- 48,54 ----
#include "replication/basebackup.h"
#include "replication/replnodes.h"
#include "replication/walprotocol.h"
+ #include "replication/walreceiver.h"
#include "replication/walsender.h"
#include "storage/fd.h"
#include "storage/ipc.h"
***************
*** 70,75 **** WalSnd *MyWalSnd = NULL;
--- 71,77 ----
/* Global state */
bool am_walsender = false; /* Am I a walsender process ? */
+ bool cascading_walsender = false; /* Am I cascading WAL to another standby ? */
/* User-settable parameters for walsender */
int max_wal_senders = 0; /* the maximum number of concurrent walsenders */
***************
*** 135,144 **** WalSenderMain(void)
{
MemoryContext walsnd_context;
! if (RecoveryInProgress())
! ereport(FATAL,
! (errcode(ERRCODE_CANNOT_CONNECT_NOW),
! errmsg("recovery is still in progress, can't accept WAL streaming connections")));
/* Create a per-walsender data structure in shared memory */
InitWalSnd();
--- 137,143 ----
{
MemoryContext walsnd_context;
! cascading_walsender = RecoveryInProgress();
/* Create a per-walsender data structure in shared memory */
InitWalSnd();
***************
*** 165,170 **** WalSenderMain(void)
--- 164,175 ----
/* Unblock signals (they were blocked when the postmaster forked us) */
PG_SETMASK(&UnBlockSig);
+ /*
+ * Use the recovery target timeline ID during recovery
+ */
+ if (cascading_walsender)
+ ThisTimeLineID = GetRecoveryTargetTLI();
+
/* Tell the standby that walsender is ready for receiving commands */
ReadyForQuery(DestRemote);
***************
*** 279,303 **** IdentifySystem(void)
char sysid[32];
char tli[11];
char xpos[MAXFNAMELEN];
XLogRecPtr logptr;
/*
! * Reply with a result set with one row, three columns. First col is
! * system ID, second is timeline ID, and third is current xlog location.
*/
snprintf(sysid, sizeof(sysid), UINT64_FORMAT,
GetSystemIdentifier());
snprintf(tli, sizeof(tli), "%u", ThisTimeLineID);
! logptr = GetInsertRecPtr();
snprintf(xpos, sizeof(xpos), "%X/%X",
logptr.xlogid, logptr.xrecoff);
/* Send a RowDescription message */
pq_beginmessage(&buf, 'T');
! pq_sendint(&buf, 3, 2); /* 3 fields */
/* first field */
pq_sendstring(&buf, "systemid"); /* col name */
--- 284,312 ----
char sysid[32];
char tli[11];
char xpos[MAXFNAMELEN];
+ char key[32];
XLogRecPtr logptr;
/*
! * Reply with a result set with one row, four columns. First col is
! * system ID, second is timeline ID, third is current xlog location,
! * and fourth is identification key.
*/
snprintf(sysid, sizeof(sysid), UINT64_FORMAT,
GetSystemIdentifier());
snprintf(tli, sizeof(tli), "%u", ThisTimeLineID);
! logptr = cascading_walsender ? GetStandbyFlushRecPtr() : GetInsertRecPtr();
snprintf(xpos, sizeof(xpos), "%X/%X",
logptr.xlogid, logptr.xrecoff);
+ snprintf(key, sizeof(key), "%d", (int32) GetWalRcvKey());
+
/* Send a RowDescription message */
pq_beginmessage(&buf, 'T');
! pq_sendint(&buf, 4, 2); /* 4 fields */
/* first field */
pq_sendstring(&buf, "systemid"); /* col name */
***************
*** 325,341 **** IdentifySystem(void)
pq_sendint(&buf, -1, 2);
pq_sendint(&buf, 0, 4);
pq_sendint(&buf, 0, 2);
pq_endmessage(&buf);
/* Send a DataRow message */
pq_beginmessage(&buf, 'D');
! pq_sendint(&buf, 3, 2); /* # of columns */
pq_sendint(&buf, strlen(sysid), 4); /* col1 len */
pq_sendbytes(&buf, (char *) &sysid, strlen(sysid));
pq_sendint(&buf, strlen(tli), 4); /* col2 len */
pq_sendbytes(&buf, (char *) tli, strlen(tli));
pq_sendint(&buf, strlen(xpos), 4); /* col3 len */
pq_sendbytes(&buf, (char *) xpos, strlen(xpos));
pq_endmessage(&buf);
--- 334,361 ----
pq_sendint(&buf, -1, 2);
pq_sendint(&buf, 0, 4);
pq_sendint(&buf, 0, 2);
+
+ /* fourth field */
+ pq_sendstring(&buf, "identificationkey");
+ pq_sendint(&buf, 0, 4);
+ pq_sendint(&buf, 0, 2);
+ pq_sendint(&buf, INT4OID, 4);
+ pq_sendint(&buf, 4, 2);
+ pq_sendint(&buf, 0, 4);
+ pq_sendint(&buf, 0, 2);
pq_endmessage(&buf);
/* Send a DataRow message */
pq_beginmessage(&buf, 'D');
! pq_sendint(&buf, 4, 2); /* # of columns */
pq_sendint(&buf, strlen(sysid), 4); /* col1 len */
pq_sendbytes(&buf, (char *) &sysid, strlen(sysid));
pq_sendint(&buf, strlen(tli), 4); /* col2 len */
pq_sendbytes(&buf, (char *) tli, strlen(tli));
pq_sendint(&buf, strlen(xpos), 4); /* col3 len */
pq_sendbytes(&buf, (char *) xpos, strlen(xpos));
+ pq_sendint(&buf, strlen(key), 4); /* col4 len */
+ pq_sendbytes(&buf, (char *) key, strlen(key));
pq_endmessage(&buf);
***************
*** 372,379 **** StartReplication(StartReplicationCmd *cmd)
* directory that was created with 'minimal'. So this is not bulletproof,
* the purpose is just to give a user-friendly error message that hints
* how to configure the system correctly.
*/
! if (wal_level == WAL_LEVEL_MINIMAL)
ereport(FATAL,
(errcode(ERRCODE_CANNOT_CONNECT_NOW),
errmsg("standby connections not allowed because wal_level=minimal")));
--- 392,403 ----
* directory that was created with 'minimal'. So this is not bulletproof,
* the purpose is just to give a user-friendly error message that hints
* how to configure the system correctly.
+ *
+ * NOTE: The existence of cascading walsender means that wal_level is set
+ * to hot_standby in the master. So we don't need to check the value of
+ * wal_level during recovery.
*/
! if (!cascading_walsender && wal_level == WAL_LEVEL_MINIMAL)
ereport(FATAL,
(errcode(ERRCODE_CANNOT_CONNECT_NOW),
errmsg("standby connections not allowed because wal_level=minimal")));
***************
*** 601,607 **** ProcessStandbyReplyMessage(void)
SpinLockRelease(&walsnd->mutex);
}
! SyncRepReleaseWaiters();
}
/*
--- 625,632 ----
SpinLockRelease(&walsnd->mutex);
}
! if (!cascading_walsender)
! SyncRepReleaseWaiters();
}
/*
***************
*** 1079,1085 **** XLogSend(char *msgbuf, bool *caughtup)
* subsequently crashes and restarts, slaves must not have applied any WAL
* that gets lost on the master.
*/
! SendRqstPtr = GetFlushRecPtr();
/* Quick exit if nothing to do */
if (XLByteLE(SendRqstPtr, sentPtr))
--- 1104,1110 ----
* subsequently crashes and restarts, slaves must not have applied any WAL
* that gets lost on the master.
*/
! SendRqstPtr = cascading_walsender ? GetStandbyFlushRecPtr() : GetFlushRecPtr();
/* Quick exit if nothing to do */
if (XLByteLE(SendRqstPtr, sentPtr))
*** a/src/include/access/xlog.h
--- b/src/include/access/xlog.h
***************
*** 293,298 **** extern bool HotStandbyActive(void);
--- 293,299 ----
extern bool XLogInsertAllowed(void);
extern void GetXLogReceiptTime(TimestampTz *rtime, bool *fromStream);
extern XLogRecPtr GetXLogReplayRecPtr(void);
+ extern XLogRecPtr GetStandbyFlushRecPtr(void);
extern void UpdateControlFile(void);
extern uint64 GetSystemIdentifier(void);
*** a/src/include/postmaster/postmaster.h
--- b/src/include/postmaster/postmaster.h
***************
*** 42,47 **** extern void ClosePostmasterPorts(bool am_syslogger);
--- 42,49 ----
extern int MaxLivePostmasterChildren(void);
+ extern long PostmasterRandom(void);
+
#ifdef EXEC_BACKEND
extern pid_t postmaster_forkexec(int argc, char *argv[]);
extern int SubPostmasterMain(int argc, char *argv[]);
*** a/src/include/replication/walreceiver.h
--- b/src/include/replication/walreceiver.h
***************
*** 43,55 **** typedef enum
typedef struct
{
/*
! * PID of currently active walreceiver process, its current state and
* start time (actually, the time at which it was requested to be
! * started).
*/
pid_t pid;
WalRcvState walRcvState;
pg_time_t startTime;
/*
* receiveStart is the first byte position that will be received. When
--- 43,56 ----
typedef struct
{
/*
! * PID of currently active walreceiver process, its current state,
* start time (actually, the time at which it was requested to be
! * started) and identification key for this walreceiver.
*/
pid_t pid;
WalRcvState walRcvState;
pg_time_t startTime;
+ long walRcvKey;
/*
* receiveStart is the first byte position that will be received. When
***************
*** 108,112 **** extern void ShutdownWalRcv(void);
--- 109,114 ----
extern bool WalRcvInProgress(void);
extern void RequestXLogStreaming(XLogRecPtr recptr, const char *conninfo);
extern XLogRecPtr GetWalRcvWriteRecPtr(XLogRecPtr *latestChunkStart);
+ extern long GetWalRcvKey(void);
#endif /* _WALRECEIVER_H */
*** a/src/include/replication/walsender.h
--- b/src/include/replication/walsender.h
***************
*** 92,97 **** extern WalSndCtlData *WalSndCtl;
--- 92,98 ----
/* global state */
extern bool am_walsender;
+ extern bool cascading_walsender;
extern volatile sig_atomic_t walsender_shutdown_requested;
extern volatile sig_atomic_t walsender_ready_to_stop;
--
Sent via pgsql-hackers mailing list (pgsql-hackers@postgresql.org)
To make changes to your subscription:
http://www.postgresql.org/mailpref/pgsql-hackers
