1 files changed, 429 insertions, 1 deletions

diff --git a/src/backend/postmaster/autovacuum.c b/src/backend/postmaster/autovacuum.c
index 33ca749b529..89dd3b321bc 100644
--- a/src/backend/postmaster/autovacuum.c
+++ b/src/backend/postmaster/autovacuum.c
@@ -92,7 +92,9 @@
 #include "storage/procsignal.h"
 #include "storage/sinvaladt.h"
 #include "tcop/tcopprot.h"
+#include "utils/dsa.h"
 #include "utils/fmgroids.h"
+#include "utils/fmgrprotos.h"
 #include "utils/lsyscache.h"
 #include "utils/memutils.h"
 #include "utils/ps_status.h"
@@ -252,9 +254,10 @@ typedef enum
  * av_runningWorkers the WorkerInfo non-free queue
  * av_startingWorker pointer to WorkerInfo currently being started (cleared by
  *					the worker itself as soon as it's up and running)
+ * av_dsa_handle	handle for allocatable shared memory
  *
  * This struct is protected by AutovacuumLock, except for av_signal and parts
- * of the worker list (see above).
+ * of the worker list (see above).  av_dsa_handle is readable unlocked.
  *-------------
  */
 typedef struct
@@ -264,6 +267,8 @@ typedef struct
 	dlist_head	av_freeWorkers;
 	dlist_head	av_runningWorkers;
 	WorkerInfo	av_startingWorker;
+	dsa_handle	av_dsa_handle;
+	dsa_pointer av_workitems;
 } AutoVacuumShmemStruct;
 
 static AutoVacuumShmemStruct *AutoVacuumShmem;
@@ -278,6 +283,32 @@ static MemoryContext DatabaseListCxt = NULL;
 /* Pointer to my own WorkerInfo, valid on each worker */
 static WorkerInfo MyWorkerInfo = NULL;
 
+/*
+ * Autovacuum workitem array, stored in AutoVacuumShmem->av_workitems.  This
+ * list is mostly protected by AutovacuumLock, except that if it's marked
+ * 'active' other processes must not modify the work-identifying members,
+ * though changing the list pointers is okay.
+ */
+typedef struct AutoVacuumWorkItem
+{
+	AutoVacuumWorkItemType avw_type;
+	Oid			avw_database;
+	Oid			avw_relation;
+	BlockNumber avw_blockNumber;
+	bool		avw_active;
+	dsa_pointer avw_next;		/* doubly linked list pointers */
+	dsa_pointer avw_prev;
+} AutoVacuumWorkItem;
+
+#define NUM_WORKITEMS	256
+typedef struct
+{
+	dsa_pointer avs_usedItems;
+	dsa_pointer avs_freeItems;
+} AutovacWorkItems;
+
+static dsa_area *AutoVacuumDSA = NULL;
+
 /* PID of launcher, valid only in worker while shutting down */
 int			AutovacuumLauncherPid = 0;
 
