hs_release_old_locks.v2.patch

*** a/src/backend/storage/ipc/procarray.c
--- b/src/backend/storage/ipc/procarray.c
***************
*** 499,505 **** ProcArrayApplyRecoveryInfo(RunningTransactions running)
  	 * Remove stale transactions, if any.
  	 */
  	ExpireOldKnownAssignedTransactionIds(running->oldestRunningXid);
! 	StandbyReleaseOldLocks(running->oldestRunningXid);
  
  	/*
  	 * If our snapshot is already valid, nothing else to do...
--- 499,505 ----
  	 * Remove stale transactions, if any.
  	 */
  	ExpireOldKnownAssignedTransactionIds(running->oldestRunningXid);
! 	StandbyReleaseOldLocks(running->xcnt, running->xids);
  
  	/*
  	 * If our snapshot is already valid, nothing else to do...
***************
*** 554,565 **** ProcArrayApplyRecoveryInfo(RunningTransactions running)
  	 */
  
  	/*
- 	 * Release any locks belonging to old transactions that are not running
- 	 * according to the running-xacts record.
- 	 */
- 	StandbyReleaseOldLocks(running->nextXid);
- 
- 	/*
  	 * Nobody else is running yet, but take locks anyhow
  	 */
  	LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
--- 554,559 ----
*** a/src/backend/storage/ipc/standby.c
--- b/src/backend/storage/ipc/standby.c
***************
*** 525,531 **** StandbyAcquireAccessExclusiveLock(TransactionId xid, Oid dbOid, Oid relOid)
  	LOCKTAG		locktag;
  
  	/* Already processed? */
