v10-0002-use-linked-list-in-ResourceOwner-to-hold-LOCALLO.patch

From 5ed0d4008afdfddf9b6fc082b12b1e51938402bd Mon Sep 17 00:00:00 2001
From: Heikki Linnakangas <heikki.linnakangas@iki.fi>
Date: Mon, 19 Jan 2026 15:50:48 +0200
Subject: [PATCH v10 2/5] use linked list in ResourceOwner to hold
 LOCALLOCKOWNERS

---
 src/backend/storage/lmgr/lock.c       | 362 +++++++++++---------------
 src/backend/utils/resowner/resowner.c | 144 +++++-----
 src/include/storage/lock.h            |  17 +-
 src/include/utils/resowner.h          |   6 +-
 4 files changed, 246 insertions(+), 283 deletions(-)

diff --git a/src/backend/storage/lmgr/lock.c b/src/backend/storage/lmgr/lock.c
index 7f0cd784f79..fe2821e31f7 100644
--- a/src/backend/storage/lmgr/lock.c
+++ b/src/backend/storage/lmgr/lock.c
@@ -322,6 +322,8 @@ static HTAB *LockMethodLockHash;
 static HTAB *LockMethodProcLockHash;
 static HTAB *LockMethodLocalHash;
 
+/* A memory context for storing LOCALLOCKOWNER structs */
+static MemoryContext LocalLockOwnerContext;
 
 /* private state for error cleanup */
 static LOCALLOCK *StrongLockInProgress;
@@ -416,8 +418,8 @@ static void BeginStrongLockAcquire(LOCALLOCK *locallock, uint32 fasthashcode);
 static void FinishStrongLockAcquire(void);
 static ProcWaitStatus WaitOnLock(LOCALLOCK *locallock, ResourceOwner owner);
 static void waitonlock_error_callback(void *arg);
-static void ReleaseLockIfHeld(LOCALLOCK *locallock, bool sessionLock);
-static void LockReassignOwner(LOCALLOCK *locallock, ResourceOwner parent);
+static void ReleaseLockIfHeld(LOCALLOCKOWNER *locallockowner, bool sessionLock);
+static void LockReassignOwner(LOCALLOCKOWNER *locallockowner, ResourceOwner parent);
 static bool UnGrantLock(LOCK *lock, LOCKMODE lockmode,
 						PROCLOCK *proclock, LockMethod lockMethodTable);
 static void CleanUpLock(LOCK *lock, PROCLOCK *proclock,
@@ -517,6 +519,17 @@ InitLockManagerAccess(void)
 									  16,
 									  &info,
 									  HASH_ELEM | HASH_BLOBS);
+
+	/*
+	 * Create a slab context for storing LOCALLOCKOWNERs.  Slab seems like a
+	 * good context type for this as it will manage fragmentation better than
+	 * aset.c contexts and it will free() excess memory rather than maintain
+	 * excessively long freelists after a large surge in locking requirements.
+	 */
+	LocalLockOwnerContext = SlabContextCreate(TopMemoryContext,
+											  "LOCALLOCKOWNER context",
+											  SLAB_DEFAULT_BLOCK_SIZE,
+											  sizeof(LOCALLOCKOWNER));
 }
 
 
@@ -906,25 +919,7 @@ LockAcquireExtended(const LOCKTAG *locktag,
 		locallock->nLocks = 0;
 		locallock->holdsStrongLockCount = false;
 		locallock->lockCleared = false;
-		locallock->numLockOwners = 0;
-		locallock->maxLockOwners = 8;
-		locallock->lockOwners = NULL;	/* in case next line fails */
-		locallock->lockOwners = (LOCALLOCKOWNER *)
-			MemoryContextAlloc(TopMemoryContext,
-							   locallock->maxLockOwners * sizeof(LOCALLOCKOWNER));
-	}
-	else
-	{
-		/* Make sure there will be room to remember the lock */
-		if (locallock->numLockOwners >= locallock->maxLockOwners)
-		{
-			int			newsize = locallock->maxLockOwners * 2;
-
-			locallock->lockOwners = (LOCALLOCKOWNER *)
-				repalloc(locallock->lockOwners,
-						 newsize * sizeof(LOCALLOCKOWNER));
-			locallock->maxLockOwners = newsize;
-		}
+		dlist_init(&locallock->locallockowners);
 	}
 	hashcode = locallock->hashcode;
 
