wakeup_waiters_v1.patch

*** a/doc/src/sgml/func.sgml
--- b/doc/src/sgml/func.sgml
***************
*** 13905,13910 **** SELECT set_config('log_statement_stats', 'off', false);
--- 13905,13917 ----
         <entry><type>boolean</type></entry>
         <entry>Terminate a backend</entry>
        </row>
+       <row>
+        <entry>
+         <literal><function>pg_wakeup_all_waiters()</function></literal>
+         </entry>
+        <entry><type>void</type></entry>
+        <entry>Wake up all the backends waiting for replication</entry>
+       </row>
       </tbody>
      </tgroup>
     </table>
***************
*** 13939,13944 **** SELECT set_config('log_statement_stats', 'off', false);
--- 13946,13956 ----
      subprocess.
     </para>
  
+    <para>
+     <function>pg_wakeup_all_waiters</> signals all the backends waiting
+     for replication, to wake up and complete the transaction.
+    </para>
+ 
     <indexterm>
      <primary>backup</primary>
     </indexterm>
*** a/src/backend/replication/syncrep.c
--- b/src/backend/replication/syncrep.c
***************
*** 72,78 **** static void SyncRepWaitOnQueue(XLogRecPtr XactCommitLSN);
  static void SyncRepQueueInsert(void);
  
  static int SyncRepGetStandbyPriority(void);
! static int SyncRepWakeQueue(void);
  
  /*
   * ===========================================================
--- 72,78 ----
  static void SyncRepQueueInsert(void);
  
  static int SyncRepGetStandbyPriority(void);
! static int SyncRepWakeQueue(bool wakeup_all);
  
  /*
   * ===========================================================
***************
*** 180,196 **** SyncRepWaitOnQueue(XLogRecPtr XactCommitLSN)
  			 * unlikely to be far enough, yet is possible. Next time we are
  			 * woken we should be more lucky.
  			 */
