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