From afa3f4c3d8993b775837cd04e5d170012b9d2691 Mon Sep 17 00:00:00 2001 From: Daniil Davidov Date: Fri, 16 May 2025 11:58:40 +0700 Subject: [PATCH v4 1/2] Parallel index autovacuum with bgworkers --- src/backend/access/common/reloptions.c | 12 +++ src/backend/access/heap/vacuumlazy.c | 6 +- src/backend/access/transam/parallel.c | 11 +++ src/backend/commands/vacuumparallel.c | 76 +++++++++++++------ src/backend/postmaster/autovacuum.c | 76 ++++++++++++++++++- src/backend/utils/init/globals.c | 1 + src/backend/utils/misc/guc_tables.c | 10 +++ src/backend/utils/misc/postgresql.conf.sample | 2 + src/include/miscadmin.h | 1 + src/include/postmaster/autovacuum.h | 4 + src/include/utils/guc_hooks.h | 2 + src/include/utils/rel.h | 12 +++ 12 files changed, 186 insertions(+), 27 deletions(-) diff --git a/src/backend/access/common/reloptions.c b/src/backend/access/common/reloptions.c index 46c1dce222d..6ba8da62546 100644 --- a/src/backend/access/common/reloptions.c +++ b/src/backend/access/common/reloptions.c @@ -222,6 +222,16 @@ static relopt_int intRelOpts[] = }, SPGIST_DEFAULT_FILLFACTOR, SPGIST_MIN_FILLFACTOR, 100 }, + { + { + "parallel_autovacuum_workers", + "Maximum number of parallel autovacuum workers that can be taken from bgworkers pool for processing this table. " + "If value is 0 then parallel degree will computed based on number of indexes.", + RELOPT_KIND_HEAP, + ShareUpdateExclusiveLock + }, + -1, -1, 1024 + }, { { "autovacuum_vacuum_threshold", @@ -1863,6 +1873,8 @@ default_reloptions(Datum reloptions, bool validate, relopt_kind kind) {"fillfactor", RELOPT_TYPE_INT, offsetof(StdRdOptions, fillfactor)}, {"autovacuum_enabled", RELOPT_TYPE_BOOL, offsetof(StdRdOptions, autovacuum) + offsetof(AutoVacOpts, enabled)}, + {"parallel_autovacuum_workers", RELOPT_TYPE_INT, + offsetof(StdRdOptions, autovacuum) + offsetof(AutoVacOpts, parallel_autovacuum_workers)}, {"autovacuum_vacuum_threshold", RELOPT_TYPE_INT, offsetof(StdRdOptions, autovacuum) + offsetof(AutoVacOpts, vacuum_threshold)}, {"autovacuum_vacuum_max_threshold", RELOPT_TYPE_INT, diff --git a/src/backend/access/heap/vacuumlazy.c b/src/backend/access/heap/vacuumlazy.c index f28326bad09..2614ceba139 100644 --- a/src/backend/access/heap/vacuumlazy.c +++ b/src/backend/access/heap/vacuumlazy.c @@ -3487,6 +3487,10 @@ dead_items_alloc(LVRelState *vacrel, int nworkers) autovacuum_work_mem != -1 ? autovacuum_work_mem : maintenance_work_mem; + int elevel = AmAutoVacuumWorkerProcess() || + vacrel->verbose ? + INFO : DEBUG2; + /* * Initialize state for a parallel vacuum. As of now, only one worker can * be used for an index, so we invoke parallelism only if there are at @@ -3513,7 +3517,7 @@ dead_items_alloc(LVRelState *vacrel, int nworkers) vacrel->pvs = parallel_vacuum_init(vacrel->rel, vacrel->indrels, vacrel->nindexes, nworkers, vac_work_mem, - vacrel->verbose ? INFO : DEBUG2, + elevel, vacrel->bstrategy); /* diff --git a/src/backend/access/transam/parallel.c b/src/backend/access/transam/parallel.c index 94db1ec3012..d3313774a4b 100644 --- a/src/backend/access/transam/parallel.c +++ b/src/backend/access/transam/parallel.c @@ -34,6 +34,7 @@ #include "miscadmin.h" #include "optimizer/optimizer.h" #include "pgstat.h" +#include "postmaster/autovacuum.h" #include "storage/ipc.h" #include "storage/predicate.h" #include "storage/spin.h" @@ -514,6 +515,11 @@ ReinitializeParallelDSM(ParallelContext *pcxt) { WaitForParallelWorkersToFinish(pcxt); WaitForParallelWorkersToExit(pcxt); + + /* Release all launched (i.e. reserved) parallel autovacuum workers. */ + if (AmAutoVacuumWorkerProcess()) + ParallelAutoVacuumReleaseWorkers(pcxt->nworkers_launched); + pcxt->nworkers_launched = 0; if (pcxt->known_attached_workers) { @@ -1002,6 +1008,11 @@ DestroyParallelContext(ParallelContext *pcxt) */ HOLD_INTERRUPTS(); WaitForParallelWorkersToExit(pcxt); + + /* Release all launched (i.e. reserved) parallel autovacuum workers. */ + if (AmAutoVacuumWorkerProcess()) + ParallelAutoVacuumReleaseWorkers(pcxt->nworkers_launched); + RESUME_INTERRUPTS(); /* Free the worker array itself. */ diff --git a/src/backend/commands/vacuumparallel.c b/src/backend/commands/vacuumparallel.c index 2b9d548cdeb..c63830fd2a5 100644 --- a/src/backend/commands/vacuumparallel.c +++ b/src/backend/commands/vacuumparallel.c @@ -1,16 +1,16 @@ /*------------------------------------------------------------------------- * * vacuumparallel.c - * Support routines for parallel vacuum execution. + * Support routines for parallel [auto]vacuum execution. * * This file contains routines that are intended to support setting up, using, * and tearing down a ParallelVacuumState. * - * In a parallel vacuum, we perform both index bulk deletion and index cleanup - * with parallel worker processes. Individual indexes are processed by one - * vacuum process. ParallelVacuumState contains shared information as well as - * the memory space for storing dead items allocated in the DSA area. We - * launch parallel worker processes at the start of parallel index + * In a parallel [auto]vacuum, we perform both index bulk deletion and index + * cleanup with parallel worker processes. Individual indexes are processed by + * one [auto]vacuum process. ParallelVacuumState contains shared information + * as well as the memory space for storing dead items allocated in the DSA area. + * We launch parallel worker processes at the start of parallel index * bulk-deletion and index cleanup and once all indexes are processed, the * parallel worker processes exit. Each time we process indexes in parallel, * the parallel context is re-initialized so that the same DSM can be used for @@ -34,6 +34,7 @@ #include "executor/instrument.h" #include "optimizer/paths.h" #include "pgstat.h" +#include "postmaster/autovacuum.h" #include "storage/bufmgr.h" #include "tcop/tcopprot.h" #include "utils/lsyscache.h" @@ -157,7 +158,8 @@ typedef struct PVIndStats } PVIndStats; /* - * Struct for maintaining a parallel vacuum state. typedef appears in vacuum.h. + * Struct for maintaining a parallel [auto]vacuum state. typedef appears in + * vacuum.h. */ struct ParallelVacuumState { @@ -371,10 +373,18 @@ parallel_vacuum_init(Relation rel, Relation *indrels, int nindexes, shared->relid = RelationGetRelid(rel); shared->elevel = elevel; shared->queryid = pgstat_get_my_query_id(); - shared->maintenance_work_mem_worker = - (nindexes_mwm > 0) ? - maintenance_work_mem / Min(parallel_workers, nindexes_mwm) : - maintenance_work_mem; + + if (AmAutoVacuumWorkerProcess()) + shared->maintenance_work_mem_worker = + (nindexes_mwm > 0) ? + autovacuum_work_mem / Min(parallel_workers, nindexes_mwm) : + autovacuum_work_mem; + else + shared->maintenance_work_mem_worker = + (nindexes_mwm > 0) ? + maintenance_work_mem / Min(parallel_workers, nindexes_mwm) : + maintenance_work_mem; + shared->dead_items_info.max_bytes = vac_work_mem * (size_t) 1024; /* Prepare DSA space for dead items */ @@ -541,7 +551,7 @@ parallel_vacuum_cleanup_all_indexes(ParallelVacuumState *pvs, long num_table_tup * * nrequested is the number of parallel workers that user requested. If * nrequested is 0, we compute the parallel degree based on nindexes, that is - * the number of indexes that support parallel vacuum. This function also + * the number of indexes that support parallel [auto]vacuum. This function also * sets will_parallel_vacuum to remember indexes that participate in parallel * vacuum. */ @@ -558,7 +568,9 @@ parallel_vacuum_compute_workers(Relation *indrels, int nindexes, int nrequested, * We don't allow performing parallel operation in standalone backend or * when parallelism is disabled. */ - if (!IsUnderPostmaster || max_parallel_maintenance_workers == 0) + if (!IsUnderPostmaster || + (max_parallel_autovacuum_workers == 0 && AmAutoVacuumWorkerProcess()) || + (max_parallel_maintenance_workers == 0 && !AmAutoVacuumWorkerProcess())) return 0; /* @@ -597,15 +609,17 @@ parallel_vacuum_compute_workers(Relation *indrels, int nindexes, int nrequested, parallel_workers = (nrequested > 0) ? Min(nrequested, nindexes_parallel) : nindexes_parallel; - /* Cap by max_parallel_maintenance_workers */ - parallel_workers = Min(parallel_workers, max_parallel_maintenance_workers); + /* Cap by GUC variable */ + parallel_workers = AmAutoVacuumWorkerProcess() ? + Min(parallel_workers, max_parallel_autovacuum_workers) : + Min(parallel_workers, max_parallel_maintenance_workers); return parallel_workers; } /* * Perform index vacuum or index cleanup with parallel workers. This function - * must be used by the parallel vacuum leader process. + * must be used by the parallel [auto]vacuum leader process. */ static void parallel_vacuum_process_all_indexes(ParallelVacuumState *pvs, int num_index_scans, @@ -666,6 +680,10 @@ parallel_vacuum_process_all_indexes(ParallelVacuumState *pvs, int num_index_scan /* Reset the parallel index processing and progress counters */ pg_atomic_write_u32(&(pvs->shared->idx), 0); + /* Check how many workers can provide autovacuum. */ + if (AmAutoVacuumWorkerProcess() && nworkers > 0) + nworkers = ParallelAutoVacuumReserveWorkers(nworkers); + /* Setup the shared cost-based vacuum delay and launch workers */ if (nworkers > 0) { @@ -690,6 +708,16 @@ parallel_vacuum_process_all_indexes(ParallelVacuumState *pvs, int num_index_scan LaunchParallelWorkers(pvs->pcxt); + if (AmAutoVacuumWorkerProcess() && + pvs->pcxt->nworkers_launched < nworkers) + { + /* + * Tell autovacuum that we could not launch all the previously + * reserved workers. + */ + ParallelAutoVacuumReleaseWorkers(pvs->pcxt->nworkers_launched - nworkers); + } + if (pvs->pcxt->nworkers_launched > 0) { /* @@ -706,16 +734,16 @@ parallel_vacuum_process_all_indexes(ParallelVacuumState *pvs, int num_index_scan if (vacuum) ereport(pvs->shared->elevel, - (errmsg(ngettext("launched %d parallel vacuum worker for index vacuuming (planned: %d)", - "launched %d parallel vacuum workers for index vacuuming (planned: %d)", + (errmsg(ngettext("launched %d parallel %svacuum worker for index vacuuming (planned: %d)", + "launched %d parallel %svacuum workers for index vacuuming (planned: %d)", pvs->pcxt->nworkers_launched), - pvs->pcxt->nworkers_launched, nworkers))); + pvs->pcxt->nworkers_launched, AmAutoVacuumWorkerProcess() ? "auto" : "", nworkers))); else ereport(pvs->shared->elevel, - (errmsg(ngettext("launched %d parallel vacuum worker for index cleanup (planned: %d)", - "launched %d parallel vacuum workers for index cleanup (planned: %d)", + (errmsg(ngettext("launched %d parallel %svacuum worker for index cleanup (planned: %d)", + "launched %d parallel %svacuum workers for index cleanup (planned: %d)", pvs->pcxt->nworkers_launched), - pvs->pcxt->nworkers_launched, nworkers))); + pvs->pcxt->nworkers_launched, AmAutoVacuumWorkerProcess() ? "auto" : "", nworkers))); } /* Vacuum the indexes that can be processed by only leader process */ @@ -982,8 +1010,8 @@ parallel_vacuum_index_is_parallel_safe(Relation indrel, int num_index_scans, /* * Perform work within a launched parallel process. * - * Since parallel vacuum workers perform only index vacuum or index cleanup, - * we don't need to report progress information. + * Since parallel [auto]vacuum workers perform only index vacuum or index + * cleanup, we don't need to report progress information. */ void parallel_vacuum_main(dsm_segment *seg, shm_toc *toc) diff --git a/src/backend/postmaster/autovacuum.c b/src/backend/postmaster/autovacuum.c index 981be42e3af..7f34e202589 100644 --- a/src/backend/postmaster/autovacuum.c +++ b/src/backend/postmaster/autovacuum.c @@ -285,6 +285,7 @@ typedef struct AutoVacuumWorkItem * av_workItems work item array * av_nworkersForBalance the number of autovacuum workers to use when * calculating the per worker cost limit + * av_active_parallel_workers the number of active parallel autovacuum workers * * This struct is protected by AutovacuumLock, except for av_signal and parts * of the worker list (see above). @@ -299,6 +300,7 @@ typedef struct WorkerInfo av_startingWorker; AutoVacuumWorkItem av_workItems[NUM_WORKITEMS]; pg_atomic_uint32 av_nworkersForBalance; + uint32 av_active_parallel_workers; } AutoVacuumShmemStruct; static AutoVacuumShmemStruct *AutoVacuumShmem; @@ -2840,8 +2842,12 @@ table_recheck_autovac(Oid relid, HTAB *table_toast_map, */ tab->at_params.index_cleanup = VACOPTVALUE_UNSPECIFIED; tab->at_params.truncate = VACOPTVALUE_UNSPECIFIED; - /* As of now, we don't support parallel vacuum for autovacuum */ - tab->at_params.nworkers = -1; + + /* Decide whether we need to process indexes of table in parallel. */ + tab->at_params.nworkers = avopts + ? avopts->parallel_autovacuum_workers + : -1; + tab->at_params.freeze_min_age = freeze_min_age; tab->at_params.freeze_table_age = freeze_table_age; tab->at_params.multixact_freeze_min_age = multixact_freeze_min_age; @@ -3322,6 +3328,61 @@ AutoVacuumRequestWork(AutoVacuumWorkItemType type, Oid relationId, return result; } +/* + * In order to meet the 'max_parallel_autovacuum_workers' limit, leader worker + * must call this function. It returns the number of parallel workers that + * actually can be launched and reserves (if any) these workers in global + * autovacuum state. + * + * NOTE: We will try to provide as many workers as requested, even if caller + * will occupy all available workers. + */ +int +ParallelAutoVacuumReserveWorkers(int nworkers) +{ + int can_launch; + + /* Only leader worker can call this function. */ + Assert(AmAutoVacuumWorkerProcess() && !IsParallelWorker()); + + LWLockAcquire(AutovacuumLock, LW_EXCLUSIVE); + if (AutoVacuumShmem->av_active_parallel_workers < nworkers) + { + /* Provide as many workers as we can. */ + can_launch = AutoVacuumShmem->av_active_parallel_workers; + AutoVacuumShmem->av_active_parallel_workers = 0; + } + else + { + /* OK, we can provide all requested workers. */ + can_launch = nworkers; + AutoVacuumShmem->av_active_parallel_workers -= nworkers; + } + LWLockRelease(AutovacuumLock); + + return can_launch; +} + +/* + * When parallel autovacuum worker die, leader worker must call this function + * in order to refresh global autovacuum state. Thus, other leaders will be able + * to use these workers. + * + * 'nworkers' - how many workers caller wants to release. + */ +void +ParallelAutoVacuumReleaseWorkers(int nworkers) +{ + /* Only leader worker can call this function. */ + Assert(AmAutoVacuumWorkerProcess() && !IsParallelWorker()); + + LWLockAcquire(AutovacuumLock, LW_EXCLUSIVE); + AutoVacuumShmem->av_active_parallel_workers += nworkers; + Assert(AutoVacuumShmem->av_active_parallel_workers <= + max_parallel_autovacuum_workers); + LWLockRelease(AutovacuumLock); +} + /* * autovac_init * This is called at postmaster initialization. @@ -3382,6 +3443,8 @@ AutoVacuumShmemInit(void) Assert(!found); AutoVacuumShmem->av_launcherpid = 0; + AutoVacuumShmem->av_active_parallel_workers = + max_parallel_autovacuum_workers; dclist_init(&AutoVacuumShmem->av_freeWorkers); dlist_init(&AutoVacuumShmem->av_runningWorkers); AutoVacuumShmem->av_startingWorker = NULL; @@ -3432,6 +3495,15 @@ check_autovacuum_work_mem(int *newval, void **extra, GucSource source) return true; } +bool +check_max_parallel_autovacuum_workers(int *newval, void **extra, + GucSource source) +{ + if (*newval >= max_worker_processes) + return false; + return true; +} + /* * Returns whether there is a free autovacuum worker slot available. */ diff --git a/src/backend/utils/init/globals.c b/src/backend/utils/init/globals.c index d31cb45a058..40a92ceecd5 100644 --- a/src/backend/utils/init/globals.c +++ b/src/backend/utils/init/globals.c @@ -143,6 +143,7 @@ int NBuffers = 16384; int MaxConnections = 100; int max_worker_processes = 8; int max_parallel_workers = 8; +int max_parallel_autovacuum_workers = 0; int MaxBackends = 0; /* GUC parameters for vacuum */ diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c index 2f8cbd86759..950b4300100 100644 --- a/src/backend/utils/misc/guc_tables.c +++ b/src/backend/utils/misc/guc_tables.c @@ -3604,6 +3604,16 @@ struct config_int ConfigureNamesInt[] = NULL, NULL, NULL }, + { + {"max_parallel_autovacuum_workers", PGC_POSTMASTER, RESOURCES_WORKER_PROCESSES, + gettext_noop("Maximum number of parallel autovacuum workers, that can be taken from bgworkers pool."), + gettext_noop("This parameter is capped by \"max_worker_processes\" (not by \"autovacuum_max_workers\"!)."), + }, + &max_parallel_autovacuum_workers, + 0, 0, MAX_BACKENDS, + check_max_parallel_autovacuum_workers, NULL, NULL + }, + { {"max_parallel_maintenance_workers", PGC_USERSET, RESOURCES_WORKER_PROCESSES, gettext_noop("Sets the maximum number of parallel processes per maintenance operation."), diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample index 63f991c4f93..23f5c890f78 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -221,6 +221,8 @@ #max_parallel_maintenance_workers = 2 # limited by max_parallel_workers #max_parallel_workers = 8 # number of max_worker_processes that # can be used in parallel operations +#max_parallel_autovacuum_workers = 0 # disabled by default and limited by max_parallel_workers + # (change requires restart) #parallel_leader_participation = on diff --git a/src/include/miscadmin.h b/src/include/miscadmin.h index 1bef98471c3..7c3575b6849 100644 --- a/src/include/miscadmin.h +++ b/src/include/miscadmin.h @@ -177,6 +177,7 @@ extern PGDLLIMPORT int MaxBackends; extern PGDLLIMPORT int MaxConnections; extern PGDLLIMPORT int max_worker_processes; extern PGDLLIMPORT int max_parallel_workers; +extern PGDLLIMPORT int max_parallel_autovacuum_workers; extern PGDLLIMPORT int commit_timestamp_buffers; extern PGDLLIMPORT int multixact_member_buffers; diff --git a/src/include/postmaster/autovacuum.h b/src/include/postmaster/autovacuum.h index e8135f41a1c..b5763e6ac36 100644 --- a/src/include/postmaster/autovacuum.h +++ b/src/include/postmaster/autovacuum.h @@ -64,6 +64,10 @@ pg_noreturn extern void AutoVacWorkerMain(const void *startup_data, size_t start extern bool AutoVacuumRequestWork(AutoVacuumWorkItemType type, Oid relationId, BlockNumber blkno); +/* parallel autovacuum stuff */ +extern int ParallelAutoVacuumReserveWorkers(int nworkers); +extern void ParallelAutoVacuumReleaseWorkers(int nworkers); + /* shared memory stuff */ extern Size AutoVacuumShmemSize(void); extern void AutoVacuumShmemInit(void); diff --git a/src/include/utils/guc_hooks.h b/src/include/utils/guc_hooks.h index 799fa7ace68..d4e6170d45c 100644 --- a/src/include/utils/guc_hooks.h +++ b/src/include/utils/guc_hooks.h @@ -31,6 +31,8 @@ extern void assign_application_name(const char *newval, void *extra); extern const char *show_archive_command(void); extern bool check_autovacuum_work_mem(int *newval, void **extra, GucSource source); +extern bool check_max_parallel_autovacuum_workers(int *newval, void **extra, + GucSource source); extern bool check_vacuum_buffer_usage_limit(int *newval, void **extra, GucSource source); extern bool check_backtrace_functions(char **newval, void **extra, diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h index b552359915f..16091e6a773 100644 --- a/src/include/utils/rel.h +++ b/src/include/utils/rel.h @@ -311,6 +311,8 @@ typedef struct ForeignKeyCacheInfo typedef struct AutoVacOpts { bool enabled; + int parallel_autovacuum_workers; /* max number of parallel + autovacuum workers */ int vacuum_threshold; int vacuum_max_threshold; int vacuum_ins_threshold; @@ -409,6 +411,16 @@ typedef struct StdRdOptions ((relation)->rd_options ? \ ((StdRdOptions *) (relation)->rd_options)->parallel_workers : (defaultpw)) +/* + * RelationGetParallelAutovacuumWorkers + * Returns the relation's parallel_autovacuum_workers reloption setting. + * Note multiple eval of argument! + */ +#define RelationGetParallelAutovacuumWorkers(relation, defaultpw) \ + ((relation)->rd_options ? \ + ((StdRdOptions *) (relation)->rd_options)->autovacuum.parallel_autovacuum_workers : \ + (defaultpw)) + /* ViewOptions->check_option values */ typedef enum ViewOptCheckOption { -- 2.43.0