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 "storage/proc.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)
ParallelVacuumState *	parallel_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)
TidStore *	parallel_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 50 of file vacuumparallel.c.

PARALLEL_VACUUM_KEY_INDEX_STATS

#define PARALLEL_VACUUM_KEY_INDEX_STATS   5

Definition at line 52 of file vacuumparallel.c.

PARALLEL_VACUUM_KEY_QUERY_TEXT

#define PARALLEL_VACUUM_KEY_QUERY_TEXT   2

Definition at line 49 of file vacuumparallel.c.

PARALLEL_VACUUM_KEY_SHARED

#define PARALLEL_VACUUM_KEY_SHARED   1

Definition at line 48 of file vacuumparallel.c.

PARALLEL_VACUUM_KEY_WAL_USAGE

#define PARALLEL_VACUUM_KEY_WAL_USAGE   4

Definition at line 51 of file vacuumparallel.c.

Typedef Documentation

PVIndStats

typedef struct PVIndStats PVIndStats

PVIndVacStatus

typedef enum PVIndVacStatus PVIndVacStatus

PVShared

typedef struct PVShared PVShared

Enumeration Type Documentation

PVIndVacStatus

enum PVIndVacStatus

Enumerator
PARALLEL_INDVAC_STATUS_INITIAL
PARALLEL_INDVAC_STATUS_NEED_BULKDELETE
PARALLEL_INDVAC_STATUS_NEED_CLEANUP
PARALLEL_INDVAC_STATUS_COMPLETED

Definition at line 126 of file vacuumparallel.c.

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

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 501 of file vacuumparallel.c.

{
    Assert(!IsParallelWorker());
 
    /*
     * We can only provide an approximate value of num_heap_tuples, at least
     * for now.
     */
    pvs->shared->reltuples = num_table_tuples;
    pvs->shared->estimated_count = true;
 
    parallel_vacuum_process_all_indexes(pvs, num_index_scans, true);
}

References Assert, PVShared::estimated_count, fb(), 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 520 of file vacuumparallel.c.

{
    Assert(!IsParallelWorker());
 
    /*
     * We can provide a better estimate of total number of surviving tuples
     * (we assume indexes are more interested in that than in the number of
     * nominally live tuples).
     */
    pvs->shared->reltuples = num_table_tuples;
    pvs->shared->estimated_count = estimated_count;
 
    parallel_vacuum_process_all_indexes(pvs, num_index_scans, false);
}

References Assert, PVShared::estimated_count, fb(), 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 550 of file vacuumparallel.c.

{
    int         nindexes_parallel = 0;
    int         nindexes_parallel_bulkdel = 0;
    int         nindexes_parallel_cleanup = 0;
    int         parallel_workers;
 
    /*
     * We don't allow performing parallel operation in standalone backend or
     * when parallelism is disabled.
     */
    if (!IsUnderPostmaster || max_parallel_maintenance_workers == 0)
        return 0;
 
    /*
     * Compute the number of indexes that can participate in parallel vacuum.
     */
    for (int i = 0; i < nindexes; i++)
    {
        Relation    indrel = indrels[i];
        uint8       vacoptions = indrel->rd_indam->amparallelvacuumoptions;
 
        /* Skip index that is not a suitable target for parallel index vacuum */
        if (vacoptions == VACUUM_OPTION_NO_PARALLEL ||
            RelationGetNumberOfBlocks(indrel) < min_parallel_index_scan_size)
            continue;
 
        will_parallel_vacuum[i] = true;
 
        if ((vacoptions & VACUUM_OPTION_PARALLEL_BULKDEL) != 0)
            nindexes_parallel_bulkdel++;
        if (((vacoptions & VACUUM_OPTION_PARALLEL_CLEANUP) != 0) ||
            ((vacoptions & VACUUM_OPTION_PARALLEL_COND_CLEANUP) != 0))
            nindexes_parallel_cleanup++;
    }
 
    nindexes_parallel = Max(nindexes_parallel_bulkdel,
                            nindexes_parallel_cleanup);
 
    /* The leader process takes one index */
    nindexes_parallel--;
 
    /* No index supports parallel vacuum */
    if (nindexes_parallel <= 0)
        return 0;
 
    /* Compute the parallel degree */
    parallel_workers = (nrequested > 0) ?
        Min(nrequested, nindexes_parallel) : nindexes_parallel;
 
    /* Cap by max_parallel_maintenance_workers */
    parallel_workers = Min(parallel_workers, max_parallel_maintenance_workers);
 
    return parallel_workers;
}

