mirror of https://github.com/postgres/postgres
This reverts the backend sides of commit 1fde38beaa.
I have, at least for now, left the pg_verify_checksums tool in place, as
this tool can be very valuable without the rest of the patch as well,
and since it's a read-only tool that only runs when the cluster is down
it should be a lot safer.
pull/27/merge
Magnus Hagander
8 years ago
parent
03c11796a9
commit
a228cc13ae
@ -1,855 +0,0 @@ |
||||
/*-------------------------------------------------------------------------
|
||||
* |
||||
* checksumhelper.c |
||||
* Background worker to walk the database and write checksums to pages |
||||
* |
||||
* When enabling data checksums on a database at initdb time, no extra process |
||||
* is required as each page is checksummed, and verified, at accesses. When |
||||
* enabling checksums on an already running cluster, which was not initialized |
||||
* with checksums, this helper worker will ensure that all pages are |
||||
* checksummed before verification of the checksums is turned on. |
||||
* |
||||
* Portions Copyright (c) 1996-2018, PostgreSQL Global Development Group |
||||
* Portions Copyright (c) 1994, Regents of the University of California |
||||
* |
||||
* |
||||
* IDENTIFICATION |
||||
* src/backend/postmaster/checksumhelper.c |
||||
* |
||||
*------------------------------------------------------------------------- |
||||
*/ |
||||
#include "postgres.h" |
||||
|
||||
#include "access/heapam.h" |
||||
#include "access/htup_details.h" |
||||
#include "access/xact.h" |
||||
#include "catalog/pg_database.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/checksumhelper.h" |
||||
#include "storage/bufmgr.h" |
||||
#include "storage/checksum.h" |
||||
#include "storage/lmgr.h" |
||||
#include "storage/ipc.h" |
||||
#include "storage/procarray.h" |
||||
#include "storage/smgr.h" |
||||
#include "tcop/tcopprot.h" |
||||
#include "utils/lsyscache.h" |
||||
#include "utils/ps_status.h" |
||||
|
||||
|
||||
typedef enum |
||||
{ |
||||
SUCCESSFUL = 0, |
||||
ABORTED, |
||||
FAILED |
||||
} ChecksumHelperResult; |
||||
|
||||
typedef struct ChecksumHelperShmemStruct |
||||
{ |
||||
pg_atomic_flag launcher_started; |
||||
ChecksumHelperResult success; |
||||
bool process_shared_catalogs; |
||||
bool abort; |
||||
/* Parameter values set on start */ |
||||
int cost_delay; |
||||
int cost_limit; |
||||
} ChecksumHelperShmemStruct; |
||||
|
||||
/* Shared memory segment for checksumhelper */ |
||||
static ChecksumHelperShmemStruct * ChecksumHelperShmem; |
||||
|
||||
/* Bookkeeping for work to do */ |
||||
typedef struct ChecksumHelperDatabase |
||||
{ |
||||
Oid dboid; |
||||
char *dbname; |
||||
} ChecksumHelperDatabase; |
||||
|
||||
typedef struct ChecksumHelperRelation |
||||
{ |
||||
Oid reloid; |
||||
char relkind; |
||||
} ChecksumHelperRelation; |
||||
|
||||
/* Prototypes */ |
||||
static List *BuildDatabaseList(void); |
||||
static List *BuildRelationList(bool include_shared); |
||||
static List *BuildTempTableList(void); |
||||
static ChecksumHelperResult ProcessDatabase(ChecksumHelperDatabase * db); |
||||
static void launcher_cancel_handler(SIGNAL_ARGS); |
||||
|
||||
/*
|
||||
* Main entry point for checksumhelper launcher process. |
||||
*/ |
||||
bool |
||||
StartChecksumHelperLauncher(int cost_delay, int cost_limit) |
||||
{ |
||||
BackgroundWorker bgw; |
||||
BackgroundWorkerHandle *bgw_handle; |
||||
|
||||
if (ChecksumHelperShmem->abort) |
||||
{ |
||||
ereport(ERROR, |
||||
(errmsg("could not start checksumhelper: has been cancelled"))); |
||||
} |
||||
|
||||
ChecksumHelperShmem->cost_delay = cost_delay; |
||||
ChecksumHelperShmem->cost_limit = cost_limit; |
||||
|
||||
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, "ChecksumHelperLauncherMain"); |
||||
snprintf(bgw.bgw_name, BGW_MAXLEN, "checksumhelper launcher"); |
||||
snprintf(bgw.bgw_type, BGW_MAXLEN, "checksumhelper launcher"); |
||||
bgw.bgw_restart_time = BGW_NEVER_RESTART; |
||||
bgw.bgw_notify_pid = MyProcPid; |
||||
bgw.bgw_main_arg = (Datum) 0; |
||||
|
||||
if (!pg_atomic_test_set_flag(&ChecksumHelperShmem->launcher_started)) |
||||
{ |
||||
/* Failed to set means somebody else started */ |
||||
ereport(ERROR, |
||||
(errmsg("could not start checksumhelper: already running"))); |
||||
} |
||||
|
||||
if (!RegisterDynamicBackgroundWorker(&bgw, &bgw_handle)) |
||||
{ |
||||
pg_atomic_clear_flag(&ChecksumHelperShmem->launcher_started); |
||||
return false; |
||||
} |
||||
|
||||
return true; |
||||
} |
||||
|
||||
/*
|
||||
* ShutdownChecksumHelperIfRunning |
||||
* Request shutdown of the checksumhelper |
||||
* |
||||
* This does not turn off processing immediately, it signals the checksum |
||||
* process to end when done with the current block. |
||||
*/ |
||||
void |
||||
ShutdownChecksumHelperIfRunning(void) |
||||
{ |
||||
/* If the launcher isn't started, there is nothing to shut down */ |
||||
if (pg_atomic_unlocked_test_flag(&ChecksumHelperShmem->launcher_started)) |
||||
return; |
||||
|
||||
/*
|
||||
* We don't need an atomic variable for aborting, setting it multiple |
||||
* times will not change the handling. |
||||
*/ |
||||
ChecksumHelperShmem->abort = true; |
||||
} |
||||
|
||||
/*
|
||||
* ProcessSingleRelationFork |
||||
* Enable checksums in a single relation/fork. |
||||
* |
||||
* Returns true if successful, and false if *aborted*. On error, an actual |
||||
* error is raised in the lower levels. |
||||
*/ |
||||
static bool |
||||
ProcessSingleRelationFork(Relation reln, ForkNumber forkNum, BufferAccessStrategy strategy) |
||||
{ |
||||
BlockNumber numblocks = RelationGetNumberOfBlocksInFork(reln, forkNum); |
||||
BlockNumber b; |
||||
char activity[NAMEDATALEN * 2 + 128]; |
||||
|
||||
for (b = 0; b < numblocks; b++) |
||||
{ |
||||
Buffer buf = ReadBufferExtended(reln, forkNum, b, RBM_NORMAL, strategy); |
||||
|
||||
/*
|
||||
* Report to pgstat every 100 blocks (so as not to "spam") |
||||
*/ |
||||
if ((b % 100) == 0) |
||||
{ |
||||
snprintf(activity, sizeof(activity) - 1, "processing: %s.%s (%s block %d/%d)", |
||||
get_namespace_name(RelationGetNamespace(reln)), RelationGetRelationName(reln), |
||||
forkNames[forkNum], b, numblocks); |
||||
pgstat_report_activity(STATE_RUNNING, activity); |
||||
} |
||||
|
||||
/* Need to get an exclusive lock before we can flag 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. hintbits) on the master 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. |
||||
*/ |
||||
START_CRIT_SECTION(); |
||||
MarkBufferDirty(buf); |
||||
log_newpage_buffer(buf, false); |
||||
END_CRIT_SECTION(); |
||||
|
||||
UnlockReleaseBuffer(buf); |
||||
|
||||
/*
|
||||
* This is the only place where we check if we are asked to abort, the |
||||
* abortion will bubble up from here. |
||||
*/ |
||||
if (ChecksumHelperShmem->abort) |
||||
return false; |
||||
|
||||
vacuum_delay_point(); |
||||
} |
||||
|
||||
return true; |
||||
} |
||||
|
||||
/*
|
||||
* ProcessSingleRelationByOid |
||||
* Process a single relation based on oid. |
||||
* |
||||
* Returns true if successful, and false if *aborted*. On error, an actual error |
||||
* is raised in the lower levels. |
||||
*/ |
||||
static bool |
||||
ProcessSingleRelationByOid(Oid relationId, BufferAccessStrategy strategy) |
||||
{ |
||||
Relation rel; |
||||
ForkNumber fnum; |
||||
bool aborted = false; |
||||
|
||||
StartTransactionCommand(); |
||||
|
||||
elog(DEBUG2, "Checksumhelper starting to process relation %d", relationId); |
||||
rel = try_relation_open(relationId, AccessShareLock); |
||||
if (rel == NULL) |
||||
{ |
||||
/*
|
||||
* Relation no longer exist. We consider this a success, since there |
||||
* are no pages in it that need checksums, and thus return true. |
||||
*/ |
||||
elog(DEBUG1, "Checksumhelper skipping relation %d as it no longer exists", relationId); |
||||
CommitTransactionCommand(); |
||||
pgstat_report_activity(STATE_IDLE, NULL); |
||||
return true; |
||||
} |
||||
RelationOpenSmgr(rel); |
||||
|
||||
for (fnum = 0; fnum <= MAX_FORKNUM; fnum++) |
||||
{ |
||||
if (smgrexists(rel->rd_smgr, fnum)) |
||||
{ |
||||
if (!ProcessSingleRelationFork(rel, fnum, strategy)) |
||||
{ |
||||
aborted = true; |
||||
break; |
||||
} |
||||
} |
||||
} |
||||
relation_close(rel, AccessShareLock); |
||||
elog(DEBUG2, "Checksumhelper done with relation %d: %s", |
||||
relationId, (aborted ? "aborted" : "finished")); |
||||
|
||||
CommitTransactionCommand(); |
||||
|
||||
pgstat_report_activity(STATE_IDLE, NULL); |
||||
|
||||
return !aborted; |
||||
} |
||||
|
||||
/*
|
||||
* ProcessDatabase |
||||
* Enable checksums in a single database. |
||||
* |
||||
* We do this by launching a dynamic background worker into this database, and |
||||
* waiting for it to finish. We have to do this in a separate worker, since |
||||
* each process can only be connected to one database during its lifetime. |
||||
*/ |
||||
static ChecksumHelperResult |
||||
ProcessDatabase(ChecksumHelperDatabase * db) |
||||
{ |
||||
BackgroundWorker bgw; |
||||
BackgroundWorkerHandle *bgw_handle; |
||||
BgwHandleStatus status; |
||||
pid_t pid; |
||||
char activity[NAMEDATALEN + 64]; |
||||
|
||||
ChecksumHelperShmem->success = FAILED; |
||||
|
||||
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, "ChecksumHelperWorkerMain"); |
||||
snprintf(bgw.bgw_name, BGW_MAXLEN, "checksumhelper worker"); |
||||
snprintf(bgw.bgw_type, BGW_MAXLEN, "checksumhelper worker"); |
||||
bgw.bgw_restart_time = BGW_NEVER_RESTART; |
||||
bgw.bgw_notify_pid = MyProcPid; |
||||
bgw.bgw_main_arg = ObjectIdGetDatum(db->dboid); |
||||
|
||||
if (!RegisterDynamicBackgroundWorker(&bgw, &bgw_handle)) |
||||
{ |
||||
ereport(LOG, |
||||
(errmsg("failed to start worker for checksumhelper in \"%s\"", |
||||
db->dbname))); |
||||
return FAILED; |
||||
} |
||||
|
||||
status = WaitForBackgroundWorkerStartup(bgw_handle, &pid); |
||||
if (status != BGWH_STARTED) |
||||
{ |
||||
ereport(LOG, |
||||
(errmsg("failed to wait for worker startup for checksumhelper in \"%s\"", |
||||
db->dbname))); |
||||
return FAILED; |
||||
} |
||||
|
||||
ereport(DEBUG1, |
||||
(errmsg("started background worker for checksums in \"%s\"", |
||||
db->dbname))); |
||||
|
||||
snprintf(activity, sizeof(activity) - 1, |
||||
"Waiting for worker in database %s (pid %d)", db->dbname, pid); |
||||
pgstat_report_activity(STATE_RUNNING, activity); |
||||
|
||||
|
||||
status = WaitForBackgroundWorkerShutdown(bgw_handle); |
||||
if (status != BGWH_STOPPED) |
||||
{ |
||||
ereport(LOG, |
||||
(errmsg("failed to wait for worker shutdown for checksumhelper in \"%s\"", |
||||
db->dbname))); |
||||
return FAILED; |
||||
} |
||||
|
||||
if (ChecksumHelperShmem->success == ABORTED) |
||||
ereport(LOG, |
||||
(errmsg("checksumhelper was aborted during processing in \"%s\"", |
||||
db->dbname))); |
||||
|
||||
ereport(DEBUG1, |
||||
(errmsg("background worker for checksums in \"%s\" completed", |
||||
db->dbname))); |
||||
|
||||
pgstat_report_activity(STATE_IDLE, NULL); |
||||
|
||||
return ChecksumHelperShmem->success; |
||||
} |
||||
|
||||
static void |
||||
launcher_exit(int code, Datum arg) |
||||
{ |
||||
ChecksumHelperShmem->abort = false; |
||||
pg_atomic_clear_flag(&ChecksumHelperShmem->launcher_started); |
||||
} |
||||
|
||||
static void |
||||
launcher_cancel_handler(SIGNAL_ARGS) |
||||
{ |
||||
ChecksumHelperShmem->abort = true; |
||||
} |
||||
|
||||
static void |
||||
WaitForAllTransactionsToFinish(void) |
||||
{ |
||||
TransactionId waitforxid; |
||||
|
||||
LWLockAcquire(XidGenLock, LW_SHARED); |
||||
waitforxid = ShmemVariableCache->nextXid; |
||||
LWLockRelease(XidGenLock); |
||||
|
||||
while (true) |
||||
{ |
||||
TransactionId oldestxid = GetOldestActiveTransactionId(); |
||||
|
||||
elog(DEBUG1, "Checking old transactions"); |
||||
if (TransactionIdPrecedes(oldestxid, waitforxid)) |
||||
{ |
||||
char activity[64]; |
||||
|
||||
/* Oldest running xid is older than us, so wait */ |
||||
snprintf(activity, sizeof(activity), "Waiting for current transactions to finish (waiting for %d)", waitforxid); |
||||
pgstat_report_activity(STATE_RUNNING, activity); |
||||
|
||||
/* Retry every 5 seconds */ |
||||
ResetLatch(MyLatch); |
||||
(void) WaitLatch(MyLatch, |
||||
WL_LATCH_SET | WL_TIMEOUT, |
||||
5000, |
||||
WAIT_EVENT_PG_SLEEP); |
||||
} |
||||
else |
||||
{ |
||||
pgstat_report_activity(STATE_IDLE, NULL); |
||||
return; |
||||
} |
||||
} |
||||
} |
||||
|
||||
void |
||||
ChecksumHelperLauncherMain(Datum arg) |
||||
{ |
||||
List *DatabaseList; |
||||
List *remaining = NIL; |
||||
ListCell *lc, |
||||
*lc2; |
||||
List *CurrentDatabases = NIL; |
||||
bool found_failed = false; |
||||
|
||||
on_shmem_exit(launcher_exit, 0); |
||||
|
||||
ereport(DEBUG1, |
||||
(errmsg("checksumhelper launcher started"))); |
||||
|
||||
pqsignal(SIGTERM, die); |
||||
pqsignal(SIGINT, launcher_cancel_handler); |
||||
|
||||
BackgroundWorkerUnblockSignals(); |
||||
|
||||
init_ps_display(pgstat_get_backend_desc(B_CHECKSUMHELPER_LAUNCHER), "", "", ""); |
||||
|
||||
/*
|
||||
* Initialize a connection to shared catalogs only. |
||||
*/ |
||||
BackgroundWorkerInitializeConnection(NULL, NULL, 0); |
||||
|
||||
/*
|
||||
* Set up so first run processes shared catalogs, but not once in every |
||||
* db. |
||||
*/ |
||||
ChecksumHelperShmem->process_shared_catalogs = true; |
||||
|
||||
/*
|
||||
* Wait for all existing transactions to finish. This will make sure that |
||||
* we can see all tables all databases, so we don't miss any. Anything |
||||
* created after this point is known to have checksums on all pages |
||||
* already, so we don't have to care about those. |
||||
*/ |
||||
WaitForAllTransactionsToFinish(); |
||||
|
||||
/*
|
||||
* Create a database list. We don't need to concern ourselves with |
||||
* rebuilding this list during runtime since any database created after |
||||
* this process started will be running with checksums turned on from the |
||||
* start. |
||||
*/ |
||||
DatabaseList = BuildDatabaseList(); |
||||
|
||||
/*
|
||||
* If there are no databases at all to checksum, we can exit immediately |
||||
* as there is no work to do. |
||||
*/ |
||||
if (DatabaseList == NIL || list_length(DatabaseList) == 0) |
||||
return; |
||||
|
||||
foreach(lc, DatabaseList) |
||||
{ |
||||
ChecksumHelperDatabase *db = (ChecksumHelperDatabase *) lfirst(lc); |
||||
ChecksumHelperResult processing; |
||||
|
||||
processing = ProcessDatabase(db); |
||||
|
||||
if (processing == SUCCESSFUL) |
||||
{ |
||||
pfree(db->dbname); |
||||
pfree(db); |
||||
|
||||
if (ChecksumHelperShmem->process_shared_catalogs) |
||||
|
||||
/*
|
||||
* Now that one database has completed shared catalogs, we |
||||
* don't have to process them again. |
||||
*/ |
||||
ChecksumHelperShmem->process_shared_catalogs = false; |
||||
} |
||||
else if (processing == FAILED) |
||||
{ |
||||
/*
|
||||
* Put failed databases on the remaining list. |
||||
*/ |
||||
remaining = lappend(remaining, db); |
||||
} |
||||
else |
||||
/* aborted */ |
||||
return; |
||||
} |
||||
list_free(DatabaseList); |
||||
|
||||
/*
|
||||
* remaining now has all databases not yet processed. This can be because |
||||
* they failed for some reason, or because the database was dropped |
||||
* between us getting the database list and trying to process it. Get a |
||||
* fresh list of databases to detect the second case where the database |
||||
* was dropped before we had started processing it. If a database still |
||||
* exists, but enabling checksums failed then we fail the entire |
||||
* checksumming process and exit with an error. |
||||
*/ |
||||
CurrentDatabases = BuildDatabaseList(); |
||||
|
||||
foreach(lc, remaining) |
||||
{ |
||||
ChecksumHelperDatabase *db = (ChecksumHelperDatabase *) lfirst(lc); |
||||
bool found = false; |
||||
|
||||
foreach(lc2, CurrentDatabases) |
||||
{ |
||||
ChecksumHelperDatabase *db2 = (ChecksumHelperDatabase *) lfirst(lc2); |
||||
|
||||
if (db->dboid == db2->dboid) |
||||
{ |
||||
found = true; |
||||
ereport(WARNING, |
||||
(errmsg("failed to enable checksums in \"%s\"", |
||||
db->dbname))); |
||||
break; |
||||
} |
||||
} |
||||
|
||||
if (found) |
||||
found_failed = true; |
||||
else |
||||
{ |
||||
ereport(LOG, |
||||
(errmsg("database \"%s\" has been dropped, skipping", |
||||
db->dbname))); |
||||
} |
||||
|
||||
pfree(db->dbname); |
||||
pfree(db); |
||||
} |
||||
list_free(remaining); |
||||
|
||||
/* Free the extra list of databases */ |
||||
foreach(lc, CurrentDatabases) |
||||
{ |
||||
ChecksumHelperDatabase *db = (ChecksumHelperDatabase *) lfirst(lc); |
||||
|
||||
pfree(db->dbname); |
||||
pfree(db); |
||||
} |
||||
list_free(CurrentDatabases); |
||||
|
||||
if (found_failed) |
||||
{ |
||||
/* Disable checksums on cluster, because we failed */ |
||||
SetDataChecksumsOff(); |
||||
ereport(ERROR, |
||||
(errmsg("checksumhelper failed to enable checksums in all databases, aborting"))); |
||||
} |
||||
|
||||
/*
|
||||
* Force a checkpoint to get everything out to disk. |
||||
*/ |
||||
RequestCheckpoint(CHECKPOINT_FORCE | CHECKPOINT_WAIT | CHECKPOINT_IMMEDIATE); |
||||
|
||||
/*
|
||||
* Everything has been processed, so flag checksums enabled. |
||||
*/ |
||||
SetDataChecksumsOn(); |
||||
|
||||
ereport(LOG, |
||||
(errmsg("checksums enabled, checksumhelper launcher shutting down"))); |
||||
} |
||||
|
||||
/*
|
||||
* ChecksumHelperShmemSize |
||||
* Compute required space for checksumhelper-related shared memory |
||||
*/ |
||||
Size |
||||
ChecksumHelperShmemSize(void) |
||||
{ |
||||
Size size; |
||||
|
||||
size = sizeof(ChecksumHelperShmemStruct); |
||||
size = MAXALIGN(size); |
||||
|
||||
return size; |
||||
} |
||||
|
||||
/*
|
||||
* ChecksumHelperShmemInit |
||||
* Allocate and initialize checksumhelper-related shared memory |
||||
*/ |
||||
void |
||||
ChecksumHelperShmemInit(void) |
||||
{ |
||||
bool found; |
||||
|
||||
ChecksumHelperShmem = (ChecksumHelperShmemStruct *) |
||||
ShmemInitStruct("ChecksumHelper Data", |
||||
ChecksumHelperShmemSize(), |
||||
&found); |
||||
|
||||
if (!found) |
||||
{ |
||||
MemSet(ChecksumHelperShmem, 0, ChecksumHelperShmemSize()); |
||||
pg_atomic_init_flag(&ChecksumHelperShmem->launcher_started); |
||||
} |
||||
} |
||||
|
||||
/*
|
||||
* BuildDatabaseList |
||||
* Compile a list of all currently available databases in the cluster |
||||
* |
||||
* This creates the list of databases for the checksumhelper workers to add |
||||
* checksums to. |
||||
*/ |
||||
static List * |
||||
BuildDatabaseList(void) |
||||
{ |
||||
List *DatabaseList = NIL; |
||||
Relation rel; |
||||
HeapScanDesc scan; |
||||
HeapTuple tup; |
||||
MemoryContext ctx = CurrentMemoryContext; |
||||
MemoryContext oldctx; |
||||
|
||||
StartTransactionCommand(); |
||||
|
||||
rel = heap_open(DatabaseRelationId, AccessShareLock); |
||||
scan = heap_beginscan_catalog(rel, 0, NULL); |
||||
|
||||
while (HeapTupleIsValid(tup = heap_getnext(scan, ForwardScanDirection))) |
||||
{ |
||||
Form_pg_database pgdb = (Form_pg_database) GETSTRUCT(tup); |
||||
ChecksumHelperDatabase *db; |
||||
|
||||
oldctx = MemoryContextSwitchTo(ctx); |
||||
|
||||
db = (ChecksumHelperDatabase *) palloc(sizeof(ChecksumHelperDatabase)); |
||||
|
||||
db->dboid = HeapTupleGetOid(tup); |
||||
db->dbname = pstrdup(NameStr(pgdb->datname)); |
||||
|
||||
DatabaseList = lappend(DatabaseList, db); |
||||
|
||||
MemoryContextSwitchTo(oldctx); |
||||
} |
||||
|
||||
heap_endscan(scan); |
||||
heap_close(rel, AccessShareLock); |
||||
|
||||
CommitTransactionCommand(); |
||||
|
||||
return DatabaseList; |
||||
} |
||||
|
||||
/*
|
||||
* BuildRelationList |
||||
* Compile a list of all relations in the database |
||||
* |
||||
* If shared is true, both shared relations and local ones are returned, else |
||||
* all non-shared relations are returned. |
||||
* Temp tables are not included. |
||||
*/ |
||||
static List * |
||||
BuildRelationList(bool include_shared) |
||||
{ |
||||
List *RelationList = NIL; |
||||
Relation rel; |
||||
HeapScanDesc scan; |
||||
HeapTuple tup; |
||||
MemoryContext ctx = CurrentMemoryContext; |
||||
MemoryContext oldctx; |
||||
|
||||
StartTransactionCommand(); |
||||
|
||||
rel = heap_open(RelationRelationId, AccessShareLock); |
||||
scan = heap_beginscan_catalog(rel, 0, NULL); |
||||
|
||||
while (HeapTupleIsValid(tup = heap_getnext(scan, ForwardScanDirection))) |
||||
{ |
||||
Form_pg_class pgc = (Form_pg_class) GETSTRUCT(tup); |
||||
ChecksumHelperRelation *relentry; |
||||
|
||||
if (pgc->relpersistence == 't') |
||||
continue; |
||||
|
||||
if (pgc->relisshared && !include_shared) |
||||
continue; |
||||
|
||||
/*
|
||||
* Foreign tables have by definition no local storage that can be |
||||
* checksummed, so skip. |
||||
*/ |
||||
if (pgc->relkind == RELKIND_FOREIGN_TABLE) |
||||
continue; |
||||
|
||||
oldctx = MemoryContextSwitchTo(ctx); |
||||
relentry = (ChecksumHelperRelation *) palloc(sizeof(ChecksumHelperRelation)); |
||||
|
||||
relentry->reloid = HeapTupleGetOid(tup); |
||||
relentry->relkind = pgc->relkind; |
||||
|
||||
RelationList = lappend(RelationList, relentry); |
||||
|
||||
MemoryContextSwitchTo(oldctx); |
||||
} |
||||
|
||||
heap_endscan(scan); |
||||
heap_close(rel, AccessShareLock); |
||||
|
||||
CommitTransactionCommand(); |
||||
|
||||
return RelationList; |
||||
} |
||||
|
||||
/*
|
||||
* BuildTempTableList |
||||
* Compile a list of all temporary tables in database |
||||
* |
||||
* Returns a List of oids. |
||||
*/ |
||||
static List * |
||||
BuildTempTableList(void) |
||||
{ |
||||
List *RelationList = NIL; |
||||
Relation rel; |
||||
HeapScanDesc scan; |
||||
HeapTuple tup; |
||||
MemoryContext ctx = CurrentMemoryContext; |
||||
MemoryContext oldctx; |
||||
|
||||
StartTransactionCommand(); |
||||
|
||||
rel = heap_open(RelationRelationId, AccessShareLock); |
||||
scan = heap_beginscan_catalog(rel, 0, NULL); |
||||
|
||||
while (HeapTupleIsValid(tup = heap_getnext(scan, ForwardScanDirection))) |
||||
{ |
||||
Form_pg_class pgc = (Form_pg_class) GETSTRUCT(tup); |
||||
|
||||
if (pgc->relpersistence != 't') |
||||
continue; |
||||
|
||||
oldctx = MemoryContextSwitchTo(ctx); |
||||
RelationList = lappend_oid(RelationList, HeapTupleGetOid(tup)); |
||||
MemoryContextSwitchTo(oldctx); |
||||
} |
||||
|
||||
heap_endscan(scan); |
||||
heap_close(rel, AccessShareLock); |
||||
|
||||
CommitTransactionCommand(); |
||||
|
||||
return RelationList; |
||||
} |
||||
|
||||
/*
|
||||
* Main function for enabling checksums in a single database |
||||
*/ |
||||
void |
||||
ChecksumHelperWorkerMain(Datum arg) |
||||
{ |
||||
Oid dboid = DatumGetObjectId(arg); |
||||
List *RelationList = NIL; |
||||
List *InitialTempTableList = NIL; |
||||
ListCell *lc; |
||||
BufferAccessStrategy strategy; |
||||
bool aborted = false; |
||||
|
||||
pqsignal(SIGTERM, die); |
||||
|
||||
BackgroundWorkerUnblockSignals(); |
||||
|
||||
init_ps_display(pgstat_get_backend_desc(B_CHECKSUMHELPER_WORKER), "", "", ""); |
||||
|
||||
ereport(DEBUG1, |
||||
(errmsg("checksum worker starting for database oid %d", dboid))); |
||||
|
||||
BackgroundWorkerInitializeConnectionByOid(dboid, InvalidOid, BGWORKER_BYPASS_ALLOWCONN); |
||||
|
||||
/*
|
||||
* 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 those files and modify them. |
||||
*/ |
||||
InitialTempTableList = BuildTempTableList(); |
||||
|
||||
/*
|
||||
* Enable vacuum cost delay, if any. |
||||
*/ |
||||
VacuumCostDelay = ChecksumHelperShmem->cost_delay; |
||||
VacuumCostLimit = ChecksumHelperShmem->cost_limit; |
||||
VacuumCostActive = (VacuumCostDelay > 0); |
||||
VacuumCostBalance = 0; |
||||
VacuumPageHit = 0; |
||||
VacuumPageMiss = 0; |
||||
VacuumPageDirty = 0; |
||||
|
||||
/*
|
||||
* Create and set the vacuum strategy as our buffer strategy. |
||||
*/ |
||||
strategy = GetAccessStrategy(BAS_VACUUM); |
||||
|
||||
RelationList = BuildRelationList(ChecksumHelperShmem->process_shared_catalogs); |
||||
foreach(lc, RelationList) |
||||
{ |
||||
ChecksumHelperRelation *rel = (ChecksumHelperRelation *) lfirst(lc); |
||||
|
||||
if (!ProcessSingleRelationByOid(rel->reloid, strategy)) |
||||
{ |
||||
aborted = true; |
||||
break; |
||||
} |
||||
} |
||||
list_free_deep(RelationList); |
||||
|
||||
if (aborted) |
||||
{ |
||||
ChecksumHelperShmem->success = ABORTED; |
||||
ereport(DEBUG1, |
||||
(errmsg("checksum worker aborted in database oid %d", dboid))); |
||||
return; |
||||
} |
||||
|
||||
/*
|
||||
* 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 state), so those are safe. |
||||
*/ |
||||
while (true) |
||||
{ |
||||
List *CurrentTempTables; |
||||
ListCell *lc; |
||||
int numleft; |
||||
char activity[64]; |
||||
|
||||
CurrentTempTables = BuildTempTableList(); |
||||
numleft = 0; |
||||
foreach(lc, InitialTempTableList) |
||||
{ |
||||
if (list_member_oid(CurrentTempTables, lfirst_oid(lc))) |
||||
numleft++; |
||||
} |
||||
list_free(CurrentTempTables); |
||||
|
||||
if (numleft == 0) |
||||
break; |
||||
|
||||
/* At least one temp table left to wait for */ |
||||
snprintf(activity, sizeof(activity), "Waiting for %d temp tables to be removed", numleft); |
||||
pgstat_report_activity(STATE_RUNNING, activity); |
||||
|
||||
/* Retry every 5 seconds */ |
||||
ResetLatch(MyLatch); |
||||
(void) WaitLatch(MyLatch, |
||||
WL_LATCH_SET | WL_TIMEOUT, |
||||
5000, |
||||
WAIT_EVENT_PG_SLEEP); |
||||
} |
||||
|
||||
list_free(InitialTempTableList); |
||||
|
||||
ChecksumHelperShmem->success = SUCCESSFUL; |
||||
ereport(DEBUG1, |
||||
(errmsg("checksum worker completed in database oid %d", dboid))); |
||||
} |
||||
@ -1,31 +0,0 @@ |
||||
/*-------------------------------------------------------------------------
|
||||
* |
||||
* checksumhelper.h |
||||
* header file for checksum helper background worker |
||||
* |
||||
* |
||||
* Portions Copyright (c) 1996-2018, PostgreSQL Global Development Group |
||||
* Portions Copyright (c) 1994, Regents of the University of California |
||||
* |
||||
* src/include/postmaster/checksumhelper.h |
||||
* |
||||
*------------------------------------------------------------------------- |
||||
*/ |
||||
#ifndef CHECKSUMHELPER_H |
||||
#define CHECKSUMHELPER_H |
||||
|
||||
/* Shared memory */ |
||||
extern Size ChecksumHelperShmemSize(void); |
||||
extern void ChecksumHelperShmemInit(void); |
||||
|
||||
/* Start the background processes for enabling checksums */ |
||||
bool StartChecksumHelperLauncher(int cost_delay, int cost_limit); |
||||
|
||||
/* Shutdown the background processes, if any */ |
||||
void ShutdownChecksumHelperIfRunning(void); |
||||
|
||||
/* Background worker entrypoints */ |
||||
void ChecksumHelperLauncherMain(Datum arg); |
||||
void ChecksumHelperWorkerMain(Datum arg); |
||||
|
||||
#endif /* CHECKSUMHELPER_H */ |
||||
@ -1,2 +0,0 @@ |
||||
# Generated by test suite |
||||
/tmp_check/ |
||||
@ -1,24 +0,0 @@ |
||||
#-------------------------------------------------------------------------
|
||||
#
|
||||
# Makefile for src/test/checksum
|
||||
#
|
||||
# Portions Copyright (c) 1996-2018, PostgreSQL Global Development Group
|
||||
# Portions Copyright (c) 1994, Regents of the University of California
|
||||
#
|
||||
# src/test/checksum/Makefile
|
||||
#
|
||||
#-------------------------------------------------------------------------
|
||||
|
||||
subdir = src/test/checksum
|
||||
top_builddir = ../../..
|
||||
include $(top_builddir)/src/Makefile.global |
||||
|
||||
check: |
||||
$(prove_check)
|
||||
|
||||
installcheck: |
||||
$(prove_installcheck)
|
||||
|
||||
clean distclean maintainer-clean: |
||||
rm -rf tmp_check
|
||||
|
||||
@ -1,22 +0,0 @@ |
||||
src/test/checksum/README |
||||
|
||||
Regression tests for data checksums |
||||
=================================== |
||||
|
||||
This directory contains a test suite for enabling data checksums |
||||
in a running cluster with streaming replication. |
||||
|
||||
Running the tests |
||||
================= |
||||
|
||||
make check |
||||
|
||||
or |
||||
|
||||
make installcheck |
||||
|
||||
NOTE: This creates a temporary installation (in the case of "check"), |
||||
with multiple nodes, be they master or standby(s) for the purpose of |
||||
the tests. |
||||
|
||||
NOTE: This requires the --enable-tap-tests argument to configure. |
||||
@ -1,101 +0,0 @@ |
||||
# Test suite for testing enabling data checksums with streaming replication |
||||
use strict; |
||||
use warnings; |
||||
use PostgresNode; |
||||
use TestLib; |
||||
use Test::More tests => 10; |
||||
|
||||
my $MAX_TRIES = 30; |
||||
|
||||
# Initialize master node |
||||
my $node_master = get_new_node('master'); |
||||
$node_master->init(allows_streaming => 1); |
||||
$node_master->start; |
||||
my $backup_name = 'my_backup'; |
||||
|
||||
# Take backup |
||||
$node_master->backup($backup_name); |
||||
|
||||
# Create streaming standby linking to master |
||||
my $node_standby_1 = get_new_node('standby_1'); |
||||
$node_standby_1->init_from_backup($node_master, $backup_name, |
||||
has_streaming => 1); |
||||
$node_standby_1->start; |
||||
|
||||
# Create some content on master to have un-checksummed data in the cluster |
||||
$node_master->safe_psql('postgres', |
||||
"CREATE TABLE t AS SELECT generate_series(1,10000) AS a;"); |
||||
|
||||
# Wait for standbys to catch up |
||||
$node_master->wait_for_catchup($node_standby_1, 'replay', |
||||
$node_master->lsn('insert')); |
||||
|
||||
# Check that checksums are turned off |
||||
my $result = $node_master->safe_psql('postgres', |
||||
"SELECT setting FROM pg_catalog.pg_settings WHERE name = 'data_checksums';"); |
||||
is($result, "off", 'ensure checksums are turned off on master'); |
||||
|
||||
$result = $node_standby_1->safe_psql('postgres', |
||||
"SELECT setting FROM pg_catalog.pg_settings WHERE name = 'data_checksums';"); |
||||
is($result, "off", 'ensure checksums are turned off on standby_1'); |
||||
|
||||
# Enable checksums for the cluster |
||||
$node_master->safe_psql('postgres', "SELECT pg_enable_data_checksums();"); |
||||
|
||||
# Ensure that the master has switched to inprogress immediately |
||||
$result = $node_master->safe_psql('postgres', |
||||
"SELECT setting FROM pg_catalog.pg_settings WHERE name = 'data_checksums';"); |
||||
is($result, "inprogress", 'ensure checksums are in progress on master'); |
||||
|
||||
# Wait for checksum enable to be replayed |
||||
$node_master->wait_for_catchup($node_standby_1, 'replay'); |
||||
|
||||
# Ensure that the standby has switched to inprogress |
||||
$result = $node_standby_1->safe_psql('postgres', |
||||
"SELECT setting FROM pg_catalog.pg_settings WHERE name = 'data_checksums';"); |
||||
is($result, "inprogress", 'ensure checksums are in progress on standby_1'); |
||||
|
||||
# Insert some more data which should be checksummed on INSERT |
||||
$node_master->safe_psql('postgres', |
||||
"INSERT INTO t VALUES (generate_series(1,10000));"); |
||||
|
||||
# Wait for checksums enabled on the master |
||||
for (my $i = 0; $i < $MAX_TRIES; $i++) |
||||
{ |
||||
$result = $node_master->safe_psql('postgres', |
||||
"SELECT setting FROM pg_catalog.pg_settings WHERE name = 'data_checksums';"); |
||||
last if ($result eq 'on'); |
||||
sleep(1); |
||||
} |
||||
is ($result, "on", 'ensure checksums are enabled on master'); |
||||
|
||||
# Wait for checksums enabled on the standby |
||||
for (my $i = 0; $i < $MAX_TRIES; $i++) |
||||
{ |
||||
$result = $node_standby_1->safe_psql('postgres', |
||||
"SELECT setting FROM pg_catalog.pg_settings WHERE name = 'data_checksums';"); |
||||
last if ($result eq 'on'); |
||||
sleep(1); |
||||
} |
||||
is ($result, "on", 'ensure checksums are enabled on standby'); |
||||
|
||||
$result = $node_master->safe_psql('postgres', "SELECT count(a) FROM t"); |
||||
is ($result, "20000", 'ensure we can safely read all data with checksums'); |
||||
|
||||
# Disable checksums and ensure it's propagated to standby and that we can |
||||
# still read all data |
||||
$node_master->safe_psql('postgres', "SELECT pg_disable_data_checksums();"); |
||||
$result = $node_master->safe_psql('postgres', |
||||
"SELECT setting FROM pg_catalog.pg_settings WHERE name = 'data_checksums';"); |
||||
is($result, "off", 'ensure checksums are in progress on master'); |
||||
|
||||
# Wait for checksum disable to be replayed |
||||
$node_master->wait_for_catchup($node_standby_1, 'replay'); |
||||
|
||||
# Ensure that the standby has switched to off |
||||
$result = $node_standby_1->safe_psql('postgres', |
||||
"SELECT setting FROM pg_catalog.pg_settings WHERE name = 'data_checksums';"); |
||||
is($result, "off", 'ensure checksums are in progress on standby_1'); |
||||
|
||||
$result = $node_master->safe_psql('postgres', "SELECT count(a) FROM t"); |
||||
is ($result, "20000", 'ensure we can safely read all data without checksums'); |
||||
@ -1,27 +0,0 @@ |
||||
Parsed test spec with 2 sessions |
||||
|
||||
starting permutation: c_verify_checksums_off r_seqread c_enable_checksums c_verify_checksums_inprogress c_disable_checksums c_wait_checksums_off |
||||
step c_verify_checksums_off: SELECT setting = 'off' FROM pg_catalog.pg_settings WHERE name = 'data_checksums'; |
||||
?column? |
||||
|
||||
t |
||||
step r_seqread: SELECT * FROM reader_loop(); |
||||
reader_loop |
||||
|
||||
t |
||||
step c_enable_checksums: SELECT pg_enable_data_checksums(1000); |
||||
pg_enable_data_checksums |
||||
|
||||
|
||||
step c_verify_checksums_inprogress: SELECT setting = 'inprogress' FROM pg_catalog.pg_settings WHERE name = 'data_checksums'; |
||||
?column? |
||||
|
||||
t |
||||
step c_disable_checksums: SELECT pg_disable_data_checksums(); |
||||
pg_disable_data_checksums |
||||
|
||||
|
||||
step c_wait_checksums_off: SELECT test_checksums_off(); |
||||
test_checksums_off |
||||
|
||||
t |
||||
@ -1,27 +0,0 @@ |
||||
Parsed test spec with 3 sessions |
||||
|
||||
starting permutation: c_verify_checksums_off w_insert100k r_seqread c_enable_checksums c_wait_for_checksums c_verify_checksums_on |
||||
step c_verify_checksums_off: SELECT setting = 'off' FROM pg_catalog.pg_settings WHERE name = 'data_checksums'; |
||||
?column? |
||||
|
||||
t |
||||
step w_insert100k: SELECT insert_1k(100); |
||||
insert_1k |
||||
|
||||
t |
||||
step r_seqread: SELECT * FROM reader_loop(); |
||||
reader_loop |
||||
|
||||
t |
||||
step c_enable_checksums: SELECT pg_enable_data_checksums(); |
||||
pg_enable_data_checksums |
||||
|
||||
|
||||
step c_wait_for_checksums: SELECT test_checksums_on(); |
||||
test_checksums_on |
||||
|
||||
t |
||||
step c_verify_checksums_on: SELECT setting = 'on' FROM pg_catalog.pg_settings WHERE name = 'data_checksums'; |
||||
?column? |
||||
|
||||
t |
||||
@ -1,47 +0,0 @@ |
||||
setup |
||||
{ |
||||
CREATE TABLE t1 (a serial, b integer, c text); |
||||
INSERT INTO t1 (b, c) VALUES (generate_series(1,10000), 'starting values'); |
||||
|
||||
CREATE OR REPLACE FUNCTION test_checksums_off() RETURNS boolean AS $$ |
||||
DECLARE |
||||
enabled boolean; |
||||
BEGIN |
||||
PERFORM pg_sleep(1); |
||||
SELECT setting = 'off' INTO enabled FROM pg_catalog.pg_settings WHERE name = 'data_checksums'; |
||||
RETURN enabled; |
||||
END; |
||||
$$ LANGUAGE plpgsql; |
||||
|
||||
CREATE OR REPLACE FUNCTION reader_loop() RETURNS boolean AS $$ |
||||
DECLARE |
||||
counter integer; |
||||
enabled boolean; |
||||
BEGIN |
||||
FOR counter IN 1..100 LOOP |
||||
PERFORM count(a) FROM t1; |
||||
END LOOP; |
||||
RETURN True; |
||||
END; |
||||
$$ LANGUAGE plpgsql; |
||||
} |
||||
|
||||
teardown |
||||
{ |
||||
DROP FUNCTION reader_loop(); |
||||
DROP FUNCTION test_checksums_off(); |
||||
|
||||
DROP TABLE t1; |
||||
} |
||||
|
||||
session "reader" |
||||
step "r_seqread" { SELECT * FROM reader_loop(); } |
||||
|
||||
session "checksums" |
||||
step "c_verify_checksums_off" { SELECT setting = 'off' FROM pg_catalog.pg_settings WHERE name = 'data_checksums'; } |
||||
step "c_enable_checksums" { SELECT pg_enable_data_checksums(1000); } |
||||
step "c_disable_checksums" { SELECT pg_disable_data_checksums(); } |
||||
step "c_verify_checksums_inprogress" { SELECT setting = 'inprogress' FROM pg_catalog.pg_settings WHERE name = 'data_checksums'; } |
||||
step "c_wait_checksums_off" { SELECT test_checksums_off(); } |
||||
|
||||
permutation "c_verify_checksums_off" "r_seqread" "c_enable_checksums" "c_verify_checksums_inprogress" "c_disable_checksums" "c_wait_checksums_off" |
||||
@ -1,70 +0,0 @@ |
||||
setup |
||||
{ |
||||
CREATE TABLE t1 (a serial, b integer, c text); |
||||
INSERT INTO t1 (b, c) VALUES (generate_series(1,10000), 'starting values'); |
||||
|
||||
CREATE OR REPLACE FUNCTION insert_1k(iterations int) RETURNS boolean AS $$ |
||||
DECLARE |
||||
counter integer; |
||||
BEGIN |
||||
FOR counter IN 1..$1 LOOP |
||||
INSERT INTO t1 (b, c) VALUES ( |
||||
generate_series(1, 1000), |
||||
array_to_string(array(select chr(97 + (random() * 25)::int) from generate_series(1,250)), '') |
||||
); |
||||
PERFORM pg_sleep(0.1); |
||||
END LOOP; |
||||
RETURN True; |
||||
END; |
||||
$$ LANGUAGE plpgsql; |
||||
|
||||
CREATE OR REPLACE FUNCTION test_checksums_on() RETURNS boolean AS $$ |
||||
DECLARE |
||||
enabled boolean; |
||||
BEGIN |
||||
LOOP |
||||
SELECT setting = 'on' INTO enabled FROM pg_catalog.pg_settings WHERE name = 'data_checksums'; |
||||
IF enabled THEN |
||||
EXIT; |
||||
END IF; |
||||
PERFORM pg_sleep(1); |
||||
END LOOP; |
||||
RETURN enabled; |
||||
END; |
||||
$$ LANGUAGE plpgsql; |
||||
|
||||
CREATE OR REPLACE FUNCTION reader_loop() RETURNS boolean AS $$ |
||||
DECLARE |
||||
counter integer; |
||||
BEGIN |
||||
FOR counter IN 1..30 LOOP |
||||
PERFORM count(a) FROM t1; |
||||
PERFORM pg_sleep(0.2); |
||||
END LOOP; |
||||
RETURN True; |
||||
END; |
||||
$$ LANGUAGE plpgsql; |
||||
} |
||||
|
||||
teardown |
||||
{ |
||||
DROP FUNCTION reader_loop(); |
||||
DROP FUNCTION test_checksums_on(); |
||||
DROP FUNCTION insert_1k(int); |
||||
|
||||
DROP TABLE t1; |
||||
} |
||||
|
||||
session "writer" |
||||
step "w_insert100k" { SELECT insert_1k(100); } |
||||
|
||||
session "reader" |
||||
step "r_seqread" { SELECT * FROM reader_loop(); } |
||||
|
||||
session "checksums" |
||||
step "c_verify_checksums_off" { SELECT setting = 'off' FROM pg_catalog.pg_settings WHERE name = 'data_checksums'; } |
||||
step "c_enable_checksums" { SELECT pg_enable_data_checksums(); } |
||||
step "c_wait_for_checksums" { SELECT test_checksums_on(); } |
||||
step "c_verify_checksums_on" { SELECT setting = 'on' FROM pg_catalog.pg_settings WHERE name = 'data_checksums'; } |
||||
|
||||
permutation "c_verify_checksums_off" "w_insert100k" "r_seqread" "c_enable_checksums" "c_wait_for_checksums" "c_verify_checksums_on" |
||||
Loading…
Reference in new issue