PostgreSQL Source Code git master
All Data Structures Namespaces Files Functions Variables Typedefs Enumerations Enumerator Macros Pages
vacuumparallel.c File Reference
#include "postgres.h"
#include "access/amapi.h"
#include "access/table.h"
#include "access/xact.h"
#include "commands/progress.h"
#include "commands/vacuum.h"
#include "executor/instrument.h"
#include "optimizer/paths.h"
#include "pgstat.h"
#include "storage/bufmgr.h"
#include "tcop/tcopprot.h"
#include "utils/lsyscache.h"
#include "utils/rel.h"
Include dependency graph for vacuumparallel.c:

Go to the source code of this file.

Data Structures

struct  PVShared
 
struct  PVIndStats
 
struct  ParallelVacuumState
 

Macros

#define PARALLEL_VACUUM_KEY_SHARED   1
 
#define PARALLEL_VACUUM_KEY_QUERY_TEXT   2
 
#define PARALLEL_VACUUM_KEY_BUFFER_USAGE   3
 
#define PARALLEL_VACUUM_KEY_WAL_USAGE   4
 
#define PARALLEL_VACUUM_KEY_INDEX_STATS   5
 

Typedefs

typedef struct PVShared PVShared
 
typedef enum PVIndVacStatus PVIndVacStatus
 
typedef struct PVIndStats PVIndStats
 

Enumerations

enum  PVIndVacStatus { PARALLEL_INDVAC_STATUS_INITIAL = 0 , PARALLEL_INDVAC_STATUS_NEED_BULKDELETE , PARALLEL_INDVAC_STATUS_NEED_CLEANUP , PARALLEL_INDVAC_STATUS_COMPLETED }
 

Functions

static int parallel_vacuum_compute_workers (Relation *indrels, int nindexes, int nrequested, bool *will_parallel_vacuum)
 
static void parallel_vacuum_process_all_indexes (ParallelVacuumState *pvs, int num_index_scans, bool vacuum)
 
static void parallel_vacuum_process_safe_indexes (ParallelVacuumState *pvs)
 
static void parallel_vacuum_process_unsafe_indexes (ParallelVacuumState *pvs)
 
static void parallel_vacuum_process_one_index (ParallelVacuumState *pvs, Relation indrel, PVIndStats *indstats)
 
static bool parallel_vacuum_index_is_parallel_safe (Relation indrel, int num_index_scans, bool vacuum)
 
static void parallel_vacuum_error_callback (void *arg)
 
ParallelVacuumStateparallel_vacuum_init (Relation rel, Relation *indrels, int nindexes, int nrequested_workers, int vac_work_mem, int elevel, BufferAccessStrategy bstrategy)
 
void parallel_vacuum_end (ParallelVacuumState *pvs, IndexBulkDeleteResult **istats)
 
TidStoreparallel_vacuum_get_dead_items (ParallelVacuumState *pvs, VacDeadItemsInfo **dead_items_info_p)
 
void parallel_vacuum_reset_dead_items (ParallelVacuumState *pvs)
 
void parallel_vacuum_bulkdel_all_indexes (ParallelVacuumState *pvs, long num_table_tuples, int num_index_scans)
 
void parallel_vacuum_cleanup_all_indexes (ParallelVacuumState *pvs, long num_table_tuples, int num_index_scans, bool estimated_count)
 
void parallel_vacuum_main (dsm_segment *seg, shm_toc *toc)
 

Macro Definition Documentation

◆ PARALLEL_VACUUM_KEY_BUFFER_USAGE

#define PARALLEL_VACUUM_KEY_BUFFER_USAGE   3

Definition at line 49 of file vacuumparallel.c.

◆ PARALLEL_VACUUM_KEY_INDEX_STATS

#define PARALLEL_VACUUM_KEY_INDEX_STATS   5

Definition at line 51 of file vacuumparallel.c.

◆ PARALLEL_VACUUM_KEY_QUERY_TEXT

#define PARALLEL_VACUUM_KEY_QUERY_TEXT   2

Definition at line 48 of file vacuumparallel.c.

◆ PARALLEL_VACUUM_KEY_SHARED

#define PARALLEL_VACUUM_KEY_SHARED   1

Definition at line 47 of file vacuumparallel.c.

◆ PARALLEL_VACUUM_KEY_WAL_USAGE

#define PARALLEL_VACUUM_KEY_WAL_USAGE   4

Definition at line 50 of file vacuumparallel.c.

Typedef Documentation

◆ PVIndStats

typedef struct PVIndStats PVIndStats

◆ PVIndVacStatus

◆ PVShared

typedef struct PVShared PVShared

Enumeration Type Documentation

◆ PVIndVacStatus

Enumerator
PARALLEL_INDVAC_STATUS_INITIAL 
PARALLEL_INDVAC_STATUS_NEED_BULKDELETE 
PARALLEL_INDVAC_STATUS_NEED_CLEANUP 
PARALLEL_INDVAC_STATUS_COMPLETED 

Definition at line 125 of file vacuumparallel.c.

Function Documentation

◆ parallel_vacuum_bulkdel_all_indexes()

void parallel_vacuum_bulkdel_all_indexes ( ParallelVacuumState pvs,
long  num_table_tuples,
int  num_index_scans 
)

Definition at line 500 of file vacuumparallel.c.

502{
504
505 /*
506 * We can only provide an approximate value of num_heap_tuples, at least
507 * for now.
508 */
509 pvs->shared->reltuples = num_table_tuples;
510 pvs->shared->estimated_count = true;
511
512 parallel_vacuum_process_all_indexes(pvs, num_index_scans, true);
513}
#define Assert(condition)
Definition: c.h:815
#define IsParallelWorker()
Definition: parallel.h:60
double reltuples
bool estimated_count
static void parallel_vacuum_process_all_indexes(ParallelVacuumState *pvs, int num_index_scans, bool vacuum)

References Assert, PVShared::estimated_count, IsParallelWorker, parallel_vacuum_process_all_indexes(), PVShared::reltuples, and ParallelVacuumState::shared.

Referenced by lazy_vacuum_all_indexes().

◆ parallel_vacuum_cleanup_all_indexes()

void parallel_vacuum_cleanup_all_indexes ( ParallelVacuumState pvs,
long  num_table_tuples,
int  num_index_scans,
bool  estimated_count 
)

Definition at line 519 of file vacuumparallel.c.

521{
523
524 /*
525 * We can provide a better estimate of total number of surviving tuples
526 * (we assume indexes are more interested in that than in the number of
527 * nominally live tuples).
528 */
529 pvs->shared->reltuples = num_table_tuples;
530 pvs->shared->estimated_count = estimated_count;
531
532 parallel_vacuum_process_all_indexes(pvs, num_index_scans, false);
533}

