slotsync.h File Reference

#include "replication/walreceiver.h"

Include dependency graph for slotsync.h:

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

Go to the source code of this file.

Functions
char *	CheckAndGetDbnameFromConninfo (void)
bool	ValidateSlotSyncParams (int elevel)
pg_noreturn void	ReplSlotSyncWorkerMain (const void *startup_data, size_t startup_data_len)
void	ShutDownSlotSync (void)
bool	SlotSyncWorkerCanRestart (void)
bool	IsSyncingReplicationSlots (void)
Size	SlotSyncShmemSize (void)
void	SlotSyncShmemInit (void)
void	SyncReplicationSlots (WalReceiverConn *wrconn)

Variables
PGDLLIMPORT bool	sync_replication_slots
PGDLLIMPORT char *	PrimaryConnInfo
PGDLLIMPORT char *	PrimarySlotName

Function Documentation

CheckAndGetDbnameFromConninfo()

char * CheckAndGetDbnameFromConninfo

(

void

)

Definition at line 1176 of file slotsync.c.

{
    char       *dbname;
 
    /*
     * The slot synchronization needs a database connection for walrcv_exec to
     * work.
     */
    dbname = walrcv_get_dbname_from_conninfo(PrimaryConnInfo);
    if (dbname == NULL)
        ereport(ERROR,
                errcode(ERRCODE_INVALID_PARAMETER_VALUE),
 
        /*
         * translator: first %s is a connection option; second %s is a GUC
         * variable name
         */
                errmsg("replication slot synchronization requires \"%s\" to be specified in \"%s\"",
                       "dbname", "primary_conninfo"));
    return dbname;
}

References dbname, ereport, errcode(), errmsg(), ERROR, PrimaryConnInfo, and walrcv_get_dbname_from_conninfo.

Referenced by pg_sync_replication_slots(), and ReplSlotSyncWorkerMain().

IsSyncingReplicationSlots()

bool IsSyncingReplicationSlots

(

void

)

Definition at line 1882 of file slotsync.c.

{
    return syncing_slots;
}

References syncing_slots.

Referenced by CreateDecodingContext(), GetStandbyFlushRecPtr(), and ReplicationSlotCreate().

ReplSlotSyncWorkerMain()

pg_noreturn void ReplSlotSyncWorkerMain	(	const void *	startup_data,
		size_t	startup_data_len
	)

Definition at line 1533 of file slotsync.c.

