Adding doubly linked list type which stores the number of items in the list

diff --git a/src/backend/access/heap/rewriteheap.c b/src/backend/access/heap/rewriteheap.c
index b01b39b008..a34e9b352d 100644
--- a/src/backend/access/heap/rewriteheap.c
+++ b/src/backend/access/heap/rewriteheap.c
@@ -196,8 +196,7 @@ typedef struct RewriteMappingFile
 	TransactionId xid;			/* xid that might need to see the row */
 	int			vfd;			/* fd of mappings file */
 	off_t		off;			/* how far have we written yet */
-	uint32		num_mappings;	/* number of in-memory mappings */
-	dlist_head	mappings;		/* list of in-memory mappings */
+	dclist_head mappings;		/* list of in-memory mappings */
 	char		path[MAXPGPATH];	/* path, for error messages */
 } RewriteMappingFile;
 
@@ -864,9 +863,10 @@ logical_heap_rewrite_flush_mappings(RewriteState state)
 		Oid			dboid;
 		uint32		len;
 		int			written;
+		uint32		num_mappings = dclist_count(&src->mappings);
 
 		/* this file hasn't got any new mappings */
-		if (src->num_mappings == 0)
+		if (num_mappings == 0)
 			continue;
 
 		if (state->rs_old_rel->rd_rel->relisshared)
@@ -874,7 +874,7 @@ logical_heap_rewrite_flush_mappings(RewriteState state)
 		else
 			dboid = MyDatabaseId;
 
-		xlrec.num_mappings = src->num_mappings;
+		xlrec.num_mappings = num_mappings;
 		xlrec.mapped_rel = RelationGetRelid(state->rs_old_rel);
 		xlrec.mapped_xid = src->xid;
 		xlrec.mapped_db = dboid;
@@ -882,31 +882,30 @@ logical_heap_rewrite_flush_mappings(RewriteState state)
 		xlrec.start_lsn = state->rs_begin_lsn;
 
 		/* write all mappings consecutively */
-		len = src->num_mappings * sizeof(LogicalRewriteMappingData);
+		len = num_mappings * sizeof(LogicalRewriteMappingData);
 		waldata_start = waldata = palloc(len);
 
 		/*
 		 * collect data we need to write out, but don't modify ondisk data yet
 		 */
-		dlist_foreach_modify(iter, &src->mappings)
+		dclist_foreach_modify(iter, &src->mappings)
 		{
 			RewriteMappingDataEntry *pmap;
 
-			pmap = dlist_container(RewriteMappingDataEntry, node, iter.cur);
+			pmap = dclist_container(RewriteMappingDataEntry, node, iter.cur);
 
 			memcpy(waldata, &pmap->map, sizeof(pmap->map));
 			waldata += sizeof(pmap->map);
 
 			/* remove from the list and free */
-			dlist_delete(&pmap->node);
+			dclist_delete_from(&src->mappings, &pmap->node);
 			pfree(pmap);
 
 			/* update bookkeeping */
 			state->rs_num_rewrite_mappings--;
-			src->num_mappings--;
 		}
 
-		Assert(src->num_mappings == 0);
+		Assert(dclist_count(&src->mappings) == 0);
 		Assert(waldata == waldata_start + len);
 
 		/*
@@ -1002,8 +1001,7 @@ logical_rewrite_log_mapping(RewriteState state, TransactionId xid,
 				 LSN_FORMAT_ARGS(state->rs_begin_lsn),
 				 xid, GetCurrentTransactionId());
 
-		dlist_init(&src->mappings);
-		src->num_mappings = 0;
+		dclist_init(&src->mappings);
 		src->off = 0;
 		memcpy(src->path, path, sizeof(path));
 		src->vfd = PathNameOpenFile(path,
@@ -1017,8 +1015,7 @@ logical_rewrite_log_mapping(RewriteState state, TransactionId xid,
 	pmap = MemoryContextAlloc(state->rs_cxt,
 							  sizeof(RewriteMappingDataEntry));
 	memcpy(&pmap->map, map, sizeof(LogicalRewriteMappingData));
-	dlist_push_tail(&src->mappings, &pmap->node);
-	src->num_mappings++;
+	dclist_push_tail(&src->mappings, &pmap->node);
 	state->rs_num_rewrite_mappings++;
 
 	/*
diff --git a/src/backend/access/transam/multixact.c b/src/backend/access/transam/multixact.c
index a7383f553b..a03749061b 100644
--- a/src/backend/access/transam/multixact.c
+++ b/src/backend/access/transam/multixact.c
@@ -319,8 +319,7 @@ typedef struct mXactCacheEnt
 } mXactCacheEnt;
 
 #define MAX_CACHE_ENTRIES	256
-static dlist_head MXactCache = DLIST_STATIC_INIT(MXactCache);
-static int	MXactCacheMembers = 0;
+static dclist_head MXactCache = DCLIST_STATIC_INIT(MXactCache);
 static MemoryContext MXactContext = NULL;
 
 #ifdef MULTIXACT_DEBUG
@@ -1504,9 +1503,9 @@ mXactCacheGetBySet(int nmembers, MultiXactMember *members)
 	/* sort the array so comparison is easy */
 	qsort(members, nmembers, sizeof(MultiXactMember), mxactMemberComparator);
 
-	dlist_foreach(iter, &MXactCache)
+	dclist_foreach(iter, &MXactCache)
 	{
-		mXactCacheEnt *entry = dlist_container(mXactCacheEnt, node, iter.cur);
+		mXactCacheEnt *entry = dclist_container(mXactCacheEnt, node, iter.cur);
 
 		if (entry->nmembers != nmembers)
 			continue;
@@ -1518,7 +1517,7 @@ mXactCacheGetBySet(int nmembers, MultiXactMember *members)
 		if (memcmp(members, entry->members, nmembers * sizeof(MultiXactMember)) == 0)
 		{
 			debug_elog3(DEBUG2, "CacheGet: found %u", entry->multi);
-			dlist_move_head(&MXactCache, iter.cur);
+			dclist_move_head(&MXactCache, iter.cur);
 			return entry->multi;
 		}
 	}
@@ -1542,9 +1541,9 @@ mXactCacheGetById(MultiXactId multi, MultiXactMember **members)
 
 	debug_elog3(DEBUG2, "CacheGet: looking for %u", multi);
 
