synch_rep_0722.patch

*** a/doc/src/sgml/config.sgml
--- b/doc/src/sgml/config.sgml
***************
*** 1933,1938 **** SET ENABLE_SEQSCAN TO OFF;
--- 1933,1965 ----
         </listitem>
        </varlistentry>
  
+       <varlistentry id="guc-quorum" xreflabel="quorum">
+        <term><varname>quorum</varname> (<type>integer</type>)</term>
+        <indexterm>
+         <primary><varname>quorum</> configuration parameter</primary>
+        </indexterm>
+        <listitem>
+        <para>
+         Specifies how many standby servers transaction commit will wait
+         for WAL records to be replicated to, before the command returns
+         a <quote>success</> indication to the client. The default value
+         is zero, which always doesn't make transaction commit wait for
+         replication without regard to <xref linkend="replication-mode">.
+         Also transaction commit always doesn't wait for replication to
+         asynchronous standby (i.e., its <varname>replication_mode</> is
+         set to <literal>async</>) without regard to this parameter.
+         A value of <literal>-1</> regards all connected standby servers
+         as synchronous.
+        </para>
+        <para>
+         This parameter can be changed at any time; the behavior for any
+         one transaction is determined by the setting in effect when it
+         commits.  It is therefore possible, and useful, to have some
+         transactions replicate synchronously and others asynchronously.
+        </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>
*** a/doc/src/sgml/high-availability.sgml
--- b/doc/src/sgml/high-availability.sgml
***************
*** 148,161 **** protocol to make nodes agree on a serializable transactional order.
       stream of write-ahead log (<acronym>WAL</>)
       records.  If the main server fails, the standby contains
       almost all of the data of the main server, and can be quickly
!      made the new master database server.  This is asynchronous and
!      can only be done for the entire database server.
      </para>
      <para>
       A PITR standby server can be implemented using file-based log shipping
       (<xref linkend="warm-standby">) or streaming replication (see
!      <xref linkend="streaming-replication">), or a combination of both. For
!      information on hot standby, see <xref linkend="hot-standby">.
      </para>
     </listitem>
    </varlistentry>
--- 148,163 ----
       stream of write-ahead log (<acronym>WAL</>)
       records.  If the main server fails, the standby contains
       almost all of the data of the main server, and can be quickly
!      made the new master database server.  This can only be done for
!      the entire database server.
      </para>
      <para>
       A PITR standby server can be implemented using file-based log shipping
       (<xref linkend="warm-standby">) or streaming replication (see
!      <xref linkend="streaming-replication">), or a combination of both.
!      While file-based log shipping is asynchronous, synchronization mode can
!      be chosen in streaming replication (see <xref linkend="replication-mode">).
!      For information on hot standby, see <xref linkend="hot-standby">.
      </para>
     </listitem>
    </varlistentry>
***************
*** 348,354 **** protocol to make nodes agree on a serializable transactional order.
       <entry>No master server overhead</entry>
       <entry align="center">&bull;</entry>
       <entry align="center"></entry>
!      <entry align="center">&bull;</entry>
       <entry align="center"></entry>
       <entry align="center">&bull;</entry>
       <entry align="center"></entry>
--- 350,356 ----
       <entry>No master server overhead</entry>
       <entry align="center">&bull;</entry>
       <entry align="center"></entry>
!      <entry align="center">Optional</entry>
       <entry align="center"></entry>
       <entry align="center">&bull;</entry>
       <entry align="center"></entry>
***************
*** 359,365 **** protocol to make nodes agree on a serializable transactional order.
       <entry>No waiting for multiple servers</entry>
       <entry align="center">&bull;</entry>
       <entry align="center"></entry>
!      <entry align="center">&bull;</entry>
       <entry align="center">&bull;</entry>
       <entry align="center"></entry>
       <entry align="center">&bull;</entry>
--- 361,367 ----
       <entry>No waiting for multiple servers</entry>
       <entry align="center">&bull;</entry>
       <entry align="center"></entry>
!      <entry align="center">Optional</entry>
       <entry align="center">&bull;</entry>
       <entry align="center"></entry>
       <entry align="center">&bull;</entry>
***************
*** 370,376 **** protocol to make nodes agree on a serializable transactional order.
       <entry>Master failure will never lose data</entry>
       <entry align="center">&bull;</entry>
       <entry align="center">&bull;</entry>
!      <entry align="center"></entry>
       <entry align="center"></entry>
       <entry align="center">&bull;</entry>
       <entry align="center"></entry>
--- 372,378 ----
       <entry>Master failure will never lose data</entry>
       <entry align="center">&bull;</entry>
       <entry align="center">&bull;</entry>
