wal-receiver-replies.patch

diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index 5a43774..63c6283 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -1984,6 +1984,29 @@ SET ENABLE_SEQSCAN TO OFF;
        </listitem>
       </varlistentry>
 
+      <varlistentry id="guc-wal-receiver-status-interval" xreflabel="wal_receiver_status_interval">
+       <term><varname>wal_receiver_status_interval</varname> (<type>integer</type>)</term>
+       <indexterm>
+        <primary><varname>wal_receiver_status_interval</> configuration parameter</primary>
+       </indexterm>
+       <listitem>
+       <para>
+        Specifies the minimum frequency, in seconds, for the WAL receiver
+        process on the standby to send information about replication progress
+        to the primary, where they can be seen using the
+        <literal>pg_stat_replication</literal> view.  The standby will report
+        the last transaction log position it has written, the last position it
+        has flushed to disk, and the last position it has applied.  Updates are
+        sent each time the write or flush positions changed, or at least as
+        often as specified by this parameter.  Thus, the apply position may
+        lag slightly behind the true position.  Setting this parameter to zero
+        disables status updates completely.  This parameter can only be set in
+        the <filename>postgresql.conf</> file or on the server command line.
+        The default value is 10 seconds.
+       </para>
+       </listitem>
+      </varlistentry>
+
      <varlistentry id="guc-vacuum-defer-cleanup-age" xreflabel="vacuum_defer_cleanup_age">
       <term><varname>vacuum_defer_cleanup_age</varname> (<type>integer</type>)</term>
       <indexterm>
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index ca83421..f3481bb 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -298,8 +298,11 @@ postgres: <replaceable>user</> <replaceable>database</> <replaceable>host</> <re
       <entry><structname>pg_stat_replication</><indexterm><primary>pg_stat_replication</primary></indexterm></entry>
       <entry>One row per WAL sender process, showing process <acronym>ID</>,
       user OID, user name, application name, client's address and port number,
-      time at which the server process began execution, current WAL sender
-      state and transaction log location. The columns detailing what exactly
+      time at which the server process began execution, and the current WAL
+      sender state and transaction log location.  In addition, the standby
+      reports the last transaction log position it received and wrote, the last
+      position it flushed to disk, and the last position it replayed, and this
+      information is also displayed here.  The columns detailing what exactly
       the connection is doing are only visible if the user examining the view
       is a superuser.
      </entry>
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 25c7e06..3680811 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -9029,6 +9029,25 @@ pg_last_xlog_receive_location(PG_FUNCTION_ARGS)
 }
 
 /*
+ * Get latest redo apply position.
+ *
+ * Exported to allow WALReceiver to read the pointer directly.
+ */
+XLogRecPtr
+GetXLogReplayRecPtr(void)
+{
+	/* use volatile pointer to prevent code rearrangement */
+	volatile XLogCtlData *xlogctl = XLogCtl;
+	XLogRecPtr	recptr;
+
+	SpinLockAcquire(&xlogctl->info_lck);
+	recptr = xlogctl->recoveryLastRecPtr;
+	SpinLockRelease(&xlogctl->info_lck);
+
+	return recptr;
+}
+
+/*
  * 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
@@ -9037,14 +9056,10 @@ pg_last_xlog_receive_location(PG_FUNCTION_ARGS)
 Datum
 pg_last_xlog_replay_location(PG_FUNCTION_ARGS)
 {
-	/* use volatile pointer to prevent code rearrangement */
-	volatile XLogCtlData *xlogctl = XLogCtl;
 	XLogRecPtr	recptr;
 	char		location[MAXFNAMELEN];
 
-	SpinLockAcquire(&xlogctl->info_lck);
-	recptr = xlogctl->recoveryLastRecPtr;
-	SpinLockRelease(&xlogctl->info_lck);
+	recptr = GetXLogReplayRecPtr();
 
 	if (recptr.xlogid == 0 && recptr.xrecoff == 0)
 		PG_RETURN_NULL();
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 718e996..40e94ba 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -502,7 +502,10 @@ CREATE VIEW pg_stat_replication AS
             S.client_port,
             S.backend_start,
             W.state,
