server_name_for_replication.v1.patch

diff --git a/src/backend/replication/walreceiver.c b/src/backend/replication/walreceiver.c
index ee09468..ff89035 100644
--- a/src/backend/replication/walreceiver.c
+++ b/src/backend/replication/walreceiver.c
@@ -95,6 +95,7 @@ static struct
 }	LogstreamResult;
 
 static StandbyReplyMessage	reply_message;
+static StandbyHSFeedbackMessage	feedback_message;
 
 /*
  * About SIGTERM handling:
@@ -123,6 +124,8 @@ static void XLogWalRcvProcessMsg(unsigned char type, char *buf, Size len);
 static void XLogWalRcvWrite(char *buf, Size nbytes, XLogRecPtr recptr);
 static void XLogWalRcvFlush(bool dying);
 static void XLogWalRcvSendReply(void);
+static void XLogWalRcvSendHSFeedback(void);
+static void XLogWalRcvSendInfo(void);
 
 /* Signal handlers */
 static void WalRcvSigHupHandler(SIGNAL_ARGS);
@@ -303,6 +306,7 @@ WalReceiverMain(void)
 		{
 			got_SIGHUP = false;
 			ProcessConfigFile(PGC_SIGHUP);
+			XLogWalRcvSendInfo();
 		}
 
 		/* Wait a while for data to arrive */
@@ -317,6 +321,7 @@ WalReceiverMain(void)
 
 			/* Let the master know that we received some data. */
 			XLogWalRcvSendReply();
+			XLogWalRcvSendHSFeedback();
 
 			/*
 			 * If we've written some records, flush them to disk and let the
@@ -331,6 +336,7 @@ WalReceiverMain(void)
 			 * the master anyway, to report any progress in applying WAL.
 			 */
 			XLogWalRcvSendReply();
+			XLogWalRcvSendHSFeedback();
 		}
 	}
 }
@@ -619,40 +625,84 @@ XLogWalRcvSendReply(void)
 	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);
+
+	/* 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 hot standby feedback message to primary, plus the current time,
+ * in case they don't have a watch.
+ */
+static void
+XLogWalRcvSendHSFeedback(void)
+{
+	char		buf[sizeof(StandbyHSFeedbackMessage) + 1];
+	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 (!hot_standby_feedback || !HotStandbyActive())
+		return;
+
+	/* Get current timestamp. */
+	now = GetCurrentTimestamp();
+
 	/*
 	 * Get the OldestXmin and its associated epoch
 	 */
-	if (hot_standby_feedback && HotStandbyActive())
 	{
 		TransactionId	nextXid;
 		uint32			nextEpoch;
 
-		reply_message.xmin = GetOldestXmin(true, false);
+		feedback_message.xmin = GetOldestXmin(true, false);
 
 		/*
 		 * Get epoch and adjust if nextXid and oldestXmin are different
 		 * sides of the epoch boundary.
 		 */
 		GetNextXidAndEpoch(&nextXid, &nextEpoch);
-		if (nextXid < reply_message.xmin)
+		if (nextXid < feedback_message.xmin)
 			nextEpoch--;
-		reply_message.epoch = nextEpoch;
-	}
-	else
-	{
-		reply_message.xmin = InvalidTransactionId;
-		reply_message.epoch = 0;
+		feedback_message.epoch = nextEpoch;
 	}
 
-	elog(DEBUG2, "sending write %X/%X flush %X/%X apply %X/%X xmin %u epoch %u",
-				 reply_message.write.xlogid, reply_message.write.xrecoff,
-				 reply_message.flush.xlogid, reply_message.flush.xrecoff,
-				 reply_message.apply.xlogid, reply_message.apply.xrecoff,
-				 reply_message.xmin,
-				 reply_message.epoch);
+	elog(DEBUG2, "sending xmin %u epoch %u",
+				 feedback_message.xmin,
+				 feedback_message.epoch);
 
 	/* Prepend with the message type and send it. */
-	buf[0] = 'r';
-	memcpy(&buf[1], &reply_message, sizeof(StandbyReplyMessage));
-	walrcv_send(buf, sizeof(StandbyReplyMessage) + 1);
+	buf[0] = 'h';
+	memcpy(&buf[1], &feedback_message, sizeof(StandbyHSFeedbackMessage));
+	walrcv_send(buf, sizeof(StandbyHSFeedbackMessage) + 1);
+}
+
+/*
+ * Send info message to primary.
+ */
+static void
+XLogWalRcvSendInfo(void)
+{
+	char		buf[sizeof(StandbyInfoMessage) + 1];
+	StandbyInfoMessage	info_message;
+
+	/* Get current timestamp. */
+	info_message.sendTime = GetCurrentTimestamp();
+	strncpy(info_message.servername, ServerName, strlen(ServerName));
+
+	elog(DEBUG2, "sending servername %s",
+				 info_message.servername);
+
+	/* Prepend with the message type and send it. */
+	buf[0] = 'i';
+	memcpy(&buf[1], &info_message, sizeof(StandbyInfoMessage));
+	walrcv_send(buf, sizeof(StandbyInfoMessage) + 1);
 }
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index a6a7a14..e46cd01 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -116,7 +116,10 @@ static void WalSndKill(int code, Datum arg);
 static bool XLogSend(char *msgbuf, bool *caughtup);
 static void IdentifySystem(void);
 static void StartReplication(StartReplicationCmd * cmd);