-	dlist_foreach(iter, &MXactCache)
+	dclist_foreach(iter, &MXactCache)
 	{
-		mXactCacheEnt *entry = dlist_container(mXactCacheEnt, node, iter.cur);
+		mXactCacheEnt *entry = dclist_container(mXactCacheEnt, node, iter.cur);
 
 		if (entry->multi == multi)
 		{
@@ -1566,7 +1565,7 @@ mXactCacheGetById(MultiXactId multi, MultiXactMember **members)
 			 * This is acceptable only because we exit the iteration
 			 * immediately afterwards.
 			 */
-			dlist_move_head(&MXactCache, iter.cur);
+			dclist_move_head(&MXactCache, iter.cur);
 
 			*members = ptr;
 			return entry->nmembers;
@@ -1610,14 +1609,13 @@ mXactCachePut(MultiXactId multi, int nmembers, MultiXactMember *members)
 	/* mXactCacheGetBySet assumes the entries are sorted, so sort them */
 	qsort(entry->members, nmembers, sizeof(MultiXactMember), mxactMemberComparator);
 
-	dlist_push_head(&MXactCache, &entry->node);
-	if (MXactCacheMembers++ >= MAX_CACHE_ENTRIES)
+	dclist_push_head(&MXactCache, &entry->node);
+	if (dclist_count(&MXactCache) > MAX_CACHE_ENTRIES)
 	{
 		dlist_node *node;
 
-		node = dlist_tail_node(&MXactCache);
-		dlist_delete(node);
-		MXactCacheMembers--;
+		node = dclist_tail_node(&MXactCache);
+		dclist_delete_from(&MXactCache, node);
 
 		entry = dlist_container(mXactCacheEnt, node, node);
 		debug_elog3(DEBUG2, "CachePut: pruning cached multi %u",
@@ -1699,8 +1697,7 @@ AtEOXact_MultiXact(void)
 	 * a child of TopTransactionContext, we needn't delete it explicitly.
 	 */
 	MXactContext = NULL;
-	dlist_init(&MXactCache);
-	MXactCacheMembers = 0;
+	dclist_init(&MXactCache);
 }
 
 /*
@@ -1766,8 +1763,7 @@ PostPrepare_MultiXact(TransactionId xid)
 	 * Discard the local MultiXactId cache like in AtEOXact_MultiXact.
 	 */
 	MXactContext = NULL;
-	dlist_init(&MXactCache);
-	MXactCacheMembers = 0;
+	dclist_init(&MXactCache);
 }
 
 /*
diff --git a/src/backend/lib/ilist.c b/src/backend/lib/ilist.c
index 29ef216212..eeb53d3110 100644
--- a/src/backend/lib/ilist.c
+++ b/src/backend/lib/ilist.c
@@ -52,6 +52,21 @@ slist_delete(slist_head *head, slist_node *node)
 }
 
 #ifdef ILIST_DEBUG
+/*
+ * dlist_member_check
+ *		Validate that 'node' is a member of 'head'
+ */
+void
+dlist_member_check(dlist_head *head, dlist_node *node)
+{
+	for (dlist_node *cur = &head->head; cur != NULL; cur = cur->next)
+	{
+		if (cur == node)
+			return;
+	}
+	elog(ERROR, "double linked list member check failure");
+}
+
 /*
  * Verify integrity of a doubly linked list
  */
@@ -107,5 +122,4 @@ slist_check(slist_head *head)
 	for (cur = head->head.next; cur != NULL; cur = cur->next)
 		;
 }
-
 #endif							/* ILIST_DEBUG */
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index c29894041e..0377f6d789 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -349,8 +349,6 @@ ReorderBufferAllocate(void)
 	buffer->by_txn_last_xid = InvalidTransactionId;
 	buffer->by_txn_last_txn = NULL;
 
-	buffer->catchange_ntxns = 0;
-
 	buffer->outbuf = NULL;
 	buffer->outbufsize = 0;
 	buffer->size = 0;
@@ -368,7 +366,7 @@ ReorderBufferAllocate(void)
 
 	dlist_init(&buffer->toplevel_by_lsn);
 	dlist_init(&buffer->txns_by_base_snapshot_lsn);
-	dlist_init(&buffer->catchange_txns);
+	dclist_init(&buffer->catchange_txns);
 
 	/*
 	 * Ensure there's no stale data from prior uses of this slot, in case some
@@ -1553,12 +1551,7 @@ ReorderBufferCleanupTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
 	 */
 	dlist_delete(&txn->node);
 	if (rbtxn_has_catalog_changes(txn))
-	{
-		dlist_delete(&txn->catchange_node);
-		rb->catchange_ntxns--;
-
-		Assert(rb->catchange_ntxns >= 0);
-	}
+		dclist_delete_from(&rb->catchange_txns, &txn->catchange_node);
 
 	/* now remove reference from buffer */
 	hash_search(rb->by_txn,
@@ -3309,8 +3302,7 @@ ReorderBufferXidSetCatalogChanges(ReorderBuffer *rb, TransactionId xid,
 	if (!rbtxn_has_catalog_changes(txn))
 	{
 		txn->txn_flags |= RBTXN_HAS_CATALOG_CHANGES;
-		dlist_push_tail(&rb->catchange_txns, &txn->catchange_node);
-		rb->catchange_ntxns++;
+		dclist_push_tail(&rb->catchange_txns, &txn->catchange_node);
 	}
 
 	/*
@@ -3323,8 +3315,7 @@ ReorderBufferXidSetCatalogChanges(ReorderBuffer *rb, TransactionId xid,
 	if (toptxn != NULL && !rbtxn_has_catalog_changes(toptxn))
 	{
 		toptxn->txn_flags |= RBTXN_HAS_CATALOG_CHANGES;
-		dlist_push_tail(&rb->catchange_txns, &toptxn->catchange_node);
-		rb->catchange_ntxns++;
+		dclist_push_tail(&rb->catchange_txns, &toptxn->catchange_node);
 	}
 }
 
@@ -3342,19 +3333,17 @@ ReorderBufferGetCatalogChangesXacts(ReorderBuffer *rb)
 	size_t		xcnt = 0;
 
 	/* Quick return if the list is empty */
-	if (rb->catchange_ntxns == 0)
-	{
-		Assert(dlist_is_empty(&rb->catchange_txns));
+	if (dclist_count(&rb->catchange_txns) == 0)
 		return NULL;
-	}
 
 	/* Initialize XID array */
-	xids = (TransactionId *) palloc(sizeof(TransactionId) * rb->catchange_ntxns);
-	dlist_foreach(iter, &rb->catchange_txns)
+	xids = (TransactionId *) palloc(sizeof(TransactionId) *
+									dclist_count(&rb->catchange_txns));
+	dclist_foreach(iter, &rb->catchange_txns)
 	{
-		ReorderBufferTXN *txn = dlist_container(ReorderBufferTXN,
-												catchange_node,
-												iter.cur);
+		ReorderBufferTXN *txn = dclist_container(ReorderBufferTXN,
+												 catchange_node,
+												 iter.cur);
 
 		Assert(rbtxn_has_catalog_changes(txn));
 
@@ -3363,7 +3352,7 @@ ReorderBufferGetCatalogChangesXacts(ReorderBuffer *rb)
 
 	qsort(xids, xcnt, sizeof(TransactionId), xidComparator);
 
-	Assert(xcnt == rb->catchange_ntxns);
+	Assert(xcnt == dclist_count(&rb->catchange_txns));
 	return xids;
 }
 
diff --git a/src/backend/replication/logical/snapbuild.c b/src/backend/replication/logical/snapbuild.c
index f892d19bfb..5006a5c464 100644
--- a/src/backend/replication/logical/snapbuild.c
+++ b/src/backend/replication/logical/snapbuild.c
@@ -1688,7 +1688,7 @@ SnapBuildSerialize(SnapBuild *builder, XLogRecPtr lsn)
 
 	/* Get the catalog modifying transactions that are yet not committed */
 	catchange_xip = ReorderBufferGetCatalogChangesXacts(builder->reorder);
-	catchange_xcnt = builder->reorder->catchange_ntxns;
+	catchange_xcnt = dclist_count(&builder->reorder->catchange_txns);
 
 	needed_length = sizeof(SnapBuildOnDisk) +
 		sizeof(TransactionId) * (builder->committed.xcnt + catchange_xcnt);
diff --git a/src/backend/utils/activity/pgstat_xact.c b/src/backend/utils/activity/pgstat_xact.c
index d6f660edf7..5a3aca4aef 100644
--- a/src/backend/utils/activity/pgstat_xact.c
+++ b/src/backend/utils/activity/pgstat_xact.c
@@ -70,16 +70,13 @@ AtEOXact_PgStat_DroppedStats(PgStat_SubXactStatus *xact_state, bool isCommit)
 	dlist_mutable_iter iter;
 	int			not_freed_count = 0;
 
-	if (xact_state->pending_drops_count == 0)
-	{
-		Assert(dlist_is_empty(&xact_state->pending_drops));
+	if (dclist_count(&xact_state->pending_drops) == 0)
 		return;
-	}
 
-	dlist_foreach_modify(iter, &xact_state->pending_drops)
+	dclist_foreach_modify(iter, &xact_state->pending_drops)
 	{
 		PgStat_PendingDroppedStatsItem *pending =
-		dlist_container(PgStat_PendingDroppedStatsItem, node, iter.cur);
+		dclist_container(PgStat_PendingDroppedStatsItem, node, iter.cur);
 		xl_xact_stats_item *it = &pending->item;
 
 		if (isCommit && !pending->is_create)
@@ -101,8 +98,7 @@ AtEOXact_PgStat_DroppedStats(PgStat_SubXactStatus *xact_state, bool isCommit)
 				not_freed_count++;
 		}
 
-		dlist_delete(&pending->node);
-		xact_state->pending_drops_count--;
+		dclist_delete_from(&xact_state->pending_drops, &pending->node);
 		pfree(pending);
 	}
 
@@ -144,19 +140,18 @@ AtEOSubXact_PgStat_DroppedStats(PgStat_SubXactStatus *xact_state,
 	dlist_mutable_iter iter;
 	int			not_freed_count = 0;
 
-	if (xact_state->pending_drops_count == 0)
+	if (dclist_count(&xact_state->pending_drops) == 0)
 		return;
 
 	parent_xact_state = pgstat_get_xact_stack_level(nestDepth - 1);
 
-	dlist_foreach_modify(iter, &xact_state->pending_drops)
+	dclist_foreach_modify(iter, &xact_state->pending_drops)
 	{
 		PgStat_PendingDroppedStatsItem *pending =
-		dlist_container(PgStat_PendingDroppedStatsItem, node, iter.cur);
+		dclist_container(PgStat_PendingDroppedStatsItem, node, iter.cur);
 		xl_xact_stats_item *it = &pending->item;
 
-		dlist_delete(&pending->node);
-		xact_state->pending_drops_count--;
+		dclist_delete_from(&xact_state->pending_drops, &pending->node);
 
 		if (!isCommit && pending->is_create)
 		{
@@ -175,8 +170,7 @@ AtEOSubXact_PgStat_DroppedStats(PgStat_SubXactStatus *xact_state,
 			 * remove the stats object, the surrounding transaction might
 			 * still abort. Pass it on to the parent.
 			 */
-			dlist_push_tail(&parent_xact_state->pending_drops, &pending->node);
-			parent_xact_state->pending_drops_count++;
+			dclist_push_tail(&parent_xact_state->pending_drops, &pending->node);
 		}
 		else
 		{
@@ -184,7 +178,7 @@ AtEOSubXact_PgStat_DroppedStats(PgStat_SubXactStatus *xact_state,
 		}
 	}
 
-	Assert(xact_state->pending_drops_count == 0);
+	Assert(dclist_count(&xact_state->pending_drops) == 0);
 	if (not_freed_count > 0)
 		pgstat_request_entry_refs_gc();
 }
@@ -250,8 +244,7 @@ pgstat_get_xact_stack_level(int nest_level)
 		xact_state = (PgStat_SubXactStatus *)
 			MemoryContextAlloc(TopTransactionContext,
 							   sizeof(PgStat_SubXactStatus));
-		dlist_init(&xact_state->pending_drops);
-		xact_state->pending_drops_count = 0;
+		dclist_init(&xact_state->pending_drops);
 		xact_state->nest_level = nest_level;
 		xact_state->prev = pgStatXactStack;
 		xact_state->first = NULL;
@@ -291,20 +284,20 @@ pgstat_get_transactional_drops(bool isCommit, xl_xact_stats_item **items)
 	Assert(!isCommit || xact_state->nest_level == 1);
 	Assert(!isCommit || xact_state->prev == NULL);
 
-	*items = palloc(xact_state->pending_drops_count
+	*items = palloc(dclist_count(&xact_state->pending_drops)
 					* sizeof(xl_xact_stats_item));
 
-	dlist_foreach(iter, &xact_state->pending_drops)
+	dclist_foreach(iter, &xact_state->pending_drops)
 	{
 		PgStat_PendingDroppedStatsItem *pending =
-		dlist_container(PgStat_PendingDroppedStatsItem, node, iter.cur);
+		dclist_container(PgStat_PendingDroppedStatsItem, node, iter.cur);
 
 		if (isCommit && pending->is_create)
 			continue;
 		if (!isCommit && !pending->is_create)
 			continue;
 
-		Assert(nitems < xact_state->pending_drops_count);
+		Assert(nitems < dclist_count(&xact_state->pending_drops));
 		(*items)[nitems++] = pending->item;
 	}
 
@@ -351,8 +344,7 @@ create_drop_transactional_internal(PgStat_Kind kind, Oid dboid, Oid objoid, bool
 	drop->item.dboid = dboid;
 	drop->item.objoid = objoid;
 
-	dlist_push_tail(&xact_state->pending_drops, &drop->node);
-	xact_state->pending_drops_count++;
+	dclist_push_tail(&xact_state->pending_drops, &drop->node);
 }
 
 /*
diff --git a/src/backend/utils/adt/ri_triggers.c b/src/backend/utils/adt/ri_triggers.c
index 1d503e7e01..3d24bc7dd6 100644
--- a/src/backend/utils/adt/ri_triggers.c
+++ b/src/backend/utils/adt/ri_triggers.c
@@ -176,8 +176,7 @@ typedef struct RI_CompareHashEntry
 static HTAB *ri_constraint_cache = NULL;
 static HTAB *ri_query_cache = NULL;
 static HTAB *ri_compare_cache = NULL;
-static dlist_head ri_constraint_cache_valid_list;
-static int	ri_constraint_cache_valid_count = 0;
+static dclist_head ri_constraint_cache_valid_list;
 
 
 /*
@@ -2174,8 +2173,7 @@ ri_LoadConstraintInfo(Oid constraintOid)
 	 * For efficient processing of invalidation messages below, we keep a
 	 * doubly-linked list, and a count, of all currently valid entries.
 	 */
-	dlist_push_tail(&ri_constraint_cache_valid_list, &riinfo->valid_link);
-	ri_constraint_cache_valid_count++;
+	dclist_push_tail(&ri_constraint_cache_valid_list, &riinfo->valid_link);
 
 	riinfo->valid = true;
 
@@ -2233,13 +2231,13 @@ InvalidateConstraintCacheCallBack(Datum arg, int cacheid, uint32 hashvalue)
 	 * O(N^2) behavior in situations where a session touches many foreign keys
 	 * and also does many ALTER TABLEs, such as a restore from pg_dump.
 	 */
-	if (ri_constraint_cache_valid_count > 1000)
+	if (dclist_count(&ri_constraint_cache_valid_list) > 1000)
 		hashvalue = 0;			/* pretend it's a cache reset */
 
-	dlist_foreach_modify(iter, &ri_constraint_cache_valid_list)
+	dclist_foreach_modify(iter, &ri_constraint_cache_valid_list)
 	{
-		RI_ConstraintInfo *riinfo = dlist_container(RI_ConstraintInfo,
-													valid_link, iter.cur);
+		RI_ConstraintInfo *riinfo = dclist_container(RI_ConstraintInfo,
+													 valid_link, iter.cur);
 
 		/*
 		 * We must invalidate not only entries directly matching the given
@@ -2252,8 +2250,7 @@ InvalidateConstraintCacheCallBack(Datum arg, int cacheid, uint32 hashvalue)
 		{
 			riinfo->valid = false;
 			/* Remove invalidated entries from the list, too */
-			dlist_delete(iter.cur);
-			ri_constraint_cache_valid_count--;
+			dclist_delete_from(&ri_constraint_cache_valid_list, iter.cur);
 		}
 	}
 }
diff --git a/src/include/lib/ilist.h b/src/include/lib/ilist.h
index 7ab0888f4f..d8726af6c9 100644
--- a/src/include/lib/ilist.h
+++ b/src/include/lib/ilist.h
@@ -10,20 +10,33 @@
  * the link fields in the remainder would be wasted space.  But usually,
  * it saves space to not have separately-allocated list nodes.)
  *
+ * The doubly-linked list comes in 2 forms.  dlist_head defines a head of a
+ * doubly-linked list of dlist_nodes.  Whereas dclist_head defines the head of
+ * a doubly-linked list of dlist_nodes with an additional 'count' field to
+ * keep track of how many items are contained within the given list.  For
+ * simplicity dlist_head and dclist_head share the same node type and
+ * iterator types.  Each function to manipulate a dlist_head is prefixed by
+ * dlist, whereas functions to manipulate a dclist_head are prefixed with
+ * dclist.  dclist_head comes with an additional function (dclist_count) to
+ * return the number of entries in the list.  dclists are able to store a
+ * maximum of UINT32 elements.  It is up to the caller to ensure no more than
+ * this many items are added to a dclist.
+ *
  * None of the functions here allocate any memory; they just manipulate
  * externally managed memory.  The APIs for singly and doubly linked lists
  * are identical as far as capabilities of both allow.
  *
  * Each list has a list header, which exists even when the list is empty.
  * An empty singly-linked list has a NULL pointer in its header.
- * There are two kinds of empty doubly linked lists: those that have been
- * initialized to NULL, and those that have been initialized to circularity.
- * (If a dlist is modified and then all its elements are deleted, it will be
- * in the circular state.)	We prefer circular dlists because there are some
- * operations that can be done without branches (and thus faster) on lists
- * that use circular representation.  However, it is often convenient to
- * initialize list headers to zeroes rather than setting them up with an
- * explicit initialization function, so we also allow the other case.
+ * For both dlist_head and dclist_head there are two kinds of empty doubly
+ * linked lists: those that have been initialized to NULL, and those that have
+ * been initialized to circularity.  (If a dlist is modified and then all its
+ * elements are deleted, it will be in the circular state.)	We prefer circular
+ * dlists because there are some operations that can be done without branches
+ * (and thus faster) on lists that use circular representation.  However, it
+ * is often convenient to initialize list headers to zeroes rather than
+ * setting them up with an explicit initialization function, so we also allow
+ * the other case.
  *
  * EXAMPLES
  *
@@ -182,6 +195,19 @@ typedef struct dlist_mutable_iter
 	dlist_node *end;			/* last node we'll iterate to */
 } dlist_mutable_iter;
 
+/*
+ * Head of a doubly linked list with a count of the number of items
+ *
+ * This internally makes use of a dlist to implement the actual list and
+ * wrapper functions are used to maintain the count when items are added and
+ * removed from the list.
+ */
+typedef struct dclist_head
+{
+	dlist_head	dlist;			/* the actual list header */
+	uint32		count;			/* the number of items in the list */
+} dclist_head;
+
 /*
  * Node of a singly linked list.
  *
@@ -246,6 +272,7 @@ typedef struct slist_mutable_iter
 
 /* Static initializers */
 #define DLIST_STATIC_INIT(name) {{&(name).head, &(name).head}}
+#define DCLIST_STATIC_INIT(name) {{{&(name).dlist.head, &(name).dlist.head}}, 0}
 #define SLIST_STATIC_INIT(name) {{NULL}}
 
 
@@ -255,6 +282,7 @@ typedef struct slist_mutable_iter
 extern void slist_delete(slist_head *head, slist_node *node);
 
 #ifdef ILIST_DEBUG
+extern void dlist_member_check(dlist_head *head, dlist_node *node);
 extern void dlist_check(dlist_head *head);
 extern void slist_check(slist_head *head);
 #else
@@ -264,6 +292,7 @@ extern void slist_check(slist_head *head);
  * in which functions the only point of passing the list head pointer is to be
  * able to run these checks.
  */
+#define dlist_member_check(head, node) ((void) (head))
 #define dlist_check(head)	((void) (head))
 #define slist_check(head)	((void) (head))
 #endif							/* ILIST_DEBUG */
@@ -361,6 +390,17 @@ dlist_delete(dlist_node *node)
 	node->next->prev = node->prev;
 }
 
