postgresql/src/backend/utils/resowner/resowner.c

/*-------------------------------------------------------------------------
 *
 * resowner.c
 *	  POSTGRES resource owner management code.
 *
 * Query-lifespan resources are tracked by associating them with
 * ResourceOwner objects.  This provides a simple mechanism for ensuring
 * that such resources are freed at the right time.
 * See utils/resowner/README for more info on how to use it.
 *
 * The implementation consists of a small fixed-size array and a hash table.
 * New entries are inserted to the fixed-size array, and when the array
 * fills up, all the entries are moved to the hash table.  This way, the
 * array always contains a few most recently remembered references.  To find
 * a particular reference, you need to search both the array and the hash
 * table.
 *
 * The most frequent usage is that a resource is remembered, and forgotten
 * shortly thereafter.  For example, pin a buffer, read one tuple from it,
 * release the pin.  Linearly scanning the small array handles that case
 * efficiently.  However, some resources are held for a longer time, and
 * sometimes a lot of resources need to be held simultaneously.  The hash
 * table handles those cases.
 *
 * When it's time to release the resources, we sort them according to the
 * 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
 * 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.
 *
 *
 * Portions Copyright (c) 1996-2024, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 *
 * IDENTIFICATION
 *	  src/backend/utils/resowner/resowner.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "common/hashfn.h"
#include "common/int.h"
#include "storage/ipc.h"
#include "storage/predicate.h"
#include "storage/proc.h"
#include "utils/memutils.h"
#include "utils/resowner.h"

/*
 * ResourceElem represents a reference associated with a resource owner.
 *
 * All objects managed by this code are required to fit into a Datum,
 * which is fine since they are generally pointers or integers.
 */
typedef struct ResourceElem
{
	Datum		item;
	const ResourceOwnerDesc *kind;	/* NULL indicates a free hash table slot */
} ResourceElem;

/*
 * Size of the fixed-size array to hold most-recently remembered resources.
 */
#define RESOWNER_ARRAY_SIZE 32

/*
 * Initially allocated size of a ResourceOwner's hash table.  Must be power of
 * two because we use (capacity - 1) as mask for hashing.
 */
#define RESOWNER_HASH_INIT_SIZE 64

/*
 * How many items may be stored in a hash table of given capacity.  When this
 * number is reached, we must resize.
 *
 * The hash table must always have enough free space that we can copy the
 * entries from the array to it, in ResourceOwnerSort.  We also insist that
 * the initial size is large enough that we don't hit the max size immediately
 * when it's created.  Aside from those limitations, 0.75 is a reasonable fill
 * factor.
 */
#define RESOWNER_HASH_MAX_ITEMS(capacity) \
	Min(capacity - RESOWNER_ARRAY_SIZE, (capacity)/4 * 3)

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
 */
typedef struct ResourceOwnerData
{
	ResourceOwner parent;		/* NULL if no parent (toplevel owner) */
	ResourceOwner firstchild;	/* head of linked list of children */
	ResourceOwner nextchild;	/* next child of same parent */
	const char *name;			/* name (just for debugging) */

	/*
	 * When ResourceOwnerRelease is called, we sort the 'hash' and 'arr' by
	 * the release priority.  After that, no new resources can be remembered
	 * or forgotten in retail.  We have separate flags because
	 * ResourceOwnerReleaseAllOfKind() temporarily sets 'releasing' without
	 * sorting the arrays.
	 */
	bool		releasing;
	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.)
	 */
	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 */

	/*
	 * The fixed-size array for recent resources.
	 *
	 * If 'sorted' is set, the contents are sorted by release priority.
	 */
	ResourceElem arr[RESOWNER_ARRAY_SIZE];

	/*
	 * The hash table.  Uses open-addressing.  'nhash' is the number of items
	 * present; if it would exceed 'grow_at', we enlarge it and re-hash.
	 * 'grow_at' should be rather less than 'capacity' so that we don't waste
	 * too much time searching for empty slots.
	 *
	 * If 'sorted' is set, the contents are no longer hashed, but sorted by
	 * release priority.  The first 'nhash' elements are occupied, the rest
	 * are empty.
	 */
	ResourceElem *hash;
	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 */
} ResourceOwnerData;


/*****************************************************************************
 *	  GLOBAL MEMORY															 *
 *****************************************************************************/