!      <entry align="center">Optional</entry>
       <entry align="center"></entry>
       <entry align="center">&bull;</entry>
       <entry align="center"></entry>
*** a/doc/src/sgml/protocol.sgml
--- b/doc/src/sgml/protocol.sgml
***************
*** 1337,1347 **** The commands accepted in walsender mode are:
    </varlistentry>
  
    <varlistentry>
!     <term>START_REPLICATION <replaceable>XXX</>/<replaceable>XXX</></term>
      <listitem>
       <para>
        Instructs server to start streaming WAL, starting at
!       WAL position <replaceable>XXX</>/<replaceable>XXX</>.
        The server can reply with an error, e.g. if the requested section of WAL
        has already been recycled. On success, server responds with a
        CopyOutResponse message, and then starts to stream WAL to the frontend.
--- 1337,1348 ----
    </varlistentry>
  
    <varlistentry>
!     <term>START_REPLICATION <replaceable>XXX</>/<replaceable>XXX</> SYNC_LEVEL <replaceable>mode-string</></term>
      <listitem>
       <para>
        Instructs server to start streaming WAL, starting at
!       WAL position <replaceable>XXX</>/<replaceable>XXX</>
!       with the replication mode <replaceable>mode-string</>.
        The server can reply with an error, e.g. if the requested section of WAL
        has already been recycled. On success, server responds with a
        CopyOutResponse message, and then starts to stream WAL to the frontend.
***************
*** 1360,1365 **** The commands accepted in walsender mode are:
--- 1361,1401 ----
        <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/doc/src/sgml/recovery-config.sgml
--- b/doc/src/sgml/recovery-config.sgml
***************
*** 280,285 **** restore_command = 'copy "C:\\server\\archivedir\\%f" "%p"'  # Windows
--- 280,343 ----
           </para>
          </listitem>
         </varlistentry>
+        <varlistentry id="replication-mode" xreflabel="replication_mode">
+         <term><varname>replication_mode</varname> (<type>string</type>)</term>
+         <indexterm>
+           <primary><varname>replication_mode</> recovery parameter</primary>
+         </indexterm>
+         <listitem>
+          <para>
+           Specifies the replication mode which can control how long transaction
+           commit on the master server waits for replication before the command
+           returns a <quote>success</> indication to the client. Valid modes are:
+          <itemizedlist>
+           <listitem>
+           <para>
+            <literal>async</> (doesn't make transaction commit wait for replication,
+            i.e., asynchronous replication)
+           </para>
+           </listitem>
+           <listitem>
+           <para>
+            <literal>recv</> (makes transaction commit wait until the standby has
+            received WAL records)
+           </para>
+           </listitem>
+           <listitem>
+           <para>
+            <literal>fsync</> (makes transaction commit wait until the standby has
+            received and flushed WAL records to disk)
+           </para>
+           </listitem>
+           <listitem>
+           <para>
+            <literal>replay</> (makes transaction commit wait until the standby has
+            replayed WAL records after receiving and flushing them to disk)
+           </para>
+           </listitem>
+          </itemizedlist>
+          </para>
+          <para>
+           In asynchronous replication, there can be a delay between when a
+           <quote>success</> is reported to the client and when the transaction
+           is really guaranteed to be safe against a failover. Setting this
+           parameter to <literal>async</> does not create any risk of database
+           inconsistency: a crash at the master server might result in some recent
+           allegedly-committed transactions being lost at the standby server,
+           but the database state of the standby will be just the same as if
+           those transactions had been aborted cleanly. So, turning
+           <varname>replication_mode</> <literal>async</> can be a useful
+           alternative when performance is more important than exact certainty
+           about the durability of a transaction.
+           The default setting is <literal>async</>.
+          </para>
+          <para>
+           If <xref linkend="guc-quorum"> is set to zero in the master server,
+           transaction commit always doesn't wait for replication without regard
+           to this parameter.
+          </para>
+         </listitem>
+        </varlistentry>
         <varlistentry id="trigger-file" xreflabel="trigger_file">
          <term><varname>trigger_file</varname> (<type>string</type>)</term>
          <indexterm>
*** a/src/backend/access/transam/recovery.conf.sample
--- b/src/backend/access/transam/recovery.conf.sample
***************
*** 100,105 ****
--- 100,110 ----
  #primary_conninfo = ''		# e.g. 'host=localhost port=5432'
  #
  #
+ # Specifies the synchronization mode of replication.
+ #
+ #replication_mode = 'async'		# 'async', 'recv', 'fsync' or 'replay'
+ #
+ #
  # By default, a standby server keeps streaming XLOG records from the
  # primary indefinitely. If you want to stop streaming and finish recovery,
  # opening up the system in read/write mode, specify path to a trigger file.
