From f90a43e0a6297511144f2b31b322e508bcdc3369 Mon Sep 17 00:00:00 2001 From: Masahiko Sawada Date: Thu, 2 Apr 2026 02:07:23 -0700 Subject: [PATCH] POC: use proc signals to propagate the shared vacuum delay parameters. --- src/backend/access/heap/vacuumlazy.c | 12 -- src/backend/access/transam/parallel.c | 20 +++ src/backend/commands/vacuum.c | 38 +++-- src/backend/commands/vacuumparallel.c | 146 ++++++++---------- src/backend/storage/ipc/procsignal.c | 4 + src/include/access/parallel.h | 2 + src/include/commands/vacuum.h | 5 + src/include/storage/procsignal.h | 4 +- src/test/modules/test_autovacuum/Makefile | 4 - src/test/modules/test_autovacuum/meson.build | 3 - .../t/001_parallel_autovacuum.pl | 78 ---------- 11 files changed, 122 insertions(+), 194 deletions(-) diff --git a/src/backend/access/heap/vacuumlazy.c b/src/backend/access/heap/vacuumlazy.c index 39395aed0d5..88c71cd85b6 100644 --- a/src/backend/access/heap/vacuumlazy.c +++ b/src/backend/access/heap/vacuumlazy.c @@ -152,7 +152,6 @@ #include "storage/latch.h" #include "storage/lmgr.h" #include "storage/read_stream.h" -#include "utils/injection_point.h" #include "utils/lsyscache.h" #include "utils/pg_rusage.h" #include "utils/timestamp.h" @@ -863,17 +862,6 @@ heap_vacuum_rel(Relation rel, const VacuumParams *params, lazy_check_wraparound_failsafe(vacrel); dead_items_alloc(vacrel, params->nworkers); -#ifdef USE_INJECTION_POINTS - - /* - * Used by tests to pause before parallel vacuum is launched, allowing - * test code to modify configuration that the leader then propagates to - * workers. - */ - if (AmAutoVacuumWorkerProcess() && ParallelVacuumIsActive(vacrel)) - INJECTION_POINT("autovacuum-start-parallel-vacuum", NULL); -#endif - /* * Call lazy_scan_heap to perform all required heap pruning, index * vacuuming, and heap vacuuming (plus related processing) diff --git a/src/backend/access/transam/parallel.c b/src/backend/access/transam/parallel.c index ab1dfb30e73..a8ba09832fc 100644 --- a/src/backend/access/transam/parallel.c +++ b/src/backend/access/transam/parallel.c @@ -1035,6 +1035,26 @@ ParallelContextActive(void) return !dlist_is_empty(&pcxt_list); } +/* + * Send a signal to all workers. + */ +void +SendProcSignalToParallelWorkers(ParallelContext *pcxt, ProcSignalReason reason) +{ + for (int i = 0; i < pcxt->nworkers_launched; i++) + { + pid_t pid; + + if (pcxt->worker[i].error_mqh == NULL || + pcxt->worker[i].bgwhandle == NULL || + GetBackgroundWorkerPid(pcxt->worker[i].bgwhandle, + &pid) != BGWH_STARTED) + continue; + + (void) SendProcSignal(pid, reason, ParallelLeaderProcNumber); + } +} + /* * Handle receipt of an interrupt indicating a parallel worker message. * diff --git a/src/backend/commands/vacuum.c b/src/backend/commands/vacuum.c index 62d03c8e190..2bf53debc4e 100644 --- a/src/backend/commands/vacuum.c +++ b/src/backend/commands/vacuum.c @@ -621,6 +621,9 @@ vacuum(List *relations, const VacuumParams *params, BufferAccessStrategy bstrate VacuumSharedCostBalance = NULL; VacuumActiveNWorkers = NULL; + ParallelVacuumSharedCostParams = NULL; + CurrentParallelVacuumState = NULL; + /* * Loop to process each selected relation. */ @@ -2435,19 +2438,8 @@ vacuum_delay_point(bool is_analyze) /* Always check for interrupts */ CHECK_FOR_INTERRUPTS(); - if (InterruptPending) - return; - - if (IsParallelWorker()) - { - /* - * Update cost-based vacuum delay parameters for a parallel autovacuum - * worker if any changes are detected. - */ - parallel_vacuum_update_shared_delay_params(); - } - - if (!VacuumCostActive && !ConfigReloadPending) + if (InterruptPending || + (!VacuumCostActive && !ConfigReloadPending && !ParallelVacuumParamsUpdatePending)) return; /* @@ -2469,6 +2461,26 @@ vacuum_delay_point(bool is_analyze) parallel_vacuum_propagate_shared_delay_params(); } + /* + * Update cost-based vacuum delay parameters for a parallel autovacuum + * worker if requested from the autovacuum worker (the leader process). + */ + if (ParallelVacuumParamsUpdatePending && ParallelVacuumSharedCostParams != NULL) + { + Assert(IsParallelWorker()); + + ParallelVacuumParamsUpdatePending = false; + parallel_vacuum_update_shared_delay_params(); + + elog(DEBUG2, + "parallel autovacuum worker updated cost params: cost_limit=%d, cost_delay=%g, cost_page_miss=%d, cost_page_dirty=%d, cost_page_hit=%d", + vacuum_cost_limit, + vacuum_cost_delay, + VacuumCostPageMiss, + VacuumCostPageDirty, + VacuumCostPageHit); + } + /* * If we disabled cost-based delays after reloading the config file, * return. diff --git a/src/backend/commands/vacuumparallel.c b/src/backend/commands/vacuumparallel.c index bac3bd28214..c3d7ad2366e 100644 --- a/src/backend/commands/vacuumparallel.c +++ b/src/backend/commands/vacuumparallel.c @@ -21,9 +21,8 @@ * For parallel autovacuum, we need to propagate cost-based vacuum delay * parameters from the leader to its workers, as the leader's parameters can * change even while processing a table (e.g., due to a config reload). - * The PVSharedCostParams struct manages these parameters using a - * generation counter. Each parallel worker polls this shared state and - * refreshes its local delay parameters whenever a change is detected. + * After the autovacuum process updates any vacuum delay parameters, it + * notifies its parallel vacuum workers to refresh their local parameters. * * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group * Portions Copyright (c) 1994, Regents of the University of California @@ -46,7 +45,6 @@ #include "storage/bufmgr.h" #include "storage/proc.h" #include "tcop/tcopprot.h" -#include "utils/injection_point.h" #include "utils/lsyscache.h" #include "utils/rel.h" @@ -67,17 +65,6 @@ */ typedef struct PVSharedCostParams { - /* - * The generation counter is incremented by the leader process each time - * it updates the shared cost-based vacuum delay parameters. Parallel - * vacuum workers compare it with their local generation, - * shared_params_generation_local, to detect whether they need to refresh - * their local parameters. The generation starts from 1 so that a freshly - * started worker (whose local copy is 0) will always load the initial - * parameters on its first check. - */ - pg_atomic_uint32 generation; - slock_t mutex; /* protects all fields below */ /* Parameters to share with parallel workers */ @@ -271,16 +258,21 @@ struct ParallelVacuumState PVIndVacStatus status; }; -static PVSharedCostParams *pv_shared_cost_params = NULL; +/* + * ParallelVacuumSharedCostParams and CurrentParallelVacuumState are + * used only for an autovacuum worker to propagate the shared vacuum + * delay parameters to its parallel vacuum workers. + * + * CurrentParallelVacuumState is set only in the leader process (i.e., + * autovacuum worker processes). + */ +PVSharedCostParams *ParallelVacuumSharedCostParams = NULL; +ParallelVacuumState *CurrentParallelVacuumState = NULL; /* - * Worker-local copy of the last cost-parameter generation this worker has - * applied. Initialized to 0; since the leader initializes the shared - * generation counter to 1, the first call to - * parallel_vacuum_update_shared_delay_params() will always detect a - * mismatch and read the initial parameters from shared memory. + * Is there an update on the shared vacuum delay parameters? */ -static uint32 shared_params_generation_local = 0; +volatile sig_atomic_t ParallelVacuumParamsUpdatePending = false; static int parallel_vacuum_compute_workers(Relation *indrels, int nindexes, int nrequested, bool *will_parallel_vacuum); @@ -463,10 +455,10 @@ parallel_vacuum_init(Relation rel, Relation *indrels, int nindexes, if (shared->is_autovacuum) { parallel_vacuum_set_cost_parameters(&shared->cost_params); - pg_atomic_init_u32(&shared->cost_params.generation, 1); SpinLockInit(&shared->cost_params.mutex); - pv_shared_cost_params = &(shared->cost_params); + ParallelVacuumSharedCostParams = &(shared->cost_params); + CurrentParallelVacuumState = pvs; } shm_toc_insert(pcxt->toc, PARALLEL_VACUUM_KEY_SHARED, shared); @@ -535,7 +527,10 @@ parallel_vacuum_end(ParallelVacuumState *pvs, IndexBulkDeleteResult **istats) ExitParallelMode(); if (AmAutoVacuumWorkerProcess()) - pv_shared_cost_params = NULL; + { + ParallelVacuumSharedCostParams = NULL; + CurrentParallelVacuumState = NULL; + } pfree(pvs->will_parallel_vacuum); pfree(pvs); @@ -627,6 +622,23 @@ parallel_vacuum_set_cost_parameters(PVSharedCostParams *params) params->cost_page_miss = VacuumCostPageMiss; } +/* + * Handle receipt of an interrupt indicating an update of the shared vacuum + * delay parameters. + */ +void +HandleParallelVacuumParamsUpdate(void) +{ + if (ParallelVacuumSharedCostParams == NULL) + return; + + /* + * Update the shared delay parameters at the next vacuum_delay_point() + * time. + */ + ParallelVacuumParamsUpdatePending = true; +} + /* * Updates the cost-based vacuum delay parameters for parallel autovacuum * workers. @@ -636,40 +648,18 @@ parallel_vacuum_set_cost_parameters(PVSharedCostParams *params) void parallel_vacuum_update_shared_delay_params(void) { - uint32 params_generation; - Assert(IsParallelWorker()); + Assert(ParallelVacuumSharedCostParams); - /* Quick return if the worker is not running for the autovacuum */ - if (pv_shared_cost_params == NULL) - return; - - params_generation = pg_atomic_read_u32(&pv_shared_cost_params->generation); - Assert(shared_params_generation_local <= params_generation); - - /* Return if parameters had not changed in the leader */ - if (params_generation == shared_params_generation_local) - return; - - SpinLockAcquire(&pv_shared_cost_params->mutex); - VacuumCostDelay = pv_shared_cost_params->cost_delay; - VacuumCostLimit = pv_shared_cost_params->cost_limit; - VacuumCostPageDirty = pv_shared_cost_params->cost_page_dirty; - VacuumCostPageHit = pv_shared_cost_params->cost_page_hit; - VacuumCostPageMiss = pv_shared_cost_params->cost_page_miss; - SpinLockRelease(&pv_shared_cost_params->mutex); + SpinLockAcquire(&ParallelVacuumSharedCostParams->mutex); + VacuumCostDelay = ParallelVacuumSharedCostParams->cost_delay; + VacuumCostLimit = ParallelVacuumSharedCostParams->cost_limit; + VacuumCostPageDirty = ParallelVacuumSharedCostParams->cost_page_dirty; + VacuumCostPageHit = ParallelVacuumSharedCostParams->cost_page_hit; + VacuumCostPageMiss = ParallelVacuumSharedCostParams->cost_page_miss; + SpinLockRelease(&ParallelVacuumSharedCostParams->mutex); VacuumUpdateCosts(); - - shared_params_generation_local = params_generation; - - elog(DEBUG2, - "parallel autovacuum worker updated cost params: cost_limit=%d, cost_delay=%g, cost_page_miss=%d, cost_page_dirty=%d, cost_page_hit=%d", - vacuum_cost_limit, - vacuum_cost_delay, - VacuumCostPageMiss, - VacuumCostPageDirty, - VacuumCostPageHit); } /* @@ -685,30 +675,30 @@ parallel_vacuum_propagate_shared_delay_params(void) /* * Quick return if the leader process is not sharing the delay parameters. */ - if (pv_shared_cost_params == NULL) + if (ParallelVacuumSharedCostParams == NULL) return; /* * Check if any delay parameters have changed. We can read them without * locks as only the leader can modify them. */ - if (vacuum_cost_delay == pv_shared_cost_params->cost_delay && - vacuum_cost_limit == pv_shared_cost_params->cost_limit && - VacuumCostPageDirty == pv_shared_cost_params->cost_page_dirty && - VacuumCostPageHit == pv_shared_cost_params->cost_page_hit && - VacuumCostPageMiss == pv_shared_cost_params->cost_page_miss) + if (vacuum_cost_delay == ParallelVacuumSharedCostParams->cost_delay && + vacuum_cost_limit == ParallelVacuumSharedCostParams->cost_limit && + VacuumCostPageDirty == ParallelVacuumSharedCostParams->cost_page_dirty && + VacuumCostPageHit == ParallelVacuumSharedCostParams->cost_page_hit && + VacuumCostPageMiss == ParallelVacuumSharedCostParams->cost_page_miss) return; /* Update the shared delay parameters */ - SpinLockAcquire(&pv_shared_cost_params->mutex); - parallel_vacuum_set_cost_parameters(pv_shared_cost_params); - SpinLockRelease(&pv_shared_cost_params->mutex); + SpinLockAcquire(&ParallelVacuumSharedCostParams->mutex); + parallel_vacuum_set_cost_parameters(ParallelVacuumSharedCostParams); + SpinLockRelease(&ParallelVacuumSharedCostParams->mutex); - /* - * Increment the generation of the parameters, i.e. let parallel workers - * know that they should re-read shared cost params. - */ - pg_atomic_fetch_add_u32(&pv_shared_cost_params->generation, 1); + Assert(CurrentParallelVacuumState); + SendProcSignalToParallelWorkers(CurrentParallelVacuumState->pcxt, + PROCSIG_PARALLEL_VACUUM_PARAMS_UPDATE); + + elog(LOG, "XXX SEND PROCSIGS"); } /* @@ -912,16 +902,6 @@ parallel_vacuum_process_all_indexes(ParallelVacuumState *pvs, int num_index_scan pvs->pcxt->nworkers_launched, nworkers))); } -#ifdef USE_INJECTION_POINTS - - /* - * Used by tests to pause after workers are launched but before index - * vacuuming begins. - */ - if (AmAutoVacuumWorkerProcess() && nworkers > 0) - INJECTION_POINT("autovacuum-leader-before-indexes-processing", NULL); -#endif - /* Vacuum the indexes that can be processed by only leader process */ parallel_vacuum_process_unsafe_indexes(pvs); @@ -1262,10 +1242,10 @@ parallel_vacuum_main(dsm_segment *seg, shm_toc *toc) * Parallel autovacuum workers initialize cost-based delay parameters * from the leader's shared state rather than GUC defaults, because * the leader may have applied per-table or autovacuum-specific - * overrides. pv_shared_cost_params must be set before calling - * parallel_vacuum_update_shared_delay_params(). + * overrides. ParallelVacuumSharedCostParams must be set before + * calling parallel_vacuum_update_shared_delay_params(). */ - pv_shared_cost_params = &(shared->cost_params); + ParallelVacuumSharedCostParams = &(shared->cost_params); parallel_vacuum_update_shared_delay_params(); } else @@ -1327,7 +1307,7 @@ parallel_vacuum_main(dsm_segment *seg, shm_toc *toc) FreeAccessStrategy(pvs.bstrategy); if (shared->is_autovacuum) - pv_shared_cost_params = NULL; + ParallelVacuumSharedCostParams = NULL; } /* diff --git a/src/backend/storage/ipc/procsignal.c b/src/backend/storage/ipc/procsignal.c index 7e017c8d53b..bf4f17ca8b4 100644 --- a/src/backend/storage/ipc/procsignal.c +++ b/src/backend/storage/ipc/procsignal.c @@ -19,6 +19,7 @@ #include "access/parallel.h" #include "commands/async.h" +#include "commands/vacuum.h" #include "miscadmin.h" #include "pgstat.h" #include "port/pg_bitutils.h" @@ -703,6 +704,9 @@ procsignal_sigusr1_handler(SIGNAL_ARGS) if (CheckProcSignal(PROCSIG_RECOVERY_CONFLICT)) HandleRecoveryConflictInterrupt(); + if (CheckProcSignal(PROCSIG_PARALLEL_VACUUM_PARAMS_UPDATE)) + HandleParallelVacuumParamsUpdate(); + SetLatch(MyLatch); } diff --git a/src/include/access/parallel.h b/src/include/access/parallel.h index 60f857675e0..90084581f7c 100644 --- a/src/include/access/parallel.h +++ b/src/include/access/parallel.h @@ -19,6 +19,7 @@ #include "access/xlogdefs.h" #include "lib/ilist.h" #include "postmaster/bgworker.h" +#include "storage/procsignal.h" #include "storage/shm_mq.h" #include "storage/shm_toc.h" @@ -71,6 +72,7 @@ extern void WaitForParallelWorkersToAttach(ParallelContext *pcxt); extern void WaitForParallelWorkersToFinish(ParallelContext *pcxt); extern void DestroyParallelContext(ParallelContext *pcxt); extern bool ParallelContextActive(void); +extern void SendProcSignalToParallelWorkers(ParallelContext *pcxt, ProcSignalReason reason); extern void HandleParallelMessageInterrupt(void); extern void ProcessParallelMessages(void); diff --git a/src/include/commands/vacuum.h b/src/include/commands/vacuum.h index 69fec07491b..8b59e3ae1c2 100644 --- a/src/include/commands/vacuum.h +++ b/src/include/commands/vacuum.h @@ -360,6 +360,10 @@ extern PGDLLIMPORT int vacuum_cost_limit; extern PGDLLIMPORT int64 parallel_vacuum_worker_delay_ns; +extern PGDLLIMPORT struct PVSharedCostParams *ParallelVacuumSharedCostParams; +extern PGDLLIMPORT struct ParallelVacuumState *CurrentParallelVacuumState; +extern PGDLLIMPORT volatile sig_atomic_t ParallelVacuumParamsUpdatePending; + /* in commands/vacuum.c */ extern void ExecVacuum(ParseState *pstate, VacuumStmt *vacstmt, bool isTopLevel); extern void vacuum(List *relations, const VacuumParams *params, @@ -425,6 +429,7 @@ extern void parallel_vacuum_cleanup_all_indexes(ParallelVacuumState *pvs, extern void parallel_vacuum_update_shared_delay_params(void); extern void parallel_vacuum_propagate_shared_delay_params(void); extern void parallel_vacuum_main(dsm_segment *seg, shm_toc *toc); +extern void HandleParallelVacuumParamsUpdate(void); /* in commands/analyze.c */ extern void analyze_rel(Oid relid, RangeVar *relation, diff --git a/src/include/storage/procsignal.h b/src/include/storage/procsignal.h index 348fba53a93..ac9a6e841fd 100644 --- a/src/include/storage/procsignal.h +++ b/src/include/storage/procsignal.h @@ -39,9 +39,11 @@ typedef enum PROCSIG_RECOVERY_CONFLICT, /* backend is blocking recovery, check * PGPROC->pendingRecoveryConflicts for the * reason */ + PROCSIG_PARALLEL_VACUUM_PARAMS_UPDATE, /* parallel autovacuum worker: + * reload cost params */ } ProcSignalReason; -#define NUM_PROCSIGNALS (PROCSIG_RECOVERY_CONFLICT + 1) +#define NUM_PROCSIGNALS (PROCSIG_PARALLEL_VACUUM_PARAMS_UPDATE + 1) typedef enum { diff --git a/src/test/modules/test_autovacuum/Makefile b/src/test/modules/test_autovacuum/Makefile index 188ec9f96a2..d91a3405e34 100644 --- a/src/test/modules/test_autovacuum/Makefile +++ b/src/test/modules/test_autovacuum/Makefile @@ -4,10 +4,6 @@ PGFILEDESC = "test_autovacuum - test code for parallel autovacuum" TAP_TESTS = 1 -EXTRA_INSTALL = src/test/modules/injection_points - -export enable_injection_points - ifdef USE_PGXS PG_CONFIG = pg_config PGXS := $(shell $(PG_CONFIG) --pgxs) diff --git a/src/test/modules/test_autovacuum/meson.build b/src/test/modules/test_autovacuum/meson.build index 86e392bc0de..94a3e8a038d 100644 --- a/src/test/modules/test_autovacuum/meson.build +++ b/src/test/modules/test_autovacuum/meson.build @@ -5,9 +5,6 @@ tests += { 'sd': meson.current_source_dir(), 'bd': meson.current_build_dir(), 'tap': { - 'env': { - 'enable_injection_points': get_option('injection_points') ? 'yes' : 'no', - }, 'tests': [ 't/001_parallel_autovacuum.pl', ], diff --git a/src/test/modules/test_autovacuum/t/001_parallel_autovacuum.pl b/src/test/modules/test_autovacuum/t/001_parallel_autovacuum.pl index 9e65eafb549..6536a3b265a 100644 --- a/src/test/modules/test_autovacuum/t/001_parallel_autovacuum.pl +++ b/src/test/modules/test_autovacuum/t/001_parallel_autovacuum.pl @@ -9,11 +9,6 @@ use PostgreSQL::Test::Cluster; use PostgreSQL::Test::Utils; use Test::More; -if ($ENV{enable_injection_points} ne 'yes') -{ - plan skip_all => 'Injection points not supported by this build'; -} - # Before each test we should disable autovacuum for 'test_autovac' table and # generate some dead tuples in it. Returns the current autovacuum_count of # the table test_autovac. @@ -66,20 +61,6 @@ log_autovacuum_min_duration = -1 }); $node->start; -# Check if the extension injection_points is available, as it may be -# possible that this script is run with installcheck, where the module -# would not be installed by default. -if (!$node->check_extension('injection_points')) -{ - plan skip_all => 'Extension injection_points not installed'; -} - -# Create all functions needed for testing -$node->safe_psql( - 'postgres', qq{ - CREATE EXTENSION injection_points; -}); - my $indexes_num = 3; my $initial_rows_num = 10_000; my $autovacuum_parallel_workers = 2; @@ -134,64 +115,5 @@ ok( $node->log_contains( qr/parallel workers: index vacuum: 2 planned, 2 launched in total/, $log_offset)); -# Test 2: -# Check whether parallel autovacuum leader can propagate cost-based parameters -# to the parallel workers. - -$av_count = prepare_for_next_test($node, 2); -$log_offset = -s $node->logfile; - -$node->safe_psql( - 'postgres', qq{ - SELECT injection_points_attach('autovacuum-start-parallel-vacuum', 'wait'); - SELECT injection_points_attach('autovacuum-leader-before-indexes-processing', 'wait'); - - ALTER TABLE test_autovac SET (autovacuum_parallel_workers = 1, autovacuum_enabled = true); -}); - -# Wait until parallel autovacuum is inited -$node->wait_for_event('autovacuum worker', - 'autovacuum-start-parallel-vacuum'); - -# Update the shared cost-based delay parameters. -$node->safe_psql( - 'postgres', qq{ - ALTER SYSTEM SET vacuum_cost_limit = 500; - ALTER SYSTEM SET vacuum_cost_page_miss = 10; - ALTER SYSTEM SET vacuum_cost_page_dirty = 10; - ALTER SYSTEM SET vacuum_cost_page_hit = 10; - SELECT pg_reload_conf(); -}); - -# Resume the leader process to update the shared parameters during heap scan (i.e. -# vacuum_delay_point() is called) and launch a parallel vacuum worker, but it stops -# before vacuuming indexes due to the injection point. -$node->safe_psql( - 'postgres', qq{ - SELECT injection_points_wakeup('autovacuum-start-parallel-vacuum'); -}); -$node->wait_for_event('autovacuum worker', - 'autovacuum-leader-before-indexes-processing'); - -# Check whether parallel worker successfully updated all parameters during -# index processing -$node->wait_for_log( - qr/parallel autovacuum worker updated cost params: cost_limit=500, cost_delay=2, cost_page_miss=10, cost_page_dirty=10, cost_page_hit=10/, - $log_offset); - -$node->safe_psql( - 'postgres', qq{ - SELECT injection_points_wakeup('autovacuum-leader-before-indexes-processing'); -}); - -wait_for_autovacuum_complete($node, $av_count); - -# Cleanup -$node->safe_psql( - 'postgres', qq{ - SELECT injection_points_detach('autovacuum-start-parallel-vacuum'); - SELECT injection_points_detach('autovacuum-leader-before-indexes-processing'); -}); - $node->stop; done_testing(); -- 2.53.0