References Assert, PVShared::estimated_count, IsParallelWorker, parallel_vacuum_process_all_indexes(), PVShared::reltuples, and ParallelVacuumState::shared.

Referenced by lazy_cleanup_all_indexes().

◆ parallel_vacuum_compute_workers()

static int parallel_vacuum_compute_workers ( Relation indrels,
int  nindexes,
int  nrequested,
bool *  will_parallel_vacuum 
)
static

Definition at line 549 of file vacuumparallel.c.

551{
552 int nindexes_parallel = 0;
553 int nindexes_parallel_bulkdel = 0;
554 int nindexes_parallel_cleanup = 0;
555 int parallel_workers;
556
557 /*
558 * We don't allow performing parallel operation in standalone backend or
559 * when parallelism is disabled.
560 */
562 return 0;
563
564 /*
565 * Compute the number of indexes that can participate in parallel vacuum.
566 */
567 for (int i = 0; i < nindexes; i++)
568 {
569 Relation indrel = indrels[i];
570 uint8 vacoptions = indrel->rd_indam->amparallelvacuumoptions;
571
572 /* Skip index that is not a suitable target for parallel index vacuum */
573 if (vacoptions == VACUUM_OPTION_NO_PARALLEL ||
575 continue;
576
577 will_parallel_vacuum[i] = true;
578
579 if ((vacoptions & VACUUM_OPTION_PARALLEL_BULKDEL) != 0)
580 nindexes_parallel_bulkdel++;
581 if (((vacoptions & VACUUM_OPTION_PARALLEL_CLEANUP) != 0) ||
582 ((vacoptions & VACUUM_OPTION_PARALLEL_COND_CLEANUP) != 0))
583 nindexes_parallel_cleanup++;
584 }
585
586 nindexes_parallel = Max(nindexes_parallel_bulkdel,
587 nindexes_parallel_cleanup);
588
589 /* The leader process takes one index */
590 nindexes_parallel--;
591
592 /* No index supports parallel vacuum */
593 if (nindexes_parallel <= 0)
594 return 0;
595
596 /* Compute the parallel degree */
597 parallel_workers = (nrequested > 0) ?
598 Min(nrequested, nindexes_parallel) : nindexes_parallel;
599
600 /* Cap by max_parallel_maintenance_workers */
601 parallel_workers = Min(parallel_workers, max_parallel_maintenance_workers);
602
603 return parallel_workers;
604}
int min_parallel_index_scan_size
Definition: allpaths.c:82
#define RelationGetNumberOfBlocks(reln)
Definition: bufmgr.h:273
#define Min(x, y)
Definition: c.h:961
uint8_t uint8
Definition: c.h:486
#define Max(x, y)
Definition: c.h:955
int max_parallel_maintenance_workers
Definition: globals.c:133
bool IsUnderPostmaster
Definition: globals.c:119
int i
Definition: isn.c:72
uint8 amparallelvacuumoptions
Definition: amapi.h:267
struct IndexAmRoutine * rd_indam
Definition: rel.h:206
#define VACUUM_OPTION_PARALLEL_CLEANUP
Definition: vacuum.h:63
#define VACUUM_OPTION_NO_PARALLEL
Definition: vacuum.h:42
#define VACUUM_OPTION_PARALLEL_BULKDEL
Definition: vacuum.h:48
#define VACUUM_OPTION_PARALLEL_COND_CLEANUP
Definition: vacuum.h:55

References IndexAmRoutine::amparallelvacuumoptions, i, IsUnderPostmaster, Max, max_parallel_maintenance_workers, Min, min_parallel_index_scan_size, RelationData::rd_indam, RelationGetNumberOfBlocks, VACUUM_OPTION_NO_PARALLEL, VACUUM_OPTION_PARALLEL_BULKDEL, VACUUM_OPTION_PARALLEL_CLEANUP, and VACUUM_OPTION_PARALLEL_COND_CLEANUP.

Referenced by parallel_vacuum_init().

◆ parallel_vacuum_end()

void parallel_vacuum_end ( ParallelVacuumState pvs,
IndexBulkDeleteResult **  istats 
)

Definition at line 436 of file vacuumparallel.c.

437{
439
440 /* Copy the updated statistics */
441 for (int i = 0; i < pvs->nindexes; i++)
442 {
443 PVIndStats *indstats = &(pvs->indstats[i]);
444
445 if (indstats->istat_updated)
446 {
448 memcpy(istats[i], &indstats->istat, sizeof(IndexBulkDeleteResult));
449 }
450 else
451 istats[i] = NULL;
452 }
453
455
458
460 pfree(pvs);
461}
void DestroyParallelContext(ParallelContext *pcxt)
Definition: parallel.c:946
void pfree(void *pointer)
Definition: mcxt.c:1521
void * palloc0(Size size)
Definition: mcxt.c:1347
bool istat_updated
IndexBulkDeleteResult istat
ParallelContext * pcxt
PVIndStats * indstats
void TidStoreDestroy(TidStore *ts)
Definition: tidstore.c:317
void ExitParallelMode(void)
Definition: xact.c:1063

References Assert, ParallelVacuumState::dead_items, DestroyParallelContext(), ExitParallelMode(), i, ParallelVacuumState::indstats, IsParallelWorker, PVIndStats::istat, PVIndStats::istat_updated, ParallelVacuumState::nindexes, palloc0(), ParallelVacuumState::pcxt, pfree(), TidStoreDestroy(), and ParallelVacuumState::will_parallel_vacuum.

Referenced by dead_items_cleanup().

◆ parallel_vacuum_error_callback()

static void parallel_vacuum_error_callback ( void *  arg)
static

Definition at line 1107 of file vacuumparallel.c.

1108{
1109 ParallelVacuumState *errinfo = arg;
1110
1111 switch (errinfo->status)
1112 {
1114 errcontext("while vacuuming index \"%s\" of relation \"%s.%s\"",
1115 errinfo->indname,
1116 errinfo->relnamespace,
1117 errinfo->relname);
1118 break;
1120 errcontext("while cleaning up index \"%s\" of relation \"%s.%s\"",
1121 errinfo->indname,
1122 errinfo->relnamespace,
1123 errinfo->relname);
1124 break;
1127 default:
1128 return;
1129 }
1130}
#define errcontext
Definition: elog.h:196
void * arg
PVIndVacStatus status

References arg, errcontext, ParallelVacuumState::indname, PARALLEL_INDVAC_STATUS_COMPLETED, PARALLEL_INDVAC_STATUS_INITIAL, PARALLEL_INDVAC_STATUS_NEED_BULKDELETE, PARALLEL_INDVAC_STATUS_NEED_CLEANUP, ParallelVacuumState::relname, ParallelVacuumState::relnamespace, and ParallelVacuumState::status.

Referenced by parallel_vacuum_main().

◆ parallel_vacuum_get_dead_items()

TidStore * parallel_vacuum_get_dead_items ( ParallelVacuumState pvs,
VacDeadItemsInfo **  dead_items_info_p 
)

Definition at line 467 of file vacuumparallel.c.

468{
469 *dead_items_info_p = &(pvs->shared->dead_items_info);
470 return pvs->dead_items;
471}
VacDeadItemsInfo dead_items_info

References ParallelVacuumState::dead_items, PVShared::dead_items_info, and ParallelVacuumState::shared.

Referenced by dead_items_alloc().

◆ parallel_vacuum_index_is_parallel_safe()

static bool parallel_vacuum_index_is_parallel_safe ( Relation  indrel,
int  num_index_scans,
bool  vacuum 
)
static

Definition at line 951 of file vacuumparallel.c.

953{
954 uint8 vacoptions;
955
956 vacoptions = indrel->rd_indam->amparallelvacuumoptions;
957
958 /* In parallel vacuum case, check if it supports parallel bulk-deletion */
959 if (vacuum)
960 return ((vacoptions & VACUUM_OPTION_PARALLEL_BULKDEL) != 0);
961
962 /* Not safe, if the index does not support parallel cleanup */
963 if (((vacoptions & VACUUM_OPTION_PARALLEL_CLEANUP) == 0) &&
964 ((vacoptions & VACUUM_OPTION_PARALLEL_COND_CLEANUP) == 0))
965 return false;
966
967 /*
968 * Not safe, if the index supports parallel cleanup conditionally, but we
969 * have already processed the index (for bulkdelete). We do this to avoid
970 * the need to invoke workers when parallel index cleanup doesn't need to
971 * scan the index. See the comments for option
972 * VACUUM_OPTION_PARALLEL_COND_CLEANUP to know when indexes support
973 * parallel cleanup conditionally.
974 */
975 if (num_index_scans > 0 &&
976 ((vacoptions & VACUUM_OPTION_PARALLEL_COND_CLEANUP) != 0))
977 return false;
978
979 return true;
980}
void vacuum(List *relations, VacuumParams *params, BufferAccessStrategy bstrategy, MemoryContext vac_context, bool isTopLevel)
Definition: vacuum.c:478

References IndexAmRoutine::amparallelvacuumoptions, RelationData::rd_indam, vacuum(), VACUUM_OPTION_PARALLEL_BULKDEL, VACUUM_OPTION_PARALLEL_CLEANUP, and VACUUM_OPTION_PARALLEL_COND_CLEANUP.

Referenced by parallel_vacuum_process_all_indexes().

◆ parallel_vacuum_init()

ParallelVacuumState * parallel_vacuum_init ( Relation  rel,
Relation indrels,
int  nindexes,
int  nrequested_workers,
int  vac_work_mem,
int  elevel,
BufferAccessStrategy  bstrategy 
)

Definition at line 243 of file vacuumparallel.c.

246{
248 ParallelContext *pcxt;
249 PVShared *shared;
250 TidStore *dead_items;
251 PVIndStats *indstats;
252 BufferUsage *buffer_usage;
253 WalUsage *wal_usage;
254 bool *will_parallel_vacuum;
255 Size est_indstats_len;
256 Size est_shared_len;
257 int nindexes_mwm = 0;
258 int parallel_workers = 0;
259 int querylen;
260
261 /*
262 * A parallel vacuum must be requested and there must be indexes on the
263 * relation
264 */
265 Assert(nrequested_workers >= 0);
266 Assert(nindexes > 0);
267
268 /*
269 * Compute the number of parallel vacuum workers to launch
270 */
271 will_parallel_vacuum = (bool *) palloc0(sizeof(bool) * nindexes);
272 parallel_workers = parallel_vacuum_compute_workers(indrels, nindexes,
273 nrequested_workers,
274 will_parallel_vacuum);
275 if (parallel_workers <= 0)
276 {
277 /* Can't perform vacuum in parallel -- return NULL */
278 pfree(will_parallel_vacuum);
279 return NULL;
280 }
281
283 pvs->indrels = indrels;
284 pvs->nindexes = nindexes;
285 pvs->will_parallel_vacuum = will_parallel_vacuum;
286 pvs->bstrategy = bstrategy;
287 pvs->heaprel = rel;
288
290 pcxt = CreateParallelContext("postgres", "parallel_vacuum_main",
291 parallel_workers);
292 Assert(pcxt->nworkers > 0);
293 pvs->pcxt = pcxt;
294
295 /* Estimate size for index vacuum stats -- PARALLEL_VACUUM_KEY_INDEX_STATS */
296 est_indstats_len = mul_size(sizeof(PVIndStats), nindexes);
297 shm_toc_estimate_chunk(&pcxt->estimator, est_indstats_len);
299
300 /* Estimate size for shared information -- PARALLEL_VACUUM_KEY_SHARED */
301 est_shared_len = sizeof(PVShared);
302 shm_toc_estimate_chunk(&pcxt->estimator, est_shared_len);
304
305 /*
306 * Estimate space for BufferUsage and WalUsage --
307 * PARALLEL_VACUUM_KEY_BUFFER_USAGE and PARALLEL_VACUUM_KEY_WAL_USAGE.
308 *
309 * If there are no extensions loaded that care, we could skip this. We
310 * have no way of knowing whether anyone's looking at pgBufferUsage or
311 * pgWalUsage, so do it unconditionally.
312 */
314 mul_size(sizeof(BufferUsage), pcxt->nworkers));
317 mul_size(sizeof(WalUsage), pcxt->nworkers));
319
320 /* Finally, estimate PARALLEL_VACUUM_KEY_QUERY_TEXT space */
322 {
323 querylen = strlen(debug_query_string);
324 shm_toc_estimate_chunk(&pcxt->estimator, querylen + 1);
326 }
327 else
328 querylen = 0; /* keep compiler quiet */
329
331
332 /* Prepare index vacuum stats */
333 indstats = (PVIndStats *) shm_toc_allocate(pcxt->toc, est_indstats_len);
334 MemSet(indstats, 0, est_indstats_len);
335 for (int i = 0; i < nindexes; i++)
336 {
337 Relation indrel = indrels[i];
338 uint8 vacoptions = indrel->rd_indam->amparallelvacuumoptions;
339
340 /*
341 * Cleanup option should be either disabled, always performing in
342 * parallel or conditionally performing in parallel.
343 */
344 Assert(((vacoptions & VACUUM_OPTION_PARALLEL_CLEANUP) == 0) ||
345 ((vacoptions & VACUUM_OPTION_PARALLEL_COND_CLEANUP) == 0));
347
348 if (!will_parallel_vacuum[i])
349 continue;
350
352 nindexes_mwm++;
353
354 /*
355 * Remember the number of indexes that support parallel operation for
356 * each phase.
357 */
358 if ((vacoptions & VACUUM_OPTION_PARALLEL_BULKDEL) != 0)
360 if ((vacoptions & VACUUM_OPTION_PARALLEL_CLEANUP) != 0)
362 if ((vacoptions & VACUUM_OPTION_PARALLEL_COND_CLEANUP) != 0)
364 }
366 pvs->indstats = indstats;
367
368 /* Prepare shared information */
369 shared = (PVShared *) shm_toc_allocate(pcxt->toc, est_shared_len);
370 MemSet(shared, 0, est_shared_len);
371 shared->relid = RelationGetRelid(rel);
372 shared->elevel = elevel;
375 (nindexes_mwm > 0) ?
376 maintenance_work_mem / Min(parallel_workers, nindexes_mwm) :
378 shared->dead_items_info.max_bytes = vac_work_mem * 1024L;
379
380 /* Prepare DSA space for dead items */
383 pvs->dead_items = dead_items;
384 shared->dead_items_handle = TidStoreGetHandle(dead_items);
386
387 /* Use the same buffer size for all workers */
388 shared->ring_nbuffers = GetAccessStrategyBufferCount(bstrategy);
389
390 pg_atomic_init_u32(&(shared->cost_balance), 0);
391 pg_atomic_init_u32(&(shared->active_nworkers), 0);
392 pg_atomic_init_u32(&(shared->idx), 0);
393
395 pvs->shared = shared;
396
397 /*
398 * Allocate space for each worker's BufferUsage and WalUsage; no need to
399 * initialize
400 */
401 buffer_usage = shm_toc_allocate(pcxt->toc,
402 mul_size(sizeof(BufferUsage), pcxt->nworkers));
404 pvs->buffer_usage = buffer_usage;
405 wal_usage = shm_toc_allocate(pcxt->toc,
406 mul_size(sizeof(WalUsage), pcxt->nworkers));
408 pvs->wal_usage = wal_usage;
409
410 /* Store query string for workers */
412 {
413 char *sharedquery;
414
415 sharedquery = (char *) shm_toc_allocate(pcxt->toc, querylen + 1);
416 memcpy(sharedquery, debug_query_string, querylen + 1);
417 sharedquery[querylen] = '\0';
418 shm_toc_insert(pcxt->toc,
419 PARALLEL_VACUUM_KEY_QUERY_TEXT, sharedquery);
420 }
421
422 /* Success -- return parallel vacuum state */
423 return pvs;
424}
static void pg_atomic_init_u32(volatile pg_atomic_uint32 *ptr, uint32 val)
Definition: atomics.h:221
void InitializeParallelDSM(ParallelContext *pcxt)
Definition: parallel.c:207
ParallelContext * CreateParallelContext(const char *library_name, const char *function_name, int nworkers)
Definition: parallel.c:169
uint64 pgstat_get_my_query_id(void)
#define MemSet(start, val, len)
Definition: c.h:977
size_t Size
Definition: c.h:562
dsa_handle dsa_get_handle(dsa_area *area)
Definition: dsa.c:498
int GetAccessStrategyBufferCount(BufferAccessStrategy strategy)
Definition: freelist.c:624
int maintenance_work_mem
Definition: globals.c:132
@ LWTRANCHE_PARALLEL_VACUUM_DSA
Definition: lwlock.h:217
const char * debug_query_string
Definition: postgres.c:87
#define RelationGetRelid(relation)
Definition: rel.h:505
void * shm_toc_allocate(shm_toc *toc, Size nbytes)
Definition: shm_toc.c:88
void shm_toc_insert(shm_toc *toc, uint64 key, void *address)
Definition: shm_toc.c:171
#define shm_toc_estimate_chunk(e, sz)
Definition: shm_toc.h:51
#define shm_toc_estimate_keys(e, cnt)
Definition: shm_toc.h:53
Size mul_size(Size s1, Size s2)
Definition: shmem.c:505
bool amusemaintenanceworkmem
Definition: amapi.h:263
pg_atomic_uint32 cost_balance
int maintenance_work_mem_worker
pg_atomic_uint32 active_nworkers
dsa_pointer dead_items_handle
dsa_handle dead_items_dsa_handle
uint64 queryid
int ring_nbuffers
pg_atomic_uint32 idx
shm_toc_estimator estimator
Definition: parallel.h:41
shm_toc * toc
Definition: parallel.h:44
BufferAccessStrategy bstrategy
BufferUsage * buffer_usage
size_t max_bytes
Definition: vacuum.h:287
dsa_area * TidStoreGetDSA(TidStore *ts)
Definition: tidstore.c:544
dsa_pointer TidStoreGetHandle(TidStore *ts)
Definition: tidstore.c:552
TidStore * TidStoreCreateShared(size_t max_bytes, int tranche_id)
Definition: tidstore.c:208
#define VACUUM_OPTION_MAX_VALID_VALUE
Definition: vacuum.h:66
static int parallel_vacuum_compute_workers(Relation *indrels, int nindexes, int nrequested, bool *will_parallel_vacuum)
#define PARALLEL_VACUUM_KEY_INDEX_STATS
#define PARALLEL_VACUUM_KEY_QUERY_TEXT
#define PARALLEL_VACUUM_KEY_BUFFER_USAGE
#define PARALLEL_VACUUM_KEY_SHARED
#define PARALLEL_VACUUM_KEY_WAL_USAGE
struct PVShared PVShared
void EnterParallelMode(void)
Definition: xact.c:1050

