datachecksum_state.c File Reference

#include "postgres.h"
#include "access/genam.h"
#include "access/heapam.h"
#include "access/htup_details.h"
#include "access/xact.h"
#include "access/xlog.h"
#include "access/xloginsert.h"
#include "catalog/indexing.h"
#include "catalog/pg_class.h"
#include "catalog/pg_database.h"
#include "commands/progress.h"
#include "commands/vacuum.h"
#include "common/relpath.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "postmaster/bgworker.h"
#include "postmaster/bgwriter.h"
#include "postmaster/datachecksum_state.h"
#include "storage/bufmgr.h"
#include "storage/checksum.h"
#include "storage/ipc.h"
#include "storage/latch.h"
#include "storage/lmgr.h"
#include "storage/lwlock.h"
#include "storage/procarray.h"
#include "storage/smgr.h"
#include "storage/subsystems.h"
#include "tcop/tcopprot.h"
#include "utils/builtins.h"
#include "utils/fmgroids.h"
#include "utils/injection_point.h"
#include "utils/lsyscache.h"
#include "utils/ps_status.h"
#include "utils/syscache.h"
#include "utils/wait_event.h"

Include dependency graph for datachecksum_state.c:

Go to the source code of this file.

Data Structures
struct	ChecksumBarrierCondition
struct	DataChecksumsStateStruct
struct	DataChecksumsWorkerDatabase

Macros
#define	CHECK_FOR_ABORT_REQUEST()

Typedefs
typedef struct ChecksumBarrierCondition	ChecksumBarrierCondition
typedef struct DataChecksumsStateStruct	DataChecksumsStateStruct
typedef struct DataChecksumsWorkerDatabase	DataChecksumsWorkerDatabase

Functions
static void	DataChecksumsShmemRequest (void *arg)
static bool	DatabaseExists (Oid dboid)
static List *	BuildDatabaseList (void)
static List *	BuildRelationList (bool temp_relations, bool include_shared)
static void	FreeDatabaseList (List *dblist)
static DataChecksumsWorkerResult	ProcessDatabase (DataChecksumsWorkerDatabase *db)
static bool	ProcessAllDatabases (void)
static bool	ProcessSingleRelationFork (Relation reln, ForkNumber forkNum, BufferAccessStrategy strategy)
static void	launcher_cancel_handler (SIGNAL_ARGS)
static void	WaitForAllTransactionsToFinish (void)
void	EmitAndWaitDataChecksumsBarrier (uint32 state)
bool	AbsorbDataChecksumsBarrier (ProcSignalBarrierType barrier)
Datum	disable_data_checksums (PG_FUNCTION_ARGS)
Datum	enable_data_checksums (PG_FUNCTION_ARGS)
void	StartDataChecksumsWorkerLauncher (DataChecksumsWorkerOperation op, int cost_delay, int cost_limit)
static bool	ProcessSingleRelationByOid (Oid relationId, BufferAccessStrategy strategy)
static void	launcher_exit (int code, Datum arg)
void	DataChecksumsWorkerLauncherMain (Datum arg)
void	DataChecksumsWorkerMain (Datum arg)

Variables
static const ChecksumBarrierCondition	checksum_barriers [9]
static DataChecksumsStateStruct *	DataChecksumState
static volatile sig_atomic_t	abort_requested = false
static volatile sig_atomic_t	launcher_running = false
static DataChecksumsWorkerOperation	operation
const ShmemCallbacks	DataChecksumsShmemCallbacks

Macro Definition Documentation

CHECK_FOR_ABORT_REQUEST

#define CHECK_FOR_ABORT_REQUEST

(

)

Value:

    do {                                                            \
		LWLockAcquire(DataChecksumsWorkerLock, LW_SHARED);            \
		if (DataChecksumState->launch_operation != operation)     \
            abort_requested = true;                                 \
		LWLockRelease(DataChecksumsWorkerLock);                       \
    } while (0)

Definition at line 375 of file datachecksum_state.c.

       {                                                            \
		LWLockAcquire(DataChecksumsWorkerLock, LW_SHARED);            \
		if (DataChecksumState->launch_operation != operation)     \
            abort_requested = true;                                 \
		LWLockRelease(DataChecksumsWorkerLock);                       \
    } while (0)

Typedef Documentation

ChecksumBarrierCondition

typedef struct ChecksumBarrierCondition ChecksumBarrierCondition

DataChecksumsStateStruct

typedef struct DataChecksumsStateStruct DataChecksumsStateStruct

DataChecksumsWorkerDatabase

typedef struct DataChecksumsWorkerDatabase DataChecksumsWorkerDatabase

Function Documentation

AbsorbDataChecksumsBarrier()

bool AbsorbDataChecksumsBarrier

(

ProcSignalBarrierType

barrier

)

Definition at line 430 of file datachecksum_state.c.

{
    uint32      target_state;
    int         current = data_checksums;
    bool        found = false;
 
    /*
     * Translate the barrier condition to the target state, doing it here
     * instead of in the procsignal code saves the latter from knowing about
     * checksum states.
     */
    switch (barrier)
    {
        case PROCSIGNAL_BARRIER_CHECKSUM_INPROGRESS_ON:
            target_state = PG_DATA_CHECKSUM_INPROGRESS_ON;
            break;
        case PROCSIGNAL_BARRIER_CHECKSUM_ON:
            target_state = PG_DATA_CHECKSUM_VERSION;
            break;
        case PROCSIGNAL_BARRIER_CHECKSUM_INPROGRESS_OFF:
            target_state = PG_DATA_CHECKSUM_INPROGRESS_OFF;
            break;
        case PROCSIGNAL_BARRIER_CHECKSUM_OFF:
            target_state = PG_DATA_CHECKSUM_OFF;
            break;
        default:
            elog(ERROR, "incorrect barrier \"%i\" received", barrier);
    }
 
    /*
     * If the target state matches the current state then the barrier has been
     * repeated.
     */
    if (current == target_state)
        return true;
 
    /*
     * If the cluster is in recovery we skip the validation of current state
     * since the replay is trusted.
     */
    if (RecoveryInProgress())
    {
        SetLocalDataChecksumState(target_state);
        return true;
    }
 
    /*
     * Find the barrier condition definition for the target state. Not finding
     * a condition would be a grave programmer error as the states are a
     * discrete set.
     */
    for (int i = 0; i < lengthof(checksum_barriers) && !found; i++)
    {
        if (checksum_barriers[i].from == current && checksum_barriers[i].to == target_state)
            found = true;
    }
 
    /*
     * If the relevant state criteria aren't satisfied, throw an error which
     * will be caught by the procsignal machinery for a later retry.
     */
    if (!found)
        ereport(ERROR,
                errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                errmsg("incorrect data checksum state %i for target state %i",
                       current, target_state));
 
    SetLocalDataChecksumState(target_state);
    return true;
}

References barrier, checksum_barriers, data_checksums, elog, ereport, errcode(), errmsg, ERROR, fb(), i, lengthof, PG_DATA_CHECKSUM_INPROGRESS_OFF, PG_DATA_CHECKSUM_INPROGRESS_ON, PG_DATA_CHECKSUM_OFF, PG_DATA_CHECKSUM_VERSION, PROCSIGNAL_BARRIER_CHECKSUM_INPROGRESS_OFF, PROCSIGNAL_BARRIER_CHECKSUM_INPROGRESS_ON, PROCSIGNAL_BARRIER_CHECKSUM_OFF, PROCSIGNAL_BARRIER_CHECKSUM_ON, RecoveryInProgress(), and SetLocalDataChecksumState().

Referenced by ProcessProcSignalBarrier().

BuildDatabaseList()

static List * BuildDatabaseList ( void  )

static

Definition at line 1373 of file datachecksum_state.c.

