synchrep_0910.patch

*** a/doc/src/sgml/protocol.sgml
--- b/doc/src/sgml/protocol.sgml
***************
*** 1360,1365 **** The commands accepted in walsender mode are:
--- 1360,1400 ----
        <variablelist>
        <varlistentry>
        <term>
+           XLogRecPtr (F)
+       </term>
+       <listitem>
+       <para>
+       <variablelist>
+       <varlistentry>
+       <term>
+           Byte1('l')
+       </term>
+       <listitem>
+       <para>
+           Identifies the message as an acknowledgment of replication.
+       </para>
+       </listitem>
+       </varlistentry>
+       <varlistentry>
+       <term>
+           Byte8
+       </term>
+       <listitem>
+       <para>
+           The end of the WAL data replicated to the standby, given in
+           XLogRecPtr format.
+       </para>
+       </listitem>
+       </varlistentry>
+       </variablelist>
+       </para>
+       </listitem>
+       </varlistentry>
+       </variablelist>
+ 
+       <variablelist>
+       <varlistentry>
+       <term>
            XLogData (B)
        </term>
        <listitem>
*** a/src/backend/Makefile
--- b/src/backend/Makefile
***************
*** 208,213 **** endif
--- 208,214 ----
  	$(INSTALL_DATA) $(srcdir)/libpq/pg_ident.conf.sample '$(DESTDIR)$(datadir)/pg_ident.conf.sample'
  	$(INSTALL_DATA) $(srcdir)/utils/misc/postgresql.conf.sample '$(DESTDIR)$(datadir)/postgresql.conf.sample'
  	$(INSTALL_DATA) $(srcdir)/access/transam/recovery.conf.sample '$(DESTDIR)$(datadir)/recovery.conf.sample'
+ 	$(INSTALL_DATA) $(srcdir)/replication/standbys.conf.sample '$(DESTDIR)$(datadir)/standbys.conf.sample'
  
  install-bin: postgres $(POSTGRES_IMP) installdirs
  	$(INSTALL_PROGRAM) postgres$(X) '$(DESTDIR)$(bindir)/postgres$(X)'
***************
*** 262,268 **** endif
  	rm -f '$(DESTDIR)$(datadir)/pg_hba.conf.sample' \
  	      '$(DESTDIR)$(datadir)/pg_ident.conf.sample' \
                '$(DESTDIR)$(datadir)/postgresql.conf.sample' \
! 	      '$(DESTDIR)$(datadir)/recovery.conf.sample'
  
  
  ##########################################################################
--- 263,270 ----
  	rm -f '$(DESTDIR)$(datadir)/pg_hba.conf.sample' \
  	      '$(DESTDIR)$(datadir)/pg_ident.conf.sample' \
                '$(DESTDIR)$(datadir)/postgresql.conf.sample' \
! 	      '$(DESTDIR)$(datadir)/recovery.conf.sample' \
! 	      '$(DESTDIR)$(datadir)/standbys.conf.sample'
  
  
  ##########################################################################
*** a/src/backend/access/transam/recovery.conf.sample
--- b/src/backend/access/transam/recovery.conf.sample
***************
*** 91,102 ****
  #---------------------------------------------------------------------------
  #
  # When standby_mode is enabled, the PostgreSQL server will work as
! # a standby. It tries to connect to the primary according to the
! # connection settings primary_conninfo, and receives XLOG records
! # continuously.
  #
  #standby_mode = 'off'
  #
  #primary_conninfo = ''		# e.g. 'host=localhost port=5432'
  #
  #
--- 91,104 ----
  #---------------------------------------------------------------------------
  #
  # When standby_mode is enabled, the PostgreSQL server will work as
! # a standby under the name of standby_name. It tries to connect to
! # the primary according to the connection settings primary_conninfo,
! # and receives XLOG records continuously.
  #
  #standby_mode = 'off'
  #
+ #standby_name = ''
+ #
  #primary_conninfo = ''		# e.g. 'host=localhost port=5432'
  #
  #
*** a/src/backend/access/transam/twophase.c
--- b/src/backend/access/transam/twophase.c
***************
*** 55,60 ****
--- 55,61 ----
  #include "miscadmin.h"
  #include "pg_trace.h"
  #include "pgstat.h"
+ #include "replication/walsender.h"
  #include "storage/fd.h"
  #include "storage/procarray.h"
  #include "storage/sinvaladt.h"
***************
*** 1062,1067 **** EndPrepare(GlobalTransaction gxact)
--- 1063,1080 ----
  
  	END_CRIT_SECTION();
  
+ 	/*
+ 	 * Wait for WAL to be replicated up to the PREPARE record
+ 	 * if replication is enabled. This operation has to be performed
+ 	 * after the PREPARE record is generated and before other
+ 	 * transactions know that this one has already been prepared.
+ 	 *
+ 	 * XXX: Since the caller prevents cancel/die interrupt, we cannot
+ 	 * process that while waiting. Should we remove this restriction?
+ 	 */
+ 	if (max_wal_senders > 0)
+ 		WaitXLogSend(gxact->prepare_lsn);
+ 
  	records.tail = records.head = NULL;
  }
  
***************
*** 2012,2017 **** RecordTransactionCommitPrepared(TransactionId xid,
--- 2025,2039 ----
  	MyProc->inCommit = false;
  
  	END_CRIT_SECTION();
+ 
+ 	/*
+ 	 * Wait for WAL to be replicated up to the COMMIT PREPARED record
+ 	 * if replication is enabled. This operation has to be performed
+ 	 * after the COMMIT PREPARED record is generated and before other
+ 	 * transactions know that this one has already been committed.
+ 	 */
+ 	if (max_wal_senders > 0)
+ 		WaitXLogSend(recptr);
  }
  
  /*
***************
*** 2084,2087 **** RecordTransactionAbortPrepared(TransactionId xid,
--- 2106,2118 ----
  	TransactionIdAbortTree(xid, nchildren, children);
  
  	END_CRIT_SECTION();
+ 
+ 	/*
+ 	 * Wait for WAL to be replicated up to the ABORT PREPARED record
+ 	 * if replication is enabled. This operation has to be performed
+ 	 * after the ABORT PREPARED record is generated and before other
+ 	 * transactions know that this one has already been aborted.
+ 	 */
+ 	if (max_wal_senders > 0)
+ 		WaitXLogSend(recptr);
  }
*** a/src/backend/access/transam/xact.c
--- b/src/backend/access/transam/xact.c
***************
*** 36,41 ****
--- 36,42 ----
  #include "libpq/be-fsstubs.h"
  #include "miscadmin.h"
  #include "pgstat.h"
+ #include "replication/walsender.h"
  #include "storage/bufmgr.h"
  #include "storage/fd.h"
  #include "storage/lmgr.h"
***************
*** 1110,1115 **** RecordTransactionCommit(void)
--- 1111,1128 ----
  	/* Compute latestXid while we have the child XIDs handy */
  	latestXid = TransactionIdLatest(xid, nchildren, children);
  
+ 	/*
+ 	 * Wait for WAL to be replicated up to the COMMIT record if replication
+ 	 * is enabled. This operation has to be performed after the COMMIT record
+ 	 * is generated and before other transactions know that this one has
+ 	 * already been committed.
+ 	 *
+ 	 * XXX: Since the caller prevents cancel/die interrupt, we cannot
+ 	 * process that while waiting. Should we remove this restriction?
+ 	 */
+ 	if (max_wal_senders > 0)
+ 		WaitXLogSend(XactLastRecEnd);
+ 
  	/* Reset XactLastRecEnd until the next transaction writes something */
  	XactLastRecEnd.xrecoff = 0;
  
*** a/src/backend/access/transam/xlog.c
--- b/src/backend/access/transam/xlog.c
***************
*** 189,194 **** static TimestampTz recoveryTargetTime;
--- 189,195 ----
  static bool StandbyMode = false;
  static char *PrimaryConnInfo = NULL;
  static char *TriggerFile = NULL;
+ static char *StandbyName = NULL;
  
  /* if recoveryStopsHere returns true, it saves actual stop xid/time here */
  static TransactionId recoveryStopXid;
***************
*** 532,537 **** typedef struct xl_parameter_change
--- 533,548 ----
  	int			wal_level;
  } xl_parameter_change;
  
+ /* Replication mode names */
+ const char *ReplicationModeNames[] = {
+ 	"async",				/* REPLICATION_MODE_ASYNC */
+ 	"recv",				/* REPLICATION_MODE_RECV */
+ 	"fsync",				/* REPLICATION_MODE_FSYNC */
+ 	"replay"				/* REPLICATION_MODE_REPLAY */
+ };
+ 
+ ReplicationMode		rplMode = InvalidReplicationMode;
+ 
  /*
   * Flags set by interrupt handlers for later service in the redo loop.
   */
***************
*** 5258,5263 **** readRecoveryCommandFile(void)
--- 5269,5281 ----
  					(errmsg("trigger_file = '%s'",
  							TriggerFile)));
  		}