*** 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"
***************
*** 1088,1093 **** RecordTransactionCommit(void)
--- 1089,1106 ----
  	/* 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 and quorum != 0. This operation has to be performed after
+ 	 * the COMMIT record is generated and before other transactions know that
+ 	 * this one has 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 && quorum != 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,202 ----
  static bool StandbyMode = false;
  static char *PrimaryConnInfo = NULL;
  static char *TriggerFile = NULL;
+ const struct config_enum_entry replication_mode_options[] = {
+ 	{"async", REPLICATION_MODE_ASYNC, false},
+ 	{"recv", REPLICATION_MODE_RECV, false},
+ 	{"fsync", REPLICATION_MODE_FSYNC, false},
+ 	{"replay", REPLICATION_MODE_REPLAY, false},
+ 	{NULL, 0, false}
+ };
+ int	rplMode = REPLICATION_MODE_ASYNC;
  
  /* if recoveryStopsHere returns true, it saves actual stop xid/time here */
  static TransactionId recoveryStopXid;
***************
*** 5258,5263 **** readRecoveryCommandFile(void)
--- 5266,5291 ----
  					(errmsg("trigger_file = '%s'",
  							TriggerFile)));
  		}
+ 		else if (strcmp(tok1, "replication_mode") == 0)
+ 		{
+ 			const struct config_enum_entry *entry;
+ 
+ 			for (entry = replication_mode_options; entry->name; entry++)
+ 			{
+ 				if (strcmp(tok2, entry->name) == 0)
+ 				{
+ 					rplMode = entry->val;
+ 					break;
+ 				}
+ 			}
+ 			if (entry->name == NULL)
+ 				ereport(FATAL,
+ 						(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ 						 errmsg("invalid value for parameter \"replication_mode\": \"%s\"",
+ 								tok2)));
+ 			ereport(DEBUG2,
+ 					(errmsg("replication_mode = '%s'", tok2)));
+ 		}
  		else
  			ereport(FATAL,
  					(errmsg("unrecognized recovery parameter \"%s\"",
***************
*** 6867,6872 **** GetFlushRecPtr(void)
--- 6895,6917 ----
  }
  
  /*
+  * 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
***************
*** 8824,8837 **** 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();
--- 8869,8878 ----
  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();
***************
*** 9463,9469 **** retry:
  						{
  							RequestXLogStreaming(
  									  fetching_ckpt ? RedoStartLSN : *RecPtr,
! 												 PrimaryConnInfo);
  							continue;
  						}
  					}
--- 9504,9510 ----
  						{
  							RequestXLogStreaming(
  									  fetching_ckpt ? RedoStartLSN : *RecPtr,
! 												 PrimaryConnInfo, rplMode);
  							continue;
  						}
  					}
*** a/src/backend/libpq/be-secure.c
--- b/src/backend/libpq/be-secure.c
***************
*** 71,76 ****
--- 71,86 ----
  #endif
  #endif   /* USE_SSL */
  
+ #ifdef HAVE_POLL_H
+ #include <poll.h>
+ #endif
+ #ifdef HAVE_SYS_POLL_H
+ #include <sys/poll.h>
+ #endif
+ #ifdef HAVE_SYS_SELECT_H
+ #include <sys/select.h>
+ #endif
+ 
  #include "libpq/libpq.h"
  #include "tcop/tcopprot.h"
  
***************
*** 397,402 **** wloop:
--- 407,472 ----
  	return n;
  }
  
+ /*
+  * Checks a socket, using poll or select, for data to be read.
+  * Returns >0 if there is data to read, 0 if it timed out, -1
+  * if an error occurred (including the interrupt).
+  *
+  * Timeout is specified in millisec. Timeout is infinite if
+  * timeout_ms is negative. Timeout is immediate (no blocking)
+  * if timeout_ms is 0.
+  *
+  * If SSL is in use, the SSL buffer is checked prior to
+  * checking the socket for read data directly.
+  *
+  * This function is based on pqSocketCheck and pqSocketPoll.
+  */
+ int
+ secure_poll(Port *port, int timeout_ms)
+ {
+ #ifdef USE_SSL
+ 	/* Check for SSL library buffering read bytes */
+ 	if (port->ssl && SSL_pending(port->ssl) > 0)
+ 	{
+ 		/* short-circuit the select */
+ 		return 1;
+ 	}
+ #endif
+ 
+ 	{
+ 		/* We use poll(2) if available, otherwise select(2) */
+ #ifdef HAVE_POLL
+ 		struct pollfd input_fd;
+ 
+ 		input_fd.fd = port->sock;
+ 		input_fd.events = POLLIN | POLLERR;
+ 		input_fd.revents = 0;
+ 
+ 		return poll(&input_fd, 1, timeout_ms);
+ #else							/* !HAVE_POLL */
+ 
+ 		fd_set		input_mask;
+ 		struct timeval timeout;
+ 		struct timeval *ptr_timeout;
+ 
+ 		FD_ZERO(&input_mask);
+ 		FD_SET(port->sock, &input_mask);
+ 
+ 		if (timeout_ms < 0)
+ 			ptr_timeout = NULL;
+ 		else
+ 		{
+ 			timeout.tv_sec	= timeout_ms / 1000;
+ 			timeout.tv_usec	= (timeout_ms % 1000) * 1000;
+ 			ptr_timeout		= &timeout;
+ 		}
+ 
+ 		return select(port->sock + 1, &input_mask,
+ 					  NULL, NULL, ptr_timeout);
+ #endif   /* HAVE_POLL */
+ 	}
+ }
+ 
  /* ------------------------------------------------------------ */
  /*						  SSL specific code						*/
  /* ------------------------------------------------------------ */
*** a/src/backend/libpq/pqcomm.c
--- b/src/backend/libpq/pqcomm.c
***************
*** 56,61 ****
--- 56,62 ----
   *		pq_putbytes		- send bytes to connection (not flushed until pq_flush)
   *		pq_flush		- flush pending output
   *		pq_getbyte_if_available - get a byte if available without blocking
+  *		pq_wait			- wait until we can read connection
   *
   * message-level I/O (and old-style-COPY-OUT cruft):
   *		pq_putmessage	- send a normal message (suppressed in COPY OUT mode)
***************
*** 911,916 **** pq_getbyte_if_available(unsigned char *c)
--- 912,959 ----
  }
  
  /* --------------------------------
+  *		pq_wait		- wait until we can read the connection socket.
+  *
+  *		returns >0 if there is data to read, 0 if it timed out or
+  *		interrupted, -1	if an error occurred.
+  *
+  *		this function is based on pqSocketCheck.
+  * --------------------------------
+  */
+ int
+ pq_wait(int timeout_ms)
+ {
+ 	int	result;
+ 
+ 	if (!MyProcPort)
+ 		return -1;
+ 	if (MyProcPort->sock < 0)
+ 	{
+ 		ereport(COMMERROR,
+ 				(errcode_for_socket_access(),
+ 				 errmsg("socket not open")));
+ 		return -1;
+ 	}
+ 
+ 	result = secure_poll(MyProcPort, timeout_ms);
+ 	if (result < 0)
+ 	{
+ 		if (errno == EINTR)
+ 			return 0;		/* interrupted */
+ 
+ 		/*
+ 		 * XXX: Should we suppress duplicate log messages also here,
+ 		 * like internal_flush?
+ 		 */
+ 		ereport(COMMERROR,
+ 				(errcode_for_socket_access(),
+ 				 errmsg("select() failed: %m")));
+ 	}
+ 
+ 	return result;
+ }
+ 
+ /* --------------------------------
   *		pq_getbytes		- get a known number of bytes from connection
   *
   *		returns 0 if OK, EOF if trouble
*** 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,
! 							 const char *modename);
  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,81 **** _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;
--- 78,84 ----
   * Establish the connection to the primary server for XLOG streaming
   */
  static bool