{
    List       *DatabaseList = NIL;
    Relation    rel;
    TableScanDesc scan;
    HeapTuple   tup;
    MemoryContext ctx = CurrentMemoryContext;
    MemoryContext oldctx;
 
    StartTransactionCommand();
 
    rel = table_open(DatabaseRelationId, AccessShareLock);
    scan = table_beginscan_catalog(rel, 0, NULL);
 
    while (HeapTupleIsValid(tup = heap_getnext(scan, ForwardScanDirection)))
    {
        Form_pg_database pgdb = (Form_pg_database) GETSTRUCT(tup);
        DataChecksumsWorkerDatabase *db;
 
        oldctx = MemoryContextSwitchTo(ctx);
 
        db = (DataChecksumsWorkerDatabase *) palloc0(sizeof(DataChecksumsWorkerDatabase));
 
        db->dboid = pgdb->oid;
        db->dbname = pstrdup(NameStr(pgdb->datname));
 
        DatabaseList = lappend(DatabaseList, db);
 
        MemoryContextSwitchTo(oldctx);
    }
 
    table_endscan(scan);
    table_close(rel, AccessShareLock);
 
    CommitTransactionCommand();
 
    return DatabaseList;
}

References AccessShareLock, CommitTransactionCommand(), CurrentMemoryContext, DatabaseList, DataChecksumsWorkerDatabase::dbname, DataChecksumsWorkerDatabase::dboid, fb(), Form_pg_database, ForwardScanDirection, GETSTRUCT(), heap_getnext(), HeapTupleIsValid, lappend(), MemoryContextSwitchTo(), NameStr, NIL, palloc0(), pstrdup(), StartTransactionCommand(), table_beginscan_catalog(), table_close(), table_endscan(), and table_open().

Referenced by ProcessAllDatabases().

BuildRelationList()

static List * BuildRelationList ( bool  temp_relations, bool  include_shared  )

static

Definition at line 1439 of file datachecksum_state.c.

{
    List       *RelationList = NIL;
    Relation    rel;
    TableScanDesc scan;
    HeapTuple   tup;
    MemoryContext ctx = CurrentMemoryContext;
    MemoryContext oldctx;
 
    StartTransactionCommand();
 
    rel = table_open(RelationRelationId, AccessShareLock);
    scan = table_beginscan_catalog(rel, 0, NULL);
 
    while (HeapTupleIsValid(tup = heap_getnext(scan, ForwardScanDirection)))
    {
        Form_pg_class pgc = (Form_pg_class) GETSTRUCT(tup);
 
        /* Only include temporary relations when explicitly asked to */
        if (pgc->relpersistence == RELPERSISTENCE_TEMP)
        {
            if (!temp_relations)
                continue;
        }
        else
        {
            /*
             * If we are only interested in temp relations then continue
             * immediately as the current relation isn't a temp relation.
             */
            if (temp_relations)
                continue;
 
            if (!RELKIND_HAS_STORAGE(pgc->relkind))
                continue;
 
            if (pgc->relisshared && !include_shared)
                continue;
        }
 
        oldctx = MemoryContextSwitchTo(ctx);
        RelationList = lappend_oid(RelationList, pgc->oid);
        MemoryContextSwitchTo(oldctx);
    }
 
    table_endscan(scan);
    table_close(rel, AccessShareLock);
 
    CommitTransactionCommand();
 
    return RelationList;
}

References AccessShareLock, CommitTransactionCommand(), CurrentMemoryContext, fb(), ForwardScanDirection, GETSTRUCT(), heap_getnext(), HeapTupleIsValid, lappend_oid(), MemoryContextSwitchTo(), NIL, StartTransactionCommand(), table_beginscan_catalog(), table_close(), table_endscan(), and table_open().

Referenced by DataChecksumsWorkerMain().

DatabaseExists()

static bool DatabaseExists ( Oid  dboid )

static

Definition at line 1326 of file datachecksum_state.c.

{
    Relation    rel;
    ScanKeyData skey;
    SysScanDesc scan;
    bool        found;
    HeapTuple   tuple;
    Form_pg_database pg_database_tuple;
 
    StartTransactionCommand();
 
    rel = table_open(DatabaseRelationId, AccessShareLock);
    ScanKeyInit(&skey,
                Anum_pg_database_oid,
                BTEqualStrategyNumber, F_OIDEQ,
                ObjectIdGetDatum(dboid));
    scan = systable_beginscan(rel, DatabaseOidIndexId, true, SnapshotSelf,
                              1, &skey);
    tuple = systable_getnext(scan);
    found = HeapTupleIsValid(tuple);
 
    /* If the Oid exists, ensure that it's not partially dropped */
    if (found)
    {
        pg_database_tuple = (Form_pg_database) GETSTRUCT(tuple);
        if (database_is_invalid_form(pg_database_tuple))
            found = false;
    }
 
    systable_endscan(scan);
    table_close(rel, AccessShareLock);
 
    CommitTransactionCommand();
 
    return found;
}

References AccessShareLock, BTEqualStrategyNumber, CommitTransactionCommand(), database_is_invalid_form(), fb(), Form_pg_database, GETSTRUCT(), HeapTupleIsValid, ObjectIdGetDatum(), ScanKeyInit(), SnapshotSelf, StartTransactionCommand(), systable_beginscan(), systable_endscan(), systable_getnext(), table_close(), and table_open().

Referenced by ProcessDatabase().

DataChecksumsShmemRequest()

static void DataChecksumsShmemRequest ( void *  arg )

static

Definition at line 1309 of file datachecksum_state.c.

{
    ShmemRequestStruct(.name = "DataChecksumsWorker Data",
                       .size = sizeof(DataChecksumsStateStruct),
                       .ptr = (void **) &DataChecksumState,
        );
}

References DataChecksumState, name, and ShmemRequestStruct.

DataChecksumsWorkerLauncherMain()

void DataChecksumsWorkerLauncherMain

(

Datum

arg

)

Definition at line 1049 of file datachecksum_state.c.

{
 
    ereport(DEBUG1,
            errmsg("background worker \"datachecksums launcher\" started"));
 
    pqsignal(SIGTERM, die);
    pqsignal(SIGINT, launcher_cancel_handler);
    pqsignal(SIGUSR1, procsignal_sigusr1_handler);
    pqsignal(SIGUSR2, PG_SIG_IGN);
 
    BackgroundWorkerUnblockSignals();
 
    MyBackendType = B_DATACHECKSUMSWORKER_LAUNCHER;
    init_ps_display(NULL);
 
    INJECTION_POINT("datachecksumsworker-launcher-delay", NULL);
 
    LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);
 
    if (DataChecksumState->launcher_running)
    {
        ereport(LOG,
                errmsg("background worker \"datachecksums launcher\" already running, exiting"));
        /* Launcher was already running, let it finish */
        LWLockRelease(DataChecksumsWorkerLock);
        return;
    }
 
    on_shmem_exit(launcher_exit, 0);
    launcher_running = true;
 
    /* Initialize a connection to shared catalogs only */
    BackgroundWorkerInitializeConnectionByOid(InvalidOid, InvalidOid, 0);
 
    operation = DataChecksumState->launch_operation;
    DataChecksumState->launcher_running = true;
    DataChecksumState->operation = operation;
    DataChecksumState->cost_delay = DataChecksumState->launch_cost_delay;
    DataChecksumState->cost_limit = DataChecksumState->launch_cost_limit;
    LWLockRelease(DataChecksumsWorkerLock);
 
    /*
     * The target state can change while we are busy enabling/disabling
     * checksums, if the user calls pg_disable/enable_data_checksums() before
     * we are finished with the previous request. In that case, we will loop
     * back here, to process the new request.
     */