References IndexAmRoutine::amparallelvacuumoptions, fb(), 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 437 of file vacuumparallel.c.

{
    Assert(!IsParallelWorker());
 
    /* Copy the updated statistics */
    for (int i = 0; i < pvs->nindexes; i++)
    {
        PVIndStats *indstats = &(pvs->indstats[i]);
 
        if (indstats->istat_updated)
        {
            istats[i] = palloc0_object(IndexBulkDeleteResult);
            memcpy(istats[i], &indstats->istat, sizeof(IndexBulkDeleteResult));
        }
        else
            istats[i] = NULL;
    }
 
    TidStoreDestroy(pvs->dead_items);
 
    DestroyParallelContext(pvs->pcxt);
    ExitParallelMode();
 
    pfree(pvs->will_parallel_vacuum);
    pfree(pvs);
}

References Assert, ParallelVacuumState::dead_items, DestroyParallelContext(), ExitParallelMode(), fb(), i, ParallelVacuumState::indstats, IsParallelWorker, PVIndStats::istat, PVIndStats::istat_updated, ParallelVacuumState::nindexes, palloc0_object, 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 1120 of file vacuumparallel.c.

{
    ParallelVacuumState *errinfo = arg;
 
    switch (errinfo->status)
    {
        case PARALLEL_INDVAC_STATUS_NEED_BULKDELETE:
            errcontext("while vacuuming index \"%s\" of relation \"%s.%s\"",
                       errinfo->indname,
                       errinfo->relnamespace,
                       errinfo->relname);
            break;
        case PARALLEL_INDVAC_STATUS_NEED_CLEANUP:
            errcontext("while cleaning up index \"%s\" of relation \"%s.%s\"",
                       errinfo->indname,
                       errinfo->relnamespace,
                       errinfo->relname);
            break;
        case PARALLEL_INDVAC_STATUS_INITIAL:
        case PARALLEL_INDVAC_STATUS_COMPLETED:
        default:
            return;
    }
}

References arg, errcontext, fb(), PARALLEL_INDVAC_STATUS_COMPLETED, PARALLEL_INDVAC_STATUS_INITIAL, PARALLEL_INDVAC_STATUS_NEED_BULKDELETE, and PARALLEL_INDVAC_STATUS_NEED_CLEANUP.

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 468 of file vacuumparallel.c.

{
    *dead_items_info_p = &(pvs->shared->dead_items_info);
    return pvs->dead_items;
}

References ParallelVacuumState::dead_items, PVShared::dead_items_info, fb(), and ParallelVacuumState::shared.

Referenced by dead_items_alloc(), and dead_items_reset().

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 952 of file vacuumparallel.c.

{
    uint8       vacoptions;
 
    vacoptions = indrel->rd_indam->amparallelvacuumoptions;
 
    /* In parallel vacuum case, check if it supports parallel bulk-deletion */
    if (vacuum)
        return ((vacoptions & VACUUM_OPTION_PARALLEL_BULKDEL) != 0);
 
    /* Not safe, if the index does not support parallel cleanup */
    if (((vacoptions & VACUUM_OPTION_PARALLEL_CLEANUP) == 0) &&
        ((vacoptions & VACUUM_OPTION_PARALLEL_COND_CLEANUP) == 0))
        return false;
 
    /*
     * Not safe, if the index supports parallel cleanup conditionally, but we
     * have already processed the index (for bulkdelete).  We do this to avoid
     * the need to invoke workers when parallel index cleanup doesn't need to
     * scan the index.  See the comments for option
     * VACUUM_OPTION_PARALLEL_COND_CLEANUP to know when indexes support
     * parallel cleanup conditionally.
     */
    if (num_index_scans > 0 &&
        ((vacoptions & VACUUM_OPTION_PARALLEL_COND_CLEANUP) != 0))
        return false;
 
    return true;
}

References fb(), 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 244 of file vacuumparallel.c.

