From df9be35749cc0fc80e187f9b3056419439fa15d6 Mon Sep 17 00:00:00 2001 From: Masahiko Sawada Date: Thu, 23 Jan 2025 16:36:37 -0800 Subject: [PATCH v1] PoC: Convert wal_level a PGC_SIGHUP parameter. --- src/backend/access/rmgrdesc/xlogdesc.c | 7 + src/backend/access/transam/Makefile | 1 + src/backend/access/transam/meson.build | 1 + src/backend/access/transam/xlog.c | 38 +- src/backend/access/transam/xloglevelworker.c | 614 ++++++++++++++++++ src/backend/commands/publicationcmds.c | 2 +- src/backend/postmaster/bgworker.c | 4 + src/backend/postmaster/checkpointer.c | 7 + src/backend/postmaster/pgarch.c | 8 + src/backend/postmaster/postmaster.c | 20 +- src/backend/replication/logical/decode.c | 10 + src/backend/replication/logical/logical.c | 2 +- src/backend/replication/logical/slotsync.c | 10 +- src/backend/replication/slot.c | 6 +- src/backend/replication/walsender.c | 29 + src/backend/storage/ipc/ipci.c | 2 + src/backend/storage/ipc/procsignal.c | 4 + src/backend/storage/ipc/standby.c | 6 +- .../utils/activity/wait_event_names.txt | 1 + src/backend/utils/init/postinit.c | 3 + src/backend/utils/misc/guc_tables.c | 4 +- src/include/access/xlog.h | 20 +- src/include/access/xloglevelworker.h | 18 + src/include/catalog/pg_control.h | 1 + src/include/postmaster/pgarch.h | 1 + src/include/replication/walsender.h | 1 + src/include/storage/lwlocklist.h | 1 + src/include/storage/procsignal.h | 2 + src/include/utils/guc_hooks.h | 1 + 29 files changed, 791 insertions(+), 33 deletions(-) create mode 100644 src/backend/access/transam/xloglevelworker.c create mode 100644 src/include/access/xloglevelworker.h diff --git a/src/backend/access/rmgrdesc/xlogdesc.c b/src/backend/access/rmgrdesc/xlogdesc.c index 58040f28656..da5331047ee 100644 --- a/src/backend/access/rmgrdesc/xlogdesc.c +++ b/src/backend/access/rmgrdesc/xlogdesc.c @@ -164,6 +164,13 @@ xlog_desc(StringInfo buf, XLogReaderState *record) { int wal_level; + memcpy(&wal_level, rec, sizeof(int)); + appendStringInfo(buf, "wal_level %s", get_wal_level_string(wal_level)); + } + else if (info == XLOG_UPDATE_WAL_LEVEL) + { + int wal_level; + memcpy(&wal_level, rec, sizeof(int)); appendStringInfo(buf, "wal_level %s", get_wal_level_string(wal_level)); } diff --git a/src/backend/access/transam/Makefile b/src/backend/access/transam/Makefile index 661c55a9db7..e98719aba73 100644 --- a/src/backend/access/transam/Makefile +++ b/src/backend/access/transam/Makefile @@ -32,6 +32,7 @@ OBJS = \ xlogbackup.o \ xlogfuncs.o \ xloginsert.o \ + xloglevelworker.o \ xlogprefetcher.o \ xlogreader.o \ xlogrecovery.o \ diff --git a/src/backend/access/transam/meson.build b/src/backend/access/transam/meson.build index e8ae9b13c8e..27947c8b902 100644 --- a/src/backend/access/transam/meson.build +++ b/src/backend/access/transam/meson.build @@ -20,6 +20,7 @@ backend_sources += files( 'xlogbackup.c', 'xlogfuncs.c', 'xloginsert.c', + 'xloglevelworker.c', 'xlogprefetcher.c', 'xlogrecovery.c', 'xlogstats.c', diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index bf3dbda901d..7095ded6d0b 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -5918,6 +5918,15 @@ StartupXLOG(void) if (InRecovery) ResetUnloggedRelations(UNLOGGED_RELATION_INIT); + /* + * During the recovery, we might have changed wal_level to 'minimal' and + * reloaded the configuration file. Therefore, we need to terminate some + * processes who are unavailable with 'minimal' level such as archiver and + * walsenders. + */ + if (InRecovery) + TerminateUnavailableProcessesWithMinimal(); + /* * Pre-scan prepared transactions to find out the range of XIDs present. * This information is not quite needed yet, but it is positioned here so @@ -7001,7 +7010,7 @@ CreateCheckPoint(int flags) WALInsertLockAcquireExclusive(); checkPoint.fullPageWrites = Insert->fullPageWrites; - checkPoint.wal_level = wal_level; + checkPoint.wal_level = GetActiveWalLevel(); if (shutdown) { @@ -7055,9 +7064,11 @@ CreateCheckPoint(int flags) */ if (!shutdown) { + int level = GetActiveWalLevel(); + /* Include WAL level in record for WAL summarizer's benefit. */ XLogBeginInsert(); - XLogRegisterData((char *) &wal_level, sizeof(wal_level)); + XLogRegisterData((char *) &level, sizeof(level)); (void) XLogInsert(RM_XLOG_ID, XLOG_CHECKPOINT_REDO); /* @@ -7382,7 +7393,7 @@ CreateEndOfRecoveryRecord(void) elog(ERROR, "can only be used to end recovery"); xlrec.end_time = GetCurrentTimestamp(); - xlrec.wal_level = wal_level; + xlrec.wal_level = GetActiveWalLevel(); WALInsertLockAcquireExclusive(); xlrec.ThisTimeLineID = XLogCtl->InsertTimeLineID; @@ -8511,7 +8522,7 @@ xlog_redo(XLogReaderState *record) */ if (InRecovery && InHotStandby && xlrec.wal_level < WAL_LEVEL_LOGICAL && - wal_level >= WAL_LEVEL_LOGICAL) + GetActiveWalLevel() >= WAL_LEVEL_LOGICAL) InvalidateObsoleteReplicationSlots(RS_INVAL_WAL_LEVEL, 0, InvalidOid, InvalidTransactionId); @@ -8583,6 +8594,25 @@ xlog_redo(XLogReaderState *record) { /* nothing to do here, just for informational purposes */ } + else if (info == XLOG_UPDATE_WAL_LEVEL) + { + int new_wal_level; + + memcpy(&new_wal_level, XLogRecGetData(record), sizeof(new_wal_level)); + + if (InRecovery && InHotStandby && + new_wal_level < WAL_LEVEL_LOGICAL && + GetActiveWalLevel() >= WAL_LEVEL_LOGICAL) + InvalidateObsoleteReplicationSlots(RS_INVAL_WAL_LEVEL, + 0, InvalidOid, + InvalidTransactionId); + + /* Update our copy of the wal_level parameter in pg_control */ + LWLockAcquire(ControlFileLock, LW_EXCLUSIVE); + ControlFile->wal_level = new_wal_level; + UpdateControlFile(); + LWLockRelease(ControlFileLock); + } } /* diff --git a/src/backend/access/transam/xloglevelworker.c b/src/backend/access/transam/xloglevelworker.c new file mode 100644 index 00000000000..286ee411f65 --- /dev/null +++ b/src/backend/access/transam/xloglevelworker.c @@ -0,0 +1,614 @@ +/*------------------------------------------------------------------------- + * xloglevelworker.c + * Functionality for controlling wal_level value online. + * + * This file contains the codes for maintaining the 'wal_level' on system-wide + * as well as for the wal_level control worker, a background worker who + * increases or decrease the wal_level value online. + * + * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group + * Portions Copyright (c) 1994, Regents of the University of California + * + * + * IDENTIFICATION + * src/backend/access/transam/xloginfo.c + *------------------------------------------------------------------------- + */ + +#include "postgres.h" + +#include "access/xact.h" +#include "access/xloginsert.h" +#include "access/xloglevelworker.h" +#include "catalog/pg_control.h" +#include "miscadmin.h" +#include "postmaster/bgworker.h" +#include "postmaster/bgwriter.h" +#include "postmaster/interrupt.h" +#include "postmaster/pgarch.h" +#include "replication/logicalxlog.h" +#include "replication/slot.h" +#include "storage/ipc.h" +#include "storage/lmgr.h" +#include "storage/procsignal.h" +#include "tcop/tcopprot.h" +#include "utils/guc_hooks.h" + +typedef enum WalLevelCtlState +{ + /* + * This state corresponds to wal_level 'minimal'. We don't WAL-logging for + * neither archival nor log-shipping. + */ + WAL_LEVEL_CTL_STATE_MINIMAL, + + /* + * In this state, we enable WAL-logging information required only for + * archival and log-shipping (and hot standby). In other word, + * XLogStandbyInfoActive() returns true from this state. Note that these + * functionalities are still not enabled until the + * WAL_LEVEL_CTL_STATE_REPLICA state. + * + * This is a ephemeral state where we use for changing wal_level between + * 'minimal' and 'replica'. Therefore, this state never be the "target" + * state when changing wal_level. + */ + WAL_LEVEL_CTL_STATE_STANDBY_INFO_LOGGING, + + /* + * This state corresponds to wal_level 'replica', meaning we allow + * archival, log-shipping, and hot standby. + */ + WAL_LEVEL_CTL_STATE_REPLICA, + + /* + * In this state, we enable WAL-logging information required for logical + * decoding. In other word, XLogLogicalInfoActive() returns true from this + * state. But note that the logical decoding and replication are not + * enabled yet. + * + * This is a ephemeral state where we use for changing wal_level between + * 'replica' and 'logical'. Therefore, this state never be the "target" + * state when changing wal_level. + */ + WAL_LEVEL_CTL_STATE_LOGICAL_INFO_LOGGING, + + /* + * This state corresponds to wal_level 'logical'. We can create the + * logical replication slot and use the logical decoding and logical + * replication in this state. + */ + WAL_LEVEL_CTL_STATE_LOGICAL, +} WalLevelCtlState; + +typedef struct WalLevelCtlData +{ + /* + * This state is the autoritatie value used by all backends to know the + * active wal_level (and some transition states), which affects what + * information is logged in WAL records and which WAL-related features + * such as replication, WAL archiving and logical decoding can be used in + * the system. + * + * The process-local wal_level value, which is automatically updated by + * GUC update, might not represent the up-to-date WAL level. Instead, we + * need to use GetActiveWalLevel() to get the active WAL level. + */ + WalLevelCtlState state; + + /* A valid pid if a worker is working */ + pid_t worker_pid; +} WalLevelCtlData; +WalLevelCtlData *WalLevelCtl = NULL; + +/* + * Both doStandbyInfoLogging and doLogicalInfoLogging are backend-local + * caches to determine if information required by hot standby and logical + * decoding need to be written in WAL records. These local variables are + * based on WalLevelCtl->state and are updated when (1) process startup + * and (2) processing the global barrier. + */ +static bool doStandbyInfoLogging = false; +static bool doLogicalInfoLogging = false; + +static void updateWalLevelCtlState(WalLevelCtlState new_state, bool notify); +static void writeUpdateWalLevel(int new_wal_level); + +Size +WalLevelCtlShmemSize(void) +{ + return sizeof(WalLevelCtlData); +} + +void +WalLevelCtlShmemInit(void) +{ + bool found; + + WalLevelCtl = ShmemInitStruct("wal_level control", + WalLevelCtlShmemSize(), + &found); + + if (!found) + { + WalLevelCtl->state = WAL_LEVEL_CTL_STATE_REPLICA; + WalLevelCtl->worker_pid = InvalidPid; + } +} + +/* + * This must be called during postmaster startup. + */ +void +InitializeWalLevelCtl(void) +{ + WalLevelCtlState initial_state; + + if (wal_level == WAL_LEVEL_MINIMAL) + initial_state = WAL_LEVEL_CTL_STATE_MINIMAL; + else if (wal_level == WAL_LEVEL_REPLICA) + initial_state = WAL_LEVEL_CTL_STATE_REPLICA; + else + initial_state = WAL_LEVEL_CTL_STATE_LOGICAL; + + WalLevelCtl->state = initial_state; +} + +static void +update_wal_logging_state(void) +{ + LWLockAcquire(WalLevelControlLock, LW_SHARED); + + doStandbyInfoLogging = + (WalLevelCtl->state >= WAL_LEVEL_CTL_STATE_STANDBY_INFO_LOGGING); + doLogicalInfoLogging = + (WalLevelCtl->state >= WAL_LEVEL_CTL_STATE_LOGICAL_INFO_LOGGING); + + LWLockRelease(WalLevelControlLock); +} + +/* + * Initialize process-local xlog info status. This must be called during the + * process startup time. + */ +void +InitWalLoggingState(void) +{ + update_wal_logging_state(); +} + +/* + * This function is called when we are ordered to update both doLogicalInfoLogging + * and doStandbyInfoLogging by a ProcSignalBarrier. + */ +bool +ProcessBarrierUpdateWalLoggingState(void) +{ + update_wal_logging_state(); + return true; +} + +/* + * Return true if the logical info logging is enabled. + */ +bool +LogicalInfoLoggingEnabled(void) +{ + return doLogicalInfoLogging; +} +bool +StandbyInfoLoggingEnabled(void) +{ + return doStandbyInfoLogging; +} + +/* + * Return the wal_level int value for the given WalLevelCtlState. + */ +static int +WalLevelCtlStateGetLevel(WalLevelCtlState state) +{ + if (state <= WAL_LEVEL_CTL_STATE_STANDBY_INFO_LOGGING) + return WAL_LEVEL_MINIMAL; + else if (state <= WAL_LEVEL_CTL_STATE_LOGICAL_INFO_LOGGING) + return WAL_LEVEL_REPLICA; + else + return WAL_LEVEL_LOGICAL; +} + +/* + * Return WalLevelCtlState value for the given wal_level int value. + */ +static WalLevelCtlState +WalLevelGetCtlState(int level) +{ + if (level == WAL_LEVEL_MINIMAL) + return WAL_LEVEL_CTL_STATE_MINIMAL; + else if (level == WAL_LEVEL_REPLICA) + return WAL_LEVEL_CTL_STATE_REPLICA; + else + return WAL_LEVEL_CTL_STATE_LOGICAL; +} + +int +GetActiveWalLevel(void) +{ + WalLevelCtlState state; + + LWLockAcquire(WalLevelControlLock, LW_SHARED); + state = WalLevelCtl->state; + LWLockRelease(WalLevelControlLock); + + return WalLevelCtlStateGetLevel(state); +} + +static const char * +WalLevelGetString(int level) +{ + if (level == WAL_LEVEL_MINIMAL) + return "minimal"; + else if (level == WAL_LEVEL_REPLICA) + return "replica"; + else + return "logical"; +} + +/* + * Return true if the logical decoding is ready to use. + */ +bool +LogicalDecodingEnabled(void) +{ + return GetActiveWalLevel() >= WAL_LEVEL_LOGICAL; +} + +void +TerminateUnavailableProcessesWithMinimal(void) +{ + int active_wal_level = GetActiveWalLevel(); + + if (active_wal_level == WAL_LEVEL_MINIMAL) + { + PgArchShutdown(); + + WalSndTerminate(); + WalSndWaitStopping(); + } +} + +void +WalLevelCtlWorkerLaunchIfNecessary(void) +{ + BackgroundWorker bgw; + BackgroundWorkerHandle *bgw_handle; + + /* + * During the recovery, we don't need a complex wal_level transition + * process and the wal-level control worker since any writes are not + * permitted yet. We can directly set the active wal_level value. After + * the recovery we will do some actions based on the active wal_level at + * that time, see TerminateUnavailableProcessesWithMinimal() for example. + */ + if (RecoveryInProgress()) + { + updateWalLevelCtlState(WalLevelGetCtlState(wal_level), false); + return; + } + + LWLockAcquire(WalLevelControlLock, LW_SHARED); + + if (WalLevelCtl->state == WalLevelGetCtlState(wal_level)) + { + LWLockRelease(WalLevelControlLock); + return; + } + + if (WalLevelCtl->worker_pid != InvalidPid) + { + LWLockRelease(WalLevelControlLock); + return; + } + + LWLockRelease(WalLevelControlLock); + + /* Register the new dynamic worker */ + memset(&bgw, 0, sizeof(bgw)); + bgw.bgw_flags = BGWORKER_SHMEM_ACCESS; + bgw.bgw_start_time = BgWorkerStart_RecoveryFinished; + snprintf(bgw.bgw_library_name, MAXPGPATH, "postgres"); + snprintf(bgw.bgw_function_name, BGW_MAXLEN, "WalLevelCtlWorkerMain"); + snprintf(bgw.bgw_name, BGW_MAXLEN, "wal level control worker"); + snprintf(bgw.bgw_type, BGW_MAXLEN, "wal level control worker"); + bgw.bgw_restart_time = 5; + bgw.bgw_notify_pid = MyProcPid; + + if (!RegisterDynamicBackgroundWorker(&bgw, &bgw_handle)) + ereport(WARNING, + (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED), + errmsg("out of background worker slots"), + errhint("You might need to increase \"%s\".", "max_worker_processes"))); +} + +static void +updateWalLevelCtlState(WalLevelCtlState new_state, bool notify) +{ + LWLockAcquire(WalLevelControlLock, LW_EXCLUSIVE); + WalLevelCtl->state = new_state; + LWLockRelease(WalLevelControlLock); + + if (notify) + { + WaitForProcSignalBarrier( + EmitProcSignalBarrier(PROCSIGNAL_BARRIER_UPDATE_WAL_LOGGING_STATE)); + } +} + +static void +writeUpdateWalLevel(int new_wal_level) +{ + XLogBeginInsert(); + XLogRegisterData((char *) (&new_wal_level), sizeof(bool)); + XLogInsert(RM_XLOG_ID, XLOG_UPDATE_WAL_LEVEL); +} + +static void +WalLevelIncreaseToReplica(void) +{ + /* + * Enable WAL-logging with information required by archival and + * log-shipping, and order all processes to reflect this change. Note that + * WAL archiving and physical replication are not enabled yet. + */ + updateWalLevelCtlState(WAL_LEVEL_CTL_STATE_STANDBY_INFO_LOGGING, true); + + /* + * We create a checkpoint to increase wal_level from 'minimal' so that we + * can restart from there in case of a server crash. + */ + RequestCheckpoint(CHECKPOINT_IMMEDIATE | CHECKPOINT_FORCE | CHECKPOINT_WAIT); + + START_CRIT_SECTION(); + + /* + * Switch to wal_level 'replica' now, letting the postmaster start WAL + * archiving and log-shipping from here. + * + * XXX: explain why need to set the state on shmem first and then write a + * WAL. + */ + updateWalLevelCtlState(WAL_LEVEL_CTL_STATE_REPLICA, false); + writeUpdateWalLevel(WAL_LEVEL_REPLICA); + + END_CRIT_SECTION(); + + ereport(LOG, + (errmsg("wal_level has been increased to \"replica\""))); +} + +static void +WalLevelIncreaseToLogical(void) +{ + /* + * Enable WAL-logging with logical information required by the logical + * decoding, and order all processes to reflect this change. From this + * point, it's guaranteed that all processes are writing WAL records with + * logical information. However, note that the logical decoding is still + * disabled. + */ + updateWalLevelCtlState(WAL_LEVEL_CTL_STATE_LOGICAL_INFO_LOGGING, true); + + START_CRIT_SECTION(); + + /* + * Switch to wal_level 'logical' now, allowing the logical decoding etc. + * + * XXX: explain why need to set the state on shmem first and then write a + * WAL. + */ + updateWalLevelCtlState(WAL_LEVEL_CTL_STATE_LOGICAL, false); + writeUpdateWalLevel(WAL_LEVEL_LOGICAL); + + END_CRIT_SECTION(); + + ereport(LOG, + (errmsg("wal_level has been increased to \"logical\""))); +} + +static void +WalLevelDecreaseToReplica(void) +{ + /* + * We still allow processes to write WAL records with logical information + * but disable the logical decoding so that we don't start new logical + * decoding from here. + */ + updateWalLevelCtlState(WAL_LEVEL_CTL_STATE_LOGICAL_INFO_LOGGING, false); + + /* + * Invalidate all logical replication slots, terminating processes doing + * logical decoding (and logical replication) too. + */ + InvalidateObsoleteReplicationSlots(RS_INVAL_WAL_LEVEL, + 0, InvalidOid, + InvalidTransactionId); + + START_CRIT_SECTION(); + + /* + * Switch to wal_level 'replica' now. + * + * XXX: explain why need to write a WAL first, and then set the state on + * shmem. + */ + writeUpdateWalLevel(WAL_LEVEL_REPLICA); + updateWalLevelCtlState(WAL_LEVEL_CTL_STATE_REPLICA, false); + + END_CRIT_SECTION(); + + /* Order all processed to disable logical information WAL-logging */ + WaitForProcSignalBarrier(EmitProcSignalBarrier(PROCSIGNAL_BARRIER_UPDATE_WAL_LOGGING_STATE)); + + ereport(LOG, + (errmsg("wal_level has been decreased to \"replica\""))); +} + +static void +ShutdownBackgoundProcessesForMinimal(void) +{ + PgArchShutdown(); + + WalSndTerminate(); + WalSndWaitStopping(); + + /* XXX any other process that should shutdown? */ +} + +static void +WalLevelDecreaseToMinimal(void) +{ + /* + * We still allow processes to write WAL records with information required + * by WAL archiving and log-shipping, but prevent these functionalities + * from newly starting. + */ + updateWalLevelCtlState(WAL_LEVEL_CTL_STATE_STANDBY_INFO_LOGGING, false); + + ShutdownBackgoundProcessesForMinimal(); + + START_CRIT_SECTION(); + + /* + * Switch to wal_level 'minimal' now. + * + * XXX: explain why need to write a WAL first, and then set the state on + * shmem. + */ + writeUpdateWalLevel(WAL_LEVEL_MINIMAL); + updateWalLevelCtlState(WAL_LEVEL_CTL_STATE_MINIMAL, false); + + END_CRIT_SECTION(); + + /* Order all processed to disable standby information WAL-logging */ + WaitForProcSignalBarrier(EmitProcSignalBarrier(PROCSIGNAL_BARRIER_UPDATE_WAL_LOGGING_STATE)); + + ereport(LOG, (errmsg("wal_level has been decreased to \"minimal\""))); +} + +static void +WalLevelWorkerShutdown(int code, Datum arg) +{ + WalLevelCtl->worker_pid = InvalidPid; +} + +void +WalLevelCtlWorkerMain(Datum main_arg) +{ + WalLevelCtlState state; + WalLevelCtlState target_state; + + ereport(LOG, + (errmsg("wal_level control worker started"))); + + before_shmem_exit(WalLevelWorkerShutdown, (Datum) 0); + + LWLockAcquire(WalLevelControlLock, LW_EXCLUSIVE); + + if (WalLevelCtl->worker_pid != InvalidPid) + { + LWLockRelease(WalLevelControlLock); + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + (errmsg("wal level worker is already running, cannot start")))); + } + + WalLevelCtl->worker_pid = MyProcPid; + + LWLockRelease(WalLevelControlLock); + + pqsignal(SIGHUP, SignalHandlerForConfigReload); + pqsignal(SIGTERM, die); + BackgroundWorkerUnblockSignals(); + + target_state = WalLevelGetCtlState(wal_level); + state = WalLevelCtl->state; + + ereport(LOG, + (errmsg("changing wal_level from \"%s\" to \"%s\"", + WalLevelGetString(WalLevelCtlStateGetLevel(state)), + WalLevelGetString(WalLevelCtlStateGetLevel(target_state))))); + + if (state == target_state) + proc_exit(0); + + if (state < target_state) + { + /* Increase wal_level to 'replica' or 'logical' */ + + switch (state) + { + case WAL_LEVEL_CTL_STATE_MINIMAL: + case WAL_LEVEL_CTL_STATE_STANDBY_INFO_LOGGING: + + WalLevelIncreaseToReplica(); + + /* FALL THROUGH */ + case WAL_LEVEL_CTL_STATE_REPLICA: + case WAL_LEVEL_CTL_STATE_LOGICAL_INFO_LOGGING: + { + if (target_state == WAL_LEVEL_CTL_STATE_REPLICA) + break; + + WalLevelIncreaseToLogical(); + break; + } + default: + elog(LOG, "unexpected wal level state: %d", state); + break; + } + } + else + { + /* Decrease wal_level to 'replica' or 'minimal' */ + + switch (state) + { + case WAL_LEVEL_CTL_STATE_LOGICAL: + { + WalLevelDecreaseToReplica(); + } + /* FALL THROUGH */ + case WAL_LEVEL_CTL_STATE_REPLICA: + case WAL_LEVEL_CTL_STATE_LOGICAL_INFO_LOGGING: + { + if (target_state == WAL_LEVEL_CTL_STATE_REPLICA) + break; + + WalLevelDecreaseToMinimal(); + break; + } + default: + elog(LOG, "unexpected wal level state: %d", state); + break; + } + } + + ereport(LOG, + (errmsg("successfully changed wal_level from \"%s\" to \"%s\"", + WalLevelGetString(WalLevelCtlStateGetLevel(state)), + WalLevelGetString(WalLevelCtlStateGetLevel(target_state))))); + + proc_exit(0); +} + +const char * +show_wal_level(void) +{ + int active_level = GetActiveWalLevel(); + + if (active_level == WAL_LEVEL_MINIMAL) + return "minimal"; + else if (active_level == WAL_LEVEL_REPLICA) + return "replica"; + else + return "logical"; +} diff --git a/src/backend/commands/publicationcmds.c b/src/backend/commands/publicationcmds.c index 35747b3df5f..3537103136b 100644 --- a/src/backend/commands/publicationcmds.c +++ b/src/backend/commands/publicationcmds.c @@ -898,7 +898,7 @@ CreatePublication(ParseState *pstate, CreatePublicationStmt *stmt) InvokeObjectPostCreateHook(PublicationRelationId, puboid, 0); - if (wal_level != WAL_LEVEL_LOGICAL) + if (!XLogLogicalInfoActive()) ereport(WARNING, (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), errmsg("\"wal_level\" is insufficient to publish logical changes"), diff --git a/src/backend/postmaster/bgworker.c b/src/backend/postmaster/bgworker.c index b288915cec8..efb0ff18e8a 100644 --- a/src/backend/postmaster/bgworker.c +++ b/src/backend/postmaster/bgworker.c @@ -13,6 +13,7 @@ #include "postgres.h" #include "access/parallel.h" +#include "access/xloglevelworker.h" #include "libpq/pqsignal.h" #include "miscadmin.h" #include "pgstat.h" @@ -132,6 +133,9 @@ static const struct }, { "TablesyncWorkerMain", TablesyncWorkerMain + }, + { + "WalLevelCtlWorkerMain", WalLevelCtlWorkerMain } }; diff --git a/src/backend/postmaster/checkpointer.c b/src/backend/postmaster/checkpointer.c index 9bfd0fd665c..5e91339a328 100644 --- a/src/backend/postmaster/checkpointer.c +++ b/src/backend/postmaster/checkpointer.c @@ -39,6 +39,7 @@ #include "access/xlog.h" #include "access/xlog_internal.h" #include "access/xlogrecovery.h" +#include "access/xloglevelworker.h" #include "libpq/pqsignal.h" #include "miscadmin.h" #include "pgstat.h" @@ -1336,6 +1337,12 @@ UpdateSharedMemoryConfig(void) */ UpdateFullPageWrites(); + /* + * If wal_level has been changed by SIGHUP, delegate changing wal_level + * value to the wal level control worker. + */ + WalLevelCtlWorkerLaunchIfNecessary(); + elog(DEBUG2, "checkpointer updated shared memory configuration values"); } diff --git a/src/backend/postmaster/pgarch.c b/src/backend/postmaster/pgarch.c index 12ee815a626..86206149594 100644 --- a/src/backend/postmaster/pgarch.c +++ b/src/backend/postmaster/pgarch.c @@ -291,6 +291,14 @@ PgArchWakeup(void) SetLatch(&ProcGlobal->allProcs[arch_pgprocno].procLatch); } +void +PgArchShutdown(void) +{ + int arch_pgprocno = PgArch->pgprocno; + + if (arch_pgprocno != INVALID_PROC_NUMBER) + kill(GetPGProcByNumber(arch_pgprocno)->pid, SIGUSR2); +} /* SIGUSR2 signal handler for archiver process */ static void diff --git a/src/backend/postmaster/postmaster.c b/src/backend/postmaster/postmaster.c index 5f615d0f605..5fc59f1e231 100644 --- a/src/backend/postmaster/postmaster.c +++ b/src/backend/postmaster/postmaster.c @@ -828,15 +828,6 @@ PostmasterMain(int argc, char *argv[]) MaxConnections); ExitPostmaster(1); } - if (XLogArchiveMode > ARCHIVE_MODE_OFF && wal_level == WAL_LEVEL_MINIMAL) - ereport(ERROR, - (errmsg("WAL archival cannot be enabled when \"wal_level\" is \"minimal\""))); - if (max_wal_senders > 0 && wal_level == WAL_LEVEL_MINIMAL) - ereport(ERROR, - (errmsg("WAL streaming (\"max_wal_senders\" > 0) requires \"wal_level\" to be \"replica\" or \"logical\""))); - if (summarize_wal && wal_level == WAL_LEVEL_MINIMAL) - ereport(ERROR, - (errmsg("WAL cannot be summarized when \"wal_level\" is \"minimal\""))); /* * Other one-time internal sanity checks can go here, if they are fast. @@ -1023,6 +1014,12 @@ PostmasterMain(int argc, char *argv[]) RemovePgTempFilesInDir(PG_TEMP_FILES_DIR, true, false); #endif + /* + * Initialize logical info logging and logical decoding state. We must do + * this after initializing GUCs since it reflects wal_level setting. + */ + InitializeWalLevelCtl(); + /* * Forcibly remove the files signaling a standby promotion request. * Otherwise, the existence of those files triggers a promotion too early, @@ -3213,7 +3210,7 @@ LaunchMissingBackgroundProcesses(void) * If WAL archiving is enabled always, we are allowed to start archiver * even during recovery. */ - if (PgArchPMChild == NULL && + if (GetActiveWalLevel() >= WAL_LEVEL_REPLICA && PgArchPMChild == NULL && ((XLogArchivingActive() && pmState == PM_RUN) || (XLogArchivingAlways() && (pmState == PM_RECOVERY || pmState == PM_HOT_STANDBY))) && PgArchCanRestart()) @@ -3259,7 +3256,8 @@ LaunchMissingBackgroundProcesses(void) } /* If we need to start a WAL summarizer, try to do that now */ - if (summarize_wal && WalSummarizerPMChild == NULL && + if (summarize_wal && GetActiveWalLevel() >= WAL_LEVEL_REPLICA && + WalSummarizerPMChild == NULL && (pmState == PM_RUN || pmState == PM_HOT_STANDBY) && Shutdown <= SmartShutdown) WalSummarizerPMChild = StartChildProcess(B_WAL_SUMMARIZER); diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c index 0bff0f10652..7ed476bef67 100644 --- a/src/backend/replication/logical/decode.c +++ b/src/backend/replication/logical/decode.c @@ -178,6 +178,16 @@ xlog_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) } break; } + case XLOG_UPDATE_WAL_LEVEL: + { + int new_wal_level; + + memcpy(&new_wal_level, XLogRecGetData(buf->record), sizeof(new_wal_level)); + if (new_wal_level < WAL_LEVEL_LOGICAL) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("logical decoding on standby requires \"wal_level\" >= \"logical\" on the primary"))); + } case XLOG_NOOP: case XLOG_NEXTOID: case XLOG_SWITCH: diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c index 0b25efafe2b..c67c34b19f2 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -115,7 +115,7 @@ CheckLogicalDecodingRequirements(void) * needs the same check. */ - if (wal_level < WAL_LEVEL_LOGICAL) + if (!LogicalDecodingEnabled()) ereport(ERROR, (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), errmsg("logical decoding requires \"wal_level\" >= \"logical\""))); diff --git a/src/backend/replication/logical/slotsync.c b/src/backend/replication/logical/slotsync.c index f6945af1d43..7a8927462de 100644 --- a/src/backend/replication/logical/slotsync.c +++ b/src/backend/replication/logical/slotsync.c @@ -1038,14 +1038,14 @@ ValidateSlotSyncParams(int elevel) { /* * Logical slot sync/creation requires wal_level >= logical. - * - * Since altering the wal_level requires a server restart, so error out in - * this case regardless of elevel provided by caller. */ - if (wal_level < WAL_LEVEL_LOGICAL) - ereport(ERROR, + if (!LogicalDecodingEnabled()) + { + ereport(elevel, errcode(ERRCODE_INVALID_PARAMETER_VALUE), errmsg("replication slot synchronization requires \"wal_level\" >= \"logical\"")); + return false; + } /* * A physical replication slot(primary_slot_name) is required on the diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index b30e0473e1c..2ab027f1d3d 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -1399,7 +1399,7 @@ CheckSlotRequirements(void) (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), errmsg("replication slots can only be used if \"max_replication_slots\" > 0"))); - if (wal_level < WAL_LEVEL_REPLICA) + if (GetActiveWalLevel() < WAL_LEVEL_REPLICA) ereport(ERROR, (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), errmsg("replication slots can only be used if \"wal_level\" >= \"replica\""))); @@ -2355,13 +2355,13 @@ RestoreSlotFromDisk(const char *name) * NB: Changing the requirements here also requires adapting * CheckSlotRequirements() and CheckLogicalDecodingRequirements(). */ - if (cp.slotdata.database != InvalidOid && wal_level < WAL_LEVEL_LOGICAL) + if (cp.slotdata.database != InvalidOid && GetActiveWalLevel() < WAL_LEVEL_LOGICAL) ereport(FATAL, (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), errmsg("logical replication slot \"%s\" exists, but \"wal_level\" < \"logical\"", NameStr(cp.slotdata.name)), errhint("Change \"wal_level\" to be \"logical\" or higher."))); - else if (wal_level < WAL_LEVEL_REPLICA) + else if (GetActiveWalLevel() < WAL_LEVEL_REPLICA) ereport(FATAL, (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), errmsg("physical replication slot \"%s\" exists, but \"wal_level\" < \"replica\"", diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index a0782b1bbf6..85b83584d99 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -277,6 +277,11 @@ static void WalSndSegmentOpen(XLogReaderState *state, XLogSegNo nextSegNo, void InitWalSender(void) { + if (GetActiveWalLevel() < WAL_LEVEL_REPLICA) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("WAL senders require \"wal_level\" to be \"replica\" or \"logical\""))); + am_cascading_walsender = RecoveryInProgress(); /* Create a per-walsender data structure in shared memory */ @@ -3735,6 +3740,30 @@ WalSndInitStopping(void) } } +/* + * Terminate all running walsenders. Unlike shutting down them using + * WalSndInitStopping, this function can be used to terminate walsenders + * even while normal backends are generating WAL records. + */ +void +WalSndTerminate(void) +{ + for (int i = 0; i < max_wal_senders; i++) + { + WalSnd *walsnd = &WalSndCtl->walsnds[i]; + pid_t pid; + + SpinLockAcquire(&walsnd->mutex); + pid = walsnd->pid; + SpinLockRelease(&walsnd->mutex); + + if (pid == 0) + continue; + + kill(pid, SIGUSR2); + } +} + /* * Wait that all the WAL senders have quit or reached the stopping state. This * is used by the checkpointer to control when the shutdown checkpoint can diff --git a/src/backend/storage/ipc/ipci.c b/src/backend/storage/ipc/ipci.c index 174eed70367..86ad378f169 100644 --- a/src/backend/storage/ipc/ipci.c +++ b/src/backend/storage/ipc/ipci.c @@ -148,6 +148,7 @@ CalculateShmemSize(int *num_semaphores) size = add_size(size, WaitEventCustomShmemSize()); size = add_size(size, InjectionPointShmemSize()); size = add_size(size, SlotSyncShmemSize()); + size = add_size(size, WalLevelCtlShmemSize()); /* include additional requested shmem from preload libraries */ size = add_size(size, total_addin_request); @@ -330,6 +331,7 @@ CreateOrAttachShmemStructs(void) PgArchShmemInit(); ApplyLauncherShmemInit(); SlotSyncShmemInit(); + WalLevelCtlShmemInit(); /* * Set up other modules that need some shared memory space diff --git a/src/backend/storage/ipc/procsignal.c b/src/backend/storage/ipc/procsignal.c index 7401b6e625e..b930036d3a3 100644 --- a/src/backend/storage/ipc/procsignal.c +++ b/src/backend/storage/ipc/procsignal.c @@ -18,6 +18,7 @@ #include #include "access/parallel.h" +#include "access/xlog.h" #include "commands/async.h" #include "miscadmin.h" #include "pgstat.h" @@ -573,6 +574,9 @@ ProcessProcSignalBarrier(void) case PROCSIGNAL_BARRIER_SMGRRELEASE: processed = ProcessBarrierSmgrRelease(); break; + case PROCSIGNAL_BARRIER_UPDATE_WAL_LOGGING_STATE: + processed = ProcessBarrierUpdateWalLoggingState(); + break; } /* diff --git a/src/backend/storage/ipc/standby.c b/src/backend/storage/ipc/standby.c index 2039062554d..0ddc834ebaa 100644 --- a/src/backend/storage/ipc/standby.c +++ b/src/backend/storage/ipc/standby.c @@ -498,7 +498,7 @@ ResolveRecoveryConflictWithSnapshot(TransactionId snapshotConflictHorizon, * seems OK, given that this kind of conflict should not normally be * reached, e.g. due to using a physical replication slot. */ - if (wal_level >= WAL_LEVEL_LOGICAL && isCatalogRel) + if (GetActiveWalLevel() >= WAL_LEVEL_LOGICAL && isCatalogRel) InvalidateObsoleteReplicationSlots(RS_INVAL_HORIZON, 0, locator.dbOid, snapshotConflictHorizon); } @@ -1313,13 +1313,13 @@ LogStandbySnapshot(void) * record. Fortunately this routine isn't executed frequently, and it's * only a shared lock. */ - if (wal_level < WAL_LEVEL_LOGICAL) + if (!XLogLogicalInfoActive()) LWLockRelease(ProcArrayLock); recptr = LogCurrentRunningXacts(running); /* Release lock if we kept it longer ... */ - if (wal_level >= WAL_LEVEL_LOGICAL) + if (XLogLogicalInfoActive()) LWLockRelease(ProcArrayLock); /* GetRunningTransactionData() acquired XidGenLock, we must release it */ diff --git a/src/backend/utils/activity/wait_event_names.txt b/src/backend/utils/activity/wait_event_names.txt index 0b53cba807d..d4919f9a6fc 100644 --- a/src/backend/utils/activity/wait_event_names.txt +++ b/src/backend/utils/activity/wait_event_names.txt @@ -345,6 +345,7 @@ WALSummarizer "Waiting to read or update WAL summarization state." DSMRegistry "Waiting to read or update the dynamic shared memory registry." InjectionPoint "Waiting to read or update information related to injection points." SerialControl "Waiting to read or update shared pg_serial state." +WalLevelControl "Waiting to read or update wal_level control state." # # END OF PREDEFINED LWLOCKS (DO NOT CHANGE THIS LINE) diff --git a/src/backend/utils/init/postinit.c b/src/backend/utils/init/postinit.c index 01bb6a410cb..26e0c612304 100644 --- a/src/backend/utils/init/postinit.c +++ b/src/backend/utils/init/postinit.c @@ -646,6 +646,9 @@ BaseInit(void) /* Initialize lock manager's local structs */ InitLockManagerAccess(); + /* Initialize WAL-logging state */ + InitWalLoggingState(); + /* * Initialize replication slots after pgstat. The exit hook might need to * drop ephemeral slots, which in turn triggers stats reporting. diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c index 38cb9e970d5..bd1b0a7141f 100644 --- a/src/backend/utils/misc/guc_tables.c +++ b/src/backend/utils/misc/guc_tables.c @@ -5063,13 +5063,13 @@ struct config_enum ConfigureNamesEnum[] = }, { - {"wal_level", PGC_POSTMASTER, WAL_SETTINGS, + {"wal_level", PGC_SIGHUP, WAL_SETTINGS, gettext_noop("Sets the level of information written to the WAL."), NULL }, &wal_level, WAL_LEVEL_REPLICA, wal_level_options, - NULL, NULL, NULL + NULL, NULL, show_wal_level }, { diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h index 4411c1468ac..e7cf92a6c82 100644 --- a/src/include/access/xlog.h +++ b/src/include/access/xlog.h @@ -106,7 +106,7 @@ extern PGDLLIMPORT int wal_level; * Is WAL-logging necessary for archival or log-shipping, or can we skip * WAL-logging if we fsync() the data before committing instead? */ -#define XLogIsNeeded() (wal_level >= WAL_LEVEL_REPLICA) +#define XLogIsNeeded() (StandbyInfoLoggingEnabled()) /* * Is a full-page image needed for hint bit updates? @@ -120,10 +120,10 @@ extern PGDLLIMPORT int wal_level; #define XLogHintBitIsNeeded() (DataChecksumsEnabled() || wal_log_hints) /* Do we need to WAL-log information required only for Hot Standby and logical replication? */ -#define XLogStandbyInfoActive() (wal_level >= WAL_LEVEL_REPLICA) +#define XLogStandbyInfoActive() (StandbyInfoLoggingEnabled()) /* Do we need to WAL-log information required only for logical replication? */ -#define XLogLogicalInfoActive() (wal_level >= WAL_LEVEL_LOGICAL) +#define XLogLogicalInfoActive() (LogicalInfoLoggingEnabled()) #ifdef WAL_DEBUG extern PGDLLIMPORT bool XLOG_DEBUG; @@ -270,6 +270,20 @@ extern void SetInstallXLogFileSegmentActive(void); extern bool IsInstallXLogFileSegmentActive(void); extern void XLogShutdownWalRcv(void); +/* + * Routines used by xloglevelworker.c to control WAL-logging information. + */ +extern Size WalLevelCtlShmemSize(void); +extern void WalLevelCtlShmemInit(void); +extern void InitializeWalLevelCtl(void); +extern void InitWalLoggingState(void); +extern bool ProcessBarrierUpdateWalLoggingState(void); +extern bool LogicalInfoLoggingEnabled(void); +extern bool StandbyInfoLoggingEnabled(void); +extern bool LogicalDecodingEnabled(void); +extern int GetActiveWalLevel(void); +extern void TerminateUnavailableProcessesWithMinimal(void); + /* * Routines to start, stop, and get status of a base backup. */ diff --git a/src/include/access/xloglevelworker.h b/src/include/access/xloglevelworker.h new file mode 100644 index 00000000000..1082b3994c9 --- /dev/null +++ b/src/include/access/xloglevelworker.h @@ -0,0 +1,18 @@ +/*------------------------------------------------------------------------- + * + * xloglevelworker.h + * + * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group + * + * IDENTIFICATION + * src/include/access/xloglevelworker.h + * + *------------------------------------------------------------------------- + */ +#ifndef XLOGLEVELWORKER_H +#define XLOGLEVELWORKER_H + +extern void WalLevelCtlWorkerLaunchIfNecessary(void); +extern void WalLevelCtlWorkerMain(Datum main_arg); + +#endif diff --git a/src/include/catalog/pg_control.h b/src/include/catalog/pg_control.h index 3797f25b306..62563d484ce 100644 --- a/src/include/catalog/pg_control.h +++ b/src/include/catalog/pg_control.h @@ -80,6 +80,7 @@ typedef struct CheckPoint /* 0xC0 is used in Postgres 9.5-11 */ #define XLOG_OVERWRITE_CONTRECORD 0xD0 #define XLOG_CHECKPOINT_REDO 0xE0 +#define XLOG_UPDATE_WAL_LEVEL 0xF0 /* diff --git a/src/include/postmaster/pgarch.h b/src/include/postmaster/pgarch.h index 8fc6bfeec1b..73f2d13223d 100644 --- a/src/include/postmaster/pgarch.h +++ b/src/include/postmaster/pgarch.h @@ -31,6 +31,7 @@ extern void PgArchShmemInit(void); extern bool PgArchCanRestart(void); extern void PgArchiverMain(char *startup_data, size_t startup_data_len) pg_attribute_noreturn(); extern void PgArchWakeup(void); +extern void PgArchShutdown(void); extern void PgArchForceDirScan(void); #endif /* _PGARCH_H */ diff --git a/src/include/replication/walsender.h b/src/include/replication/walsender.h index c3e8e191339..d6d426915e9 100644 --- a/src/include/replication/walsender.h +++ b/src/include/replication/walsender.h @@ -44,6 +44,7 @@ extern void WalSndSignals(void); extern Size WalSndShmemSize(void); extern void WalSndShmemInit(void); extern void WalSndWakeup(bool physical, bool logical); +extern void WalSndTerminate(void); extern void WalSndInitStopping(void); extern void WalSndWaitStopping(void); extern void HandleWalSndInitStopping(void); diff --git a/src/include/storage/lwlocklist.h b/src/include/storage/lwlocklist.h index cf565452382..979c8a4f010 100644 --- a/src/include/storage/lwlocklist.h +++ b/src/include/storage/lwlocklist.h @@ -83,3 +83,4 @@ PG_LWLOCK(49, WALSummarizer) PG_LWLOCK(50, DSMRegistry) PG_LWLOCK(51, InjectionPoint) PG_LWLOCK(52, SerialControl) +PG_LWLOCK(53, WalLevelControl) diff --git a/src/include/storage/procsignal.h b/src/include/storage/procsignal.h index 022fd8ed933..fc5b238ec20 100644 --- a/src/include/storage/procsignal.h +++ b/src/include/storage/procsignal.h @@ -54,6 +54,8 @@ typedef enum typedef enum { PROCSIGNAL_BARRIER_SMGRRELEASE, /* ask smgr to close files */ + PROCSIGNAL_BARRIER_UPDATE_WAL_LOGGING_STATE, /* ask to update xlog info + * state */ } ProcSignalBarrierType; /* diff --git a/src/include/utils/guc_hooks.h b/src/include/utils/guc_hooks.h index 87999218d68..22180b2016c 100644 --- a/src/include/utils/guc_hooks.h +++ b/src/include/utils/guc_hooks.h @@ -169,6 +169,7 @@ extern bool check_wal_buffers(int *newval, void **extra, GucSource source); extern bool check_wal_consistency_checking(char **newval, void **extra, GucSource source); extern void assign_wal_consistency_checking(const char *newval, void *extra); +extern const char *show_wal_level(void); extern bool check_wal_segment_size(int *newval, void **extra, GucSource source); extern void assign_wal_sync_method(int new_wal_sync_method, void *extra); extern bool check_synchronized_standby_slots(char **newval, void **extra, -- 2.43.5