logicalctl.c File Reference

#include "postgres.h"
#include "access/xloginsert.h"
#include "catalog/pg_control.h"
#include "miscadmin.h"
#include "replication/slot.h"
#include "storage/ipc.h"
#include "storage/lmgr.h"
#include "storage/proc.h"
#include "storage/procarray.h"
#include "storage/procsignal.h"
#include "storage/subsystems.h"
#include "utils/injection_point.h"

Include dependency graph for logicalctl.c:

Go to the source code of this file.

Data Structures
struct	LogicalDecodingCtlData

Typedefs
typedef struct LogicalDecodingCtlData	LogicalDecodingCtlData

Functions
static void	LogicalDecodingCtlShmemRequest (void *arg)
static void	update_xlog_logical_info (void)
static void	abort_logical_decoding_activation (int code, Datum arg)
static void	write_logical_decoding_status_update_record (bool status)
void	StartupLogicalDecodingStatus (bool last_status)
void	InitializeProcessXLogLogicalInfo (void)
bool	ProcessBarrierUpdateXLogLogicalInfo (void)
bool	IsLogicalDecodingEnabled (void)
bool	IsXLogLogicalInfoEnabled (void)
void	AtEOXact_LogicalCtl (void)
void	EnsureLogicalDecodingEnabled (void)
void	EnableLogicalDecoding (void)
void	RequestDisableLogicalDecoding (void)
void	DisableLogicalDecodingIfNecessary (void)
void	DisableLogicalDecoding (void)
void	UpdateLogicalDecodingStatusEndOfRecovery (void)

Variables
static LogicalDecodingCtlData *	LogicalDecodingCtl = NULL
const ShmemCallbacks	LogicalDecodingCtlShmemCallbacks
bool	XLogLogicalInfo = false
static bool	XLogLogicalInfoUpdatePending = false

Typedef Documentation

LogicalDecodingCtlData

typedef struct LogicalDecodingCtlData LogicalDecodingCtlData

Function Documentation

abort_logical_decoding_activation()

static void abort_logical_decoding_activation ( int  code, Datum  arg  )

static

Definition at line 263 of file logicalctl.c.

{
    Assert(MyReplicationSlot);
    Assert(!LogicalDecodingCtl->logical_decoding_enabled);
 
    elog(DEBUG1, "aborting logical decoding activation process");
 
    /*
     * Abort the change to xlog_logical_info. We don't need to check
     * CheckLogicalSlotExists() as we're still holding a logical slot.
     */
    LWLockAcquire(LogicalDecodingControlLock, LW_EXCLUSIVE);
    LogicalDecodingCtl->xlog_logical_info = false;
    LWLockRelease(LogicalDecodingControlLock);
 
    /*
     * Some processes might have already started logical info WAL logging, so
     * tell all running processes to update their caches. We don't need to
     * wait for all processes to disable xlog_logical_info locally as it's
     * always safe to write logical information to WAL records, even when not
     * strictly required.
     */
    EmitProcSignalBarrier(PROCSIGNAL_BARRIER_UPDATE_XLOG_LOGICAL_INFO);
}

References Assert, DEBUG1, elog, EmitProcSignalBarrier(), fb(), LogicalDecodingCtlData::logical_decoding_enabled, LogicalDecodingCtl, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MyReplicationSlot, PROCSIGNAL_BARRIER_UPDATE_XLOG_LOGICAL_INFO, and LogicalDecodingCtlData::xlog_logical_info.

Referenced by EnsureLogicalDecodingEnabled().

AtEOXact_LogicalCtl()

void AtEOXact_LogicalCtl

(

void

)

Definition at line 233 of file logicalctl.c.

{
    /* Update the local cache if there is a pending update */
    if (XLogLogicalInfoUpdatePending)
    {
        update_xlog_logical_info();
        XLogLogicalInfoUpdatePending = false;
    }
}

References update_xlog_logical_info(), and XLogLogicalInfoUpdatePending.

Referenced by AbortTransaction(), CommitTransaction(), and PrepareTransaction().

DisableLogicalDecoding()

void DisableLogicalDecoding

(

void

)

Definition at line 489 of file logicalctl.c.