again:
 
    pgstat_progress_start_command(PROGRESS_COMMAND_DATACHECKSUMS,
                                  InvalidOid);
 
    if (operation == ENABLE_DATACHECKSUMS)
    {
        /*
         * If we are asked to enable checksums in a cluster which already has
         * checksums enabled, exit immediately as there is nothing more to do.
         */
        if (DataChecksumsNeedVerify())
            goto done;
 
        ereport(LOG,
                errmsg("enabling data checksums requested, starting data checksum calculation"));
 
        /*
         * Set the state to inprogress-on and wait on the procsignal barrier.
         */
        pgstat_progress_update_param(PROGRESS_DATACHECKSUMS_PHASE,
                                     PROGRESS_DATACHECKSUMS_PHASE_ENABLING);
        SetDataChecksumsOnInProgress();
 
        /*
         * All backends are now in inprogress-on state and are writing data
         * checksums.  Start processing all data at rest.
         */
        if (!ProcessAllDatabases())
        {
            /*
             * If the target state changed during processing then it's not a
             * failure, so restart processing instead.
             */
            LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);
            if (DataChecksumState->launch_operation != operation)
            {
                LWLockRelease(DataChecksumsWorkerLock);
                goto done;
            }
            LWLockRelease(DataChecksumsWorkerLock);
            ereport(ERROR,
                    errcode(ERRCODE_INSUFFICIENT_RESOURCES),
                    errmsg("unable to enable data checksums in cluster"));
        }
 
        /*
         * Data checksums have been set on all pages, set the state to on in
         * order to instruct backends to validate checksums on reading.
         */
        SetDataChecksumsOn();
 
        ereport(LOG,
                errmsg("data checksums are now enabled"));
    }
    else if (operation == DISABLE_DATACHECKSUMS)
    {
        ereport(LOG,
                errmsg("disabling data checksums requested"));
 
        pgstat_progress_update_param(PROGRESS_DATACHECKSUMS_PHASE,
                                     PROGRESS_DATACHECKSUMS_PHASE_DISABLING);
        SetDataChecksumsOff();
        ereport(LOG,
                errmsg("data checksums are now disabled"));
    }
    else
        Assert(false);
 
done:
 
    /*
     * This state will only be displayed for a fleeting moment, but for the
     * sake of correctness it is still added before ending the command.
     */
    pgstat_progress_update_param(PROGRESS_DATACHECKSUMS_PHASE,
                                 PROGRESS_DATACHECKSUMS_PHASE_DONE);
 
    /*
     * All done. But before we exit, check if the target state was changed
     * while we were running. In that case we will have to start all over
     * again.
     */
    LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);
    if (DataChecksumState->launch_operation != operation)
    {
        DataChecksumState->operation = DataChecksumState->launch_operation;
        operation = DataChecksumState->launch_operation;
        DataChecksumState->cost_delay = DataChecksumState->launch_cost_delay;
        DataChecksumState->cost_limit = DataChecksumState->launch_cost_limit;
        LWLockRelease(DataChecksumsWorkerLock);
        goto again;
    }
 
    /* Shut down progress reporting as we are done */
    pgstat_progress_end_command();
 
    launcher_running = false;
    DataChecksumState->launcher_running = false;
    LWLockRelease(DataChecksumsWorkerLock);
}

References Assert, B_DATACHECKSUMSWORKER_LAUNCHER, BackgroundWorkerInitializeConnectionByOid(), BackgroundWorkerUnblockSignals(), DataChecksumsStateStruct::cost_delay, DataChecksumsStateStruct::cost_limit, DataChecksumsNeedVerify(), DataChecksumState, DEBUG1, die, DISABLE_DATACHECKSUMS, ENABLE_DATACHECKSUMS, ereport, errcode(), errmsg, ERROR, fb(), init_ps_display(), INJECTION_POINT, InvalidOid, DataChecksumsStateStruct::launch_cost_delay, DataChecksumsStateStruct::launch_cost_limit, DataChecksumsStateStruct::launch_operation, launcher_cancel_handler(), launcher_exit(), DataChecksumsStateStruct::launcher_running, launcher_running, LOG, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MyBackendType, on_shmem_exit(), DataChecksumsStateStruct::operation, operation, PG_SIG_IGN, pgstat_progress_end_command(), pgstat_progress_start_command(), pgstat_progress_update_param(), pqsignal, ProcessAllDatabases(), procsignal_sigusr1_handler(), PROGRESS_COMMAND_DATACHECKSUMS, PROGRESS_DATACHECKSUMS_PHASE, PROGRESS_DATACHECKSUMS_PHASE_DISABLING, PROGRESS_DATACHECKSUMS_PHASE_DONE, PROGRESS_DATACHECKSUMS_PHASE_ENABLING, SetDataChecksumsOff(), SetDataChecksumsOn(), SetDataChecksumsOnInProgress(), SIGUSR1, and SIGUSR2.

DataChecksumsWorkerMain()

void DataChecksumsWorkerMain

(

Datum

arg

)

Definition at line 1504 of file datachecksum_state.c.

{
    Oid         dboid = DatumGetObjectId(arg);
    List       *RelationList = NIL;
    List       *InitialTempTableList = NIL;
    BufferAccessStrategy strategy;
    bool        aborted = false;
    int64       rels_done;
#ifdef USE_INJECTION_POINTS
    bool        retried = false;
#endif
 
    operation = ENABLE_DATACHECKSUMS;
 
    pqsignal(SIGTERM, die);
    pqsignal(SIGUSR1, procsignal_sigusr1_handler);
 
    BackgroundWorkerUnblockSignals();
 
    MyBackendType = B_DATACHECKSUMSWORKER_WORKER;
    init_ps_display(NULL);
 
    BackgroundWorkerInitializeConnectionByOid(dboid, InvalidOid,
                                              BGWORKER_BYPASS_ALLOWCONN);
 
    /* worker will have a separate entry in pg_stat_progress_data_checksums */
    pgstat_progress_start_command(PROGRESS_COMMAND_DATACHECKSUMS,
                                  InvalidOid);
 
    /*
     * Get a list of all temp tables present as we start in this database. We
     * need to wait until they are all gone until we are done, since we cannot
     * access these relations and modify them.
     */
    InitialTempTableList = BuildRelationList(true, false);
 
    /*
     * Enable vacuum cost delay, if any.  While this process isn't doing any
     * vacuuming, we are re-using the infrastructure that vacuum cost delay
     * provides rather than inventing something bespoke. This is an internal
     * implementation detail and care should be taken to avoid it bleeding
     * through to the user to avoid confusion.
     *
     * VacuumUpdateCosts() propagates the values to the variables actually
     * read by vacuum_delay_point().
     */
    VacuumCostDelay = DataChecksumState->cost_delay;
    VacuumCostLimit = DataChecksumState->cost_limit;
    VacuumUpdateCosts();
    VacuumCostBalance = 0;
 
    /*
     * Create and set the vacuum strategy as our buffer strategy.
     */
    strategy = GetAccessStrategy(BAS_VACUUM);
 
    RelationList = BuildRelationList(false,
                                     DataChecksumState->process_shared_catalogs);
 
    /* Update the total number of relations to be processed in this DB. */
    {
        const int   index[] = {
            PROGRESS_DATACHECKSUMS_RELS_TOTAL,
            PROGRESS_DATACHECKSUMS_RELS_DONE
        };
 
        int64       vals[2];
 
        vals[0] = list_length(RelationList);
        vals[1] = 0;
 
        pgstat_progress_update_multi_param(2, index, vals);
    }
 
    /* Process the relations */
    rels_done = 0;
    foreach_oid(reloid, RelationList)
    {
        bool        costs_updated = false;
 
        if (!ProcessSingleRelationByOid(reloid, strategy))
        {
            aborted = true;
            break;
        }
 
        pgstat_progress_update_param(PROGRESS_DATACHECKSUMS_RELS_DONE,
                                     ++rels_done);
        CHECK_FOR_INTERRUPTS();
        CHECK_FOR_ABORT_REQUEST();
 
        if (abort_requested)
            break;
 
        /*
         * Check if the cost settings changed during runtime and if so, update
         * to reflect the new values and signal that the access strategy needs
         * to be refreshed.
         */
        LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);
        if ((DataChecksumState->launch_cost_delay != DataChecksumState->cost_delay)
            || (DataChecksumState->launch_cost_limit != DataChecksumState->cost_limit))
        {
            costs_updated = true;
            VacuumCostDelay = DataChecksumState->launch_cost_delay;
            VacuumCostLimit = DataChecksumState->launch_cost_limit;
            VacuumUpdateCosts();
 
            DataChecksumState->cost_delay = DataChecksumState->launch_cost_delay;
            DataChecksumState->cost_limit = DataChecksumState->launch_cost_limit;
        }
        else
            costs_updated = false;
        LWLockRelease(DataChecksumsWorkerLock);
 
        if (costs_updated)
        {
            FreeAccessStrategy(strategy);
            strategy = GetAccessStrategy(BAS_VACUUM);
        }
    }
 
    list_free(RelationList);
    FreeAccessStrategy(strategy);
 
    if (aborted || abort_requested)
    {
        LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);
        DataChecksumState->success = DATACHECKSUMSWORKER_ABORTED;
        LWLockRelease(DataChecksumsWorkerLock);
        ereport(DEBUG1,
                errmsg("data checksum processing aborted in database OID %u",
                       dboid));
        return;
    }
 
    /* The worker is about to wait for temporary tables to go away. */
    pgstat_progress_update_param(PROGRESS_DATACHECKSUMS_PHASE,
                                 PROGRESS_DATACHECKSUMS_PHASE_WAITING_TEMPREL);
 
    /*
     * Wait for all temp tables that existed when we started to go away. This
     * is necessary since we cannot "reach" them to enable checksums. Any temp
     * tables created after we started will already have checksums in them
     * (due to the "inprogress-on" state), so no need to wait for those.
     */
    for (;;)
    {
        List       *CurrentTempTables;
        int         numleft;
        char        activity[64];
 
        CurrentTempTables = BuildRelationList(true, false);
        numleft = 0;
        foreach_oid(tmptbloid, InitialTempTableList)
        {
            if (list_member_oid(CurrentTempTables, tmptbloid))
                numleft++;
        }
        list_free(CurrentTempTables);
 
#ifdef USE_INJECTION_POINTS
        if (IS_INJECTION_POINT_ATTACHED("datachecksumsworker-fake-temptable-wait"))
        {
            /* Make sure to just cause one retry */
            if (!retried && numleft == 0)
            {
                numleft = 1;
                retried = true;
 
                INJECTION_POINT_CACHED("datachecksumsworker-fake-temptable-wait", NULL);
            }
        }
#endif
 
        if (numleft == 0)
            break;
 
        /*
         * At least one temp table is left to wait for, indicate in pgstat
         * activity and progress reporting.
         */
        snprintf(activity,
                 sizeof(activity),
                 "Waiting for %d temp tables to be removed", numleft);
        pgstat_report_activity(STATE_RUNNING, activity);
 
        /* Retry every 3 seconds */
        ResetLatch(MyLatch);
        (void) WaitLatch(MyLatch,
                         WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
                         3000,
                         WAIT_EVENT_CHECKSUM_ENABLE_TEMPTABLE_WAIT);
 
        CHECK_FOR_INTERRUPTS();
        CHECK_FOR_ABORT_REQUEST();
 
        if (aborted || abort_requested)
        {
            LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);
            DataChecksumState->success = DATACHECKSUMSWORKER_ABORTED;
            LWLockRelease(DataChecksumsWorkerLock);
            ereport(LOG,
                    errmsg("data checksum processing aborted in database OID %u",
                           dboid));
            return;
        }
    }
 
    list_free(InitialTempTableList);
 
    /* worker done */
    pgstat_progress_end_command();
 
    LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);
    DataChecksumState->success = DATACHECKSUMSWORKER_SUCCESSFUL;
    LWLockRelease(DataChecksumsWorkerLock);
}