@@ -1475,17 +1470,18 @@ CheckAndSetLockHeld(LOCALLOCK *locallock, bool acquired)
 static void
 RemoveLocalLock(LOCALLOCK *locallock)
 {
-	int			i;
+	dlist_mutable_iter iter;
 
-	for (i = locallock->numLockOwners - 1; i >= 0; i--)
+	dlist_foreach_modify(iter, &locallock->locallockowners)
 	{
-		if (locallock->lockOwners[i].owner != NULL)
-			ResourceOwnerForgetLock(locallock->lockOwners[i].owner, locallock);
+		LOCALLOCKOWNER *locallockowner = dlist_container(LOCALLOCKOWNER, locallock_node, iter.cur);
+
+		dlist_delete(&locallockowner->locallock_node);
+		if (locallockowner->owner != NULL)
+			ResourceOwnerForgetLock(locallockowner);
+		pfree(locallockowner);
 	}
-	locallock->numLockOwners = 0;
-	if (locallock->lockOwners != NULL)
-		pfree(locallock->lockOwners);
-	locallock->lockOwners = NULL;
+	Assert(dlist_is_empty(&locallock->locallockowners));
 
 	if (locallock->holdsStrongLockCount)
 	{
@@ -1791,26 +1787,38 @@ CleanUpLock(LOCK *lock, PROCLOCK *proclock,
 static void
 GrantLockLocal(LOCALLOCK *locallock, ResourceOwner owner)
 {
-	LOCALLOCKOWNER *lockOwners = locallock->lockOwners;
-	int			i;
+	LOCALLOCKOWNER *locallockowner;
+	dlist_iter	iter;
 
-	Assert(locallock->numLockOwners < locallock->maxLockOwners);
 	/* Count the total */
 	locallock->nLocks++;
 	/* Count the per-owner lock */
-	for (i = 0; i < locallock->numLockOwners; i++)
+	dlist_foreach(iter, &locallock->locallockowners)
 	{
-		if (lockOwners[i].owner == owner)
+		locallockowner = dlist_container(LOCALLOCKOWNER, locallock_node, iter.cur);
+
+		if (locallockowner->owner == owner)
 		{
-			lockOwners[i].nLocks++;
+			locallockowner->nLocks++;
 			return;
 		}
 	}
-	lockOwners[i].owner = owner;
-	lockOwners[i].nLocks = 1;
-	locallock->numLockOwners++;
+	locallockowner = MemoryContextAlloc(LocalLockOwnerContext, sizeof(LOCALLOCKOWNER));
+	locallockowner->owner = owner;
+	locallockowner->nLocks = 1;
+	locallockowner->locallock = locallock;
+
+	dlist_push_tail(&locallock->locallockowners, &locallockowner->locallock_node);
+
 	if (owner != NULL)
-		ResourceOwnerRememberLock(owner, locallock);
+		ResourceOwnerRememberLock(owner, locallockowner);
+	else
+	{
+		/* session lock */
+		LOCKMETHODID lockmethodid = LOCALLOCK_LOCKMETHOD(*locallockowner->locallock);
+
+		Assert(lockmethodid > 0 && lockmethodid <= 2);
+	}
 
 	/* Indicate that the lock is acquired for certain types of locks. */
 	CheckAndSetLockHeld(locallock, true);
@@ -2148,9 +2156,9 @@ LockRelease(const LOCKTAG *locktag, LOCKMODE lockmode, bool sessionLock)
 	 * Decrease the count for the resource owner.
 	 */
 	{
-		LOCALLOCKOWNER *lockOwners = locallock->lockOwners;
 		ResourceOwner owner;
-		int			i;
+		dlist_mutable_iter iter;
+		bool		found = false;
 
 		/* Identify owner for lock */
 		if (sessionLock)
@@ -2158,24 +2166,25 @@ LockRelease(const LOCKTAG *locktag, LOCKMODE lockmode, bool sessionLock)
 		else
 			owner = CurrentResourceOwner;
 
-		for (i = locallock->numLockOwners - 1; i >= 0; i--)
+		dlist_foreach_modify(iter, &locallock->locallockowners)
 		{
-			if (lockOwners[i].owner == owner)
+			LOCALLOCKOWNER *locallockowner = dlist_container(LOCALLOCKOWNER, locallock_node, iter.cur);
+
+			if (locallockowner->owner == owner)
 			{
-				Assert(lockOwners[i].nLocks > 0);
-				if (--lockOwners[i].nLocks == 0)
+				Assert(locallockowner->nLocks > 0);
+				if (--locallockowner->nLocks == 0)
 				{
+					dlist_delete(&locallockowner->locallock_node);
 					if (owner != NULL)
-						ResourceOwnerForgetLock(owner, locallock);
-					/* compact out unused slot */
-					locallock->numLockOwners--;
-					if (i < locallock->numLockOwners)
-						lockOwners[i] = lockOwners[locallock->numLockOwners];
+						ResourceOwnerForgetLock(locallockowner);
+					pfree(locallockowner);
 				}
+				found = true;
 				break;
 			}
 		}
-		if (i < 0)
+		if (!found)
 		{
 			/* don't release a lock belonging to another owner */
 			elog(WARNING, "you don't own a lock of type %s",
@@ -2368,29 +2377,30 @@ LockReleaseAll(LOCKMETHODID lockmethodid, bool allLocks)
 		 */
 		if (!allLocks)
 		{
-			LOCALLOCKOWNER *lockOwners = locallock->lockOwners;
+			dlist_mutable_iter iter;
+			int			session_locks = 0;
 
-			/* If session lock is above array position 0, move it down to 0 */
-			for (i = 0; i < locallock->numLockOwners; i++)
+			dlist_foreach_modify(iter, &locallock->locallockowners)
 			{
-				if (lockOwners[i].owner == NULL)
-					lockOwners[0] = lockOwners[i];
+				LOCALLOCKOWNER *locallockowner = dlist_container(LOCALLOCKOWNER, locallock_node, iter.cur);
+
+				if (locallockowner->owner != NULL)
+				{
+					dlist_delete(&locallockowner->locallock_node);
+					ResourceOwnerForgetLock(locallockowner);
+					pfree(locallockowner);
+				}
 				else
-					ResourceOwnerForgetLock(lockOwners[i].owner, locallock);
+					session_locks += locallockowner->nLocks;
 			}
 
-			if (locallock->numLockOwners > 0 &&
-				lockOwners[0].owner == NULL &&
-				lockOwners[0].nLocks > 0)
+			/* Fix the locallock to show just the session locks */
+			locallock->nLocks = session_locks;
+			if (session_locks > 0)
 			{
-				/* Fix the locallock to show just the session locks */
-				locallock->nLocks = lockOwners[0].nLocks;
-				locallock->numLockOwners = 1;
 				/* We aren't deleting this locallock, so done */
 				continue;
 			}
-			else
-				locallock->numLockOwners = 0;
 		}
 
 #ifdef USE_ASSERT_CHECKING
@@ -2590,52 +2600,41 @@ LockReleaseSession(LOCKMETHODID lockmethodid)
 
 	while ((locallock = (LOCALLOCK *) hash_seq_search(&status)) != NULL)
 	{
+		dlist_mutable_iter iter;
+
 		/* Ignore items that are not of the specified lock method */
 		if (LOCALLOCK_LOCKMETHOD(*locallock) != lockmethodid)
 			continue;
 
-		ReleaseLockIfHeld(locallock, true);
+		dlist_foreach_modify(iter, &locallock->locallockowners)
+		{
+			LOCALLOCKOWNER *locallockowner = dlist_container(LOCALLOCKOWNER, locallock_node, iter.cur);
+
+			if (locallockowner->owner == NULL)
+				ReleaseLockIfHeld(locallockowner, true);
+		}
 	}
 }
 
 /*
  * LockReleaseCurrentOwner
- *		Release all locks belonging to CurrentResourceOwner
- *
- * If the caller knows what those locks are, it can pass them as an array.
- * That speeds up the call significantly, when a lot of locks are held.
- * Otherwise, pass NULL for locallocks, and we'll traverse through our hash
- * table to find them.
+ *		Release all locks belonging to 'owner'
  */
 void
-LockReleaseCurrentOwner(LOCALLOCK **locallocks, int nlocks)
+LockReleaseCurrentOwner(struct ResourceOwnerData *owner, LOCALLOCKOWNER *locallockowner)
 {
-	if (locallocks == NULL)
-	{
-		HASH_SEQ_STATUS status;
-		LOCALLOCK  *locallock;
-
-		hash_seq_init(&status, LockMethodLocalHash);
+	Assert(locallockowner->owner == owner);
 
-		while ((locallock = (LOCALLOCK *) hash_seq_search(&status)) != NULL)
-			ReleaseLockIfHeld(locallock, false);
-	}
-	else
-	{
-		int			i;
-
-		for (i = nlocks - 1; i >= 0; i--)
-			ReleaseLockIfHeld(locallocks[i], false);
-	}
+	ReleaseLockIfHeld(locallockowner, false);
 }
 
 /*
  * ReleaseLockIfHeld
- *		Release any session-level locks on this lockable object if sessionLock
- *		is true; else, release any locks held by CurrentResourceOwner.
+ *		Release any session-level locks on this 'locallockowner' if sessionLock
+ *		is true; else, release any locks held by 'locallockowner'.
  *
  * It is tempting to pass this a ResourceOwner pointer (or NULL for session
- * locks), but without refactoring LockRelease() we cannot support releasing
+ * locks), but without refactoring LockRelease() we cannot support releasing XXX
  * locks belonging to resource owners other than CurrentResourceOwner.
  * If we were to refactor, it'd be a good idea to fix it so we don't have to
  * do a hashtable lookup of the locallock, too.  However, currently this
@@ -2643,52 +2642,39 @@ LockReleaseCurrentOwner(LOCALLOCK **locallocks, int nlocks)
  * convenience.
  */
 static void
-ReleaseLockIfHeld(LOCALLOCK *locallock, bool sessionLock)
+ReleaseLockIfHeld(LOCALLOCKOWNER *locallockowner, bool sessionLock)
 {
-	ResourceOwner owner;
-	LOCALLOCKOWNER *lockOwners;
-	int			i;
+	LOCALLOCK  *locallock = locallockowner->locallock;
 
-	/* Identify owner for lock (must match LockRelease!) */
+	/* release all references to the lock by this resource owner */
 	if (sessionLock)
-		owner = NULL;
+		Assert(locallockowner->owner == NULL);
 	else
-		owner = CurrentResourceOwner;
+		Assert(locallockowner->owner != NULL);
 
-	/* Scan to see if there are any locks belonging to the target owner */
-	lockOwners = locallock->lockOwners;
-	for (i = locallock->numLockOwners - 1; i >= 0; i--)
+	if (locallockowner->nLocks < locallock->nLocks)
 	{
-		if (lockOwners[i].owner == owner)
-		{
-			Assert(lockOwners[i].nLocks > 0);
-			if (lockOwners[i].nLocks < locallock->nLocks)
-			{
-				/*
-				 * We will still hold this lock after forgetting this
-				 * ResourceOwner.
-				 */
-				locallock->nLocks -= lockOwners[i].nLocks;
-				/* compact out unused slot */
-				locallock->numLockOwners--;
-				if (owner != NULL)
-					ResourceOwnerForgetLock(owner, locallock);
-				if (i < locallock->numLockOwners)
-					lockOwners[i] = lockOwners[locallock->numLockOwners];
-			}
-			else
-			{
-				Assert(lockOwners[i].nLocks == locallock->nLocks);
-				/* We want to call LockRelease just once */
-				lockOwners[i].nLocks = 1;
-				locallock->nLocks = 1;
-				if (!LockRelease(&locallock->tag.lock,
-								 locallock->tag.mode,
-								 sessionLock))
-					elog(WARNING, "ReleaseLockIfHeld: failed??");
-			}
-			break;
-		}
+		/*
+		 * We will still hold this lock after forgetting this ResourceOwner.
+		 */
+		locallock->nLocks -= locallockowner->nLocks;
+
+		dlist_delete(&locallockowner->locallock_node);
+		if (locallockowner->owner)
+			ResourceOwnerForgetLock(locallockowner);
+		pfree(locallockowner);
+	}
+	else
+	{
+		Assert(locallockowner->nLocks == locallock->nLocks);
+		/* We want to call LockRelease just once */
+		locallockowner->nLocks = 1;
+		locallock->nLocks = 1;
+
+		if (!LockRelease(&locallock->tag.lock,
+						 locallock->tag.mode,
+						 sessionLock))
+			elog(WARNING, "ReleaseLockIfHeld: failed??");
 	}
 }
 
@@ -2696,82 +2682,47 @@ ReleaseLockIfHeld(LOCALLOCK *locallock, bool sessionLock)
  * LockReassignCurrentOwner
  *		Reassign all locks belonging to CurrentResourceOwner to belong
  *		to its parent resource owner.
- *
- * If the caller knows what those locks are, it can pass them as an array.
- * That speeds up the call significantly, when a lot of locks are held
- * (e.g pg_dump with a large schema).  Otherwise, pass NULL for locallocks,
- * and we'll traverse through our hash table to find them.
  */
 void
-LockReassignCurrentOwner(LOCALLOCK **locallocks, int nlocks)
+LockReassignCurrentOwner(LOCALLOCKOWNER *locallockowner)
 {
 	ResourceOwner parent = ResourceOwnerGetParent(CurrentResourceOwner);
 
-	Assert(parent != NULL);
-
-	if (locallocks == NULL)
-	{
-		HASH_SEQ_STATUS status;
-		LOCALLOCK  *locallock;
-
-		hash_seq_init(&status, LockMethodLocalHash);
-
-		while ((locallock = (LOCALLOCK *) hash_seq_search(&status)) != NULL)
-			LockReassignOwner(locallock, parent);
-	}
-	else
-	{
-		int			i;
-
-		for (i = nlocks - 1; i >= 0; i--)
-			LockReassignOwner(locallocks[i], parent);
-	}
+	LockReassignOwner(locallockowner, parent);
 }
 
 /*
- * Subroutine of LockReassignCurrentOwner. Reassigns a given lock belonging to
- * CurrentResourceOwner to its parent.
+ * Subroutine of LockReassignCurrentOwner. Reassigns the given
+ *'locallockowner' to 'parent'.
  */
 static void
-LockReassignOwner(LOCALLOCK *locallock, ResourceOwner parent)
+LockReassignOwner(LOCALLOCKOWNER *locallockowner, ResourceOwner parent)
 {
-	LOCALLOCKOWNER *lockOwners;
-	int			i;
-	int			ic = -1;
-	int			ip = -1;
+	LOCALLOCK  *locallock = locallockowner->locallock;
+	dlist_iter	iter;
 
-	/*
-	 * Scan to see if there are any locks belonging to current owner or its
-	 * parent
-	 */
-	lockOwners = locallock->lockOwners;
-	for (i = locallock->numLockOwners - 1; i >= 0; i--)
+	ResourceOwnerForgetLock(locallockowner);
+
+	dlist_foreach(iter, &locallock->locallockowners)
 	{
-		if (lockOwners[i].owner == CurrentResourceOwner)
-			ic = i;
-		else if (lockOwners[i].owner == parent)
-			ip = i;
-	}
+		LOCALLOCKOWNER *parentlocalowner = dlist_container(LOCALLOCKOWNER, locallock_node, iter.cur);
 
-	if (ic < 0)
-		return;					/* no current locks */
+		Assert(parentlocalowner->locallock == locallock);
 
-	if (ip < 0)
-	{
-		/* Parent has no slot, so just give it the child's slot */
-		lockOwners[ic].owner = parent;
-		ResourceOwnerRememberLock(parent, locallock);
-	}
-	else
-	{
-		/* Merge child's count with parent's */
-		lockOwners[ip].nLocks += lockOwners[ic].nLocks;
-		/* compact out unused slot */
-		locallock->numLockOwners--;
-		if (ic < locallock->numLockOwners)
-			lockOwners[ic] = lockOwners[locallock->numLockOwners];
+		if (parentlocalowner->owner != parent)
+			continue;
+
+		parentlocalowner->nLocks += locallockowner->nLocks;
+
+		locallockowner->nLocks = 0;
+		dlist_delete(&locallockowner->locallock_node);
+		pfree(locallockowner);
+		return;
 	}
-	ResourceOwnerForgetLock(CurrentResourceOwner, locallock);
+
+	/* reassign locallockowner to parent resowner */
+	locallockowner->owner = parent;
+	ResourceOwnerRememberLock(parent, locallockowner);
 }
 
 /*
@@ -3414,10 +3365,9 @@ CheckForSessionAndXactLocks(void)
 
 	while ((locallock = (LOCALLOCK *) hash_seq_search(&status)) != NULL)
 	{
-		LOCALLOCKOWNER *lockOwners = locallock->lockOwners;
+		dlist_iter	iter;
 		PerLockTagEntry *hentry;
 		bool		found;
-		int			i;
 
 		/*
 		 * Ignore VXID locks.  We don't want those to be held by prepared
@@ -3438,9 +3388,11 @@ CheckForSessionAndXactLocks(void)
 			hentry->sessLock = hentry->xactLock = false;
 
 		/* Scan to see if we hold lock at session or xact level or both */
-		for (i = locallock->numLockOwners - 1; i >= 0; i--)
+		dlist_foreach(iter, &locallock->locallockowners)
 		{
-			if (lockOwners[i].owner == NULL)
+			LOCALLOCKOWNER *locallockowner = dlist_container(LOCALLOCKOWNER, locallock_node, iter.cur);
+
+			if (locallockowner->owner == NULL)
 				hentry->sessLock = true;
 			else
 				hentry->xactLock = true;
@@ -3487,10 +3439,9 @@ AtPrepare_Locks(void)
 	while ((locallock = (LOCALLOCK *) hash_seq_search(&status)) != NULL)
 	{
 		TwoPhaseLockRecord record;
-		LOCALLOCKOWNER *lockOwners = locallock->lockOwners;
+		dlist_iter	iter;
 		bool		haveSessionLock;
 		bool		haveXactLock;
-		int			i;
 
 		/*
 		 * Ignore VXID locks.  We don't want those to be held by prepared
@@ -3505,9 +3456,11 @@ AtPrepare_Locks(void)
 
 		/* Scan to see whether we hold it at session or transaction level */
 		haveSessionLock = haveXactLock = false;
-		for (i = locallock->numLockOwners - 1; i >= 0; i--)
+		dlist_foreach(iter, &locallock->locallockowners)
 		{
-			if (lockOwners[i].owner == NULL)
+			LOCALLOCKOWNER *locallockowner = dlist_container(LOCALLOCKOWNER, locallock_node, iter.cur);
+
+			if (locallockowner->owner == NULL)
 				haveSessionLock = true;
 			else
 				haveXactLock = true;
@@ -3599,10 +3552,9 @@ PostPrepare_Locks(FullTransactionId fxid)
 
 	while ((locallock = (LOCALLOCK *) hash_seq_search(&status)) != NULL)
 	{
-		LOCALLOCKOWNER *lockOwners = locallock->lockOwners;
+		dlist_iter	iter;
 		bool		haveSessionLock;
 		bool		haveXactLock;
-		int			i;
 
 		if (locallock->proclock == NULL || locallock->lock == NULL)
 		{
@@ -3621,9 +3573,11 @@ PostPrepare_Locks(FullTransactionId fxid)
 
 		/* Scan to see whether we hold it at session or transaction level */
 		haveSessionLock = haveXactLock = false;
-		for (i = locallock->numLockOwners - 1; i >= 0; i--)
+		dlist_foreach(iter, &locallock->locallockowners)
 		{
-			if (lockOwners[i].owner == NULL)
+			LOCALLOCKOWNER *locallockowner = dlist_container(LOCALLOCKOWNER, locallock_node, iter.cur);
+
+			if (locallockowner->owner == NULL)
 				haveSessionLock = true;
 			else
 				haveXactLock = true;
diff --git a/src/backend/utils/resowner/resowner.c b/src/backend/utils/resowner/resowner.c
index 06e1121c5ff..b327c9ee610 100644
--- a/src/backend/utils/resowner/resowner.c
+++ b/src/backend/utils/resowner/resowner.c
@@ -26,12 +26,11 @@
  * release-priority of each resource, and release them in that order.
  *
  * Local lock references are special, they are not stored in the array or
- * the hash table.  Instead, each resource owner has a separate small cache
+ * the hash table.  Instead, each resource owner has a separate linked list
  * of locks it owns.  The lock manager has the same information in its local
- * lock hash table, and we fall back on that if the cache overflows, but
- * traversing the hash table is slower when there are a lot of locks
- * belonging to other resource owners.  This is to speed up bulk releasing
- * or reassigning locks from a resource owner to its parent.
+ * lock hash table, but traversing the hash table is slower when there are a
+ * lot of locks belonging to other resource owners.  This is to speed up
+ * bulk releasing or reassigning locks from a resource owner to its parent.
  *
  *
  * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
@@ -94,18 +93,6 @@ typedef struct ResourceElem
 StaticAssertDecl(RESOWNER_HASH_MAX_ITEMS(RESOWNER_HASH_INIT_SIZE) >= RESOWNER_ARRAY_SIZE,
 				 "initial hash size too small compared to array size");
 
-/*
- * MAX_RESOWNER_LOCKS is the size of the per-resource owner locks cache. It's
- * chosen based on some testing with pg_dump with a large schema. When the
- * tests were done (on 9.2), resource owners in a pg_dump run contained up
- * to 9 locks, regardless of the schema size, except for the top resource
- * owner which contained much more (overflowing the cache). 15 seems like a
- * nice round number that's somewhat higher than what pg_dump needs. Note that
- * making this number larger is not free - the bigger the cache, the slower
- * it is to release locks (in retail), when a resource owner holds many locks.
- */
-#define MAX_RESOWNER_LOCKS 15
-
 /*
  * ResourceOwner objects look like this
  */
@@ -127,10 +114,9 @@ struct ResourceOwnerData
 	bool		sorted;			/* are 'hash' and 'arr' sorted by priority? */
 
 	/*
-	 * Number of items in the locks cache, array, and hash table respectively.
-	 * (These are packed together to avoid padding in the struct.)
+	 * Number of items in the array and hash table respectively. (These are
+	 * packed together to avoid padding in the struct.)
 	 */
-	uint8		nlocks;			/* number of owned locks */
 	uint8		narr;			/* how many items are stored in the array */
 	uint32		nhash;			/* how many items are stored in the hash */
 
@@ -155,8 +141,7 @@ struct ResourceOwnerData
 	uint32		capacity;		/* allocated length of hash[] */
 	uint32		grow_at;		/* grow hash when reach this */
 
-	/* The local locks cache. */
-	LOCALLOCK  *locks[MAX_RESOWNER_LOCKS];	/* list of owned locks */
+	dlist_head	locks;			/* dlist of owned locks */
 
 	/*
 	 * AIO handles need be registered in critical sections and therefore
@@ -422,6 +407,7 @@ ResourceOwnerCreate(ResourceOwner parent, const char *name)
 	owner = (ResourceOwner) MemoryContextAllocZero(TopMemoryContext,
 												   sizeof(struct ResourceOwnerData));
 	owner->name = name;
+	dlist_init(&owner->locks);
 
 	if (parent)
 	{
@@ -742,8 +728,19 @@ ResourceOwnerReleaseInternal(ResourceOwner owner,
 	}
 	else if (phase == RESOURCE_RELEASE_LOCKS)
 	{
+		dlist_mutable_iter iter;
+
 		if (isTopLevel)
 		{
+			dlist_foreach_modify(iter, &owner->locks)
+			{
+				LOCALLOCKOWNER *locallockowner = dlist_container(LOCALLOCKOWNER, resowner_node, iter.cur);
+
+				LockReleaseCurrentOwner(owner, locallockowner);
+			}
+
+			Assert(dlist_is_empty(&owner->locks));
+
 			/*
 			 * For a top-level xact we are going to release all locks (or at
 			 * least all non-session locks), so just do a single lmgr call at
@@ -762,30 +759,30 @@ ResourceOwnerReleaseInternal(ResourceOwner owner,
 			 * subtransaction, we do NOT release its locks yet, but transfer
 			 * them to the parent.
 			 */
-			LOCALLOCK **locks;
-			int			nlocks;
+			if (isCommit)
+			{
+				dlist_foreach_modify(iter, &owner->locks)
+				{
+					LOCALLOCKOWNER *locallockowner = dlist_container(LOCALLOCKOWNER,
+																	 resowner_node,
+																	 iter.cur);
 
-			Assert(owner->parent != NULL);
+					LockReassignCurrentOwner(locallockowner);
+				}
 
-			/*
-			 * Pass the list of locks owned by this resource owner to the lock
-			 * manager, unless it has overflowed.
-			 */
-			if (owner->nlocks > MAX_RESOWNER_LOCKS)
-			{
-				locks = NULL;
-				nlocks = 0;
+				Assert(dlist_is_empty(&owner->locks));
 			}
 			else
 			{
-				locks = owner->locks;
-				nlocks = owner->nlocks;
-			}
+				dlist_foreach_modify(iter, &owner->locks)
+				{
+					LOCALLOCKOWNER *locallockowner = dlist_container(LOCALLOCKOWNER, resowner_node, iter.cur);
 
-			if (isCommit)
-				LockReassignCurrentOwner(locks, nlocks);
-			else
-				LockReleaseCurrentOwner(locks, nlocks);
+					LockReleaseCurrentOwner(owner, locallockowner);
+				}
+
+				Assert(dlist_is_empty(&owner->locks));
+			}
 		}
 	}
 	else if (phase == RESOURCE_RELEASE_AFTER_LOCKS)
@@ -873,7 +870,7 @@ ResourceOwnerDelete(ResourceOwner owner)
 	/* And it better not own any resources, either */
 	Assert(owner->narr == 0);
 	Assert(owner->nhash == 0);
-	Assert(owner->nlocks == 0 || owner->nlocks == MAX_RESOWNER_LOCKS + 1);
+	Assert(dlist_is_empty(&owner->locks));
 
 	/*
 	 * Delete children.  The recursive call will delink the child from me, so
@@ -1047,54 +1044,59 @@ ReleaseAuxProcessResourcesCallback(int code, Datum arg)
 
 /*
  * Remember that a Local Lock is owned by a ResourceOwner
- *
- * This is different from the generic ResourceOwnerRemember in that the list of
- * locks is only a lossy cache.  It can hold up to MAX_RESOWNER_LOCKS entries,
- * and when it overflows, we stop tracking locks.  The point of only remembering
- * only up to MAX_RESOWNER_LOCKS entries is that if a lot of locks are held,
- * ResourceOwnerForgetLock doesn't need to scan through a large array to find
- * the entry.
  */
 void
-ResourceOwnerRememberLock(ResourceOwner owner, LOCALLOCK *locallock)
+ResourceOwnerRememberLock(ResourceOwner owner, LOCALLOCKOWNER *locallockowner)
 {
-	Assert(locallock != NULL);
-
-	if (owner->nlocks > MAX_RESOWNER_LOCKS)
-		return;					/* we have already overflowed */
+	Assert(owner != NULL);
+	Assert(locallockowner != NULL);
 
-	if (owner->nlocks < MAX_RESOWNER_LOCKS)
-		owner->locks[owner->nlocks] = locallock;
-	else
+#ifdef USE_ASSERT_CHECKING
 	{
-		/* overflowed */
+		dlist_iter	iter;
+
+		dlist_foreach(iter, &owner->locks)
+		{
+			LOCALLOCKOWNER *i = dlist_container(LOCALLOCKOWNER, resowner_node, iter.cur);
+
+			Assert(i->locallock != locallockowner->locallock);
+		}
 	}
-	owner->nlocks++;
+#endif
+
+	dlist_push_tail(&owner->locks, &locallockowner->resowner_node);
 }
 
 /*
- * Forget that a Local Lock is owned by a ResourceOwner
+ * Forget that a Local Lock is owned by the given LOCALLOCKOWNER.
  */
 void
-ResourceOwnerForgetLock(ResourceOwner owner, LOCALLOCK *locallock)
+ResourceOwnerForgetLock(LOCALLOCKOWNER *locallockowner)
 {
-	int			i;
+#ifdef USE_ASSERT_CHECKING
+	ResourceOwner owner;
 
-	if (owner->nlocks > MAX_RESOWNER_LOCKS)
-		return;					/* we have overflowed */
+	Assert(locallockowner != NULL);
+
+	owner = locallockowner->owner;
 
-	Assert(owner->nlocks > 0);
-	for (i = owner->nlocks - 1; i >= 0; i--)
 	{
-		if (locallock == owner->locks[i])
+		dlist_iter	iter;
+		bool		found = false;
+
+		dlist_foreach(iter, &owner->locks)
 		{
-			owner->locks[i] = owner->locks[owner->nlocks - 1];
-			owner->nlocks--;
-			return;
+			if (locallockowner == dlist_container(LOCALLOCKOWNER, resowner_node, iter.cur))
+			{
+				Assert(!found);
+				found = true;
+			}
 		}
+
+		Assert(found);
 	}
-	elog(ERROR, "lock reference %p is not owned by resource owner %s",
-		 locallock, owner->name);
+#endif
+	dlist_delete(&locallockowner->resowner_node);
 }
 
 void
diff --git a/src/include/storage/lock.h b/src/include/storage/lock.h
index 55ffaa5e4a5..7db33b9abbe 100644
--- a/src/include/storage/lock.h
+++ b/src/include/storage/lock.h
@@ -422,6 +422,13 @@ typedef struct LOCALLOCKOWNER
 	 * Must use a forward struct reference to avoid circularity.
 	 */
 	struct ResourceOwnerData *owner;
+
+	dlist_node	resowner_node;	/* dlist link for ResourceOwner.locks */
+
+	dlist_node	locallock_node; /* dlist link for LOCALLOCK.locallockowners */
+
+	struct LOCALLOCK *locallock;	/* pointer to the corresponding LOCALLOCK */
+
 	int64		nLocks;			/* # of times held by this owner */
 } LOCALLOCKOWNER;
 
@@ -435,9 +442,9 @@ typedef struct LOCALLOCK
 	LOCK	   *lock;			/* associated LOCK object, if any */
 	PROCLOCK   *proclock;		/* associated PROCLOCK object, if any */
 	int64		nLocks;			/* total number of times lock is held */
-	int			numLockOwners;	/* # of relevant ResourceOwners */
-	int			maxLockOwners;	/* allocated size of array */
-	LOCALLOCKOWNER *lockOwners; /* dynamically resizable array */
+
+	dlist_head	locallockowners;	/* dlist of LOCALLOCKOWNER */
+
 	bool		holdsStrongLockCount;	/* bumped FastPathStrongRelationLocks */
 	bool		lockCleared;	/* we read all sinval msgs for lock */
 } LOCALLOCK;
@@ -570,8 +577,8 @@ extern bool LockRelease(const LOCKTAG *locktag,
 						LOCKMODE lockmode, bool sessionLock);
 extern void LockReleaseAll(LOCKMETHODID lockmethodid, bool allLocks);
 extern void LockReleaseSession(LOCKMETHODID lockmethodid);
-extern void LockReleaseCurrentOwner(LOCALLOCK **locallocks, int nlocks);
-extern void LockReassignCurrentOwner(LOCALLOCK **locallocks, int nlocks);
+extern void LockReleaseCurrentOwner(struct ResourceOwnerData *owner, LOCALLOCKOWNER *locallockowner);
+extern void LockReassignCurrentOwner(LOCALLOCKOWNER *locallockowner);
 extern bool LockHeldByMe(const LOCKTAG *locktag,
 						 LOCKMODE lockmode, bool orstronger);
 #ifdef USE_ASSERT_CHECKING
diff --git a/src/include/utils/resowner.h b/src/include/utils/resowner.h
index eb6033b4fdb..05ef56edc85 100644
--- a/src/include/utils/resowner.h
+++ b/src/include/utils/resowner.h
@@ -160,9 +160,9 @@ extern void CreateAuxProcessResourceOwner(void);
 extern void ReleaseAuxProcessResources(bool isCommit);
 
 /* special support for local lock management */
-struct LOCALLOCK;
-extern void ResourceOwnerRememberLock(ResourceOwner owner, struct LOCALLOCK *locallock);
-extern void ResourceOwnerForgetLock(ResourceOwner owner, struct LOCALLOCK *locallock);
+struct LOCALLOCKOWNER;
+extern void ResourceOwnerRememberLock(ResourceOwner owner, struct LOCALLOCKOWNER *locallockowner);
+extern void ResourceOwnerForgetLock(struct LOCALLOCKOWNER *locallockowner);
 
 /* special support for AIO */
 struct dlist_node;
-- 
2.47.3