Allow dynamic allocation of shared memory segments.

Patch by myself and Amit Kapila.  Design help from Noah Misch.  Review
by Andres Freund.

src/backend/utils/resowner/resowner.c

@@ -98,6 +98,11 @@ typedef struct ResourceOwnerData
 	int			nfiles;			/* number of owned temporary files */
 	File	   *files;			/* dynamically allocated array */
 	int			maxfiles;		/* currently allocated array size */
+
+	/* We have built-in support for remembering dynamic shmem segments */
+	int			ndsms;			/* number of owned shmem segments */
+	dsm_segment **dsms;			/* dynamically allocated array */
+	int			maxdsms;		/* currently allocated array size */
 }	ResourceOwnerData;
 
 
@@ -132,6 +137,7 @@ static void PrintPlanCacheLeakWarning(CachedPlan *plan);
 static void PrintTupleDescLeakWarning(TupleDesc tupdesc);
 static void PrintSnapshotLeakWarning(Snapshot snapshot);
 static void PrintFileLeakWarning(File file);
+static void PrintDSMLeakWarning(dsm_segment *seg);
 
 
 /*****************************************************************************
@@ -271,6 +277,21 @@ ResourceOwnerReleaseInternal(ResourceOwner owner,
 				PrintRelCacheLeakWarning(owner->relrefs[owner->nrelrefs - 1]);
 			RelationClose(owner->relrefs[owner->nrelrefs - 1]);
 		}
+
+		/*
+		 * Release dynamic shared memory segments.  Note that dsm_detach()
+		 * will remove the segment from my list, so I just have to iterate
+		 * until there are none.
+		 *
+		 * As in the preceding cases, warn if there are leftover at commit
+		 * time.
+		 */
+		while (owner->ndsms > 0)
+		{
+			if (isCommit)
+				PrintDSMLeakWarning(owner->dsms[owner->ndsms - 1]);
+			dsm_detach(owner->dsms[owner->ndsms - 1]);
+		}
 	}
 	else if (phase == RESOURCE_RELEASE_LOCKS)
 	{
@@ -402,6 +423,7 @@ ResourceOwnerDelete(ResourceOwner owner)
 	Assert(owner->ncatrefs == 0);
 	Assert(owner->ncatlistrefs == 0);
 	Assert(owner->nrelrefs == 0);
+	Assert(owner->ndsms == 0);
 	Assert(owner->nplanrefs == 0);
 	Assert(owner->ntupdescs == 0);
 	Assert(owner->nsnapshots == 0);
@@ -438,6 +460,8 @@ ResourceOwnerDelete(ResourceOwner owner)
 		pfree(owner->snapshots);
 	if (owner->files)
 		pfree(owner->files);
+	if (owner->dsms)
+		pfree(owner->dsms);
 
 	pfree(owner);
 }
@@ -1230,3 +1254,88 @@ PrintFileLeakWarning(File file)
 		 "temporary file leak: File %d still referenced",
 		 file);
 }
+
+/*
+ * Make sure there is room for at least one more entry in a ResourceOwner's
+ * dynamic shmem segment reference array.
+ *
+ * This is separate from actually inserting an entry because if we run out
+ * of memory, it's critical to do so *before* acquiring the resource.
+ */
+void
+ResourceOwnerEnlargeDSMs(ResourceOwner owner)
+{
+	int			newmax;
+
+	if (owner->ndsms < owner->maxdsms)
+		return;					/* nothing to do */
+
+	if (owner->dsms == NULL)
+	{
+		newmax = 16;
+		owner->dsms = (dsm_segment **)
+			MemoryContextAlloc(TopMemoryContext,
+							   newmax * sizeof(dsm_segment *));
+		owner->maxdsms = newmax;
+	}
+	else
+	{
+		newmax = owner->maxdsms * 2;
+		owner->dsms = (dsm_segment **)
+			repalloc(owner->dsms, newmax * sizeof(dsm_segment *));
+		owner->maxdsms = newmax;
+	}
+}
+
+/*
+ * Remember that a dynamic shmem segment is owned by a ResourceOwner
+ *
+ * Caller must have previously done ResourceOwnerEnlargeDSMs()
+ */
+void
+ResourceOwnerRememberDSM(ResourceOwner owner, dsm_segment *seg)
+{
+	Assert(owner->ndsms < owner->maxdsms);
+	owner->dsms[owner->ndsms] = seg;
+	owner->ndsms++;
+}
+
+/*
+ * Forget that a temporary file is owned by a ResourceOwner
+ */
+void
+ResourceOwnerForgetDSM(ResourceOwner owner, dsm_segment *seg)
+{
+	dsm_segment **dsms = owner->dsms;
+	int			ns1 = owner->ndsms - 1;
+	int			i;
+
+	for (i = ns1; i >= 0; i--)
+	{
+		if (dsms[i] == seg)
+		{
+			while (i < ns1)
+			{
+				dsms[i] = dsms[i + 1];
+				i++;
+			}
+			owner->ndsms = ns1;
+			return;
+		}
+	}
+	elog(ERROR,
+		 "dynamic shared memory segment %u is not owned by resource owner %s",
+		 dsm_segment_handle(seg), owner->name);
+}
+
+
+/*
+ * Debugging subroutine
+ */
+static void
+PrintDSMLeakWarning(dsm_segment *seg)
+{
+	elog(WARNING,
+		 "dynamic shared memory leak: segment %u still referenced",
+		 dsm_segment_handle(seg));
+}