change_poll_loop_in_walsender_0727.patch

*** 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"
***************
*** 1025,1030 **** EndPrepare(GlobalTransaction gxact)
--- 1026,1038 ----
  
  	/* If we crash now, we have prepared: WAL replay will fix things */
  
+ 	/*
+ 	 * Wake up all walsenders to send WAL up to the PREPARE record
+ 	 * immediately if replication is enabled
+ 	 */
+ 	if (max_wal_senders > 0)
+ 		WalSndWakeup();
+ 
  	/* write correct CRC and close file */
  	if ((write(fd, &statefile_crc, sizeof(pg_crc32))) != sizeof(pg_crc32))
  	{
***************
*** 2005,2010 **** RecordTransactionCommitPrepared(TransactionId xid,
--- 2013,2025 ----
  	/* Flush XLOG to disk */
  	XLogFlush(recptr);
  
+ 	/*
+ 	 * Wake up all walsenders to send WAL up to the COMMIT PREPARED record
+ 	 * immediately if replication is enabled
+ 	 */
+ 	if (max_wal_senders > 0)
+ 		WalSndWakeup();
+ 
  	/* Mark the transaction committed in pg_clog */
  	TransactionIdCommitTree(xid, nchildren, children);
  
***************
*** 2078,2083 **** RecordTransactionAbortPrepared(TransactionId xid,
--- 2093,2105 ----
  	XLogFlush(recptr);
  
  	/*
+ 	 * Wake up all walsenders to send WAL up to the ABORT PREPARED record
+ 	 * immediately if replication is enabled
+ 	 */
+ 	if (max_wal_senders > 0)
+ 		WalSndWakeup();
+ 
+ 	/*
  	 * Mark the transaction aborted in clog.  This is not absolutely necessary
  	 * but we may as well do it while we are here.
  	 */
*** 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"
***************
*** 1068,1073 **** RecordTransactionCommit(void)
--- 1069,1081 ----
  		XLogFlush(XactLastRecEnd);
  
  		/*
+ 		 * Wake up all walsenders to send WAL up to the COMMIT record
+ 		 * immediately if replication is enabled
+ 		 */
+ 		if (max_wal_senders > 0)
+ 			WalSndWakeup();
+ 
+ 		/*
  		 * Now we may update the CLOG, if we wrote a COMMIT record above
  		 */
  		if (markXidCommitted)
*** a/src/backend/replication/walsender.c
--- b/src/backend/replication/walsender.c
***************
*** 87,98 **** static XLogRecPtr sentPtr = {0, 0};
--- 87,100 ----
  /* 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 xlogsend_requested = false;
  static volatile sig_atomic_t ready_to_stop = false;
  
  /* Signal handlers */
  static void WalSndSigHupHandler(SIGNAL_ARGS);
  static void WalSndShutdownHandler(SIGNAL_ARGS);
  static void WalSndQuickDieHandler(SIGNAL_ARGS);
+ static void WalSndXLogSendHandler(SIGNAL_ARGS);
  static void WalSndLastCycleHandler(SIGNAL_ARGS);
  
  /* Prototypes for private functions */
***************
*** 418,423 **** WalSndLoop(void)
--- 420,428 ----
  			proc_exit(0);
  		}
  
+ 		if (xlogsend_requested)
+ 			xlogsend_requested = false;
+ 
  		/*
  		 * If we had sent all accumulated WAL in last round, nap for the
  		 * configured time before retrying.
***************
*** 433,439 **** WalSndLoop(void)
  			while (remain > 0)
  			{
  				/* Check for interrupts */
! 				if (got_SIGHUP || shutdown_requested || ready_to_stop)
  					break;
  
  				/* Sleep and check that the connection is still alive */
--- 438,445 ----
  			while (remain > 0)
  			{
  				/* Check for interrupts */
! 				if (got_SIGHUP || shutdown_requested || xlogsend_requested ||
! 					ready_to_stop)
  					break;
  
  				/* Sleep and check that the connection is still alive */
***************
*** 665,670 **** XLogSend(char *msgbuf, bool *caughtup)
--- 671,678 ----
  	Size		nbytes;
  	WalDataMessageHeader msghdr;
  
+ 	MyWalSnd->sndrqst = false;
+ 
  	/*
  	 * Attempt to send all data that's already been written out and fsync'd to
  	 * disk.  We cannot go further than what's been written out given the
***************
*** 678,684 **** XLogSend(char *msgbuf, bool *caughtup)
  	/* Quick exit if nothing to do */
  	if (XLByteLE(SendRqstPtr, sentPtr))
  	{
! 		*caughtup = true;
  		return true;
  	}
  
--- 686,692 ----
  	/* Quick exit if nothing to do */
  	if (XLByteLE(SendRqstPtr, sentPtr))
  	{
! 		*caughtup = !MyWalSnd->sndrqst;
  		return true;
  	}
  
***************
*** 718,730 **** XLogSend(char *msgbuf, bool *caughtup)
  	if (XLByteLE(SendRqstPtr, endptr))
  	{
  		endptr = SendRqstPtr;
! 		*caughtup = true;
  	}
  	else
  	{
  		/* round down to page boundary. */
  		endptr.xrecoff -= (endptr.xrecoff % XLOG_BLCKSZ);
  		*caughtup = false;
  	}
  
  	nbytes = endptr.xrecoff - startptr.xrecoff;
--- 726,739 ----
  	if (XLByteLE(SendRqstPtr, endptr))
  	{
  		endptr = SendRqstPtr;
! 		*caughtup = !MyWalSnd->sndrqst;
  	}
  	else
  	{
  		/* round down to page boundary. */
  		endptr.xrecoff -= (endptr.xrecoff % XLOG_BLCKSZ);
  		*caughtup = false;
+ 		MyWalSnd->sndrqst = true;
  	}
  
  	nbytes = endptr.xrecoff - startptr.xrecoff;
***************
*** 828,833 **** WalSndQuickDieHandler(SIGNAL_ARGS)
--- 837,849 ----
  	exit(2);
  }
  
+ /* SIGUSR1: set flag to send WAL records */
+ static void
+ WalSndXLogSendHandler(SIGNAL_ARGS)
+ {
+ 	xlogsend_requested = true;
+ }
+ 
  /* SIGUSR2: set flag to do a last cycle and shut down afterwards */
  static void
  WalSndLastCycleHandler(SIGNAL_ARGS)
***************
*** 847,853 **** WalSndSignals(void)
  	pqsignal(SIGQUIT, WalSndQuickDieHandler);	/* hard crash time */
  	pqsignal(SIGALRM, SIG_IGN);
  	pqsignal(SIGPIPE, SIG_IGN);
! 	pqsignal(SIGUSR1, SIG_IGN); /* not used */
  	pqsignal(SIGUSR2, WalSndLastCycleHandler);	/* request a last cycle and
  												 * shutdown */
  
--- 863,869 ----
  	pqsignal(SIGQUIT, WalSndQuickDieHandler);	/* hard crash time */
  	pqsignal(SIGALRM, SIG_IGN);
  	pqsignal(SIGPIPE, SIG_IGN);
! 	pqsignal(SIGUSR1, WalSndXLogSendHandler);	/* request WAL sending */
  	pqsignal(SIGUSR2, WalSndLastCycleHandler);	/* request a last cycle and
  												 * shutdown */
  
***************
*** 895,900 **** WalSndShmemInit(void)
--- 911,937 ----
  	}
  }
  