-            W.sent_location
+            W.sent_location,
+            W.write_location,
+            W.flush_location,
+            W.apply_location
     FROM pg_stat_get_activity(NULL) AS S, pg_authid U,
             pg_stat_get_wal_senders() AS W
     WHERE S.usesysid = U.oid AND
diff --git a/src/backend/replication/walreceiver.c b/src/backend/replication/walreceiver.c
index 7005307..35cd121 100644
--- a/src/backend/replication/walreceiver.c
+++ b/src/backend/replication/walreceiver.c
@@ -53,6 +53,7 @@
 
 /* Global variable to indicate if this process is a walreceiver process */
 bool		am_walreceiver;
+int			wal_receiver_status_interval;
 
 /* libpqreceiver hooks to these when loaded */
 walrcv_connect_type walrcv_connect = NULL;
@@ -88,6 +89,8 @@ static struct
 	XLogRecPtr	Flush;			/* last byte + 1 flushed in the standby */
 }	LogstreamResult;
 
+static StandbyReplyMessage	reply_message;
+
 /*
  * About SIGTERM handling:
  *
@@ -114,6 +117,7 @@ static void WalRcvDie(int code, Datum arg);
 static void XLogWalRcvProcessMsg(unsigned char type, char *buf, Size len);
 static void XLogWalRcvWrite(char *buf, Size nbytes, XLogRecPtr recptr);
 static void XLogWalRcvFlush(void);
+static void XLogWalRcvSendReply(void);
 
 /* Signal handlers */
 static void WalRcvSigHupHandler(SIGNAL_ARGS);
@@ -306,12 +310,24 @@ WalReceiverMain(void)
 			while (walrcv_receive(0, &type, &buf, &len))
 				XLogWalRcvProcessMsg(type, buf, len);
 
+			/* Let the master know that we received some data. */
+			XLogWalRcvSendReply();
+
 			/*
 			 * If we've written some records, flush them to disk and let the
 			 * startup process know about them.
 			 */
 			XLogWalRcvFlush();
 		}
+
+		/*
+		 * Send a status update to the master.
+		 *
+		 * If we received any data this cycle, the flush position will have
+		 * advanced; and the apply position may have advanced whether we got
+		 * any new data or not.
+		 */
+		XLogWalRcvSendReply();
 	}
 }
 
@@ -548,3 +564,51 @@ XLogWalRcvFlush(void)
 		}
 	}
 }
+
+/*
+ * Send reply message to primary, indicating our current XLOG positions and
+ * the current time.
+ */
+static void
+XLogWalRcvSendReply(void)
+{
+	TimestampTz		now;
+
+	/*
+	 * If the user doesn't want status to be reported to the master, be sure
+	 * to exit before doing anything at all.
+	 */
+	if (wal_receiver_status_interval <= 0)
+		return;
+
+	/* Get current timestamp. */
+	now = GetCurrentTimestamp();
+
+	/*
+	 * We can compare the write and flush positions to the last message we sent
+	 * without taking any lock, but the apply position requires a spin lock, so
+	 * we don't check that unless something else has changed or 10 seconds have
+	 * passed.  This means that the apply log position will appear, from the
+	 * master's point of view, to lag slightly, but since this is only for
+	 * reporting purposes and only on idle systems, that's probably OK.
+	 */
+	if (XLByteEQ(reply_message.write, LogstreamResult.Write)
+		&& XLByteEQ(reply_message.flush, LogstreamResult.Flush)
+		&& !TimestampDifferenceExceeds(reply_message.sendTime, now,
+			wal_receiver_status_interval * 1000))
+		return;
+
+	/* Construct a new message. */
+	reply_message.write = LogstreamResult.Write;
+	reply_message.flush = LogstreamResult.Flush;
+	reply_message.apply = GetXLogReplayRecPtr();
+	reply_message.sendTime = now;
+
+	elog(DEBUG2, "sending write %X/%X flush %X/%X apply %X/%X",
+				 reply_message.write.xlogid, reply_message.write.xrecoff,
+				 reply_message.flush.xlogid, reply_message.flush.xrecoff,
+				 reply_message.apply.xlogid, reply_message.apply.xrecoff);
+
+	/* Send it. */
+	walrcv_send((char *) &reply_message, sizeof(StandbyReplyMessage));
+}
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 78963c1..fcb5a32 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -39,6 +39,7 @@
 
 #include "funcapi.h"
 #include "access/xlog_internal.h"