! 	if (TransactionIdDidCommit(xid) || TransactionIdDidAbort(xid))
  		return;
  
  	elog(trace_recovery(DEBUG4),
--- 525,533 ----
  	LOCKTAG		locktag;
  
  	/* Already processed? */
! 	if (!TransactionIdIsValid(xid) ||
! 		TransactionIdDidCommit(xid) ||
! 		TransactionIdDidAbort(xid))
  		return;
  
  	elog(trace_recovery(DEBUG4),
***************
*** 607,640 **** StandbyReleaseLockTree(TransactionId xid, int nsubxids, TransactionId *subxids)
  }
  
  /*
!  * StandbyReleaseLocksMany
!  *		Release standby locks held by XIDs < removeXid
!  *
!  * If keepPreparedXacts is true, keep prepared transactions even if
!  * they're older than removeXid
   */
! static void
! StandbyReleaseLocksMany(TransactionId removeXid, bool keepPreparedXacts)
  {
  	ListCell   *cell,
  			   *prev,
  			   *next;
  	LOCKTAG		locktag;
  
- 	/*
- 	 * Release all matching locks.
- 	 */
  	prev = NULL;
  	for (cell = list_head(RecoveryLockList); cell; cell = next)
  	{
  		xl_standby_lock *lock = (xl_standby_lock *) lfirst(cell);
  
  		next = lnext(cell);
  
! 		if (!TransactionIdIsValid(removeXid) || TransactionIdPrecedes(lock->xid, removeXid))
  		{
- 			if (keepPreparedXacts && StandbyTransactionIdIsPrepared(lock->xid))
- 				continue;
  			elog(trace_recovery(DEBUG4),
  				 "releasing recovery lock: xid %u db %u rel %u",
  				 lock->xid, lock->dbOid, lock->relOid);
--- 609,691 ----
  }
  
  /*
!  * Called at end of recovery and when we see a shutdown checkpoint.
   */
! void
! StandbyReleaseAllLocks(void)
! {
! 	ListCell   *cell,
! 			   *prev,
! 			   *next;
! 	LOCKTAG		locktag;
! 
! 	elog(trace_recovery(DEBUG2), "release all standby locks");
! 
! 	prev = NULL;
! 	for (cell = list_head(RecoveryLockList); cell; cell = next)
! 	{
! 		xl_standby_lock *lock = (xl_standby_lock *) lfirst(cell);
! 
! 		next = lnext(cell);
! 
! 		elog(trace_recovery(DEBUG4),
! 			 "releasing recovery lock: xid %u db %u rel %u",
! 			 lock->xid, lock->dbOid, lock->relOid);
! 		SET_LOCKTAG_RELATION(locktag, lock->dbOid, lock->relOid);
! 		if (!LockRelease(&locktag, AccessExclusiveLock, true))
! 			elog(LOG,
! 				 "RecoveryLockList contains entry for lock no longer recorded by lock manager: xid %u database %u relation %u",
! 				 lock->xid, lock->dbOid, lock->relOid);
! 		RecoveryLockList = list_delete_cell(RecoveryLockList, cell, prev);
! 		pfree(lock);
! 	}
! }
! 
! /*
!  * StandbyReleaseOldLocks
!  *		Release standby locks held by XIDs that aren't running, as long
!  *		as they're not prepared transactions.
!  */
! void
! StandbyReleaseOldLocks(int nxids, TransactionId *xids)
  {
  	ListCell   *cell,
  			   *prev,
  			   *next;
  	LOCKTAG		locktag;
  
  	prev = NULL;
  	for (cell = list_head(RecoveryLockList); cell; cell = next)
  	{
  		xl_standby_lock *lock = (xl_standby_lock *) lfirst(cell);
+ 		bool	remove = false;
  
  		next = lnext(cell);
  
! 		Assert(TransactionIdIsValid(lock->xid));
! 
! 		if (StandbyTransactionIdIsPrepared(lock->xid))
! 			remove = false;
! 		else
! 		{
! 			int		i;
! 			bool	found = false;
! 
! 			for (i = 0; i < nxids; i++)
! 			{
! 				if (lock->xid == xids[i])
! 					found = true;
! 			}
! 
! 			/*
! 			 * If its not a running transaction, remove it.
! 			 */
! 			if (!found)
! 				remove = true;
! 		}
! 
! 		if (remove)
  		{
  			elog(trace_recovery(DEBUG4),
  				 "releasing recovery lock: xid %u db %u rel %u",
  				 lock->xid, lock->dbOid, lock->relOid);
***************
*** 652,678 **** StandbyReleaseLocksMany(TransactionId removeXid, bool keepPreparedXacts)
  }
  
  /*
-  * Called at end of recovery and when we see a shutdown checkpoint.
-  */
- void
- StandbyReleaseAllLocks(void)
- {
- 	elog(trace_recovery(DEBUG2), "release all standby locks");
- 	StandbyReleaseLocksMany(InvalidTransactionId, false);
- }
- 
- /*
-  * StandbyReleaseOldLocks
-  *		Release standby locks held by XIDs < removeXid, as long
-  *		as they're not prepared transactions.
-  */
- void
- StandbyReleaseOldLocks(TransactionId removeXid)
- {
- 	StandbyReleaseLocksMany(removeXid, true);
- }
- 
- /*
   * --------------------------------------------------------------------
   *		Recovery handling for Rmgr RM_STANDBY_ID
   *
--- 703,708 ----
***************
*** 813,818 **** standby_desc(StringInfo buf, uint8 xl_info, char *rec)
--- 843,855 ----
   * Later, when we apply the running xact data we must be careful to ignore
   * transactions already committed, since those commits raced ahead when
   * making WAL entries.
+  *
+  * The loose timing also means that locks may be recorded that have a
+  * zero xid, since xids are removed from procs before locks are removed.
+  * So we must prune the lock list down to ensure we hold locks only for
+  * currently running xids, performed by StandbyReleaseOldLocks().
+  * Zero xids should no longer be possible, but we may be replaying WAL
+  * from a time when they were possible.
   */
  void
  LogStandbySnapshot(TransactionId *nextXid)
*** a/src/backend/storage/lmgr/lock.c
--- b/src/backend/storage/lmgr/lock.c
***************
*** 3190,3195 **** GetRunningTransactionLocks(int *nlocks)
--- 3190,3205 ----
  			PGPROC	   *proc = proclock->tag.myProc;
  			PGXACT	   *pgxact = &ProcGlobal->allPgXact[proc->pgprocno];
  			LOCK	   *lock = proclock->tag.myLock;
+ 			TransactionId xid = pgxact->xid;
+ 
+ 			/*
+ 			 * Don't record locks for transactions if we know they have already
+ 			 * issued their WAL record for commit but not yet released lock.
+ 			 * It is still possible that we see locks held by already complete
+ 			 * transactions, if they haven't yet zeroed their xids.
+ 			 */
+ 			if (!TransactionIdIsValid(xid))
+ 				continue;
  
  			accessExclusiveLocks[index].xid = pgxact->xid;
  			accessExclusiveLocks[index].dbOid = lock->tag.locktag_field1;
*** a/src/include/storage/standby.h
--- b/src/include/storage/standby.h
***************
*** 48,54 **** extern void StandbyAcquireAccessExclusiveLock(TransactionId xid, Oid dbOid, Oid
  extern void StandbyReleaseLockTree(TransactionId xid,
  					   int nsubxids, TransactionId *subxids);
  extern void StandbyReleaseAllLocks(void);
! extern void StandbyReleaseOldLocks(TransactionId removeXid);
  
  /*
   * XLOG message types
--- 48,54 ----
  extern void StandbyReleaseLockTree(TransactionId xid,
  					   int nsubxids, TransactionId *subxids);
  extern void StandbyReleaseAllLocks(void);
! extern void StandbyReleaseOldLocks(int nxids, TransactionId *xids);
  
  /*
   * XLOG message types