+/*
+ * As dlist_delete but performs checks in ILIST_DEBUG to ensure that 'node'
+ * belongs to 'head'.
+ */
+static inline void
+dlist_delete_from(dlist_head *head, dlist_node *node)
+{
+	dlist_member_check(head, node);
+	dlist_delete(node);
+}
+
 /*
  * Remove and return the first node from a list (there must be one).
  */
@@ -562,6 +602,299 @@ dlist_tail_node(dlist_head *head)
 		 (iter).cur != (iter).end;											\
 		 (iter).cur = (iter).cur->prev)
 
+/* double linked list with count implementation */
+
+/*
+ * Initialize a doubly linked count list.
+ * Previous state will be thrown away without any cleanup.
+ */
+static inline void
+dclist_init(dclist_head *head)
+{
+	dlist_init(&head->dlist);
+	head->count = 0;
+}
+
+/*
+ * dclist_is_empty
+ *		Returns true if the list is empty, otherwise false.
+ */
+static inline bool
+dclist_is_empty(dclist_head *head)
+{
+	Assert(dlist_is_empty(&head->dlist) == (head->count == 0));
+
+	return (head->count == 0);
+}
+
+/*
+ * dclist_push_head
+ *		Insert a node at the beginning of the list.
+ */
+static inline void
+dclist_push_head(dclist_head *head, dlist_node *node)
+{
+	if (head->dlist.head.next == NULL)	/* convert NULL header to circular */
+		dclist_init(head);
+
+	dlist_push_head(&head->dlist, node);
+	head->count++;
+}
+
+/*
+ * dclist_push_tail
+ *		Insert a node at the end of the list.
+ */
+static inline void
+dclist_push_tail(dclist_head *head, dlist_node *node)
+{
+	if (head->dlist.head.next == NULL)	/* convert NULL header to circular */
+		dclist_init(head);
+
+	dlist_push_tail(&head->dlist, node);
+	head->count++;
+}
+
+/*
+ * dclist_insert_after
+ *		Insert a node after another *in the same list*
+ *
+ * Caller must ensure that 'after' is a member of 'head'.
+ */
+static inline void
+dclist_insert_after(dclist_head *head, dlist_node *after, dlist_node *node)
+{
+	dlist_member_check(&head->dlist, after);
+	Assert(head->count > 0);
+
+	dlist_insert_after(after, node);
+	head->count++;
+}
+
+/*
+ * dclist_insert_before
+ *		Insert a node before another *in the same list*
+ *
+ * Caller must ensure that 'before' is a member of 'head'.
+ */
+static inline void
+dclist_insert_before(dclist_head *head, dlist_node *before, dlist_node *node)
+{
+	dlist_member_check(&head->dlist, before);
+	Assert(head->count > 0);
+
+	dlist_insert_before(before, node);
+	head->count++;
+}
+
+/*
+ * dclist_delete_from
+ *		Deletes 'node' from 'head'.
+ *
+ * Caution: 'node' must be a member of 'head'.
+ */
+static inline void
+dclist_delete_from(dclist_head *head, dlist_node *node)
+{
+	Assert(head->count > 0);
+	dlist_delete_from(&head->dlist, node);
+	head->count--;
+}
+
+/*
+ * dclist_pop_head_node
+ *		Remove and return the first node from a list (there must be one).
+ */
+static inline dlist_node *
+dclist_pop_head_node(dclist_head *head)
+{
+	dlist_node *node;
+
+	Assert(head->count > 0);
+
+	node = dlist_pop_head_node(&head->dlist);
+	head->count--;
+
+	return node;
+}
+
+/*
+ * dclist_move_head
+ *		Move 'node' from its current position in the list to the head position
+ *		in 'head'.
+ *
+ * Caution: 'node' must be a member of 'head'.
+ */
+static inline void
+dclist_move_head(dclist_head *head, dlist_node *node)
+{
+	dlist_member_check(&head->dlist, node);
+	Assert(head->count > 0);
+
+	dlist_move_head(&head->dlist, node);
+}
+
+/*
+ * dclist_move_tail
+ *		Move 'node' from its current position in the list to the tail position
+ *		in 'head'.
+ *
+ * Caution: 'node' must be a member of 'head'.
+ */
+static inline void
+dclist_move_tail(dclist_head *head, dlist_node *node)
+{
+	dlist_member_check(&head->dlist, node);
+	Assert(head->count > 0);
+
+	dlist_move_tail(&head->dlist, node);
+}
+
+/*
+ * dclist_has_next
+ *		Check whether 'node' has a following node.
+ *
+ * Caution: 'node' must be a member of 'head'.
+ */
+static inline bool
+dclist_has_next(dclist_head *head, dlist_node *node)
+{
+	dlist_member_check(&head->dlist, node);
+	Assert(head->count > 0);
+
+	return dlist_has_next(&head->dlist, node);
+}
+
+/*
+ * dclist_has_prev
+ *		Check whether 'node' has a preceding node.
+ *
+ * Caution: 'node' must be a member of 'head'.
+ */
+static inline bool
+dclist_has_prev(dclist_head *head, dlist_node *node)
+{
+	dlist_member_check(&head->dlist, node);
+
+	Assert(head->count > 0);
+
+	return dlist_has_prev(&head->dlist, node);
+}
+
+/*
+ * dclist_next_node
+ *		Return the next node in the list (there must be one).
+ */
+static inline dlist_node *
+dclist_next_node(dclist_head *head, dlist_node *node)
+{
+	Assert(head->count > 0);
+
+	return dlist_next_node(&head->dlist, node);
+}
+
+/*
+ * dclist_prev_node
+ *		Return the prev node in the list (there must be one).
+ */
+static inline dlist_node *
+dclist_prev_node(dclist_head *head, dlist_node *node)
+{
+	Assert(head->count > 0);
+
+	return dlist_prev_node(&head->dlist, node);
+}
+
+/* internal support function to get address of head element's struct */
+static inline void *
+dclist_head_element_off(dclist_head *head, size_t off)
+{
+	Assert(!dclist_is_empty(head));
+	return (char *) head->dlist.head.next - off;
+}
+
+/*
+ * dclist_head_node
+ *		Return the first node in the list (there must be one).
+ */
+static inline dlist_node *
+dclist_head_node(dclist_head *head)
+{
+	Assert(head->count > 0);
+
+	return (dlist_node *) dlist_head_element_off(&head->dlist, 0);
+}
+
+/* internal support function to get address of tail element's struct */
+static inline void *
+dclist_tail_element_off(dclist_head *head, size_t off)
+{
+	Assert(!dclist_is_empty(head));
+	return (char *) head->dlist.head.prev - off;
+}
+
+/*
+ * Return the last node in the list (there must be one).
+ */
+static inline dlist_node *
+dclist_tail_node(dclist_head *head)
+{
+	Assert(head->count > 0);
+
+	return (dlist_node *) dlist_tail_element_off(&head->dlist, 0);
+}
+
+/*
+ * dclist_count
+ *		Returns the stored number of entries in 'head'
+ */
+static inline uint32
+dclist_count(dclist_head *head)
+{
+	Assert(dlist_is_empty(&head->dlist) == (head->count == 0));
+
+	return head->count;
+}
+
+/*
+ * Return the containing struct of 'type' where 'membername' is the dlist_node
+ * pointed at by 'ptr'.
+ *
+ * This is used to convert a dlist_node * back to its containing struct.
+ *
+ * Note: This is effectively just the same as dlist_container, so reuse that.
+ */
+#define dclist_container(type, membername, ptr) \
+		dlist_container(type, membername, ptr)
+
+ /*
+  * Return the address of the first element in the list.
+  *
+  * The list must not be empty.
+  */
+#define dclist_head_element(type, membername, lhead)							\
+	(AssertVariableIsOfTypeMacro(((type *) NULL)->membername, dlist_node),	\
+	 (type *) dclist_head_element_off(lhead, offsetof(type, membername)))
+
+ /*
+  * Return the address of the last element in the list.
+  *
+  * The list must not be empty.
+  */
+#define dclist_tail_element(type, membername, lhead)							\
+	(AssertVariableIsOfTypeMacro(((type *) NULL)->membername, dlist_node),	\
+	 ((type *) dclist_tail_element_off(lhead, offsetof(type, membername))))
+
+
+/* Iterators for dclists */
+#define dclist_foreach(iter, lhead) \
+	dlist_foreach(iter, &((lhead)->dlist))
+
+#define dclist_foreach_modify(iter, lhead) \
+	dlist_foreach_modify(iter, &((lhead)->dlist))
+
+#define dclist_reverse_foreach(iter, lhead) \
+	dlist_reverse_foreach(iter, &((lhead)->dlist))
 
 /* singly linked list implementation */
 
diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h
index 02b59a1931..b23d8cc4f9 100644
--- a/src/include/replication/reorderbuffer.h
+++ b/src/include/replication/reorderbuffer.h
@@ -534,8 +534,7 @@ struct ReorderBuffer
 	/*
 	 * Transactions and subtransactions that have modified system catalogs.
 	 */
-	dlist_head	catchange_txns;
-	int			catchange_ntxns;
+	dclist_head catchange_txns;
 
 	/*
 	 * one-entry sized cache for by_txn. Very frequently the same txn gets
diff --git a/src/include/utils/pgstat_internal.h b/src/include/utils/pgstat_internal.h
index 627c1389e4..9f32a6f6d3 100644
--- a/src/include/utils/pgstat_internal.h
+++ b/src/include/utils/pgstat_internal.h
@@ -162,8 +162,7 @@ typedef struct PgStat_SubXactStatus
 	 * if the transaction commits/aborts. To handle replicas and crashes,
 	 * stats drops are included in commit / abort records.
 	 */
-	dlist_head	pending_drops;
-	int			pending_drops_count;
+	dclist_head pending_drops;
 
 	/*
 	 * Tuple insertion/deletion counts for an open transaction can't be
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 2f02cc8f42..c15f990cbb 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -3194,6 +3194,7 @@ datapagemap_t
 dateKEY
 datetkn
 dce_uuid_t
+dclist_head
 decimal
 deparse_columns
 deparse_context

diff --git a/src/backend/access/heap/rewriteheap.c b/src/backend/access/heap/rewriteheap.c
index b01b39b008..a34e9b352d 100644
--- a/src/backend/access/heap/rewriteheap.c
+++ b/src/backend/access/heap/rewriteheap.c
@@ -196,8 +196,7 @@ typedef struct RewriteMappingFile
 	TransactionId xid;			/* xid that might need to see the row */
 	int			vfd;			/* fd of mappings file */
 	off_t		off;			/* how far have we written yet */
-	uint32		num_mappings;	/* number of in-memory mappings */
-	dlist_head	mappings;		/* list of in-memory mappings */
+	dclist_head mappings;		/* list of in-memory mappings */
 	char		path[MAXPGPATH];	/* path, for error messages */
 } RewriteMappingFile;
 
@@ -864,9 +863,10 @@ logical_heap_rewrite_flush_mappings(RewriteState state)
 		Oid			dboid;
 		uint32		len;
 		int			written;
+		uint32		num_mappings = dclist_count(&src->mappings);
 
 		/* this file hasn't got any new mappings */
-		if (src->num_mappings == 0)
+		if (num_mappings == 0)
 			continue;
 
 		if (state->rs_old_rel->rd_rel->relisshared)
@@ -874,7 +874,7 @@ logical_heap_rewrite_flush_mappings(RewriteState state)
 		else
 			dboid = MyDatabaseId;
 
-		xlrec.num_mappings = src->num_mappings;
+		xlrec.num_mappings = num_mappings;
 		xlrec.mapped_rel = RelationGetRelid(state->rs_old_rel);
 		xlrec.mapped_xid = src->xid;
 		xlrec.mapped_db = dboid;
@@ -882,31 +882,30 @@ logical_heap_rewrite_flush_mappings(RewriteState state)
 		xlrec.start_lsn = state->rs_begin_lsn;
 
 		/* write all mappings consecutively */
-		len = src->num_mappings * sizeof(LogicalRewriteMappingData);
+		len = num_mappings * sizeof(LogicalRewriteMappingData);
 		waldata_start = waldata = palloc(len);
 
 		/*
 		 * collect data we need to write out, but don't modify ondisk data yet
 		 */
-		dlist_foreach_modify(iter, &src->mappings)
+		dclist_foreach_modify(iter, &src->mappings)
 		{
 			RewriteMappingDataEntry *pmap;
 
-			pmap = dlist_container(RewriteMappingDataEntry, node, iter.cur);
+			pmap = dclist_container(RewriteMappingDataEntry, node, iter.cur);
 
 			memcpy(waldata, &pmap->map, sizeof(pmap->map));
 			waldata += sizeof(pmap->map);
 
 			/* remove from the list and free */
-			dlist_delete(&pmap->node);
+			dclist_delete_from(&src->mappings, &pmap->node);
 			pfree(pmap);
 
 			/* update bookkeeping */
 			state->rs_num_rewrite_mappings--;
-			src->num_mappings--;
 		}
 
-		Assert(src->num_mappings == 0);
+		Assert(dclist_count(&src->mappings) == 0);
 		Assert(waldata == waldata_start + len);
 
 		/*
@@ -1002,8 +1001,7 @@ logical_rewrite_log_mapping(RewriteState state, TransactionId xid,
 				 LSN_FORMAT_ARGS(state->rs_begin_lsn),
 				 xid, GetCurrentTransactionId());
 
-		dlist_init(&src->mappings);
-		src->num_mappings = 0;
+		dclist_init(&src->mappings);
 		src->off = 0;
 		memcpy(src->path, path, sizeof(path));
 		src->vfd = PathNameOpenFile(path,
@@ -1017,8 +1015,7 @@ logical_rewrite_log_mapping(RewriteState state, TransactionId xid,
 	pmap = MemoryContextAlloc(state->rs_cxt,
 							  sizeof(RewriteMappingDataEntry));
 	memcpy(&pmap->map, map, sizeof(LogicalRewriteMappingData));
-	dlist_push_tail(&src->mappings, &pmap->node);
-	src->num_mappings++;
+	dclist_push_tail(&src->mappings, &pmap->node);
 	state->rs_num_rewrite_mappings++;
 
 	/*
diff --git a/src/backend/access/transam/multixact.c b/src/backend/access/transam/multixact.c
index 5eff87665b..09c9409c31 100644
--- a/src/backend/access/transam/multixact.c
+++ b/src/backend/access/transam/multixact.c
@@ -319,8 +319,7 @@ typedef struct mXactCacheEnt
 } mXactCacheEnt;
 
 #define MAX_CACHE_ENTRIES	256
-static dlist_head MXactCache = DLIST_STATIC_INIT(MXactCache);
-static int	MXactCacheMembers = 0;
+static dclist_head MXactCache = DCLIST_STATIC_INIT(MXactCache);
 static MemoryContext MXactContext = NULL;
 
 #ifdef MULTIXACT_DEBUG
@@ -1504,9 +1503,9 @@ mXactCacheGetBySet(int nmembers, MultiXactMember *members)
 	/* sort the array so comparison is easy */
 	qsort(members, nmembers, sizeof(MultiXactMember), mxactMemberComparator);
 
-	dlist_foreach(iter, &MXactCache)
+	dclist_foreach(iter, &MXactCache)
 	{
-		mXactCacheEnt *entry = dlist_container(mXactCacheEnt, node, iter.cur);
+		mXactCacheEnt *entry = dclist_container(mXactCacheEnt, node, iter.cur);
 
 		if (entry->nmembers != nmembers)
 			continue;
@@ -1518,7 +1517,7 @@ mXactCacheGetBySet(int nmembers, MultiXactMember *members)
 		if (memcmp(members, entry->members, nmembers * sizeof(MultiXactMember)) == 0)
 		{
 			debug_elog3(DEBUG2, "CacheGet: found %u", entry->multi);
-			dlist_move_head(&MXactCache, iter.cur);
+			dclist_move_head(&MXactCache, iter.cur);
 			return entry->multi;
 		}
 	}
@@ -1542,9 +1541,9 @@ mXactCacheGetById(MultiXactId multi, MultiXactMember **members)
 
 	debug_elog3(DEBUG2, "CacheGet: looking for %u", multi);
 