{
    ParallelVacuumState *pvs;
    ParallelContext *pcxt;
    PVShared   *shared;
    TidStore   *dead_items;
    PVIndStats *indstats;
    BufferUsage *buffer_usage;
    WalUsage   *wal_usage;
    bool       *will_parallel_vacuum;
    Size        est_indstats_len;
    Size        est_shared_len;
    int         nindexes_mwm = 0;
    int         parallel_workers = 0;
    int         querylen;
 
    /*
     * A parallel vacuum must be requested and there must be indexes on the
     * relation
     */
    Assert(nrequested_workers >= 0);
    Assert(nindexes > 0);
 
    /*
     * Compute the number of parallel vacuum workers to launch
     */
    will_parallel_vacuum = palloc0_array(bool, nindexes);
    parallel_workers = parallel_vacuum_compute_workers(indrels, nindexes,
                                                       nrequested_workers,
                                                       will_parallel_vacuum);
    if (parallel_workers <= 0)
    {
        /* Can't perform vacuum in parallel -- return NULL */
        pfree(will_parallel_vacuum);
        return NULL;
    }
 
    pvs = palloc0_object(ParallelVacuumState);
    pvs->indrels = indrels;
    pvs->nindexes = nindexes;
    pvs->will_parallel_vacuum = will_parallel_vacuum;
    pvs->bstrategy = bstrategy;
    pvs->heaprel = rel;
 
    EnterParallelMode();
    pcxt = CreateParallelContext("postgres", "parallel_vacuum_main",
                                 parallel_workers);
    Assert(pcxt->nworkers > 0);
    pvs->pcxt = pcxt;
 
    /* Estimate size for index vacuum stats -- PARALLEL_VACUUM_KEY_INDEX_STATS */
    est_indstats_len = mul_size(sizeof(PVIndStats), nindexes);
    shm_toc_estimate_chunk(&pcxt->estimator, est_indstats_len);
    shm_toc_estimate_keys(&pcxt->estimator, 1);
 
    /* Estimate size for shared information -- PARALLEL_VACUUM_KEY_SHARED */
    est_shared_len = sizeof(PVShared);
    shm_toc_estimate_chunk(&pcxt->estimator, est_shared_len);
    shm_toc_estimate_keys(&pcxt->estimator, 1);
 
    /*
     * Estimate space for BufferUsage and WalUsage --
     * PARALLEL_VACUUM_KEY_BUFFER_USAGE and PARALLEL_VACUUM_KEY_WAL_USAGE.
     *
     * If there are no extensions loaded that care, we could skip this.  We
     * have no way of knowing whether anyone's looking at pgBufferUsage or
     * pgWalUsage, so do it unconditionally.
     */
    shm_toc_estimate_chunk(&pcxt->estimator,
                           mul_size(sizeof(BufferUsage), pcxt->nworkers));
    shm_toc_estimate_keys(&pcxt->estimator, 1);
    shm_toc_estimate_chunk(&pcxt->estimator,
                           mul_size(sizeof(WalUsage), pcxt->nworkers));
    shm_toc_estimate_keys(&pcxt->estimator, 1);
 
    /* Finally, estimate PARALLEL_VACUUM_KEY_QUERY_TEXT space */
    if (debug_query_string)
    {
        querylen = strlen(debug_query_string);
        shm_toc_estimate_chunk(&pcxt->estimator, querylen + 1);
        shm_toc_estimate_keys(&pcxt->estimator, 1);
    }
    else
        querylen = 0;           /* keep compiler quiet */
 
    InitializeParallelDSM(pcxt);
 
    /* Prepare index vacuum stats */
    indstats = (PVIndStats *) shm_toc_allocate(pcxt->toc, est_indstats_len);
    MemSet(indstats, 0, est_indstats_len);
    for (int i = 0; i < nindexes; i++)
    {
        Relation    indrel = indrels[i];
        uint8       vacoptions = indrel->rd_indam->amparallelvacuumoptions;
 
        /*
         * Cleanup option should be either disabled, always performing in
         * parallel or conditionally performing in parallel.
         */
        Assert(((vacoptions & VACUUM_OPTION_PARALLEL_CLEANUP) == 0) ||
               ((vacoptions & VACUUM_OPTION_PARALLEL_COND_CLEANUP) == 0));
        Assert(vacoptions <= VACUUM_OPTION_MAX_VALID_VALUE);
 
        if (!will_parallel_vacuum[i])
            continue;
 
        if (indrel->rd_indam->amusemaintenanceworkmem)
            nindexes_mwm++;
 
        /*
         * Remember the number of indexes that support parallel operation for
         * each phase.
         */
        if ((vacoptions & VACUUM_OPTION_PARALLEL_BULKDEL) != 0)
            pvs->nindexes_parallel_bulkdel++;
        if ((vacoptions & VACUUM_OPTION_PARALLEL_CLEANUP) != 0)
            pvs->nindexes_parallel_cleanup++;
        if ((vacoptions & VACUUM_OPTION_PARALLEL_COND_CLEANUP) != 0)
            pvs->nindexes_parallel_condcleanup++;
    }
    shm_toc_insert(pcxt->toc, PARALLEL_VACUUM_KEY_INDEX_STATS, indstats);
    pvs->indstats = indstats;
 
    /* Prepare shared information */
    shared = (PVShared *) shm_toc_allocate(pcxt->toc, est_shared_len);
    MemSet(shared, 0, est_shared_len);
    shared->relid = RelationGetRelid(rel);
    shared->elevel = elevel;
    shared->queryid = pgstat_get_my_query_id();
    shared->maintenance_work_mem_worker =
        (nindexes_mwm > 0) ?
        maintenance_work_mem / Min(parallel_workers, nindexes_mwm) :
        maintenance_work_mem;
    shared->dead_items_info.max_bytes = vac_work_mem * (size_t) 1024;
 
    /* Prepare DSA space for dead items */
    dead_items = TidStoreCreateShared(shared->dead_items_info.max_bytes,
                                      LWTRANCHE_PARALLEL_VACUUM_DSA);
    pvs->dead_items = dead_items;
    shared->dead_items_handle = TidStoreGetHandle(dead_items);
    shared->dead_items_dsa_handle = dsa_get_handle(TidStoreGetDSA(dead_items));
 
    /* Use the same buffer size for all workers */
    shared->ring_nbuffers = GetAccessStrategyBufferCount(bstrategy);
 
    pg_atomic_init_u32(&(shared->cost_balance), 0);
    pg_atomic_init_u32(&(shared->active_nworkers), 0);
    pg_atomic_init_u32(&(shared->idx), 0);
 
    shm_toc_insert(pcxt->toc, PARALLEL_VACUUM_KEY_SHARED, shared);
    pvs->shared = shared;
 
    /*
     * Allocate space for each worker's BufferUsage and WalUsage; no need to
     * initialize
     */
    buffer_usage = shm_toc_allocate(pcxt->toc,
                                    mul_size(sizeof(BufferUsage), pcxt->nworkers));
    shm_toc_insert(pcxt->toc, PARALLEL_VACUUM_KEY_BUFFER_USAGE, buffer_usage);
    pvs->buffer_usage = buffer_usage;
    wal_usage = shm_toc_allocate(pcxt->toc,
                                 mul_size(sizeof(WalUsage), pcxt->nworkers));
    shm_toc_insert(pcxt->toc, PARALLEL_VACUUM_KEY_WAL_USAGE, wal_usage);
    pvs->wal_usage = wal_usage;
 
    /* Store query string for workers */
    if (debug_query_string)
    {
        char       *sharedquery;
 
        sharedquery = (char *) shm_toc_allocate(pcxt->toc, querylen + 1);
        memcpy(sharedquery, debug_query_string, querylen + 1);
        sharedquery[querylen] = '\0';
        shm_toc_insert(pcxt->toc,
                       PARALLEL_VACUUM_KEY_QUERY_TEXT, sharedquery);
    }
 
    /* Success -- return parallel vacuum state */
    return pvs;
}