+#include "access/transam.h"
 #include "catalog/pg_type.h"
 #include "libpq/libpq.h"
 #include "libpq/pqformat.h"
@@ -51,6 +52,7 @@
 #include "storage/fd.h"
 #include "storage/ipc.h"
 #include "storage/pmsignal.h"
+#include "storage/proc.h"
 #include "tcop/tcopprot.h"
 #include "utils/builtins.h"
 #include "utils/guc.h"
@@ -106,9 +108,10 @@ static void InitWalSnd(void);
 static void WalSndHandshake(void);
 static void WalSndKill(int code, Datum arg);
 static bool XLogSend(char *msgbuf, bool *caughtup);
-static void CheckClosedConnection(void);
 static void IdentifySystem(void);
 static void StartReplication(StartReplicationCmd * cmd);
+static void ProcessStandbyReplyMessage(void);
+static void ProcessRepliesIfAny(void);
 
 
 /* Main entry point for walsender process */
@@ -173,7 +176,7 @@ WalSenderMain(void)
 static void
 WalSndHandshake(void)
 {
-	StringInfoData input_message;
+	static StringInfoData input_message;
 	bool		replication_started = false;
 
 	initStringInfo(&input_message);
@@ -442,7 +445,7 @@ HandleReplicationCommand(const char *cmd_string)
  * Check if the remote end has closed the connection.
  */
 static void
-CheckClosedConnection(void)
+ProcessRepliesIfAny(void)
 {
 	unsigned char firstchar;
 	int			r;
@@ -466,6 +469,13 @@ CheckClosedConnection(void)
 	switch (firstchar)
 	{
 			/*
+			 * 'd' means a standby reply wrapped in a COPY BOTH packet.
+			 */
+		case 'd':
+			ProcessStandbyReplyMessage();
+			break;
+
+			/*
 			 * 'X' means that the standby is closing down the socket.
 			 */
 		case 'X':
@@ -479,6 +489,54 @@ CheckClosedConnection(void)
 	}
 }
 
+/*
+ * Receive StandbyReplyMessage. False if message send failed.
+ */
+static void
+ProcessStandbyReplyMessage(void)
+{
+	static StringInfoData input_message;
+	StandbyReplyMessage	reply;
+
+	initStringInfo(&input_message);
+
+	/*
+	 * Read the message contents.
+	 */
+	if (pq_getmessage(&input_message, 0))
+	{
+		ereport(COMMERROR,
+				(errcode(ERRCODE_PROTOCOL_VIOLATION),
+				 errmsg("unexpected EOF on standby connection")));
+		proc_exit(0);
+	}
+
+	pq_copymsgbytes(&input_message, (char *) &reply, sizeof(StandbyReplyMessage));
+
+	elog(DEBUG2, "write %X/%X flush %X/%X apply %X/%X ",
+		 reply.write.xlogid, reply.write.xrecoff,
+		 reply.flush.xlogid, reply.flush.xrecoff,
+		 reply.apply.xlogid, reply.apply.xrecoff);
+
+	/*
+	 * Update shared state for this WalSender process
+	 * based on reply data from standby.
+	 */
+	{
+		/* use volatile pointer to prevent code rearrangement */
+		volatile WalSnd *walsnd = MyWalSnd;
+
+		SpinLockAcquire(&walsnd->mutex);
+		if (XLByteLT(walsnd->write, reply.write))
+			walsnd->write = reply.write;
+		if (XLByteLT(walsnd->flush, reply.flush))
+			walsnd->flush = reply.flush;
+		if (XLByteLT(walsnd->apply, reply.apply))
+			walsnd->apply = reply.apply;
+		SpinLockRelease(&walsnd->mutex);
+	}
+}
+
 /* Main loop of walsender process */
 static int
 WalSndLoop(void)
@@ -518,6 +576,7 @@ WalSndLoop(void)
 		{
 			if (!XLogSend(output_message, &caughtup))
 				break;
+			ProcessRepliesIfAny();
 			if (caughtup)
 				walsender_shutdown_requested = true;
 		}
@@ -561,9 +620,6 @@ WalSndLoop(void)
 				WaitLatchOrSocket(&MyWalSnd->latch, MyProcPort->sock,
 								  WalSndDelay * 1000L);
 			}