-	dlist_foreach(iter, &MXactCache)
+	dclist_foreach(iter, &MXactCache)
 	{
-		mXactCacheEnt *entry = dlist_container(mXactCacheEnt, node, iter.cur);
+		mXactCacheEnt *entry = dclist_container(mXactCacheEnt, node, iter.cur);
 
 		if (entry->multi == multi)
 		{
@@ -1566,7 +1565,7 @@ mXactCacheGetById(MultiXactId multi, MultiXactMember **members)
 			 * This is acceptable only because we exit the iteration
 			 * immediately afterwards.
 			 */
-			dlist_move_head(&MXactCache, iter.cur);
+			dclist_move_head(&MXactCache, iter.cur);
 
 			*members = ptr;
 			return entry->nmembers;
@@ -1610,16 +1609,15 @@ mXactCachePut(MultiXactId multi, int nmembers, MultiXactMember *members)
 	/* mXactCacheGetBySet assumes the entries are sorted, so sort them */
 	qsort(entry->members, nmembers, sizeof(MultiXactMember), mxactMemberComparator);
 
-	dlist_push_head(&MXactCache, &entry->node);
-	if (MXactCacheMembers++ >= MAX_CACHE_ENTRIES)
+	dclist_push_head(&MXactCache, &entry->node);
+	if (dclist_count(&MXactCache) > MAX_CACHE_ENTRIES)
 	{
 		dlist_node *node;
 
-		node = dlist_tail_node(&MXactCache);
-		dlist_delete(node);
-		MXactCacheMembers--;
+		node = dclist_tail_node(&MXactCache);
+		dclist_delete_from(&MXactCache, node);
 
-		entry = dlist_container(mXactCacheEnt, node, node);
+		entry = dclist_container(mXactCacheEnt, node, node);
 		debug_elog3(DEBUG2, "CachePut: pruning cached multi %u",
 					entry->multi);
 
@@ -1699,8 +1697,7 @@ AtEOXact_MultiXact(void)
 	 * a child of TopTransactionContext, we needn't delete it explicitly.
 	 */
 	MXactContext = NULL;
-	dlist_init(&MXactCache);
-	MXactCacheMembers = 0;
+	dclist_init(&MXactCache);
 }
 
 /*
@@ -1766,8 +1763,7 @@ PostPrepare_MultiXact(TransactionId xid)
 	 * Discard the local MultiXactId cache like in AtEOXact_MultiXact.
 	 */
 	MXactContext = NULL;
-	dlist_init(&MXactCache);
-	MXactCacheMembers = 0;
+	dclist_init(&MXactCache);
 }
 
 /*
diff --git a/src/backend/lib/ilist.c b/src/backend/lib/ilist.c
index 29ef216212..e8ea981176 100644
--- a/src/backend/lib/ilist.c
+++ b/src/backend/lib/ilist.c
@@ -52,6 +52,23 @@ slist_delete(slist_head *head, slist_node *node)
 }
 
 #ifdef ILIST_DEBUG
+/*
+ * dlist_member_check
+ *		Validate that 'node' is a member of 'head'
+ */
+void
+dlist_member_check(dlist_head *head, dlist_node *node)
+{
+	dlist_iter	iter;
+
+	dlist_foreach(iter, head)
+	{
+		if (iter.cur == node)
+			return;
+	}
+	elog(ERROR, "double linked list member check failure");
+}
+
 /*
  * Verify integrity of a doubly linked list
  */
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index 20c9cd139a..22a15a482a 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -349,8 +349,6 @@ ReorderBufferAllocate(void)
 	buffer->by_txn_last_xid = InvalidTransactionId;
 	buffer->by_txn_last_txn = NULL;
 
-	buffer->catchange_ntxns = 0;
-
 	buffer->outbuf = NULL;
 	buffer->outbufsize = 0;
 	buffer->size = 0;
@@ -368,7 +366,7 @@ ReorderBufferAllocate(void)
 
 	dlist_init(&buffer->toplevel_by_lsn);
 	dlist_init(&buffer->txns_by_base_snapshot_lsn);
-	dlist_init(&buffer->catchange_txns);
+	dclist_init(&buffer->catchange_txns);
 
 	/*
 	 * Ensure there's no stale data from prior uses of this slot, in case some
@@ -1553,12 +1551,7 @@ ReorderBufferCleanupTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
 	 */
 	dlist_delete(&txn->node);
 	if (rbtxn_has_catalog_changes(txn))
-	{
-		dlist_delete(&txn->catchange_node);
-		rb->catchange_ntxns--;
-
-		Assert(rb->catchange_ntxns >= 0);
-	}
+		dclist_delete_from(&rb->catchange_txns, &txn->catchange_node);
 
 	/* now remove reference from buffer */
 	hash_search(rb->by_txn,
@@ -3309,8 +3302,7 @@ ReorderBufferXidSetCatalogChanges(ReorderBuffer *rb, TransactionId xid,
 	if (!rbtxn_has_catalog_changes(txn))
 	{
 		txn->txn_flags |= RBTXN_HAS_CATALOG_CHANGES;
-		dlist_push_tail(&rb->catchange_txns, &txn->catchange_node);
-		rb->catchange_ntxns++;
+		dclist_push_tail(&rb->catchange_txns, &txn->catchange_node);
 	}
 
 	/*
@@ -3323,8 +3315,7 @@ ReorderBufferXidSetCatalogChanges(ReorderBuffer *rb, TransactionId xid,
 	if (toptxn != NULL && !rbtxn_has_catalog_changes(toptxn))
 	{
 		toptxn->txn_flags |= RBTXN_HAS_CATALOG_CHANGES;
-		dlist_push_tail(&rb->catchange_txns, &toptxn->catchange_node);
-		rb->catchange_ntxns++;
+		dclist_push_tail(&rb->catchange_txns, &toptxn->catchange_node);
 	}
 }
 
@@ -3342,19 +3333,17 @@ ReorderBufferGetCatalogChangesXacts(ReorderBuffer *rb)
 	size_t		xcnt = 0;
 
 	/* Quick return if the list is empty */
-	if (rb->catchange_ntxns == 0)
-	{
-		Assert(dlist_is_empty(&rb->catchange_txns));
+	if (dclist_count(&rb->catchange_txns) == 0)
 		return NULL;
-	}
 
 	/* Initialize XID array */
-	xids = (TransactionId *) palloc(sizeof(TransactionId) * rb->catchange_ntxns);
-	dlist_foreach(iter, &rb->catchange_txns)
+	xids = (TransactionId *) palloc(sizeof(TransactionId) *
+									dclist_count(&rb->catchange_txns));
+	dclist_foreach(iter, &rb->catchange_txns)
 	{
-		ReorderBufferTXN *txn = dlist_container(ReorderBufferTXN,
-												catchange_node,
-												iter.cur);
+		ReorderBufferTXN *txn = dclist_container(ReorderBufferTXN,
+												 catchange_node,
+												 iter.cur);
 
 		Assert(rbtxn_has_catalog_changes(txn));
 
@@ -3363,7 +3352,7 @@ ReorderBufferGetCatalogChangesXacts(ReorderBuffer *rb)
 
 	qsort(xids, xcnt, sizeof(TransactionId), xidComparator);
 
-	Assert(xcnt == rb->catchange_ntxns);
+	Assert(xcnt == dclist_count(&rb->catchange_txns));
 	return xids;
 }
 
diff --git a/src/backend/replication/logical/snapbuild.c b/src/backend/replication/logical/snapbuild.c
index f892d19bfb..5006a5c464 100644
--- a/src/backend/replication/logical/snapbuild.c
+++ b/src/backend/replication/logical/snapbuild.c
@@ -1688,7 +1688,7 @@ SnapBuildSerialize(SnapBuild *builder, XLogRecPtr lsn)
 
 	/* Get the catalog modifying transactions that are yet not committed */
 	catchange_xip = ReorderBufferGetCatalogChangesXacts(builder->reorder);
-	catchange_xcnt = builder->reorder->catchange_ntxns;
+	catchange_xcnt = dclist_count(&builder->reorder->catchange_txns);
 
 	needed_length = sizeof(SnapBuildOnDisk) +
 		sizeof(TransactionId) * (builder->committed.xcnt + catchange_xcnt);
diff --git a/src/backend/utils/activity/pgstat_xact.c b/src/backend/utils/activity/pgstat_xact.c
index d6f660edf7..5a3aca4aef 100644
--- a/src/backend/utils/activity/pgstat_xact.c
+++ b/src/backend/utils/activity/pgstat_xact.c
@@ -70,16 +70,13 @@ AtEOXact_PgStat_DroppedStats(PgStat_SubXactStatus *xact_state, bool isCommit)
 	dlist_mutable_iter iter;
 	int			not_freed_count = 0;
 
-	if (xact_state->pending_drops_count == 0)
-	{
-		Assert(dlist_is_empty(&xact_state->pending_drops));
+	if (dclist_count(&xact_state->pending_drops) == 0)
 		return;
-	}
 
-	dlist_foreach_modify(iter, &xact_state->pending_drops)
+	dclist_foreach_modify(iter, &xact_state->pending_drops)
 	{
 		PgStat_PendingDroppedStatsItem *pending =
-		dlist_container(PgStat_PendingDroppedStatsItem, node, iter.cur);
+		dclist_container(PgStat_PendingDroppedStatsItem, node, iter.cur);
 		xl_xact_stats_item *it = &pending->item;
 
 		if (isCommit && !pending->is_create)
@@ -101,8 +98,7 @@ AtEOXact_PgStat_DroppedStats(PgStat_SubXactStatus *xact_state, bool isCommit)
 				not_freed_count++;
 		}
 
-		dlist_delete(&pending->node);
-		xact_state->pending_drops_count--;
+		dclist_delete_from(&xact_state->pending_drops, &pending->node);
 		pfree(pending);
 	}
 
@@ -144,19 +140,18 @@ AtEOSubXact_PgStat_DroppedStats(PgStat_SubXactStatus *xact_state,
 	dlist_mutable_iter iter;
 	int			not_freed_count = 0;
 
-	if (xact_state->pending_drops_count == 0)
+	if (dclist_count(&xact_state->pending_drops) == 0)
 		return;
 
 	parent_xact_state = pgstat_get_xact_stack_level(nestDepth - 1);
 
-	dlist_foreach_modify(iter, &xact_state->pending_drops)
+	dclist_foreach_modify(iter, &xact_state->pending_drops)
 	{
 		PgStat_PendingDroppedStatsItem *pending =
-		dlist_container(PgStat_PendingDroppedStatsItem, node, iter.cur);
+		dclist_container(PgStat_PendingDroppedStatsItem, node, iter.cur);
 		xl_xact_stats_item *it = &pending->item;
 
-		dlist_delete(&pending->node);
-		xact_state->pending_drops_count--;
+		dclist_delete_from(&xact_state->pending_drops, &pending->node);
 
 		if (!isCommit && pending->is_create)
 		{
@@ -175,8 +170,7 @@ AtEOSubXact_PgStat_DroppedStats(PgStat_SubXactStatus *xact_state,
 			 * remove the stats object, the surrounding transaction might
 			 * still abort. Pass it on to the parent.
 			 */
-			dlist_push_tail(&parent_xact_state->pending_drops, &pending->node);
-			parent_xact_state->pending_drops_count++;
+			dclist_push_tail(&parent_xact_state->pending_drops, &pending->node);
 		}
 		else
 		{
@@ -184,7 +178,7 @@ AtEOSubXact_PgStat_DroppedStats(PgStat_SubXactStatus *xact_state,
 		}
 	}
 
-	Assert(xact_state->pending_drops_count == 0);
+	Assert(dclist_count(&xact_state->pending_drops) == 0);
 	if (not_freed_count > 0)
 		pgstat_request_entry_refs_gc();
 }
@@ -250,8 +244,7 @@ pgstat_get_xact_stack_level(int nest_level)
 		xact_state = (PgStat_SubXactStatus *)
 			MemoryContextAlloc(TopTransactionContext,
 							   sizeof(PgStat_SubXactStatus));
-		dlist_init(&xact_state->pending_drops);
-		xact_state->pending_drops_count = 0;
+		dclist_init(&xact_state->pending_drops);
 		xact_state->nest_level = nest_level;
 		xact_state->prev = pgStatXactStack;
 		xact_state->first = NULL;
@@ -291,20 +284,20 @@ pgstat_get_transactional_drops(bool isCommit, xl_xact_stats_item **items)
 	Assert(!isCommit || xact_state->nest_level == 1);
 	Assert(!isCommit || xact_state->prev == NULL);
 
-	*items = palloc(xact_state->pending_drops_count
+	*items = palloc(dclist_count(&xact_state->pending_drops)
 					* sizeof(xl_xact_stats_item));
 
-	dlist_foreach(iter, &xact_state->pending_drops)
+	dclist_foreach(iter, &xact_state->pending_drops)
 	{
 		PgStat_PendingDroppedStatsItem *pending =
-		dlist_container(PgStat_PendingDroppedStatsItem, node, iter.cur);
+		dclist_container(PgStat_PendingDroppedStatsItem, node, iter.cur);
 
 		if (isCommit && pending->is_create)
 			continue;
 		if (!isCommit && !pending->is_create)
 			continue;
 
-		Assert(nitems < xact_state->pending_drops_count);
+		Assert(nitems < dclist_count(&xact_state->pending_drops));
 		(*items)[nitems++] = pending->item;
 	}
 
@@ -351,8 +344,7 @@ create_drop_transactional_internal(PgStat_Kind kind, Oid dboid, Oid objoid, bool
 	drop->item.dboid = dboid;
 	drop->item.objoid = objoid;
 
-	dlist_push_tail(&xact_state->pending_drops, &drop->node);
-	xact_state->pending_drops_count++;
+	dclist_push_tail(&xact_state->pending_drops, &drop->node);
 }
 
 /*
diff --git a/src/backend/utils/adt/ri_triggers.c b/src/backend/utils/adt/ri_triggers.c
index 1d503e7e01..61c2eecaca 100644
--- a/src/backend/utils/adt/ri_triggers.c
+++ b/src/backend/utils/adt/ri_triggers.c
@@ -176,8 +176,7 @@ typedef struct RI_CompareHashEntry
 static HTAB *ri_constraint_cache = NULL;
 static HTAB *ri_query_cache = NULL;
 static HTAB *ri_compare_cache = NULL;
-static dlist_head ri_constraint_cache_valid_list;
-static int	ri_constraint_cache_valid_count = 0;
+static dclist_head ri_constraint_cache_valid_list;
 
 
 /*
@@ -2172,10 +2171,9 @@ ri_LoadConstraintInfo(Oid constraintOid)
 
 	/*
 	 * For efficient processing of invalidation messages below, we keep a
-	 * doubly-linked list, and a count, of all currently valid entries.
+	 * doubly-linked count list of all currently valid entries.
 	 */
-	dlist_push_tail(&ri_constraint_cache_valid_list, &riinfo->valid_link);
-	ri_constraint_cache_valid_count++;
+	dclist_push_tail(&ri_constraint_cache_valid_list, &riinfo->valid_link);
 
 	riinfo->valid = true;
 
@@ -2233,13 +2231,13 @@ InvalidateConstraintCacheCallBack(Datum arg, int cacheid, uint32 hashvalue)
 	 * O(N^2) behavior in situations where a session touches many foreign keys
 	 * and also does many ALTER TABLEs, such as a restore from pg_dump.
 	 */
-	if (ri_constraint_cache_valid_count > 1000)
+	if (dclist_count(&ri_constraint_cache_valid_list) > 1000)
 		hashvalue = 0;			/* pretend it's a cache reset */
 
-	dlist_foreach_modify(iter, &ri_constraint_cache_valid_list)
+	dclist_foreach_modify(iter, &ri_constraint_cache_valid_list)
 	{
-		RI_ConstraintInfo *riinfo = dlist_container(RI_ConstraintInfo,
-													valid_link, iter.cur);
+		RI_ConstraintInfo *riinfo = dclist_container(RI_ConstraintInfo,
+													 valid_link, iter.cur);
 
 		/*
 		 * We must invalidate not only entries directly matching the given
@@ -2252,8 +2250,7 @@ InvalidateConstraintCacheCallBack(Datum arg, int cacheid, uint32 hashvalue)
 		{
 			riinfo->valid = false;
 			/* Remove invalidated entries from the list, too */
-			dlist_delete(iter.cur);
-			ri_constraint_cache_valid_count--;
+			dclist_delete_from(&ri_constraint_cache_valid_list, iter.cur);
 		}
 	}
 }
diff --git a/src/include/lib/ilist.h b/src/include/lib/ilist.h
index 7ab0888f4f..4ac0afe5e7 100644
--- a/src/include/lib/ilist.h
+++ b/src/include/lib/ilist.h
@@ -10,20 +10,36 @@
  * the link fields in the remainder would be wasted space.  But usually,
  * it saves space to not have separately-allocated list nodes.)
  *
+ * The doubly-linked list comes in 2 forms.  dlist_head defines a head of a
+ * doubly-linked list of dlist_nodes.  Whereas dclist_head defines the head of
+ * a doubly-linked list of dlist_nodes with an additional 'count' field to
+ * keep track of how many items are contained within the given list.  For
+ * simplicity dlist_head and dclist_head share the same node and iterator
+ * types.  The functions to manipulate a dlist_head always have a name
+ * starting with "dlist", whereas functions to manipulate a dclist_head have a
+ * name starting with "dclist".  dclist_head comes with an additional function
+ * (dclist_count) to return the number of entries in the list.  dclists are
+ * able to store a maximum of PG_UINT32_MAX elements.  It is up to the caller
+ * to ensure no more than this many items are added to a dclist.
+ *
  * None of the functions here allocate any memory; they just manipulate
- * externally managed memory.  The APIs for singly and doubly linked lists
- * are identical as far as capabilities of both allow.
+ * externally managed memory.  With the exception doubly-linked count lists
+ * providing the ability to obtain the number of items in the list, the APIs
+ * for singly and both doubly linked lists are identical as far as
+ * capabilities of both allow.
  *
  * Each list has a list header, which exists even when the list is empty.
  * An empty singly-linked list has a NULL pointer in its header.
- * There are two kinds of empty doubly linked lists: those that have been
- * initialized to NULL, and those that have been initialized to circularity.
+ *
+ * For both doubly-linked list types, there are two valid ways to represent an
+ * empty list.  The head's 'next' pointer can either be NULL or the head's
+ * 'next' and 'prev' links can both point back to the list head (circular).
  * (If a dlist is modified and then all its elements are deleted, it will be
- * in the circular state.)	We prefer circular dlists because there are some
+ * in the circular state.).  We prefer circular dlists because there are some
  * operations that can be done without branches (and thus faster) on lists
  * that use circular representation.  However, it is often convenient to
  * initialize list headers to zeroes rather than setting them up with an
- * explicit initialization function, so we also allow the other case.
+ * explicit initialization function, so we also allow the NULL initalization.
  *
  * EXAMPLES
  *
@@ -146,15 +162,17 @@ typedef struct dlist_head
 
 
 /*
- * Doubly linked list iterator.
+ * Doubly linked list iterator type for dlist_head and and dclist_head types.
+ *
+ * Used as state in dlist_foreach() and dlist_reverse_foreach() (and the
+ * dclist variant thereof).
  *
- * Used as state in dlist_foreach() and dlist_reverse_foreach(). To get the
- * current element of the iteration use the 'cur' member.
+ * To get the current element of the iteration use the 'cur' member.
  *
  * Iterations using this are *not* allowed to change the list while iterating!
  *
  * NB: We use an extra "end" field here to avoid multiple evaluations of
- * arguments in the dlist_foreach() macro.
+ * arguments in the dlist_foreach() and dclist_foreach() macros.
  */
 typedef struct dlist_iter
 {
@@ -163,10 +181,12 @@ typedef struct dlist_iter
 } dlist_iter;
 
 /*
- * Doubly linked list iterator allowing some modifications while iterating.
+ * Doubly linked list iterator for both dlist_head and dclist_head types.
+ * This iterator type allows some modifications while iterating.
  *
- * Used as state in dlist_foreach_modify(). To get the current element of the
- * iteration use the 'cur' member.
+ * Used as state in dlist_foreach_modify() and dclist_foreach_modify().
+ *
+ * To get the current element of the iteration use the 'cur' member.
  *
  * Iterations using this are only allowed to change the list at the current
  * point of iteration. It is fine to delete the current node, but it is *not*
@@ -182,6 +202,19 @@ typedef struct dlist_mutable_iter
 	dlist_node *end;			/* last node we'll iterate to */
 } dlist_mutable_iter;
 