References PVShared::active_nworkers, IndexAmRoutine::amparallelvacuumoptions, 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, fb(), GetAccessStrategyBufferCount(), ParallelVacuumState::heaprel, i, PVShared::idx, ParallelVacuumState::indrels, ParallelVacuumState::indstats, InitializeParallelDSM(), 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_array, palloc0_object, 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 990 of file vacuumparallel.c.

{
    ParallelVacuumState pvs;
    Relation    rel;
    Relation   *indrels;
    PVIndStats *indstats;
    PVShared   *shared;
    TidStore   *dead_items;
    BufferUsage *buffer_usage;
    WalUsage   *wal_usage;
    int         nindexes;
    char       *sharedquery;
    ErrorContextCallback errcallback;
 
    /*
     * A parallel vacuum worker must have only PROC_IN_VACUUM flag since we
     * don't support parallel vacuum for autovacuum as of now.
     */
    Assert(MyProc->statusFlags == PROC_IN_VACUUM);
 
    elog(DEBUG1, "starting parallel vacuum worker");
 
    shared = (PVShared *) shm_toc_lookup(toc, PARALLEL_VACUUM_KEY_SHARED, false);
 
    /* Set debug_query_string for individual workers */
    sharedquery = shm_toc_lookup(toc, PARALLEL_VACUUM_KEY_QUERY_TEXT, true);
    debug_query_string = sharedquery;
    pgstat_report_activity(STATE_RUNNING, debug_query_string);
 
    /* Track query ID */
    pgstat_report_query_id(shared->queryid, false);
 
    /*
     * Open table.  The lock mode is the same as the leader process.  It's
     * okay because the lock mode does not conflict among the parallel
     * workers.
     */
    rel = table_open(shared->relid, ShareUpdateExclusiveLock);
 
    /*
     * Open all indexes. indrels are sorted in order by OID, which should be
     * matched to the leader's one.
     */
    vac_open_indexes(rel, RowExclusiveLock, &nindexes, &indrels);
    Assert(nindexes > 0);
 
    /*
     * Apply the desired value of maintenance_work_mem within this process.
     * Really we should use SetConfigOption() to change a GUC, but since we're
     * already in parallel mode guc.c would complain about that.  Fortunately,
     * by the same token guc.c will not let any user-defined code change it.
     * So just avert your eyes while we do this:
     */
    if (shared->maintenance_work_mem_worker > 0)
        maintenance_work_mem = shared->maintenance_work_mem_worker;
 
    /* Set index statistics */
    indstats = (PVIndStats *) shm_toc_lookup(toc,
                                             PARALLEL_VACUUM_KEY_INDEX_STATS,
                                             false);
 
    /* Find dead_items in shared memory */
    dead_items = TidStoreAttach(shared->dead_items_dsa_handle,
                                shared->dead_items_handle);
 
    /* Set cost-based vacuum delay */
    VacuumUpdateCosts();
    VacuumCostBalance = 0;
    VacuumCostBalanceLocal = 0;
    VacuumSharedCostBalance = &(shared->cost_balance);
    VacuumActiveNWorkers = &(shared->active_nworkers);
 
    /* Set parallel vacuum state */
    pvs.indrels = indrels;
    pvs.nindexes = nindexes;
    pvs.indstats = indstats;
    pvs.shared = shared;
    pvs.dead_items = dead_items;
    pvs.relnamespace = get_namespace_name(RelationGetNamespace(rel));
    pvs.relname = pstrdup(RelationGetRelationName(rel));
    pvs.heaprel = rel;
 
    /* These fields will be filled during index vacuum or cleanup */
    pvs.indname = NULL;
    pvs.status = PARALLEL_INDVAC_STATUS_INITIAL;
 
    /* Each parallel VACUUM worker gets its own access strategy. */
    pvs.bstrategy = GetAccessStrategyWithSize(BAS_VACUUM,
                                              shared->ring_nbuffers * (BLCKSZ / 1024));
 
    /* Setup error traceback support for ereport() */
    errcallback.callback = parallel_vacuum_error_callback;
    errcallback.arg = &pvs;
    errcallback.previous = error_context_stack;
    error_context_stack = &errcallback;
 
    /* Prepare to track buffer usage during parallel execution */
    InstrStartParallelQuery();
 
    /* Process indexes to perform vacuum/cleanup */
    parallel_vacuum_process_safe_indexes(&pvs);
 
    /* Report buffer/WAL usage during parallel execution */
    buffer_usage = shm_toc_lookup(toc, PARALLEL_VACUUM_KEY_BUFFER_USAGE, false);
    wal_usage = shm_toc_lookup(toc, PARALLEL_VACUUM_KEY_WAL_USAGE, false);
    InstrEndParallelQuery(&buffer_usage[ParallelWorkerNumber],
                          &wal_usage[ParallelWorkerNumber]);
 
    /* Report any remaining cost-based vacuum delay time */
    if (track_cost_delay_timing)
        pgstat_progress_parallel_incr_param(PROGRESS_VACUUM_DELAY_TIME,
                                            parallel_vacuum_worker_delay_ns);
 
    TidStoreDetach(dead_items);
 
    /* Pop the error context stack */
    error_context_stack = errcallback.previous;
 
    vac_close_indexes(nindexes, indrels, RowExclusiveLock);
    table_close(rel, ShareUpdateExclusiveLock);
    FreeAccessStrategy(pvs.bstrategy);
}

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, fb(), 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(), parallel_vacuum_worker_delay_ns, ParallelWorkerNumber, pgstat_progress_parallel_incr_param(), pgstat_report_activity(), pgstat_report_query_id(), ErrorContextCallback::previous, PROC_IN_VACUUM, PROGRESS_VACUUM_DELAY_TIME, 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(), track_cost_delay_timing, 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 612 of file vacuumparallel.c.