References abort_requested, arg, B_DATACHECKSUMSWORKER_WORKER, BackgroundWorkerInitializeConnectionByOid(), BackgroundWorkerUnblockSignals(), BAS_VACUUM, BGWORKER_BYPASS_ALLOWCONN, BuildRelationList(), CHECK_FOR_ABORT_REQUEST, CHECK_FOR_INTERRUPTS, DataChecksumsStateStruct::cost_delay, DataChecksumsStateStruct::cost_limit, DataChecksumState, DATACHECKSUMSWORKER_ABORTED, DATACHECKSUMSWORKER_SUCCESSFUL, DatumGetObjectId(), DEBUG1, die, ENABLE_DATACHECKSUMS, ereport, errmsg, fb(), foreach_oid, FreeAccessStrategy(), GetAccessStrategy(), init_ps_display(), INJECTION_POINT_CACHED, InvalidOid, IS_INJECTION_POINT_ATTACHED, DataChecksumsStateStruct::launch_cost_delay, DataChecksumsStateStruct::launch_cost_limit, list_free(), list_length(), list_member_oid(), LOG, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MyBackendType, MyLatch, NIL, operation, pgstat_progress_end_command(), pgstat_progress_start_command(), pgstat_progress_update_multi_param(), pgstat_progress_update_param(), pgstat_report_activity(), pqsignal, DataChecksumsStateStruct::process_shared_catalogs, ProcessSingleRelationByOid(), procsignal_sigusr1_handler(), PROGRESS_COMMAND_DATACHECKSUMS, PROGRESS_DATACHECKSUMS_PHASE, PROGRESS_DATACHECKSUMS_PHASE_WAITING_TEMPREL, PROGRESS_DATACHECKSUMS_RELS_DONE, PROGRESS_DATACHECKSUMS_RELS_TOTAL, ResetLatch(), SIGUSR1, snprintf, STATE_RUNNING, DataChecksumsStateStruct::success, VacuumCostBalance, VacuumCostDelay, VacuumCostLimit, VacuumUpdateCosts(), WaitLatch(), WL_EXIT_ON_PM_DEATH, WL_LATCH_SET, and WL_TIMEOUT.

disable_data_checksums()

Datum disable_data_checksums

(

PG_FUNCTION_ARGS

)

Definition at line 507 of file datachecksum_state.c.

{
    PreventCommandDuringRecovery("pg_disable_data_checksums()");
 
    if (!superuser())
        ereport(ERROR,
                errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
                errmsg("must be superuser to change data checksum state"));
 
    StartDataChecksumsWorkerLauncher(DISABLE_DATACHECKSUMS, 0, 0);
    PG_RETURN_VOID();
}

References DISABLE_DATACHECKSUMS, ereport, errcode(), errmsg, ERROR, fb(), PG_RETURN_VOID, PreventCommandDuringRecovery(), StartDataChecksumsWorkerLauncher(), and superuser().

EmitAndWaitDataChecksumsBarrier()

void EmitAndWaitDataChecksumsBarrier

(

uint32

state

)

Definition at line 389 of file datachecksum_state.c.

{
    uint64      barrier;
 
    switch (state)
    {
        case PG_DATA_CHECKSUM_INPROGRESS_ON:
            barrier = EmitProcSignalBarrier(PROCSIGNAL_BARRIER_CHECKSUM_INPROGRESS_ON);
            WaitForProcSignalBarrier(barrier);
            break;
 
        case PG_DATA_CHECKSUM_INPROGRESS_OFF:
            barrier = EmitProcSignalBarrier(PROCSIGNAL_BARRIER_CHECKSUM_INPROGRESS_OFF);
            WaitForProcSignalBarrier(barrier);
            break;
 
        case PG_DATA_CHECKSUM_VERSION:
            barrier = EmitProcSignalBarrier(PROCSIGNAL_BARRIER_CHECKSUM_ON);
            WaitForProcSignalBarrier(barrier);
            break;
 
        case PG_DATA_CHECKSUM_OFF:
            barrier = EmitProcSignalBarrier(PROCSIGNAL_BARRIER_CHECKSUM_OFF);
            WaitForProcSignalBarrier(barrier);
            break;
 
        default:
            Assert(false);
    }
}

References Assert, barrier, EmitProcSignalBarrier(), PG_DATA_CHECKSUM_INPROGRESS_OFF, PG_DATA_CHECKSUM_INPROGRESS_ON, PG_DATA_CHECKSUM_OFF, PG_DATA_CHECKSUM_VERSION, PROCSIGNAL_BARRIER_CHECKSUM_INPROGRESS_OFF, PROCSIGNAL_BARRIER_CHECKSUM_INPROGRESS_ON, PROCSIGNAL_BARRIER_CHECKSUM_OFF, PROCSIGNAL_BARRIER_CHECKSUM_ON, and WaitForProcSignalBarrier().

Referenced by StartupXLOG(), xlog2_redo(), and xlog_redo().

enable_data_checksums()

Datum enable_data_checksums

(

PG_FUNCTION_ARGS

)

Definition at line 526 of file datachecksum_state.c.