+/*
+ * Head of a doubly linked list with a count of the number of items
+ *
+ * This internally makes use of a dlist to implement the actual list and
+ * wrapper functions are used to maintain the count when items are added and
+ * removed from the list.
+ */
+typedef struct dclist_head
+{
+	dlist_head	dlist;			/* the actual list header */
+	uint32		count;			/* the number of items in the list */
+} dclist_head;
+
 /*
  * Node of a singly linked list.
  *
@@ -246,6 +279,7 @@ typedef struct slist_mutable_iter
 
 /* Static initializers */
 #define DLIST_STATIC_INIT(name) {{&(name).head, &(name).head}}
+#define DCLIST_STATIC_INIT(name) {{{&(name).dlist.head, &(name).dlist.head}}, 0}
 #define SLIST_STATIC_INIT(name) {{NULL}}
 
 
@@ -255,6 +289,7 @@ typedef struct slist_mutable_iter
 extern void slist_delete(slist_head *head, slist_node *node);
 
 #ifdef ILIST_DEBUG
+extern void dlist_member_check(dlist_head *head, dlist_node *node);
 extern void dlist_check(dlist_head *head);
 extern void slist_check(slist_head *head);
 #else
@@ -264,6 +299,7 @@ extern void slist_check(slist_head *head);
  * in which functions the only point of passing the list head pointer is to be
  * able to run these checks.
  */
+#define dlist_member_check(head, node) ((void) (head))
 #define dlist_check(head)	((void) (head))
 #define slist_check(head)	((void) (head))
 #endif							/* ILIST_DEBUG */
@@ -361,6 +397,17 @@ dlist_delete(dlist_node *node)
 	node->next->prev = node->prev;
 }
 
+/*
+ * Same as dlist_delete, but performs checks in ILIST_DEBUG builds to ensure
+ * that 'node' belongs to 'head'.
+ */
+static inline void
+dlist_delete_from(dlist_head *head, dlist_node *node)
+{
+	dlist_member_check(head, node);
+	dlist_delete(node);
+}
+
 /*
  * Remove and return the first node from a list (there must be one).
  */
