Branch data Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * vacuumparallel.c
4 : : * Support routines for parallel vacuum and autovacuum execution. In the
5 : : * comments below, the word "vacuum" will refer to both vacuum and
6 : : * autovacuum.
7 : : *
8 : : * This file contains routines that are intended to support setting up, using,
9 : : * and tearing down a ParallelVacuumState.
10 : : *
11 : : * In a parallel vacuum, we perform both index bulk deletion and index cleanup
12 : : * with parallel worker processes. Individual indexes are processed by one
13 : : * vacuum process. ParallelVacuumState contains shared information as well as
14 : : * the memory space for storing dead items allocated in the DSA area. We
15 : : * launch parallel worker processes at the start of parallel index
16 : : * bulk-deletion and index cleanup and once all indexes are processed, the
17 : : * parallel worker processes exit. Each time we process indexes in parallel,
18 : : * the parallel context is re-initialized so that the same DSM can be used for
19 : : * multiple passes of index bulk-deletion and index cleanup.
20 : : *
21 : : * For parallel autovacuum, we need to propagate cost-based vacuum delay
22 : : * parameters from the leader to its workers, as the leader's parameters can
23 : : * change even while processing a table (e.g., due to a config reload).
24 : : * The PVSharedCostParams struct manages these parameters using a
25 : : * generation counter. Each parallel worker polls this shared state and
26 : : * refreshes its local delay parameters whenever a change is detected.
27 : : *
28 : : * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
29 : : * Portions Copyright (c) 1994, Regents of the University of California
30 : : *
31 : : * IDENTIFICATION
32 : : * src/backend/commands/vacuumparallel.c
33 : : *
34 : : *-------------------------------------------------------------------------
35 : : */
36 : : #include "postgres.h"
37 : :
38 : : #include "access/amapi.h"
39 : : #include "access/table.h"
40 : : #include "access/xact.h"
41 : : #include "commands/progress.h"
42 : : #include "commands/vacuum.h"
43 : : #include "executor/instrument.h"
44 : : #include "optimizer/paths.h"
45 : : #include "pgstat.h"
46 : : #include "postmaster/interrupt.h"
47 : : #include "storage/bufmgr.h"
48 : : #include "storage/proc.h"
49 : : #include "tcop/tcopprot.h"
50 : : #include "utils/injection_point.h"
51 : : #include "utils/lsyscache.h"
52 : : #include "utils/rel.h"
53 : :
54 : : /*
55 : : * DSM keys for parallel vacuum. Unlike other parallel execution code, since
56 : : * we don't need to worry about DSM keys conflicting with plan_node_id we can
57 : : * use small integers.
58 : : */
59 : : #define PARALLEL_VACUUM_KEY_SHARED 1
60 : : #define PARALLEL_VACUUM_KEY_QUERY_TEXT 2
61 : : #define PARALLEL_VACUUM_KEY_BUFFER_USAGE 3
62 : : #define PARALLEL_VACUUM_KEY_WAL_USAGE 4
63 : : #define PARALLEL_VACUUM_KEY_INDEX_STATS 5
64 : :
65 : : /*
66 : : * Struct for cost-based vacuum delay related parameters to share among an
67 : : * autovacuum worker and its parallel vacuum workers.
68 : : */
69 : : typedef struct PVSharedCostParams
70 : : {
71 : : /*
72 : : * The generation counter is incremented by the leader process each time
73 : : * it updates the shared cost-based vacuum delay parameters. Parallel
74 : : * vacuum workers compare it with their local generation,
75 : : * shared_params_generation_local, to detect whether they need to refresh
76 : : * their local parameters. The generation starts from 1 so that a freshly
77 : : * started worker (whose local copy is 0) will always load the initial
78 : : * parameters on its first check.
79 : : */
80 : : pg_atomic_uint32 generation;
81 : :
82 : : slock_t mutex; /* protects all fields below */
83 : :
84 : : /* Parameters to share with parallel workers */
85 : : double cost_delay;
86 : : int cost_limit;
87 : : int cost_page_dirty;
88 : : int cost_page_hit;
89 : : int cost_page_miss;
90 : : bool track_cost_delay_timing;
91 : : } PVSharedCostParams;
92 : :
93 : : /*
94 : : * Shared information among parallel workers. So this is allocated in the DSM
95 : : * segment.
96 : : */
97 : : typedef struct PVShared
98 : : {
99 : : /*
100 : : * Target table relid, log level (for messages about parallel workers
101 : : * launched during VACUUM VERBOSE) and query ID. These fields are not
102 : : * modified during the parallel vacuum.
103 : : */
104 : : Oid relid;
105 : : int elevel;
106 : : int64 queryid;
107 : :
108 : : /*
109 : : * Fields for both index vacuum and cleanup.
110 : : *
111 : : * reltuples is the total number of input heap tuples. We set either old
112 : : * live tuples in the index vacuum case or the new live tuples in the
113 : : * index cleanup case.
114 : : *
115 : : * estimated_count is true if reltuples is an estimated value. (Note that
116 : : * reltuples could be -1 in this case, indicating we have no idea.)
117 : : */
118 : : double reltuples;
119 : : bool estimated_count;
120 : :
121 : : /*
122 : : * In single process vacuum we could consume more memory during index
123 : : * vacuuming or cleanup apart from the memory for heap scanning. In
124 : : * parallel vacuum, since individual vacuum workers can consume memory
125 : : * equal to maintenance_work_mem, the new maintenance_work_mem for each
126 : : * worker is set such that the parallel operation doesn't consume more
127 : : * memory than single process vacuum.
128 : : */
129 : : int maintenance_work_mem_worker;
130 : :
131 : : /*
132 : : * The number of buffers each worker's Buffer Access Strategy ring should
133 : : * contain.
134 : : */
135 : : int ring_nbuffers;
136 : :
137 : : /*
138 : : * Shared vacuum cost balance. During parallel vacuum,
139 : : * VacuumSharedCostBalance points to this value and it accumulates the
140 : : * balance of each parallel vacuum worker.
141 : : */
142 : : pg_atomic_uint32 cost_balance;
143 : :
144 : : /*
145 : : * Number of active parallel workers. This is used for computing the
146 : : * minimum threshold of the vacuum cost balance before a worker sleeps for
147 : : * cost-based delay.
148 : : */
149 : : pg_atomic_uint32 active_nworkers;
150 : :
151 : : /* Counter for vacuuming and cleanup */
152 : : pg_atomic_uint32 idx;
153 : :
154 : : /* DSA handle where the TidStore lives */
155 : : dsa_handle dead_items_dsa_handle;
156 : :
157 : : /* DSA pointer to the shared TidStore */
158 : : dsa_pointer dead_items_handle;
159 : :
160 : : /* Statistics of shared dead items */
161 : : VacDeadItemsInfo dead_items_info;
162 : :
163 : : /*
164 : : * If 'true' then we are running parallel autovacuum. Otherwise, we are
165 : : * running parallel maintenance VACUUM.
166 : : */
167 : : bool is_autovacuum;
168 : :
169 : : /*
170 : : * Cost-based vacuum delay parameters shared between the autovacuum leader
171 : : * and its parallel workers.
172 : : */
173 : : PVSharedCostParams cost_params;
174 : : } PVShared;
175 : :
176 : : /* Status used during parallel index vacuum or cleanup */
177 : : typedef enum PVIndVacStatus
178 : : {
179 : : PARALLEL_INDVAC_STATUS_INITIAL = 0,
180 : : PARALLEL_INDVAC_STATUS_NEED_BULKDELETE,
181 : : PARALLEL_INDVAC_STATUS_NEED_CLEANUP,
182 : : PARALLEL_INDVAC_STATUS_COMPLETED,
183 : : } PVIndVacStatus;
184 : :
185 : : /*
186 : : * Struct for index vacuum statistics of an index that is used for parallel vacuum.
187 : : * This includes the status of parallel index vacuum as well as index statistics.
188 : : */
189 : : typedef struct PVIndStats
190 : : {
191 : : /*
192 : : * The following two fields are set by leader process before executing
193 : : * parallel index vacuum or parallel index cleanup. These fields are not
194 : : * fixed for the entire VACUUM operation. They are only fixed for an
195 : : * individual parallel index vacuum and cleanup.
196 : : *
197 : : * parallel_workers_can_process is true if both leader and worker can
198 : : * process the index, otherwise only leader can process it.
199 : : */
200 : : PVIndVacStatus status;
201 : : bool parallel_workers_can_process;
202 : :
203 : : /*
204 : : * Individual worker or leader stores the result of index vacuum or
205 : : * cleanup.
206 : : */
207 : : bool istat_updated; /* are the stats updated? */
208 : : IndexBulkDeleteResult istat;
209 : : } PVIndStats;
210 : :
211 : : /*
212 : : * Struct for maintaining a parallel vacuum state. typedef appears in vacuum.h.
213 : : */
214 : : struct ParallelVacuumState
215 : : {
216 : : /* NULL for worker processes */
217 : : ParallelContext *pcxt;
218 : :
219 : : /* Parent Heap Relation */
220 : : Relation heaprel;
221 : :
222 : : /* Target indexes */
223 : : Relation *indrels;
224 : : int nindexes;
225 : :
226 : : /* Shared information among parallel vacuum workers */
227 : : PVShared *shared;
228 : :
229 : : /*
230 : : * Shared index statistics among parallel vacuum workers. The array
231 : : * element is allocated for every index, even those indexes where parallel
232 : : * index vacuuming is unsafe or not worthwhile (e.g.,
233 : : * will_parallel_vacuum[] is false). During parallel vacuum,
234 : : * IndexBulkDeleteResult of each index is kept in DSM and is copied into
235 : : * local memory at the end of parallel vacuum.
236 : : */
237 : : PVIndStats *indstats;
238 : :
239 : : /* Shared dead items space among parallel vacuum workers */
240 : : TidStore *dead_items;
241 : :
242 : : /* Points to buffer usage area in DSM */
243 : : BufferUsage *buffer_usage;
244 : :
245 : : /* Points to WAL usage area in DSM */
246 : : WalUsage *wal_usage;
247 : :
248 : : /*
249 : : * False if the index is totally unsuitable target for all parallel
250 : : * processing. For example, the index could be <
251 : : * min_parallel_index_scan_size cutoff.
252 : : */
253 : : bool *will_parallel_vacuum;
254 : :
255 : : /*
256 : : * The number of indexes that support parallel index bulk-deletion and
257 : : * parallel index cleanup respectively.
258 : : */
259 : : int nindexes_parallel_bulkdel;
260 : : int nindexes_parallel_cleanup;
261 : : int nindexes_parallel_condcleanup;
262 : :
263 : : /* Buffer access strategy used by leader process */
264 : : BufferAccessStrategy bstrategy;
265 : :
266 : : /*
267 : : * Error reporting state. The error callback is set only for workers
268 : : * processes during parallel index vacuum.
269 : : */
270 : : char *relnamespace;
271 : : char *relname;
272 : : char *indname;
273 : : PVIndVacStatus status;
274 : : };
275 : :
276 : : static PVSharedCostParams *pv_shared_cost_params = NULL;
277 : :
278 : : /*
279 : : * Worker-local copy of the last cost-parameter generation this worker has
280 : : * applied. Initialized to 0; since the leader initializes the shared
281 : : * generation counter to 1, the first call to
282 : : * parallel_vacuum_update_shared_delay_params() will always detect a
283 : : * mismatch and read the initial parameters from shared memory.
284 : : */
285 : : static uint32 shared_params_generation_local = 0;
286 : :
287 : : static int parallel_vacuum_compute_workers(Relation *indrels, int nindexes, int nrequested,
288 : : bool *will_parallel_vacuum);
289 : : static void parallel_vacuum_process_all_indexes(ParallelVacuumState *pvs, int num_index_scans,
290 : : bool vacuum, PVWorkerStats *wstats);
291 : : static void parallel_vacuum_process_safe_indexes(ParallelVacuumState *pvs);
292 : : static void parallel_vacuum_process_unsafe_indexes(ParallelVacuumState *pvs);
293 : : static void parallel_vacuum_process_one_index(ParallelVacuumState *pvs, Relation indrel,
294 : : PVIndStats *indstats);
295 : : static bool parallel_vacuum_index_is_parallel_safe(Relation indrel, int num_index_scans,
296 : : bool vacuum);
297 : : static void parallel_vacuum_error_callback(void *arg);
298 : : static inline void parallel_vacuum_set_cost_parameters(PVSharedCostParams *params);
299 : : static void parallel_vacuum_dsm_detach(dsm_segment *seg, Datum arg);
300 : :
301 : : /*
302 : : * Try to enter parallel mode and create a parallel context. Then initialize
303 : : * shared memory state.
304 : : *
305 : : * On success, return parallel vacuum state. Otherwise return NULL.
306 : : */
307 : : ParallelVacuumState *
308 : 42985 : parallel_vacuum_init(Relation rel, Relation *indrels, int nindexes,
309 : : int nrequested_workers, int vac_work_mem,
310 : : int elevel, BufferAccessStrategy bstrategy)
311 : : {
312 : : ParallelVacuumState *pvs;
313 : : ParallelContext *pcxt;
314 : : PVShared *shared;
315 : : TidStore *dead_items;
316 : : PVIndStats *indstats;
317 : : BufferUsage *buffer_usage;
318 : : WalUsage *wal_usage;
319 : : bool *will_parallel_vacuum;
320 : : Size est_indstats_len;
321 : : Size est_shared_len;
322 : 42985 : int nindexes_mwm = 0;
323 : 42985 : int parallel_workers = 0;
324 : : int querylen;
325 : :
326 : : /*
327 : : * A parallel vacuum must be requested and there must be indexes on the
328 : : * relation
329 : : */
330 : : Assert(nrequested_workers >= 0);
331 : : Assert(nindexes > 0);
332 : :
333 : : /*
334 : : * Compute the number of parallel vacuum workers to launch
335 : : */
336 : 42985 : will_parallel_vacuum = palloc0_array(bool, nindexes);
337 : 42985 : parallel_workers = parallel_vacuum_compute_workers(indrels, nindexes,
338 : : nrequested_workers,
339 : : will_parallel_vacuum);
340 [ + + ]: 42985 : if (parallel_workers <= 0)
341 : : {
342 : : /* Can't perform vacuum in parallel -- return NULL */
343 : 42958 : pfree(will_parallel_vacuum);
344 : 42958 : return NULL;
345 : : }
346 : :
347 : 27 : pvs = palloc0_object(ParallelVacuumState);
348 : 27 : pvs->indrels = indrels;
349 : 27 : pvs->nindexes = nindexes;
350 : 27 : pvs->will_parallel_vacuum = will_parallel_vacuum;
351 : 27 : pvs->bstrategy = bstrategy;
352 : 27 : pvs->heaprel = rel;
353 : :
354 : 27 : EnterParallelMode();
355 : 27 : pcxt = CreateParallelContext("postgres", "parallel_vacuum_main",
356 : : parallel_workers);
357 : : Assert(pcxt->nworkers > 0);
358 : 27 : pvs->pcxt = pcxt;
359 : :
360 : : /* Estimate size for index vacuum stats -- PARALLEL_VACUUM_KEY_INDEX_STATS */
361 : 27 : est_indstats_len = mul_size(sizeof(PVIndStats), nindexes);
362 : 27 : shm_toc_estimate_chunk(&pcxt->estimator, est_indstats_len);
363 : 27 : shm_toc_estimate_keys(&pcxt->estimator, 1);
364 : :
365 : : /* Estimate size for shared information -- PARALLEL_VACUUM_KEY_SHARED */
366 : 27 : est_shared_len = sizeof(PVShared);
367 : 27 : shm_toc_estimate_chunk(&pcxt->estimator, est_shared_len);
368 : 27 : shm_toc_estimate_keys(&pcxt->estimator, 1);
369 : :
370 : : /*
371 : : * Estimate space for BufferUsage and WalUsage --
372 : : * PARALLEL_VACUUM_KEY_BUFFER_USAGE and PARALLEL_VACUUM_KEY_WAL_USAGE.
373 : : *
374 : : * If there are no extensions loaded that care, we could skip this. We
375 : : * have no way of knowing whether anyone's looking at pgBufferUsage or
376 : : * pgWalUsage, so do it unconditionally.
377 : : */
378 : 27 : shm_toc_estimate_chunk(&pcxt->estimator,
379 : : mul_size(sizeof(BufferUsage), pcxt->nworkers));
380 : 27 : shm_toc_estimate_keys(&pcxt->estimator, 1);
381 : 27 : shm_toc_estimate_chunk(&pcxt->estimator,
382 : : mul_size(sizeof(WalUsage), pcxt->nworkers));
383 : 27 : shm_toc_estimate_keys(&pcxt->estimator, 1);
384 : :
385 : : /* Finally, estimate PARALLEL_VACUUM_KEY_QUERY_TEXT space */
386 [ + + ]: 27 : if (debug_query_string)
387 : : {
388 : 23 : querylen = strlen(debug_query_string);
389 : 23 : shm_toc_estimate_chunk(&pcxt->estimator, querylen + 1);
390 : 23 : shm_toc_estimate_keys(&pcxt->estimator, 1);
391 : : }
392 : : else
393 : 4 : querylen = 0; /* keep compiler quiet */
394 : :
395 : 27 : InitializeParallelDSM(pcxt);
396 : :
397 : : /* Prepare index vacuum stats */
398 : 27 : indstats = (PVIndStats *) shm_toc_allocate(pcxt->toc, est_indstats_len);
399 [ + - + - : 603 : MemSet(indstats, 0, est_indstats_len);
+ - + - +
+ ]
400 [ + + ]: 123 : for (int i = 0; i < nindexes; i++)
401 : : {
402 : 96 : Relation indrel = indrels[i];
403 : 96 : uint8 vacoptions = indrel->rd_indam->amparallelvacuumoptions;
404 : :
405 : : /*
406 : : * Cleanup option should be either disabled, always performing in
407 : : * parallel or conditionally performing in parallel.
408 : : */
409 : : Assert(((vacoptions & VACUUM_OPTION_PARALLEL_CLEANUP) == 0) ||
410 : : ((vacoptions & VACUUM_OPTION_PARALLEL_COND_CLEANUP) == 0));
411 : : Assert(vacoptions <= VACUUM_OPTION_MAX_VALID_VALUE);
412 : :
413 [ + + ]: 96 : if (!will_parallel_vacuum[i])
414 : 4 : continue;
415 : :
416 [ + + ]: 92 : if (indrel->rd_indam->amusemaintenanceworkmem)
417 : 8 : nindexes_mwm++;
418 : :
419 : : /*
420 : : * Remember the number of indexes that support parallel operation for
421 : : * each phase.
422 : : */
423 [ + + ]: 92 : if ((vacoptions & VACUUM_OPTION_PARALLEL_BULKDEL) != 0)
424 : 84 : pvs->nindexes_parallel_bulkdel++;
425 [ + + ]: 92 : if ((vacoptions & VACUUM_OPTION_PARALLEL_CLEANUP) != 0)
426 : 16 : pvs->nindexes_parallel_cleanup++;
427 [ + + ]: 92 : if ((vacoptions & VACUUM_OPTION_PARALLEL_COND_CLEANUP) != 0)
428 : 68 : pvs->nindexes_parallel_condcleanup++;
429 : : }
430 : 27 : shm_toc_insert(pcxt->toc, PARALLEL_VACUUM_KEY_INDEX_STATS, indstats);
431 : 27 : pvs->indstats = indstats;
432 : :
433 : : /* Prepare shared information */
434 : 27 : shared = (PVShared *) shm_toc_allocate(pcxt->toc, est_shared_len);
435 [ + - + - : 459 : MemSet(shared, 0, est_shared_len);
+ - + - +
+ ]
436 : 27 : shared->relid = RelationGetRelid(rel);
437 : 27 : shared->elevel = elevel;
438 : 27 : shared->queryid = pgstat_get_my_query_id();
439 : 27 : shared->maintenance_work_mem_worker =
440 : : (nindexes_mwm > 0) ?
441 [ + + ]: 27 : vac_work_mem / Min(parallel_workers, nindexes_mwm) :
442 : : vac_work_mem;
443 : :
444 : 27 : shared->dead_items_info.max_bytes = vac_work_mem * (size_t) 1024;
445 : :
446 : : /* Prepare DSA space for dead items */
447 : 27 : dead_items = TidStoreCreateShared(shared->dead_items_info.max_bytes,
448 : : LWTRANCHE_PARALLEL_VACUUM_DSA);
449 : 27 : pvs->dead_items = dead_items;
450 : 27 : shared->dead_items_handle = TidStoreGetHandle(dead_items);
451 : 27 : shared->dead_items_dsa_handle = dsa_get_handle(TidStoreGetDSA(dead_items));
452 : :
453 : : /* Use the same buffer size for all workers */
454 : 27 : shared->ring_nbuffers = GetAccessStrategyBufferCount(bstrategy);
455 : :
456 : 27 : pg_atomic_init_u32(&(shared->cost_balance), 0);
457 : 27 : pg_atomic_init_u32(&(shared->active_nworkers), 0);
458 : 27 : pg_atomic_init_u32(&(shared->idx), 0);
459 : :
460 : 27 : shared->is_autovacuum = AmAutoVacuumWorkerProcess();
461 : :
462 : : /*
463 : : * Initialize shared cost-based vacuum delay parameters if it's for
464 : : * autovacuum and the parallel context has workers. Note that the parallel
465 : : * context falls back to the leader's private memory with no workers and
466 : : * no segment when the maximum number of DSM segments has been reached.
467 : : * There are then no workers to propagate the parameters to, and no
468 : : * segment to register the detach callback on.
469 : : */
470 [ + + + - ]: 27 : if (shared->is_autovacuum && pcxt->nworkers > 0)
471 : : {
472 : 4 : parallel_vacuum_set_cost_parameters(&shared->cost_params);
473 : 4 : pg_atomic_init_u32(&shared->cost_params.generation, 1);
474 : 4 : SpinLockInit(&shared->cost_params.mutex);
475 : :
476 : 4 : pv_shared_cost_params = &(shared->cost_params);
477 : 4 : on_dsm_detach(pcxt->seg, parallel_vacuum_dsm_detach, (Datum) 0);
478 : : }
479 : :
480 : 27 : shm_toc_insert(pcxt->toc, PARALLEL_VACUUM_KEY_SHARED, shared);
481 : 27 : pvs->shared = shared;
482 : :
483 : : /*
484 : : * Allocate space for each worker's BufferUsage and WalUsage; no need to
485 : : * initialize
486 : : */
487 : 27 : buffer_usage = shm_toc_allocate(pcxt->toc,
488 : 27 : mul_size(sizeof(BufferUsage), pcxt->nworkers));
489 : 27 : shm_toc_insert(pcxt->toc, PARALLEL_VACUUM_KEY_BUFFER_USAGE, buffer_usage);
490 : 27 : pvs->buffer_usage = buffer_usage;
491 : 27 : wal_usage = shm_toc_allocate(pcxt->toc,
492 : 27 : mul_size(sizeof(WalUsage), pcxt->nworkers));
493 : 27 : shm_toc_insert(pcxt->toc, PARALLEL_VACUUM_KEY_WAL_USAGE, wal_usage);
494 : 27 : pvs->wal_usage = wal_usage;
495 : :
496 : : /* Store query string for workers */
497 [ + + ]: 27 : if (debug_query_string)
498 : : {
499 : : char *sharedquery;
500 : :
501 : 23 : sharedquery = (char *) shm_toc_allocate(pcxt->toc, querylen + 1);
502 : 23 : memcpy(sharedquery, debug_query_string, querylen + 1);
503 : 23 : sharedquery[querylen] = '\0';
504 : 23 : shm_toc_insert(pcxt->toc,
505 : : PARALLEL_VACUUM_KEY_QUERY_TEXT, sharedquery);
506 : : }
507 : :
508 : : /* Success -- return parallel vacuum state */
509 : 27 : return pvs;
510 : : }
511 : :
512 : : /*
513 : : * Destroy the parallel context, and end parallel mode.
514 : : *
515 : : * Since writes are not allowed during parallel mode, copy the
516 : : * updated index statistics from DSM into local memory and then later use that
517 : : * to update the index statistics. One might think that we can exit from
518 : : * parallel mode, update the index statistics and then destroy parallel
519 : : * context, but that won't be safe (see ExitParallelMode).
520 : : */
521 : : void
522 : 26 : parallel_vacuum_end(ParallelVacuumState *pvs, IndexBulkDeleteResult **istats)
523 : : {
524 : : Assert(!IsParallelWorker());
525 : :
526 : : /* Copy the updated statistics */
527 [ + + ]: 120 : for (int i = 0; i < pvs->nindexes; i++)
528 : : {
529 : 94 : PVIndStats *indstats = &(pvs->indstats[i]);
530 : :
531 [ + + ]: 94 : if (indstats->istat_updated)
532 : : {
533 : 66 : istats[i] = palloc0_object(IndexBulkDeleteResult);
534 : 66 : memcpy(istats[i], &indstats->istat, sizeof(IndexBulkDeleteResult));
535 : : }
536 : : else
537 : 28 : istats[i] = NULL;
538 : : }
539 : :
540 : 26 : TidStoreDestroy(pvs->dead_items);
541 : :
542 : 26 : DestroyParallelContext(pvs->pcxt);
543 : 26 : ExitParallelMode();
544 : :
545 [ + + ]: 26 : if (AmAutoVacuumWorkerProcess())
546 : 3 : pv_shared_cost_params = NULL;
547 : :
548 : 26 : pfree(pvs->will_parallel_vacuum);
549 : 26 : pfree(pvs);
550 : 26 : }
551 : :
552 : : /*
553 : : * DSM detach callback. This is invoked when an autovacuum worker detaches
554 : : * from the DSM segment holding PVShared. It ensures to reset the local pointer
555 : : * to the shared state even if parallel vacuum raises an error and doesn't
556 : : * call parallel_vacuum_end().
557 : : */
558 : : static void
559 : 4 : parallel_vacuum_dsm_detach(dsm_segment *seg, Datum arg)
560 : : {
561 : : Assert(AmAutoVacuumWorkerProcess());
562 : 4 : pv_shared_cost_params = NULL;
563 : 4 : }
564 : :
565 : : /*
566 : : * Returns the dead items space and dead items information.
567 : : */
568 : : TidStore *
569 : 47 : parallel_vacuum_get_dead_items(ParallelVacuumState *pvs, VacDeadItemsInfo **dead_items_info_p)
570 : : {
571 : 47 : *dead_items_info_p = &(pvs->shared->dead_items_info);
572 : 47 : return pvs->dead_items;
573 : : }
574 : :
575 : : /* Forget all items in dead_items */
576 : : void
577 : 20 : parallel_vacuum_reset_dead_items(ParallelVacuumState *pvs)
578 : : {
579 : 20 : VacDeadItemsInfo *dead_items_info = &(pvs->shared->dead_items_info);
580 : :
581 : : /*
582 : : * Free the current tidstore and return allocated DSA segments to the
583 : : * operating system. Then we recreate the tidstore with the same max_bytes
584 : : * limitation we just used.
585 : : */
586 : 20 : TidStoreDestroy(pvs->dead_items);
587 : 20 : pvs->dead_items = TidStoreCreateShared(dead_items_info->max_bytes,
588 : : LWTRANCHE_PARALLEL_VACUUM_DSA);
589 : :
590 : : /* Update the DSA pointer for dead_items to the new one */
591 : 20 : pvs->shared->dead_items_dsa_handle = dsa_get_handle(TidStoreGetDSA(pvs->dead_items));
592 : 20 : pvs->shared->dead_items_handle = TidStoreGetHandle(pvs->dead_items);
593 : :
594 : : /* Reset the counter */
595 : 20 : dead_items_info->num_items = 0;
596 : 20 : }
597 : :
598 : : /*
599 : : * Do parallel index bulk-deletion with parallel workers.
600 : : */
601 : : void
602 : 21 : parallel_vacuum_bulkdel_all_indexes(ParallelVacuumState *pvs, long num_table_tuples,
603 : : int num_index_scans, PVWorkerStats *wstats)
604 : : {
605 : : Assert(!IsParallelWorker());
606 : :
607 : : /*
608 : : * We can only provide an approximate value of num_heap_tuples, at least
609 : : * for now.
610 : : */
611 : 21 : pvs->shared->reltuples = num_table_tuples;
612 : 21 : pvs->shared->estimated_count = true;
613 : :
614 : 21 : parallel_vacuum_process_all_indexes(pvs, num_index_scans, true, wstats);
615 : 20 : }
616 : :
617 : : /*
618 : : * Do parallel index cleanup with parallel workers.
619 : : */
620 : : void
621 : 26 : parallel_vacuum_cleanup_all_indexes(ParallelVacuumState *pvs, long num_table_tuples,
622 : : int num_index_scans, bool estimated_count,
623 : : PVWorkerStats *wstats)
624 : : {
625 : : Assert(!IsParallelWorker());
626 : :
627 : : /*
628 : : * We can provide a better estimate of total number of surviving tuples
629 : : * (we assume indexes are more interested in that than in the number of
630 : : * nominally live tuples).
631 : : */
632 : 26 : pvs->shared->reltuples = num_table_tuples;
633 : 26 : pvs->shared->estimated_count = estimated_count;
634 : :
635 : 26 : parallel_vacuum_process_all_indexes(pvs, num_index_scans, false, wstats);
636 : 26 : }
637 : :
638 : : /*
639 : : * Fill in the given structure with cost-based vacuum delay parameter values.
640 : : */
641 : : static inline void
642 : 6 : parallel_vacuum_set_cost_parameters(PVSharedCostParams *params)
643 : : {
644 : 6 : params->cost_delay = vacuum_cost_delay;
645 : 6 : params->cost_limit = vacuum_cost_limit;
646 : 6 : params->cost_page_dirty = VacuumCostPageDirty;
647 : 6 : params->cost_page_hit = VacuumCostPageHit;
648 : 6 : params->cost_page_miss = VacuumCostPageMiss;
649 : 6 : params->track_cost_delay_timing = track_cost_delay_timing;
650 : 6 : }
651 : :
652 : : /*
653 : : * Updates the cost-based vacuum delay parameters for parallel autovacuum
654 : : * workers.
655 : : *
656 : : * For non-autovacuum parallel workers, this function will have no effect.
657 : : */
658 : : void
659 : 363 : parallel_vacuum_update_shared_delay_params(void)
660 : : {
661 : : uint32 params_generation;
662 : :
663 : : Assert(IsParallelWorker());
664 : :
665 : : /* Quick return if the worker is not running for the autovacuum */
666 [ - + ]: 363 : if (pv_shared_cost_params == NULL)
667 : 0 : return;
668 : :
669 : 363 : params_generation = pg_atomic_read_u32(&pv_shared_cost_params->generation);
670 : : Assert(shared_params_generation_local <= params_generation);
671 : :
672 : : /* Return if parameters had not changed in the leader */
673 [ + + ]: 363 : if (params_generation == shared_params_generation_local)
674 : 359 : return;
675 : :
676 : 4 : SpinLockAcquire(&pv_shared_cost_params->mutex);
677 : 4 : VacuumCostDelay = pv_shared_cost_params->cost_delay;
678 : 4 : VacuumCostLimit = pv_shared_cost_params->cost_limit;
679 : 4 : VacuumCostPageDirty = pv_shared_cost_params->cost_page_dirty;
680 : 4 : VacuumCostPageHit = pv_shared_cost_params->cost_page_hit;
681 : 4 : VacuumCostPageMiss = pv_shared_cost_params->cost_page_miss;
682 : 4 : track_cost_delay_timing = pv_shared_cost_params->track_cost_delay_timing;
683 : 4 : SpinLockRelease(&pv_shared_cost_params->mutex);
684 : :
685 : 4 : VacuumUpdateCosts();
686 : :
687 : 4 : shared_params_generation_local = params_generation;
688 : :
689 [ + - + + ]: 4 : elog(DEBUG2,
690 : : "parallel autovacuum worker updated cost params: cost_limit=%d, cost_delay=%g, cost_page_miss=%d, cost_page_dirty=%d, cost_page_hit=%d, track_cost_delay_timing=%s",
691 : : vacuum_cost_limit,
692 : : vacuum_cost_delay,
693 : : VacuumCostPageMiss,
694 : : VacuumCostPageDirty,
695 : : VacuumCostPageHit,
696 : : track_cost_delay_timing ? "yes" : "no");
697 : : }
698 : :
699 : : /*
700 : : * Store the cost-based vacuum delay parameters in the shared memory so that
701 : : * parallel vacuum workers can consume them (see
702 : : * parallel_vacuum_update_shared_delay_params()).
703 : : */
704 : : void
705 : 3737 : parallel_vacuum_propagate_shared_delay_params(void)
706 : : {
707 : : Assert(AmAutoVacuumWorkerProcess());
708 : :
709 : : /*
710 : : * Quick return if the leader process is not sharing the delay parameters.
711 : : */
712 [ + + ]: 3737 : if (pv_shared_cost_params == NULL)
713 : 3687 : return;
714 : :
715 : : /*
716 : : * Check if any delay parameters have changed. We can read them without
717 : : * locks as only the leader can modify them.
718 : : */
719 [ + + ]: 50 : if (vacuum_cost_delay == pv_shared_cost_params->cost_delay &&
720 [ + + ]: 49 : vacuum_cost_limit == pv_shared_cost_params->cost_limit &&
721 [ + - ]: 48 : VacuumCostPageDirty == pv_shared_cost_params->cost_page_dirty &&
722 [ + - ]: 48 : VacuumCostPageHit == pv_shared_cost_params->cost_page_hit &&
723 [ + - ]: 48 : VacuumCostPageMiss == pv_shared_cost_params->cost_page_miss &&
724 [ + - ]: 48 : track_cost_delay_timing == pv_shared_cost_params->track_cost_delay_timing)
725 : 48 : return;
726 : :
727 : : /* Update the shared delay parameters */
728 : 2 : SpinLockAcquire(&pv_shared_cost_params->mutex);
729 : 2 : parallel_vacuum_set_cost_parameters(pv_shared_cost_params);
730 : 2 : SpinLockRelease(&pv_shared_cost_params->mutex);
731 : :
732 : : /*
733 : : * Increment the generation of the parameters, i.e. let parallel workers
734 : : * know that they should re-read shared cost params.
735 : : */
736 : 2 : pg_atomic_fetch_add_u32(&pv_shared_cost_params->generation, 1);
737 : : }
738 : :
739 : : /*
740 : : * Reload the configuration file if requested, and refresh the cost-based
741 : : * delay parameters of an autovacuum worker running a parallel vacuum
742 : : * (leader), propagating any change to its parallel workers.
743 : : */
744 : : void
745 : 16 : parallel_vacuum_refresh_cost_params(void)
746 : : {
747 : : Assert(AmAutoVacuumWorkerProcess());
748 : :
749 : : /*
750 : : * Quick return if the leader is not sharing the delay parameters with
751 : : * parallel workers.
752 : : */
753 [ - + ]: 16 : if (pv_shared_cost_params == NULL)
754 : 0 : return;
755 : :
756 [ - + ]: 16 : if (ConfigReloadPending)
757 : : {
758 : 0 : ConfigReloadPending = false;
759 : 0 : ProcessConfigFile(PGC_SIGHUP);
760 : :
761 : : /* This re-divides the cost limit too */
762 : 0 : VacuumUpdateCosts();
763 : : }
764 : : else
765 : : {
766 : : /* The number of workers sharing the cost limit may have changed */
767 : 16 : AutoVacuumUpdateCostLimit();
768 : : }
769 : :
770 : 16 : parallel_vacuum_propagate_shared_delay_params();
771 : : }
772 : :
773 : : /*
774 : : * Compute the number of parallel worker processes to request. Both index
775 : : * vacuum and index cleanup can be executed with parallel workers.
776 : : * The index is eligible for parallel vacuum iff its size is greater than
777 : : * min_parallel_index_scan_size as invoking workers for very small indexes
778 : : * can hurt performance.
779 : : *
780 : : * nrequested is the number of parallel workers that user requested. If
781 : : * nrequested is 0, we compute the parallel degree based on nindexes, that is
782 : : * the number of indexes that support parallel vacuum. This function also
783 : : * sets will_parallel_vacuum to remember indexes that participate in parallel
784 : : * vacuum.
785 : : */
786 : : static int
787 : 42985 : parallel_vacuum_compute_workers(Relation *indrels, int nindexes, int nrequested,
788 : : bool *will_parallel_vacuum)
789 : : {
790 : 42985 : int nindexes_parallel = 0;
791 : 42985 : int nindexes_parallel_bulkdel = 0;
792 : 42985 : int nindexes_parallel_cleanup = 0;
793 : : int parallel_workers;
794 : : int max_workers;
795 : :
796 : 85970 : max_workers = AmAutoVacuumWorkerProcess() ?
797 [ + + ]: 42985 : autovacuum_max_parallel_workers :
798 : : max_parallel_maintenance_workers;
799 : :
800 : : /*
801 : : * We don't allow performing parallel operation in standalone backend or
802 : : * when parallelism is disabled.
803 : : */
804 [ + + + + ]: 42985 : if (!IsUnderPostmaster || max_workers == 0)
805 : 39530 : return 0;
806 : :
807 : : /*
808 : : * Compute the number of indexes that can participate in parallel vacuum.
809 : : */
810 [ + + ]: 11276 : for (int i = 0; i < nindexes; i++)
811 : : {
812 : 7821 : Relation indrel = indrels[i];
813 : 7821 : uint8 vacoptions = indrel->rd_indam->amparallelvacuumoptions;
814 : :
815 : : /* Skip index that is not a suitable target for parallel index vacuum */
816 [ + - ]: 7821 : if (vacoptions == VACUUM_OPTION_NO_PARALLEL ||
817 [ + + ]: 7821 : RelationGetNumberOfBlocks(indrel) < min_parallel_index_scan_size)
818 : 7717 : continue;
819 : :
820 : 104 : will_parallel_vacuum[i] = true;
821 : :
822 [ + + ]: 104 : if ((vacoptions & VACUUM_OPTION_PARALLEL_BULKDEL) != 0)
823 : 96 : nindexes_parallel_bulkdel++;
824 [ + + ]: 104 : if (((vacoptions & VACUUM_OPTION_PARALLEL_CLEANUP) != 0) ||
825 [ + + ]: 88 : ((vacoptions & VACUUM_OPTION_PARALLEL_COND_CLEANUP) != 0))
826 : 96 : nindexes_parallel_cleanup++;
827 : : }
828 : :
829 : 3455 : nindexes_parallel = Max(nindexes_parallel_bulkdel,
830 : : nindexes_parallel_cleanup);
831 : :
832 : : /* The leader process takes one index */
833 : 3455 : nindexes_parallel--;
834 : :
835 : : /* No index supports parallel vacuum */
836 [ + + ]: 3455 : if (nindexes_parallel <= 0)
837 : 3428 : return 0;
838 : :
839 : : /* Compute the parallel degree */
840 : 27 : parallel_workers = (nrequested > 0) ?
841 [ + + ]: 27 : Min(nrequested, nindexes_parallel) : nindexes_parallel;
842 : :
843 : : /* Cap by GUC variable */
844 : 27 : parallel_workers = Min(parallel_workers, max_workers);
845 : :
846 : 27 : return parallel_workers;
847 : : }
848 : :
849 : : /*
850 : : * Perform index vacuum or index cleanup with parallel workers. This function
851 : : * must be used by the parallel vacuum leader process.
852 : : *
853 : : * If wstats is not NULL, the parallel worker statistics are updated.
854 : : */
855 : : static void
856 : 47 : parallel_vacuum_process_all_indexes(ParallelVacuumState *pvs, int num_index_scans,
857 : : bool vacuum, PVWorkerStats *wstats)
858 : : {
859 : : int nworkers;
860 : : PVIndVacStatus new_status;
861 : :
862 : : Assert(!IsParallelWorker());
863 : :
864 [ + + ]: 47 : if (vacuum)
865 : : {
866 : 21 : new_status = PARALLEL_INDVAC_STATUS_NEED_BULKDELETE;
867 : :
868 : : /* Determine the number of parallel workers to launch */
869 : 21 : nworkers = pvs->nindexes_parallel_bulkdel;
870 : : }
871 : : else
872 : : {
873 : 26 : new_status = PARALLEL_INDVAC_STATUS_NEED_CLEANUP;
874 : :
875 : : /* Determine the number of parallel workers to launch */
876 : 26 : nworkers = pvs->nindexes_parallel_cleanup;
877 : :
878 : : /* Add conditionally parallel-aware indexes if in the first time call */
879 [ + + ]: 26 : if (num_index_scans == 0)
880 : 14 : nworkers += pvs->nindexes_parallel_condcleanup;
881 : : }
882 : :
883 : : /* The leader process will participate */
884 : 47 : nworkers--;
885 : :
886 : : /*
887 : : * It is possible that parallel context is initialized with fewer workers
888 : : * than the number of indexes that need a separate worker in the current
889 : : * phase, so we need to consider it. See
890 : : * parallel_vacuum_compute_workers().
891 : : */
892 : 47 : nworkers = Min(nworkers, pvs->pcxt->nworkers);
893 : :
894 : : /* Update the statistics, if we asked to */
895 [ + - + + ]: 47 : if (wstats != NULL && nworkers > 0)
896 : 36 : wstats->nplanned += nworkers;
897 : :
898 : : /*
899 : : * Set index vacuum status and mark whether parallel vacuum worker can
900 : : * process it.
901 : : */
902 [ + + ]: 197 : for (int i = 0; i < pvs->nindexes; i++)
903 : : {
904 : 150 : PVIndStats *indstats = &(pvs->indstats[i]);
905 : :
906 : : Assert(indstats->status == PARALLEL_INDVAC_STATUS_INITIAL);
907 : 150 : indstats->status = new_status;
908 : 150 : indstats->parallel_workers_can_process =
909 [ + + + + ]: 292 : (pvs->will_parallel_vacuum[i] &&
910 : 142 : parallel_vacuum_index_is_parallel_safe(pvs->indrels[i],
911 : : num_index_scans,
912 : 150 : vacuum));
913 : : }
914 : :
915 : : /* Reset the parallel index processing and progress counters */
916 : 47 : pg_atomic_write_u32(&(pvs->shared->idx), 0);
917 : :
918 : : /* Setup the shared cost-based vacuum delay and launch workers */
919 [ + + ]: 47 : if (nworkers > 0)
920 : : {
921 : : /* Reinitialize parallel context to relaunch parallel workers */
922 [ + + ]: 36 : if (num_index_scans > 0)
923 : 9 : ReinitializeParallelDSM(pvs->pcxt);
924 : :
925 : : /*
926 : : * Set up shared cost balance and the number of active workers for
927 : : * vacuum delay. We need to do this before launching workers as
928 : : * otherwise, they might not see the updated values for these
929 : : * parameters.
930 : : */
931 : 36 : pg_atomic_write_u32(&(pvs->shared->cost_balance), VacuumCostBalance);
932 : 36 : pg_atomic_write_u32(&(pvs->shared->active_nworkers), 0);
933 : :
934 : : /*
935 : : * The number of workers can vary between bulkdelete and cleanup
936 : : * phase.
937 : : */
938 : 36 : ReinitializeParallelWorkers(pvs->pcxt, nworkers);
939 : :
940 : 36 : LaunchParallelWorkers(pvs->pcxt);
941 : :
942 [ + - ]: 36 : if (pvs->pcxt->nworkers_launched > 0)
943 : : {
944 : : /*
945 : : * Reset the local cost values for leader backend as we have
946 : : * already accumulated the remaining balance of heap.
947 : : */
948 : 36 : VacuumCostBalance = 0;
949 : 36 : VacuumCostBalanceLocal = 0;
950 : :
951 : : /* Enable shared cost balance for leader backend */
952 : 36 : VacuumSharedCostBalance = &(pvs->shared->cost_balance);
953 : 36 : VacuumActiveNWorkers = &(pvs->shared->active_nworkers);
954 : :
955 : : /* Update the statistics, if we asked to */
956 [ + - ]: 36 : if (wstats != NULL)
957 : 36 : wstats->nlaunched += pvs->pcxt->nworkers_launched;
958 : : }
959 : :
960 [ + + ]: 36 : if (vacuum)
961 [ + + ]: 21 : ereport(pvs->shared->elevel,
962 : : (errmsg(ngettext("launched %d parallel vacuum worker for index vacuuming (planned: %d)",
963 : : "launched %d parallel vacuum workers for index vacuuming (planned: %d)",
964 : : pvs->pcxt->nworkers_launched),
965 : : pvs->pcxt->nworkers_launched, nworkers)));
966 : : else
967 [ - + ]: 15 : ereport(pvs->shared->elevel,
968 : : (errmsg(ngettext("launched %d parallel vacuum worker for index cleanup (planned: %d)",
969 : : "launched %d parallel vacuum workers for index cleanup (planned: %d)",
970 : : pvs->pcxt->nworkers_launched),
971 : : pvs->pcxt->nworkers_launched, nworkers)));
972 : : }
973 : :
974 : : /* Vacuum the indexes that can be processed by only leader process */
975 : 47 : parallel_vacuum_process_unsafe_indexes(pvs);
976 : :
977 : : /*
978 : : * Join as a parallel worker. The leader vacuums alone processes all
979 : : * parallel-safe indexes in the case where no workers are launched.
980 : : */
981 : 47 : parallel_vacuum_process_safe_indexes(pvs);
982 : :
983 : : /*
984 : : * Next, accumulate buffer and WAL usage. (This must wait for the workers
985 : : * to finish, or we might get incomplete data.)
986 : : */
987 [ + + ]: 47 : if (nworkers > 0)
988 : : {
989 : : /* Wait for all vacuum workers to finish */
990 : 36 : WaitForParallelWorkersToFinish(pvs->pcxt);
991 : :
992 [ + + ]: 79 : for (int i = 0; i < pvs->pcxt->nworkers_launched; i++)
993 : 44 : InstrAccumParallelQuery(&pvs->buffer_usage[i], &pvs->wal_usage[i]);
994 : : }
995 : :
996 : : /*
997 : : * Reset all index status back to initial (while checking that we have
998 : : * vacuumed all indexes).
999 : : */
1000 [ + + ]: 194 : for (int i = 0; i < pvs->nindexes; i++)
1001 : : {
1002 : 148 : PVIndStats *indstats = &(pvs->indstats[i]);
1003 : :
1004 [ - + ]: 148 : if (indstats->status != PARALLEL_INDVAC_STATUS_COMPLETED)
1005 [ # # ]: 0 : elog(ERROR, "parallel index vacuum on index \"%s\" is not completed",
1006 : : RelationGetRelationName(pvs->indrels[i]));
1007 : :
1008 : 148 : indstats->status = PARALLEL_INDVAC_STATUS_INITIAL;
1009 : : }
1010 : :
1011 : : /*
1012 : : * Carry the shared balance value to heap scan and disable shared costing
1013 : : */
1014 [ + + ]: 46 : if (VacuumSharedCostBalance)
1015 : : {
1016 : 35 : VacuumCostBalance = pg_atomic_read_u32(VacuumSharedCostBalance);
1017 : 35 : VacuumSharedCostBalance = NULL;
1018 : 35 : VacuumActiveNWorkers = NULL;
1019 : : }
1020 : 46 : }
1021 : :
1022 : : /*
1023 : : * Index vacuum/cleanup routine used by the leader process and parallel
1024 : : * vacuum worker processes to vacuum the indexes in parallel.
1025 : : */
1026 : : static void
1027 : 91 : parallel_vacuum_process_safe_indexes(ParallelVacuumState *pvs)
1028 : : {
1029 : : /*
1030 : : * Increment the active worker count if we are able to launch any worker.
1031 : : */
1032 [ + + ]: 91 : if (VacuumActiveNWorkers)
1033 : 80 : pg_atomic_add_fetch_u32(VacuumActiveNWorkers, 1);
1034 : :
1035 : : /* Loop until all indexes are vacuumed */
1036 : : for (;;)
1037 : 150 : {
1038 : : int idx;
1039 : : PVIndStats *indstats;
1040 : :
1041 : : /* Get an index number to process */
1042 : 241 : idx = pg_atomic_fetch_add_u32(&(pvs->shared->idx), 1);
1043 : :
1044 : : /* Done for all indexes? */
1045 [ + + ]: 241 : if (idx >= pvs->nindexes)
1046 : 91 : break;
1047 : :
1048 : 150 : indstats = &(pvs->indstats[idx]);
1049 : :
1050 : : /*
1051 : : * Skip vacuuming index that is unsafe for workers or has an
1052 : : * unsuitable target for parallel index vacuum (this is vacuumed in
1053 : : * parallel_vacuum_process_unsafe_indexes() by the leader).
1054 : : */
1055 [ + + ]: 150 : if (!indstats->parallel_workers_can_process)
1056 : 48 : continue;
1057 : :
1058 : : /* Do vacuum or cleanup of the index */
1059 : 102 : parallel_vacuum_process_one_index(pvs, pvs->indrels[idx], indstats);
1060 : : }
1061 : :
1062 : : /*
1063 : : * We have completed the index vacuum so decrement the active worker
1064 : : * count.
1065 : : */
1066 [ + + ]: 91 : if (VacuumActiveNWorkers)
1067 : 80 : pg_atomic_sub_fetch_u32(VacuumActiveNWorkers, 1);
1068 : 91 : }
1069 : :
1070 : : /*
1071 : : * Perform parallel vacuuming of indexes in leader process.
1072 : : *
1073 : : * Handles index vacuuming (or index cleanup) for indexes that are not
1074 : : * parallel safe. It's possible that this will vary for a given index, based
1075 : : * on details like whether we're performing index cleanup right now.
1076 : : *
1077 : : * Also performs vacuuming of smaller indexes that fell under the size cutoff
1078 : : * enforced by parallel_vacuum_compute_workers().
1079 : : */
1080 : : static void
1081 : 47 : parallel_vacuum_process_unsafe_indexes(ParallelVacuumState *pvs)
1082 : : {
1083 : : Assert(!IsParallelWorker());
1084 : :
1085 : : /*
1086 : : * Increment the active worker count if we are able to launch any worker.
1087 : : */
1088 [ + + ]: 47 : if (VacuumActiveNWorkers)
1089 : 36 : pg_atomic_add_fetch_u32(VacuumActiveNWorkers, 1);
1090 : :
1091 [ + + ]: 197 : for (int i = 0; i < pvs->nindexes; i++)
1092 : : {
1093 : 150 : PVIndStats *indstats = &(pvs->indstats[i]);
1094 : :
1095 : : /* Skip, indexes that are safe for workers */
1096 [ + + ]: 150 : if (indstats->parallel_workers_can_process)
1097 : 102 : continue;
1098 : :
1099 : : /* Do vacuum or cleanup of the index */
1100 : 48 : parallel_vacuum_process_one_index(pvs, pvs->indrels[i], indstats);
1101 : : }
1102 : :
1103 : : /*
1104 : : * We have completed the index vacuum so decrement the active worker
1105 : : * count.
1106 : : */
1107 [ + + ]: 47 : if (VacuumActiveNWorkers)
1108 : 36 : pg_atomic_sub_fetch_u32(VacuumActiveNWorkers, 1);
1109 : 47 : }
1110 : :
1111 : : /*
1112 : : * Vacuum or cleanup index either by leader process or by one of the worker
1113 : : * process. After vacuuming the index this function copies the index
1114 : : * statistics returned from ambulkdelete and amvacuumcleanup to the DSM
1115 : : * segment.
1116 : : */
1117 : : static void
1118 : 150 : parallel_vacuum_process_one_index(ParallelVacuumState *pvs, Relation indrel,
1119 : : PVIndStats *indstats)
1120 : : {
1121 : 150 : IndexBulkDeleteResult *istat = NULL;
1122 : : IndexBulkDeleteResult *istat_res;
1123 : : IndexVacuumInfo ivinfo;
1124 : 150 : const int progress_index[] = {
1125 : : PROGRESS_VACUUM_PHASE,
1126 : : PROGRESS_VACUUM_CURRENT_INDEX_RELID
1127 : : };
1128 : : int64 progress_val[2];
1129 : 150 : const int reset_index[] = {
1130 : : PROGRESS_VACUUM_CURRENT_INDEX_RELID,
1131 : : PROGRESS_SCAN_BLOCKS_TOTAL,
1132 : : PROGRESS_SCAN_BLOCKS_DONE
1133 : : };
1134 : 150 : const int64 reset_val[] = {(int64) InvalidOid, 0, 0};
1135 : :
1136 : : /*
1137 : : * Update the pointer to the corresponding bulk-deletion result if someone
1138 : : * has already updated it
1139 : : */
1140 [ + + ]: 150 : if (indstats->istat_updated)
1141 : 54 : istat = &(indstats->istat);
1142 : :
1143 : 150 : ivinfo.index = indrel;
1144 : 150 : ivinfo.heaprel = pvs->heaprel;
1145 : 150 : ivinfo.analyze_only = false;
1146 : 150 : ivinfo.is_autovacuum = pvs->shared->is_autovacuum;
1147 : 150 : ivinfo.report_progress = true;
1148 : 150 : ivinfo.message_level = DEBUG2;
1149 : 150 : ivinfo.estimated_count = pvs->shared->estimated_count;
1150 : 150 : ivinfo.num_heap_tuples = pvs->shared->reltuples;
1151 : 150 : ivinfo.strategy = pvs->bstrategy;
1152 : :
1153 : : /* Update error traceback information */
1154 : 150 : pvs->indname = pstrdup(RelationGetRelationName(indrel));
1155 : 150 : pvs->status = indstats->status;
1156 : :
1157 : : /* Report the phase and the index we're about to process */
1158 : 300 : progress_val[0] = (indstats->status == PARALLEL_INDVAC_STATUS_NEED_BULKDELETE)
1159 : : ? PROGRESS_VACUUM_PHASE_VACUUM_INDEX
1160 [ + + ]: 150 : : PROGRESS_VACUUM_PHASE_INDEX_CLEANUP;
1161 : 150 : progress_val[1] = (int64) RelationGetRelid(indrel);
1162 : 150 : pgstat_progress_update_multi_param(2, progress_index, progress_val);
1163 : :
1164 [ + + - ]: 150 : switch (indstats->status)
1165 : : {
1166 : 56 : case PARALLEL_INDVAC_STATUS_NEED_BULKDELETE:
1167 : 56 : istat_res = vac_bulkdel_one_index(&ivinfo, istat, pvs->dead_items,
1168 : 56 : &pvs->shared->dead_items_info);
1169 : 56 : break;
1170 : 94 : case PARALLEL_INDVAC_STATUS_NEED_CLEANUP:
1171 : 94 : istat_res = vac_cleanup_one_index(&ivinfo, istat);
1172 : 94 : break;
1173 : 0 : default:
1174 [ # # ]: 0 : elog(ERROR, "unexpected parallel vacuum index status %d for index \"%s\"",
1175 : : indstats->status,
1176 : : RelationGetRelationName(indrel));
1177 : : }
1178 : :
1179 : 150 : pgstat_progress_update_multi_param(3, reset_index, reset_val);
1180 : :
1181 : : /*
1182 : : * Copy the index bulk-deletion result returned from ambulkdelete and
1183 : : * amvacuumcleanup to the DSM segment if it's the first cycle because they
1184 : : * allocate locally and it's possible that an index will be vacuumed by a
1185 : : * different vacuum process the next cycle. Copying the result normally
1186 : : * happens only the first time an index is vacuumed. For any additional
1187 : : * vacuum pass, we directly point to the result on the DSM segment and
1188 : : * pass it to vacuum index APIs so that workers can update it directly.
1189 : : *
1190 : : * Since all vacuum workers write the bulk-deletion result at different
1191 : : * slots we can write them without locking.
1192 : : */
1193 [ + + + + ]: 150 : if (!indstats->istat_updated && istat_res != NULL)
1194 : : {
1195 : 68 : memcpy(&(indstats->istat), istat_res, sizeof(IndexBulkDeleteResult));
1196 : 68 : indstats->istat_updated = true;
1197 : :
1198 : : /* Free the locally-allocated bulk-deletion result */
1199 : 68 : pfree(istat_res);
1200 : : }
1201 : :
1202 : : /*
1203 : : * Update the status to completed. No need to lock here since each worker
1204 : : * touches different indexes.
1205 : : */
1206 : 150 : indstats->status = PARALLEL_INDVAC_STATUS_COMPLETED;
1207 : :
1208 : : /* Reset error traceback information */
1209 : 150 : pvs->status = PARALLEL_INDVAC_STATUS_COMPLETED;
1210 : 150 : pfree(pvs->indname);
1211 : 150 : pvs->indname = NULL;
1212 : :
1213 : : /*
1214 : : * Call the parallel variant of pgstat_progress_incr_param so workers can
1215 : : * report progress of index vacuum to the leader.
1216 : : */
1217 : 150 : pgstat_progress_parallel_incr_param(PROGRESS_VACUUM_INDEXES_PROCESSED, 1);
1218 : 150 : }
1219 : :
1220 : : /*
1221 : : * Returns false, if the given index can't participate in the next execution of
1222 : : * parallel index vacuum or parallel index cleanup.
1223 : : */
1224 : : static bool
1225 : 142 : parallel_vacuum_index_is_parallel_safe(Relation indrel, int num_index_scans,
1226 : : bool vacuum)
1227 : : {
1228 : : uint8 vacoptions;
1229 : :
1230 : 142 : vacoptions = indrel->rd_indam->amparallelvacuumoptions;
1231 : :
1232 : : /* In parallel vacuum case, check if it supports parallel bulk-deletion */
1233 [ + + ]: 142 : if (vacuum)
1234 : 52 : return ((vacoptions & VACUUM_OPTION_PARALLEL_BULKDEL) != 0);
1235 : :
1236 : : /* Not safe, if the index does not support parallel cleanup */
1237 [ + + ]: 90 : if (((vacoptions & VACUUM_OPTION_PARALLEL_CLEANUP) == 0) &&
1238 [ + + ]: 74 : ((vacoptions & VACUUM_OPTION_PARALLEL_COND_CLEANUP) == 0))
1239 : 8 : return false;
1240 : :
1241 : : /*
1242 : : * Not safe, if the index supports parallel cleanup conditionally, but we
1243 : : * have already processed the index (for bulkdelete). We do this to avoid
1244 : : * the need to invoke workers when parallel index cleanup doesn't need to
1245 : : * scan the index. See the comments for option
1246 : : * VACUUM_OPTION_PARALLEL_COND_CLEANUP to know when indexes support
1247 : : * parallel cleanup conditionally.
1248 : : */
1249 [ + + ]: 82 : if (num_index_scans > 0 &&
1250 [ + + ]: 33 : ((vacoptions & VACUUM_OPTION_PARALLEL_COND_CLEANUP) != 0))
1251 : 31 : return false;
1252 : :
1253 : 51 : return true;
1254 : : }
1255 : :
1256 : : /*
1257 : : * Perform work within a launched parallel process.
1258 : : *
1259 : : * Parallel vacuum workers perform only index vacuum or index cleanup; they
1260 : : * report progress for the index they are processing.
1261 : : */
1262 : : void
1263 : 45 : parallel_vacuum_main(dsm_segment *seg, shm_toc *toc)
1264 : : {
1265 : : ParallelVacuumState pvs;
1266 : : Relation rel;
1267 : : Relation *indrels;
1268 : : PVIndStats *indstats;
1269 : : PVShared *shared;
1270 : : TidStore *dead_items;
1271 : : BufferUsage *buffer_usage;
1272 : : WalUsage *wal_usage;
1273 : : int nindexes;
1274 : : char *sharedquery;
1275 : : ErrorContextCallback errcallback;
1276 : :
1277 : : /*
1278 : : * A parallel vacuum worker carries only the PROC_IN_VACUUM flag. The
1279 : : * leader, whether it's a backend running a VACUUM command or an
1280 : : * autovacuum worker, sets PROC_IN_VACUUM when it starts vacuuming the
1281 : : * table, and the worker inherits the flag by importing the leader's
1282 : : * snapshot (see ProcArrayInstallRestoredXmin). The leader's other flags
1283 : : * don't reach the worker: the snapshot import copies only the
1284 : : * PROC_XMIN_FLAGS bits, so PROC_VACUUM_FOR_WRAPAROUND isn't carried over,
1285 : : * and PROC_IS_AUTOVACUUM is never set on the worker in the first place
1286 : : * since parallel workers run as regular background workers, not
1287 : : * autovacuum workers.
1288 : : */
1289 : : Assert(MyProc->statusFlags == PROC_IN_VACUUM);
1290 : :
1291 [ + + ]: 45 : elog(DEBUG1, "starting parallel vacuum worker");
1292 : :
1293 : : /* Used by tests to hold a worker while it is attached to its leader */
1294 : 45 : INJECTION_POINT("parallel-vacuum-worker-start", NULL);
1295 : :
1296 : 44 : shared = (PVShared *) shm_toc_lookup(toc, PARALLEL_VACUUM_KEY_SHARED, false);
1297 : :
1298 : : /* Set debug_query_string for individual workers */
1299 : 44 : sharedquery = shm_toc_lookup(toc, PARALLEL_VACUUM_KEY_QUERY_TEXT, true);
1300 : 44 : debug_query_string = sharedquery;
1301 : 44 : pgstat_report_activity(STATE_RUNNING, debug_query_string);
1302 : :
1303 : : /* Track query ID */
1304 : 44 : pgstat_report_query_id(shared->queryid, false);
1305 : :
1306 : : /*
1307 : : * Open table. The lock mode is the same as the leader process. It's
1308 : : * okay because the lock mode does not conflict among the parallel
1309 : : * workers.
1310 : : */
1311 : 44 : rel = table_open(shared->relid, ShareUpdateExclusiveLock);
1312 : :
1313 : : /*
1314 : : * Open all indexes. indrels are sorted in order by OID, which should be
1315 : : * matched to the leader's one.
1316 : : */
1317 : 44 : vac_open_indexes(rel, RowExclusiveLock, &nindexes, &indrels);
1318 : : Assert(nindexes > 0);
1319 : :
1320 : : /*
1321 : : * Apply the desired value of maintenance_work_mem within this process.
1322 : : * Really we should use SetConfigOption() to change a GUC, but since we're
1323 : : * already in parallel mode guc.c would complain about that. Fortunately,
1324 : : * by the same token guc.c will not let any user-defined code change it.
1325 : : * So just avert your eyes while we do this:
1326 : : */
1327 [ + - ]: 44 : if (shared->maintenance_work_mem_worker > 0)
1328 : 44 : maintenance_work_mem = shared->maintenance_work_mem_worker;
1329 : :
1330 : : /* Set index statistics */
1331 : 44 : indstats = (PVIndStats *) shm_toc_lookup(toc,
1332 : : PARALLEL_VACUUM_KEY_INDEX_STATS,
1333 : : false);
1334 : :
1335 : : /* Find dead_items in shared memory */
1336 : 44 : dead_items = TidStoreAttach(shared->dead_items_dsa_handle,
1337 : : shared->dead_items_handle);
1338 : :
1339 : : /* Set cost-based vacuum delay */
1340 [ + + ]: 44 : if (shared->is_autovacuum)
1341 : : {
1342 : : /*
1343 : : * Parallel autovacuum workers initialize cost-based delay parameters
1344 : : * from the leader's shared state rather than GUC defaults, because
1345 : : * the leader may have applied per-table or autovacuum-specific
1346 : : * overrides. pv_shared_cost_params must be set before calling
1347 : : * parallel_vacuum_update_shared_delay_params().
1348 : : */
1349 : 4 : pv_shared_cost_params = &(shared->cost_params);
1350 : 4 : parallel_vacuum_update_shared_delay_params();
1351 : : }
1352 : : else
1353 : 40 : VacuumUpdateCosts();
1354 : :
1355 : 44 : VacuumCostBalance = 0;
1356 : 44 : VacuumCostBalanceLocal = 0;
1357 : 44 : VacuumSharedCostBalance = &(shared->cost_balance);
1358 : 44 : VacuumActiveNWorkers = &(shared->active_nworkers);
1359 : :
1360 : : /* Set parallel vacuum state */
1361 : 44 : pvs.indrels = indrels;
1362 : 44 : pvs.nindexes = nindexes;
1363 : 44 : pvs.indstats = indstats;
1364 : 44 : pvs.shared = shared;
1365 : 44 : pvs.dead_items = dead_items;
1366 : 44 : pvs.relnamespace = get_namespace_name(RelationGetNamespace(rel));
1367 : 44 : pvs.relname = pstrdup(RelationGetRelationName(rel));
1368 : 44 : pvs.heaprel = rel;
1369 : :
1370 : : /* These fields will be filled during index vacuum or cleanup */
1371 : 44 : pvs.indname = NULL;
1372 : 44 : pvs.status = PARALLEL_INDVAC_STATUS_INITIAL;
1373 : :
1374 : : /* Each parallel VACUUM worker gets its own access strategy. */
1375 : 88 : pvs.bstrategy = GetAccessStrategyWithSize(BAS_VACUUM,
1376 : 44 : shared->ring_nbuffers * (BLCKSZ / 1024));
1377 : :
1378 : : /* Setup error traceback support for ereport() */
1379 : 44 : errcallback.callback = parallel_vacuum_error_callback;
1380 : 44 : errcallback.arg = &pvs;
1381 : 44 : errcallback.previous = error_context_stack;
1382 : 44 : error_context_stack = &errcallback;
1383 : :
1384 : : /* Prepare to track buffer usage during parallel execution */
1385 : 44 : InstrStartParallelQuery();
1386 : :
1387 : : /* Register this worker for vacuum progress reporting */
1388 : 44 : pgstat_progress_start_command(PROGRESS_COMMAND_VACUUM, shared->relid);
1389 : :
1390 : : /* Process indexes to perform vacuum/cleanup */
1391 : 44 : parallel_vacuum_process_safe_indexes(&pvs);
1392 : :
1393 : : /* Report buffer/WAL usage during parallel execution */
1394 : 44 : buffer_usage = shm_toc_lookup(toc, PARALLEL_VACUUM_KEY_BUFFER_USAGE, false);
1395 : 44 : wal_usage = shm_toc_lookup(toc, PARALLEL_VACUUM_KEY_WAL_USAGE, false);
1396 : 44 : InstrEndParallelQuery(&buffer_usage[ParallelWorkerNumber],
1397 : 44 : &wal_usage[ParallelWorkerNumber]);
1398 : :
1399 : : /* Report any remaining cost-based vacuum delay time */
1400 [ + + ]: 44 : if (parallel_vacuum_worker_delay_ns > 0)
1401 : 2 : pgstat_progress_parallel_incr_param(PROGRESS_VACUUM_DELAY_TIME,
1402 : : parallel_vacuum_worker_delay_ns);
1403 : :
1404 : 44 : TidStoreDetach(dead_items);
1405 : :
1406 : : /* Pop the error context stack */
1407 : 44 : error_context_stack = errcallback.previous;
1408 : :
1409 : 44 : pgstat_progress_end_command();
1410 : :
1411 : 44 : vac_close_indexes(nindexes, indrels, RowExclusiveLock);
1412 : 44 : table_close(rel, ShareUpdateExclusiveLock);
1413 : 44 : FreeAccessStrategy(pvs.bstrategy);
1414 : :
1415 [ + + ]: 44 : if (shared->is_autovacuum)
1416 : 4 : pv_shared_cost_params = NULL;
1417 : 44 : }
1418 : :
1419 : : /*
1420 : : * Error context callback for errors occurring during parallel index vacuum.
1421 : : * The error context messages should match the messages set in the lazy vacuum
1422 : : * error context. If you change this function, change vacuum_error_callback()
1423 : : * as well.
1424 : : */
1425 : : static void
1426 : 7 : parallel_vacuum_error_callback(void *arg)
1427 : : {
1428 : 7 : ParallelVacuumState *errinfo = arg;
1429 : :
1430 [ + - - ]: 7 : switch (errinfo->status)
1431 : : {
1432 : 7 : case PARALLEL_INDVAC_STATUS_NEED_BULKDELETE:
1433 : 7 : errcontext("while vacuuming index \"%s\" of relation \"%s.%s\"",
1434 : : errinfo->indname,
1435 : : errinfo->relnamespace,
1436 : : errinfo->relname);
1437 : 7 : break;
1438 : 0 : case PARALLEL_INDVAC_STATUS_NEED_CLEANUP:
1439 : 0 : errcontext("while cleaning up index \"%s\" of relation \"%s.%s\"",
1440 : : errinfo->indname,
1441 : : errinfo->relnamespace,
1442 : : errinfo->relname);
1443 : 0 : break;
1444 : 0 : case PARALLEL_INDVAC_STATUS_INITIAL:
1445 : : case PARALLEL_INDVAC_STATUS_COMPLETED:
1446 : : default:
1447 : 0 : return;
1448 : : }
1449 : : }
|