References PVShared::active_nworkers, IndexAmRoutine::amparallelvacuumoptions, IndexAmRoutine::amusemaintenanceworkmem, Assert, ParallelVacuumState::bstrategy, ParallelVacuumState::buffer_usage, PVShared::cost_balance, CreateParallelContext(), ParallelVacuumState::dead_items, PVShared::dead_items_dsa_handle, PVShared::dead_items_handle, PVShared::dead_items_info, debug_query_string, dsa_get_handle(), PVShared::elevel, EnterParallelMode(), ParallelContext::estimator, GetAccessStrategyBufferCount(), ParallelVacuumState::heaprel, i, PVShared::idx, ParallelVacuumState::indrels, ParallelVacuumState::indstats, InitializeParallelDSM(), LWTRANCHE_PARALLEL_VACUUM_DSA, maintenance_work_mem, PVShared::maintenance_work_mem_worker, VacDeadItemsInfo::max_bytes, MemSet, Min, mul_size(), ParallelVacuumState::nindexes, ParallelVacuumState::nindexes_parallel_bulkdel, ParallelVacuumState::nindexes_parallel_cleanup, ParallelVacuumState::nindexes_parallel_condcleanup, ParallelContext::nworkers, palloc0(), parallel_vacuum_compute_workers(), PARALLEL_VACUUM_KEY_BUFFER_USAGE, PARALLEL_VACUUM_KEY_INDEX_STATS, PARALLEL_VACUUM_KEY_QUERY_TEXT, PARALLEL_VACUUM_KEY_SHARED, PARALLEL_VACUUM_KEY_WAL_USAGE, ParallelVacuumState::pcxt, pfree(), pg_atomic_init_u32(), pgstat_get_my_query_id(), PVShared::queryid, RelationData::rd_indam, RelationGetRelid, PVShared::relid, PVShared::ring_nbuffers, ParallelVacuumState::shared, shm_toc_allocate(), shm_toc_estimate_chunk, shm_toc_estimate_keys, shm_toc_insert(), TidStoreCreateShared(), TidStoreGetDSA(), TidStoreGetHandle(), ParallelContext::toc, VACUUM_OPTION_MAX_VALID_VALUE, VACUUM_OPTION_PARALLEL_BULKDEL, VACUUM_OPTION_PARALLEL_CLEANUP, VACUUM_OPTION_PARALLEL_COND_CLEANUP, ParallelVacuumState::wal_usage, and ParallelVacuumState::will_parallel_vacuum.