{
    int         cost_delay = PG_GETARG_INT32(0);
    int         cost_limit = PG_GETARG_INT32(1);
 
    PreventCommandDuringRecovery("pg_enable_data_checksums()");
 
    if (!superuser())
        ereport(ERROR,
                errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
                errmsg("must be superuser to change data checksum state"));
 
    if (cost_delay < 0)
        ereport(ERROR,
                errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                errmsg("cost delay cannot be a negative value"));
 
    if (cost_limit <= 0)
        ereport(ERROR,
                errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                errmsg("cost limit must be greater than zero"));
 
    StartDataChecksumsWorkerLauncher(ENABLE_DATACHECKSUMS, cost_delay, cost_limit);
 
    PG_RETURN_VOID();
}

References ENABLE_DATACHECKSUMS, ereport, errcode(), errmsg, ERROR, fb(), PG_GETARG_INT32, PG_RETURN_VOID, PreventCommandDuringRecovery(), StartDataChecksumsWorkerLauncher(), and superuser().

FreeDatabaseList()

static void FreeDatabaseList ( List *  dblist )

static

Definition at line 1413 of file datachecksum_state.c.

{
    if (!dblist)
        return;
 
    foreach_ptr(DataChecksumsWorkerDatabase, db, dblist)
    {
        if (db->dbname != NULL)
            pfree(db->dbname);
    }
 
    list_free_deep(dblist);
}

References dblist, fb(), foreach_ptr, list_free_deep(), and pfree().

Referenced by ProcessAllDatabases().

launcher_cancel_handler()

static void launcher_cancel_handler ( SIGNAL_ARGS  )

static

Definition at line 962 of file datachecksum_state.c.

{
    int         save_errno = errno;
 
    abort_requested = true;
 
    /*
     * There is no sleeping in the main loop, the flag will be checked
     * periodically in ProcessSingleRelationFork. The worker does however
     * sleep when waiting for concurrent transactions to end so we still need
     * to set the latch.
     */
    SetLatch(MyLatch);
 
    errno = save_errno;
}

References abort_requested, fb(), MyLatch, and SetLatch().

Referenced by DataChecksumsWorkerLauncherMain().

launcher_exit()

static void launcher_exit ( int  code, Datum  arg  )

static

Definition at line 925 of file datachecksum_state.c.

{
    abort_requested = false;
 
    if (launcher_running)
    {
        LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);
        if (DataChecksumState->worker_pid != InvalidPid)
        {
            ereport(LOG,
                    errmsg("data checksums launcher exiting while worker is still running, signalling worker"));
            kill(DataChecksumState->worker_pid, SIGTERM);
        }
        LWLockRelease(DataChecksumsWorkerLock);
    }
 
    /*
     * If the launcher is exiting before data checksums are enabled then set
     * the state to off since processing cannot be resumed.
     */
    if (DataChecksumsInProgressOn())
        SetDataChecksumsOff();
 
    LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);
    launcher_running = false;
    DataChecksumState->launcher_running = false;
    LWLockRelease(DataChecksumsWorkerLock);
}

References abort_requested, DataChecksumsInProgressOn(), DataChecksumState, ereport, errmsg, fb(), InvalidPid, kill, DataChecksumsStateStruct::launcher_running, launcher_running, LOG, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), SetDataChecksumsOff(), and DataChecksumsStateStruct::worker_pid.

Referenced by DataChecksumsWorkerLauncherMain().

ProcessAllDatabases()

static bool ProcessAllDatabases ( void  )

static

Definition at line 1208 of file datachecksum_state.c.

{
    List       *DatabaseList;
    int         cumulative_total = 0;
 
    /* Set up so first run processes shared catalogs, not once in every db */
    LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);
    DataChecksumState->process_shared_catalogs = true;
    LWLockRelease(DataChecksumsWorkerLock);
 
    /* Get a list of all databases to process */
    WaitForAllTransactionsToFinish();
    DatabaseList = BuildDatabaseList();
 
    /*
     * Update progress reporting with the total number of databases we need to
     * process.  This number should not be changed during processing, the
     * columns for processed databases is instead increased such that it can
     * be compared against the total.
     */
    {
        const int   index[] = {
            PROGRESS_DATACHECKSUMS_DBS_TOTAL,
            PROGRESS_DATACHECKSUMS_DBS_DONE,
            PROGRESS_DATACHECKSUMS_RELS_TOTAL,
            PROGRESS_DATACHECKSUMS_RELS_DONE,
            PROGRESS_DATACHECKSUMS_BLOCKS_TOTAL,
            PROGRESS_DATACHECKSUMS_BLOCKS_DONE,
        };
 
        int64       vals[6];
 
        vals[0] = list_length(DatabaseList);
        vals[1] = 0;
        /* translated to NULL */
        vals[2] = -1;
        vals[3] = -1;
        vals[4] = -1;
        vals[5] = -1;
 
        pgstat_progress_update_multi_param(6, index, vals);
    }
 
    foreach_ptr(DataChecksumsWorkerDatabase, db, DatabaseList)
    {
        DataChecksumsWorkerResult result;
 
        result = ProcessDatabase(db);
 
#ifdef USE_INJECTION_POINTS
        /* Allow a test process to alter the result of the operation */
        if (IS_INJECTION_POINT_ATTACHED("datachecksumsworker-fail-db-result"))
        {
            result = DATACHECKSUMSWORKER_FAILED;
            INJECTION_POINT_CACHED("datachecksumsworker-fail-db-result",
                                   db->dbname);
        }
#endif
 
        pgstat_progress_update_param(PROGRESS_DATACHECKSUMS_DBS_DONE,
                                     ++cumulative_total);
 
        if (result == DATACHECKSUMSWORKER_FAILED)
        {
            /*
             * Disable checksums on cluster, because we failed one of the
             * databases and this is an all or nothing process.
             */
            SetDataChecksumsOff();
            ereport(ERROR,
                    errcode(ERRCODE_INSUFFICIENT_RESOURCES),
                    errmsg("data checksums failed to get enabled in all databases, aborting"),
                    errhint("The server log might have more information on the cause of the error."));
        }
        else if (result == DATACHECKSUMSWORKER_ABORTED || abort_requested)
        {
            /* Abort flag set, so exit the whole process */
            return false;
        }
 
        /*
         * When one database has completed, it will have done shared catalogs
         * so we don't have to process them again.
         */
        LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);
        DataChecksumState->process_shared_catalogs = false;
        LWLockRelease(DataChecksumsWorkerLock);
    }
 
    FreeDatabaseList(DatabaseList);
 
    pgstat_progress_update_param(PROGRESS_DATACHECKSUMS_PHASE,
                                 PROGRESS_DATACHECKSUMS_PHASE_WAITING_BARRIER);
    return true;
}

References abort_requested, BuildDatabaseList(), DatabaseList, DataChecksumState, DATACHECKSUMSWORKER_ABORTED, DATACHECKSUMSWORKER_FAILED, ereport, errcode(), errhint(), errmsg, ERROR, fb(), foreach_ptr, FreeDatabaseList(), INJECTION_POINT_CACHED, IS_INJECTION_POINT_ATTACHED, list_length(), LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), pgstat_progress_update_multi_param(), pgstat_progress_update_param(), DataChecksumsStateStruct::process_shared_catalogs, ProcessDatabase(), PROGRESS_DATACHECKSUMS_BLOCKS_DONE, PROGRESS_DATACHECKSUMS_BLOCKS_TOTAL, PROGRESS_DATACHECKSUMS_DBS_DONE, PROGRESS_DATACHECKSUMS_DBS_TOTAL, PROGRESS_DATACHECKSUMS_PHASE, PROGRESS_DATACHECKSUMS_PHASE_WAITING_BARRIER, PROGRESS_DATACHECKSUMS_RELS_DONE, PROGRESS_DATACHECKSUMS_RELS_TOTAL, result, SetDataChecksumsOff(), and WaitForAllTransactionsToFinish().

Referenced by DataChecksumsWorkerLauncherMain().

ProcessDatabase()

static DataChecksumsWorkerResult ProcessDatabase ( DataChecksumsWorkerDatabase *  db )

static

Definition at line 793 of file datachecksum_state.c.