{
    WalReceiverConn *wrconn = NULL;
    char       *dbname;
    char       *err;
    sigjmp_buf  local_sigjmp_buf;
    StringInfoData app_name;
 
    Assert(startup_data_len == 0);
 
    MyBackendType = B_SLOTSYNC_WORKER;
 
    init_ps_display(NULL);
 
    Assert(GetProcessingMode() == InitProcessing);
 
    /*
     * Create a per-backend PGPROC struct in shared memory.  We must do this
     * before we access any shared memory.
     */
    InitProcess();
 
    /*
     * Early initialization.
     */
    BaseInit();
 
    Assert(SlotSyncCtx != NULL);
 
    /*
     * If an exception is encountered, processing resumes here.
     *
     * We just need to clean up, report the error, and go away.
     *
     * If we do not have this handling here, then since this worker process
     * operates at the bottom of the exception stack, ERRORs turn into FATALs.
     * Therefore, we create our own exception handler to catch ERRORs.
     */
    if (sigsetjmp(local_sigjmp_buf, 1) != 0)
    {
        /* since not using PG_TRY, must reset error stack by hand */
        error_context_stack = NULL;
 
        /* Prevents interrupts while cleaning up */
        HOLD_INTERRUPTS();
 
        /* Report the error to the server log */
        EmitErrorReport();
 
        /*
         * We can now go away.  Note that because we called InitProcess, a
         * callback was registered to do ProcKill, which will clean up
         * necessary state.
         */
        proc_exit(0);
    }
 
    /* We can now handle ereport(ERROR) */
    PG_exception_stack = &local_sigjmp_buf;
 
    /* Setup signal handling */
    pqsignal(SIGHUP, SignalHandlerForConfigReload);
    pqsignal(SIGINT, StatementCancelHandler);
    pqsignal(SIGTERM, die);
    pqsignal(SIGFPE, FloatExceptionHandler);
    pqsignal(SIGUSR1, procsignal_sigusr1_handler);
    pqsignal(SIGUSR2, SIG_IGN);
    pqsignal(SIGPIPE, SIG_IGN);
    pqsignal(SIGCHLD, SIG_DFL);
 
    check_and_set_sync_info(MyProcPid);
 
    ereport(LOG, errmsg("slot sync worker started"));
 
    /* Register it as soon as SlotSyncCtx->pid is initialized. */
    before_shmem_exit(slotsync_worker_onexit, (Datum) 0);
 
    /*
     * Establishes SIGALRM handler and initialize timeout module. It is needed
     * by InitPostgres to register different timeouts.
     */
    InitializeTimeouts();
 
    /* Load the libpq-specific functions */
    load_file("libpqwalreceiver", false);
 
    /*
     * Unblock signals (they were blocked when the postmaster forked us)
     */
    sigprocmask(SIG_SETMASK, &UnBlockSig, NULL);
 
    /*
     * Set always-secure search path, so malicious users can't redirect user
     * code (e.g. operators).
     *
     * It's not strictly necessary since we won't be scanning or writing to
     * any user table locally, but it's good to retain it here for added
     * precaution.
     */
    SetConfigOption("search_path", "", PGC_SUSET, PGC_S_OVERRIDE);
 
    dbname = CheckAndGetDbnameFromConninfo();
 
    /*
     * Connect to the database specified by the user in primary_conninfo. We
     * need a database connection for walrcv_exec to work which we use to
     * fetch slot information from the remote node. See comments atop
     * libpqrcv_exec.
     *
     * We do not specify a specific user here since the slot sync worker will
     * operate as a superuser. This is safe because the slot sync worker does
     * not interact with user tables, eliminating the risk of executing
     * arbitrary code within triggers.
     */
    InitPostgres(dbname, InvalidOid, NULL, InvalidOid, 0, NULL);
 
    SetProcessingMode(NormalProcessing);
 
    initStringInfo(&app_name);
    if (cluster_name[0])
        appendStringInfo(&app_name, "%s_%s", cluster_name, "slotsync worker");
    else
        appendStringInfoString(&app_name, "slotsync worker");
 
    /*
     * Establish the connection to the primary server for slot
     * synchronization.
     */
    wrconn = walrcv_connect(PrimaryConnInfo, false, false, false,
                            app_name.data, &err);
 
    if (!wrconn)
        ereport(ERROR,
                errcode(ERRCODE_CONNECTION_FAILURE),
                errmsg("synchronization worker \"%s\" could not connect to the primary server: %s",
                       app_name.data, err));
 
    pfree(app_name.data);
 
    /*
     * Register the disconnection callback.
     *
     * XXX: This can be combined with previous cleanup registration of
     * slotsync_worker_onexit() but that will need the connection to be made
     * global and we want to avoid introducing global for this purpose.
     */
    before_shmem_exit(slotsync_worker_disconnect, PointerGetDatum(wrconn));
 
    /*
     * Using the specified primary server connection, check that we are not a
     * cascading standby and slot configured in 'primary_slot_name' exists on
     * the primary server.
     */
    validate_remote_info(wrconn);
 
    /* Main loop to synchronize slots */
    for (;;)
    {
        bool        some_slot_updated = false;
        bool        started_tx = false;
        List       *remote_slots;
 
        ProcessSlotSyncInterrupts();
 
        /*
         * The syscache access in fetch_remote_slots() needs a transaction
         * env.
         */
        if (!IsTransactionState())
        {
            StartTransactionCommand();
            started_tx = true;
        }
 
        remote_slots = fetch_remote_slots(wrconn, NIL);
        some_slot_updated = synchronize_slots(wrconn, remote_slots, NULL);
        list_free_deep(remote_slots);
 
        if (started_tx)
            CommitTransactionCommand();
 
        wait_for_slot_activity(some_slot_updated);
    }
 
    /*
     * The slot sync worker can't get here because it will only stop when it
     * receives a stop request from the startup process, or when there is an
     * error.
     */
    Assert(false);
}