! libpqrcv_connect(char *conninfo, XLogRecPtr startpoint, const char *modename)
  {
  	char		conninfo_repl[MAXCONNINFO + 37];
  	char	   *primary_sysid;
***************
*** 154,161 **** libpqrcv_connect(char *conninfo, XLogRecPtr startpoint)
  	ThisTimeLineID = primary_tli;
  
  	/* Start streaming from the point requested by startup process */
! 	snprintf(cmd, sizeof(cmd), "START_REPLICATION %X/%X",
! 			 startpoint.xlogid, startpoint.xrecoff);
  	res = libpqrcv_PQexec(cmd);
  	if (PQresultStatus(res) != PGRES_COPY_OUT)
  	{
--- 157,164 ----
  	ThisTimeLineID = primary_tli;
  
  	/* Start streaming from the point requested by startup process */
! 	snprintf(cmd, sizeof(cmd), "START_REPLICATION %X/%X SYNC_LEVEL %s",
! 			 startpoint.xlogid, startpoint.xrecoff, modename);
  	res = libpqrcv_PQexec(cmd);
  	if (PQresultStatus(res) != PGRES_COPY_OUT)
  	{
***************
*** 398,400 **** libpqrcv_receive(int timeout, unsigned char *type, char **buffer, int *len)
--- 401,418 ----
  
  	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))));
+ }
*** 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];
+ 	const char *modename;
  	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);