{
    BackgroundWorker bgw;
    BackgroundWorkerHandle *bgw_handle;
    BgwHandleStatus status;
    pid_t       pid;
    char        activity[NAMEDATALEN + 64];
 
    LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);
    DataChecksumState->success = DATACHECKSUMSWORKER_FAILED;
    LWLockRelease(DataChecksumsWorkerLock);
 
    memset(&bgw, 0, sizeof(bgw));
    bgw.bgw_flags = BGWORKER_SHMEM_ACCESS | BGWORKER_BACKEND_DATABASE_CONNECTION;
    bgw.bgw_start_time = BgWorkerStart_RecoveryFinished;
    snprintf(bgw.bgw_library_name, BGW_MAXLEN, "postgres");
    snprintf(bgw.bgw_function_name, BGW_MAXLEN, "%s", "DataChecksumsWorkerMain");
    snprintf(bgw.bgw_name, BGW_MAXLEN, "datachecksums worker");
    snprintf(bgw.bgw_type, BGW_MAXLEN, "datachecksums worker");
    bgw.bgw_restart_time = BGW_NEVER_RESTART;
    bgw.bgw_notify_pid = MyProcPid;
    bgw.bgw_main_arg = ObjectIdGetDatum(db->dboid);
 
    /*
     * If there are no worker slots available, there is little we can do.  If
     * we retry in a bit it's still unlikely that the user has managed to
     * reconfigure in the meantime and we'd be run through retries fast.
     */
    if (!RegisterDynamicBackgroundWorker(&bgw, &bgw_handle))
    {
        ereport(WARNING,
                errmsg("could not start background worker for enabling data checksums in database \"%s\"",
                       db->dbname),
                errhint("The \"%s\" setting might be too low.", "max_worker_processes"));
        return DATACHECKSUMSWORKER_FAILED;
    }
 
    status = WaitForBackgroundWorkerStartup(bgw_handle, &pid);
    if (status == BGWH_STOPPED)
    {
        /*
         * If the worker managed to start, and stop, before we got to waiting
         * for it we can see a STOPPED status here without it being a failure.
         */
        LWLockAcquire(DataChecksumsWorkerLock, LW_SHARED);
        if (DataChecksumState->success == DATACHECKSUMSWORKER_SUCCESSFUL)
        {
            LWLockRelease(DataChecksumsWorkerLock);
            pgstat_report_activity(STATE_IDLE, NULL);
            LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);
            DataChecksumState->worker_pid = InvalidPid;
            LWLockRelease(DataChecksumsWorkerLock);
            return DataChecksumState->success;
        }
        LWLockRelease(DataChecksumsWorkerLock);
 
        ereport(WARNING,
                errmsg("could not start background worker for enabling data checksums in database \"%s\"",
                       db->dbname),
                errhint("More details on the error might be found in the server log."));
 
        /*
         * Heuristic to see if the database was dropped, and if it was we can
         * treat it as not an error, else treat as fatal and error out.
         */
        if (DatabaseExists(db->dboid))
            return DATACHECKSUMSWORKER_FAILED;
        else
            return DATACHECKSUMSWORKER_DROPDB;
    }
 
    /*
     * If the postmaster crashed we cannot end up with a processed database so
     * we have no alternative other than exiting. When enabling checksums we
     * won't at this time have changed the data checksums state in pg_control
     * to enabled so when the cluster comes back up processing will have to be
     * restarted.
     */
    if (status == BGWH_POSTMASTER_DIED)
        ereport(FATAL,
                errcode(ERRCODE_ADMIN_SHUTDOWN),
                errmsg("cannot enable data checksums without the postmaster process"),
                errhint("Restart the database and restart data checksum processing by calling pg_enable_data_checksums()."));
 
    Assert(status == BGWH_STARTED);
    ereport(LOG,
            errmsg("initiating data checksum processing in database \"%s\"",
                   db->dbname));
 
    /* Save the pid of the worker so we can signal it later */
    LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);
    DataChecksumState->worker_pid = pid;
    LWLockRelease(DataChecksumsWorkerLock);
 
    snprintf(activity, sizeof(activity) - 1,
             "Waiting for worker in database %s (pid %ld)", db->dbname, (long) pid);
    pgstat_report_activity(STATE_RUNNING, activity);
 
    status = WaitForBackgroundWorkerShutdown(bgw_handle);
    if (status == BGWH_POSTMASTER_DIED)
        ereport(FATAL,
                errcode(ERRCODE_ADMIN_SHUTDOWN),
                errmsg("postmaster exited during data checksum processing in \"%s\"",
                       db->dbname),
                errhint("Restart the database and restart data checksum processing by calling pg_enable_data_checksums()."));
 
    LWLockAcquire(DataChecksumsWorkerLock, LW_SHARED);
    if (DataChecksumState->success == DATACHECKSUMSWORKER_ABORTED)
        ereport(LOG,
                errmsg("data checksums processing was aborted in database \"%s\"",
                       db->dbname));
    LWLockRelease(DataChecksumsWorkerLock);
 
    pgstat_report_activity(STATE_IDLE, NULL);
    LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);
    DataChecksumState->worker_pid = InvalidPid;
    LWLockRelease(DataChecksumsWorkerLock);
 
    return DataChecksumState->success;
}

References Assert, BGW_MAXLEN, BGW_NEVER_RESTART, BGWH_POSTMASTER_DIED, BGWH_STARTED, BGWH_STOPPED, BGWORKER_BACKEND_DATABASE_CONNECTION, BGWORKER_SHMEM_ACCESS, BgWorkerStart_RecoveryFinished, DatabaseExists(), DataChecksumState, DATACHECKSUMSWORKER_ABORTED, DATACHECKSUMSWORKER_DROPDB, DATACHECKSUMSWORKER_FAILED, DATACHECKSUMSWORKER_SUCCESSFUL, DataChecksumsWorkerDatabase::dbname, DataChecksumsWorkerDatabase::dboid, ereport, errcode(), errhint(), errmsg, FATAL, fb(), InvalidPid, LOG, LW_EXCLUSIVE, LW_SHARED, LWLockAcquire(), LWLockRelease(), MyProcPid, NAMEDATALEN, ObjectIdGetDatum(), pgstat_report_activity(), RegisterDynamicBackgroundWorker(), snprintf, STATE_IDLE, STATE_RUNNING, DataChecksumsStateStruct::success, WaitForBackgroundWorkerShutdown(), WaitForBackgroundWorkerStartup(), WARNING, and DataChecksumsStateStruct::worker_pid.

Referenced by ProcessAllDatabases().

ProcessSingleRelationByOid()

static bool ProcessSingleRelationByOid ( Oid  relationId, BufferAccessStrategy  strategy  )

static

Definition at line 742 of file datachecksum_state.c.

{
    Relation    rel;
    bool        aborted = false;
 
    StartTransactionCommand();
 
    rel = try_relation_open(relationId, AccessShareLock);
    if (rel == NULL)
    {
        /*
         * Relation no longer exists. We don't consider this an error since
         * there are no pages in it that need data checksums, and thus return
         * true. The worker operates off a list of relations generated at the
         * start of processing, so relations being dropped in the meantime is
         * to be expected.
         */
        CommitTransactionCommand();
        pgstat_report_activity(STATE_IDLE, NULL);
        return true;
    }
    RelationGetSmgr(rel);
 
    for (ForkNumber fnum = 0; fnum <= MAX_FORKNUM; fnum++)
    {
        if (smgrexists(rel->rd_smgr, fnum))
        {
            if (!ProcessSingleRelationFork(rel, fnum, strategy))
            {
                aborted = true;
                break;
            }
        }
    }
    relation_close(rel, AccessShareLock);
 
    CommitTransactionCommand();
    pgstat_report_activity(STATE_IDLE, NULL);
 
    return !aborted;
}

References AccessShareLock, CommitTransactionCommand(), fb(), MAX_FORKNUM, pgstat_report_activity(), ProcessSingleRelationFork(), RelationData::rd_smgr, relation_close(), RelationGetSmgr(), smgrexists(), StartTransactionCommand(), STATE_IDLE, and try_relation_open().

Referenced by DataChecksumsWorkerMain().

ProcessSingleRelationFork()

static bool ProcessSingleRelationFork ( Relation  reln, ForkNumber  forkNum, BufferAccessStrategy  strategy  )

static

Definition at line 652 of file datachecksum_state.c.