Referenced by dead_items_alloc().

◆ parallel_vacuum_main()

void parallel_vacuum_main ( dsm_segment seg,
shm_toc toc 
)

Definition at line 989 of file vacuumparallel.c.

990{
992 Relation rel;
993 Relation *indrels;
994 PVIndStats *indstats;
995 PVShared *shared;
996 TidStore *dead_items;
997 BufferUsage *buffer_usage;
998 WalUsage *wal_usage;
999 int nindexes;
1000 char *sharedquery;
1001 ErrorContextCallback errcallback;
1002
1003 /*
1004 * A parallel vacuum worker must have only PROC_IN_VACUUM flag since we
1005 * don't support parallel vacuum for autovacuum as of now.
1006 */
1008
1009 elog(DEBUG1, "starting parallel vacuum worker");
1010
1011 shared = (PVShared *) shm_toc_lookup(toc, PARALLEL_VACUUM_KEY_SHARED, false);
1012
1013 /* Set debug_query_string for individual workers */
1014 sharedquery = shm_toc_lookup(toc, PARALLEL_VACUUM_KEY_QUERY_TEXT, true);
1015 debug_query_string = sharedquery;
1017
1018 /* Track query ID */
1019 pgstat_report_query_id(shared->queryid, false);
1020
1021 /*
1022 * Open table. The lock mode is the same as the leader process. It's
1023 * okay because the lock mode does not conflict among the parallel
1024 * workers.
1025 */
1027
1028 /*
1029 * Open all indexes. indrels are sorted in order by OID, which should be
1030 * matched to the leader's one.
1031 */
1032 vac_open_indexes(rel, RowExclusiveLock, &nindexes, &indrels);
1033 Assert(nindexes > 0);
1034
1035 if (shared->maintenance_work_mem_worker > 0)
1037
1038 /* Set index statistics */
1039 indstats = (PVIndStats *) shm_toc_lookup(toc,
1041 false);
1042
1043 /* Find dead_items in shared memory */
1044 dead_items = TidStoreAttach(shared->dead_items_dsa_handle,
1045 shared->dead_items_handle);
1046
1047 /* Set cost-based vacuum delay */
1053
1054 /* Set parallel vacuum state */
1055 pvs.indrels = indrels;
1056 pvs.nindexes = nindexes;
1057 pvs.indstats = indstats;
1058 pvs.shared = shared;
1059 pvs.dead_items = dead_items;
1062 pvs.heaprel = rel;
1063
1064 /* These fields will be filled during index vacuum or cleanup */
1065 pvs.indname = NULL;
1067
1068 /* Each parallel VACUUM worker gets its own access strategy. */
1070 shared->ring_nbuffers * (BLCKSZ / 1024));
1071
1072 /* Setup error traceback support for ereport() */
1074 errcallback.arg = &pvs;
1075 errcallback.previous = error_context_stack;
1076 error_context_stack = &errcallback;
1077
1078 /* Prepare to track buffer usage during parallel execution */
1080
1081 /* Process indexes to perform vacuum/cleanup */
1083
1084 /* Report buffer/WAL usage during parallel execution */
1085 buffer_usage = shm_toc_lookup(toc, PARALLEL_VACUUM_KEY_BUFFER_USAGE, false);
1086 wal_usage = shm_toc_lookup(toc, PARALLEL_VACUUM_KEY_WAL_USAGE, false);
1088 &wal_usage[ParallelWorkerNumber]);
1089
1090 TidStoreDetach(dead_items);
1091
1092 /* Pop the error context stack */
1093 error_context_stack = errcallback.previous;
1094
1095 vac_close_indexes(nindexes, indrels, RowExclusiveLock);
1098}
void VacuumUpdateCosts(void)
Definition: autovacuum.c:1651
int ParallelWorkerNumber
Definition: parallel.c:114
void pgstat_report_query_id(uint64 query_id, bool force)
void pgstat_report_activity(BackendState state, const char *cmd_str)
@ STATE_RUNNING
@ BAS_VACUUM
Definition: bufmgr.h:39
ErrorContextCallback * error_context_stack
Definition: elog.c:94
#define DEBUG1
Definition: elog.h:30
#define elog(elevel,...)
Definition: elog.h:225
BufferAccessStrategy GetAccessStrategyWithSize(BufferAccessStrategyType btype, int ring_size_kb)
Definition: freelist.c:584
void FreeAccessStrategy(BufferAccessStrategy strategy)
Definition: freelist.c:681
int VacuumCostBalance
Definition: globals.c:156
void InstrEndParallelQuery(BufferUsage *bufusage, WalUsage *walusage)
Definition: instrument.c:208
void InstrStartParallelQuery(void)
Definition: instrument.c:200
#define ShareUpdateExclusiveLock
Definition: lockdefs.h:39
#define RowExclusiveLock
Definition: lockdefs.h:38
char * get_namespace_name(Oid nspid)
Definition: lsyscache.c:3366
char * pstrdup(const char *in)
Definition: mcxt.c:1696
#define PROC_IN_VACUUM
Definition: proc.h:58
#define RelationGetRelationName(relation)
Definition: rel.h:539
#define RelationGetNamespace(relation)
Definition: rel.h:546
void * shm_toc_lookup(shm_toc *toc, uint64 key, bool noError)
Definition: shm_toc.c:232
PGPROC * MyProc
Definition: proc.c:66
struct ErrorContextCallback * previous
Definition: elog.h:296
void(* callback)(void *arg)
Definition: elog.h:297
uint8 statusFlags
Definition: proc.h:242
void table_close(Relation relation, LOCKMODE lockmode)
Definition: table.c:126
Relation table_open(Oid relationId, LOCKMODE lockmode)
Definition: table.c:40
void TidStoreDetach(TidStore *ts)
Definition: tidstore.c:269
TidStore * TidStoreAttach(dsa_handle area_handle, dsa_pointer handle)
Definition: tidstore.c:244
pg_atomic_uint32 * VacuumActiveNWorkers
Definition: vacuum.c:102
void vac_open_indexes(Relation relation, LOCKMODE lockmode, int *nindexes, Relation **Irel)
Definition: vacuum.c:2297
int VacuumCostBalanceLocal
Definition: vacuum.c:103
void vac_close_indexes(int nindexes, Relation *Irel, LOCKMODE lockmode)
Definition: vacuum.c:2340
pg_atomic_uint32 * VacuumSharedCostBalance
Definition: vacuum.c:101
static void parallel_vacuum_error_callback(void *arg)
static void parallel_vacuum_process_safe_indexes(ParallelVacuumState *pvs)

