resqueue: Fix statement leak for holdable cursors

This is a 6X backport of commit: 48d9db0

There were minor conflicts due to the use of FALSE v false for
holdsStrongLockCount, differences in the type for rsqcountlimit and
rsqcountvalue, and a change from session #4 -> #3 in the ans file.

Original commit message follows:

Creating cursors "with hold" would clean up the locallock prematurely
during transaction commit for the DECLARE CURSOR command:

RemoveLocalLock lock.c:1406
LockReleaseAll lock.c:2313
ProcReleaseLocks proc.c:900
ResourceOwnerReleaseInternal resowner.c:320
ResourceOwnerRelease resowner.c:225
CommitTransaction xact.c:2818

This would cause an early exit in ResLockRelease():
LOG: Resource queue %d: no lock to release

and then subsequently the assertion failure:
FailedAssertion(""!(SHMQueueEmpty(&(MyProc->myProcLocks[i])))"", File: ""proc.c"", Line: 994

indicating a failure to clean up the resource queue locks, leading to a
statement leak in the associated resource queue.

This behavior was inadvertently introduced when we merged upstream
commit 3cba899, for 6X_STABLE.

The fix here is to ensure we don't accidentally remove a LOCALLOCK
associated with a resource queue, in LockReleaseAll().

We also take the trouble to add more commentary about how we expect
LOCALLOCK fields to be used in the context of resource queues.

Co-authored-by: xuejing zhao <zxuejing@vmware.com>
Co-authored-by: Zhenghua Lyu <kainwen@gmail.com>

src/backend/storage/lmgr/lock.c

@@ -2271,6 +2271,8 @@ LockReleaseAll(LOCKMETHODID lockmethodid, bool allLocks)
 	int			partition;
 	bool		have_fast_path_lwlock = false;
 
+	Assert(lockmethodid != RESOURCE_LOCKMETHOD);
+
 	if (lockmethodid <= 0 || lockmethodid >= lengthof(LockMethods))
 		elog(ERROR, "unrecognized lock method: %d", lockmethodid);
 	lockMethodTable = LockMethods[lockmethodid];
@@ -2307,11 +2309,21 @@ LockReleaseAll(LOCKMETHODID lockmethodid, bool allLocks)
 		 * If the LOCALLOCK entry is unused, we must've run out of shared
 		 * memory while trying to set up this lock.  Just forget the local
 		 * entry.
+		 *
+		 * GPDB: Add an exception for resource queue based locallocks. Neither
+		 * do we maintain nLocks for them, nor do we use the resource owner
+		 * mechanism for them.
 		 */
 		if (locallock->nLocks == 0)
 		{
-			RemoveLocalLock(locallock);
-			continue;
+			if (locallock->lock &&
+				locallock->lock->tag.locktag_type == LOCKTAG_RESOURCE_QUEUE)
+				Assert(locallock->numLockOwners == 0);
+			else
+			{
+				RemoveLocalLock(locallock);
+				continue;
+			}
 		}
 
 		/* Ignore items that are not of the lockmethod to be removed */

src/backend/utils/resscheduler/resqueue.c

@@ -211,13 +211,16 @@ ResLockAcquire(LOCKTAG *locktag, ResPortalIncrement *incrementSet)
 		locallock->lock = NULL;
 		locallock->proclock = NULL;
 		locallock->hashcode = LockTagHashCode(&(localtag.lock));
-		locallock->istemptable = false;
+
+		/* must remain 0 for the entire lifecycle of the LOCALLOCK */
 		locallock->nLocks = 0;
 		locallock->numLockOwners = 0;
-		locallock->maxLockOwners = 8;
+
+		/* initialized but unused for the entire lifecycle of the LOCALLOCK */
+		locallock->istemptable = false;
 		locallock->holdsStrongLockCount = FALSE;
 		locallock->lockCleared = false;
-		locallock->lockOwners = NULL;
+		locallock->maxLockOwners = 8;
 		locallock->lockOwners = (LOCALLOCKOWNER *)
 			MemoryContextAlloc(TopMemoryContext, locallock->maxLockOwners * sizeof(LOCALLOCKOWNER));
 	}

src/include/storage/lock.h

@@ -460,6 +460,13 @@ typedef struct LOCALLOCKOWNER
 	int64		nLocks;			/* # of times held by this owner */
 } LOCALLOCKOWNER;
 
+/*
+ * GPDB: For resource queue based LOCALLOCKs, we don't maintain nLocks or any
+ * of the owner related fields, after their initialization in ResLockAcquire().
+ * We don't use the resource owner mechanism for resource queue based
+ * LOCALLOCKs. This is key to avoid inadvertently operating on these in upstream
+ * lock routines where such LOCALLOCKs aren't meant to be manipulated.
+ */
 typedef struct LOCALLOCK
 {
 	/* tag */

src/test/isolation2/expected/resource_queue.out

@@ -58,6 +58,34 @@ END
  1.0           | 0.0           
 (1 row)
 
+-- Introduce a holdable cursor.
+3:SET role role_concurrency_test;
+SET
+3:DECLARE c_hold CURSOR WITH HOLD FOR SELECT 1;
+DECLARE
+
+-- Sanity check: The holdable cursor should be accounted for in pg_locks.
+0:SELECT granted, locktype, mode FROM pg_locks where locktype = 'resource queue' and pid != pg_backend_pid();
+ granted | locktype       | mode          
+---------+----------------+---------------
+ t       | resource queue | ExclusiveLock 
+(1 row)
+
+3q: ... <quitting>
+
+-- Sanity check: Ensure that all locks were released.
+0:SELECT granted, locktype, mode FROM pg_locks where locktype = 'resource queue' and pid != pg_backend_pid();
+ granted | locktype | mode 
+---------+----------+------
+(0 rows)
+
+-- Sanity check: Ensure that the resource queue is now empty.
+0: SELECT rsqcountlimit, rsqcountvalue from pg_resqueue_status WHERE rsqname = 'rq_concurrency_test';
+ rsqcountlimit | rsqcountvalue 
+---------------+---------------
+ 1.0           | 0.0           
+(1 row)
+
 0:DROP role role_concurrency_test;
 DROP
 0:DROP RESOURCE QUEUE rq_concurrency_test;

src/test/isolation2/sql/resource_queue.sql

@@ -25,5 +25,20 @@
 -- Sanity check: Ensure that the resource queue is now empty.
 0: SELECT rsqcountlimit, rsqcountvalue from pg_resqueue_status WHERE rsqname = 'rq_concurrency_test';
 
+-- Introduce a holdable cursor.
+3:SET role role_concurrency_test;
+3:DECLARE c_hold CURSOR WITH HOLD FOR SELECT 1;
+
+-- Sanity check: The holdable cursor should be accounted for in pg_locks.
+0:SELECT granted, locktype, mode FROM pg_locks where locktype = 'resource queue' and pid != pg_backend_pid();
+
+3q:
+
+-- Sanity check: Ensure that all locks were released.
+0:SELECT granted, locktype, mode FROM pg_locks where locktype = 'resource queue' and pid != pg_backend_pid();
+
+-- Sanity check: Ensure that the resource queue is now empty.
+0: SELECT rsqcountlimit, rsqcountvalue from pg_resqueue_status WHERE rsqname = 'rq_concurrency_test';
+
 0:DROP role role_concurrency_test;
 0:DROP RESOURCE QUEUE rq_concurrency_test;