{
    BlockNumber numblocks = RelationGetNumberOfBlocksInFork(reln, forkNum);
    char        activity[NAMEDATALEN * 2 + 128];
    char       *relns;
 
    relns = get_namespace_name(RelationGetNamespace(reln));
 
    /* Report the current relation to pg_stat_activity */
    snprintf(activity, sizeof(activity) - 1, "processing: %s.%s (%s, %u blocks)",
             (relns ? relns : ""), RelationGetRelationName(reln), forkNames[forkNum], numblocks);
    pgstat_report_activity(STATE_RUNNING, activity);
    pgstat_progress_update_param(PROGRESS_DATACHECKSUMS_BLOCKS_TOTAL, numblocks);
    if (relns)
        pfree(relns);
 
    /*
     * We are looping over the blocks which existed at the time of process
     * start, which is safe since new blocks are created with checksums set
     * already due to the state being "inprogress-on".
     */
    for (BlockNumber blknum = 0; blknum < numblocks; blknum++)
    {
        Buffer      buf = ReadBufferExtended(reln, forkNum, blknum, RBM_NORMAL, strategy);
 
        /* Need to get an exclusive lock to mark the buffer as dirty */
        LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
 
        /*
         * Mark the buffer as dirty and force a full page write.  We have to
         * re-write the page to WAL even if the checksum hasn't changed,
         * because if there is a replica it might have a slightly different
         * version of the page with an invalid checksum, caused by unlogged
         * changes (e.g. hint bits) on the primary happening while checksums
         * were off. This can happen if there was a valid checksum on the page
         * at one point in the past, so only when checksums are first on, then
         * off, and then turned on again.  TODO: investigate if this could be
         * avoided if the checksum is calculated to be correct and wal_level
         * is set to "minimal".
         *
         * Unlogged relations don't need WAL since they are reset to their
         * init fork on recovery.  We still dirty the buffer so that the
         * checksum is written to disk at the next checkpoint.
         *
         * The init fork is an exception: it is WAL-logged so the standby can
         * materialize the relation after promotion (see
         * ResetUnloggedRelations()).  Skipping it here would leave the
         * standby with a stale init fork that, once copied to the main fork
         * on promotion, would fail checksum verification on every read.
         */
        START_CRIT_SECTION();
        MarkBufferDirty(buf);
        if (RelationNeedsWAL(reln) || forkNum == INIT_FORKNUM)
            log_newpage_buffer(buf, false);
        END_CRIT_SECTION();
 
        UnlockReleaseBuffer(buf);
 
        /* Check if we are asked to abort, the abortion will bubble up. */
        Assert(operation == ENABLE_DATACHECKSUMS);
        LWLockAcquire(DataChecksumsWorkerLock, LW_SHARED);
        if (DataChecksumState->launch_operation == DISABLE_DATACHECKSUMS)
            abort_requested = true;
        LWLockRelease(DataChecksumsWorkerLock);
 
        if (abort_requested)
            return false;
 
        /* update the block counter */
        pgstat_progress_update_param(PROGRESS_DATACHECKSUMS_BLOCKS_DONE,
                                     (blknum + 1));
 
        /*
         * Processing is re-using the vacuum cost delay for process
         * throttling, hence why we call vacuum APIs here.
         */
        vacuum_delay_point(false);
    }
 
    return true;
}

References abort_requested, Assert, buf, BUFFER_LOCK_EXCLUSIVE, DataChecksumState, DISABLE_DATACHECKSUMS, ENABLE_DATACHECKSUMS, END_CRIT_SECTION, fb(), forkNames, get_namespace_name(), INIT_FORKNUM, DataChecksumsStateStruct::launch_operation, LockBuffer(), log_newpage_buffer(), LW_SHARED, LWLockAcquire(), LWLockRelease(), MarkBufferDirty(), NAMEDATALEN, operation, pfree(), pgstat_progress_update_param(), pgstat_report_activity(), PROGRESS_DATACHECKSUMS_BLOCKS_DONE, PROGRESS_DATACHECKSUMS_BLOCKS_TOTAL, RBM_NORMAL, ReadBufferExtended(), RelationGetNamespace, RelationGetNumberOfBlocksInFork(), RelationGetRelationName, RelationNeedsWAL, snprintf, START_CRIT_SECTION, STATE_RUNNING, UnlockReleaseBuffer(), and vacuum_delay_point().

Referenced by ProcessSingleRelationByOid().

StartDataChecksumsWorkerLauncher()

void StartDataChecksumsWorkerLauncher	(	DataChecksumsWorkerOperation	op,
		int	cost_delay,
		int	cost_limit
	)

Definition at line 566 of file datachecksum_state.c.

{
    BackgroundWorker bgw;
    BackgroundWorkerHandle *bgw_handle;
    bool        running;
 
#ifdef USE_ASSERT_CHECKING
    /* The cost delay settings have no effect when disabling */
    if (op == DISABLE_DATACHECKSUMS)
        Assert(cost_delay == 0 && cost_limit == 0);
#endif
 
    INJECTION_POINT("datachecksumsworker-startup-delay", NULL);
 
    /* Store the desired state in shared memory */
    LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);
 
    DataChecksumState->launch_operation = op;
    DataChecksumState->launch_cost_delay = cost_delay;
    DataChecksumState->launch_cost_limit = cost_limit;
 
    /* Is the launcher already running? If so, what is it doing? */
    running = DataChecksumState->launcher_running;
 
    LWLockRelease(DataChecksumsWorkerLock);
 
    /*
     * Launch a new launcher process, if it's not running already.
     *
     * If the launcher is currently busy enabling the checksums, and we want
     * them disabled (or vice versa), the launcher will notice that at latest
     * when it's about to exit, and will loop back to process the new request.
     * So if the launcher is already running, we don't need to do anything
     * more here to abort it.
     *
     * If you call pg_enable/disable_data_checksums() twice in a row, before
     * the launcher has had a chance to start up, we still end up launching it
     * twice.  That's OK, the second invocation will see that a launcher is
     * already running and exit quickly.
     */
    if (!running)
    {
        if ((op == ENABLE_DATACHECKSUMS && DataChecksumsOn()) ||
            (op == DISABLE_DATACHECKSUMS && DataChecksumsOff()))
        {
            ereport(LOG,
                    errmsg("data checksums already in desired state, exiting"));
            return;
        }
 
        /*
         * Prepare the BackgroundWorker and launch it.
         */
        memset(&bgw, 0, sizeof(bgw));
        bgw.bgw_flags = BGWORKER_SHMEM_ACCESS | BGWORKER_BACKEND_DATABASE_CONNECTION;
        bgw.bgw_start_time = BgWorkerStart_RecoveryFinished;
        snprintf(bgw.bgw_library_name, BGW_MAXLEN, "postgres");
        snprintf(bgw.bgw_function_name, BGW_MAXLEN, "DataChecksumsWorkerLauncherMain");
        snprintf(bgw.bgw_name, BGW_MAXLEN, "datachecksums launcher");
        snprintf(bgw.bgw_type, BGW_MAXLEN, "datachecksums launcher");
        bgw.bgw_restart_time = BGW_NEVER_RESTART;
        bgw.bgw_notify_pid = MyProcPid;
        bgw.bgw_main_arg = (Datum) 0;
 
        if (!RegisterDynamicBackgroundWorker(&bgw, &bgw_handle))
            ereport(ERROR,
                    errcode(ERRCODE_INSUFFICIENT_RESOURCES),
                    errmsg("failed to start background worker to process data checksums"));
    }
    else
    {
        ereport(LOG,
                errmsg("data checksum processing already running"));
    }
}

References Assert, BGW_MAXLEN, BGW_NEVER_RESTART, BGWORKER_BACKEND_DATABASE_CONNECTION, BGWORKER_SHMEM_ACCESS, BgWorkerStart_RecoveryFinished, DataChecksumsOff(), DataChecksumsOn(), DataChecksumState, DISABLE_DATACHECKSUMS, ENABLE_DATACHECKSUMS, ereport, errcode(), errmsg, ERROR, fb(), INJECTION_POINT, DataChecksumsStateStruct::launch_cost_delay, DataChecksumsStateStruct::launch_cost_limit, DataChecksumsStateStruct::launch_operation, DataChecksumsStateStruct::launcher_running, LOG, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MyProcPid, RegisterDynamicBackgroundWorker(), and snprintf.