ResourceOwner CurrentResourceOwner = NULL;
ResourceOwner CurTransactionResourceOwner = NULL;
ResourceOwner TopTransactionResourceOwner = NULL;
ResourceOwner AuxProcessResourceOwner = NULL;

/* #define RESOWNER_STATS */

#ifdef RESOWNER_STATS
static int	narray_lookups = 0;
static int	nhash_lookups = 0;
#endif

/*
 * List of add-on callbacks for resource releasing
 */
typedef struct ResourceReleaseCallbackItem
{
	struct ResourceReleaseCallbackItem *next;
	ResourceReleaseCallback callback;
	void	   *arg;
} ResourceReleaseCallbackItem;

static ResourceReleaseCallbackItem *ResourceRelease_callbacks = NULL;


/* Internal routines */
static inline uint32 hash_resource_elem(Datum value, const ResourceOwnerDesc *kind);
static void ResourceOwnerAddToHash(ResourceOwner owner, Datum value,
								   const ResourceOwnerDesc *kind);
static int	resource_priority_cmp(const void *a, const void *b);
static void ResourceOwnerSort(ResourceOwner owner);
static void ResourceOwnerReleaseAll(ResourceOwner owner,
									ResourceReleasePhase phase,
									bool printLeakWarnings);
static void ResourceOwnerReleaseInternal(ResourceOwner owner,
										 ResourceReleasePhase phase,
										 bool isCommit,
										 bool isTopLevel);
static void ReleaseAuxProcessResourcesCallback(int code, Datum arg);


/*****************************************************************************
 *	  INTERNAL ROUTINES														 *
 *****************************************************************************/

/*
 * Hash function for value+kind combination.
 */
static inline uint32
hash_resource_elem(Datum value, const ResourceOwnerDesc *kind)
{
	/*
	 * Most resource kinds store a pointer in 'value', and pointers are unique
	 * all on their own.  But some resources store plain integers (Files and
	 * Buffers as of this writing), so we want to incorporate the 'kind' in
	 * the hash too, otherwise those resources will collide a lot.  But
	 * because there are only a few resource kinds like that - and only a few
	 * resource kinds to begin with - we don't need to work too hard to mix
	 * 'kind' into the hash.  Just add it with hash_combine(), it perturbs the
	 * result enough for our purposes.
	 */
#if SIZEOF_DATUM == 8
	return hash_combine64(murmurhash64((uint64) value), (uint64) kind);
#else
	return hash_combine(murmurhash32((uint32) value), (uint32) kind);
#endif
}

/*
 * Adds 'value' of given 'kind' to the ResourceOwner's hash table
 */
static void
ResourceOwnerAddToHash(ResourceOwner owner, Datum value, const ResourceOwnerDesc *kind)
{
	uint32		mask = owner->capacity - 1;
	uint32		idx;

	Assert(kind != NULL);

	/* Insert into first free slot at or after hash location. */
	idx = hash_resource_elem(value, kind) & mask;
	for (;;)
	{
		if (owner->hash[idx].kind == NULL)
			break;				/* found a free slot */
		idx = (idx + 1) & mask;
	}
	owner->hash[idx].item = value;
	owner->hash[idx].kind = kind;
	owner->nhash++;
}

/*
 * Comparison function to sort by release phase and priority
 */
static int
resource_priority_cmp(const void *a, const void *b)
{
	const ResourceElem *ra = (const ResourceElem *) a;
	const ResourceElem *rb = (const ResourceElem *) b;

	/* Note: reverse order */
	if (ra->kind->release_phase == rb->kind->release_phase)
		return pg_cmp_u32(rb->kind->release_priority, ra->kind->release_priority);
	else if (ra->kind->release_phase > rb->kind->release_phase)
		return -1;
	else
		return 1;
}

/*
 * Sort resources in reverse release priority.
 *
 * If the hash table is in use, all the elements from the fixed-size array are
 * moved to the hash table, and then the hash table is sorted.  If there is no
 * hash table, then the fixed-size array is sorted directly.  In either case,
 * the result is one sorted array that contains all the resources.
 */