References appendStringInfo(), appendStringInfoString(), Assert(), B_SLOTSYNC_WORKER, BaseInit(), before_shmem_exit(), check_and_set_sync_info(), CheckAndGetDbnameFromConninfo(), cluster_name, CommitTransactionCommand(), StringInfoData::data, dbname, die, EmitErrorReport(), ereport, err(), errcode(), errmsg(), ERROR, error_context_stack, fetch_remote_slots(), FloatExceptionHandler(), GetProcessingMode, HOLD_INTERRUPTS, init_ps_display(), InitializeTimeouts(), InitPostgres(), InitProcess(), InitProcessing, initStringInfo(), InvalidOid, IsTransactionState(), list_free_deep(), load_file(), LOG, MyBackendType, MyProcPid, NIL, NormalProcessing, pfree(), PG_exception_stack, PGC_S_OVERRIDE, PGC_SUSET, PointerGetDatum(), pqsignal, PrimaryConnInfo, proc_exit(), ProcessSlotSyncInterrupts(), procsignal_sigusr1_handler(), SetConfigOption(), SetProcessingMode, SIGCHLD, SIGHUP, SignalHandlerForConfigReload(), SIGPIPE, SIGUSR1, SIGUSR2, slotsync_worker_disconnect(), slotsync_worker_onexit(), SlotSyncCtx, StartTransactionCommand(), StatementCancelHandler(), synchronize_slots(), UnBlockSig, validate_remote_info(), wait_for_slot_activity(), walrcv_connect, and wrconn.

ShutDownSlotSync()

void ShutDownSlotSync

(

void

)

Definition at line 1784 of file slotsync.c.

{
    pid_t       sync_process_pid;
 
    SpinLockAcquire(&SlotSyncCtx->mutex);
 
    SlotSyncCtx->stopSignaled = true;
 
    /*
     * Return if neither the slot sync worker is running nor the function
     * pg_sync_replication_slots() is executing.
     */
    if (!SlotSyncCtx->syncing)
    {
        SpinLockRelease(&SlotSyncCtx->mutex);
        update_synced_slots_inactive_since();
        return;
    }
 
    sync_process_pid = SlotSyncCtx->pid;
 
    SpinLockRelease(&SlotSyncCtx->mutex);
 
    /*
     * Signal process doing slotsync, if any. The process will stop upon
     * detecting that the stopSignaled flag is set to true.
     */
    if (sync_process_pid != InvalidPid)
        kill(sync_process_pid, SIGUSR1);
 
    /* Wait for slot sync to end */
    for (;;)
    {
        int         rc;
 
        /* Wait a bit, we don't expect to have to wait long */
        rc = WaitLatch(MyLatch,
                       WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
                       10L, WAIT_EVENT_REPLICATION_SLOTSYNC_SHUTDOWN);
 
        if (rc & WL_LATCH_SET)
        {
            ResetLatch(MyLatch);
            CHECK_FOR_INTERRUPTS();
        }
 
        SpinLockAcquire(&SlotSyncCtx->mutex);
 
        /* Ensure that no process is syncing the slots. */
        if (!SlotSyncCtx->syncing)
            break;
 
        SpinLockRelease(&SlotSyncCtx->mutex);
    }
 
    SpinLockRelease(&SlotSyncCtx->mutex);
 
    update_synced_slots_inactive_since();
}

References CHECK_FOR_INTERRUPTS, InvalidPid, kill, SlotSyncCtxStruct::mutex, MyLatch, SlotSyncCtxStruct::pid, ResetLatch(), SIGUSR1, SlotSyncCtx, SpinLockAcquire, SpinLockRelease, SlotSyncCtxStruct::stopSignaled, SlotSyncCtxStruct::syncing, update_synced_slots_inactive_since(), WaitLatch(), WL_EXIT_ON_PM_DEATH, WL_LATCH_SET, and WL_TIMEOUT.

Referenced by FinishWalRecovery().

SlotSyncShmemInit()

void SlotSyncShmemInit

(

void

)

Definition at line 1900 of file slotsync.c.