-
-			/* Check if the connection was closed */
-			CheckClosedConnection();
 		}
 		else
 		{
@@ -574,6 +630,7 @@ WalSndLoop(void)
 
 		/* Update our state to indicate if we're behind or not */
 		WalSndSetState(caughtup ? WALSNDSTATE_STREAMING : WALSNDSTATE_CATCHUP);
+		ProcessRepliesIfAny();
 	}
 
 	/*
@@ -1104,7 +1161,7 @@ WalSndGetStateString(WalSndState state)
 Datum
 pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
 {
-#define PG_STAT_GET_WAL_SENDERS_COLS 	3
+#define PG_STAT_GET_WAL_SENDERS_COLS 	6
 	ReturnSetInfo	   *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
 	TupleDesc			tupdesc;
 	Tuplestorestate	   *tupstore;
@@ -1141,8 +1198,11 @@ pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
 	{
 		/* use volatile pointer to prevent code rearrangement */
 		volatile WalSnd *walsnd = &WalSndCtl->walsnds[i];
-		char		sent_location[MAXFNAMELEN];
+		char		location[MAXFNAMELEN];
 		XLogRecPtr	sentPtr;
+		XLogRecPtr	write;
+		XLogRecPtr	flush;
+		XLogRecPtr	apply;
 		WalSndState	state;
 		Datum		values[PG_STAT_GET_WAL_SENDERS_COLS];
 		bool		nulls[PG_STAT_GET_WAL_SENDERS_COLS];
@@ -1153,13 +1213,14 @@ pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
 		SpinLockAcquire(&walsnd->mutex);
 		sentPtr = walsnd->sentPtr;
 		state = walsnd->state;
+		write = walsnd->write;
+		flush = walsnd->flush;
+		apply = walsnd->apply;
 		SpinLockRelease(&walsnd->mutex);
 
-		snprintf(sent_location, sizeof(sent_location), "%X/%X",
-					sentPtr.xlogid, sentPtr.xrecoff);
-
 		memset(nulls, 0, sizeof(nulls));
 		values[0] = Int32GetDatum(walsnd->pid);
+
 		if (!superuser())
 		{
 			/*
@@ -1168,11 +1229,35 @@ pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
 			 */
 			nulls[1] = true;
 			nulls[2] = true;
+			nulls[3] = true;
+			nulls[4] = true;
+			nulls[5] = true;
 		}
 		else
 		{
 			values[1] = CStringGetTextDatum(WalSndGetStateString(state));
-			values[2] = CStringGetTextDatum(sent_location);
+
+			snprintf(location, sizeof(location), "%X/%X",
+					 sentPtr.xlogid, sentPtr.xrecoff);
+			values[2] = CStringGetTextDatum(location);
+
+			if (write.xlogid == 0 && write.xrecoff == 0)
+				nulls[4] = true;
+			snprintf(location, sizeof(location), "%X/%X",
+					 write.xlogid, write.xrecoff);
+			values[3] = CStringGetTextDatum(location);
+
+			if (flush.xlogid == 0 && flush.xrecoff == 0)
+				nulls[5] = true;
+			snprintf(location, sizeof(location), "%X/%X",
+					flush.xlogid, flush.xrecoff);
+			values[4] = CStringGetTextDatum(location);
+
+			if (apply.xlogid == 0 && apply.xrecoff == 0)
+				nulls[6] = true;
+			snprintf(location, sizeof(location), "%X/%X",
+					 apply.xlogid, apply.xrecoff);
+			values[5] = CStringGetTextDatum(location);
 		}
 
 		tuplestore_putvalues(tupstore, tupdesc, values, nulls);
diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c
index 216236b..5ede280 100644
--- a/src/backend/utils/misc/guc.c
+++ b/src/backend/utils/misc/guc.c
@@ -55,6 +55,7 @@
 #include "postmaster/postmaster.h"
 #include "postmaster/syslogger.h"
 #include "postmaster/walwriter.h"
+#include "replication/walreceiver.h"
 #include "replication/walsender.h"
 #include "storage/bufmgr.h"
 #include "storage/standby.h"
