From 878dc4587fdf4774fc78c54ffadd938ed7595030 Mon Sep 17 00:00:00 2001
From: Amit Kapila <amit.kapila@enterprisedb.com>
Date: Tue, 3 Dec 2019 15:36:46 +0530
Subject: [PATCH] Code review.

---
 src/backend/access/heap/vacuumlazy.c  | 441 +++++++++++++++++-----------------
 src/backend/access/transam/parallel.c |   4 +-
 src/backend/commands/vacuum.c         | 115 +++++----
 src/include/access/heapam.h           |   5 +-
 src/include/commands/vacuum.h         |   5 +
 5 files changed, 290 insertions(+), 280 deletions(-)

diff --git a/src/backend/access/heap/vacuumlazy.c b/src/backend/access/heap/vacuumlazy.c
index 36f4ef1..e1cbdb1 100644
--- a/src/backend/access/heap/vacuumlazy.c
+++ b/src/backend/access/heap/vacuumlazy.c
@@ -33,11 +33,12 @@
  * processes exit.  And then the leader process re-initializes the parallel
  * context so that it can use the same DSM for multiple passses of index
  * vacuum and for performing index cleanup.  Note that all parallel workers
- * live during either index vacuuming or index cleanup but the leader process
- * neither exits from the parallel mode nor destroys the parallel context.
- * For updating the index statistics, since any updates are not allowed during
- * parallel mode we update the index statistics after exited from the parallel
- * mode.
+ * are alive only during index vacuum or index cleanup but the leader process
+ * neither exits from the parallel mode nor destroys the parallel context till
+ * the entire parallel operation is finished.  For updating the index
+ * statistics, we need to update the system table and since updates are not
+ * allowed during parallel mode we update the index statistics after exiting
+ * from the parallel mode.
  *
  * Portions Copyright (c) 1996-2019, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