+ 	rplMode	= walrcv->rplMode;
  	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");
  
  	/*
***************
*** 259,267 **** WalReceiverMain(void)
  	/* Unblock signals (they were blocked when the postmaster forked us) */
  	PG_SETMASK(&UnBlockSig);
  
  	/* Establish the connection to the primary for XLOG streaming */
  	EnableWalRcvImmediateExit();
! 	walrcv_connect(conninfo, startpoint);
  	DisableWalRcvImmediateExit();
  
  	/* Loop until end-of-streaming or error */
--- 264,287 ----
  	/* Unblock signals (they were blocked when the postmaster forked us) */
  	PG_SETMASK(&UnBlockSig);
  
+ 	/* Lookup the name of replication mode with the "rplMode" */
+ 	{
+ 		const struct config_enum_entry *entry;
+ 
+ 		for (entry = replication_mode_options; entry->name; entry++)
+ 		{
+ 			if (entry->val == rplMode)
+ 			{
+ 				modename = entry->name;
+ 				break;
+ 			}
+ 		}
+ 		Assert(entry->name != NULL);
+ 	}
+ 
  	/* Establish the connection to the primary for XLOG streaming */
  	EnableWalRcvImmediateExit();
! 	walrcv_connect(conninfo, startpoint, modename);
  	DisableWalRcvImmediateExit();
  
  	/* Loop until end-of-streaming or error */
***************
*** 311,316 **** WalReceiverMain(void)
--- 331,355 ----
  			 */
  			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)
--- 445,463 ----
  				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)
--- 575,588 ----
  
  		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)
--- 601,624 ----
  		}
  	}
  }
+ 
+ /* 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,178 ----
  /*
   * Request postmaster to start walreceiver.
   *
!  * recptr indicates the position where streaming should begin, conninfo is
!  * a libpq connection string to use, and mode is a replication mode.
   */
  void
! RequestXLogStreaming(XLogRecPtr recptr, const char *conninfo, int mode)
  {
  	/* use volatile pointer to prevent code rearrangement */
  	volatile WalRcvData *walrcv = WalRcv;
***************
*** 196,201 **** RequestXLogStreaming(XLogRecPtr recptr, const char *conninfo)
--- 196,202 ----
  		strlcpy((char *) walrcv->conninfo, conninfo, MAXCONNINFO);
  	else
  		walrcv->conninfo[0] = '\0';
+ 	walrcv->rplMode = mode;
  	walrcv->walRcvState = WALRCV_STARTING;
  	walrcv->startTime = now;
  
*** a/src/backend/replication/walsender.c
--- b/src/backend/replication/walsender.c
***************
*** 65,73 **** bool		am_walsender = false;		/* Am I a walsender process ? */
  /* User-settable parameters for walsender */
  int			max_wal_senders = 0;	/* the maximum number of concurrent walsenders */
  int			WalSndDelay = 200;	/* max sleep time between some actions */
  
! #define NAPTIME_PER_CYCLE 100000L		/* max sleep time between cycles
! 										 * (100ms) */
  
  /*
   * These variables are used similarly to openLogFile/Id/Seg/Off,
--- 65,73 ----
  /* User-settable parameters for walsender */
  int			max_wal_senders = 0;	/* the maximum number of concurrent walsenders */
  int			WalSndDelay = 200;	/* max sleep time between some actions */
+ int			quorum = 0;		/* the maximum number of synchronous walsenders */
  
! #define NAPTIME_PER_CYCLE 100L	/* max sleep time between cycles (100ms) */
  
  /*
   * These variables are used similarly to openLogFile/Id/Seg/Off,
***************
*** 84,89 **** static uint32 sendOff = 0;
--- 84,96 ----
   */
  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;
***************
*** 101,107 **** 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 */
--- 108,114 ----
  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);
  
  
  /* Main entry point for walsender process */