{
    Size        size = SlotSyncShmemSize();
    bool        found;
 
    SlotSyncCtx = (SlotSyncCtxStruct *)
        ShmemInitStruct("Slot Sync Data", size, &found);
 
    if (!found)
    {
        memset(SlotSyncCtx, 0, size);
        SlotSyncCtx->pid = InvalidPid;
        SpinLockInit(&SlotSyncCtx->mutex);
    }
}

References InvalidPid, SlotSyncCtxStruct::mutex, SlotSyncCtxStruct::pid, ShmemInitStruct(), SlotSyncCtx, SlotSyncShmemSize(), and SpinLockInit.

Referenced by CreateOrAttachShmemStructs().

SlotSyncShmemSize()

Size SlotSyncShmemSize

(

void

)

Definition at line 1891 of file slotsync.c.

{
    return sizeof(SlotSyncCtxStruct);
}

Referenced by CalculateShmemSize(), and SlotSyncShmemInit().

SlotSyncWorkerCanRestart()

bool SlotSyncWorkerCanRestart

(

void

)

Definition at line 1857 of file slotsync.c.

{
    time_t      curtime = time(NULL);
 
    /*
     * If first time through, or time somehow went backwards, always update
     * last_start_time to match the current clock and allow worker start.
     * Otherwise allow it only once enough time has elapsed.
     */
    if (SlotSyncCtx->last_start_time == 0 ||
        curtime < SlotSyncCtx->last_start_time ||
        curtime - SlotSyncCtx->last_start_time >= SLOTSYNC_RESTART_INTERVAL_SEC)
    {
        SlotSyncCtx->last_start_time = curtime;
        return true;
    }
    return false;
}

References SlotSyncCtxStruct::last_start_time, SLOTSYNC_RESTART_INTERVAL_SEC, and SlotSyncCtx.

Referenced by LaunchMissingBackgroundProcesses().

SyncReplicationSlots()

void SyncReplicationSlots

(

WalReceiverConn *

wrconn

)

Definition at line 1982 of file slotsync.c.

{
    PG_ENSURE_ERROR_CLEANUP(slotsync_failure_callback, PointerGetDatum(wrconn));
    {
        List       *remote_slots = NIL;
        List       *slot_names = NIL;   /* List of slot names to track */
 
        check_and_set_sync_info(MyProcPid);
 
        /* Check for interrupts and config changes */
        ProcessSlotSyncInterrupts();
 
        validate_remote_info(wrconn);
 
        /* Retry until all the slots are sync-ready */
        for (;;)
        {
            bool        slot_persistence_pending = false;
            bool        some_slot_updated = false;
 
            /* Check for interrupts and config changes */
            ProcessSlotSyncInterrupts();
 
            /* We must be in a valid transaction state */
            Assert(IsTransactionState());
 
            /*
             * Fetch remote slot info for the given slot_names. If slot_names
             * is NIL, fetch all failover-enabled slots. Note that we reuse
             * slot_names from the first iteration; re-fetching all failover
             * slots each time could cause an endless loop. Instead of
             * reprocessing only the pending slots in each iteration, it's
             * better to process all the slots received in the first
             * iteration. This ensures that by the time we're done, all slots
             * reflect the latest values.
             */
            remote_slots = fetch_remote_slots(wrconn, slot_names);
 
            /* Attempt to synchronize slots */
            some_slot_updated = synchronize_slots(wrconn, remote_slots,
                                                  &slot_persistence_pending);
 
            /*
             * If slot_persistence_pending is true, extract slot names for
             * future iterations (only needed if we haven't done it yet)
             */
            if (slot_names == NIL && slot_persistence_pending)
                slot_names = extract_slot_names(remote_slots);
 
            /* Free the current remote_slots list */
            list_free_deep(remote_slots);
 
            /* Done if all slots are persisted i.e are sync-ready */
            if (!slot_persistence_pending)
                break;
 
            /* wait before retrying again */
            wait_for_slot_activity(some_slot_updated);
        }
 
        if (slot_names)
            list_free_deep(slot_names);
 
        /* Cleanup the synced temporary slots */
        ReplicationSlotCleanup(true);
 
        /* We are done with sync, so reset sync flag */
        reset_syncing_flag();
    }
    PG_END_ENSURE_ERROR_CLEANUP(slotsync_failure_callback, PointerGetDatum(wrconn));
}