{
    bool        in_recovery = RecoveryInProgress();
 
    LWLockAcquire(LogicalDecodingControlLock, LW_EXCLUSIVE);
 
    /*
     * Check if we can disable logical decoding.
     *
     * Skip CheckLogicalSlotExists() check during recovery because the
     * existing slots will be invalidated after disabling logical decoding.
     */
    if (!LogicalDecodingCtl->logical_decoding_enabled ||
        (!in_recovery && CheckLogicalSlotExists()))
    {
        LogicalDecodingCtl->pending_disable = false;
        LWLockRelease(LogicalDecodingControlLock);
        return;
    }
 
    START_CRIT_SECTION();
 
    /*
     * We need to disable logical decoding first and then disable logical
     * information WAL logging in order to ensure that no logical decoding
     * processes WAL records with insufficient information.
     */
    LogicalDecodingCtl->logical_decoding_enabled = false;
 
    /* Write the WAL to disable logical decoding on standbys too */
    if (!in_recovery)
        write_logical_decoding_status_update_record(false);
 
    /* Now disable logical information WAL logging */
    LogicalDecodingCtl->xlog_logical_info = false;
    LogicalDecodingCtl->pending_disable = false;
 
    END_CRIT_SECTION();
 
    if (!in_recovery)
        ereport(LOG,
                errmsg("logical decoding is disabled because there are no valid logical replication slots"));
 
    LWLockRelease(LogicalDecodingControlLock);
 
    /*
     * Tell all running processes to reflect the xlog_logical_info update.
     * Unlike when enabling logical decoding, we don't need to wait for all
     * processes to complete it in this case. We already disabled logical
     * decoding and it's always safe to write logical information to WAL
     * records, even when not strictly required. Therefore, we don't need to
     * wait for all running transactions to finish either.
     */
    EmitProcSignalBarrier(PROCSIGNAL_BARRIER_UPDATE_XLOG_LOGICAL_INFO);
}

References CheckLogicalSlotExists(), EmitProcSignalBarrier(), END_CRIT_SECTION, ereport, errmsg, fb(), LOG, LogicalDecodingCtlData::logical_decoding_enabled, LogicalDecodingCtl, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), LogicalDecodingCtlData::pending_disable, PROCSIGNAL_BARRIER_UPDATE_XLOG_LOGICAL_INFO, RecoveryInProgress(), START_CRIT_SECTION, write_logical_decoding_status_update_record(), and LogicalDecodingCtlData::xlog_logical_info.

Referenced by DisableLogicalDecodingIfNecessary(), and xlog_redo().

DisableLogicalDecodingIfNecessary()

void DisableLogicalDecodingIfNecessary

(

void

)

Definition at line 458 of file logicalctl.c.

{
    bool        pending_disable;
 
    if (wal_level != WAL_LEVEL_REPLICA)
        return;
 
    /*
     * Sanity check as we cannot disable logical decoding while holding a
     * logical slot.
     */
    Assert(!MyReplicationSlot);
 
    if (RecoveryInProgress())
        return;
 
    LWLockAcquire(LogicalDecodingControlLock, LW_SHARED);
    pending_disable = LogicalDecodingCtl->pending_disable;
    LWLockRelease(LogicalDecodingControlLock);
 
    /* Quick return if no pending disable request */
    if (!pending_disable)
        return;
 
    DisableLogicalDecoding();
}

References Assert, DisableLogicalDecoding(), fb(), LogicalDecodingCtl, LW_SHARED, LWLockAcquire(), LWLockRelease(), MyReplicationSlot, LogicalDecodingCtlData::pending_disable, RecoveryInProgress(), wal_level, and WAL_LEVEL_REPLICA.

Referenced by CheckpointerMain().

EnableLogicalDecoding()

void EnableLogicalDecoding

(

void

)

Definition at line 338 of file logicalctl.c.