***************
*** 195,200 **** WalSndHandshake(void)
--- 202,209 ----
  		/* Handle the very limited subset of commands expected in this phase */
  		switch (firstchar)
  		{
+ 			char	modename[8];
+ 
  			case 'Q':			/* Query message */
  				{
  					const char *query_string;
***************
*** 255,262 **** WalSndHandshake(void)
  						ReadyForQuery(DestRemote);
  						/* ReadyForQuery did pq_flush for us */
  					}
! 					else if (sscanf(query_string, "START_REPLICATION %X/%X",
! 									&recptr.xlogid, &recptr.xrecoff) == 2)
  					{
  						StringInfoData buf;
  
--- 264,271 ----
  						ReadyForQuery(DestRemote);
  						/* ReadyForQuery did pq_flush for us */
  					}
! 					else if (sscanf(query_string, "START_REPLICATION %X/%X SYNC_LEVEL %8s",
! 									&recptr.xlogid, &recptr.xrecoff, modename) == 3)
  					{
  						StringInfoData buf;
  
***************
*** 277,282 **** WalSndHandshake(void)
--- 286,325 ----
  									(errcode(ERRCODE_CANNOT_CONNECT_NOW),
  									 errmsg("standby connections not allowed because wal_level=minimal")));
  
+ 						/* Verify that the specified replication mode is valid */
+ 						{
+ 							const struct config_enum_entry *entry;
+ 
+ 							for (entry = replication_mode_options; entry->name; entry++)
+ 							{
+ 								if (strcmp(modename, entry->name) == 0)
+ 								{
+ 									rplMode = entry->val;
+ 									break;
+ 								}
+ 							}
+ 							if (entry->name == NULL)
+ 								ereport(FATAL,
+ 										(errcode(ERRCODE_PROTOCOL_VIOLATION),
+ 										 errmsg("invalid replication mode: %s", modename)));
+ 						}
+ 						MyWalSnd->rplMode = rplMode;
+ 
+ 						/*
+ 						 * Update the current number of synchronous standbys if replication
+ 						 * mode is "synchronous"
+ 						 */
+ 						if (rplMode >= REPLICATION_MODE_RECV)
+ 						{
+ 							/* use volatile pointer to prevent code rearrangement */
+ 							volatile WalSndCtlData *walsndctl = WalSndCtl;
+ 
+ 							SpinLockAcquire(&walsndctl->info_lck);
+ 							walsndctl->num_sync_sbys++;
+ 							Assert(walsndctl->num_sync_sbys > 0);
+ 							SpinLockRelease(&walsndctl->info_lck);
+ 						}
+ 
  						/* Send a CopyOutResponse message, and start streaming */
  						pq_beginmessage(&buf, 'H');
  						pq_sendbyte(&buf, 0);
***************
*** 285,294 **** 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;
--- 328,337 ----
  						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;
***************
*** 322,364 **** 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)));
  	}
  }
  
--- 365,473 ----
  }
  
  /*
!  * 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);
  	}
  }
  
***************
*** 366,374 **** CheckClosedConnection(void)
--- 475,486 ----
  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
***************
*** 428,443 **** WalSndLoop(void)
  		 */
  		if (caughtup)
  		{
! 			remain = WalSndDelay * 1000L;
  			while (remain > 0)
  			{
  				/* Check for interrupts */
  				if (got_SIGHUP || shutdown_requested || ready_to_stop)
  					break;
  
! 				/* Sleep and check that the connection is still alive */
! 				pg_usleep(remain > NAPTIME_PER_CYCLE ? NAPTIME_PER_CYCLE : remain);
! 				CheckClosedConnection();
  
  				remain -= NAPTIME_PER_CYCLE;
  			}
--- 540,569 ----
  		 */
  		if (caughtup)
  		{
! 			remain = WalSndDelay;
  			while (remain > 0)
  			{
+ 				int		res;
+ 
  				/* Check for interrupts */
  				if (got_SIGHUP || shutdown_requested || ready_to_stop)
  					break;
  
! 				/*
! 				 * Check to see whether a message from the standby or an interrupt
! 				 * from other processes has arrived.
! 				 */
! 				res = pq_wait(remain > NAPTIME_PER_CYCLE ? NAPTIME_PER_CYCLE : remain);
! 				if (res < 0)
! 				{
! 					/* unexpected error or EOF */
! 					ereport(COMMERROR,
! 							(errcode(ERRCODE_PROTOCOL_VIOLATION),
! 							 errmsg("unexpected EOF on standby connection")));
! 					proc_exit(0);
! 				}
! 				if (res > 0)
! 					ProcessStreamMsgs(&input_message);
  
  				remain -= NAPTIME_PER_CYCLE;
  			}
***************
*** 496,501 **** InitWalSnd(void)
--- 622,628 ----
  			MyWalSnd = (WalSnd *) walsnd;
  			walsnd->pid = MyProcPid;
  			MemSet(&MyWalSnd->sentPtr, 0, sizeof(XLogRecPtr));
+ 			MemSet(&MyWalSnd->ackdPtr, 0, sizeof(XLogRecPtr));
  			SpinLockRelease(&walsnd->mutex);
  			break;
  		}
***************
*** 523,528 **** WalSndKill(int code, Datum arg)
--- 650,670 ----
  	 */
  	MyWalSnd->pid = 0;
  
+ 	/*
+ 	 * Update the current number of synchronous standbys if replication
+ 	 * mode is "synchronous"
+ 	 */
+ 	if (rplMode >= REPLICATION_MODE_RECV)
+ 	{
+ 		/* use volatile pointer to prevent code rearrangement */
+ 		volatile WalSndCtlData *walsndctl = WalSndCtl;
+ 
+ 		SpinLockAcquire(&walsndctl->info_lck);
+ 		Assert(walsndctl->num_sync_sbys > 0);
+ 		walsndctl->num_sync_sbys--;
+ 		SpinLockRelease(&walsndctl->info_lck);
+ 	}
+ 
  	/* WalSnd struct isn't mine anymore */
  	MyWalSnd = NULL;
  }