@@ -562,6 +609,309 @@ dlist_tail_node(dlist_head *head)
 		 (iter).cur != (iter).end;											\
 		 (iter).cur = (iter).cur->prev)
 
+/* doubly-linked count list implementation */
+
+/*
+ * dclist_init
+ *		Initialize a doubly linked count list.
+ *
+ * Previous state will be thrown away without any cleanup.
+ */
+static inline void
+dclist_init(dclist_head *head)
+{
+	dlist_init(&head->dlist);
+	head->count = 0;
+}
+
+/*
+ * dclist_is_empty
+ *		Returns true if the list is empty, otherwise false.
+ */
+static inline bool
+dclist_is_empty(dclist_head *head)
+{
+	Assert(dlist_is_empty(&head->dlist) == (head->count == 0));
+	return (head->count == 0);
+}
+
+/*
+ * dclist_push_head
+ *		Insert a node at the beginning of the list.
+ */
+static inline void
+dclist_push_head(dclist_head *head, dlist_node *node)
+{
+	if (head->dlist.head.next == NULL)	/* convert NULL header to circular */
+		dclist_init(head);
+
+	dlist_push_head(&head->dlist, node);
+	head->count++;
+
+	Assert(head->count > 0);	/* count overflow check */
+}
+
+/*
+ * dclist_push_tail
+ *		Insert a node at the end of the list.
+ */
+static inline void
+dclist_push_tail(dclist_head *head, dlist_node *node)
+{
+	if (head->dlist.head.next == NULL)	/* convert NULL header to circular */
+		dclist_init(head);
+
+	dlist_push_tail(&head->dlist, node);
+	head->count++;
+
+	Assert(head->count > 0);	/* count overflow check */
+}
+
+/*
+ * dclist_insert_after
+ *		Insert a node after another *in the same list*
+ *
+ * Caution: 'after' must be a member of 'head'.
+ */
+static inline void
+dclist_insert_after(dclist_head *head, dlist_node *after, dlist_node *node)
+{
+	dlist_member_check(&head->dlist, after);
+	Assert(head->count > 0);	/* must be at least 1 already */
+
+	dlist_insert_after(after, node);
+	head->count++;
+
+	Assert(head->count > 0);	/* count overflow check */
+}
+
+/*
+ * dclist_insert_before
+ *		Insert a node before another *in the same list*
+ *
+ * Caution: 'before' must be a member of 'head'.
+ */
+static inline void
+dclist_insert_before(dclist_head *head, dlist_node *before, dlist_node *node)
+{
+	dlist_member_check(&head->dlist, before);
+	Assert(head->count > 0);	/* must be at least 1 already */
+
+	dlist_insert_before(before, node);
+	head->count++;
+
+	Assert(head->count > 0);	/* count overflow check */
+}
+
+/*
+ * dclist_delete_from
+ *		Deletes 'node' from 'head'.
+ *
+ * Caution: 'node' must be a member of 'head'.
+ */
+static inline void
+dclist_delete_from(dclist_head *head, dlist_node *node)
+{
+	Assert(head->count > 0);
+
+	dlist_delete_from(&head->dlist, node);
+	head->count--;
+}
+
+/*
+ * dclist_pop_head_node
+ *		Remove and return the first node from a list (there must be one).
+ */
+static inline dlist_node *
+dclist_pop_head_node(dclist_head *head)
+{
+	dlist_node *node;
+
+	Assert(head->count > 0);
+
+	node = dlist_pop_head_node(&head->dlist);
+	head->count--;
+	return node;
+}
+
+/*
+ * dclist_move_head
+ *		Move 'node' from its current position in the list to the head position
+ *		in 'head'.
+ *
+ * Caution: 'node' must be a member of 'head'.
+ */
+static inline void
+dclist_move_head(dclist_head *head, dlist_node *node)
+{
+	dlist_member_check(&head->dlist, node);
+	Assert(head->count > 0);
+
+	dlist_move_head(&head->dlist, node);
+}
+
+/*
+ * dclist_move_tail
+ *		Move 'node' from its current position in the list to the tail position
+ *		in 'head'.
+ *
+ * Caution: 'node' must be a member of 'head'.
+ */
+static inline void
+dclist_move_tail(dclist_head *head, dlist_node *node)
+{
+	dlist_member_check(&head->dlist, node);
+	Assert(head->count > 0);
+
+	dlist_move_tail(&head->dlist, node);
+}
+
+/*
+ * dclist_has_next
+ *		Check whether 'node' has a following node.
+ *
+ * Caution: 'node' must be a member of 'head'.
+ */
+static inline bool
+dclist_has_next(dclist_head *head, dlist_node *node)
+{
+	dlist_member_check(&head->dlist, node);
+	Assert(head->count > 0);
+
+	return dlist_has_next(&head->dlist, node);
+}
+
+/*
+ * dclist_has_prev
+ *		Check whether 'node' has a preceding node.
+ *
+ * Caution: 'node' must be a member of 'head'.
+ */
+static inline bool
+dclist_has_prev(dclist_head *head, dlist_node *node)
+{
+	dlist_member_check(&head->dlist, node);
+	Assert(head->count > 0);
+
+	return dlist_has_prev(&head->dlist, node);
+}
+
+/*
+ * dclist_next_node
+ *		Return the next node in the list (there must be one).
+ */
+static inline dlist_node *
+dclist_next_node(dclist_head *head, dlist_node *node)
+{
+	Assert(head->count > 0);
+
+	return dlist_next_node(&head->dlist, node);
+}
+
+/*
+ * dclist_prev_node
+ *		Return the prev node in the list (there must be one).
+ */
+static inline dlist_node *
+dclist_prev_node(dclist_head *head, dlist_node *node)
+{
+	Assert(head->count > 0);
+
+	return dlist_prev_node(&head->dlist, node);
+}
+
+/* internal support function to get address of head element's struct */
+static inline void *
+dclist_head_element_off(dclist_head *head, size_t off)
+{
+	Assert(!dclist_is_empty(head));
+
+	return (char *) head->dlist.head.next - off;
+}
+
+/*
+ * dclist_head_node
+ *		Return the first node in the list (there must be one).
+ */
+static inline dlist_node *
+dclist_head_node(dclist_head *head)
+{
+	Assert(head->count > 0);
+
+	return (dlist_node *) dlist_head_element_off(&head->dlist, 0);
+}
+
+/* internal support function to get address of tail element's struct */
+static inline void *
+dclist_tail_element_off(dclist_head *head, size_t off)
+{
+	Assert(!dclist_is_empty(head));
+
+	return (char *) head->dlist.head.prev - off;
+}
+
+/*
+ * Return the last node in the list (there must be one).
+ */
+static inline dlist_node *
+dclist_tail_node(dclist_head *head)
+{
+	Assert(head->count > 0);
+
+	return (dlist_node *) dlist_tail_element_off(&head->dlist, 0);
+}
+
+/*
+ * dclist_count
+ *		Returns the stored number of entries in 'head'
+ */
+static inline uint32
+dclist_count(dclist_head *head)
+{
+	Assert(dlist_is_empty(&head->dlist) == (head->count == 0));
+
+	return head->count;
+}
+
+/*
+ * Return the containing struct of 'type' where 'membername' is the dlist_node
+ * pointed at by 'ptr'.
+ *
+ * This is used to convert a dlist_node * back to its containing struct.
+ *
+ * Note: This is effectively just the same as dlist_container, so reuse that.
+ */
+#define dclist_container(type, membername, ptr) \
+		dlist_container(type, membername, ptr)
+
+ /*
+  * Return the address of the first element in the list.
+  *
+  * The list must not be empty.
+  */
+#define dclist_head_element(type, membername, lhead)							\
+	(AssertVariableIsOfTypeMacro(((type *) NULL)->membername, dlist_node),	\
+	 (type *) dclist_head_element_off(lhead, offsetof(type, membername)))
+
+ /*
+  * Return the address of the last element in the list.
+  *
+  * The list must not be empty.
+  */
+#define dclist_tail_element(type, membername, lhead)							\
+	(AssertVariableIsOfTypeMacro(((type *) NULL)->membername, dlist_node),	\
+	 ((type *) dclist_tail_element_off(lhead, offsetof(type, membername))))
+
+
+/* Iterators for dclists */
+#define dclist_foreach(iter, lhead) \
+	dlist_foreach(iter, &((lhead)->dlist))
+
+#define dclist_foreach_modify(iter, lhead) \
+	dlist_foreach_modify(iter, &((lhead)->dlist))
+
+#define dclist_reverse_foreach(iter, lhead) \
+	dlist_reverse_foreach(iter, &((lhead)->dlist))
 
 /* singly linked list implementation */
 
diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h
index 02b59a1931..b23d8cc4f9 100644
--- a/src/include/replication/reorderbuffer.h
+++ b/src/include/replication/reorderbuffer.h
@@ -534,8 +534,7 @@ struct ReorderBuffer
 	/*
 	 * Transactions and subtransactions that have modified system catalogs.
 	 */
-	dlist_head	catchange_txns;
-	int			catchange_ntxns;
+	dclist_head catchange_txns;
 
 	/*
 	 * one-entry sized cache for by_txn. Very frequently the same txn gets
diff --git a/src/include/utils/pgstat_internal.h b/src/include/utils/pgstat_internal.h
index c869533b28..e2c7b59324 100644
--- a/src/include/utils/pgstat_internal.h
+++ b/src/include/utils/pgstat_internal.h
@@ -162,8 +162,7 @@ typedef struct PgStat_SubXactStatus
 	 * if the transaction commits/aborts. To handle replicas and crashes,
 	 * stats drops are included in commit / abort records.
 	 */
-	dlist_head	pending_drops;
-	int			pending_drops_count;
+	dclist_head pending_drops;
 
 	/*
 	 * Tuple insertion/deletion counts for an open transaction can't be
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 2f02cc8f42..c15f990cbb 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -3194,6 +3194,7 @@ datapagemap_t
 dateKEY
 datetkn
 dce_uuid_t
+dclist_head
 decimal
 deparse_columns
 deparse_context

From 2aab14a052de2d09ee7171cfffe94aa88175c72d Mon Sep 17 00:00:00 2001
From: Aleksander Alekseev <aleksander@timescale.com>
Date: Wed, 2 Nov 2022 11:26:23 +0300
Subject: [PATCH v1] Constify the arguments of ilist.c/h functions

Const qualifiers ensure that we don't do something stupid in the function
implementation. Additionally they clarify the interface. As an example:

void
slist_delete(slist_head *head, const slist_node *node)

Here one can instantly tell that node->next is not going to be set to NULL.
Finally, const qualifiers potentially allow the compiler to do more
optimizations. This being said no benchmarking was done for this patch.

Author: Aleksander Alekseev
Reviewed-by: Bharath Rupireddy, David Rowley
Discussion: https://postgr.es/m/CAApHDvrtVxr+FXEX0VbViCFKDGxA3tWDgw9oFewNXCJMmwLjLg@mail.gmail.com
---
 src/backend/lib/ilist.c |  8 +++---
 src/include/lib/ilist.h | 56 ++++++++++++++++++++---------------------
 2 files changed, 32 insertions(+), 32 deletions(-)

diff --git a/src/backend/lib/ilist.c b/src/backend/lib/ilist.c
index e8ea981176..a1f236a9e1 100644
--- a/src/backend/lib/ilist.c
+++ b/src/backend/lib/ilist.c
@@ -28,7 +28,7 @@
  * Caution: this is O(n); consider using slist_delete_current() instead.
  */
 void