{
    int         nworkers;
    PVIndVacStatus new_status;
 
    Assert(!IsParallelWorker());
 
    if (vacuum)
    {
        new_status = PARALLEL_INDVAC_STATUS_NEED_BULKDELETE;
 
        /* Determine the number of parallel workers to launch */
        nworkers = pvs->nindexes_parallel_bulkdel;
    }
    else
    {
        new_status = PARALLEL_INDVAC_STATUS_NEED_CLEANUP;
 
        /* Determine the number of parallel workers to launch */
        nworkers = pvs->nindexes_parallel_cleanup;
 
        /* Add conditionally parallel-aware indexes if in the first time call */
        if (num_index_scans == 0)
            nworkers += pvs->nindexes_parallel_condcleanup;
    }
 
    /* The leader process will participate */
    nworkers--;
 
    /*
     * It is possible that parallel context is initialized with fewer workers
     * than the number of indexes that need a separate worker in the current
     * phase, so we need to consider it.  See
     * parallel_vacuum_compute_workers().
     */
    nworkers = Min(nworkers, pvs->pcxt->nworkers);
 
    /*
     * Set index vacuum status and mark whether parallel vacuum worker can
     * process it.
     */
    for (int i = 0; i < pvs->nindexes; i++)
    {
        PVIndStats *indstats = &(pvs->indstats[i]);
 
        Assert(indstats->status == PARALLEL_INDVAC_STATUS_INITIAL);
        indstats->status = new_status;
        indstats->parallel_workers_can_process =
            (pvs->will_parallel_vacuum[i] &&
             parallel_vacuum_index_is_parallel_safe(pvs->indrels[i],
                                                    num_index_scans,
                                                    vacuum));
    }
 
    /* Reset the parallel index processing and progress counters */
    pg_atomic_write_u32(&(pvs->shared->idx), 0);
 
    /* Setup the shared cost-based vacuum delay and launch workers */
    if (nworkers > 0)
    {
        /* Reinitialize parallel context to relaunch parallel workers */
        if (num_index_scans > 0)
            ReinitializeParallelDSM(pvs->pcxt);
 
        /*
         * Set up shared cost balance and the number of active workers for
         * vacuum delay.  We need to do this before launching workers as
         * otherwise, they might not see the updated values for these
         * parameters.
         */
        pg_atomic_write_u32(&(pvs->shared->cost_balance), VacuumCostBalance);
        pg_atomic_write_u32(&(pvs->shared->active_nworkers), 0);
 
        /*
         * The number of workers can vary between bulkdelete and cleanup
         * phase.
         */
        ReinitializeParallelWorkers(pvs->pcxt, nworkers);
 
        LaunchParallelWorkers(pvs->pcxt);
 
        if (pvs->pcxt->nworkers_launched > 0)
        {
            /*
             * Reset the local cost values for leader backend as we have
             * already accumulated the remaining balance of heap.
             */
            VacuumCostBalance = 0;
            VacuumCostBalanceLocal = 0;
 
            /* Enable shared cost balance for leader backend */
            VacuumSharedCostBalance = &(pvs->shared->cost_balance);
            VacuumActiveNWorkers = &(pvs->shared->active_nworkers);
        }
 
        if (vacuum)
            ereport(pvs->shared->elevel,
                    (errmsg(ngettext("launched %d parallel vacuum worker for index vacuuming (planned: %d)",
                                     "launched %d parallel vacuum workers for index vacuuming (planned: %d)",
                                     pvs->pcxt->nworkers_launched),
                            pvs->pcxt->nworkers_launched, nworkers)));
        else
            ereport(pvs->shared->elevel,
                    (errmsg(ngettext("launched %d parallel vacuum worker for index cleanup (planned: %d)",
                                     "launched %d parallel vacuum workers for index cleanup (planned: %d)",
                                     pvs->pcxt->nworkers_launched),
                            pvs->pcxt->nworkers_launched, nworkers)));
    }
 
    /* Vacuum the indexes that can be processed by only leader process */
    parallel_vacuum_process_unsafe_indexes(pvs);
 
    /*
     * Join as a parallel worker.  The leader vacuums alone processes all
     * parallel-safe indexes in the case where no workers are launched.
     */
    parallel_vacuum_process_safe_indexes(pvs);
 
    /*
     * Next, accumulate buffer and WAL usage.  (This must wait for the workers
     * to finish, or we might get incomplete data.)
     */
    if (nworkers > 0)
    {
        /* Wait for all vacuum workers to finish */
        WaitForParallelWorkersToFinish(pvs->pcxt);
 
        for (int i = 0; i < pvs->pcxt->nworkers_launched; i++)
            InstrAccumParallelQuery(&pvs->buffer_usage[i], &pvs->wal_usage[i]);
    }
 
    /*
     * Reset all index status back to initial (while checking that we have
     * vacuumed all indexes).
     */
    for (int i = 0; i < pvs->nindexes; i++)
    {
        PVIndStats *indstats = &(pvs->indstats[i]);
 
        if (indstats->status != PARALLEL_INDVAC_STATUS_COMPLETED)
            elog(ERROR, "parallel index vacuum on index \"%s\" is not completed",
                 RelationGetRelationName(pvs->indrels[i]));
 
        indstats->status = PARALLEL_INDVAC_STATUS_INITIAL;
    }
 
    /*
     * Carry the shared balance value to heap scan and disable shared costing
     */
    if (VacuumSharedCostBalance)
    {
        VacuumCostBalance = pg_atomic_read_u32(VacuumSharedCostBalance);
        VacuumSharedCostBalance = NULL;
        VacuumActiveNWorkers = NULL;
    }
}