{
    bool        in_recovery;
 
    LWLockAcquire(LogicalDecodingControlLock, LW_EXCLUSIVE);
 
    /* Return if it is already enabled */
    if (LogicalDecodingCtl->logical_decoding_enabled)
    {
        LogicalDecodingCtl->pending_disable = false;
        LWLockRelease(LogicalDecodingControlLock);
        return;
    }
 
    /*
     * Set logical info WAL logging in shmem. All process starts after this
     * point will include the information required by logical decoding to WAL
     * records.
     */
    LogicalDecodingCtl->xlog_logical_info = true;
 
    LWLockRelease(LogicalDecodingControlLock);
 
    /*
     * Tell all running processes to reflect the xlog_logical_info update, and
     * wait. This ensures that all running processes have enabled logical
     * information WAL logging.
     */
    WaitForProcSignalBarrier(
                             EmitProcSignalBarrier(PROCSIGNAL_BARRIER_UPDATE_XLOG_LOGICAL_INFO));
 
    INJECTION_POINT("logical-decoding-activation", NULL);
 
    in_recovery = RecoveryInProgress();
 
    /*
     * There could be some transactions that might have started with the old
     * status, but we don't need to wait for these transactions to complete as
     * long as they have valid XIDs. These transactions will appear in the
     * xl_running_xacts record and therefore the snapshot builder will not try
     * to decode the transaction during the logical decoding initialization.
     *
     * There is a theoretical case where a transaction decides whether to
     * include logical-info to WAL records before getting an XID. In this
     * case, the transaction won't appear in xl_running_xacts.
     *
     * For operations that do not require an XID assignment, the process
     * starts including logical-info immediately upon receiving the signal
     * (barrier). If such an operation checks the effective_wal_level multiple
     * times within a single execution, the resulting WAL records might be
     * inconsistent (i.e., logical-info is included in some records but not in
     * others). However, this is harmless because logical decoding generally
     * ignores WAL records that are not associated with an assigned XID.
     *
     * One might think we need to wait for all running transactions, including
     * those without XIDs and read-only transactions, to finish before
     * enabling logical decoding. However, such a requirement would force the
     * slot creation to wait for a potentially very long time due to
     * long-running read queries, which is practically unacceptable.
     */
 
    START_CRIT_SECTION();
 
    /*
     * We enable logical decoding first, followed by writing the WAL record.
     * This sequence ensures logical decoding becomes available on the primary
     * first.
     */
    LWLockAcquire(LogicalDecodingControlLock, LW_EXCLUSIVE);
 
    LogicalDecodingCtl->logical_decoding_enabled = true;
 
    if (!in_recovery)
        write_logical_decoding_status_update_record(true);
 
    LogicalDecodingCtl->pending_disable = false;
 
    LWLockRelease(LogicalDecodingControlLock);
 
    END_CRIT_SECTION();
 
    if (!in_recovery)
        ereport(LOG,
                errmsg("logical decoding is enabled upon creating a new logical replication slot"));
}

References EmitProcSignalBarrier(), END_CRIT_SECTION, ereport, errmsg, fb(), INJECTION_POINT, LOG, LogicalDecodingCtlData::logical_decoding_enabled, LogicalDecodingCtl, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), LogicalDecodingCtlData::pending_disable, PROCSIGNAL_BARRIER_UPDATE_XLOG_LOGICAL_INFO, RecoveryInProgress(), START_CRIT_SECTION, WaitForProcSignalBarrier(), write_logical_decoding_status_update_record(), and LogicalDecodingCtlData::xlog_logical_info.

Referenced by EnsureLogicalDecodingEnabled(), and xlog_redo().

EnsureLogicalDecodingEnabled()

void EnsureLogicalDecodingEnabled

(

void

)

Definition at line 303 of file logicalctl.c.

{
    Assert(MyReplicationSlot);
    Assert(wal_level >= WAL_LEVEL_REPLICA);
 
    /* Logical decoding is always enabled */
    if (wal_level >= WAL_LEVEL_LOGICAL)
        return;
 
    if (RecoveryInProgress())
    {
        /*
         * CheckLogicalDecodingRequirements() must have already errored out if
         * logical decoding is not enabled since we cannot enable the logical
         * decoding status during recovery.
         */
        Assert(IsLogicalDecodingEnabled());
        return;
    }
 
    /*
     * Ensure to abort the activation process in cases where there in an
     * interruption during the wait.
     */
    PG_ENSURE_ERROR_CLEANUP(abort_logical_decoding_activation, (Datum) 0);
    {
        EnableLogicalDecoding();
    }
    PG_END_ENSURE_ERROR_CLEANUP(abort_logical_decoding_activation, (Datum) 0);
}