-slist_delete(slist_head *head, slist_node *node)
+slist_delete(slist_head *head, const slist_node *node)
 {
 	slist_node *last = &head->head;
 	slist_node *cur;
@@ -57,7 +57,7 @@ slist_delete(slist_head *head, slist_node *node)
  *		Validate that 'node' is a member of 'head'
  */
 void
-dlist_member_check(dlist_head *head, dlist_node *node)
+dlist_member_check(const dlist_head *head, const dlist_node *node)
 {
 	dlist_iter	iter;
 
@@ -73,7 +73,7 @@ dlist_member_check(dlist_head *head, dlist_node *node)
  * Verify integrity of a doubly linked list
  */
 void
-dlist_check(dlist_head *head)
+dlist_check(const dlist_head *head)
 {
 	dlist_node *cur;
 
@@ -110,7 +110,7 @@ dlist_check(dlist_head *head)
  * Verify integrity of a singly linked list
  */
 void
-slist_check(slist_head *head)
+slist_check(const slist_head *head)
 {
 	slist_node *cur;
 
diff --git a/src/include/lib/ilist.h b/src/include/lib/ilist.h
index 3c543e7c36..6c2b05721b 100644
--- a/src/include/lib/ilist.h
+++ b/src/include/lib/ilist.h
@@ -286,12 +286,12 @@ typedef struct slist_mutable_iter
 /* Prototypes for functions too big to be inline */
 
 /* Caution: this is O(n); consider using slist_delete_current() instead */
-extern void slist_delete(slist_head *head, slist_node *node);
+extern void slist_delete(slist_head *head, const slist_node *node);
 
 #ifdef ILIST_DEBUG
-extern void dlist_member_check(dlist_head *head, dlist_node *node);
-extern void dlist_check(dlist_head *head);
-extern void slist_check(slist_head *head);
+extern void dlist_member_check(const dlist_head *head, const dlist_node *node);
+extern void dlist_check(const dlist_head *head);
+extern void slist_check(const slist_head *head);
 #else
 /*
  * These seemingly useless casts to void are here to keep the compiler quiet
@@ -322,7 +322,7 @@ dlist_init(dlist_head *head)
  * An empty list has either its first 'next' pointer set to NULL, or to itself.
  */
 static inline bool
-dlist_is_empty(dlist_head *head)
+dlist_is_empty(const dlist_head *head)
 {
 	dlist_check(head);
 
@@ -465,7 +465,7 @@ dlist_move_tail(dlist_head *head, dlist_node *node)
  * Caution: unreliable if 'node' is not in the list.
  */
 static inline bool
-dlist_has_next(dlist_head *head, dlist_node *node)
+dlist_has_next(const dlist_head *head, const dlist_node *node)
 {
 	return node->next != &head->head;
 }
@@ -475,7 +475,7 @@ dlist_has_next(dlist_head *head, dlist_node *node)
  * Caution: unreliable if 'node' is not in the list.
  */
 static inline bool
-dlist_has_prev(dlist_head *head, dlist_node *node)
+dlist_has_prev(const dlist_head *head, const dlist_node *node)
 {
 	return node->prev != &head->head;
 }
@@ -484,7 +484,7 @@ dlist_has_prev(dlist_head *head, dlist_node *node)
  * Return the next node in the list (there must be one).
  */
 static inline dlist_node *
-dlist_next_node(dlist_head *head, dlist_node *node)
+dlist_next_node(const dlist_head *head, const dlist_node *node)
 {
 	Assert(dlist_has_next(head, node));
 	return node->next;
@@ -494,7 +494,7 @@ dlist_next_node(dlist_head *head, dlist_node *node)
  * Return previous node in the list (there must be one).
  */
 static inline dlist_node *
-dlist_prev_node(dlist_head *head, dlist_node *node)
+dlist_prev_node(const dlist_head *head, const dlist_node *node)
 {
 	Assert(dlist_has_prev(head, node));
 	return node->prev;
@@ -502,7 +502,7 @@ dlist_prev_node(dlist_head *head, dlist_node *node)
 
 /* internal support function to get address of head element's struct */
 static inline void *
-dlist_head_element_off(dlist_head *head, size_t off)
+dlist_head_element_off(const dlist_head *head, size_t off)
 {
 	Assert(!dlist_is_empty(head));
 	return (char *) head->head.next - off;
@@ -512,14 +512,14 @@ dlist_head_element_off(dlist_head *head, size_t off)
  * Return the first node in the list (there must be one).
  */
 static inline dlist_node *
-dlist_head_node(dlist_head *head)
+dlist_head_node(const dlist_head *head)
 {
 	return (dlist_node *) dlist_head_element_off(head, 0);
 }
 
 /* internal support function to get address of tail element's struct */
 static inline void *
-dlist_tail_element_off(dlist_head *head, size_t off)
+dlist_tail_element_off(const dlist_head *head, size_t off)
 {
 	Assert(!dlist_is_empty(head));
 	return (char *) head->head.prev - off;
@@ -529,7 +529,7 @@ dlist_tail_element_off(dlist_head *head, size_t off)
  * Return the last node in the list (there must be one).
  */
 static inline dlist_node *
-dlist_tail_node(dlist_head *head)
+dlist_tail_node(const dlist_head *head)
 {
 	return (dlist_node *) dlist_tail_element_off(head, 0);
 }
@@ -629,7 +629,7 @@ dclist_init(dclist_head *head)
  *		Returns true if the list is empty, otherwise false.
  */
 static inline bool
-dclist_is_empty(dclist_head *head)
+dclist_is_empty(const dclist_head *head)
 {
 	Assert(dlist_is_empty(&head->dlist) == (head->count == 0));
 	return (head->count == 0);
@@ -773,7 +773,7 @@ dclist_move_tail(dclist_head *head, dlist_node *node)
  * Caution: 'node' must be a member of 'head'.
  */
 static inline bool
-dclist_has_next(dclist_head *head, dlist_node *node)
+dclist_has_next(const dclist_head *head, const dlist_node *node)
 {
 	dlist_member_check(&head->dlist, node);
 	Assert(head->count > 0);
@@ -788,7 +788,7 @@ dclist_has_next(dclist_head *head, dlist_node *node)
  * Caution: 'node' must be a member of 'head'.
  */
 static inline bool
-dclist_has_prev(dclist_head *head, dlist_node *node)
+dclist_has_prev(const dclist_head *head, const dlist_node *node)
 {
 	dlist_member_check(&head->dlist, node);
 	Assert(head->count > 0);
@@ -801,7 +801,7 @@ dclist_has_prev(dclist_head *head, dlist_node *node)
  *		Return the next node in the list (there must be one).
  */
 static inline dlist_node *
-dclist_next_node(dclist_head *head, dlist_node *node)
+dclist_next_node(const dclist_head *head, const dlist_node *node)
 {
 	Assert(head->count > 0);
 
@@ -813,7 +813,7 @@ dclist_next_node(dclist_head *head, dlist_node *node)
  *		Return the prev node in the list (there must be one).
  */
 static inline dlist_node *
-dclist_prev_node(dclist_head *head, dlist_node *node)
+dclist_prev_node(const dclist_head *head, const dlist_node *node)
 {
 	Assert(head->count > 0);
 
@@ -822,7 +822,7 @@ dclist_prev_node(dclist_head *head, dlist_node *node)
 
 /* internal support function to get address of head element's struct */
 static inline void *
-dclist_head_element_off(dclist_head *head, size_t off)
+dclist_head_element_off(const dclist_head *head, size_t off)
 {
 	Assert(!dclist_is_empty(head));
 
@@ -834,7 +834,7 @@ dclist_head_element_off(dclist_head *head, size_t off)
  *		Return the first node in the list (there must be one).
  */
 static inline dlist_node *
-dclist_head_node(dclist_head *head)
+dclist_head_node(const dclist_head *head)
 {
 	Assert(head->count > 0);
 
@@ -843,7 +843,7 @@ dclist_head_node(dclist_head *head)
 
 /* internal support function to get address of tail element's struct */
 static inline void *
-dclist_tail_element_off(dclist_head *head, size_t off)
+dclist_tail_element_off(const dclist_head *head, size_t off)
 {
 	Assert(!dclist_is_empty(head));
 
@@ -854,7 +854,7 @@ dclist_tail_element_off(dclist_head *head, size_t off)
  * Return the last node in the list (there must be one).
  */
 static inline dlist_node *
-dclist_tail_node(dclist_head *head)
+dclist_tail_node(const dclist_head *head)
 {
 	Assert(head->count > 0);
 
@@ -866,7 +866,7 @@ dclist_tail_node(dclist_head *head)
  *		Returns the stored number of entries in 'head'
  */
 static inline uint32
-dclist_count(dclist_head *head)
+dclist_count(const dclist_head *head)
 {
 	Assert(dlist_is_empty(&head->dlist) == (head->count == 0));
 
@@ -929,7 +929,7 @@ slist_init(slist_head *head)
  * Is the list empty?
  */
 static inline bool
-slist_is_empty(slist_head *head)
+slist_is_empty(const slist_head *head)
 {
 	slist_check(head);
 
@@ -977,7 +977,7 @@ slist_pop_head_node(slist_head *head)
  * Check whether 'node' has a following node.
  */
 static inline bool
-slist_has_next(slist_head *head, slist_node *node)
+slist_has_next(const slist_head *head, const slist_node *node)
 {
 	slist_check(head);
 
@@ -988,7 +988,7 @@ slist_has_next(slist_head *head, slist_node *node)
  * Return the next node in the list (there must be one).
  */
 static inline slist_node *
-slist_next_node(slist_head *head, slist_node *node)
+slist_next_node(const slist_head *head, const slist_node *node)
 {
 	Assert(slist_has_next(head, node));
 	return node->next;
@@ -996,7 +996,7 @@ slist_next_node(slist_head *head, slist_node *node)
 
 /* internal support function to get address of head element's struct */
 static inline void *
-slist_head_element_off(slist_head *head, size_t off)
+slist_head_element_off(const slist_head *head, size_t off)
 {
 	Assert(!slist_is_empty(head));
 	return (char *) head->head.next - off;
@@ -1006,7 +1006,7 @@ slist_head_element_off(slist_head *head, size_t off)
  * Return the first node in the list (there must be one).
  */
 static inline slist_node *
-slist_head_node(slist_head *head)
+slist_head_node(const slist_head *head)
 {
 	return (slist_node *) slist_head_element_off(head, 0);
 }
-- 
2.38.0