+ 		else if (strcmp(tok1, "standby_name") == 0)
+ 		{
+ 			StandbyName = pstrdup(tok2);
+ 			ereport(DEBUG2,
+ 					(errmsg("standby_name = '%s'",
+ 							StandbyName)));
+ 		}
  		else
  			ereport(FATAL,
  					(errmsg("unrecognized recovery parameter \"%s\"",
***************
*** 6867,6872 **** GetFlushRecPtr(void)
--- 6885,6907 ----
  }
  
  /*
+  * GetReplayRecPtr -- Returns the last replay position.
+  */
+ XLogRecPtr
+ GetReplayRecPtr(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;
+ }
+ 
+ /*
   * Get the time of the last xlog segment switch
   */
  pg_time_t
***************
*** 8828,8842 **** 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);
! 
  	if (recptr.xlogid == 0 && recptr.xrecoff == 0)
  		PG_RETURN_NULL();
  
--- 8863,8872 ----
  Datum
  pg_last_xlog_replay_location(PG_FUNCTION_ARGS)
  {
  	XLogRecPtr	recptr;
  	char		location[MAXFNAMELEN];
  
! 	recptr = GetReplayRecPtr();
  	if (recptr.xlogid == 0 && recptr.xrecoff == 0)
  		PG_RETURN_NULL();
  
***************
*** 9467,9473 **** retry:
  						{
  							RequestXLogStreaming(
  									  fetching_ckpt ? RedoStartLSN : *RecPtr,
! 												 PrimaryConnInfo);
  							continue;
  						}
  					}
--- 9497,9503 ----
  						{
  							RequestXLogStreaming(
  									  fetching_ckpt ? RedoStartLSN : *RecPtr,
! 												 PrimaryConnInfo, StandbyName);
  							continue;
  						}
  					}
***************
*** 9681,9683 **** CheckForStandbyTrigger(void)
--- 9711,9727 ----
  	}
  	return false;
  }
+ 
+ /*
+  * Look up replication mode value by name.
+  */
+ ReplicationMode
+ ReplicationModeNameGetValue(char *name)
+ {
+ 	ReplicationMode	mode;
+ 
+ 	for (mode = 0; mode <= MAXREPLICATIONMODE; mode++)
+ 		if (strcmp(ReplicationModeNames[mode], name) == 0)
+ 			return mode;
+ 	return InvalidReplicationMode;
+ }
*** a/src/backend/libpq/hba.c
--- b/src/backend/libpq/hba.c
***************
*** 38,46 ****
  #define atooid(x)  ((Oid) strtoul((x), NULL, 10))
  #define atoxid(x)  ((TransactionId) strtoul((x), NULL, 10))
  
- /* This is used to separate values in multi-valued column strings */
- #define MULTI_VALUE_SEP "\001"
- 
  #define MAX_TOKEN	256
  
  /* callback data for check_network_callback */
--- 38,43 ----
***************
*** 54,59 **** typedef struct check_network_data
--- 51,59 ----
  /* pre-parsed content of HBA config file: list of HbaLine structs */
  static List *parsed_hba_lines = NIL;
  
+ static const char *hba_keywords[] = {"all", "sameuser", "samegroup", "samerole",
+ 									 "replication", NULL};
+ 
  /*
   * These variables hold the pre-parsed contents of the ident usermap
   * configuration file.	ident_lines is a list of sublists, one sublist for
***************
*** 67,76 **** static List *ident_lines = NIL;
  static List *ident_line_nums = NIL;
  
  
- static void tokenize_file(const char *filename, FILE *file,
- 			  List **lines, List **line_nums);
  static char *tokenize_inc_file(const char *outer_filename,
! 				  const char *inc_filename);
  
  /*
   * isblank() exists in the ISO C99 spec, but it's not very portable yet,
--- 67,74 ----
  static List *ident_line_nums = NIL;
  
  
  static char *tokenize_inc_file(const char *outer_filename,
! 				  const char *inc_filename, const char **keywords);
  
  /*
   * isblank() exists in the ISO C99 spec, but it's not very portable yet,
***************
*** 108,114 **** pg_isblank(const char c)
   * token.
   */
  static bool