References abort_logical_decoding_activation(), Assert, EnableLogicalDecoding(), IsLogicalDecodingEnabled(), MyReplicationSlot, PG_END_ENSURE_ERROR_CLEANUP, PG_ENSURE_ERROR_CLEANUP, RecoveryInProgress(), wal_level, WAL_LEVEL_LOGICAL, and WAL_LEVEL_REPLICA.

Referenced by create_logical_replication_slot(), CreateReplicationSlot(), and repack_setup_logical_decoding().

InitializeProcessXLogLogicalInfo()

void InitializeProcessXLogLogicalInfo

(

void

)

Definition at line 174 of file logicalctl.c.

{
    update_xlog_logical_info();
}

References update_xlog_logical_info().

Referenced by BaseInit().

IsLogicalDecodingEnabled()

bool IsLogicalDecodingEnabled

(

void

)

Definition at line 202 of file logicalctl.c.

{
    bool        enabled;
 
    LWLockAcquire(LogicalDecodingControlLock, LW_SHARED);
    enabled = LogicalDecodingCtl->logical_decoding_enabled;
    LWLockRelease(LogicalDecodingControlLock);
 
    return enabled;
}

References fb(), LogicalDecodingCtlData::logical_decoding_enabled, LogicalDecodingCtl, LW_SHARED, LWLockAcquire(), and LWLockRelease().

Referenced by CheckLogicalDecodingRequirements(), create_logical_replication_slot(), CreateCheckPoint(), CreateReplicationSlot(), EnsureLogicalDecodingEnabled(), LogStandbySnapshot(), ResolveRecoveryConflictWithSnapshot(), UpdateLogicalDecodingStatusEndOfRecovery(), and ValidateSlotSyncParams().

IsXLogLogicalInfoEnabled()

bool IsXLogLogicalInfoEnabled

(

void

)

Definition at line 218 of file logicalctl.c.

{
    bool        xlog_logical_info;
 
    LWLockAcquire(LogicalDecodingControlLock, LW_SHARED);
    xlog_logical_info = LogicalDecodingCtl->xlog_logical_info;
    LWLockRelease(LogicalDecodingControlLock);
 
    return xlog_logical_info;
}

References fb(), LogicalDecodingCtl, LW_SHARED, LWLockAcquire(), LWLockRelease(), and LogicalDecodingCtlData::xlog_logical_info.

Referenced by show_effective_wal_level(), update_xlog_logical_info(), and UpdateLogicalDecodingStatusEndOfRecovery().

LogicalDecodingCtlShmemRequest()

static void LogicalDecodingCtlShmemRequest ( void *  arg )

static

Definition at line 131 of file logicalctl.c.

{
    ShmemRequestStruct(.name = "Logical decoding control",
                       .size = sizeof(LogicalDecodingCtlData),
                       .ptr = (void **) &LogicalDecodingCtl,
        );
}

References LogicalDecodingCtl, name, and ShmemRequestStruct.

ProcessBarrierUpdateXLogLogicalInfo()

bool ProcessBarrierUpdateXLogLogicalInfo

(

void

)

Definition at line 184 of file logicalctl.c.

{
    if (GetTopTransactionIdIfAny() != InvalidTransactionId)
    {
        /* Delay updating XLogLogicalInfo until the transaction end */
        XLogLogicalInfoUpdatePending = true;
    }
    else
        update_xlog_logical_info();
 
    return true;
}

References GetTopTransactionIdIfAny(), InvalidTransactionId, update_xlog_logical_info(), and XLogLogicalInfoUpdatePending.

Referenced by ProcessProcSignalBarrier().

RequestDisableLogicalDecoding()

void RequestDisableLogicalDecoding

(

void

)

Definition at line 431 of file logicalctl.c.

{
    if (wal_level != WAL_LEVEL_REPLICA)
        return;
 
    /*
     * It's possible that we might not actually need to disable logical
     * decoding if someone creates a new logical slot concurrently. We set the
     * flag anyway and the checkpointer will check it and disable logical
     * decoding if necessary.
     */
    LWLockAcquire(LogicalDecodingControlLock, LW_EXCLUSIVE);
    LogicalDecodingCtl->pending_disable = true;
    LWLockRelease(LogicalDecodingControlLock);
 
    WakeupCheckpointer();
 
    elog(DEBUG1, "requested disabling logical decoding");
}