static void
ResourceOwnerSort(ResourceOwner owner)
{
	ResourceElem *items;
	uint32		nitems;

	if (owner->nhash == 0)
	{
		items = owner->arr;
		nitems = owner->narr;
	}
	else
	{
		/*
		 * Compact the hash table, so that all the elements are in the
		 * beginning of the 'hash' array, with no empty elements.
		 */
		uint32		dst = 0;

		for (int idx = 0; idx < owner->capacity; idx++)
		{
			if (owner->hash[idx].kind != NULL)
			{
				if (dst != idx)
					owner->hash[dst] = owner->hash[idx];
				dst++;
			}
		}

		/*
		 * Move all entries from the fixed-size array to 'hash'.
		 *
		 * RESOWNER_HASH_MAX_ITEMS is defined so that there is always enough
		 * free space to move all the elements from the fixed-size array to
		 * the hash.
		 */
		Assert(dst + owner->narr <= owner->capacity);
		for (int idx = 0; idx < owner->narr; idx++)
		{
			owner->hash[dst] = owner->arr[idx];
			dst++;
		}
		Assert(dst == owner->nhash + owner->narr);
		owner->narr = 0;
		owner->nhash = dst;

		items = owner->hash;
		nitems = owner->nhash;
	}

	qsort(items, nitems, sizeof(ResourceElem), resource_priority_cmp);
}

/*
 * Call the ReleaseResource callback on entries with given 'phase'.
 */
static void
ResourceOwnerReleaseAll(ResourceOwner owner, ResourceReleasePhase phase,
						bool printLeakWarnings)
{
	ResourceElem *items;
	uint32		nitems;

	/*
	 * ResourceOwnerSort must've been called already.  All the resources are
	 * either in the array or the hash.
	 */
	Assert(owner->releasing);
	Assert(owner->sorted);
	if (owner->nhash == 0)
	{
		items = owner->arr;
		nitems = owner->narr;
	}
	else
	{
		Assert(owner->narr == 0);
		items = owner->hash;
		nitems = owner->nhash;
	}

	/*
	 * The resources are sorted in reverse priority order.  Release them
	 * starting from the end, until we hit the end of the phase that we are
	 * releasing now.  We will continue from there when called again for the
	 * next phase.
	 */
	while (nitems > 0)
	{
		uint32		idx = nitems - 1;
		Datum		value = items[idx].item;
		const ResourceOwnerDesc *kind = items[idx].kind;

		if (kind->release_phase > phase)
			break;
		Assert(kind->release_phase == phase);

		if (printLeakWarnings)
		{
			char	   *res_str;

			res_str = kind->DebugPrint ?
				kind->DebugPrint(value)
				: psprintf("%s %p", kind->name, DatumGetPointer(value));
			elog(WARNING, "resource was not closed: %s", res_str);
			pfree(res_str);
		}
		kind->ReleaseResource(value);
		nitems--;
	}
	if (owner->nhash == 0)
		owner->narr = nitems;
	else
		owner->nhash = nitems;
}


/*****************************************************************************
 *	  EXPORTED ROUTINES														 *
 *****************************************************************************/


/*
 * ResourceOwnerCreate
 *		Create an empty ResourceOwner.
 *
 * All ResourceOwner objects are kept in TopMemoryContext, since they should
 * only be freed explicitly.
 */
ResourceOwner
ResourceOwnerCreate(ResourceOwner parent, const char *name)
{
	ResourceOwner owner;

	owner = (ResourceOwner) MemoryContextAllocZero(TopMemoryContext,
												   sizeof(ResourceOwnerData));
	owner->name = name;

	if (parent)
	{
		owner->parent = parent;
		owner->nextchild = parent->firstchild;
		parent->firstchild = owner;
	}

	return owner;
}

/*
 * Make sure there is room for at least one more resource in an array.
 *
 * This is separate from actually inserting a resource because if we run out
 * of memory, it's critical to do so *before* acquiring the resource.
 *
 * NB: Make sure there are no unrelated ResourceOwnerRemember() calls between
 * your ResourceOwnerEnlarge() call and the ResourceOwnerRemember() call that
 * you reserved the space for!
 */