References PVShared::active_nworkers, Assert, ParallelVacuumState::buffer_usage, PVShared::cost_balance, PVShared::elevel, elog, ereport, errmsg, ERROR, fb(), 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 866 of file vacuumparallel.c.

{
    IndexBulkDeleteResult *istat = NULL;
    IndexBulkDeleteResult *istat_res;
    IndexVacuumInfo ivinfo;
 
    /*
     * Update the pointer to the corresponding bulk-deletion result if someone
     * has already updated it
     */
    if (indstats->istat_updated)
        istat = &(indstats->istat);
 
    ivinfo.index = indrel;
    ivinfo.heaprel = pvs->heaprel;
    ivinfo.analyze_only = false;
    ivinfo.report_progress = false;
    ivinfo.message_level = DEBUG2;
    ivinfo.estimated_count = pvs->shared->estimated_count;
    ivinfo.num_heap_tuples = pvs->shared->reltuples;
    ivinfo.strategy = pvs->bstrategy;
 
    /* Update error traceback information */
    pvs->indname = pstrdup(RelationGetRelationName(indrel));
    pvs->status = indstats->status;
 
    switch (indstats->status)
    {
        case PARALLEL_INDVAC_STATUS_NEED_BULKDELETE:
            istat_res = vac_bulkdel_one_index(&ivinfo, istat, pvs->dead_items,
                                              &pvs->shared->dead_items_info);
            break;
        case PARALLEL_INDVAC_STATUS_NEED_CLEANUP:
            istat_res = vac_cleanup_one_index(&ivinfo, istat);
            break;
        default:
            elog(ERROR, "unexpected parallel vacuum index status %d for index \"%s\"",
                 indstats->status,
                 RelationGetRelationName(indrel));
    }
 
    /*
     * Copy the index bulk-deletion result returned from ambulkdelete and
     * amvacuumcleanup to the DSM segment if it's the first cycle because they
     * allocate locally and it's possible that an index will be vacuumed by a
     * different vacuum process the next cycle.  Copying the result normally
     * happens only the first time an index is vacuumed.  For any additional
     * vacuum pass, we directly point to the result on the DSM segment and
     * pass it to vacuum index APIs so that workers can update it directly.
     *
     * Since all vacuum workers write the bulk-deletion result at different
     * slots we can write them without locking.
     */
    if (!indstats->istat_updated && istat_res != NULL)
    {
        memcpy(&(indstats->istat), istat_res, sizeof(IndexBulkDeleteResult));
        indstats->istat_updated = true;
 
        /* Free the locally-allocated bulk-deletion result */
        pfree(istat_res);
    }
 
    /*
     * Update the status to completed. No need to lock here since each worker
     * touches different indexes.
     */
    indstats->status = PARALLEL_INDVAC_STATUS_COMPLETED;
 
    /* Reset error traceback information */
    pvs->status = PARALLEL_INDVAC_STATUS_COMPLETED;
    pfree(pvs->indname);
    pvs->indname = NULL;
 
    /*
     * Call the parallel variant of pgstat_progress_incr_param so workers can
     * report progress of index vacuum to the leader.
     */
    pgstat_progress_parallel_incr_param(PROGRESS_VACUUM_INDEXES_PROCESSED, 1);
}