References DEBUG1, elog, fb(), LogicalDecodingCtl, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), LogicalDecodingCtlData::pending_disable, WakeupCheckpointer(), wal_level, and WAL_LEVEL_REPLICA.

Referenced by InvalidateObsoleteReplicationSlots(), ReplicationSlotCleanup(), ReplicationSlotDrop(), ReplicationSlotRelease(), and ReplicationSlotsDropDBSlots().

StartupLogicalDecodingStatus()

void StartupLogicalDecodingStatus

(

bool

last_status

)

Definition at line 144 of file logicalctl.c.

{
    /* Logical decoding is always disabled when 'minimal' WAL level */
    if (wal_level == WAL_LEVEL_MINIMAL)
        return;
 
    /*
     * Set the initial logical decoding status based on the last status. If
     * logical decoding was enabled before the last shutdown, it remains
     * enabled as we might have set wal_level='logical' or have at least one
     * logical slot.
     */
    LogicalDecodingCtl->xlog_logical_info = last_status;
    LogicalDecodingCtl->logical_decoding_enabled = last_status;
}

References fb(), LogicalDecodingCtlData::logical_decoding_enabled, LogicalDecodingCtl, wal_level, WAL_LEVEL_MINIMAL, and LogicalDecodingCtlData::xlog_logical_info.

Referenced by StartupXLOG().

update_xlog_logical_info()

static void update_xlog_logical_info ( void  )

inlinestatic

Definition at line 164 of file logicalctl.c.

{
    XLogLogicalInfo = IsXLogLogicalInfoEnabled();
}

References IsXLogLogicalInfoEnabled(), and XLogLogicalInfo.

Referenced by AtEOXact_LogicalCtl(), InitializeProcessXLogLogicalInfo(), and ProcessBarrierUpdateXLogLogicalInfo().

UpdateLogicalDecodingStatusEndOfRecovery()

void UpdateLogicalDecodingStatusEndOfRecovery

(

void

)

Definition at line 551 of file logicalctl.c.

{
    bool        new_status = false;
 
    Assert(RecoveryInProgress());
 
    /*
     * With 'minimal' WAL level, there are no logical replication slots during
     * recovery. Logical decoding is always disabled, so there is no need to
     * synchronize XLogLogicalInfo.
     */
    if (wal_level == WAL_LEVEL_MINIMAL)
    {
        Assert(!IsXLogLogicalInfoEnabled() && !IsLogicalDecodingEnabled());
        return;
    }
 
    LWLockAcquire(LogicalDecodingControlLock, LW_EXCLUSIVE);
 
    if (wal_level == WAL_LEVEL_LOGICAL || CheckLogicalSlotExists())
        new_status = true;
 
    /*
     * When recovery ends, we need to either enable or disable logical
     * decoding based on the wal_level setting and the presence of logical
     * slots. We need to note that concurrent slot creation and deletion could
     * happen but WAL writes are still not permitted until recovery fully
     * completes. Here's how we handle concurrent toggling of logical
     * decoding:
     *
     * For 'enable' case, if there's a concurrent disable request before
     * recovery fully completes, the checkpointer will handle it after
     * recovery is done. This means there might be a brief period after
     * recovery where logical decoding remains enabled even with no logical
     * replication slots present. This temporary state is not new - it can
     * already occur due to the checkpointer's asynchronous deactivation
     * process.
     *
     * For 'disable' case, backend cannot create logical replication slots
     * during recovery (see checks in CheckLogicalDecodingRequirements()),
     * which prevents a race condition between disabling logical decoding and
     * concurrent slot creation.
     */
    if (new_status != LogicalDecodingCtl->logical_decoding_enabled)
    {
        /*
         * Update both the logical decoding status and logical WAL logging
         * status. Unlike toggling these status during non-recovery, we don't
         * need to worry about the operation order as WAL writes are still not
         * permitted.
         */
        LogicalDecodingCtl->xlog_logical_info = new_status;
        LogicalDecodingCtl->logical_decoding_enabled = new_status;
 
        elog(DEBUG1,
             "update logical decoding status to %d at the end of recovery",
             new_status);
 
        /*
         * Now that we updated the logical decoding status, clear the pending
         * disable flag. It's possible that a concurrent process drops the
         * last logical slot and initiates the pending disable again. The
         * checkpointer process will check it.
         */
        LogicalDecodingCtl->pending_disable = false;
 
        LWLockRelease(LogicalDecodingControlLock);
 
        write_logical_decoding_status_update_record(new_status);
    }
    else
        LWLockRelease(LogicalDecodingControlLock);
 
    /*
     * Ensure all running processes have the updated status. We don't need to
     * wait for running transactions to finish as we don't accept any writes
     * yet. On the other hand, we need to wait for synchronizing
     * XLogLogicalInfo even if we've not updated the status above as the
     * status have been turned on and off during recovery, having running
     * processes have different status on their local caches.
     */
    if (IsUnderPostmaster)
        WaitForProcSignalBarrier(
                                 EmitProcSignalBarrier(PROCSIGNAL_BARRIER_UPDATE_XLOG_LOGICAL_INFO));
 
    INJECTION_POINT("startup-logical-decoding-status-change-end-of-recovery", NULL);
}