+ /* Wake all walsenders up by signaling */
+ void
+ WalSndWakeup(void)
+ {
+ 	int		i;
+ 
+ 	for (i = 0; i < max_wal_senders; i++)
+ 	{
+ 		/* use volatile pointer to prevent code rearrangement */
+ 		volatile WalSnd *walsnd = &WalSndCtl->walsnds[i];
+ 		pid_t	pid = walsnd->pid;
+ 
+ 		/* we don't need to signal to waking walsenders */
+ 		if (pid != 0 && !walsnd->sndrqst)
+ 		{
+ 			walsnd->sndrqst = true;
+ 			kill(pid, SIGUSR1);
+ 		}
+ 	}
+ }
+ 
  /*
   * This isn't currently used for anything. Monitoring tools might be
   * interested in the future, and we'll need something like this in the
*** a/src/include/replication/walsender.h
--- b/src/include/replication/walsender.h
***************
*** 12,17 ****
--- 12,21 ----
  #ifndef _WALSENDER_H
  #define _WALSENDER_H
  
+ #include "postgres.h"
+ 
+ #include <signal.h>
+ 
  #include "access/xlog.h"
  #include "storage/spin.h"
  
***************
*** 23,28 **** typedef struct WalSnd
--- 27,42 ----
  	pid_t		pid;			/* this walsender's process id, or 0 */
  	XLogRecPtr	sentPtr;		/* WAL has been sent up to this point */
  
+ 	/*
+ 	 * The request for WAL sending has already been sent? If false,
+ 	 * we set sndrqst to true and signal walsender to send WAL up to
+ 	 * the current WAL write location immeidately. If true, since
+ 	 * walsender is ready for sending all the currently-written WAL,
+ 	 * we don't need to signal. This value is used to suppress
+ 	 * redundant signaling.
+ 	 */
+ 	sig_atomic_t	sndrqst;
+ 
  	slock_t		mutex;			/* locks shared variables shown above */
  } WalSnd;
  
***************
*** 45,49 **** extern int	WalSenderMain(void);
--- 59,64 ----
  extern void WalSndSignals(void);
  extern Size WalSndShmemSize(void);
  extern void WalSndShmemInit(void);
+ extern void WalSndWakeup(void);
  
  #endif   /* _WALSENDER_H */