datachecksum_state.h File Reference

#include "storage/procsignal.h"

Include dependency graph for datachecksum_state.h:

This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Functions
bool	AbsorbDataChecksumsBarrier (ProcSignalBarrierType barrier)
void	EmitAndWaitDataChecksumsBarrier (uint32 state)
void	DataChecksumsWorkerLauncherMain (Datum arg)
void	DataChecksumsWorkerMain (Datum arg)

Function Documentation

AbsorbDataChecksumsBarrier()

bool AbsorbDataChecksumsBarrier

(

ProcSignalBarrierType

barrier

)

Definition at line 467 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().

DataChecksumsWorkerLauncherMain()

void DataChecksumsWorkerLauncherMain

(

Datum

arg

)

Definition at line 1093 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.
             */
            CHECK_FOR_LAUNCHER_ABORT_REQUEST();
            if (abort_requested)
                goto done;
            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 abort_requested, Assert, B_DATACHECKSUMSWORKER_LAUNCHER, BackgroundWorkerInitializeConnectionByOid(), BackgroundWorkerUnblockSignals(), CHECK_FOR_LAUNCHER_ABORT_REQUEST, 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 1544 of file datachecksum_state.c.

{
    Oid         dboid;
    List       *RelationList = NIL;
    List       *InitialTempTableList = NIL;
    BufferAccessStrategy strategy;
    bool        aborted = false;
    int64       rels_done;
    bool        process_shared;
#ifdef USE_INJECTION_POINTS
    bool        retried = false;
#endif
 
    worker_invocation = DatumGetUInt64(arg);
 
    operation = ENABLE_DATACHECKSUMS;
 
    pqsignal(SIGTERM, die);
    pqsignal(SIGUSR1, procsignal_sigusr1_handler);
 
    BackgroundWorkerUnblockSignals();
 
    MyBackendType = B_DATACHECKSUMSWORKER_WORKER;
    init_ps_display(NULL);
 
    LWLockAcquire(DataChecksumsWorkerLock, LW_SHARED);
    if (DataChecksumState->worker_invocation != worker_invocation)
    {
        LWLockRelease(DataChecksumsWorkerLock);
        return;
    }
    dboid = DataChecksumState->database_oid;
    LWLockRelease(DataChecksumsWorkerLock);
 
    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 before we exit.  For the list of
     * relations to enable checksums in, check if shared catalogs have been
     * processed already.
     */
    InitialTempTableList = BuildRelationList(true, false);
    LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);
    if (DataChecksumState->worker_invocation != worker_invocation)
    {
        LWLockRelease(DataChecksumsWorkerLock);
        return;
    }
    process_shared = DataChecksumState->process_shared_catalogs;
 
    /*
     * 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;
    LWLockRelease(DataChecksumsWorkerLock);
    VacuumUpdateCosts();
    VacuumCostBalance = 0;
 
    /*
     * Create and set the vacuum strategy as our buffer strategy.
     */
    strategy = GetAccessStrategy(BAS_VACUUM);
 
    RelationList = BuildRelationList(false, process_shared);
 
    /* 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_WORKER_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->worker_invocation != worker_invocation)
        {
            LWLockRelease(DataChecksumsWorkerLock);
            break;
        }
        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);
        if (DataChecksumState->worker_invocation == worker_invocation)
            DataChecksumState->worker_result = 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_WORKER_ABORT_REQUEST();
 
        if (aborted || abort_requested)
        {
            LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);
            if (DataChecksumState->worker_invocation == worker_invocation)
                DataChecksumState->worker_result = 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);
    if (DataChecksumState->worker_invocation == worker_invocation)
        DataChecksumState->worker_result = DATACHECKSUMSWORKER_SUCCESSFUL;
    LWLockRelease(DataChecksumsWorkerLock);
}

References abort_requested, arg, B_DATACHECKSUMSWORKER_WORKER, BackgroundWorkerInitializeConnectionByOid(), BackgroundWorkerUnblockSignals(), BAS_VACUUM, BGWORKER_BYPASS_ALLOWCONN, BuildRelationList(), CHECK_FOR_INTERRUPTS, CHECK_FOR_WORKER_ABORT_REQUEST, DataChecksumsStateStruct::cost_delay, DataChecksumsStateStruct::cost_limit, DataChecksumsStateStruct::database_oid, DataChecksumState, DATACHECKSUMSWORKER_ABORTED, DATACHECKSUMSWORKER_SUCCESSFUL, DatumGetUInt64(), 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, LW_SHARED, 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, VacuumCostBalance, VacuumCostDelay, VacuumCostLimit, VacuumUpdateCosts(), WaitLatch(), WL_EXIT_ON_PM_DEATH, WL_LATCH_SET, WL_TIMEOUT, DataChecksumsStateStruct::worker_invocation, worker_invocation, and DataChecksumsStateStruct::worker_result.

EmitAndWaitDataChecksumsBarrier()

void EmitAndWaitDataChecksumsBarrier

(

uint32

state

)

Definition at line 426 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().