diff --git a/src/backend/access/heap/vacuumlazy.c b/src/backend/access/heap/vacuumlazy.c index 6f90a4dc40..0bf62cf2c5 100644 --- a/src/backend/access/heap/vacuumlazy.c +++ b/src/backend/access/heap/vacuumlazy.c @@ -30,8 +30,8 @@ * shared information as well as the memory space for storing dead tuples. * When starting either index vacuuming or index cleanup, we launch parallel * worker processes. Once all indexes are processed the parallel worker - * 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 + * processes exit. After that, the leader process re-initializes the parallel + * context so that it can use the same DSM for multiple passes of index * vacuum and for performing index cleanup. 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 @@ -140,7 +140,7 @@ /* * Macro to check if we are in a parallel lazy vacuum. If true, we are - * in the parallel mode and prepared the DSM segment. + * in the parallel mode and the DSM segment is initialized. */ #define ParallelVacuumIsActive(lps) (((LVParallelState *) (lps)) != NULL) @@ -299,6 +299,7 @@ static MultiXactId MultiXactCutoff; static BufferAccessStrategy vac_strategy; + /* non-export function prototypes */ static void lazy_scan_heap(Relation onerel, VacuumParams *params, LVRelStats *vacrelstats, Relation *Irel, int nindexes, @@ -675,9 +676,9 @@ vacuum_log_cleanup_info(Relation rel, LVRelStats *vacrelstats) * parallel workers are launched at beginning of index vacuuming and index * cleanup and they exit once done with all indexes. At the end of this * function we exit from parallel mode. Index bulk-deletion results are - * stored in the DSM segment and update index statistics as a whole after - * exited from parallel mode since all writes are not allowed during parallel - * mode. + * stored in the DSM segment and we update index statistics for all the + * indexes after exiting from parallel mode since writes are not allowed + * during parallel mode. * * If there are no indexes then we can reclaim line pointers on the fly; * dead line pointers need only be retained until all index pointers that @@ -1969,8 +1970,11 @@ lazy_parallel_vacuum_indexes(Relation *Irel, IndexBulkDeleteResult **stats, else nworkers = lps->nindexes_parallel_bulkdel - 1; - /* Cap by the worker we computed at the beginning of parallel lazy vacuum */ - nworkers = Min(nworkers, lps->pcxt->nworkers); + /* + * The number of workers required for parallel vacuum phase must be less + * than the number of workers with which parallel context is initialized. + */ + Assert(lps->pcxt->nworkers >= nworkers); /* Setup the shared cost-based vacuum delay and launch workers */ if (nworkers > 0) @@ -1982,7 +1986,7 @@ lazy_parallel_vacuum_indexes(Relation *Irel, IndexBulkDeleteResult **stats, pg_atomic_write_u32(&(lps->lvshared->nprocessed), 0); /* Reinitialize the parallel context to relaunch parallel workers */ - ReinitializeParallelDSM(lps->pcxt, nworkers); + ReinitializeParallelDSM(lps->pcxt); } /* Enable shared cost balance */ @@ -1996,13 +2000,19 @@ lazy_parallel_vacuum_indexes(Relation *Irel, IndexBulkDeleteResult **stats, pg_atomic_write_u32(VacuumSharedCostBalance, VacuumCostBalance); pg_atomic_write_u32(VacuumActiveNWorkers, 0); + /* + * The number of workers can vary between and bulkdelete and cleanup + * phase. + */ + ReinitializeParallelWorkers(lps->pcxt, nworkers); + LaunchParallelWorkers(lps->pcxt); if (lps->pcxt->nworkers_launched > 0) { /* - * Reset the local value so that we compute cost balance during - * parallel index vacuuming. + * Reset the local cost values for leader backend as we have + * already accumulated the remaining balance of heap. */ VacuumCostBalance = 0; VacuumCostBalanceLocal = 0; @@ -2101,7 +2111,7 @@ parallel_vacuum_index(Relation *Irel, IndexBulkDeleteResult **stats, */ Assert(shared_indstats); - /* Do vacuum or cleanup one index */ + /* Do vacuum or cleanup of the index */ vacuum_one_index(Irel[idx], &(stats[idx]), lvshared, shared_indstats, dead_tuples); } @@ -2181,7 +2191,7 @@ vacuum_one_index(Relation indrel, IndexBulkDeleteResult **stats, *stats = bulkdelete_res; } - /* Do vacuum or cleanup one index */ + /* Do vacuum or cleanup of the index */ if (lvshared->for_cleanup) lazy_cleanup_index(indrel, stats, lvshared->reltuples, lvshared->estimated_count); @@ -2194,7 +2204,7 @@ vacuum_one_index(Relation indrel, IndexBulkDeleteResult **stats, * amvacuumcleanup to the DSM segment if it's the first time to get it * from them, because they allocate it locally and it's possible that an * index will be vacuumed by the different vacuum process at the next - * time. The copying the result normally happens only after the first + * time. The copying of the result normally happens only after the first * time of index vacuuming. From the second time, we pass the result on * the DSM segment so that they then update it directly. * @@ -2207,8 +2217,8 @@ vacuum_one_index(Relation indrel, IndexBulkDeleteResult **stats, shared_indstats->updated = true; /* - * no longer need the locally allocated result and now stats[idx] - * points to the DSM segment. + * Now that the stats[idx] points to the DSM segment, we don't need + * the locally allocated results. */ pfree(*stats); *stats = bulkdelete_res; @@ -2965,7 +2975,7 @@ compute_parallel_vacuum_workers(Relation *Irel, int nindexes, int nrequested) if ((vacoptions & VACUUM_OPTION_PARALLEL_BULKDEL) != 0) nindexes_parallel_bulkdel++; if (((vacoptions & VACUUM_OPTION_PARALLEL_CLEANUP) != 0) || - ((vacoptions & VACUUM_OPTION_PARALLEL_CLEANUP) != 0)) + ((vacoptions & VACUUM_OPTION_PARALLEL_COND_CLEANUP) != 0)) nindexes_parallel_cleanup++; } @@ -3158,7 +3168,6 @@ begin_parallel_vacuum(Oid relid, Relation *Irel, LVRelStats *vacrelstats, MemSet(shared, 0, est_shared); shared->relid = relid; shared->elevel = elevel; - shared->disable_delay = (params->options & VACOPT_FAST); shared->maintenance_work_mem_worker = (nindexes_mwm > 0) ? maintenance_work_mem / Min(parallel_workers, nindexes_mwm) : diff --git a/src/backend/access/transam/parallel.c b/src/backend/access/transam/parallel.c index b90305cde6..aae81a571a 100644 --- a/src/backend/access/transam/parallel.c +++ b/src/backend/access/transam/parallel.c @@ -449,12 +449,10 @@ InitializeParallelDSM(ParallelContext *pcxt) * to launch in the next execution. */ void -ReinitializeParallelDSM(ParallelContext *pcxt, int nworkers) +ReinitializeParallelDSM(ParallelContext *pcxt) { FixedParallelState *fps; - Assert(nworkers >= 0 && pcxt->nworkers >= nworkers); - /* Wait for any old workers to exit. */ if (pcxt->nworkers_launched > 0) { @@ -494,6 +492,23 @@ ReinitializeParallelDSM(ParallelContext *pcxt, int nworkers) } } +/* + * Reinitialize parallel workers for a parallel context such that we could + * launch the different number of workers. This is required for cases where + * we need to reuse the same DSM segment, but the number of workers can + * vary from run-to-run. + */ +void +ReinitializeParallelWorkers(ParallelContext *pcxt, int nworkers_to_launch) +{ + /* + * The number of workers that need to be launched must be less than the + * number of workers with which the parallel context is initialized. + */ + Assert(pcxt->nworkers >= nworkers_to_launch); + pcxt->nworkers_to_launch = nworkers_to_launch; +} + /* * Launch parallel workers. */ diff --git a/src/backend/executor/execParallel.c b/src/backend/executor/execParallel.c index 0dd1bf4472..683b06cdd6 100644 --- a/src/backend/executor/execParallel.c +++ b/src/backend/executor/execParallel.c @@ -873,7 +873,7 @@ ExecParallelReinitialize(PlanState *planstate, */ ExecSetParamPlanMulti(sendParams, GetPerTupleExprContext(estate)); - ReinitializeParallelDSM(pei->pcxt, pei->pcxt->nworkers); + ReinitializeParallelDSM(pei->pcxt); pei->tqueue = ExecParallelSetupTupleQueues(pei->pcxt, true); pei->reader = NULL; pei->finished = false; diff --git a/src/include/access/parallel.h b/src/include/access/parallel.h index 86d24650ba..5add74a6dd 100644 --- a/src/include/access/parallel.h +++ b/src/include/access/parallel.h @@ -63,7 +63,8 @@ extern PGDLLIMPORT bool InitializingParallelWorker; extern ParallelContext *CreateParallelContext(const char *library_name, const char *function_name, int nworkers); extern void InitializeParallelDSM(ParallelContext *pcxt); -extern void ReinitializeParallelDSM(ParallelContext *pcxt, int nworkers); +extern void ReinitializeParallelDSM(ParallelContext *pcxt); +extern void ReinitializeParallelWorkers(ParallelContext *pcxt, int nworkers_to_launch); extern void LaunchParallelWorkers(ParallelContext *pcxt); extern void WaitForParallelWorkersToAttach(ParallelContext *pcxt); extern void WaitForParallelWorkersToFinish(ParallelContext *pcxt);