! 			if (XLByteLE(XactCommitLSN, walsndctl->lsn))
  			{
  				SHMQueueDelete(&(MyProc->syncrep_links));
  				LWLockRelease(SyncRepLock);
  
- 				/*
- 				 * Reset our waitLSN.
- 				 */
- 				MyProc->waitLSN.xlogid = 0;
- 				MyProc->waitLSN.xrecoff = 0;
- 
  				if (new_status)
  				{
  					/* Reset ps display */
--- 180,192 ----
  			 * unlikely to be far enough, yet is possible. Next time we are
  			 * woken we should be more lucky.
  			 */
! 			if (XLByteLE(XactCommitLSN, walsndctl->lsn) ||
! 				(MyProc->waitLSN.xlogid == 0 &&
! 				 MyProc->waitLSN.xrecoff == 0))
  			{
  				SHMQueueDelete(&(MyProc->syncrep_links));
  				LWLockRelease(SyncRepLock);
  
  				if (new_status)
  				{
  					/* Reset ps display */
***************
*** 347,353 **** SyncRepReleaseWaiters(void)
  		 * release up to this location.
  		 */
  		walsndctl->lsn = MyWalSnd->flush;
! 		numprocs = SyncRepWakeQueue();
  	}
  
  	LWLockRelease(SyncRepLock);
--- 343,349 ----
  		 * release up to this location.
  		 */
  		walsndctl->lsn = MyWalSnd->flush;
! 		numprocs = SyncRepWakeQueue(false);
  	}
  
  	LWLockRelease(SyncRepLock);
***************
*** 427,436 **** SyncRepGetStandbyPriority(void)
   * to be woken. We don't modify the queue, we leave that for individual
   * procs to release themselves.
   *
   * Must hold SyncRepLock
   */
  static int
! SyncRepWakeQueue(void)
  {
  	volatile WalSndCtlData *walsndctl = WalSndCtl;
  	PGPROC	*proc;
--- 423,434 ----
   * to be woken. We don't modify the queue, we leave that for individual
   * procs to release themselves.
   *
+  * If 'wakeup_all' is true, set the latches of all procs in the queue.
+  *
   * Must hold SyncRepLock
   */
  static int
! SyncRepWakeQueue(bool wakeup_all)
  {
  	volatile WalSndCtlData *walsndctl = WalSndCtl;
  	PGPROC	*proc;
***************
*** 445,454 **** SyncRepWakeQueue(void)
  		/*
  		 * Assume the queue is ordered by LSN
  		 */
! 		if (XLByteLT(walsndctl->lsn, proc->waitLSN))
  			return numprocs;
  
  		numprocs++;
  		SetLatch(&(proc->waitLatch));
  		proc = (PGPROC *) SHMQueueNext(&(WalSndCtl->SyncRepQueue),
  									   &(proc->syncrep_links),
--- 443,454 ----
  		/*
  		 * Assume the queue is ordered by LSN
  		 */
! 		if (!wakeup_all && XLByteLT(walsndctl->lsn, proc->waitLSN))
  			return numprocs;
  
  		numprocs++;
+ 		proc->waitLSN.xlogid = 0;
+ 		proc->waitLSN.xrecoff = 0;
  		SetLatch(&(proc->waitLatch));
  		proc = (PGPROC *) SHMQueueNext(&(WalSndCtl->SyncRepQueue),
  									   &(proc->syncrep_links),
***************
*** 457,459 **** SyncRepWakeQueue(void)
--- 457,477 ----
  
  	return numprocs;
  }
+ 
+ /*
+  * Wake up all the waiting backends
+  */
+ Datum
+ pg_wakeup_all_waiters(PG_FUNCTION_ARGS)
+ {
+ 	if (!superuser())
+ 		ereport(ERROR,
+ 				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+ 			(errmsg("must be superuser to signal other server processes"))));
+ 
+ 	LWLockAcquire(SyncRepLock, LW_EXCLUSIVE);
+ 	SyncRepWakeQueue(true);
+ 	LWLockRelease(SyncRepLock);
+ 
+ 	PG_RETURN_VOID();
+ }
*** a/src/include/catalog/pg_proc.h
--- b/src/include/catalog/pg_proc.h
***************
*** 2869,2874 **** DATA(insert OID = 3821 ( pg_last_xlog_replay_location	PGNSP PGUID 12 1 0 0 f f f
--- 2869,2876 ----
  DESCR("last xlog replay location");
  DATA(insert OID = 3830 ( pg_last_xact_replay_timestamp	PGNSP PGUID 12 1 0 0 f f f t f v 0 0 1184 "" _null_ _null_ _null_ _null_ pg_last_xact_replay_timestamp _null_ _null_ _null_ ));
  DESCR("timestamp of last replay xact");
+ DATA(insert OID = 3831 ( pg_wakeup_all_waiters		PGNSP PGUID 12 1 0 0 f f f t f v 0 0 2278 "" _null_ _null_ _null_ _null_ pg_wakeup_all_waiters _null_ _null_ _null_ ));
+ DESCR("wake up all waiters");
  
  DATA(insert OID = 3071 ( pg_xlog_replay_pause		PGNSP PGUID 12 1 0 0 f f f t f v 0 0 2278 "" _null_ _null_ _null_ _null_ pg_xlog_replay_pause _null_ _null_ _null_ ));
  DESCR("pause xlog replay");
*** a/src/include/replication/syncrep.h
--- b/src/include/replication/syncrep.h
***************
*** 34,37 **** extern void SyncRepCleanupAtProcExit(int code, Datum arg);
--- 34,40 ----
  extern void SyncRepInitConfig(void);
  extern void SyncRepReleaseWaiters(void);
  
+ /* system administration functions */
+ extern Datum pg_wakeup_all_waiters(PG_FUNCTION_ARGS);
+ 
  #endif   /* _SYNCREP_H */