Referenced by disable_data_checksums(), and enable_data_checksums().

WaitForAllTransactionsToFinish()

static void WaitForAllTransactionsToFinish ( void  )

static

Definition at line 992 of file datachecksum_state.c.

{
    TransactionId waitforxid;
 
    LWLockAcquire(XidGenLock, LW_SHARED);
    waitforxid = XidFromFullTransactionId(TransamVariables->nextXid);
    LWLockRelease(XidGenLock);
 
    while (TransactionIdPrecedes(GetOldestActiveTransactionId(false, true), waitforxid))
    {
        char        activity[64];
        int         rc;
 
        /* Oldest running xid is older than us, so wait */
        snprintf(activity,
                 sizeof(activity),
                 "Waiting for transactions older than %u to end",
                 waitforxid);
        pgstat_report_activity(STATE_RUNNING, activity);
 
        /* Retry every 3 seconds */
        ResetLatch(MyLatch);
        rc = WaitLatch(MyLatch,
                       WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH,
                       3000,
                       WAIT_EVENT_CHECKSUM_ENABLE_STARTCONDITION);
 
        /*
         * If the postmaster died we won't be able to enable checksums
         * cluster-wide so abort and hope to continue when restarted.
         */
        if (rc & WL_POSTMASTER_DEATH)
            ereport(FATAL,
                    errcode(ERRCODE_ADMIN_SHUTDOWN),
                    errmsg("postmaster exited during data checksums processing"),
                    errhint("Data checksums processing must be restarted manually after cluster restart."));
 
        CHECK_FOR_INTERRUPTS();
        CHECK_FOR_ABORT_REQUEST();
 
        if (abort_requested)
            break;
    }
 
    pgstat_report_activity(STATE_IDLE, NULL);
    return;
}

References abort_requested, CHECK_FOR_ABORT_REQUEST, CHECK_FOR_INTERRUPTS, ereport, errcode(), errhint(), errmsg, FATAL, fb(), GetOldestActiveTransactionId(), LW_SHARED, LWLockAcquire(), LWLockRelease(), MyLatch, TransamVariablesData::nextXid, pgstat_report_activity(), ResetLatch(), snprintf, STATE_IDLE, STATE_RUNNING, TransactionIdPrecedes(), TransamVariables, WaitLatch(), WL_LATCH_SET, WL_POSTMASTER_DEATH, WL_TIMEOUT, and XidFromFullTransactionId.

Referenced by ProcessAllDatabases().

Variable Documentation

abort_requested

volatile sig_atomic_t abort_requested = false

static

Definition at line 348 of file datachecksum_state.c.

Referenced by DataChecksumsWorkerMain(), launcher_cancel_handler(), launcher_exit(), ProcessAllDatabases(), ProcessSingleRelationFork(), and WaitForAllTransactionsToFinish().

checksum_barriers

const ChecksumBarrierCondition checksum_barriers[9]

static

Initial value:

=
{
    
    {PG_DATA_CHECKSUM_VERSION, PG_DATA_CHECKSUM_INPROGRESS_OFF},
    {PG_DATA_CHECKSUM_INPROGRESS_OFF, PG_DATA_CHECKSUM_OFF},
 
    
    {PG_DATA_CHECKSUM_INPROGRESS_ON, PG_DATA_CHECKSUM_OFF},
 
    
    {PG_DATA_CHECKSUM_OFF, PG_DATA_CHECKSUM_INPROGRESS_ON},
    {PG_DATA_CHECKSUM_INPROGRESS_ON, PG_DATA_CHECKSUM_VERSION},
 
    
    {PG_DATA_CHECKSUM_INPROGRESS_OFF, PG_DATA_CHECKSUM_VERSION},
 
    
    {PG_DATA_CHECKSUM_INPROGRESS_ON, PG_DATA_CHECKSUM_INPROGRESS_OFF},
 
    
    {PG_DATA_CHECKSUM_INPROGRESS_OFF, PG_DATA_CHECKSUM_INPROGRESS_ON},
    {PG_DATA_CHECKSUM_OFF, PG_DATA_CHECKSUM_INPROGRESS_OFF},
}

Definition at line 238 of file datachecksum_state.c.

{
    /*
     * Disabling checksums: If checksums are currently enabled, disabling must
     * go through the 'inprogress-off' state.
     */
    {PG_DATA_CHECKSUM_VERSION, PG_DATA_CHECKSUM_INPROGRESS_OFF},
    {PG_DATA_CHECKSUM_INPROGRESS_OFF, PG_DATA_CHECKSUM_OFF},
 
    /*
     * If checksums are in the process of being enabled, but are not yet being
     * verified, we can abort by going back to 'off' state.
     */
    {PG_DATA_CHECKSUM_INPROGRESS_ON, PG_DATA_CHECKSUM_OFF},
 
    /*
     * Enabling checksums must normally go through the 'inprogress-on' state.
     */
    {PG_DATA_CHECKSUM_OFF, PG_DATA_CHECKSUM_INPROGRESS_ON},
    {PG_DATA_CHECKSUM_INPROGRESS_ON, PG_DATA_CHECKSUM_VERSION},
 
    /*
     * If checksums are being disabled but all backends are still computing
     * checksums, we can go straight back to 'on'
     */
    {PG_DATA_CHECKSUM_INPROGRESS_OFF, PG_DATA_CHECKSUM_VERSION},
 
    /*
     * If checksums are being enabled when launcher_exit is executed, state is
     * set to off since we cannot reach on at that point.
     */
    {PG_DATA_CHECKSUM_INPROGRESS_ON, PG_DATA_CHECKSUM_INPROGRESS_OFF},
 
    /*
     * Transitions that can happen when a new request is made while another is
     * currently being processed.
     */
    {PG_DATA_CHECKSUM_INPROGRESS_OFF, PG_DATA_CHECKSUM_INPROGRESS_ON},
    {PG_DATA_CHECKSUM_OFF, PG_DATA_CHECKSUM_INPROGRESS_OFF},
};

Referenced by AbsorbDataChecksumsBarrier().

DataChecksumsShmemCallbacks

const ShmemCallbacks DataChecksumsShmemCallbacks

Initial value:

= {
    .request_fn = DataChecksumsShmemRequest,
}

Definition at line 371 of file datachecksum_state.c.

                                                   {
    .request_fn = DataChecksumsShmemRequest,
};

DataChecksumState

DataChecksumsStateStruct* DataChecksumState

static

Definition at line 339 of file datachecksum_state.c.

Referenced by DataChecksumsShmemRequest(), DataChecksumsWorkerLauncherMain(), DataChecksumsWorkerMain(), launcher_exit(), ProcessAllDatabases(), ProcessDatabase(), ProcessSingleRelationFork(), and StartDataChecksumsWorkerLauncher().

launcher_running

volatile sig_atomic_t launcher_running = false

static

Definition at line 354 of file datachecksum_state.c.

Referenced by DataChecksumsWorkerLauncherMain(), and launcher_exit().

operation

DataChecksumsWorkerOperation operation

static

Definition at line 357 of file datachecksum_state.c.

Referenced by _SPI_pquery(), apply_handle_tuple_routing(), AsyncReadBuffers(), ATExecSetRelOptions(), check_foreign_key(), CheckReadBuffersOperation(), CheckValidResultRel(), create_foreign_modify(), create_modifytable_path(), DataChecksumsWorkerLauncherMain(), DataChecksumsWorkerMain(), ExecInitModifyTable(), ExecModifyTable(), execute_foreign_modify(), ExecutePlan(), ExplainNode(), InitPlan(), make_modifytable(), postgresPlanDirectModify(), postgresPlanForeignModify(), ProcessReadBuffersResult(), ProcessSingleRelationFork(), read_buffers(), ReadBuffer_common(), set_deparse_plan(), show_modifytable_info(), show_plan_tlist(), standard_ExecutorRun(), start_lo_xact(), StartReadBuffer(), StartReadBuffers(), StartReadBuffersImpl(), triggered_change_notification(), and WaitReadBuffers().