References Assert(), check_and_set_sync_info(), extract_slot_names(), fetch_remote_slots(), IsTransactionState(), list_free_deep(), MyProcPid, NIL, PG_END_ENSURE_ERROR_CLEANUP, PG_ENSURE_ERROR_CLEANUP, PointerGetDatum(), ProcessSlotSyncInterrupts(), ReplicationSlotCleanup(), reset_syncing_flag(), slotsync_failure_callback(), synchronize_slots(), validate_remote_info(), wait_for_slot_activity(), and wrconn.

Referenced by pg_sync_replication_slots().

ValidateSlotSyncParams()

bool ValidateSlotSyncParams

(

int

elevel

)

Definition at line 1203 of file slotsync.c.

{
    /*
     * Logical slot sync/creation requires logical decoding to be enabled.
     */
    if (!IsLogicalDecodingEnabled())
    {
        ereport(elevel,
                errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                errmsg("replication slot synchronization requires \"effective_wal_level\" >= \"logical\" on the primary"),
                errhint("To enable logical decoding on primary, set \"wal_level\" >= \"logical\" or create at least one logical slot when \"wal_level\" = \"replica\"."));
 
        return false;
    }
 
    /*
     * A physical replication slot(primary_slot_name) is required on the
     * primary to ensure that the rows needed by the standby are not removed
     * after restarting, so that the synchronized slot on the standby will not
     * be invalidated.
     */
    if (PrimarySlotName == NULL || *PrimarySlotName == '\0')
    {
        ereport(elevel,
                errcode(ERRCODE_INVALID_PARAMETER_VALUE),
        /* translator: %s is a GUC variable name */
                errmsg("replication slot synchronization requires \"%s\" to be set", "primary_slot_name"));
        return false;
    }
 
    /*
     * hot_standby_feedback must be enabled to cooperate with the physical
     * replication slot, which allows informing the primary about the xmin and
     * catalog_xmin values on the standby.
     */
    if (!hot_standby_feedback)
    {
        ereport(elevel,
                errcode(ERRCODE_INVALID_PARAMETER_VALUE),
        /* translator: %s is a GUC variable name */
                errmsg("replication slot synchronization requires \"%s\" to be enabled",
                       "hot_standby_feedback"));
        return false;
    }
 
    /*
     * The primary_conninfo is required to make connection to primary for
     * getting slots information.
     */
    if (PrimaryConnInfo == NULL || *PrimaryConnInfo == '\0')
    {
        ereport(elevel,
                errcode(ERRCODE_INVALID_PARAMETER_VALUE),
        /* translator: %s is a GUC variable name */
                errmsg("replication slot synchronization requires \"%s\" to be set",
                       "primary_conninfo"));
        return false;
    }
 
    return true;
}

References ereport, errcode(), errhint(), errmsg(), hot_standby_feedback, IsLogicalDecodingEnabled(), PrimaryConnInfo, and PrimarySlotName.

Referenced by LaunchMissingBackgroundProcesses(), and pg_sync_replication_slots().

Variable Documentation

PrimaryConnInfo

PGDLLIMPORT char* PrimaryConnInfo

extern

Definition at line 99 of file xlogrecovery.c.

Referenced by CheckAndGetDbnameFromConninfo(), pg_sync_replication_slots(), ReplSlotSyncWorkerMain(), slotsync_reread_config(), StartupRereadConfig(), validateRecoveryParameters(), ValidateSlotSyncParams(), and WaitForWALToBecomeAvailable().

PrimarySlotName

PGDLLIMPORT char* PrimarySlotName

extern

Definition at line 100 of file xlogrecovery.c.

Referenced by slotsync_reread_config(), StartupRereadConfig(), validate_remote_info(), ValidateSlotSyncParams(), and WaitForWALToBecomeAvailable().

sync_replication_slots

PGDLLIMPORT bool sync_replication_slots

extern

Definition at line 117 of file slotsync.c.

Referenced by LaunchMissingBackgroundProcesses(), PostmasterMain(), slotsync_reread_config(), and xlog_redo().