@@ -175,7 +176,7 @@ typedef struct LVDeadTuples
 typedef struct LVShared
 {
 	/*
-	 * Target table relid and log level. These fields are not modified
+	 * Target table relid and log level.  These fields are not modified
 	 * during the lazy vacuum.
 	 */
 	Oid		relid;
@@ -190,13 +191,13 @@ typedef struct LVShared
 	bool	first_time;
 
 	/*
-	 * Fields for both index vacuuming and index cleanup.
+	 * Fields for both index vacuum and cleanup.
 	 *
 	 * reltuples is the total number of input heap tuples.  We set either
-	 * an old live tuples in index vacuuming case or the new live tuples in
-	 * index cleanup case.
+	 * old live tuples in the index vacuum case or the new live tuples in
+	 * the index cleanup case.
 	 *
-	 * estimated_count is true if the reltuples is estimated value.
+	 * estimated_count is true if the reltuples is an estimated value.
 	 */
 	double	reltuples;
 	bool	estimated_count;
@@ -318,27 +319,6 @@ static MultiXactId MultiXactCutoff;
 
 static BufferAccessStrategy vac_strategy;
 
-/*
- * Variables for cost-based vacuum delay for parallel index vacuuming.
- * The basic idea of cost-based vacuum delay for parallel index vacuuming
- * is to allow all parallel vacuum workers including the leader process
- * to have a shared view of cost related parameters (mainly VacuumCostBalance)
- * and allow each worker to update it and then based on that decide
- * whether it needs to sleep.  Besides, we allow any worker to sleep
- * only if it has performed the I/O above a certain threshold, which is
- * calculated based on the number of active workers (VacuumActiveNWorkers),
- * and the overall cost balance is more than VacuumCostLimit set by the
- * system.  Then we will allow the worker to sleep proportional to the work
- * done and reduce the VacuumSharedCostBalance by the amount which is
- * consumed by the current worker (VacuumCostBalanceLocal).  This can
- * avoid letting the workers sleep which has done less or no I/O as compared
- * to other workers, and therefore can ensure that workers who are doing
- * more I/O got throttled more.
- */
-pg_atomic_uint32	*VacuumSharedCostBalance = NULL;
-pg_atomic_uint32	*VacuumActiveNWorkers = NULL;
-int					VacuumCostBalanceLocal = 0;
-
 /* non-export function prototypes */
 static void lazy_scan_heap(Relation onerel, VacuumParams *params,
 						   LVRelStats *vacrelstats, Relation *Irel, int nindexes,
@@ -364,38 +344,37 @@ static bool lazy_tid_reaped(ItemPointer itemptr, void *state);
 static int	vac_cmp_itemptr(const void *left, const void *right);
 static bool heap_page_is_all_visible(Relation rel, Buffer buf,
 									 TransactionId *visibility_cutoff_xid, bool *all_frozen);
-static void lazy_parallel_vacuum_or_cleanup_indexes(LVRelStats *vacrelstats, Relation *Irel,
-													int nindexes, IndexBulkDeleteResult **stats,
-													LVParallelState *lps);
-static void vacuum_or_cleanup_indexes_worker(Relation *Irel, int nindexes,
-											 IndexBulkDeleteResult **stats,
-											 LVShared *lvshared,
-											 LVDeadTuples *dead_tuples);
-static void vacuum_or_cleanup_skipped_indexes(LVRelStats *vacrelstats, Relation *Irel,
-											  int nindexes, IndexBulkDeleteResult **stats,
-											  LVParallelState *lps);
-static void vacuum_or_cleanup_one_index_worker(Relation indrel, IndexBulkDeleteResult **stats,
-											   LVShared *lvshared, LVSharedIndStats *shared_indstats,
-											   LVDeadTuples *dead_tuples);
-static void lazy_vacuum_indexes(LVRelStats *vacrelstats, Relation *Irel,
-								int nindexes, IndexBulkDeleteResult **stats,
-								LVParallelState *lps);
-static void lazy_cleanup_indexes(LVRelStats *vacrelstats, Relation *Irel,
-								 int nindexes, IndexBulkDeleteResult **stats,
-								 LVParallelState *lps);
-static void update_index_statistics(Relation *Irel, IndexBulkDeleteResult **stats,
-									int nindexes);
+static void lazy_parallel_vacuum_indexes(Relation *Irel, IndexBulkDeleteResult **stats,
+										 LVRelStats *vacrelstats, LVParallelState *lps,
+										 int nindexes);
+static void parallel_vacuum_index(Relation *Irel, IndexBulkDeleteResult **stats,
+								  LVShared *lvshared, LVDeadTuples *dead_tuples,
+								  int nindexes);
+static void vacuum_skipped_indexes(Relation *Irel, IndexBulkDeleteResult **stats,
+								   LVRelStats *vacrelstats, LVParallelState *lps,
+								   int nindexes);
+static void vacuum_one_index(Relation indrel, IndexBulkDeleteResult **stats,
+							 LVShared *lvshared, LVSharedIndStats *shared_indstats,
+							 LVDeadTuples *dead_tuples);
+static void lazy_vacuum_indexes(Relation *Irel, IndexBulkDeleteResult **stats,
+								LVRelStats *vacrelstats, LVParallelState *lps,
+								int nindexes);
+static void lazy_cleanup_indexes(Relation *Irel, IndexBulkDeleteResult **stats,
+								 LVRelStats *vacrelstats, LVParallelState *lps,
+								 int nindexes);
 static long compute_max_dead_tuples(BlockNumber relblocks, bool hasindex);
-static int compute_parallel_workers(Relation *Irel, int nindexes, int nrequested);
-static void prepare_index_statistics(LVShared *lvshared, Relation *Irel, int nindexes,
+static int compute_parallel_vacuum_workers(Relation *Irel, int nindexes, int nrequested);
+static void prepare_index_statistics(Relation *Irel, LVShared *lvshared, int nindexes,
 									 int nworkers);
-static LVParallelState *begin_parallel_vacuum(LVRelStats *vacrelstats, Oid relid,
-											  BlockNumber nblocks, Relation *Irel,
-											  int nindexes, int nrequested);
-static void end_parallel_vacuum(LVParallelState *lps, Relation *Irel, int nindexes,
-								IndexBulkDeleteResult **stats);
+static void update_index_statistics(Relation *Irel, IndexBulkDeleteResult **stats,
+									int nindexes);
+static LVParallelState *begin_parallel_vacuum(Oid relid, Relation *Irel,
+							LVRelStats *vacrelstats, BlockNumber nblocks,
+							int nindexes, int nrequested);
+static void end_parallel_vacuum(Relation *Irel, IndexBulkDeleteResult **stats,
+								LVParallelState *lps, int nindexes);
 static LVSharedIndStats *get_indstats(LVShared *lvshared, int n);
-static bool skip_parallel_index_vacuum(Relation indrel, LVShared *lvshared);
+static bool skip_parallel_vacuum_index(Relation indrel, LVShared *lvshared);
 
 
 /*
@@ -753,7 +732,6 @@ lazy_scan_heap(Relation onerel, VacuumParams *params, LVRelStats *vacrelstats,
 	bool		skipping_blocks;
 	xl_heap_freeze_tuple *frozen;
 	StringInfoData buf;
-	int			parallel_workers = 0;
 	const int	initprog_index[] = {
 		PROGRESS_VACUUM_PHASE,
 		PROGRESS_VACUUM_TOTAL_HEAP_BLKS,
@@ -790,32 +768,19 @@ lazy_scan_heap(Relation onerel, VacuumParams *params, LVRelStats *vacrelstats,
 	vacrelstats->latestRemovedXid = InvalidTransactionId;
 
 	/*
-	 * Compute the number of parallel vacuum workers to launch if the parallel
-	 * vacuum is requested and we need to vacuum the indexes.
+	 * Try to initialize the parallel vacuum if requested
 	 */
 	if (params->nworkers >= 0 && vacrelstats->useindex)
-		parallel_workers = compute_parallel_workers(Irel, nindexes,
-													params->nworkers);
+		lps = begin_parallel_vacuum(RelationGetRelid(onerel), Irel,
+									vacrelstats, nblocks, nindexes,
+									params->nworkers);
 
-	if (parallel_workers > 0)
-	{
-		/*
-		 * Enter parallel mode, create the parallel context and allocate the
-		 * DSM segment.
-		 */
-		lps = begin_parallel_vacuum(vacrelstats,
-									RelationGetRelid(onerel),
-									nblocks, Irel, nindexes,
-									parallel_workers);
-	}
-	else
-	{
-		/*
-		 * Use single process vacuum. We allocate the memory space for dead
-		 * tuples locally.
-		 */
+	/*
+	 * Allocate the space for dead tuples in case the parallel vacuum is not
+	 * initialized.
+	 */
+	if (!ParallelVacuumIsActive(lps))
 		lazy_space_alloc(vacrelstats, nblocks);
-	}
 
 	dead_tuples = vacrelstats->dead_tuples;
 	frozen = palloc(sizeof(xl_heap_freeze_tuple) * MaxHeapTuplesPerPage);
@@ -1030,7 +995,7 @@ lazy_scan_heap(Relation onerel, VacuumParams *params, LVRelStats *vacrelstats,
 										 PROGRESS_VACUUM_PHASE_VACUUM_INDEX);
 
 			/* Remove index entries */
-			lazy_vacuum_indexes(vacrelstats, Irel, nindexes, indstats, lps);
+			lazy_vacuum_indexes(Irel, indstats, vacrelstats, lps, nindexes);
 
 			/*
 			 * Report that we are now vacuuming the heap.  We also increase
@@ -1695,7 +1660,7 @@ lazy_scan_heap(Relation onerel, VacuumParams *params, LVRelStats *vacrelstats,
 									 PROGRESS_VACUUM_PHASE_VACUUM_INDEX);
 
 		/* Remove index entries */
-		lazy_vacuum_indexes(vacrelstats, Irel, nindexes, indstats, lps);
+		lazy_vacuum_indexes(Irel, indstats, vacrelstats, lps, nindexes);
 
 		/* Report that we are now vacuuming the heap */
 		hvp_val[0] = PROGRESS_VACUUM_PHASE_VACUUM_HEAP;
@@ -1723,14 +1688,14 @@ lazy_scan_heap(Relation onerel, VacuumParams *params, LVRelStats *vacrelstats,
 
 	/* Do post-vacuum cleanup */
 	if (vacrelstats->useindex)
-		lazy_cleanup_indexes(vacrelstats, Irel, nindexes, indstats, lps);
+		lazy_cleanup_indexes(Irel, indstats, vacrelstats, lps, nindexes);
 
 	/*
 	 * End parallel mode before updating index statistics as we cannot write
 	 * during parallel mode.
 	 */
 	if (ParallelVacuumIsActive(lps))
-		end_parallel_vacuum(lps, Irel, nindexes, indstats);
+		end_parallel_vacuum(Irel, indstats, lps, nindexes);
 
 	/* Update index statistics */
 	 update_index_statistics(Irel, indstats, nindexes);
@@ -2004,9 +1969,9 @@ lazy_check_needs_freeze(Buffer buf, bool *hastup)
  * or cleanup.
  */
 static void
-lazy_parallel_vacuum_or_cleanup_indexes(LVRelStats *vacrelstats, Relation *Irel,
-										int nindexes, IndexBulkDeleteResult **stats,
-										LVParallelState *lps)
+lazy_parallel_vacuum_indexes(Relation *Irel, IndexBulkDeleteResult **stats,
+							 LVRelStats *vacrelstats, LVParallelState *lps,
+							 int nindexes)
 {
 	int	nworkers;
 
@@ -2030,31 +1995,39 @@ lazy_parallel_vacuum_or_cleanup_indexes(LVRelStats *vacrelstats, Relation *Irel,
 	/* Cap by the worker we computed at the beginning of parallel lazy vacuum */
 	nworkers = Min(nworkers, lps->pcxt->nworkers);
 
-	/* Setup the shared cost-based vacuum delay and launch workers*/
+	/* Setup the shared cost-based vacuum delay and launch workers */
 	if (nworkers > 0)
 	{
+		/* Enable shared cost balance */
+		VacuumSharedCostBalance = &(lps->lvshared->cost_balance);
+		VacuumActiveNWorkers = &(lps->lvshared->active_nworkers);
+
 		/*
-		 * Reset the local value so that we compute cost balance during
-		 * parallel index vacuuming.
+		 * Set up shared cost balance and the number of active workers for
+		 * vacuum delay.
 		 */
-		VacuumCostBalance = 0;
-		VacuumCostBalanceLocal = 0;
+		pg_atomic_write_u32(VacuumSharedCostBalance, VacuumCostBalance);
+		pg_atomic_write_u32(VacuumActiveNWorkers, 0);
 
 		LaunchParallelWorkers(lps->pcxt, nworkers);
 
-		/* Enable shared costing iff we process indexes in parallel. */
 		if (lps->pcxt->nworkers_launched > 0)
 		{
-			/* Enable shared cost balance */
-			VacuumSharedCostBalance = &(lps->lvshared->cost_balance);
-			VacuumActiveNWorkers = &(lps->lvshared->active_nworkers);
-
 			/*
-			 * Set up shared cost balance and the number of active workers for
-			 * vacuum delay.
+			 * Reset the local value so that we compute cost balance during
+			 * parallel index vacuuming.
+			 */
+			VacuumCostBalance = 0;
+			VacuumCostBalanceLocal = 0;
+		}
+		else
+		{
+			/*
+			 * Disable shared cost balance if we are not able to launch
+			 * workers.
 			 */
-			pg_atomic_write_u32(VacuumSharedCostBalance, VacuumCostBalance);
-			pg_atomic_write_u32(VacuumActiveNWorkers, 0);
+			VacuumSharedCostBalance = NULL;
+			VacuumActiveNWorkers = NULL;
 		}
 
 		if (lps->lvshared->for_cleanup)
@@ -2072,20 +2045,24 @@ lazy_parallel_vacuum_or_cleanup_indexes(LVRelStats *vacrelstats, Relation *Irel,
 	}
 
 	/*
-	 * Join as a parallel worker. The leader process alone does that in
-	 * case where no workers launched.
+	 * Join as a parallel worker.  The leader process alone processes all the
+	 * indexes in the case where no workers are launched.
 	 */
 	if (lps->leaderparticipates || lps->pcxt->nworkers_launched == 0)
-		vacuum_or_cleanup_indexes_worker(Irel, nindexes, stats, lps->lvshared,
-										 vacrelstats->dead_tuples);
+		parallel_vacuum_index(Irel, stats, lps->lvshared,
+							  vacrelstats->dead_tuples, nindexes);
 
 	/*
-	 * Here, the indexes that had been skipped during parallel index vacuuming
-	 * are remaining. If there are such indexes the leader process does vacuum
-	 * or cleanup them one by one.
+	 * Now process the indexes that had been skipped during parallel index
+	 * vacuuming.
+	 *
+	 * XXX - We can do this before the leader backend participates as a worker
+	 * and that might make a few cases faster where indexes are of equal size,
+	 * but OTOH, it might slow down a few cases where indexes are of unequal
+	 * size.  In anycase, it won't matter much as we don't have indexes that
+	 * don't support parallel vacuum yet.
 	 */
-	vacuum_or_cleanup_skipped_indexes(vacrelstats, Irel, nindexes, stats,
-									  lps);
+	vacuum_skipped_indexes(Irel, stats, vacrelstats, lps, nindexes);
 
 	/* Wait for all vacuum workers to finish */
 	WaitForParallelWorkersToFinish(lps->pcxt);
@@ -2099,9 +2076,9 @@ lazy_parallel_vacuum_or_cleanup_indexes(LVRelStats *vacrelstats, Relation *Irel,
 	VacuumActiveNWorkers = NULL;
 
 	/*
-	 * In cleanup case we don't need to reinitialize the parallel
-	 * context as no more index vacuuming and index cleanup will be
-	 * performed after that.
+	 * In the cleanup case, we don't need to reinitialize the parallel context
+	 * as no more index vacuuming or index cleanup will be performed after
+	 * that.
 	 */
 	if (!lps->lvshared->for_cleanup)
 	{
@@ -2118,15 +2095,13 @@ lazy_parallel_vacuum_or_cleanup_indexes(LVRelStats *vacrelstats, Relation *Irel,
 }
 
 /*
- * Index vacuuming and index cleanup routine used by parallel vacuum
- * worker processes and the leader process to process the indexes in
- * parallel.
+ * Index vacuum/cleanup routine used by the leader process and parallel
+ * vacuum worker processes to process the indexes in parallel.
  */
 static void
-vacuum_or_cleanup_indexes_worker(Relation *Irel, int nindexes,
-								 IndexBulkDeleteResult **stats,
-								 LVShared *lvshared,
-								 LVDeadTuples *dead_tuples)
+parallel_vacuum_index(Relation *Irel, IndexBulkDeleteResult **stats,
+					  LVShared *lvshared, LVDeadTuples *dead_tuples,
+					  int nindexes)
 {
 	/*
 	 * Increment the active worker count if we are able to launch any worker.
@@ -2148,7 +2123,7 @@ vacuum_or_cleanup_indexes_worker(Relation *Irel, int nindexes,
 			break;
 
 		/* Skip processing indexes that doesn't support parallel operation */
-		if (skip_parallel_index_vacuum(Irel[idx], lvshared))
+		if (skip_parallel_vacuum_index(Irel[idx], lvshared))
 			continue;
 
 		/* Increment the processing count */
@@ -2164,9 +2139,8 @@ vacuum_or_cleanup_indexes_worker(Relation *Irel, int nindexes,
 		Assert(shared_indstats);
 
 		/* Do vacuum or cleanup one index */
-		vacuum_or_cleanup_one_index_worker(Irel[idx], &(stats[idx]),
-										   lvshared, shared_indstats,
-										   dead_tuples);
+		vacuum_one_index(Irel[idx], &(stats[idx]), lvshared, shared_indstats,
+						 dead_tuples);
 	}
 
 	/*
@@ -2179,13 +2153,13 @@ vacuum_or_cleanup_indexes_worker(Relation *Irel, int nindexes,
 
 /*
  * Vacuum or cleanup indexes that have been skipped during parallel operation
- * because these indexes don't support parallel operation at that phase.  Therefore
- * this function must be called by the leader process.
+ * because these indexes don't support parallel operation at that phase.
+ * Therefore this function must be called by the leader process.
  */
 static void
-vacuum_or_cleanup_skipped_indexes(LVRelStats *vacrelstats, Relation *Irel,
-								  int nindexes, IndexBulkDeleteResult **stats,
-								  LVParallelState *lps)
+vacuum_skipped_indexes(Relation *Irel, IndexBulkDeleteResult **stats,
+					   LVRelStats *vacrelstats, LVParallelState *lps,
+					   int nindexes)
 {
 	int nindexes_remains;
 	int i;
@@ -2207,15 +2181,15 @@ vacuum_or_cleanup_skipped_indexes(LVRelStats *vacrelstats, Relation *Irel,
 
 	for (i = 0; i < nindexes; i++)
 	{
-		bool processed = !skip_parallel_index_vacuum(Irel[i], lps->lvshared);
+		bool processed = !skip_parallel_vacuum_index(Irel[i], lps->lvshared);
 
 		/* Skip the already processed indexes */
 		if (processed)
 			continue;
 
-		vacuum_or_cleanup_one_index_worker(Irel[i], &(stats[i]),
-										   lps->lvshared, get_indstats(lps->lvshared, i),
-										   vacrelstats->dead_tuples);
+		vacuum_one_index(Irel[i], &(stats[i]), lps->lvshared,
+						 get_indstats(lps->lvshared, i),
+						 vacrelstats->dead_tuples);
 	}
 
 	/*
@@ -2227,15 +2201,15 @@ vacuum_or_cleanup_skipped_indexes(LVRelStats *vacrelstats, Relation *Irel,
 }
 
 /*
- * Vacuum or cleanup one index by worker processing including the leader
- * process.  After finished each indexes this function copies the index
+ * Vacuum or cleanup index either by leader process or by one of the worker
+ * process.  After processing the index this function copies the index
  * statistics returned from ambulkdelete and amvacuumcleanup to the DSM
  * segment.
  */
 static void
-vacuum_or_cleanup_one_index_worker(Relation indrel, IndexBulkDeleteResult **stats,
-								   LVShared *lvshared, LVSharedIndStats *shared_indstats,
-								   LVDeadTuples *dead_tuples)
+vacuum_one_index(Relation indrel, IndexBulkDeleteResult **stats,
+				 LVShared *lvshared, LVSharedIndStats *shared_indstats,
+				 LVDeadTuples *dead_tuples)
 {
 	IndexBulkDeleteResult *bulkdelete_res = NULL;
 
@@ -2292,9 +2266,9 @@ vacuum_or_cleanup_one_index_worker(Relation indrel, IndexBulkDeleteResult **stat
  * parallel vacuum.
  */
 static void
-lazy_vacuum_indexes(LVRelStats *vacrelstats, Relation *Irel,
-					int nindexes, IndexBulkDeleteResult **stats,
-					LVParallelState *lps)
+lazy_vacuum_indexes(Relation *Irel, IndexBulkDeleteResult **stats,
+					LVRelStats *vacrelstats, LVParallelState *lps,
+					int nindexes)
 {
 	int		idx;
 
@@ -2315,8 +2289,7 @@ lazy_vacuum_indexes(LVRelStats *vacrelstats, Relation *Irel,
 		lps->lvshared->reltuples = vacrelstats->old_live_tuples;
 		lps->lvshared->estimated_count = true;
 
-		lazy_parallel_vacuum_or_cleanup_indexes(vacrelstats, Irel, nindexes,
-												stats, lps);
+		lazy_parallel_vacuum_indexes(Irel, stats, vacrelstats, lps, nindexes);
 	}
 	else
 	{
@@ -2331,9 +2304,9 @@ lazy_vacuum_indexes(LVRelStats *vacrelstats, Relation *Irel,
  * parallel vacuum.
  */
 static void
-lazy_cleanup_indexes(LVRelStats *vacrelstats, Relation *Irel,
-					 int nindexes, IndexBulkDeleteResult **stats,
-					 LVParallelState *lps)
+lazy_cleanup_indexes(Relation *Irel, IndexBulkDeleteResult **stats,
+					 LVRelStats *vacrelstats, LVParallelState *lps,
+					 int nindexes)
 {
 	int		idx;
 
@@ -2360,8 +2333,7 @@ lazy_cleanup_indexes(LVRelStats *vacrelstats, Relation *Irel,
 		lps->lvshared->estimated_count =
 					(vacrelstats->tupcount_pages < vacrelstats->rel_pages);
 
-		lazy_parallel_vacuum_or_cleanup_indexes(vacrelstats, Irel, nindexes,
-												stats, lps);
+		lazy_parallel_vacuum_indexes(Irel, stats, vacrelstats, lps, nindexes);
 	}
 	else
 	{
@@ -2376,16 +2348,17 @@ lazy_cleanup_indexes(LVRelStats *vacrelstats, Relation *Irel,
  *	lazy_vacuum_index() -- vacuum one index relation.
  *
  *		Delete all the index entries pointing to tuples listed in
- *		vacrelstats->dead_tuples, and update running statistics.
+ *		dead_tuples, and update running statistics.
+ *
  *		reltuples is the number of heap tuples to be passed to the
- *		bulk delete callback.
+ *		bulkdelete callback.
  */
 static void
 lazy_vacuum_index(Relation indrel, IndexBulkDeleteResult **stats,
 				  LVDeadTuples *dead_tuples, double reltuples)
 {
 	IndexVacuumInfo ivinfo;
-	char		*msgfmt;
+	const char	*msg;
 	PGRUsage	ru0;
 
 	pg_rusage_init(&ru0);
@@ -2403,12 +2376,12 @@ lazy_vacuum_index(Relation indrel, IndexBulkDeleteResult **stats,
 							   lazy_tid_reaped, (void *) dead_tuples);
 
 	if (IsParallelWorker())
-		msgfmt = gettext_noop("scanned index \"%s\" to remove %d row versions by parallel vacuum worker");
+		msg = gettext_noop("scanned index \"%s\" to remove %d row versions by parallel vacuum worker");
 	else
-		msgfmt = gettext_noop("scanned index \"%s\" to remove %d row versions");
+		msg = gettext_noop("scanned index \"%s\" to remove %d row versions");
 
 	ereport(elevel,
-			(errmsg(msgfmt,
+			(errmsg(msg,
 					RelationGetRelationName(indrel),
 					dead_tuples->num_tuples),
 			 errdetail_internal("%s", pg_rusage_show(&ru0))));
@@ -2426,7 +2399,7 @@ lazy_cleanup_index(Relation indrel,
 				   double reltuples, bool estimated_count)
 {
 	IndexVacuumInfo ivinfo;
-	char		*msgfmt;
+	const char	*msg;
 	PGRUsage	ru0;
 
 	pg_rusage_init(&ru0);
@@ -2446,12 +2419,12 @@ lazy_cleanup_index(Relation indrel,
 		return;
 
 	if (IsParallelWorker())
-		msgfmt = gettext_noop("index \"%s\" now contains %.0f row versions in %u pages, reported by parallel vacuum worker");
+		msg = gettext_noop("index \"%s\" now contains %.0f row versions in %u pages as reported by parallel vacuum worker");
 	else
-		msgfmt = gettext_noop("index \"%s\" now contains %.0f row versions in %u pages");
+		msg = gettext_noop("index \"%s\" now contains %.0f row versions in %u pages");
 
 	ereport(elevel,
-			(errmsg(msgfmt,
+			(errmsg(msg,
 					RelationGetRelationName(indrel),
 					(*stats)->num_index_tuples,
 					(*stats)->num_pages),
@@ -2464,35 +2437,6 @@ lazy_cleanup_index(Relation indrel,
 }
 
 /*
- * Update index statistics in pg_class if the statistics is accurate.
- */
-static void
-update_index_statistics(Relation *Irel, IndexBulkDeleteResult **stats,
-						int nindexes)
-{
-	int i;
-
-	Assert(!IsInParallelMode());
-
-	for (i = 0; i < nindexes; i++)
-	{
-		if (stats[i] == NULL || stats[i]->estimated_count)
-			continue;
-
-		/* Update index statistics */
-		vac_update_relstats(Irel[i],
-							stats[i]->num_pages,
-							stats[i]->num_index_tuples,
-							0,
-							false,
-							InvalidTransactionId,
-							InvalidMultiXactId,
-							false);
-		pfree(stats[i]);
-	}
-}
-
-/*
  * should_attempt_truncation - should we attempt to truncate the heap?
  *
  * Don't even think about it unless we have a shot at releasing a goodly
@@ -3032,7 +2976,7 @@ heap_page_is_all_visible(Relation rel, Buffer buf,
 }
 
 /*
- * Compute the number of parallel worker processes to request. Both index
+ * Compute the number of parallel worker processes to request.  Both index
  * vacuuming and index cleanup can be executed together with parallel workers.
  * The relation sizes of table and indexes don't affect to the parallel
  * degree for now. nrequested is the number of parallel workers that user
@@ -3041,7 +2985,7 @@ heap_page_is_all_visible(Relation rel, Buffer buf,
  * vacuuming.
  */
 static int
-compute_parallel_workers(Relation *Irel, int nindexes, int nrequested)
+compute_parallel_vacuum_workers(Relation *Irel, int nindexes, int nrequested)
 {
 	bool	leaderparticipates = true;
 	int		nindexes_parallel = 0;
@@ -3050,10 +2994,11 @@ compute_parallel_workers(Relation *Irel, int nindexes, int nrequested)
 	int		parallel_workers;
 	int		i;
 
-	Assert(nrequested >= 0);
-
-	/* Return immediately when parallelism disabled */
-	if (max_parallel_maintenance_workers == 0)
+	/*
+	 * We don't allow to perform parallel operation in standalone backend or
+	 * when parallelism is disabled.
+	 */
+	if (!IsUnderPostmaster || max_parallel_maintenance_workers == 0)
 		return 0;
 
 	/*
@@ -3104,8 +3049,8 @@ compute_parallel_workers(Relation *Irel, int nindexes, int nrequested)
  * for autovacuum we don't need to care about autovacuum_work_mem.
  */
 static void
-prepare_index_statistics(LVShared *lvshared, Relation *Irel, int nindexes,
-	int nworkers)
+prepare_index_statistics(Relation *Irel, LVShared *lvshared, int nindexes,
+						 int nworkers)
 {
 	char *p = (char *) GetSharedIndStats(lvshared);
 	int nindexes_mwm = 0;
@@ -3145,13 +3090,44 @@ prepare_index_statistics(LVShared *lvshared, Relation *Irel, int nindexes,
 }
 
 /*
- * Enter parallel mode, allocate and initialize the DSM segment.
+ * Update index statistics in pg_class if the statistics is accurate.
+ */
+static void
+update_index_statistics(Relation *Irel, IndexBulkDeleteResult **stats,
+						int nindexes)
+{
+	int i;
+
+	Assert(!IsInParallelMode());
+
+	for (i = 0; i < nindexes; i++)
+	{
+		if (stats[i] == NULL || stats[i]->estimated_count)
+			continue;
+
+		/* Update index statistics */
+		vac_update_relstats(Irel[i],
+							stats[i]->num_pages,
+							stats[i]->num_index_tuples,
+							0,
+							false,
+							InvalidTransactionId,
+							InvalidMultiXactId,
+							false);
+		pfree(stats[i]);
+	}
+}
+
+/*
+ * This function prepares and returns parallel vacuum state if we can launch
+ * even one worker.  This function is responsible to create a parallel
+ * context, enter parallel mode, and then initialize the DSM segment.
  */
 static LVParallelState *
-begin_parallel_vacuum(LVRelStats *vacrelstats, Oid relid, BlockNumber nblocks,
-					  Relation *Irel, int nindexes, int nrequested)
+begin_parallel_vacuum(Oid relid, Relation *Irel, LVRelStats *vacrelstats,
+					  BlockNumber nblocks, int nindexes, int nrequested)
 {
-	LVParallelState *lps = (LVParallelState *) palloc0(sizeof(LVParallelState));
+	LVParallelState *lps = NULL;
 	ParallelContext *pcxt;
 	LVShared		*shared;
 	LVDeadTuples	*dead_tuples;
@@ -3159,12 +3135,29 @@ begin_parallel_vacuum(LVRelStats *vacrelstats, Oid relid, BlockNumber nblocks,
 	char	*sharedquery;
 	Size	est_shared;
 	Size	est_deadtuples;
+	int		parallel_workers = 0;
 	int		querylen;
 	int		i;
 
-	Assert(nrequested > 0);
+	/*
+	 * a parallel vacuum must be requested and there must be indexes on the
+	 * relation
+	 */
+	Assert(nrequested >= 0);
 	Assert(nindexes > 0);
 
+	/*
+	 * Compute the number of parallel vacuum workers to launch
+	 */
+	parallel_workers = compute_parallel_vacuum_workers(Irel, nindexes,
+													   nrequested);
+
+	/* Can't perform vacuum in parallel */
+	if (parallel_workers <= 0)
+		return lps;
+
+	lps = (LVParallelState *) palloc0(sizeof(LVParallelState));
+
 	lps->leaderparticipates = true;
 
 #ifdef PARALLEL_VACUUM_DISABLE_LEADER_PARTICIPATION
@@ -3172,8 +3165,8 @@ begin_parallel_vacuum(LVRelStats *vacrelstats, Oid relid, BlockNumber nblocks,
 #endif
 
 	EnterParallelMode();
-	pcxt = CreateParallelContext("postgres", "heap_parallel_vacuum_main",
-								 nrequested);
+	pcxt = CreateParallelContext("postgres", "parallel_vacuum_main",
+								 parallel_workers);
 	lps->pcxt = pcxt;
 	Assert(pcxt->nworkers > 0);
 
@@ -3199,7 +3192,7 @@ begin_parallel_vacuum(LVRelStats *vacrelstats, Oid relid, BlockNumber nblocks,
 
 			/*
 			 * Remember the number of indexes that support parallel operation
-			 * for each phases.
+			 * for each phase.
 			 */
 			if ((vacoptions & VACUUM_OPTION_PARALLEL_BULKDEL) != 0)
 				lps->nindexes_parallel_bulkdel++;
@@ -3237,7 +3230,7 @@ begin_parallel_vacuum(LVRelStats *vacrelstats, Oid relid, BlockNumber nblocks,
 	 * in that way.
 	 */
 	shared->offset = MAXALIGN(add_size(SizeOfLVShared, BITMAPLEN(nindexes)));
-	prepare_index_statistics(shared, Irel, nindexes, nrequested);
+	prepare_index_statistics(Irel, shared, nindexes, parallel_workers);
 	pg_atomic_init_u32(&(shared->idx), 0);
 	pg_atomic_init_u32(&(shared->nprocessed), 0);
 	pg_atomic_init_u32(&(shared->cost_balance), 0);
@@ -3266,14 +3259,15 @@ begin_parallel_vacuum(LVRelStats *vacrelstats, Oid relid, BlockNumber nblocks,
 /*
  * Destroy the parallel context, and end parallel mode.
  *
- * All writes are not allowed during parallel mode and it might not be
- * safe to exit from the parallel mode while keeping the parallel context.
- * So we copy the updated index statistics to a local memory and then later
- * use that to update the index statistics.
+ * Since writes are not allowed during the parallel mode, so we copy the
+ * updated index statistics from DSM in local memory and then later use that
+ * to update the index statistics.  One might think that we can exit from
+ * parallel mode, update the index statistics and then destroy parallel
+ * context, but that won't be safe (see ExitParallelMode).
  */
 static void
-end_parallel_vacuum(LVParallelState *lps, Relation *Irel, int nindexes,
-					IndexBulkDeleteResult **stats)
+end_parallel_vacuum(Relation *Irel, IndexBulkDeleteResult **stats,
+					LVParallelState *lps, int nindexes)
 {
 	int i;
 
@@ -3285,8 +3279,8 @@ end_parallel_vacuum(LVParallelState *lps, Relation *Irel, int nindexes,
 		LVSharedIndStats *indstats = get_indstats(lps->lvshared, i);
 
 		/*
-		 * Skip unused slot.  The statistics of this index are already
-		 * stored in local memory.
+		 * Skip unused slot.  The statistics of this index are already stored
+		 * in local memory.
 		 */
 		if (indstats == NULL)
 			continue;
@@ -3333,11 +3327,11 @@ get_indstats(LVShared *lvshared, int n)
 }
 
 /*
- * Check if the given index participates in parallel index vacuuming
- * or parallel index cleanup.
+ * Check if the given index participates in parallel index vacuum or parallel
+ * index cleanup.
  */
 static bool
-skip_parallel_index_vacuum(Relation indrel, LVShared *lvshared)
+skip_parallel_vacuum_index(Relation indrel, LVShared *lvshared)
 {
 	uint8 vacoptions = indrel->rd_indam->amparallelvacuumoptions;
 
@@ -3373,11 +3367,11 @@ skip_parallel_index_vacuum(Relation indrel, LVShared *lvshared)
 /*
  * Perform work within a launched parallel process.
  *
- * Since parallel vacuum workers work only within index vacuuming and index
- * cleanup, no need to report the progress information.
+ * Since parallel vacuum workers perform only index vacuum or index cleanup,
+ * we don't need to report the progress information.
  */
 void
-heap_parallel_vacuum_main(dsm_segment *seg, shm_toc *toc)
+parallel_vacuum_main(dsm_segment *seg, shm_toc *toc)
 {
 	Relation	onerel;
 	Relation	*indrels;
@@ -3434,9 +3428,8 @@ heap_parallel_vacuum_main(dsm_segment *seg, shm_toc *toc)
 	if (lvshared->maintenance_work_mem_worker > 0)
 		maintenance_work_mem = lvshared->maintenance_work_mem_worker;
 
-	/* Do either vacuuming indexes or cleaning indexes */
-	vacuum_or_cleanup_indexes_worker(indrels, nindexes, stats, lvshared,
-									 dead_tuples);
+	/* Process indexes to perform vacuum/cleanup */
+	parallel_vacuum_index(indrels, stats, lvshared, dead_tuples, nindexes);
 
 	vac_close_indexes(nindexes, indrels, RowExclusiveLock);
 	table_close(onerel, ShareUpdateExclusiveLock);
diff --git a/src/backend/access/transam/parallel.c b/src/backend/access/transam/parallel.c
index 157c309..a387932 100644
--- a/src/backend/access/transam/parallel.c
+++ b/src/backend/access/transam/parallel.c
@@ -142,7 +142,7 @@ static const struct
 		"_bt_parallel_build_main", _bt_parallel_build_main
 	},
 	{
-		"heap_parallel_vacuum_main", heap_parallel_vacuum_main
+		"parallel_vacuum_main", parallel_vacuum_main
 	}
 };
 
@@ -498,8 +498,8 @@ LaunchParallelWorkers(ParallelContext *pcxt, int nworkers)
 {
 	MemoryContext oldcontext;
 	BackgroundWorker worker;
-	int			nworkers_to_launch = Min(nworkers, pcxt->nworkers);;
 	int			i;
+	int			nworkers_to_launch = Min(nworkers, pcxt->nworkers);
 	bool		any_registrations_failed = false;
 
 	/* Skip this if we have no workers. */
diff --git a/src/backend/commands/vacuum.c b/src/backend/commands/vacuum.c
index 4b7f480..f6dc890 100644
--- a/src/backend/commands/vacuum.c
+++ b/src/backend/commands/vacuum.c
@@ -69,6 +69,14 @@ static MemoryContext vac_context = NULL;
 static BufferAccessStrategy vac_strategy;
 
 
+/*
+ * Variables for cost-based parallel vacuum.  See comments atop
+ * compute_parallel_delay to understand how it works.
+ */
+pg_atomic_uint32	*VacuumSharedCostBalance = NULL;
+pg_atomic_uint32	*VacuumActiveNWorkers = NULL;
+int					VacuumCostBalanceLocal = 0;
+
 /* non-export function prototypes */
 static List *expand_vacuum_rel(VacuumRelation *vrel, int options);
 static List *get_all_vacuum_rels(int options);
@@ -77,6 +85,7 @@ static void vac_truncate_clog(TransactionId frozenXID,
 							  TransactionId lastSaneFrozenXid,
 							  MultiXactId lastSaneMinMulti);
 static bool vacuum_rel(Oid relid, RangeVar *relation, VacuumParams *params);
+static double compute_parallel_delay(void);
 static VacOptTernaryValue get_vacopt_ternary_value(DefElem *def);
 
 /*
@@ -1994,58 +2003,11 @@ vacuum_delay_point(void)
 		return;
 
 	/*
-	 * If the vacuum cost balance is shared among parallel workers we
-	 * decide whether to sleep based on that.
+	 * For parallel vacuum, the delay is computed based on the shared cost
+	 * balance.  See compute_parallel_delay.
 	 */
 	if (VacuumSharedCostBalance != NULL)
-	{
-		int nworkers = pg_atomic_read_u32(VacuumActiveNWorkers);
-
-		/* At least count itself */
-		Assert(nworkers >= 1);
-
-		/* Update the shared cost balance value atomically */
-		while (true)
-		{
-			uint32 shared_balance;
-			uint32 new_balance;
-			uint32 local_balance;
-
-			msec = 0;
-
-			/* compute new balance by adding the local value */
-			shared_balance = pg_atomic_read_u32(VacuumSharedCostBalance);
-			new_balance = shared_balance + VacuumCostBalance;
-
-			/* also compute the total local balance */
-			local_balance = VacuumCostBalanceLocal + VacuumCostBalance;
-
-			if ((new_balance >= VacuumCostLimit) &&
-				(local_balance > 0.5 * (VacuumCostLimit / nworkers)))
-			{
-				/* compute sleep time based on the local cost balance */
-				msec = VacuumCostDelay * VacuumCostBalanceLocal / VacuumCostLimit;
-				new_balance = shared_balance - VacuumCostBalanceLocal;
-				VacuumCostBalanceLocal = 0;
-			}
-
-			if (pg_atomic_compare_exchange_u32(VacuumSharedCostBalance,
-											   &shared_balance,
-											   new_balance))
-			{
-				/* Updated successfully, break */
-				break;
-			}
-		}
-
-		VacuumCostBalanceLocal += VacuumCostBalance;
-
-		/*
-		 * Reset the local balance as we accumulated it into the shared
-		 * value.
-		 */
-		VacuumCostBalance = 0;
-	}
+		msec = compute_parallel_delay();
 	else if (VacuumCostBalance >= VacuumCostLimit)
 		msec = VacuumCostDelay * VacuumCostBalance / VacuumCostLimit;
 
@@ -2068,6 +2030,59 @@ vacuum_delay_point(void)
 }
 
 /*
+ * Computes the vacuum delay for parallel workers.
+ *
+ * The basic idea of a cost-based vacuum delay for parallel index vacuuming
+ * is to allow all parallel vacuum workers including the leader process to
+ * have a shared view of cost related parameters (mainly VacuumCostBalance)
+ * and allow each worker to update it and then based on that decide
+ * whether it needs to sleep.  Besides, we allow any worker to sleep only
+ * if it has performed the I/O above a certain threshold, which is calculated
+ * based on the number of active workers (VacuumActiveNWorkers), and the
+ * overall cost balance is more than VacuumCostLimit set by the system.  Then
+ * we will allow the worker to sleep proportional to the work done and reduce
+ * the VacuumSharedCostBalance by the amount which is consumed by the current
+ * worker (VacuumCostBalanceLocal).  This can avoid letting the workers sleep
+ * who have done less or no I/O as compared to other workers and therefore can
+ * ensure that workers who are doing more I/O got throttled more.
+ */
+static double
+compute_parallel_delay(void)
+{
+	double	msec = 0;
+	uint32	shared_balance;
+	int		nworkers = pg_atomic_read_u32(VacuumActiveNWorkers);
+
+	/* At least count itself */
+	Assert(nworkers >= 1);
+	/* parallel vacuum must be active */
+	Assert(VacuumSharedCostBalance);
+
+	/* Update the shared cost balance value atomically */
+	shared_balance = pg_atomic_add_fetch_u32(VacuumSharedCostBalance, VacuumCostBalance);
+
+	/* also compute the total local balance */
+	VacuumCostBalanceLocal += VacuumCostBalance;
+
+	if ((shared_balance >= VacuumCostLimit) &&
+		(VacuumCostBalanceLocal > 0.5 * (VacuumCostLimit / nworkers)))
+	{
+		/* compute sleep time based on the local cost balance */
+		msec = VacuumCostDelay * VacuumCostBalanceLocal / VacuumCostLimit;
+		shared_balance = pg_atomic_sub_fetch_u32(VacuumSharedCostBalance, VacuumCostBalanceLocal);
+		VacuumCostBalanceLocal = 0;
+	}
+
+	/*
+	 * Reset the local balance as we accumulated it into the shared
+	 * value.
+	 */
+	VacuumCostBalance = 0;
+
+	return msec;
+}
+
+/*
  * A wrapper function of defGetBoolean().
  *
  * This function returns VACOPT_TERNARY_ENABLED and VACOPT_TERNARY_DISABLED
diff --git a/src/include/access/heapam.h b/src/include/access/heapam.h
index 61725e7..e89c125 100644
--- a/src/include/access/heapam.h
+++ b/src/include/access/heapam.h
@@ -192,13 +192,10 @@ extern void SyncScanShmemInit(void);
 extern Size SyncScanShmemSize(void);
 
 /* in heap/vacuumlazy.c */
-extern pg_atomic_uint32	*VacuumSharedCostBalance;
-extern pg_atomic_uint32	*VacuumActiveNWorkers;
-extern int	VacuumCostBalanceLocal;
 struct VacuumParams;
 extern void heap_vacuum_rel(Relation onerel,
 							struct VacuumParams *params, BufferAccessStrategy bstrategy);
-extern void heap_parallel_vacuum_main(dsm_segment *seg, shm_toc *toc);
+extern void parallel_vacuum_main(dsm_segment *seg, shm_toc *toc);
 
 /* in heap/heapam_visibility.c */
 extern bool HeapTupleSatisfiesVisibility(HeapTuple stup, Snapshot snapshot,
diff --git a/src/include/commands/vacuum.h b/src/include/commands/vacuum.h
index 0a586dc..5222e33 100644
--- a/src/include/commands/vacuum.h
+++ b/src/include/commands/vacuum.h
@@ -232,6 +232,11 @@ extern int	vacuum_freeze_table_age;
 extern int	vacuum_multixact_freeze_min_age;
 extern int	vacuum_multixact_freeze_table_age;
 
+/* Variables for cost-based parallel vacuum  */
+extern pg_atomic_uint32	*VacuumSharedCostBalance;
+extern pg_atomic_uint32	*VacuumActiveNWorkers;
+extern int	VacuumCostBalanceLocal;
+
 
 /* in commands/vacuum.c */
 extern void ExecVacuum(ParseState *pstate, VacuumStmt *vacstmt, bool isTopLevel);
-- 
1.8.3.1