References PVShared::active_nworkers, ErrorContextCallback::arg, Assert, BAS_VACUUM, ParallelVacuumState::bstrategy, ErrorContextCallback::callback, PVShared::cost_balance, ParallelVacuumState::dead_items, PVShared::dead_items_dsa_handle, PVShared::dead_items_handle, DEBUG1, debug_query_string, elog, error_context_stack, FreeAccessStrategy(), get_namespace_name(), GetAccessStrategyWithSize(), ParallelVacuumState::heaprel, ParallelVacuumState::indname, ParallelVacuumState::indrels, ParallelVacuumState::indstats, InstrEndParallelQuery(), InstrStartParallelQuery(), maintenance_work_mem, PVShared::maintenance_work_mem_worker, MyProc, ParallelVacuumState::nindexes, PARALLEL_INDVAC_STATUS_INITIAL, parallel_vacuum_error_callback(), PARALLEL_VACUUM_KEY_BUFFER_USAGE, PARALLEL_VACUUM_KEY_INDEX_STATS, PARALLEL_VACUUM_KEY_QUERY_TEXT, PARALLEL_VACUUM_KEY_SHARED, PARALLEL_VACUUM_KEY_WAL_USAGE, parallel_vacuum_process_safe_indexes(), ParallelWorkerNumber, pgstat_report_activity(), pgstat_report_query_id(), ErrorContextCallback::previous, PROC_IN_VACUUM, pstrdup(), PVShared::queryid, RelationGetNamespace, RelationGetRelationName, PVShared::relid, ParallelVacuumState::relname, ParallelVacuumState::relnamespace, PVShared::ring_nbuffers, RowExclusiveLock, ParallelVacuumState::shared, ShareUpdateExclusiveLock, shm_toc_lookup(), STATE_RUNNING, ParallelVacuumState::status, PGPROC::statusFlags, table_close(), table_open(), TidStoreAttach(), TidStoreDetach(), vac_close_indexes(), vac_open_indexes(), VacuumActiveNWorkers, VacuumCostBalance, VacuumCostBalanceLocal, VacuumSharedCostBalance, and VacuumUpdateCosts().