! next_token(FILE *fp, char *buf, int bufsz, bool *initial_quote)
  {
  	int			c;
  	char	   *start_buf = buf;
--- 106,113 ----
   * token.
   */
  static bool
! next_token(const char *filename, FILE *fp, char *buf, int bufsz,
! 		   bool *initial_quote, const char **keywords)
  {
  	int			c;
  	char	   *start_buf = buf;
***************
*** 155,162 **** next_token(FILE *fp, char *buf, int bufsz, bool *initial_quote)
  			*buf = '\0';
  			ereport(LOG,
  					(errcode(ERRCODE_CONFIG_FILE_ERROR),
! 			   errmsg("authentication file token too long, skipping: \"%s\"",
! 					  start_buf)));
  			/* Discard remainder of line */
  			while ((c = getc(fp)) != EOF && c != '\n')
  				;
--- 154,161 ----
  			*buf = '\0';
  			ereport(LOG,
  					(errcode(ERRCODE_CONFIG_FILE_ERROR),
! 			   errmsg("configuration file \"%s\" token too long, skipping: \"%s\"",
! 					  filename, start_buf)));
  			/* Discard remainder of line */
  			while ((c = getc(fp)) != EOF && c != '\n')
  				;
***************
*** 196,211 **** next_token(FILE *fp, char *buf, int bufsz, bool *initial_quote)
  
  	*buf = '\0';
  
! 	if (!saw_quote &&
! 		(strcmp(start_buf, "all") == 0 ||
! 		 strcmp(start_buf, "sameuser") == 0 ||
! 		 strcmp(start_buf, "samegroup") == 0 ||
! 		 strcmp(start_buf, "samerole") == 0 ||
! 		 strcmp(start_buf, "replication") == 0))
  	{
! 		/* append newline to a magical keyword */
! 		*buf++ = '\n';
! 		*buf = '\0';
  	}
  
  	return (saw_quote || buf > start_buf);
--- 195,214 ----
  
  	*buf = '\0';
  
! 	if (!saw_quote)
  	{
! 		const char	**entry;
! 
! 		for (entry = keywords; *entry != NULL; entry++)
! 		{
! 			if (strcmp(start_buf, *entry) == 0)
! 			{
! 				/* append newline to a magical keyword */
! 				*buf++ = '\n';
! 				*buf = '\0';
! 				break;
! 			}
! 		}
  	}
  
  	return (saw_quote || buf > start_buf);
***************
*** 219,225 **** next_token(FILE *fp, char *buf, int bufsz, bool *initial_quote)
   * The result is a palloc'd string, or NULL if we have reached EOL.
   */
  static char *
! next_token_expand(const char *filename, FILE *file)
  {
  	char		buf[MAX_TOKEN];
  	char	   *comma_str = pstrdup("");
--- 222,228 ----
   * The result is a palloc'd string, or NULL if we have reached EOL.
   */
  static char *
! next_token_expand(const char *filename, FILE *file, const char **keywords)
  {
  	char		buf[MAX_TOKEN];
  	char	   *comma_str = pstrdup("");
***************
*** 231,237 **** next_token_expand(const char *filename, FILE *file)
  
  	do
  	{
! 		if (!next_token(file, buf, sizeof(buf), &initial_quote))
  			break;
  
  		got_something = true;
--- 234,241 ----
  
  	do
  	{
! 		if (!next_token(filename, file, buf, sizeof(buf), &initial_quote,
! 						keywords))
  			break;
  
  		got_something = true;
***************
*** 246,252 **** next_token_expand(const char *filename, FILE *file)
  
  		/* Is this referencing a file? */
  		if (!initial_quote && buf[0] == '@' && buf[1] != '\0')
! 			incbuf = tokenize_inc_file(filename, buf + 1);
  		else
  			incbuf = pstrdup(buf);
  
--- 250,256 ----
  
  		/* Is this referencing a file? */
  		if (!initial_quote && buf[0] == '@' && buf[1] != '\0')
! 			incbuf = tokenize_inc_file(filename, buf + 1, keywords);
  		else
  			incbuf = pstrdup(buf);
  
***************
*** 273,279 **** next_token_expand(const char *filename, FILE *file)
  /*
   * Free memory used by lines/tokens (i.e., structure built by tokenize_file)
   */
! static void
  free_lines(List **lines, List **line_nums)
  {
  	/*
--- 277,283 ----
  /*
   * Free memory used by lines/tokens (i.e., structure built by tokenize_file)
   */
! void
  free_lines(List **lines, List **line_nums)
  {
  	/*
***************
*** 318,324 **** free_lines(List **lines, List **line_nums)
  
  static char *
  tokenize_inc_file(const char *outer_filename,
! 				  const char *inc_filename)
  {
  	char	   *inc_fullname;
  	FILE	   *inc_file;
--- 322,328 ----
  
  static char *
  tokenize_inc_file(const char *outer_filename,
! 				  const char *inc_filename, const char **keywords)
  {
  	char	   *inc_fullname;
  	FILE	   *inc_file;
***************
*** 348,354 **** tokenize_inc_file(const char *outer_filename,
  	{
  		ereport(LOG,
  				(errcode_for_file_access(),
! 				 errmsg("could not open secondary authentication file \"@%s\" as \"%s\": %m",
  						inc_filename, inc_fullname)));
  		pfree(inc_fullname);
  
--- 352,358 ----
  	{
  		ereport(LOG,
  				(errcode_for_file_access(),
! 				 errmsg("could not open secondary configuration file \"@%s\" as \"%s\": %m",
  						inc_filename, inc_fullname)));
  		pfree(inc_fullname);
  
***************
*** 357,363 **** tokenize_inc_file(const char *outer_filename,
  	}
  
  	/* There is possible recursion here if the file contains @ */
! 	tokenize_file(inc_fullname, inc_file, &inc_lines, &inc_line_nums);
  
  	FreeFile(inc_file);
  	pfree(inc_fullname);
--- 361,368 ----
  	}
  
  	/* There is possible recursion here if the file contains @ */
! 	tokenize_file(inc_fullname, inc_file, &inc_lines, &inc_line_nums,
! 				  keywords);
  
  	FreeFile(inc_file);
  	pfree(inc_fullname);
***************
*** 404,412 **** tokenize_inc_file(const char *outer_filename,
   *
   * filename must be the absolute path to the target file.
   */
! static void
  tokenize_file(const char *filename, FILE *file,
! 			  List **lines, List **line_nums)
  {
  	List	   *current_line = NIL;
  	int			line_number = 1;
--- 409,417 ----
   *
   * filename must be the absolute path to the target file.
   */
! void
  tokenize_file(const char *filename, FILE *file,
! 			  List **lines, List **line_nums, const char **keywords)
  {
  	List	   *current_line = NIL;
  	int			line_number = 1;
***************
*** 416,422 **** tokenize_file(const char *filename, FILE *file,
  
  	while (!feof(file) && !ferror(file))
  	{
! 		buf = next_token_expand(filename, file);
  
  		/* add token to list, unless we are at EOL or comment start */
  		if (buf)
--- 421,427 ----
  
  	while (!feof(file) && !ferror(file))
  	{
! 		buf = next_token_expand(filename, file, keywords);
  
  		/* add token to list, unless we are at EOL or comment start */
  		if (buf)
***************
*** 1490,1496 **** load_hba(void)
  		return false;
  	}
  
! 	tokenize_file(HbaFileName, file, &hba_lines, &hba_line_nums);
  	FreeFile(file);
  
  	/* Now parse all the lines */
--- 1495,1501 ----
  		return false;
  	}
  
! 	tokenize_file(HbaFileName, file, &hba_lines, &hba_line_nums, hba_keywords);
  	FreeFile(file);
  
  	/* Now parse all the lines */
***************
*** 1809,1815 **** load_ident(void)
  	}
  	else
  	{
! 		tokenize_file(IdentFileName, file, &ident_lines, &ident_line_nums);
  		FreeFile(file);
  	}
  }
--- 1814,1821 ----
  	}
  	else
  	{
! 		tokenize_file(IdentFileName, file, &ident_lines, &ident_line_nums,
! 					  hba_keywords);
  		FreeFile(file);
  	}
  }
*** a/src/backend/postmaster/postmaster.c
--- b/src/backend/postmaster/postmaster.c
***************
*** 1063,1069 **** PostmasterMain(int argc, char *argv[])
  	autovac_init();
  
  	/*
! 	 * Load configuration files for client authentication.
  	 */
  	if (!load_hba())
  	{
--- 1063,1069 ----
  	autovac_init();
  
  	/*
! 	 * Load configuration files for client authentication and replication.
  	 */
  	if (!load_hba())
  	{
***************
*** 1075,1080 **** PostmasterMain(int argc, char *argv[])
--- 1075,1085 ----
  				(errmsg("could not load pg_hba.conf")));
  	}
  	load_ident();
+ 	if (max_wal_senders > 0 && !load_standbys())
+ 	{
+ 		ereport(FATAL,
+ 				(errmsg("could not load standbys.conf")));
+ 	}
  
  	/*
  	 * Remember postmaster startup time
***************
*** 1713,1718 **** retry1:
--- 1718,1725 ----
  							(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
  							 errmsg("invalid value for boolean option \"replication\"")));
  			}
+ 			else if (strcmp(nameptr, "standby_name") == 0)
+ 			    standby_name = pstrdup(valptr);
  			else
  			{
  				/* Assume it's a generic GUC option */
***************
*** 2129,2134 **** SIGHUP_handler(SIGNAL_ARGS)
--- 2136,2146 ----
  
  		load_ident();
  
+ 		/* Reload standbys configuration file too */
+ 		if (max_wal_senders > 0 && !load_standbys())
+ 			ereport(WARNING,
+ 					(errmsg("standbys.conf not reloaded")));
+ 
  #ifdef EXEC_BACKEND
  		/* Update the starting-point file for future children */
  		write_nondefault_variables(PGC_SIGHUP);
*** a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
--- b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
***************
*** 47,55 **** static bool justconnected = false;
  static char *recvBuf = NULL;
  
  /* Prototypes for interface functions */
! static bool libpqrcv_connect(char *conninfo, XLogRecPtr startpoint);
  static bool libpqrcv_receive(int timeout, unsigned char *type,
  				 char **buffer, int *len);
  static void libpqrcv_disconnect(void);
  
  /* Prototypes for private functions */
--- 47,57 ----
  static char *recvBuf = NULL;
  
  /* Prototypes for interface functions */
! static bool libpqrcv_connect(char *conninfo, XLogRecPtr startpoint,
! 							 char *standbyName);
  static bool libpqrcv_receive(int timeout, unsigned char *type,
  				 char **buffer, int *len);
+ static void libpqrcv_send(const char *buffer, int nbytes);
  static void libpqrcv_disconnect(void);
  
  /* Prototypes for private functions */
***************
*** 64,73 **** _PG_init(void)
  {
  	/* Tell walreceiver how to reach us */
  	if (walrcv_connect != NULL || walrcv_receive != NULL ||
! 		walrcv_disconnect != NULL)
  		elog(ERROR, "libpqwalreceiver already loaded");
  	walrcv_connect = libpqrcv_connect;
  	walrcv_receive = libpqrcv_receive;
  	walrcv_disconnect = libpqrcv_disconnect;
  }
  
--- 66,76 ----
  {
  	/* Tell walreceiver how to reach us */
  	if (walrcv_connect != NULL || walrcv_receive != NULL ||
! 		walrcv_send != NULL || walrcv_disconnect != NULL)
  		elog(ERROR, "libpqwalreceiver already loaded");
  	walrcv_connect = libpqrcv_connect;
  	walrcv_receive = libpqrcv_receive;
+ 	walrcv_send = libpqrcv_send;
  	walrcv_disconnect = libpqrcv_disconnect;
  }
  
***************
*** 75,98 **** _PG_init(void)
   * Establish the connection to the primary server for XLOG streaming
   */
  static bool
! libpqrcv_connect(char *conninfo, XLogRecPtr startpoint)
  {
! 	char		conninfo_repl[MAXCONNINFO + 37];
  	char	   *primary_sysid;
  	char		standby_sysid[32];
  	TimeLineID	primary_tli;
  	TimeLineID	standby_tli;
  	PGresult   *res;
  	char		cmd[64];
  
  	/*
! 	 * Connect using deliberately undocumented parameter: replication. The
! 	 * database name is ignored by the server in replication mode, but specify
! 	 * "replication" for .pgpass lookup.
  	 */
! 	snprintf(conninfo_repl, sizeof(conninfo_repl),
! 			 "%s dbname=replication replication=true",
! 			 conninfo);
  
  	streamConn = PQconnectdb(conninfo_repl);
  	if (PQstatus(streamConn) != CONNECTION_OK)
--- 78,107 ----
   * Establish the connection to the primary server for XLOG streaming
   */
  static bool
! libpqrcv_connect(char *conninfo, XLogRecPtr startpoint, char *standbyName)
  {
! 	char		conninfo_repl[MAXCONNINFO + MAXSTANDBYNAME + 37];
  	char	   *primary_sysid;
  	char		standby_sysid[32];
  	TimeLineID	primary_tli;
  	TimeLineID	standby_tli;
+ 	char	   *primary_rplMode;
  	PGresult   *res;
  	char		cmd[64];
  
  	/*
! 	 * Connect using deliberately undocumented parameter: replication
! 	 * and standby_name. The database name is ignored by the server in
! 	 * replication mode, but specify "replication" for .pgpass lookup.
  	 */
! 	if (standbyName)
! 		snprintf(conninfo_repl, sizeof(conninfo_repl),
! 				 "%s dbname=replication replication=true standby_name=%s",
! 				 conninfo, standbyName);
! 	else
! 		snprintf(conninfo_repl, sizeof(conninfo_repl),
! 				 "%s dbname=replication replication=true",
! 				 conninfo);
  
  	streamConn = PQconnectdb(conninfo_repl);
  	if (PQstatus(streamConn) != CONNECTION_OK)
***************
*** 109,119 **** 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) != 2 || PQntuples(res) != 1)
  	{
  		int			ntuples = PQntuples(res);
  		int			nfields = PQnfields(res);
--- 118,128 ----
  	{
  		PQclear(res);
  		ereport(ERROR,
! 				(errmsg("could not receive database system identifier, timeline ID and "
! 						"replication mode from the primary server: %s",
  						PQerrorMessage(streamConn))));
  	}
! 	if (PQnfields(res) != 3 || PQntuples(res) != 1)
  	{
  		int			ntuples = PQntuples(res);
  		int			nfields = PQnfields(res);
***************
*** 121,131 **** libpqrcv_connect(char *conninfo, XLogRecPtr startpoint)
  		PQclear(res);
  		ereport(ERROR,
  				(errmsg("invalid response from primary server"),
! 				 errdetail("Expected 1 tuple with 2 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.
--- 130,141 ----
  		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);
+ 	primary_rplMode = PQgetvalue(res, 0, 2);
  
  	/*
  	 * Confirm that the system identifier of the primary is the same as ours.
***************
*** 146,158 **** 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);
--- 156,180 ----
  	 * 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 passed replication mode is valid.
+ 	 */
+ 	rplMode = ReplicationModeNameGetValue(primary_rplMode);
+ 	PQclear(res);
+ 	if (rplMode == InvalidReplicationMode)
+ 		ereport(ERROR,
+ 				(errmsg("invalid replication mode \"%s\"",
+ 						primary_rplMode)));
+ 
  	/* Start streaming from the point requested by startup process */
  	snprintf(cmd, sizeof(cmd), "START_REPLICATION %X/%X",
  			 startpoint.xlogid, startpoint.xrecoff);
***************
*** 398,400 **** libpqrcv_receive(int timeout, unsigned char *type, char **buffer, int *len)
--- 420,437 ----
  
  	return true;
  }
+ 
+ /*
+  * Send a message to XLOG stream.
+  *
+  * ereports on error.
+  */
+ static void
+ libpqrcv_send(const char *buffer, int nbytes)
+ {
+ 	if (PQputCopyData(streamConn, buffer, nbytes) <= 0 ||
+ 		PQflush(streamConn))
+ 		ereport(ERROR,
+ 				(errmsg("could not send data to WAL stream: %s",
+ 						PQerrorMessage(streamConn))));
+ }
*** /dev/null
--- b/src/backend/replication/standbys.conf.sample
***************
*** 0 ****
--- 1,35 ----
+ # PostgreSQL Standbys Configuration File
+ # ===================================================
+ #
+ # Refer to the "Streaming Replication" section in the PostgreSQL
+ # documentation for a complete description of this file.  A short
+ # synopsis follows.
+ #
+ # This file controls which replication mode each standby uses.
+ # Records are of the form:
+ #
+ # STANDBY-NAME  REPLICATION-MODE
+ #
+ # (The uppercase items must be replaced by actual values.)
+ #
+ # STANDBY-NAME can be "all", standby name, or a comma-separated list
+ # thereof.
+ #
+ # REPLICATION-MODE specifies how long transaction commit waits for
+ # replication before the commit command returns a "success" to a
+ # client. The valid modes are "async", "recv", "fsync" and "replay".
+ #
+ # Standby name containing spaces, commas, quotes and other special
+ # characters must be quoted.  Quoting one of the keyword "all" makes
+ # the name lose its special character, and just match standby with
+ # that name.
+ #
+ # This file is read on server startup and when the postmaster receives
+ # a SIGHUP signal.  If you edit the file on a running system, you have
+ # to SIGHUP the postmaster for the changes to take effect.  You can
+ # use "pg_ctl reload" to do that.
+ 
+ # Put your actual configuration here
+ # ----------------------------------
+ 
+ # STANDBY-NAME       REPLICATION-MODE
*** a/src/backend/replication/walreceiver.c
--- b/src/backend/replication/walreceiver.c
***************
*** 57,62 **** bool		am_walreceiver;
--- 57,63 ----
  /* libpqreceiver hooks to these when loaded */
  walrcv_connect_type walrcv_connect = NULL;
  walrcv_receive_type walrcv_receive = NULL;
+ walrcv_send_type walrcv_send = NULL;
  walrcv_disconnect_type walrcv_disconnect = NULL;
  
  #define NAPTIME_PER_CYCLE 100	/* max sleep time between cycles (100ms) */
***************
*** 113,118 **** static void WalRcvDie(int code, Datum arg);
--- 114,120 ----
  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 XLogWalRcvSendRecPtr(XLogRecPtr recptr);
  
  /* Signal handlers */
  static void WalRcvSigHupHandler(SIGNAL_ARGS);
***************
*** 158,164 **** void
--- 160,168 ----
  WalReceiverMain(void)
  {
  	char		conninfo[MAXCONNINFO];
+ 	char		standbyName[MAXSTANDBYNAME];
  	XLogRecPtr	startpoint;
+ 	XLogRecPtr	ackedpoint = {0, 0};
  
  	/* use volatile pointer to prevent code rearrangement */
  	volatile WalRcvData *walrcv = WalRcv;
***************
*** 206,211 **** WalReceiverMain(void)
--- 210,216 ----
  
  	/* Fetch information required to start streaming */
  	strlcpy(conninfo, (char *) walrcv->conninfo, MAXCONNINFO);
+ 	strlcpy(standbyName, (char *) walrcv->standbyName, MAXSTANDBYNAME);
  	startpoint = walrcv->receivedUpto;
  	SpinLockRelease(&walrcv->mutex);
  
***************
*** 247,253 **** WalReceiverMain(void)
  	/* Load the libpq-specific functions */
  	load_file("libpqwalreceiver", false);
  	if (walrcv_connect == NULL || walrcv_receive == NULL ||
! 		walrcv_disconnect == NULL)
  		elog(ERROR, "libpqwalreceiver didn't initialize correctly");
  
  	/*
--- 252,258 ----
  	/* Load the libpq-specific functions */
  	load_file("libpqwalreceiver", false);
  	if (walrcv_connect == NULL || walrcv_receive == NULL ||
! 		walrcv_send == NULL || walrcv_disconnect == NULL)
  		elog(ERROR, "libpqwalreceiver didn't initialize correctly");
  
  	/*
***************
*** 261,267 **** WalReceiverMain(void)
  
  	/* Establish the connection to the primary for XLOG streaming */
  	EnableWalRcvImmediateExit();
! 	walrcv_connect(conninfo, startpoint);
  	DisableWalRcvImmediateExit();
  
  	/* Loop until end-of-streaming or error */
--- 266,272 ----
  
  	/* Establish the connection to the primary for XLOG streaming */
  	EnableWalRcvImmediateExit();
! 	walrcv_connect(conninfo, startpoint, standbyName);
  	DisableWalRcvImmediateExit();
  
  	/* Loop until end-of-streaming or error */
***************
*** 311,316 **** WalReceiverMain(void)
--- 316,340 ----
  			 */
  			XLogWalRcvFlush();
  		}
+ 
+ 		/*
+ 		 * If replication_mode is "replay", send the last WAL replay location
+ 		 * to the primary, to acknowledge that replication has been completed
+ 		 * up to that. This occurs only when WAL records were replayed since
+ 		 * the last acknowledgement.
+ 		 */
+ 		if (rplMode == REPLICATION_MODE_REPLAY &&
+ 			XLByteLT(ackedpoint, LogstreamResult.Flush))
+ 		{
+ 			XLogRecPtr	recptr;
+ 
+ 			recptr = GetReplayRecPtr();
+ 			if (XLByteLT(ackedpoint, recptr))
+ 			{
+ 				XLogWalRcvSendRecPtr(recptr);
+ 				ackedpoint = recptr;
+ 			}
+ 		}
  	}
  }
  
***************
*** 406,411 **** XLogWalRcvProcessMsg(unsigned char type, char *buf, Size len)
--- 430,448 ----
  				buf += sizeof(WalDataMessageHeader);
  				len -= sizeof(WalDataMessageHeader);
  
+ 				/*
+ 				 * If replication_mode is "recv", send the last WAL receive
+ 				 * location to the primary, to acknowledge that replication
+ 				 * has been completed up to that.
+ 				 */
+ 				if (rplMode == REPLICATION_MODE_RECV)
+ 				{
+ 					XLogRecPtr	endptr = msghdr.dataStart;
+ 
+ 					XLByteAdvance(endptr, len);
+ 					XLogWalRcvSendRecPtr(endptr);
+ 				}
+ 
  				XLogWalRcvWrite(buf, len, msghdr.dataStart);
  				break;
  			}
***************
*** 523,528 **** XLogWalRcvFlush(void)
--- 560,573 ----
  
  		LogstreamResult.Flush = LogstreamResult.Write;
  
+ 		/*
+ 		 * If replication_mode is "fsync", send the last WAL flush
+ 		 * location to the primary, to acknowledge that replication
+ 		 * has been completed up to that.
+ 		 */
+ 		if (rplMode == REPLICATION_MODE_FSYNC)
+ 			XLogWalRcvSendRecPtr(LogstreamResult.Flush);
+ 
  		/* Update shared-memory status */
  		SpinLockAcquire(&walrcv->mutex);
  		walrcv->latestChunkStart = walrcv->receivedUpto;
***************
*** 541,543 **** XLogWalRcvFlush(void)
--- 586,609 ----
  		}
  	}
  }
+ 
+ /* Send the lsn to the primary server */
+ static void
+ XLogWalRcvSendRecPtr(XLogRecPtr recptr)
+ {
+ 	static char	   *msgbuf = NULL;
+ 	WalAckMessageData	msgdata;
+ 
+ 	/*
+ 	 * Allocate buffer that will be used for each output message if first
+ 	 * time through.  We do this just once to reduce palloc overhead.
+ 	 * The buffer must be made large enough for maximum-sized messages.
+ 	 */
+ 	if (msgbuf == NULL)
+ 		msgbuf = palloc(1 + sizeof(WalAckMessageData));
+ 
+ 	msgbuf[0] = 'l';
+ 	msgdata.ackEnd = recptr;
+ 	memcpy(msgbuf + 1, &msgdata, sizeof(WalAckMessageData));
+ 	walrcv_send(msgbuf, 1 + sizeof(WalAckMessageData));
+ }
*** a/src/backend/replication/walreceiverfuncs.c
--- b/src/backend/replication/walreceiverfuncs.c
***************
*** 168,178 **** ShutdownWalRcv(void)
  /*
   * Request postmaster to start walreceiver.
   *
!  * recptr indicates the position where streaming should begin, and conninfo
!  * is a libpq connection string to use.
   */
  void
! RequestXLogStreaming(XLogRecPtr recptr, const char *conninfo)
  {
  	/* use volatile pointer to prevent code rearrangement */
  	volatile WalRcvData *walrcv = WalRcv;
--- 168,180 ----
  /*
   * Request postmaster to start walreceiver.
   *
!  * recptr indicates the position where streaming should begin, conninfo
!  * is a libpq connection string to use, and standbyName is name of this
!  * standby.
   */
  void
! RequestXLogStreaming(XLogRecPtr recptr, const char *conninfo,
! 					 const char *standbyName)
  {
  	/* use volatile pointer to prevent code rearrangement */
  	volatile WalRcvData *walrcv = WalRcv;
***************
*** 196,201 **** RequestXLogStreaming(XLogRecPtr recptr, const char *conninfo)
--- 198,207 ----
  		strlcpy((char *) walrcv->conninfo, conninfo, MAXCONNINFO);
  	else
  		walrcv->conninfo[0] = '\0';
+ 	if (standbyName != NULL)
+ 		strlcpy((char *) walrcv->standbyName, standbyName, MAXSTANDBYNAME);
+ 	else
+ 		walrcv->standbyName[0] = '\0';
  	walrcv->walRcvState = WALRCV_STARTING;
  	walrcv->startTime = now;
  
*** a/src/backend/replication/walsender.c
--- b/src/backend/replication/walsender.c
***************
*** 38,43 ****
--- 38,44 ----
  
  #include "access/xlog_internal.h"
  #include "catalog/pg_type.h"
+ #include "libpq/hba.h"
  #include "libpq/libpq.h"
  #include "libpq/pqformat.h"
  #include "libpq/pqsignal.h"
***************
*** 61,66 **** static WalSnd *MyWalSnd = NULL;
--- 62,68 ----
  
  /* Global state */
  bool		am_walsender = false;		/* Am I a walsender process ? */
+ char	   *standby_name = NULL;		/* Name of connected standby */
  
  /* User-settable parameters for walsender */
  int			max_wal_senders = 0;	/* the maximum number of concurrent walsenders */
***************
*** 84,94 **** static uint32 sendOff = 0;
--- 86,114 ----
   */
  static XLogRecPtr sentPtr = {0, 0};
  
+ /*
+  * How far have we completed replication already? This is also
+  * advertised in MyWalSnd->ackdPtr. This is not used in asynchronous
+  * replication case.
+  */
+ static XLogRecPtr ackdPtr = {0, 0};
+ 
  /* Flags set by signal handlers for later service in main loop */
  static volatile sig_atomic_t got_SIGHUP = false;
  static volatile sig_atomic_t shutdown_requested = false;
  static volatile sig_atomic_t ready_to_stop = false;
  
+ /*
+  * pre-parsed content of standbys configuration file: list of
+  * StandbysLine structs
+  */
+ static List *parsed_standbys_lines = NIL;
+ 
+ static const char *standbys_keywords[] = {"all", NULL};
+ 
+ /* Path of standbys configuration file (relative to $PGDATA) */
+ #define STANDBYS_FILE		"standbys.conf"
+ 
  /* Signal handlers */
  static void WalSndSigHupHandler(SIGNAL_ARGS);
  static void WalSndShutdownHandler(SIGNAL_ARGS);
***************
*** 102,108 **** static void WalSndHandshake(void);
  static void WalSndKill(int code, Datum arg);
  static void XLogRead(char *buf, XLogRecPtr recptr, Size nbytes);
  static bool XLogSend(char *msgbuf, bool *caughtup);
! static void CheckClosedConnection(void);
  
  
  /* Main entry point for walsender process */
--- 122,132 ----
  static void WalSndKill(int code, Datum arg);
  static void XLogRead(char *buf, XLogRecPtr recptr, Size nbytes);
  static bool XLogSend(char *msgbuf, bool *caughtup);
! static void ProcessStreamMsgs(StringInfo inMsg);
! 
! static bool parse_standbys_line(List *line, int line_num, StandbysLine *parsedline);
! static void free_standbys_record(StandbysLine *record);
! static void clean_standbys_list(List *lines);
  
  
  /* Main entry point for walsender process */
***************
*** 209,227 **** WalSndHandshake(void)
  						StringInfoData buf;
  						char		sysid[32];
  						char		tli[11];
  
  						/*
! 						 * Reply with a result set with one row, two columns.
! 						 * First col is system ID, and second is timeline ID
  						 */
  
  						snprintf(sysid, sizeof(sysid), UINT64_FORMAT,
  								 GetSystemIdentifier());
  						snprintf(tli, sizeof(tli), "%u", ThisTimeLineID);
  
  						/* Send a RowDescription message */
  						pq_beginmessage(&buf, 'T');
! 						pq_sendint(&buf, 2, 2); /* 2 fields */
  
  						/* first field */
  						pq_sendstring(&buf, "systemid");		/* col name */
--- 233,254 ----
  						StringInfoData buf;
  						char		sysid[32];
  						char		tli[11];
+ 						char		mode[8];
  
  						/*
! 						 * Reply with a result set with one row, three columns.
! 						 * First col is system ID, second is timeline ID, and
! 						 * third is replication mode.
  						 */
  
  						snprintf(sysid, sizeof(sysid), UINT64_FORMAT,
  								 GetSystemIdentifier());
  						snprintf(tli, sizeof(tli), "%u", ThisTimeLineID);
+ 						snprintf(mode, sizeof(mode), "%s", ReplicationModeNames[rplMode]);
  
  						/* Send a RowDescription message */
  						pq_beginmessage(&buf, 'T');
! 						pq_sendint(&buf, 3, 2); /* 3 fields */
  
  						/* first field */
  						pq_sendstring(&buf, "systemid");		/* col name */
***************
*** 240,254 **** WalSndHandshake(void)
  						pq_sendint(&buf, 4, 2); /* typlen */
  						pq_sendint(&buf, 0, 4); /* typmod */
  						pq_sendint(&buf, 0, 2); /* format code */
  						pq_endmessage(&buf);
  
  						/* Send a DataRow message */
  						pq_beginmessage(&buf, 'D');
! 						pq_sendint(&buf, 2, 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_endmessage(&buf);
  
  						/* Send CommandComplete and ReadyForQuery messages */
--- 267,292 ----
  						pq_sendint(&buf, 4, 2); /* typlen */
  						pq_sendint(&buf, 0, 4); /* typmod */
  						pq_sendint(&buf, 0, 2); /* format code */
+ 
+ 						/* third field */
+ 						pq_sendstring(&buf, "replication_mode");	/* col name */
+ 						pq_sendint(&buf, 0, 4); /* table oid */
+ 						pq_sendint(&buf, 0, 2); /* attnum */
+ 						pq_sendint(&buf, TEXTOID, 4);	/* type oid */
+ 						pq_sendint(&buf, -1, 2);		/* typlen */
+ 						pq_sendint(&buf, 0, 4); /* typmod */
+ 						pq_sendint(&buf, 0, 2); /* format code */
  						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(mode), 4);	/* col3 len */
+ 						pq_sendbytes(&buf, (char *) &mode, strlen(mode));
  						pq_endmessage(&buf);
  
  						/* Send CommandComplete and ReadyForQuery messages */
***************
*** 286,295 **** WalSndHandshake(void)
  						pq_flush();
  
  						/*
! 						 * Initialize position to the received one, then the
  						 * xlog records begin to be shipped from that position
  						 */
! 						sentPtr = recptr;
  
  						/* break out of the loop */
  						replication_started = true;
--- 324,333 ----
  						pq_flush();
  
  						/*
! 						 * Initialize positions to the received one, then the
  						 * xlog records begin to be shipped from that position
  						 */
! 						sentPtr = ackdPtr = recptr;
  
  						/* break out of the loop */
  						replication_started = true;
***************
*** 323,375 **** WalSndHandshake(void)
  }
  
  /*
!  * Check if the remote end has closed the connection.
   */
  static void
! CheckClosedConnection(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)
  	{
! 		/* no data available without blocking */
! 		return;
! 	}
  
- 	/* Handle the very limited subset of commands expected in this phase */
- 	switch (firstchar)
- 	{
  			/*
  			 * '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)));
  	}
  }
  
  /* Main loop of walsender process */
  static int
  WalSndLoop(void)
  {
  	char	   *output_message;
  	bool		caughtup = false;
  
  	/*
  	 * Allocate buffer that will be used for each output message.  We do this
  	 * just once to reduce palloc overhead.  The buffer must be made large
--- 361,482 ----
  }
  
  /*
!  * Process messages received from the standby.
!  *
!  * ereports on error.
   */
  static void
! ProcessStreamMsgs(StringInfo inMsg)
  {
! 	bool	acked = false;
  
! 	/* Loop to process successive complete messages available */
! 	for (;;)
  	{
! 		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)
! 		{
! 			/* no data available without blocking */
! 			break;
! 		}
! 
! 		/* Handle the very limited subset of commands expected in this phase */
! 		switch (firstchar)
! 		{
! 			case 'd':       /* CopyData message */
! 			{
! 				unsigned char	rpltype;
! 
! 				/*
! 				 * Read the message contents. This is expected to be done without
! 				 * blocking because we've been able to get message type code.
! 				 */
! 				if (pq_getmessage(inMsg, 0))
! 					proc_exit(0);		/* suitable message already logged */
! 
! 				/* Read the replication message type from CopyData message */
! 				rpltype = pq_getmsgbyte(inMsg);
! 				switch (rpltype)
! 				{
! 					case 'l':
! 					{
! 						WalAckMessageData  *msgdata;
! 
! 						msgdata = (WalAckMessageData *) pq_getmsgbytes(inMsg, sizeof(WalAckMessageData));
! 
! 						/*
! 						 * Update local status.
! 						 *
! 						 * The ackd ptr received from standby should not
! 						 * go backwards.
! 						 */
! 						if (XLByteLE(ackdPtr, msgdata->ackEnd))
! 							ackdPtr = msgdata->ackEnd;
! 						else
! 							ereport(FATAL,
! 									(errmsg("replication completion location went back from "
! 											"%X/%X to %X/%X",
! 											ackdPtr.xlogid, ackdPtr.xrecoff,
! 											msgdata->ackEnd.xlogid, msgdata->ackEnd.xrecoff)));
! 
! 						acked = true;	/* also need to update shared position */
! 						break;
! 					}
! 					default:
! 						ereport(FATAL,
! 								(errcode(ERRCODE_PROTOCOL_VIOLATION),
! 								 errmsg("invalid replication message type %d",
! 										rpltype)));
! 				}
! 				break;
! 			}
  
  			/*
  			 * '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)));
! 		}
  	}
+ 
+ 	if (acked)
+ 	{
+ 		/* use volatile pointer to prevent code rearrangement */
+ 		volatile WalSnd *walsnd = MyWalSnd;
+ 
+ 		SpinLockAcquire(&walsnd->mutex);
+ 		walsnd->ackdPtr = ackdPtr;
+ 		SpinLockRelease(&walsnd->mutex);
+  	}
  }
  
  /* Main loop of walsender process */
  static int
  WalSndLoop(void)
  {
+ 	StringInfoData	input_message;
  	char	   *output_message;
  	bool		caughtup = false;
  
+ 	initStringInfo(&input_message);
+ 
  	/*
  	 * Allocate buffer that will be used for each output message.  We do this
  	 * just once to reduce palloc overhead.  The buffer must be made large
***************
*** 438,444 **** WalSndLoop(void)
  
  				/* Sleep and check that the connection is still alive */
  				pg_usleep(remain > NAPTIME_PER_CYCLE ? NAPTIME_PER_CYCLE : remain);
! 				CheckClosedConnection();
  
  				remain -= NAPTIME_PER_CYCLE;
  			}
--- 545,551 ----
  
  				/* Sleep and check that the connection is still alive */
  				pg_usleep(remain > NAPTIME_PER_CYCLE ? NAPTIME_PER_CYCLE : remain);
! 				ProcessStreamMsgs(&input_message);
  
  				remain -= NAPTIME_PER_CYCLE;
  			}
***************
*** 497,502 **** InitWalSnd(void)
--- 604,611 ----
  			MyWalSnd = (WalSnd *) walsnd;
  			walsnd->pid = MyProcPid;
  			MemSet(&MyWalSnd->sentPtr, 0, sizeof(XLogRecPtr));
+ 			MemSet(&MyWalSnd->ackdPtr, 0, sizeof(XLogRecPtr));
+ 			walsnd->rplMode = rplMode;
  			SpinLockRelease(&walsnd->mutex);
  			break;
  		}
***************
*** 523,528 **** WalSndKill(int code, Datum arg)
--- 632,638 ----
  	 * for this.
  	 */
  	MyWalSnd->pid = 0;
+ 	MyWalSnd->rplMode = InvalidReplicationMode;
  
  	/* WalSnd struct isn't mine anymore */
  	MyWalSnd = NULL;
***************
*** 896,938 **** WalSndShmemInit(void)
  }
  
  /*
!  * This isn't currently used for anything. Monitoring tools might be
!  * interested in the future, and we'll need something like this in the
!  * future for synchronous replication.
   */
! #ifdef NOT_USED
  /*
!  * Returns the oldest Send position among walsenders. Or InvalidXLogRecPtr
!  * if none.
   */
! XLogRecPtr
! GetOldestWALSendPointer(void)
  {
! 	XLogRecPtr	oldest = {0, 0};
! 	int			i;
! 	bool		found = false;
  
! 	for (i = 0; i < max_wal_senders; i++)
  	{
! 		/* use volatile pointer to prevent code rearrangement */
! 		volatile WalSnd *walsnd = &WalSndCtl->walsnds[i];
! 		XLogRecPtr	recptr;
  
! 		if (walsnd->pid == 0)
! 			continue;
  
! 		SpinLockAcquire(&walsnd->mutex);
! 		recptr = walsnd->sentPtr;
! 		SpinLockRelease(&walsnd->mutex);
  
! 		if (recptr.xlogid == 0 && recptr.xrecoff == 0)
! 			continue;
  
! 		if (!found || XLByteLT(recptr, oldest))
! 			oldest = recptr;
! 		found = true;
  	}
! 	return oldest;
  }
  
! #endif
--- 1006,1253 ----
  }
  
  /*
!  * Ensure that all xlog records through the given position is
!  * replicated to the standby servers.
!  *
!  * XXX: We should replace the poll loop in this function with a latch.
!  */
! void
! WaitXLogSend(XLogRecPtr record)
! {
! 	Assert(max_wal_senders > 0);
! 
! 	/*
! 	 * XXX: We should track the number of currently connected standbys
! 	 * and skip waiting if it's zero.
! 	 */
! 
! 	for (;;)
! 	{
! 		int		i;
! 		bool	ackd = true;
! 
! 		for (i = 0; i < max_wal_senders; i++)
! 		{
! 			/* use volatile pointer to prevent code rearrangement */
! 			volatile WalSnd *walsnd = &WalSndCtl->walsnds[i];
! 			XLogRecPtr		recptr;
! 
! 			/* Don't wait for unconnected and asynchronous standbys */
! 			if (walsnd->pid == 0 || walsnd->rplMode <= REPLICATION_MODE_ASYNC)
! 				continue;
! 
! 			SpinLockAcquire(&walsnd->mutex);
! 			recptr = walsnd->ackdPtr;
! 			SpinLockRelease(&walsnd->mutex);
! 
! 			if (XLByteLT(recptr, record))
! 			{
! 				ackd = false;
! 				break;
! 			}
! 		}
! 
! 		if (ackd)
! 			return;
! 
! 		pg_usleep(100000L);     /* 100ms */
! 	}
! }
! 
! 
! /* ----------
!  * Routines to handle standbys configuration file
!  * ----------
!  */
! 
! /*
!  * Scan the (pre-parsed) standbys configuration file line by line,
!  * looking for a match to the standby name passed from the standby.
   */
! bool
! check_standbys(void)
! {
! 	ListCell   *line;
! 	StandbysLine *standbys;
! 
! 	foreach(line, parsed_standbys_lines)
! 	{
! 		char	   *tok;
! 
! 		standbys = (StandbysLine *) lfirst(line);
! 
! 		/* Check standby name */
! 		for (tok = strtok(standbys->standbyName, MULTI_VALUE_SEP);
! 			 tok != NULL;
! 			 tok = strtok(NULL, MULTI_VALUE_SEP))
! 		{
! 			if (strcmp(tok, "all\n") == 0 ||
! 				(standby_name != NULL &&
! 				 strcmp(tok, standby_name) == 0))
! 			{
! 				rplMode = standbys->rplMode;
! 				return true;
! 			}
! 		}
! 	}
! 	return false;
! }
! 
  /*
!  * Parse one line in the standbys configuration file and store
!  * the result in a StandbysLine structure.
   */
! static bool
! parse_standbys_line(List *line, int line_num, StandbysLine *parsedline)
  {
! 	char	   *token;
! 	ListCell   *line_item;
  
! 	line_item = list_head(line);
! 
! 	parsedline->linenumber = line_num;
! 
! 	/* Get the standby name. */
! 	parsedline->standbyName = pstrdup(lfirst(line_item));
! 
! 	/* Get the mode. */
! 	line_item = lnext(line_item);
! 	if (!line_item)
  	{
! 		ereport(LOG,
! 				(errcode(ERRCODE_CONFIG_FILE_ERROR),
! 				 errmsg("end-of-line before mode specification"),
! 				 errcontext("line %d of configuration file \"%s\"",
! 							line_num, STANDBYS_FILE)));
! 		return false;
! 	}
! 	token = lfirst(line_item);
  
! 	parsedline->rplMode = ReplicationModeNameGetValue(token);
! 	if (parsedline->rplMode == InvalidReplicationMode)
! 	{
! 		ereport(LOG,
! 				(errcode(ERRCODE_CONFIG_FILE_ERROR),
! 				 errmsg("invalid replication mode \"%s\"",
! 						token),
! 				 errcontext("line %d of configuration file \"%s\"",
! 							line_num, STANDBYS_FILE)));
! 		return false;
! 	}
  
! 	/* Ignore remaining tokens */
  
! 	return true;
! }
! 
! /*
!  * Free an StandbysLine structure
!  */
! static void
! free_standbys_record(StandbysLine *record)
! {
! 	if (record->standbyName)
! 		pfree(record->standbyName);
! 	pfree(record);
! }
! 
! /*
!  * Free all records on the parsed Standbys list
!  */
! static void
! clean_standbys_list(List *lines)
! {
! 	ListCell   *line;
! 
! 	foreach(line, lines)
! 	{
! 		StandbysLine    *parsed = (StandbysLine *) lfirst(line);
  
! 		if (parsed)
! 			free_standbys_record(parsed);
  	}
! 	list_free(lines);
  }
  
! /*
!  * Read the config file and create a List of StandbysLine records for the contents.
!  *
!  * The configuration is read into a temporary list, and if any parse error occurs
!  * the old list is kept in place and false is returned. Only if the whole file
!  * parses Ok is the list replaced, and the function returns true.
!  */
! bool
! load_standbys(void)
! {
! 	FILE	   *file;
! 	List	   *standbys_lines = NIL;
! 	List	   *standbys_line_nums = NIL;
! 	ListCell   *line,
! 			   *line_num;
! 	List	   *new_parsed_lines = NIL;
! 	bool		ok = true;
! 
! 	/* Ignore standbys.conf if replication is not enabled */
! 	if (max_wal_senders <= 0)
! 		return true;
! 
! 	file = AllocateFile(STANDBYS_FILE, "r");
! 	if (file == NULL)
! 	{
! 		ereport(LOG,
! 				(errcode_for_file_access(),
! 				 errmsg("could not open configuration file \"%s\": %m",
! 						STANDBYS_FILE)));
! 
! 		/*
! 		 * Caller will take care of making this a FATAL error in case this is
! 		 * the initial startup. If it happens on reload, we just keep the old
! 		 * version around.
! 		 */
! 		return false;
! 	}
! 
! 	tokenize_file(STANDBYS_FILE, file, &standbys_lines, &standbys_line_nums,
! 				  standbys_keywords);
! 	FreeFile(file);
! 
! 	/* Now parse all the lines */
! 	forboth(line, standbys_lines, line_num, standbys_line_nums)
! 	{
! 		StandbysLine    *newline;
! 
! 		newline = palloc0(sizeof(StandbysLine));
! 
! 		if (!parse_standbys_line(lfirst(line), lfirst_int(line_num), newline))
! 		{
! 			/* Parse error in the file, so indicate there's a problem */
! 			free_standbys_record(newline);
! 			ok = false;
! 
! 			/*
! 			 * Keep parsing the rest of the file so we can report errors on
! 			 * more than the first row. Error has already been reported in the
! 			 * parsing function, so no need to log it here.
! 			 */
! 			continue;
! 		}
! 
! 		new_parsed_lines = lappend(new_parsed_lines, newline);
! 	}
! 
! 	/* Free the temporary lists */
! 	free_lines(&standbys_lines, &standbys_line_nums);
! 
! 	if (!ok)
! 	{
! 		/* Parsing failed at one or more rows, so bail out */
! 		clean_standbys_list(new_parsed_lines);
! 		return false;
! 	}
! 
! 	/* Loaded new file successfully, replace the one we use */
! 	clean_standbys_list(parsed_standbys_lines);
! 	parsed_standbys_lines = new_parsed_lines;
! 
! 	return true;
! }
*** a/src/backend/utils/init/postinit.c
--- b/src/backend/utils/init/postinit.c
***************
*** 661,666 **** InitPostgres(const char *in_dbname, Oid dboid, const char *username,
--- 661,688 ----
  			ereport(FATAL,
  					(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
  					 errmsg("must be superuser to start walsender")));
+ 
+ 		/*
+ 		 * In EXEC_BACKEND case, we didn't inherit the contents of standbys.conf
+ 		 * etcetera from the postmaster, and have to load them ourselves.  Note we
+ 		 * are loading them into the startup transaction's memory context, not
+ 		 * PostmasterContext, but that shouldn't matter.
+ 		 *
+ 		 * FIXME: [fork/exec] Ugh.	Is there a way around this overhead?
+ 		 */
+ #ifdef EXEC_BACKEND
+ 		if (!load_standbys())
+ 		{
+ 			ereport(FATAL,
+ 					(errmsg("could not load standbys.conf")));
+ 		}
+ #endif
+ 
+ 		if (!check_standbys())
+ 			ereport(FATAL,
+ 					(errmsg("no standbys.conf entry for standby name \"%s\"",
+ 							standby_name)));
+ 
  		/* report this backend in the PgBackendStatus array */
  		pgstat_bestart();
  		/* close the transaction we started above */
*** a/src/include/access/xlog.h
--- b/src/include/access/xlog.h
***************
*** 189,194 **** typedef enum
--- 189,229 ----
  
  extern XLogRecPtr XactLastRecEnd;
  
+ /*
+  * Replication mode. This is used to identify how long transaction
+  * commit should wait for replication.
+  *
+  * REPLICATION_MODE_ASYNC doesn't make transaction commit wait for
+  * replication, i.e., asynchronous replication.
+  *
+  * REPLICATION_MODE_RECV makes transaction commit wait for XLOG
+  * records to be received on the standby.
+  *
+  * REPLICATION_MODE_FSYNC makes transaction commit wait for XLOG
+  * records to be received and fsync'd on the standby.
+  *
+  * REPLICATION_MODE_REPLAY makes transaction commit wait for XLOG
+  * records to be received, fsync'd and replayed on the standby.
+  */
+ typedef enum ReplicationMode
+ {
+ 	InvalidReplicationMode = -1,
+ 	REPLICATION_MODE_ASYNC = 0,
+ 	REPLICATION_MODE_RECV,
+ 	REPLICATION_MODE_FSYNC,
+ 	REPLICATION_MODE_REPLAY
+ 
+ 	/*
+ 	 * NOTE: if you add a new mode, change MAXREPLICATIONMODE below
+ 	 * and update the ReplicationModeNames array in xlog.c
+ 	 */
+ } ReplicationMode;
+ 
+ #define MAXREPLICATIONMODE		REPLICATION_MODE_REPLAY
+ 
+ extern const char *ReplicationModeNames[];
+ extern ReplicationMode	rplMode;
+ 
  /* these variables are GUC parameters related to XLOG */
  extern int	CheckPointSegments;
  extern int	wal_keep_segments;
***************
*** 298,307 **** extern void XLogPutNextOid(Oid nextOid);
--- 333,345 ----
  extern XLogRecPtr GetRedoRecPtr(void);
  extern XLogRecPtr GetInsertRecPtr(void);
  extern XLogRecPtr GetFlushRecPtr(void);
+ extern XLogRecPtr GetReplayRecPtr(void);
  extern void GetNextXidAndEpoch(TransactionId *xid, uint32 *epoch);
  extern TimeLineID GetRecoveryTargetTLI(void);
  
  extern void HandleStartupProcInterrupts(void);
  extern void StartupProcessMain(void);
  
+ extern ReplicationMode ReplicationModeNameGetValue(char *name);
+ 
  #endif   /* XLOG_H */
*** a/src/include/libpq/hba.h
--- b/src/include/libpq/hba.h
***************
*** 15,20 ****
--- 15,24 ----
  #include "libpq/pqcomm.h"
  
  
+ /* This is used to separate values in multi-valued column strings */
+ #define MULTI_VALUE_SEP "\001"
+ 
+ 
  typedef enum UserAuth
  {
  	uaReject,
***************
*** 89,93 **** extern int check_usermap(const char *usermap_name,
--- 93,100 ----
  			  const char *pg_role, const char *auth_user,
  			  bool case_sensitive);
  extern bool pg_isblank(const char c);
+ extern void tokenize_file(const char *filename, FILE *file,
+ 			  List **lines, List **line_nums, const char **keywords);
+ extern void free_lines(List **lines, List **line_nums);
  
  #endif   /* HBA_H */
*** a/src/include/replication/walprotocol.h
--- b/src/include/replication/walprotocol.h
***************
*** 50,53 **** typedef struct
--- 50,63 ----
   */
  #define MAX_SEND_SIZE (XLOG_BLCKSZ * 16)
  
+ /*
+  * Body for a WAL acknowledgment message (message type 'l'). This is wrapped
+  * within a CopyData message at the FE/BE protocol level.
+  */
+ typedef struct
+ {
+ 	/* End of WAL replicated to the standby */
+ 	XLogRecPtr	ackEnd;
+ } WalAckMessageData;
+ 
  #endif   /* _WALPROTOCOL_H */
*** a/src/include/replication/walreceiver.h
--- b/src/include/replication/walreceiver.h
***************
*** 26,31 **** extern bool am_walreceiver;
--- 26,38 ----
  #define MAXCONNINFO		1024
  
  /*
+  * MAXSTANDBYNAME: maximum size of standby name.
+  *
+  * XXX: Should this move to pg_config_manual.h?
+  */
+ #define MAXSTANDBYNAME	64
+ 
+ /*
   * Values for WalRcv->walRcvState.
   */
  typedef enum
***************
*** 71,89 **** typedef struct
  	 */
  	char		conninfo[MAXCONNINFO];
  
  	slock_t		mutex;			/* locks shared variables shown above */
  } WalRcvData;
  
  extern WalRcvData *WalRcv;
  
  /* libpqwalreceiver hooks */
! typedef bool (*walrcv_connect_type) (char *conninfo, XLogRecPtr startpoint);
  extern PGDLLIMPORT walrcv_connect_type walrcv_connect;
  
  typedef bool (*walrcv_receive_type) (int timeout, unsigned char *type,
  												 char **buffer, int *len);
  extern PGDLLIMPORT walrcv_receive_type walrcv_receive;
  
  typedef void (*walrcv_disconnect_type) (void);
  extern PGDLLIMPORT walrcv_disconnect_type walrcv_disconnect;
  
--- 78,106 ----
  	 */
  	char		conninfo[MAXCONNINFO];
  
+ 	/*
+ 	 * standby name; is used for the master to determine replication mode
+ 	 * from standbys configuration file.
+ 	 */
+ 	char		standbyName[MAXSTANDBYNAME];
+ 
  	slock_t		mutex;			/* locks shared variables shown above */
  } WalRcvData;
  
  extern WalRcvData *WalRcv;
  
  /* libpqwalreceiver hooks */
! typedef bool (*walrcv_connect_type) (char *conninfo, XLogRecPtr startpoint,
! 									 char *standbyName);
  extern PGDLLIMPORT walrcv_connect_type walrcv_connect;
  
  typedef bool (*walrcv_receive_type) (int timeout, unsigned char *type,
  												 char **buffer, int *len);
  extern PGDLLIMPORT walrcv_receive_type walrcv_receive;
  
+ typedef void (*walrcv_send_type) (const char *buffer, int nbytes);
+ extern PGDLLIMPORT walrcv_send_type walrcv_send;
+ 
  typedef void (*walrcv_disconnect_type) (void);
  extern PGDLLIMPORT walrcv_disconnect_type walrcv_disconnect;
  
***************
*** 93,99 **** extern void WalRcvShmemInit(void);
  extern void ShutdownWalRcv(void);
  extern bool WalRcvInProgress(void);
  extern XLogRecPtr WaitNextXLogAvailable(XLogRecPtr recptr, bool *finished);
! extern void RequestXLogStreaming(XLogRecPtr recptr, const char *conninfo);
  extern XLogRecPtr GetWalRcvWriteRecPtr(XLogRecPtr *latestChunkStart);
  
  #endif   /* _WALRECEIVER_H */
--- 110,117 ----
  extern void ShutdownWalRcv(void);
  extern bool WalRcvInProgress(void);
  extern XLogRecPtr WaitNextXLogAvailable(XLogRecPtr recptr, bool *finished);
! extern void RequestXLogStreaming(XLogRecPtr recptr, const char *conninfo,
! 								 const char *standbyName);
  extern XLogRecPtr GetWalRcvWriteRecPtr(XLogRecPtr *latestChunkStart);
  
  #endif   /* _WALRECEIVER_H */
*** a/src/include/replication/walsender.h
--- b/src/include/replication/walsender.h
***************
*** 22,27 **** typedef struct WalSnd
--- 22,30 ----
  {
  	pid_t		pid;			/* this walsender's process id, or 0 */
  	XLogRecPtr	sentPtr;		/* WAL has been sent up to this point */
+ 	XLogRecPtr	ackdPtr;		/* WAL has been replicated up to this point */
+ 
+ 	ReplicationMode	rplMode;	/* replication mode */
  
  	slock_t		mutex;			/* locks shared variables shown above */
  } WalSnd;
***************
*** 36,49 **** extern WalSndCtlData *WalSndCtl;
--- 39,65 ----
  
  /* global state */
  extern bool am_walsender;
+ extern char *standby_name;
  
  /* user-settable parameters */
  extern int	WalSndDelay;
  extern int	max_wal_senders;
  
+ /* struct definition for standbys configuration file */
+ typedef struct
+ {
+ 	int			linenumber;
+ 	char	   *standbyName;
+ 	ReplicationMode	rplMode;
+ } StandbysLine;
+ 
  extern int	WalSenderMain(void);
  extern void WalSndSignals(void);
  extern Size WalSndShmemSize(void);
  extern void WalSndShmemInit(void);
+ extern void WaitXLogSend(XLogRecPtr record);
+ 
+ extern bool check_standbys(void);
+ extern bool load_standbys(void);
  
  #endif   /* _WALSENDER_H */
*** a/src/interfaces/libpq/fe-connect.c
--- b/src/interfaces/libpq/fe-connect.c
***************
*** 254,259 **** static const PQconninfoOption PQconninfoOptions[] = {
--- 254,262 ----
  	{"replication", NULL, NULL, NULL,
  	"Replication", "D", 5},
  
+ 	{"standby_name", NULL, NULL, NULL,
+ 	"Standby-Name", "D", 64},
+ 
  	/* Terminating entry --- MUST BE LAST */
  	{NULL, NULL, NULL, NULL,
  	NULL, NULL, 0}
***************
*** 613,618 **** fillPGconn(PGconn *conn, PQconninfoOption *connOptions)
--- 616,623 ----
  #endif
  	tmp = conninfo_getval(connOptions, "replication");
  	conn->replication = tmp ? strdup(tmp) : NULL;
+ 	tmp = conninfo_getval(connOptions, "standby_name");
+ 	conn->standbyName = tmp ? strdup(tmp) : NULL;
  }
  
  /*
***************
*** 2622,2627 **** freePGconn(PGconn *conn)
--- 2627,2634 ----
  		free(conn->dbName);
  	if (conn->replication)
  		free(conn->replication);
+ 	if (conn->standbyName)
+ 		free(conn->standbyName);
  	if (conn->pguser)
  		free(conn->pguser);
  	if (conn->pgpass)
*** a/src/interfaces/libpq/fe-exec.c
--- b/src/interfaces/libpq/fe-exec.c
***************
*** 2002,2007 **** PQnotifies(PGconn *conn)
--- 2002,2010 ----
  /*
   * PQputCopyData - send some data to the backend during COPY IN
   *
+  * This function can be called by walreceiver even during COPY OUT
+  * to send a message to the master.
+  *
   * Returns 1 if successful, 0 if data could not be sent (only possible
   * in nonblock mode), or -1 if an error occurs.
   */
***************
*** 2010,2016 **** PQputCopyData(PGconn *conn, const char *buffer, int nbytes)
  {
  	if (!conn)
  		return -1;
! 	if (conn->asyncStatus != PGASYNC_COPY_IN)
  	{
  		printfPQExpBuffer(&conn->errorMessage,
  						  libpq_gettext("no COPY in progress\n"));
--- 2013,2020 ----
  {
  	if (!conn)
  		return -1;
! 	if (conn->asyncStatus != PGASYNC_COPY_IN &&
! 		conn->asyncStatus != PGASYNC_COPY_OUT)
  	{
  		printfPQExpBuffer(&conn->errorMessage,
  						  libpq_gettext("no COPY in progress\n"));
*** a/src/interfaces/libpq/fe-protocol3.c
--- b/src/interfaces/libpq/fe-protocol3.c
***************
*** 1911,1916 **** build_startup_packet(const PGconn *conn, char *packet,
--- 1911,1918 ----
  		ADD_STARTUP_OPTION("database", conn->dbName);
  	if (conn->replication && conn->replication[0])
  		ADD_STARTUP_OPTION("replication", conn->replication);
+ 	if (conn->standbyName && conn->standbyName[0])
+ 		ADD_STARTUP_OPTION("standby_name", conn->standbyName);
  	if (conn->pgoptions && conn->pgoptions[0])
  		ADD_STARTUP_OPTION("options", conn->pgoptions);
  	if (conn->send_appname)
*** a/src/interfaces/libpq/libpq-int.h
--- b/src/interfaces/libpq/libpq-int.h
***************
*** 297,302 **** struct pg_conn
--- 297,303 ----
  	char	   *fbappname;		/* fallback application name */
  	char	   *dbName;			/* database name */
  	char	   *replication;	/* connect as the replication standby? */
+ 	char	   *standbyName;	/* standby name */
  	char	   *pguser;			/* Postgres username and password, if any */
  	char	   *pgpass;
  	char	   *keepalives;		/* use TCP keepalives? */