cascade_replication_v0.patch

*** 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;