◆ parallel_vacuum_process_all_indexes()

static void parallel_vacuum_process_all_indexes ( ParallelVacuumState pvs,
int  num_index_scans,
bool  vacuum 
)
static

Definition at line 611 of file vacuumparallel.c.

613{
614 int nworkers;
615 PVIndVacStatus new_status;
616
618
619 if (vacuum)
620 {
622
623 /* Determine the number of parallel workers to launch */
624 nworkers = pvs->nindexes_parallel_bulkdel;
625 }
626 else
627 {
629
630 /* Determine the number of parallel workers to launch */
631 nworkers = pvs->nindexes_parallel_cleanup;
632
633 /* Add conditionally parallel-aware indexes if in the first time call */
634 if (num_index_scans == 0)
635 nworkers += pvs->nindexes_parallel_condcleanup;
636 }
637
638 /* The leader process will participate */
639 nworkers--;
640
641 /*
642 * It is possible that parallel context is initialized with fewer workers
643 * than the number of indexes that need a separate worker in the current
644 * phase, so we need to consider it. See
645 * parallel_vacuum_compute_workers().
646 */
647 nworkers = Min(nworkers, pvs->pcxt->nworkers);
648
649 /*
650 * Set index vacuum status and mark whether parallel vacuum worker can
651 * process it.
652 */
653 for (int i = 0; i < pvs->nindexes; i++)
654 {
655 PVIndStats *indstats = &(pvs->indstats[i]);
656
658 indstats->status = new_status;
660 (pvs->will_parallel_vacuum[i] &&
662 num_index_scans,
663 vacuum));
664 }
665
666 /* Reset the parallel index processing and progress counters */
667 pg_atomic_write_u32(&(pvs->shared->idx), 0);
668
669 /* Setup the shared cost-based vacuum delay and launch workers */
670 if (nworkers > 0)
671 {
672 /* Reinitialize parallel context to relaunch parallel workers */
673 if (num_index_scans > 0)
675
676 /*
677 * Set up shared cost balance and the number of active workers for
678 * vacuum delay. We need to do this before launching workers as
679 * otherwise, they might not see the updated values for these
680 * parameters.
681 */
684
685 /*
686 * The number of workers can vary between bulkdelete and cleanup
687 * phase.
688 */
689 ReinitializeParallelWorkers(pvs->pcxt, nworkers);
690
692
693 if (pvs->pcxt->nworkers_launched > 0)
694 {
695 /*
696 * Reset the local cost values for leader backend as we have
697 * already accumulated the remaining balance of heap.
698 */
701
702 /* Enable shared cost balance for leader backend */
705 }
706
707 if (vacuum)
708 ereport(pvs->shared->elevel,
709 (errmsg(ngettext("launched %d parallel vacuum worker for index vacuuming (planned: %d)",
710 "launched %d parallel vacuum workers for index vacuuming (planned: %d)",
711 pvs->pcxt->nworkers_launched),
712 pvs->pcxt->nworkers_launched, nworkers)));
713 else
714 ereport(pvs->shared->elevel,
715 (errmsg(ngettext("launched %d parallel vacuum worker for index cleanup (planned: %d)",
716 "launched %d parallel vacuum workers for index cleanup (planned: %d)",
717 pvs->pcxt->nworkers_launched),
718 pvs->pcxt->nworkers_launched, nworkers)));
719 }
720
721 /* Vacuum the indexes that can be processed by only leader process */
723
724 /*
725 * Join as a parallel worker. The leader vacuums alone processes all
726 * parallel-safe indexes in the case where no workers are launched.
727 */
729
730 /*
731 * Next, accumulate buffer and WAL usage. (This must wait for the workers
732 * to finish, or we might get incomplete data.)
733 */
734 if (nworkers > 0)
735 {
736 /* Wait for all vacuum workers to finish */
738
739 for (int i = 0; i < pvs->pcxt->nworkers_launched; i++)
741 }
742
743 /*
744 * Reset all index status back to initial (while checking that we have
745 * vacuumed all indexes).
746 */
747 for (int i = 0; i < pvs->nindexes; i++)
748 {
749 PVIndStats *indstats = &(pvs->indstats[i]);
750
752 elog(ERROR, "parallel index vacuum on index \"%s\" is not completed",
754
756 }
757
758 /*
759 * Carry the shared balance value to heap scan and disable shared costing
760 */
762 {
766 }
767}
static void pg_atomic_write_u32(volatile pg_atomic_uint32 *ptr, uint32 val)
Definition: atomics.h:276
static uint32 pg_atomic_read_u32(volatile pg_atomic_uint32 *ptr)
Definition: atomics.h:239
void WaitForParallelWorkersToFinish(ParallelContext *pcxt)
Definition: parallel.c:792
void LaunchParallelWorkers(ParallelContext *pcxt)
Definition: parallel.c:569
void ReinitializeParallelDSM(ParallelContext *pcxt)
Definition: parallel.c:504
void ReinitializeParallelWorkers(ParallelContext *pcxt, int nworkers_to_launch)
Definition: parallel.c:554
#define ngettext(s, p, n)
Definition: c.h:1138
int errmsg(const char *fmt,...)
Definition: elog.c:1070
#define ERROR
Definition: elog.h:39
#define ereport(elevel,...)
Definition: elog.h:149
void InstrAccumParallelQuery(BufferUsage *bufusage, WalUsage *walusage)
Definition: instrument.c:218
bool parallel_workers_can_process
PVIndVacStatus status
int nworkers_launched
Definition: parallel.h:37
static bool parallel_vacuum_index_is_parallel_safe(Relation indrel, int num_index_scans, bool vacuum)
static void parallel_vacuum_process_unsafe_indexes(ParallelVacuumState *pvs)