void
ResourceOwnerEnlarge(ResourceOwner owner)
{
	/*
	 * Mustn't try to remember more resources after we have already started
	 * releasing
	 */
	if (owner->releasing)
		elog(ERROR, "ResourceOwnerEnlarge called after release started");

	if (owner->narr < RESOWNER_ARRAY_SIZE)
		return;					/* no work needed */

	/*
	 * Is there space in the hash? If not, enlarge it.
	 */
	if (owner->narr + owner->nhash >= owner->grow_at)
	{
		uint32		i,
					oldcap,
					newcap;
		ResourceElem *oldhash;
		ResourceElem *newhash;

		oldhash = owner->hash;
		oldcap = owner->capacity;

		/* Double the capacity (it must stay a power of 2!) */
		newcap = (oldcap > 0) ? oldcap * 2 : RESOWNER_HASH_INIT_SIZE;
		newhash = (ResourceElem *) MemoryContextAllocZero(TopMemoryContext,
														  newcap * sizeof(ResourceElem));

		/*
		 * We assume we can't fail below this point, so OK to scribble on the
		 * owner
		 */
		owner->hash = newhash;
		owner->capacity = newcap;
		owner->grow_at = RESOWNER_HASH_MAX_ITEMS(newcap);
		owner->nhash = 0;

		if (oldhash != NULL)
		{
			/*
			 * Transfer any pre-existing entries into the new hash table; they
			 * don't necessarily go where they were before, so this simple
			 * logic is the best way.
			 */
			for (i = 0; i < oldcap; i++)
			{
				if (oldhash[i].kind != NULL)
					ResourceOwnerAddToHash(owner, oldhash[i].item, oldhash[i].kind);
			}

			/* And release old hash table. */
			pfree(oldhash);
		}
	}

	/* Move items from the array to the hash */
	for (int i = 0; i < owner->narr; i++)
		ResourceOwnerAddToHash(owner, owner->arr[i].item, owner->arr[i].kind);
	owner->narr = 0;

	Assert(owner->nhash <= owner->grow_at);
}

/*
 * Remember that an object is owned by a ResourceOwner
 *
 * Caller must have previously done ResourceOwnerEnlarge()
 */
void
ResourceOwnerRemember(ResourceOwner owner, Datum value, const ResourceOwnerDesc *kind)
{
	uint32		idx;

	/* sanity check the ResourceOwnerDesc */
	Assert(kind->release_phase != 0);
	Assert(kind->release_priority != 0);

	/*
	 * Mustn't try to remember more resources after we have already started
	 * releasing.  We already checked this in ResourceOwnerEnlarge.
	 */
	Assert(!owner->releasing);
	Assert(!owner->sorted);

	if (owner->narr >= RESOWNER_ARRAY_SIZE)
	{
		/* forgot to call ResourceOwnerEnlarge? */
		elog(ERROR, "ResourceOwnerRemember called but array was full");
	}

	/* Append to the array. */
	idx = owner->narr;
	owner->arr[idx].item = value;
	owner->arr[idx].kind = kind;
	owner->narr++;
}

/*
 * Forget that an object is owned by a ResourceOwner
 *
 * Note: If same resource ID is associated with the ResourceOwner more than
 * once, one instance is removed.
 *
 * Note: Forgetting a resource does not guarantee that there is room to
 * remember a new resource.  One exception is when you forget the most
 * recently remembered resource; that does make room for a new remember call.
 * Some code callers rely on that exception.
 */
void
ResourceOwnerForget(ResourceOwner owner, Datum value, const ResourceOwnerDesc *kind)
{
	/*
	 * Mustn't call this after we have already started releasing resources.
	 * (Release callback functions are not allowed to release additional
	 * resources.)
	 */
	if (owner->releasing)
		elog(ERROR, "ResourceOwnerForget called for %s after release started", kind->name);
	Assert(!owner->sorted);

	/* Search through all items in the array first. */
	for (int i = owner->narr - 1; i >= 0; i--)
	{
		if (owner->arr[i].item == value &&
			owner->arr[i].kind == kind)
		{
			owner->arr[i] = owner->arr[owner->narr - 1];
			owner->narr--;

#ifdef RESOWNER_STATS
			narray_lookups++;
#endif
			return;
		}
	}

	/* Search hash */
	if (owner->nhash > 0)
	{
		uint32		mask = owner->capacity - 1;
		uint32		idx;

		idx = hash_resource_elem(value, kind) & mask;
		for (uint32 i = 0; i < owner->capacity; i++)
		{
			if (owner->hash[idx].item == value &&
				owner->hash[idx].kind == kind)
			{
				owner->hash[idx].item = (Datum) 0;
				owner->hash[idx].kind = NULL;
				owner->nhash--;

#ifdef RESOWNER_STATS
				nhash_lookups++;
#endif
				return;
			}
			idx = (idx + 1) & mask;
		}
	}

	/*
	 * Use %p to print the reference, since most objects tracked by a resource
	 * owner are pointers.  It's a bit misleading if it's not a pointer, but
	 * this is a programmer error, anyway.
	 */
	elog(ERROR, "%s %p is not owned by resource owner %s",
		 kind->name, DatumGetPointer(value), owner->name);
}