References Assert, CheckLogicalSlotExists(), DEBUG1, elog, EmitProcSignalBarrier(), fb(), INJECTION_POINT, IsLogicalDecodingEnabled(), IsUnderPostmaster, IsXLogLogicalInfoEnabled(), LogicalDecodingCtlData::logical_decoding_enabled, LogicalDecodingCtl, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), LogicalDecodingCtlData::pending_disable, PROCSIGNAL_BARRIER_UPDATE_XLOG_LOGICAL_INFO, RecoveryInProgress(), WaitForProcSignalBarrier(), wal_level, WAL_LEVEL_LOGICAL, WAL_LEVEL_MINIMAL, write_logical_decoding_status_update_record(), and LogicalDecodingCtlData::xlog_logical_info.

Referenced by StartupXLOG().

write_logical_decoding_status_update_record()

static void write_logical_decoding_status_update_record ( bool  status )

static

Definition at line 248 of file logicalctl.c.

{
    XLogRecPtr  recptr;
 
    XLogBeginInsert();
    XLogRegisterData(&status, sizeof(bool));
    recptr = XLogInsert(RM_XLOG_ID, XLOG_LOGICAL_DECODING_STATUS_CHANGE);
    XLogFlush(recptr);
}

References fb(), XLOG_LOGICAL_DECODING_STATUS_CHANGE, XLogBeginInsert(), XLogFlush(), XLogInsert(), and XLogRegisterData().

Referenced by DisableLogicalDecoding(), EnableLogicalDecoding(), and UpdateLogicalDecodingStatusEndOfRecovery().

Variable Documentation

LogicalDecodingCtl

LogicalDecodingCtlData* LogicalDecodingCtl = NULL

static

Definition at line 100 of file logicalctl.c.

Referenced by abort_logical_decoding_activation(), DisableLogicalDecoding(), DisableLogicalDecodingIfNecessary(), EnableLogicalDecoding(), IsLogicalDecodingEnabled(), IsXLogLogicalInfoEnabled(), LogicalDecodingCtlShmemRequest(), RequestDisableLogicalDecoding(), StartupLogicalDecodingStatus(), and UpdateLogicalDecodingStatusEndOfRecovery().

LogicalDecodingCtlShmemCallbacks

const ShmemCallbacks LogicalDecodingCtlShmemCallbacks

Initial value:

= {
    .request_fn = LogicalDecodingCtlShmemRequest,
}

Definition at line 104 of file logicalctl.c.

                                                        {
    .request_fn = LogicalDecodingCtlShmemRequest,
};

XLogLogicalInfo

bool XLogLogicalInfo = false

Definition at line 115 of file logicalctl.c.

Referenced by update_xlog_logical_info().

XLogLogicalInfoUpdatePending

bool XLogLogicalInfoUpdatePending = false

static

Definition at line 124 of file logicalctl.c.

Referenced by AtEOXact_LogicalCtl(), and ProcessBarrierUpdateXLogLogicalInfo().