References PVShared::active_nworkers, Assert, ParallelVacuumState::buffer_usage, PVShared::cost_balance, PVShared::elevel, elog, ereport, errmsg(), ERROR, i, PVShared::idx, ParallelVacuumState::indrels, ParallelVacuumState::indstats, InstrAccumParallelQuery(), IsParallelWorker, LaunchParallelWorkers(), Min, ngettext, ParallelVacuumState::nindexes, ParallelVacuumState::nindexes_parallel_bulkdel, ParallelVacuumState::nindexes_parallel_cleanup, ParallelVacuumState::nindexes_parallel_condcleanup, ParallelContext::nworkers, ParallelContext::nworkers_launched, PARALLEL_INDVAC_STATUS_COMPLETED, PARALLEL_INDVAC_STATUS_INITIAL, PARALLEL_INDVAC_STATUS_NEED_BULKDELETE, PARALLEL_INDVAC_STATUS_NEED_CLEANUP, parallel_vacuum_index_is_parallel_safe(), parallel_vacuum_process_safe_indexes(), parallel_vacuum_process_unsafe_indexes(), PVIndStats::parallel_workers_can_process, ParallelVacuumState::pcxt, pg_atomic_read_u32(), pg_atomic_write_u32(), ReinitializeParallelDSM(), ReinitializeParallelWorkers(), RelationGetRelationName, ParallelVacuumState::shared, PVIndStats::status, vacuum(), VacuumActiveNWorkers, VacuumCostBalance, VacuumCostBalanceLocal, VacuumSharedCostBalance, WaitForParallelWorkersToFinish(), ParallelVacuumState::wal_usage, and ParallelVacuumState::will_parallel_vacuum.

Referenced by parallel_vacuum_bulkdel_all_indexes(), and parallel_vacuum_cleanup_all_indexes().

◆ parallel_vacuum_process_one_index()

static void parallel_vacuum_process_one_index ( ParallelVacuumState pvs,
Relation  indrel,
PVIndStats indstats 
)
static

Definition at line 865 of file vacuumparallel.c.