/*
 * ResourceOwnerRelease
 *		Release all resources owned by a ResourceOwner and its descendants,
 *		but don't delete the owner objects themselves.
 *
 * Note that this executes just one phase of release, and so typically
 * must be called three times.  We do it this way because (a) we want to
 * do all the recursion separately for each phase, thereby preserving
 * the needed order of operations; and (b) xact.c may have other operations
 * to do between the phases.
 *
 * phase: release phase to execute
 * isCommit: true for successful completion of a query or transaction,
 *			false for unsuccessful
 * isTopLevel: true if completing a main transaction, else false
 *
 * isCommit is passed because some modules may expect that their resources
 * were all released already if the transaction or portal finished normally.
 * If so it is reasonable to give a warning (NOT an error) should any
 * unreleased resources be present.  When isCommit is false, such warnings
 * are generally inappropriate.
 *
 * isTopLevel is passed when we are releasing TopTransactionResourceOwner
 * at completion of a main transaction.  This generally means that *all*
 * resources will be released, and so we can optimize things a bit.
 *
 * NOTE: After starting the release process, by calling this function, no new
 * resources can be remembered in the resource owner.  You also cannot call
 * ResourceOwnerForget on any previously remembered resources to release
 * resources "in retail" after that, you must let the bulk release take care
 * of them.
 */
void
ResourceOwnerRelease(ResourceOwner owner,
					 ResourceReleasePhase phase,
					 bool isCommit,
					 bool isTopLevel)
{
	/* There's not currently any setup needed before recursing */
	ResourceOwnerReleaseInternal(owner, phase, isCommit, isTopLevel);

#ifdef RESOWNER_STATS
	if (isTopLevel)
	{
		elog(LOG, "RESOWNER STATS: lookups: array %d, hash %d",
			 narray_lookups, nhash_lookups);
		narray_lookups = 0;
		nhash_lookups = 0;
	}
#endif
}