References ParallelVacuumState::bstrategy, ParallelVacuumState::dead_items, PVShared::dead_items_info, DEBUG2, elog, ERROR, PVShared::estimated_count, fb(), ParallelVacuumState::heaprel, ParallelVacuumState::indname, PVIndStats::istat, PVIndStats::istat_updated, 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, ParallelVacuumState::shared, PVIndStats::status, ParallelVacuumState::status, 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 775 of file vacuumparallel.c.

{
    /*
     * Increment the active worker count if we are able to launch any worker.
     */
    if (VacuumActiveNWorkers)
        pg_atomic_add_fetch_u32(VacuumActiveNWorkers, 1);
 
    /* Loop until all indexes are vacuumed */
    for (;;)
    {
        int         idx;
        PVIndStats *indstats;
 
        /* Get an index number to process */
        idx = pg_atomic_fetch_add_u32(&(pvs->shared->idx), 1);
 
        /* Done for all indexes? */
        if (idx >= pvs->nindexes)
            break;
 
        indstats = &(pvs->indstats[idx]);
 
        /*
         * Skip vacuuming index that is unsafe for workers or has an
         * unsuitable target for parallel index vacuum (this is vacuumed in
         * parallel_vacuum_process_unsafe_indexes() by the leader).
         */
        if (!indstats->parallel_workers_can_process)
            continue;
 
        /* Do vacuum or cleanup of the index */
        parallel_vacuum_process_one_index(pvs, pvs->indrels[idx], indstats);
    }
 
    /*
     * We have completed the index vacuum so decrement the active worker
     * count.
     */
    if (VacuumActiveNWorkers)
        pg_atomic_sub_fetch_u32(VacuumActiveNWorkers, 1);
}

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 829 of file vacuumparallel.c.