@@ -316,11 +347,16 @@ static AutoVacOpts *extract_autovac_opts(HeapTuple tup,
 static PgStat_StatTabEntry *get_pgstat_tabentry_relid(Oid relid, bool isshared,
 						  PgStat_StatDBEntry *shared,
 						  PgStat_StatDBEntry *dbentry);
+static void perform_work_item(AutoVacuumWorkItem *workitem);
 static void autovac_report_activity(autovac_table *tab);
+static void autovac_report_workitem(AutoVacuumWorkItem *workitem,
+						const char *nspname, const char *relname);
 static void av_sighup_handler(SIGNAL_ARGS);
 static void avl_sigusr2_handler(SIGNAL_ARGS);
 static void avl_sigterm_handler(SIGNAL_ARGS);
 static void autovac_refresh_stats(void);
+static void remove_wi_from_list(dsa_pointer *list, dsa_pointer wi_ptr);
+static void add_wi_to_list(dsa_pointer *list, dsa_pointer wi_ptr);
 
 
 
@@ -574,6 +610,28 @@ AutoVacLauncherMain(int argc, char *argv[])
 	 */
 	rebuild_database_list(InvalidOid);
 
+	/*
+	 * Set up our DSA so that backends can install work-item requests.  It may
+	 * already exist as created by a previous launcher.
+	 */
+	if (!AutoVacuumShmem->av_dsa_handle)
+	{
+		LWLockAcquire(AutovacuumLock, LW_EXCLUSIVE);
+		AutoVacuumDSA = dsa_create(AutovacuumLock->tranche);
+		/* make sure it doesn't go away even if we do */
+		dsa_pin(AutoVacuumDSA);
+		dsa_pin_mapping(AutoVacuumDSA);
+		AutoVacuumShmem->av_dsa_handle = dsa_get_handle(AutoVacuumDSA);
+		/* delay array allocation until first request */
+		AutoVacuumShmem->av_workitems = InvalidDsaPointer;
+		LWLockRelease(AutovacuumLock);
+	}
+	else
+	{
+		AutoVacuumDSA = dsa_attach(AutoVacuumShmem->av_dsa_handle);
+		dsa_pin_mapping(AutoVacuumDSA);
+	}
+
 	/* loop until shutdown request */
 	while (!got_SIGTERM)
 	{
@@ -1617,6 +1675,14 @@ AutoVacWorkerMain(int argc, char *argv[])
 	{
 		char		dbname[NAMEDATALEN];
 
+		if (AutoVacuumShmem->av_dsa_handle)
+		{
+			/* First use of DSA in this worker, so attach to it */
+			Assert(!AutoVacuumDSA);
+			AutoVacuumDSA = dsa_attach(AutoVacuumShmem->av_dsa_handle);
+			dsa_pin_mapping(AutoVacuumDSA);
+		}
+
 		/*
 		 * Report autovac startup to the stats collector.  We deliberately do
 		 * this before InitPostgres, so that the last_autovac_time will get
@@ -2467,6 +2533,69 @@ deleted:
 	}
 
 	/*
+	 * Perform additional work items, as requested by backends.
+	 */
+	if (AutoVacuumShmem->av_workitems)
+	{
+		dsa_pointer wi_ptr;
+		AutovacWorkItems *workitems;
+
+		LWLockAcquire(AutovacuumLock, LW_EXCLUSIVE);
+
+		/*
+		 * Scan the list of pending items, and process the inactive ones in
+		 * our database.
+		 */
+		workitems = (AutovacWorkItems *)
+			dsa_get_address(AutoVacuumDSA, AutoVacuumShmem->av_workitems);
+		wi_ptr = workitems->avs_usedItems;
+
+		while (wi_ptr != InvalidDsaPointer)
+		{
+			AutoVacuumWorkItem *workitem;
+
+			workitem = (AutoVacuumWorkItem *)
+				dsa_get_address(AutoVacuumDSA, wi_ptr);
+
+			if (workitem->avw_database == MyDatabaseId && !workitem->avw_active)
+			{
+				dsa_pointer next_ptr;
+
+				/* claim this one */
+				workitem->avw_active = true;
+
+				LWLockRelease(AutovacuumLock);
+
+				perform_work_item(workitem);
+
+				/*
+				 * Check for config changes before acquiring lock for further
+				 * jobs.
+				 */
+				CHECK_FOR_INTERRUPTS();
+				if (got_SIGHUP)
+				{
+					got_SIGHUP = false;
+					ProcessConfigFile(PGC_SIGHUP);
+				}
+
+				LWLockAcquire(AutovacuumLock, LW_EXCLUSIVE);
+
+				/* Put the array item back for the next user */
+				next_ptr = workitem->avw_next;
+				remove_wi_from_list(&workitems->avs_usedItems, wi_ptr);
+				add_wi_to_list(&workitems->avs_freeItems, wi_ptr);
+				wi_ptr = next_ptr;
+			}
+			else
+				wi_ptr = workitem->avw_next;
+		}
+
+		/* all done */
+		LWLockRelease(AutovacuumLock);
+	}
+
+	/*
 	 * We leak table_toast_map here (among other things), but since we're
 	 * going away soon, it's not a problem.
 	 */
@@ -2499,6 +2628,103 @@ deleted:
 }
 
 /*
+ * Execute a previously registered work item.
+ */
+static void
+perform_work_item(AutoVacuumWorkItem *workitem)
+{
+	char	   *cur_datname = NULL;
+	char	   *cur_nspname = NULL;
+	char	   *cur_relname = NULL;
+
+	/*
+	 * Note we do not store table info in MyWorkerInfo, since this is not
+	 * vacuuming proper.
+	 */
+
+	/*
+	 * Save the relation name for a possible error message, to avoid a catalog
+	 * lookup in case of an error.  If any of these return NULL, then the
+	 * relation has been dropped since last we checked; skip it. Note: they
+	 * must live in a long-lived memory context because we call vacuum and
+	 * analyze in different transactions.
+	 */
+
+	cur_relname = get_rel_name(workitem->avw_relation);
+	cur_nspname = get_namespace_name(get_rel_namespace(workitem->avw_relation));
+	cur_datname = get_database_name(MyDatabaseId);
+	if (!cur_relname || !cur_nspname || !cur_datname)
+		goto deleted2;
+
+	autovac_report_workitem(workitem, cur_nspname, cur_datname);
+
+	/*
+	 * We will abort the current work item if something errors out, and
+	 * continue with the next one; in particular, this happens if we are
+	 * interrupted with SIGINT.  Note that this means that the work item list
+	 * can be lossy.
+	 */
+	PG_TRY();
+	{
+		/* have at it */
+		MemoryContextSwitchTo(TopTransactionContext);
+
+		switch (workitem->avw_type)
+		{
+			case AVW_BRINSummarizeRange:
+				DirectFunctionCall2(brin_summarize_range,
+									ObjectIdGetDatum(workitem->avw_relation),
+						   Int64GetDatum((int64) workitem->avw_blockNumber));
+				break;
+			default:
+				elog(WARNING, "unrecognized work item found: type %d",
+					 workitem->avw_type);
+				break;
+		}
+
+		/*
+		 * Clear a possible query-cancel signal, to avoid a late reaction to
+		 * an automatically-sent signal because of vacuuming the current table
+		 * (we're done with it, so it would make no sense to cancel at this
+		 * point.)
+		 */
+		QueryCancelPending = false;
+	}
+	PG_CATCH();
+	{
+		/*
+		 * Abort the transaction, start a new one, and proceed with the next
+		 * table in our list.
+		 */
+		HOLD_INTERRUPTS();
+		errcontext("processing work entry for relation \"%s.%s.%s\"",
+				   cur_datname, cur_nspname, cur_relname);
+		EmitErrorReport();
+
+		/* this resets the PGXACT flags too */
+		AbortOutOfAnyTransaction();
+		FlushErrorState();
+		MemoryContextResetAndDeleteChildren(PortalContext);
+
+		/* restart our transaction for the following operations */
+		StartTransactionCommand();
+		RESUME_INTERRUPTS();
+	}
+	PG_END_TRY();
+
+	/* We intentionally do not set did_vacuum here */
+
+	/* be tidy */
+deleted2:
+	if (cur_datname)
+		pfree(cur_datname);
+	if (cur_nspname)
+		pfree(cur_nspname);
+	if (cur_relname)
+		pfree(cur_relname);
+}
+
+/*
  * extract_autovac_opts
  *
  * Given a relation's pg_class tuple, return the AutoVacOpts portion of
@@ -2946,6 +3172,45 @@ autovac_report_activity(autovac_table *tab)
 }
 
 /*
+ * autovac_report_workitem
+ *		Report to pgstat that autovacuum is processing a work item
+ */
+static void
+autovac_report_workitem(AutoVacuumWorkItem *workitem,
+						const char *nspname, const char *relname)
+{
+	char		activity[MAX_AUTOVAC_ACTIV_LEN + 12 + 2];
+	char		blk[12 + 2];
+	int			len;
+
+	switch (workitem->avw_type)
+	{
+		case AVW_BRINSummarizeRange:
+			snprintf(activity, MAX_AUTOVAC_ACTIV_LEN,
+					 "autovacuum: BRIN summarize");
+			break;
+	}
+
+	/*
+	 * Report the qualified name of the relation, and the block number if any
+	 */
+	len = strlen(activity);
+
+	if (BlockNumberIsValid(workitem->avw_blockNumber))
+		snprintf(blk, sizeof(blk), " %u", workitem->avw_blockNumber);
+	else
+		blk[0] = '\0';
+
+	snprintf(activity + len, MAX_AUTOVAC_ACTIV_LEN - len,
+			 " %s.%s%s", nspname, relname, blk);
+
+	/* Set statement_timestamp() to current time for pg_stat_activity */
+	SetCurrentStatementStartTimestamp();
+
+	pgstat_report_activity(STATE_RUNNING, activity);
+}
+
+/*
  * AutoVacuumingActive
  *		Check GUC vars and report whether the autovacuum process should be
  *		running.
@@ -2959,6 +3224,113 @@ AutoVacuumingActive(void)
 }
 
 /*
+ * Request one work item to the next autovacuum run processing our database.
+ */
+void
+AutoVacuumRequestWork(AutoVacuumWorkItemType type, Oid relationId,
+					  BlockNumber blkno)
+{
+	AutovacWorkItems *workitems;
+	dsa_pointer wi_ptr;
+	AutoVacuumWorkItem *workitem;
+
+	LWLockAcquire(AutovacuumLock, LW_EXCLUSIVE);
+
+	/*
+	 * It may be useful to de-duplicate the list upon insertion.  For the only
+	 * currently existing caller, this is not necessary.
+	 */
+
+	/* First use in this process?  Set up DSA */
+	if (!AutoVacuumDSA)
+	{
+		if (!AutoVacuumShmem->av_dsa_handle)
+		{
+			/* autovacuum launcher not started; nothing can be done */
+			LWLockRelease(AutovacuumLock);
+			return;
+		}
+		AutoVacuumDSA = dsa_attach(AutoVacuumShmem->av_dsa_handle);
+		dsa_pin_mapping(AutoVacuumDSA);
+	}
+
+	/* First use overall?  Allocate work items array */
+	if (AutoVacuumShmem->av_workitems == InvalidDsaPointer)
+	{
+		int			i;
+		AutovacWorkItems *workitems;
+
+		AutoVacuumShmem->av_workitems =
+			dsa_allocate_extended(AutoVacuumDSA,
+								  sizeof(AutovacWorkItems) +
+								  NUM_WORKITEMS * sizeof(AutoVacuumWorkItem),
+								  DSA_ALLOC_NO_OOM);
+		/* if out of memory, silently disregard the request */
+		if (AutoVacuumShmem->av_workitems == InvalidDsaPointer)
+		{
+			LWLockRelease(AutovacuumLock);
+			dsa_detach(AutoVacuumDSA);
+			AutoVacuumDSA = NULL;
+			return;
+		}
+
+		/* Initialize each array entry as a member of the free list */
+		workitems = dsa_get_address(AutoVacuumDSA, AutoVacuumShmem->av_workitems);
+
+		workitems->avs_usedItems = InvalidDsaPointer;
+		workitems->avs_freeItems = InvalidDsaPointer;
+		for (i = 0; i < NUM_WORKITEMS; i++)
+		{
+			/* XXX surely there is a simpler way to do this */
+			wi_ptr = AutoVacuumShmem->av_workitems + sizeof(AutovacWorkItems) +
+				sizeof(AutoVacuumWorkItem) * i;
+			workitem = (AutoVacuumWorkItem *) dsa_get_address(AutoVacuumDSA, wi_ptr);
+
+			workitem->avw_type = 0;
+			workitem->avw_database = InvalidOid;
+			workitem->avw_relation = InvalidOid;
+			workitem->avw_active = false;
+
+			/* put this item in the free list */
+			workitem->avw_next = workitems->avs_freeItems;
+			workitems->avs_freeItems = wi_ptr;
+		}
+	}
+
+	workitems = (AutovacWorkItems *)
+		dsa_get_address(AutoVacuumDSA, AutoVacuumShmem->av_workitems);
+
+	/* If array is full, disregard the request */
+	if (workitems->avs_freeItems == InvalidDsaPointer)
+	{
+		LWLockRelease(AutovacuumLock);
+		dsa_detach(AutoVacuumDSA);
+		AutoVacuumDSA = NULL;
+		return;
+	}
+
+	/* remove workitem struct from free list ... */
+	wi_ptr = workitems->avs_freeItems;
+	remove_wi_from_list(&workitems->avs_freeItems, wi_ptr);
+
+	/* ... initialize it ... */
+	workitem = dsa_get_address(AutoVacuumDSA, wi_ptr);
+	workitem->avw_type = type;
+	workitem->avw_database = MyDatabaseId;
+	workitem->avw_relation = relationId;
+	workitem->avw_blockNumber = blkno;
+	workitem->avw_active = false;
+
+	/* ... and put it on autovacuum's to-do list */
+	add_wi_to_list(&workitems->avs_usedItems, wi_ptr);
+
+	LWLockRelease(AutovacuumLock);
+
+	dsa_detach(AutoVacuumDSA);
+	AutoVacuumDSA = NULL;
+}
+
+/*
  * autovac_init
  *		This is called at postmaster initialization.
  *
@@ -3079,3 +3451,59 @@ autovac_refresh_stats(void)
 
 	pgstat_clear_snapshot();
 }
+
+/*
+ * Simplistic open-coded list implementation for objects stored in DSA.
+ * Each item is doubly linked, but we have no tail pointer, and the "prev"
+ * element of the first item is null, not the list.
+ */
+
+/*
+ * Remove a work item from the given list.
+ */
+static void
+remove_wi_from_list(dsa_pointer *list, dsa_pointer wi_ptr)
+{
+	AutoVacuumWorkItem *workitem = dsa_get_address(AutoVacuumDSA, wi_ptr);
+	dsa_pointer next = workitem->avw_next;
+	dsa_pointer prev = workitem->avw_prev;
+
+	workitem->avw_next = workitem->avw_prev = InvalidDsaPointer;
+
+	if (next != InvalidDsaPointer)
+	{
+		workitem = dsa_get_address(AutoVacuumDSA, next);
+		workitem->avw_prev = prev;
+	}
+
+	if (prev != InvalidDsaPointer)
+	{
+		workitem = dsa_get_address(AutoVacuumDSA, prev);
+		workitem->avw_next = next;
+	}
+	else
+		*list = next;
+}
+
+/*
+ * Add a workitem to the given list
+ */
+static void
+add_wi_to_list(dsa_pointer *list, dsa_pointer wi_ptr)
+{
+	if (*list == InvalidDsaPointer)
+	{
+		/* list is empty; item is now singleton */
+		*list = wi_ptr;
+	}
+	else
+	{
+		AutoVacuumWorkItem *workitem = dsa_get_address(AutoVacuumDSA, wi_ptr);
+		AutoVacuumWorkItem *old = dsa_get_address(AutoVacuumDSA, *list);
+
+		/* Put item at head of list */
+		workitem->avw_next = *list;
+		old->avw_prev = wi_ptr;
+		*list = wi_ptr;
+	}
+}