***************
*** 884,889 **** WalSndShmemInit(void)
--- 1026,1032 ----
  	{
  		/* First time through, so initialize */
  		MemSet(WalSndCtl, 0, WalSndShmemSize());
+ 		SpinLockInit(&WalSndCtl->info_lck);
  
  		for (i = 0; i < max_wal_senders; i++)
  		{
***************
*** 895,937 **** 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
--- 1038,1109 ----
  }
  
  /*
!  * Ensure that all xlog records through the given position is
!  * replicated to the standby
   */
! void
! WaitXLogSend(XLogRecPtr record)
  {
! 	Assert(max_wal_senders > 0);
! 	Assert(quorum != 0);
  
! 	for (;;)
  	{
  		/* use volatile pointer to prevent code rearrangement */
! 		volatile WalSndCtlData *walsndctl = WalSndCtl;
! 		int		i;
! 		int		already_acked = 0;
! 		bool	unacked = false;
  
! 		/* Don't need to wait if there is no synchronous standbys */
! 		if (walsndctl->num_sync_sbys == 0)
! 			return;
  
! 		/*
! 		 * Count walsenders which have already received the ACK meaning
! 		 * completion of replication up to the given position. If the
! 		 * sum is more than or equal to the "quorum", the backend breaks
! 		 * out of this loop and returns a "success" of the transaction
! 		 * to a client.
! 		 */
! 		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 count inactive or asynchronous walsenders */
! 			if (walsnd->pid == 0 ||
! 				walsnd->rplMode == REPLICATION_MODE_ASYNC)
! 				continue;
! 
! 			SpinLockAcquire(&walsnd->mutex);
! 			recptr = walsnd->ackdPtr;
! 			SpinLockRelease(&walsnd->mutex);
  
! 			if ((recptr.xlogid == 0 && recptr.xrecoff == 0) ||
! 				XLByteLT(recptr, record))
! 			{
! 				unacked = true;
! 				continue;
! 			}
! 
! 			if (quorum > -1 && ++already_acked >= quorum)
! 				return;
! 		}
! 
! 		/*
! 		 * If synchronous walsender was not found in the WalSnd array,
! 		 * we no longer need to wait. This can happen if all synchronous
! 		 * walsenders are terminated while searching the array.
! 		 *
! 		 * If all synchronous walsenders have already received the ACK,
! 		 * we no longer need to wait, too. This can happen when the
! 		 * "quorum" is more than max_wal_senders or is equal to -1.
! 		 */
! 		if (!unacked)
! 			return;
! 
! 		pg_usleep(100000L);     /* 100ms */
  	}
  }