static void
ResourceOwnerReleaseInternal(ResourceOwner owner,
							 ResourceReleasePhase phase,
							 bool isCommit,
							 bool isTopLevel)
{
	ResourceOwner child;
	ResourceOwner save;
	ResourceReleaseCallbackItem *item;
	ResourceReleaseCallbackItem *next;

	/* Recurse to handle descendants */
	for (child = owner->firstchild; child != NULL; child = child->nextchild)
		ResourceOwnerReleaseInternal(child, phase, isCommit, isTopLevel);

	/*
	 * To release the resources in the right order, sort them by phase and
	 * priority.
	 *
	 * The ReleaseResource callback functions are not allowed to remember or
	 * forget any other resources after this. Otherwise we lose track of where
	 * we are in processing the hash/array.
	 */
	if (!owner->releasing)
	{
		Assert(phase == RESOURCE_RELEASE_BEFORE_LOCKS);
		Assert(!owner->sorted);
		owner->releasing = true;
	}
	else
	{
		/*
		 * Phase is normally > RESOURCE_RELEASE_BEFORE_LOCKS, if this is not
		 * the first call to ResourceOwnerRelease. But if an error happens
		 * between the release phases, we might get called again for the same
		 * ResourceOwner from AbortTransaction.
		 */
	}
	if (!owner->sorted)
	{
		ResourceOwnerSort(owner);
		owner->sorted = true;
	}

	/*
	 * Make CurrentResourceOwner point to me, so that the release callback
	 * functions know which resource owner is been released.
	 */
	save = CurrentResourceOwner;
	CurrentResourceOwner = owner;

	if (phase == RESOURCE_RELEASE_BEFORE_LOCKS)
	{
		/*
		 * Release all resources that need to be released before the locks.
		 *
		 * During a commit, there shouldn't be any remaining resources ---
		 * that would indicate failure to clean up the executor correctly ---
		 * so issue warnings.  In the abort case, just clean up quietly.
		 */
		ResourceOwnerReleaseAll(owner, phase, isCommit);
	}
	else if (phase == RESOURCE_RELEASE_LOCKS)
	{
		if (isTopLevel)
		{
			/*
			 * 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
			 * the top of the recursion.
			 */
			if (owner == TopTransactionResourceOwner)
			{
				ProcReleaseLocks(isCommit);
				ReleasePredicateLocks(isCommit, false);
			}
		}
		else
		{
			/*
			 * Release locks retail.  Note that if we are committing a
			 * subtransaction, we do NOT release its locks yet, but transfer
			 * them to the parent.
			 */
			LOCALLOCK **locks;
			int			nlocks;

			Assert(owner->parent != NULL);

			/*
			 * 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;
			}
			else
			{
				locks = owner->locks;
				nlocks = owner->nlocks;
			}

			if (isCommit)
				LockReassignCurrentOwner(locks, nlocks);
			else
				LockReleaseCurrentOwner(locks, nlocks);
		}
	}
	else if (phase == RESOURCE_RELEASE_AFTER_LOCKS)
	{
		/*
		 * Release all resources that need to be released after the locks.
		 */
		ResourceOwnerReleaseAll(owner, phase, isCommit);
	}

	/* Let add-on modules get a chance too */
	for (item = ResourceRelease_callbacks; item; item = next)
	{
		/* allow callbacks to unregister themselves when called */
		next = item->next;
		item->callback(phase, isCommit, isTopLevel, item->arg);
	}

	CurrentResourceOwner = save;
}

/*
 * ResourceOwnerReleaseAllOfKind
 *		Release all resources of a certain type held by this owner.
 */
void
ResourceOwnerReleaseAllOfKind(ResourceOwner owner, const ResourceOwnerDesc *kind)
{
	/* Mustn't call this after we have already started releasing resources. */
	if (owner->releasing)
		elog(ERROR, "ResourceOwnerForget called for %s after release started", kind->name);
	Assert(!owner->sorted);

	/*
	 * Temporarily set 'releasing', to prevent calls to ResourceOwnerRemember
	 * while we're scanning the owner.  Enlarging the hash would cause us to
	 * lose track of the point we're scanning.
	 */
	owner->releasing = true;

	/* Array first */
	for (int i = 0; i < owner->narr; i++)
	{
		if (owner->arr[i].kind == kind)
		{
			Datum		value = owner->arr[i].item;

			owner->arr[i] = owner->arr[owner->narr - 1];
			owner->narr--;
			i--;

			kind->ReleaseResource(value);
		}
	}

	/* Then hash */
	for (int i = 0; i < owner->capacity; i++)
	{
		if (owner->hash[i].kind == kind)
		{
			Datum		value = owner->hash[i].item;

			owner->hash[i].item = (Datum) 0;
			owner->hash[i].kind = NULL;
			owner->nhash--;

			kind->ReleaseResource(value);
		}
	}
	owner->releasing = false;
}

/*
 * ResourceOwnerDelete
 *		Delete an owner object and its descendants.
 *
 * The caller must have already released all resources in the object tree.
 */
void
ResourceOwnerDelete(ResourceOwner owner)
{
	/* We had better not be deleting CurrentResourceOwner ... */
	Assert(owner != CurrentResourceOwner);

	/* 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);

	/*
	 * Delete children.  The recursive call will delink the child from me, so
	 * just iterate as long as there is a child.
	 */
	while (owner->firstchild != NULL)
		ResourceOwnerDelete(owner->firstchild);

	/*
	 * We delink the owner from its parent before deleting it, so that if
	 * there's an error we won't have deleted/busted owners still attached to
	 * the owner tree.  Better a leak than a crash.
	 */
	ResourceOwnerNewParent(owner, NULL);

	/* And free the object. */
	if (owner->hash)
		pfree(owner->hash);
	pfree(owner);
}

/*
 * Fetch parent of a ResourceOwner (returns NULL if top-level owner)
 */
ResourceOwner
ResourceOwnerGetParent(ResourceOwner owner)
{
	return owner->parent;
}

/*
 * Reassign a ResourceOwner to have a new parent
 */
void
ResourceOwnerNewParent(ResourceOwner owner,
					   ResourceOwner newparent)
{
	ResourceOwner oldparent = owner->parent;

	if (oldparent)
	{
		if (owner == oldparent->firstchild)
			oldparent->firstchild = owner->nextchild;
		else
		{
			ResourceOwner child;

			for (child = oldparent->firstchild; child; child = child->nextchild)
			{
				if (owner == child->nextchild)
				{
					child->nextchild = owner->nextchild;
					break;
				}
			}
		}
	}

	if (newparent)
	{
		Assert(owner != newparent);
		owner->parent = newparent;
		owner->nextchild = newparent->firstchild;
		newparent->firstchild = owner;
	}
	else
	{
		owner->parent = NULL;
		owner->nextchild = NULL;
	}
}

/*
 * Register or deregister callback functions for resource cleanup
 *
 * These functions can be used by dynamically loaded modules.  These used
 * to be the only way for an extension to register custom resource types
 * with a resource owner, but nowadays it is easier to define a new
 * ResourceOwnerDesc with custom callbacks.
 */
void
RegisterResourceReleaseCallback(ResourceReleaseCallback callback, void *arg)
{
	ResourceReleaseCallbackItem *item;

	item = (ResourceReleaseCallbackItem *)
		MemoryContextAlloc(TopMemoryContext,
						   sizeof(ResourceReleaseCallbackItem));
	item->callback = callback;
	item->arg = arg;
	item->next = ResourceRelease_callbacks;
	ResourceRelease_callbacks = item;
}

void
UnregisterResourceReleaseCallback(ResourceReleaseCallback callback, void *arg)
{
	ResourceReleaseCallbackItem *item;
	ResourceReleaseCallbackItem *prev;

	prev = NULL;
	for (item = ResourceRelease_callbacks; item; prev = item, item = item->next)
	{
		if (item->callback == callback && item->arg == arg)
		{
			if (prev)
				prev->next = item->next;
			else
				ResourceRelease_callbacks = item->next;
			pfree(item);
			break;
		}
	}
}

/*
 * Establish an AuxProcessResourceOwner for the current process.
 */
void
CreateAuxProcessResourceOwner(void)
{
	Assert(AuxProcessResourceOwner == NULL);
	Assert(CurrentResourceOwner == NULL);
	AuxProcessResourceOwner = ResourceOwnerCreate(NULL, "AuxiliaryProcess");
	CurrentResourceOwner = AuxProcessResourceOwner;

	/*
	 * Register a shmem-exit callback for cleanup of aux-process resource
	 * owner.  (This needs to run after, e.g., ShutdownXLOG.)
	 */
	on_shmem_exit(ReleaseAuxProcessResourcesCallback, 0);
}

/*
 * Convenience routine to release all resources tracked in
 * AuxProcessResourceOwner (but that resowner is not destroyed here).
 * Warn about leaked resources if isCommit is true.
 */
void
ReleaseAuxProcessResources(bool isCommit)
{
	/*
	 * At this writing, the only thing that could actually get released is
	 * buffer pins; but we may as well do the full release protocol.
	 */
	ResourceOwnerRelease(AuxProcessResourceOwner,
						 RESOURCE_RELEASE_BEFORE_LOCKS,
						 isCommit, true);
	ResourceOwnerRelease(AuxProcessResourceOwner,
						 RESOURCE_RELEASE_LOCKS,
						 isCommit, true);
	ResourceOwnerRelease(AuxProcessResourceOwner,
						 RESOURCE_RELEASE_AFTER_LOCKS,
						 isCommit, true);
	/* allow it to be reused */
	AuxProcessResourceOwner->releasing = false;
	AuxProcessResourceOwner->sorted = false;
}

/*
 * Shmem-exit callback for the same.
 * Warn about leaked resources if process exit code is zero (ie normal).
 */
static void
ReleaseAuxProcessResourcesCallback(int code, Datum arg)
{
	bool		isCommit = (code == 0);

	ReleaseAuxProcessResources(isCommit);
}

/*
 * 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)
{
	Assert(locallock != NULL);

	if (owner->nlocks > MAX_RESOWNER_LOCKS)
		return;					/* we have already overflowed */

	if (owner->nlocks < MAX_RESOWNER_LOCKS)
		owner->locks[owner->nlocks] = locallock;
	else
	{
		/* overflowed */
	}
	owner->nlocks++;
}

/*
 * Forget that a Local Lock is owned by a ResourceOwner
 */
void
ResourceOwnerForgetLock(ResourceOwner owner, LOCALLOCK *locallock)
{
	int			i;

	if (owner->nlocks > MAX_RESOWNER_LOCKS)
		return;					/* we have overflowed */

	Assert(owner->nlocks > 0);
	for (i = owner->nlocks - 1; i >= 0; i--)
	{
		if (locallock == owner->locks[i])
		{
			owner->locks[i] = owner->locks[owner->nlocks - 1];
			owner->nlocks--;
			return;
		}
	}
	elog(ERROR, "lock reference %p is not owned by resource owner %s",
		 locallock, owner->name);
}