@@ -1755,6 +1756,16 @@ static struct config_int ConfigureNamesInt[] =
 	},
 
 	{
+		{"wal_receiver_status_interval", PGC_SIGHUP, WAL_STANDBY_SERVERS,
+			gettext_noop("Sets the maximum interval between WAL receiver status reports to the master."),
+			NULL,
+			GUC_UNIT_S
+		},
+		&wal_receiver_status_interval,
+		10, 0, INT_MAX/1000, NULL, NULL
+	},
+
+	{
 		{"checkpoint_segments", PGC_SIGHUP, WAL_CHECKPOINTS,
 			gettext_noop("Sets the maximum distance in log segments between automatic WAL checkpoints."),
 			NULL
diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample
index fe80c4d..1b02aa0 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -194,6 +194,7 @@
 
 # - Standby Servers -
 
+#wal_receiver_status_interval = 10s	# replies at least this often, 0 disables
 #hot_standby = off			# "on" allows queries during recovery
 					# (change requires restart)
 #max_standby_archive_delay = 30s	# max delay before canceling queries
diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h
index 122e96b..352b9a4 100644
--- a/src/include/access/xlog.h
+++ b/src/include/access/xlog.h
@@ -290,6 +290,7 @@ extern void issue_xlog_fsync(int fd, uint32 log, uint32 seg);
 extern bool RecoveryInProgress(void);
 extern bool XLogInsertAllowed(void);
 extern void GetXLogReceiptTime(TimestampTz *rtime, bool *fromStream);
+extern XLogRecPtr GetXLogReplayRecPtr(void);
 
 extern void UpdateControlFile(void);
 extern uint64 GetSystemIdentifier(void);
diff --git a/src/include/catalog/pg_proc.h b/src/include/catalog/pg_proc.h
index f8b5d4d..f842a50 100644
--- a/src/include/catalog/pg_proc.h
+++ b/src/include/catalog/pg_proc.h
@@ -3075,7 +3075,7 @@ DATA(insert OID = 1936 (  pg_stat_get_backend_idset		PGNSP PGUID 12 1 100 0 f f
 DESCR("statistics: currently active backend IDs");
 DATA(insert OID = 2022 (  pg_stat_get_activity			PGNSP PGUID 12 1 100 0 f f f f t s 1 0 2249 "23" "{23,26,23,26,25,25,16,1184,1184,1184,869,23}" "{i,o,o,o,o,o,o,o,o,o,o,o}" "{pid,datid,procpid,usesysid,application_name,current_query,waiting,xact_start,query_start,backend_start,client_addr,client_port}" _null_ pg_stat_get_activity _null_ _null_ _null_ ));
 DESCR("statistics: information about currently active backends");
-DATA(insert OID = 3099 (  pg_stat_get_wal_senders	PGNSP PGUID 12 1 10 0 f f f f t s 0 0 2249 "" "{23,25,25}" "{o,o,o}" "{procpid,state,sent_location}" _null_ pg_stat_get_wal_senders _null_ _null_ _null_ ));
+DATA(insert OID = 3099 (  pg_stat_get_wal_senders	PGNSP PGUID 12 1 10 0 f f f f t s 0 0 2249 "" "{23,25,25,25,25,25}" "{o,o,o,o,o,o}" "{procpid,state,sent_location,write_location,flush_location,apply_location}" _null_ pg_stat_get_wal_senders _null_ _null_ _null_ ));
 DESCR("statistics: information about currently active replication");
 DATA(insert OID = 2026 (  pg_backend_pid				PGNSP PGUID 12 1 0 0 f f f t f s 0 0 23 "" _null_ _null_ _null_ _null_ pg_backend_pid _null_ _null_ _null_ ));
 DESCR("statistics: current backend PID");
diff --git a/src/include/replication/walprotocol.h b/src/include/replication/walprotocol.h
index 1993851..c69ca9d 100644
--- a/src/include/replication/walprotocol.h
+++ b/src/include/replication/walprotocol.h
@@ -40,6 +40,40 @@ typedef struct
 } WalDataMessageHeader;
 
 /*
+ * Reply message from standby (message type 'r').  This is wrapped within
+ * a CopyData message at the FE/BE protocol level.
+ *
+ * Note that the data length is not specified here.
+ */
+typedef struct
+{
+	/*
+	 * The xlog location that has been written to WAL file by standby-side.
+	 * This may be invalid if the standby-side does not choose to offer
+	 * a synchronous replication reply service, or is unable to offer
+	 * a valid reply for data that has only been written, not fsynced.
+	 */
+	XLogRecPtr	write;
+
+	/*
+	 * The xlog location that has been fsynced onto disk by standby-side.
+	 * This may be invalid if the standby-side does not choose to offer
+	 * a synchronous replication reply service, or is unable to.
+	 */
+	XLogRecPtr	flush;
+
+	/*
+	 * The xlog location that has been applied by standby Startup process.
+	 * This may be invalid if the standby-side does not support apply,
+	 * or does not choose to apply records, as yet.
+	 */
+	XLogRecPtr	apply;
+
+	/* Sender's system clock at the time of transmission */
+	TimestampTz sendTime;
+} StandbyReplyMessage;
+
+/*
  * 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
diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h
index 24ad438..aa5bfb7 100644
--- a/src/include/replication/walreceiver.h
+++ b/src/include/replication/walreceiver.h
@@ -17,6 +17,7 @@
 #include "pgtime.h"
 
 extern bool am_walreceiver;
+extern int wal_receiver_status_interval;
 
 /*
  * MAXCONNINFO: maximum size of a connection string.
diff --git a/src/include/replication/walsender.h b/src/include/replication/walsender.h
index 9a196ab..abee380 100644
--- a/src/include/replication/walsender.h
+++ b/src/include/replication/walsender.h
@@ -35,13 +35,34 @@ typedef struct WalSnd
 	WalSndState state;			/* this walsender's state */
 	XLogRecPtr	sentPtr;		/* WAL has been sent up to this point */
 
-	slock_t		mutex;			/* locks shared variables shown above */
+	/*
+	 * The xlog location that has been written to WAL file by standby-side.
+	 * This may be invalid if the standby-side has not offered a value yet.
+	 */
+	XLogRecPtr	write;
+
+	/*
+	 * The xlog location that has been fsynced onto disk by standby-side.
+	 * This may be invalid if the standby-side has not offered a value yet.
+	 */
+	XLogRecPtr	flush;
+
+	/*
+	 * The xlog location that has been applied by standby Startup process.
+	 * This may be invalid if the standby-side has not offered a value yet.
+	 */
+	XLogRecPtr	apply;
 
 	/*
 	 * Latch used by backends to wake up this walsender when it has work
 	 * to do.
 	 */
 	Latch		latch;
+
+	/*
+	 * Locks shared variables shown above.
+	 */
+	slock_t         mutex;
 } WalSnd;
 
 /* There is one WalSndCtl struct for the whole database cluster */
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 72e5630..1dbd1e5 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1296,7 +1296,7 @@ SELECT viewname, definition FROM pg_views WHERE schemaname <> 'information_schem
  pg_stat_bgwriter            | SELECT pg_stat_get_bgwriter_timed_checkpoints() AS checkpoints_timed, pg_stat_get_bgwriter_requested_checkpoints() AS checkpoints_req, pg_stat_get_bgwriter_buf_written_checkpoints() AS buffers_checkpoint, pg_stat_get_bgwriter_buf_written_clean() AS buffers_clean, pg_stat_get_bgwriter_maxwritten_clean() AS maxwritten_clean, pg_stat_get_buf_written_backend() AS buffers_backend, pg_stat_get_buf_fsync_backend() AS buffers_backend_fsync, pg_stat_get_buf_alloc() AS buffers_alloc;
  pg_stat_database            | SELECT d.oid AS datid, d.datname, pg_stat_get_db_numbackends(d.oid) AS numbackends, pg_stat_get_db_xact_commit(d.oid) AS xact_commit, pg_stat_get_db_xact_rollback(d.oid) AS xact_rollback, (pg_stat_get_db_blocks_fetched(d.oid) - pg_stat_get_db_blocks_hit(d.oid)) AS blks_read, pg_stat_get_db_blocks_hit(d.oid) AS blks_hit, pg_stat_get_db_tuples_returned(d.oid) AS tup_returned, pg_stat_get_db_tuples_fetched(d.oid) AS tup_fetched, pg_stat_get_db_tuples_inserted(d.oid) AS tup_inserted, pg_stat_get_db_tuples_updated(d.oid) AS tup_updated, pg_stat_get_db_tuples_deleted(d.oid) AS tup_deleted, pg_stat_get_db_conflict_all(d.oid) AS conflicts FROM pg_database d;
  pg_stat_database_conflicts  | SELECT d.oid AS datid, d.datname, pg_stat_get_db_conflict_tablespace(d.oid) AS confl_tablespace, pg_stat_get_db_conflict_lock(d.oid) AS confl_lock, pg_stat_get_db_conflict_snapshot(d.oid) AS confl_snapshot, pg_stat_get_db_conflict_bufferpin(d.oid) AS confl_bufferpin, pg_stat_get_db_conflict_startup_deadlock(d.oid) AS confl_deadlock FROM pg_database d;
- pg_stat_replication         | SELECT s.procpid, s.usesysid, u.rolname AS usename, s.application_name, s.client_addr, s.client_port, s.backend_start, w.state, w.sent_location FROM pg_stat_get_activity(NULL::integer) s(datid, procpid, usesysid, application_name, current_query, waiting, xact_start, query_start, backend_start, client_addr, client_port), pg_authid u, pg_stat_get_wal_senders() w(procpid, state, sent_location) WHERE ((s.usesysid = u.oid) AND (s.procpid = w.procpid));
+ pg_stat_replication         | SELECT s.procpid, s.usesysid, u.rolname AS usename, s.application_name, s.client_addr, s.client_port, s.backend_start, w.state, w.sent_location, w.write_location, w.flush_location, w.apply_location FROM pg_stat_get_activity(NULL::integer) s(datid, procpid, usesysid, application_name, current_query, waiting, xact_start, query_start, backend_start, client_addr, client_port), pg_authid u, pg_stat_get_wal_senders() w(procpid, state, sent_location, write_location, flush_location, apply_location) WHERE ((s.usesysid = u.oid) AND (s.procpid = w.procpid));
  pg_stat_sys_indexes         | SELECT pg_stat_all_indexes.relid, pg_stat_all_indexes.indexrelid, pg_stat_all_indexes.schemaname, pg_stat_all_indexes.relname, pg_stat_all_indexes.indexrelname, pg_stat_all_indexes.idx_scan, pg_stat_all_indexes.idx_tup_read, pg_stat_all_indexes.idx_tup_fetch FROM pg_stat_all_indexes WHERE ((pg_stat_all_indexes.schemaname = ANY (ARRAY['pg_catalog'::name, 'information_schema'::name])) OR (pg_stat_all_indexes.schemaname ~ '^pg_toast'::text));
  pg_stat_sys_tables          | SELECT pg_stat_all_tables.relid, pg_stat_all_tables.schemaname, pg_stat_all_tables.relname, pg_stat_all_tables.seq_scan, pg_stat_all_tables.seq_tup_read, pg_stat_all_tables.idx_scan, pg_stat_all_tables.idx_tup_fetch, pg_stat_all_tables.n_tup_ins, pg_stat_all_tables.n_tup_upd, pg_stat_all_tables.n_tup_del, pg_stat_all_tables.n_tup_hot_upd, pg_stat_all_tables.n_live_tup, pg_stat_all_tables.n_dead_tup, pg_stat_all_tables.last_vacuum, pg_stat_all_tables.last_autovacuum, pg_stat_all_tables.last_analyze, pg_stat_all_tables.last_autoanalyze, pg_stat_all_tables.vacuum_count, pg_stat_all_tables.autovacuum_count, pg_stat_all_tables.analyze_count, pg_stat_all_tables.autoanalyze_count FROM pg_stat_all_tables WHERE ((pg_stat_all_tables.schemaname = ANY (ARRAY['pg_catalog'::name, 'information_schema'::name])) OR (pg_stat_all_tables.schemaname ~ '^pg_toast'::text));
  pg_stat_user_functions      | SELECT p.oid AS funcid, n.nspname AS schemaname, p.proname AS funcname, pg_stat_get_function_calls(p.oid) AS calls, (pg_stat_get_function_time(p.oid) / 1000) AS total_time, (pg_stat_get_function_self_time(p.oid) / 1000) AS self_time FROM (pg_proc p LEFT JOIN pg_namespace n ON ((n.oid = p.pronamespace))) WHERE ((p.prolang <> (12)::oid) AND (pg_stat_get_function_calls(p.oid) IS NOT NULL));