+static void ProcessStandbyMessage(void);
 static void ProcessStandbyReplyMessage(void);
+static void ProcessStandbyHSFeedbackMessage(void);
+static void ProcessStandbyInfoMessage(void);
 static void ProcessRepliesIfAny(void);
 
 
@@ -456,42 +459,45 @@ ProcessRepliesIfAny(void)
 	unsigned char firstchar;
 	int			r;
 
-	r = pq_getbyte_if_available(&firstchar);
-	if (r < 0)
-	{
-		/* unexpected error or EOF */
-		ereport(COMMERROR,
-				(errcode(ERRCODE_PROTOCOL_VIOLATION),
-				 errmsg("unexpected EOF on standby connection")));
-		proc_exit(0);
-	}
-	if (r == 0)
+	for (;;)
 	{
-		/* no data available without blocking */
-		return;
-	}
+		r = pq_getbyte_if_available(&firstchar);
+		if (r < 0)
+		{
+			/* unexpected error or EOF */
+			ereport(COMMERROR,
+					(errcode(ERRCODE_PROTOCOL_VIOLATION),
+					 errmsg("unexpected EOF on standby connection")));
+			proc_exit(0);
+		}
+		if (r == 0)
+		{
+			/* no data available without blocking */
+			return;
+		}
 
-	/* Handle the very limited subset of commands expected in this phase */
-	switch (firstchar)
-	{
-			/*
-			 * 'd' means a standby reply wrapped in a CopyData packet.
-			 */
-		case 'd':
-			ProcessStandbyReplyMessage();
-			break;
+		/* Handle the very limited subset of commands expected in this phase */
+		switch (firstchar)
+		{
+				/*
+				 * 'd' means a standby reply wrapped in a CopyData packet.
+				 */
+			case 'd':
+				ProcessStandbyMessage();
+				break;
 
-			/*
-			 * 'X' means that the standby is closing down the socket.
-			 */
-		case 'X':
-			proc_exit(0);
+				/*
+				 * 'X' means that the standby is closing down the socket.
+				 */
+			case 'X':
+				proc_exit(0);
 
-		default:
-			ereport(FATAL,
-					(errcode(ERRCODE_PROTOCOL_VIOLATION),
-					 errmsg("invalid standby closing message type %d",
-							firstchar)));
+			default:
+				ereport(FATAL,
+						(errcode(ERRCODE_PROTOCOL_VIOLATION),
+						 errmsg("invalid standby closing message type %d",
+								firstchar)));
+		}
 	}
 }
 
@@ -499,11 +505,9 @@ ProcessRepliesIfAny(void)
  * Process a status update message received from standby.
  */
 static void