*** a/src/backend/utils/misc/guc.c
--- b/src/backend/utils/misc/guc.c
***************
*** 1782,1787 **** static struct config_int ConfigureNamesInt[] =
--- 1782,1796 ----
  	},
  
  	{
+ 		{"quorum", PGC_USERSET, WAL_REPLICATION,
+ 			gettext_noop("Sets the maximum number of synchronous standby servers."),
+ 			NULL
+ 		},
+ 		&quorum,
+ 		0, -1, INT_MAX / 4, NULL, NULL
+ 	},
+ 
+ 	{
  		{"commit_delay", PGC_USERSET, WAL_SETTINGS,
  			gettext_noop("Sets the delay in microseconds between transaction commit and "
  						 "flushing WAL to disk."),
*** a/src/backend/utils/misc/postgresql.conf.sample
--- b/src/backend/utils/misc/postgresql.conf.sample
***************
*** 188,193 ****
--- 188,195 ----
  #max_wal_senders = 0		# max number of walsender processes
  #wal_sender_delay = 200ms	# walsender cycle time, 1-10000 milliseconds
  #wal_keep_segments = 0		# in logfile segments, 16MB each; 0 disables
+ #quorum = 0				# max number of synchronous standbys;
+ 					# -1 regards all connected standbys as synchronous
  #vacuum_defer_cleanup_age = 0	# number of xacts by which cleanup is delayed
  
  # - Standby Servers -
*** a/src/include/access/xlog.h
--- b/src/include/access/xlog.h
***************
*** 15,20 ****
--- 15,21 ----
  #include "access/xlogdefs.h"
  #include "lib/stringinfo.h"
  #include "storage/buf.h"
+ #include "utils/guc.h"
  #include "utils/pg_crc.h"
  #include "utils/timestamp.h"
  
***************
*** 261,266 **** typedef struct CheckpointStatsData
--- 262,291 ----
  
  extern CheckpointStatsData CheckpointStats;
  
+ /*
+  * Synchronization mode of replication. These modes identify how far
+  * we 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 server.
+  *
+  * REPLICATION_MODE_FSYNC makes transaction commit wait for XLOG
+  * records to be received and fsync'd on the standby server.
+  *
+  * REPLICATION_MODE_REPLAY makes transaction commit wait for XLOG
+  * records to be received, fsync'd and replayed on the standby server.
+  */
+ #define	REPLICATION_MODE_ASYNC	0
+ #define	REPLICATION_MODE_RECV	1
+ #define	REPLICATION_MODE_FSYNC	2
+ #define	REPLICATION_MODE_REPLAY	3
+ extern int rplMode;
+ extern const struct config_enum_entry replication_mode_options[];
+ 
+ 
  extern XLogRecPtr XLogInsert(RmgrId rmid, uint8 info, XLogRecData *rdata);
  extern void XLogFlush(XLogRecPtr RecPtr);
  extern void XLogBackgroundFlush(void);
***************
*** 298,303 **** extern void XLogPutNextOid(Oid nextOid);
--- 323,329 ----
  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);
  
*** a/src/include/libpq/libpq.h
--- b/src/include/libpq/libpq.h
***************
*** 58,63 **** extern int	pq_getmessage(StringInfo s, int maxlen);
--- 58,64 ----
  extern int	pq_getbyte(void);
  extern int	pq_peekbyte(void);
  extern int	pq_getbyte_if_available(unsigned char *c);
+ extern int	pq_wait(int timeout_ms);
  extern int	pq_putbytes(const char *s, size_t len);
  extern int	pq_flush(void);
  extern int	pq_putmessage(char msgtype, const char *s, size_t len);
***************
*** 74,78 **** extern int	secure_open_server(Port *port);
--- 75,80 ----
  extern void secure_close(Port *port);
  extern ssize_t secure_read(Port *port, void *ptr, size_t len);
  extern ssize_t secure_write(Port *port, void *ptr, size_t len);
+ extern int	secure_poll(Port *port, int timeout_ms);
  
  #endif   /* LIBPQ_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
***************
*** 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;
  
--- 71,99 ----
  	 */
  	char		conninfo[MAXCONNINFO];
  
+ 	/*
+ 	 * replication mode; controls how long transaction commit on the primary
+ 	 * server waits for replication.
+ 	 */
+ 	int		rplMode;
+ 
  	slock_t		mutex;			/* locks shared variables shown above */
  } WalRcvData;
  
  extern WalRcvData *WalRcv;
  
  /* libpqwalreceiver hooks */
! typedef bool (*walrcv_connect_type) (char *conninfo, XLogRecPtr startpoint,
! 									 const char *modename);
  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 */
--- 103,110 ----
  extern void ShutdownWalRcv(void);
  extern bool WalRcvInProgress(void);
  extern XLogRecPtr WaitNextXLogAvailable(XLogRecPtr recptr, bool *finished);
! extern void RequestXLogStreaming(XLogRecPtr recptr, const char *conninfo,
! 								 int mode);
  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 */
+ 
+ 	int		rplMode;	/* replication mode */
  
  	slock_t		mutex;			/* locks shared variables shown above */
  } WalSnd;
***************
*** 29,34 **** typedef struct WalSnd
--- 32,41 ----
  /* There is one WalSndCtl struct for the whole database cluster */
  typedef struct
  {
+ 	int			num_sync_sbys;	/* current # of synchronous standbys */
+ 
+ 	slock_t		info_lck;		/* protects the variable shown above */
+ 
  	WalSnd		walsnds[1];		/* VARIABLE LENGTH ARRAY */
  } WalSndCtlData;
  
***************
*** 40,49 **** extern bool am_walsender;
--- 47,58 ----
  /* user-settable parameters */
  extern int	WalSndDelay;
  extern int	max_wal_senders;
+ extern int	quorum;
  
  extern int	WalSenderMain(void);
  extern void WalSndSignals(void);
  extern Size WalSndShmemSize(void);
  extern void WalSndShmemInit(void);
+ extern void WaitXLogSend(XLogRecPtr recptr);
  
  #endif   /* _WALSENDER_H */
*** 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
   *
+  * Note that this function might be called by walreceiver even during
+  * COPY OUT to send a message to XLOG stream.
+  *
   * 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"));