{
    Assert(!IsParallelWorker());
 
    /*
     * Increment the active worker count if we are able to launch any worker.
     */
    if (VacuumActiveNWorkers)
        pg_atomic_add_fetch_u32(VacuumActiveNWorkers, 1);
 
    for (int i = 0; i < pvs->nindexes; i++)
    {
        PVIndStats *indstats = &(pvs->indstats[i]);
 
        /* Skip, indexes that are safe for workers */
        if (indstats->parallel_workers_can_process)
            continue;
 
        /* Do vacuum or cleanup of the index */
        parallel_vacuum_process_one_index(pvs, pvs->indrels[i], indstats);
    }
 
    /*
     * We have completed the index vacuum so decrement the active worker
     * count.
     */
    if (VacuumActiveNWorkers)
        pg_atomic_sub_fetch_u32(VacuumActiveNWorkers, 1);
}

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 476 of file vacuumparallel.c.

{
    VacDeadItemsInfo *dead_items_info = &(pvs->shared->dead_items_info);
 
    /*
     * Free the current tidstore and return allocated DSA segments to the
     * operating system. Then we recreate the tidstore with the same max_bytes
     * limitation we just used.
     */
    TidStoreDestroy(pvs->dead_items);
    pvs->dead_items = TidStoreCreateShared(dead_items_info->max_bytes,
                                           LWTRANCHE_PARALLEL_VACUUM_DSA);
 
    /* Update the DSA pointer for dead_items to the new one */
    pvs->shared->dead_items_dsa_handle = dsa_get_handle(TidStoreGetDSA(pvs->dead_items));
    pvs->shared->dead_items_handle = TidStoreGetHandle(pvs->dead_items);
 
    /* Reset the counter */
    dead_items_info->num_items = 0;
}

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

Referenced by dead_items_reset().