-ProcessStandbyReplyMessage(void)
+ProcessStandbyMessage(void)
 {
-	StandbyReplyMessage	reply;
 	char msgtype;
-	TransactionId newxmin = InvalidTransactionId;
 
 	resetStringInfo(&reply_message);
 
@@ -523,22 +527,43 @@ ProcessStandbyReplyMessage(void)
 	 * one type.
 	 */
 	msgtype = pq_getmsgbyte(&reply_message);
-	if (msgtype != 'r')
+
+	switch (msgtype)
 	{
-		ereport(COMMERROR,
-				(errcode(ERRCODE_PROTOCOL_VIOLATION),
-				 errmsg("unexpected message type %c", msgtype)));
-		proc_exit(0);
+		case 'r':
+			ProcessStandbyReplyMessage();
+			break;
+
+		case 'h':
+			ProcessStandbyHSFeedbackMessage();
+			break;
+
+		case 'i':
+			ProcessStandbyInfoMessage();
+			break;
+
+		default:
+			ereport(COMMERROR,
+					(errcode(ERRCODE_PROTOCOL_VIOLATION),
+					 errmsg("unexpected message type %c", msgtype)));
+			proc_exit(0);
 	}
+}
+
+/*
+ * Regular reply from standby advising of WAL positions on standby server.
+ */
+static void
+ProcessStandbyReplyMessage(void)
+{
+	StandbyReplyMessage	reply;
 
 	pq_copymsgbytes(&reply_message, (char *) &reply, sizeof(StandbyReplyMessage));
 
-	elog(DEBUG2, "write %X/%X flush %X/%X apply %X/%X xmin %u epoch %u",
+	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,
-		 reply.xmin,
-		 reply.epoch);
+		 reply.apply.xlogid, reply.apply.xrecoff);
 
 	/*
 	 * Update shared state for this WalSender process
@@ -554,6 +579,22 @@ ProcessStandbyReplyMessage(void)
 		walsnd->apply = reply.apply;
 		SpinLockRelease(&walsnd->mutex);
 	}
+}
+
+/*
+ * Hot Standby feedback
+ */
+static void
+ProcessStandbyHSFeedbackMessage(void)
+{
+	StandbyHSFeedbackMessage	reply;
+	TransactionId newxmin = InvalidTransactionId;
+
+	pq_copymsgbytes(&reply_message, (char *) &reply, sizeof(StandbyHSFeedbackMessage));
+
+	elog(DEBUG2, "xmin %u epoch %u",
+		 reply.xmin,
+		 reply.epoch);
 
 	/*
 	 * Update the WalSender's proc xmin to allow it to be visible
@@ -619,6 +660,16 @@ ProcessStandbyReplyMessage(void)
 	}
 }
 
+static void
+ProcessStandbyInfoMessage(void)
+{
+	StandbyInfoMessage	info;
+
+	pq_copymsgbytes(&reply_message, (char *) &info, sizeof(StandbyInfoMessage));
+
+	elog(DEBUG2, "server name %s", info.servername);
+}
+
 /* Main loop of walsender process */
 static int
 WalSndLoop(void)
diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c
index 55cbf75..de6de82 100644
--- a/src/backend/utils/misc/guc.c
+++ b/src/backend/utils/misc/guc.c
@@ -405,6 +405,9 @@ int			tcp_keepalives_idle;
 int			tcp_keepalives_interval;
 int			tcp_keepalives_count;
 
+char		*ServerName = NULL;
+
+
 /*
  * These variables are all dummies that don't do anything, except in some
  * cases provide the value for SHOW to display.  The real state is elsewhere
@@ -2365,6 +2368,15 @@ static struct config_string ConfigureNamesString[] =
 	},
 
 	{
+		{"server_name", PGC_POSTMASTER, CLIENT_CONN_STATEMENT,
+			gettext_noop("Allows setting of a unique name for this server."),
+			NULL
+		},
+		&ServerName,
+		"", NULL, NULL
+	},
+
+	{
 		{"temp_tablespaces", PGC_USERSET, CLIENT_CONN_STATEMENT,
 			gettext_noop("Sets the tablespace(s) to use for temporary tables and sort files."),
 			NULL,
diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample
index 6726733..cbe6fb2 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -56,6 +56,7 @@
 
 # - Connection Settings -
 
+#server_name = ''	# optional server name for use when clustering servers
 #listen_addresses = 'localhost'		# what IP address(es) to listen on;
 					# comma-separated list of addresses;
 					# defaults to 'localhost', '*' = all
diff --git a/src/include/miscadmin.h b/src/include/miscadmin.h
index aa8cce5..bf6c262 100644
--- a/src/include/miscadmin.h
+++ b/src/include/miscadmin.h
@@ -218,6 +218,7 @@ extern int	CTimeZone;
 
 #define MAXTZLEN		10		/* max TZ name len, not counting tr. null */
 
+extern char	*ServerName;
 extern bool enableFsync;
 extern bool allowSystemTableMods;
 extern PGDLLIMPORT int work_mem;
diff --git a/src/include/replication/walprotocol.h b/src/include/replication/walprotocol.h
index da94b6b..48cb333 100644
--- a/src/include/replication/walprotocol.h
+++ b/src/include/replication/walprotocol.h
@@ -56,6 +56,18 @@ typedef struct
 	XLogRecPtr	flush;
 	XLogRecPtr	apply;
 
+	/* Sender's system clock at the time of transmission */
+	TimestampTz sendTime;
+} StandbyReplyMessage;
+
+/*
+ * Hot Standby feedback from standby (message type 'h').  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 current xmin and epoch from the standby, for Hot Standby feedback.
 	 * This may be invalid if the standby-side does not support feedback,
@@ -64,10 +76,23 @@ typedef struct
 	TransactionId	xmin;
 	uint32			epoch;
 
+	/* Sender's system clock at the time of transmission */
+	TimestampTz sendTime;
+} StandbyHSFeedbackMessage;
+
+/*
+ * Info message from standby (message type 'i').  This is wrapped within
+ * a CopyData message at the FE/BE protocol level.
+ *
+ * Note that the data length is not specified here.
+ */
+typedef struct
+{
+	char		servername[64];
 
 	/* Sender's system clock at the time of transmission */
 	TimestampTz sendTime;
-} StandbyReplyMessage;
+} StandbyInfoMessage;
 
 /*
  * Maximum data payload in a WAL data message.	Must be >= XLOG_BLCKSZ.