867{
868 IndexBulkDeleteResult *istat = NULL;
869 IndexBulkDeleteResult *istat_res;
870 IndexVacuumInfo ivinfo;
871
872 /*
873 * Update the pointer to the corresponding bulk-deletion result if someone
874 * has already updated it
875 */
876 if (indstats->istat_updated)
877 istat = &(indstats->istat);
878
879 ivinfo.index = indrel;
880 ivinfo.heaprel = pvs->heaprel;
881 ivinfo.analyze_only = false;
882 ivinfo.report_progress = false;
883 ivinfo.message_level = DEBUG2;
885 ivinfo.num_heap_tuples = pvs->shared->reltuples;
886 ivinfo.strategy = pvs->bstrategy;
887
888 /* Update error traceback information */
890 pvs->status = indstats->status;
891
892 switch (indstats->status)
893 {
895 istat_res = vac_bulkdel_one_index(&ivinfo, istat, pvs->dead_items,
896 &pvs->shared->dead_items_info);
897 break;
899 istat_res = vac_cleanup_one_index(&ivinfo, istat);
900 break;
901 default:
902 elog(ERROR, "unexpected parallel vacuum index status %d for index \"%s\"",
903 indstats->status,
905 }
906
907 /*
908 * Copy the index bulk-deletion result returned from ambulkdelete and
909 * amvacuumcleanup to the DSM segment if it's the first cycle because they
910 * allocate locally and it's possible that an index will be vacuumed by a
911 * different vacuum process the next cycle. Copying the result normally
912 * happens only the first time an index is vacuumed. For any additional
913 * vacuum pass, we directly point to the result on the DSM segment and
914 * pass it to vacuum index APIs so that workers can update it directly.
915 *
916 * Since all vacuum workers write the bulk-deletion result at different
917 * slots we can write them without locking.
918 */
919 if (!indstats->istat_updated && istat_res != NULL)
920 {
921 memcpy(&(indstats->istat), istat_res, sizeof(IndexBulkDeleteResult));
922 indstats->istat_updated = true;
923
924 /* Free the locally-allocated bulk-deletion result */
925 pfree(istat_res);
926 }
927
928 /*
929 * Update the status to completed. No need to lock here since each worker
930 * touches different indexes.
931 */
933
934 /* Reset error traceback information */
936 pfree(pvs->indname);
937 pvs->indname = NULL;
938
939 /*
940 * Call the parallel variant of pgstat_progress_incr_param so workers can
941 * report progress of index vacuum to the leader.
942 */
944}
void pgstat_progress_parallel_incr_param(int index, int64 incr)
#define DEBUG2
Definition: elog.h:29
#define PROGRESS_VACUUM_INDEXES_PROCESSED
Definition: progress.h:30
Relation index
Definition: genam.h:48
double num_heap_tuples
Definition: genam.h:54
bool analyze_only
Definition: genam.h:50
BufferAccessStrategy strategy
Definition: genam.h:55
Relation heaprel
Definition: genam.h:49
bool report_progress
Definition: genam.h:51
int message_level
Definition: genam.h:53
bool estimated_count
Definition: genam.h:52
IndexBulkDeleteResult * vac_cleanup_one_index(IndexVacuumInfo *ivinfo, IndexBulkDeleteResult *istat)
Definition: vacuum.c:2536
IndexBulkDeleteResult * vac_bulkdel_one_index(IndexVacuumInfo *ivinfo, IndexBulkDeleteResult *istat, TidStore *dead_items, VacDeadItemsInfo *dead_items_info)
Definition: vacuum.c:2515

References IndexVacuumInfo::analyze_only, ParallelVacuumState::bstrategy, ParallelVacuumState::dead_items, PVShared::dead_items_info, DEBUG2, elog, ERROR, PVShared::estimated_count, IndexVacuumInfo::estimated_count, ParallelVacuumState::heaprel, IndexVacuumInfo::heaprel, IndexVacuumInfo::index, ParallelVacuumState::indname, PVIndStats::istat, PVIndStats::istat_updated, IndexVacuumInfo::message_level, IndexVacuumInfo::num_heap_tuples, PARALLEL_INDVAC_STATUS_COMPLETED, PARALLEL_INDVAC_STATUS_NEED_BULKDELETE, PARALLEL_INDVAC_STATUS_NEED_CLEANUP, pfree(), pgstat_progress_parallel_incr_param(), PROGRESS_VACUUM_INDEXES_PROCESSED, pstrdup(), RelationGetRelationName, PVShared::reltuples, IndexVacuumInfo::report_progress, ParallelVacuumState::shared, PVIndStats::status, ParallelVacuumState::status, IndexVacuumInfo::strategy, vac_bulkdel_one_index(), and vac_cleanup_one_index().

Referenced by parallel_vacuum_process_safe_indexes(), and parallel_vacuum_process_unsafe_indexes().

◆ parallel_vacuum_process_safe_indexes()

static void parallel_vacuum_process_safe_indexes ( ParallelVacuumState pvs)
static

Definition at line 774 of file vacuumparallel.c.

775{
776 /*
777 * Increment the active worker count if we are able to launch any worker.
778 */
781
782 /* Loop until all indexes are vacuumed */
783 for (;;)
784 {
785 int idx;
786 PVIndStats *indstats;
787
788 /* Get an index number to process */
789 idx = pg_atomic_fetch_add_u32(&(pvs->shared->idx), 1);
790
791 /* Done for all indexes? */
792 if (idx >= pvs->nindexes)
793 break;
794
795 indstats = &(pvs->indstats[idx]);
796
797 /*
798 * Skip vacuuming index that is unsafe for workers or has an
799 * unsuitable target for parallel index vacuum (this is vacuumed in
800 * parallel_vacuum_process_unsafe_indexes() by the leader).
801 */
802 if (!indstats->parallel_workers_can_process)
803 continue;
804
805 /* Do vacuum or cleanup of the index */
806 parallel_vacuum_process_one_index(pvs, pvs->indrels[idx], indstats);
807 }
808
809 /*
810 * We have completed the index vacuum so decrement the active worker
811 * count.
812 */
815}
Datum idx(PG_FUNCTION_ARGS)
Definition: _int_op.c:259
static uint32 pg_atomic_sub_fetch_u32(volatile pg_atomic_uint32 *ptr, int32 sub_)
Definition: atomics.h:439
static uint32 pg_atomic_fetch_add_u32(volatile pg_atomic_uint32 *ptr, int32 add_)
Definition: atomics.h:366
static uint32 pg_atomic_add_fetch_u32(volatile pg_atomic_uint32 *ptr, int32 add_)
Definition: atomics.h:424
static void parallel_vacuum_process_one_index(ParallelVacuumState *pvs, Relation indrel, PVIndStats *indstats)

References idx(), PVShared::idx, ParallelVacuumState::indrels, ParallelVacuumState::indstats, ParallelVacuumState::nindexes, parallel_vacuum_process_one_index(), PVIndStats::parallel_workers_can_process, pg_atomic_add_fetch_u32(), pg_atomic_fetch_add_u32(), pg_atomic_sub_fetch_u32(), ParallelVacuumState::shared, and VacuumActiveNWorkers.

Referenced by parallel_vacuum_main(), and parallel_vacuum_process_all_indexes().

◆ parallel_vacuum_process_unsafe_indexes()

static void parallel_vacuum_process_unsafe_indexes ( ParallelVacuumState pvs)
static

Definition at line 828 of file vacuumparallel.c.

829{
831
832 /*
833 * Increment the active worker count if we are able to launch any worker.
834 */
837
838 for (int i = 0; i < pvs->nindexes; i++)
839 {
840 PVIndStats *indstats = &(pvs->indstats[i]);
841
842 /* Skip, indexes that are safe for workers */
843 if (indstats->parallel_workers_can_process)
844 continue;
845
846 /* Do vacuum or cleanup of the index */
847 parallel_vacuum_process_one_index(pvs, pvs->indrels[i], indstats);
848 }
849
850 /*
851 * We have completed the index vacuum so decrement the active worker
852 * count.
853 */
856}

References Assert, i, ParallelVacuumState::indrels, ParallelVacuumState::indstats, IsParallelWorker, ParallelVacuumState::nindexes, parallel_vacuum_process_one_index(), PVIndStats::parallel_workers_can_process, pg_atomic_add_fetch_u32(), pg_atomic_sub_fetch_u32(), and VacuumActiveNWorkers.

Referenced by parallel_vacuum_process_all_indexes().

◆ parallel_vacuum_reset_dead_items()

void parallel_vacuum_reset_dead_items ( ParallelVacuumState pvs)

Definition at line 475 of file vacuumparallel.c.

476{
477 VacDeadItemsInfo *dead_items_info = &(pvs->shared->dead_items_info);
478
479 /*
480 * Free the current tidstore and return allocated DSA segments to the
481 * operating system. Then we recreate the tidstore with the same max_bytes
482 * limitation we just used.
483 */
485 pvs->dead_items = TidStoreCreateShared(dead_items_info->max_bytes,
487
488 /* Update the DSA pointer for dead_items to the new one */
491
492 /* Reset the counter */
493 dead_items_info->num_items = 0;
494}
int64 num_items
Definition: vacuum.h:288

References ParallelVacuumState::dead_items, PVShared::dead_items_dsa_handle, PVShared::dead_items_handle, PVShared::dead_items_info, dsa_get_handle(), LWTRANCHE_PARALLEL_VACUUM_DSA, VacDeadItemsInfo::max_bytes, VacDeadItemsInfo::num_items, ParallelVacuumState::shared, TidStoreCreateShared(), TidStoreDestroy(), TidStoreGetDSA(), and TidStoreGetHandle().

Referenced by dead_items_reset().