walsender.h File Reference

#include <signal.h>

Include dependency graph for walsender.h:

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

Go to the source code of this file.

Macros
#define	WalSndWakeupRequest()   do { wake_wal_senders = true; } while (0)
#define	WalSndWakeupProcessRequests()

Enumerations
enum	CRSSnapshotAction { CRS_EXPORT_SNAPSHOT, CRS_NOEXPORT_SNAPSHOT, CRS_USE_SNAPSHOT }

Functions
void	InitWalSender (void)
bool	exec_replication_command (const char *query_string)
void	WalSndErrorCleanup (void)
void	WalSndResourceCleanup (bool isCommit)
void	WalSndSignals (void)
Size	WalSndShmemSize (void)
void	WalSndShmemInit (void)
void	WalSndWakeup (void)
void	WalSndInitStopping (void)
void	WalSndWaitStopping (void)
void	HandleWalSndInitStopping (void)
void	WalSndRqstFileReload (void)

Variables
bool	am_walsender
bool	am_cascading_walsender
bool	am_db_walsender
bool	wake_wal_senders
int	max_wal_senders
int	wal_sender_timeout
bool	log_replication_commands

Macro Definition Documentation

WalSndWakeupProcessRequests

#define WalSndWakeupProcessRequests

(

)

Value:

do                                      \
    {                                       \
        if (wake_wal_senders)               \
        {                                   \
            wake_wal_senders = false;       \
            if (max_wal_senders > 0)        \
                WalSndWakeup();             \
        }                                   \
    } while (0)

Definition at line 63 of file walsender.h.

Referenced by XLogBackgroundFlush(), and XLogFlush().

WalSndWakeupRequest

#define WalSndWakeupRequest

(

)

do { wake_wal_senders = true; } while (0)

Definition at line 57 of file walsender.h.

Referenced by XLogWrite().

Enumeration Type Documentation

CRSSnapshotAction

enum CRSSnapshotAction

Enumerator
CRS_EXPORT_SNAPSHOT
CRS_NOEXPORT_SNAPSHOT
CRS_USE_SNAPSHOT

Definition at line 20 of file walsender.h.

{
    CRS_EXPORT_SNAPSHOT,
    CRS_NOEXPORT_SNAPSHOT,
    CRS_USE_SNAPSHOT
} CRSSnapshotAction;

Function Documentation

exec_replication_command()

bool exec_replication_command

(

const char *

query_string

)

Definition at line 1516 of file walsender.c.

References ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, Assert, CHECK_FOR_INTERRUPTS, CommitTransactionCommand(), CreateDestReceiver(), CreateReplicationSlot(), CurrentMemoryContext, DEBUG1, generate_unaccent_rules::dest, DestRemote, DestRemoteSimple, DropReplicationSlot(), elog, EndCommand(), ereport, errcode(), errmsg(), errmsg_internal(), ERROR, GetPGVariable(), got_STOPPING, IdentifySystem(), initStringInfo(), InvalidOid, IsA, IsAbortedTransactionBlockState(), IsTransactionBlock(), StartReplicationCmd::kind, LOG, log_replication_commands, MemoryContextDelete(), MemoryContextSwitchTo(), MyDatabaseId, VariableShowStmt::name, pgstat_report_activity(), PreventInTransactionBlock(), REPLICATION_KIND_PHYSICAL, replication_parse_result, replication_scanner_init(), replication_yyparse(), SendBaseBackup(), SendTimeLineHistory(), SetQueryCompletion(), SnapBuildClearExportedSnapshot(), StartLogicalReplication(), StartReplication(), StartTransactionCommand(), WalSnd::state, STATE_IDLE, STATE_RUNNING, T_BaseBackupCmd, T_CreateReplicationSlotCmd, T_DropReplicationSlotCmd, T_IdentifySystemCmd, T_SQLCmd, T_StartReplicationCmd, T_TimeLineHistoryCmd, T_VariableShowStmt, TimeLineHistoryCmd::type, Node::type, WalSndSetState(), and WALSNDSTATE_STOPPING.

Referenced by PostgresMain().

{
    int         parse_rc;
    Node       *cmd_node;
    MemoryContext cmd_context;
    MemoryContext old_context;
    QueryCompletion qc;

    /*
     * If WAL sender has been told that shutdown is getting close, switch its
     * status accordingly to handle the next replication commands correctly.
     */
    if (got_STOPPING)
        WalSndSetState(WALSNDSTATE_STOPPING);

    /*
     * Throw error if in stopping mode.  We need prevent commands that could
     * generate WAL while the shutdown checkpoint is being written.  To be
     * safe, we just prohibit all new commands.
     */
    if (MyWalSnd->state == WALSNDSTATE_STOPPING)
        ereport(ERROR,
                (errmsg("cannot execute new commands while WAL sender is in stopping mode")));

    /*
     * CREATE_REPLICATION_SLOT ... LOGICAL exports a snapshot until the next
     * command arrives. Clean up the old stuff if there's anything.
     */
    SnapBuildClearExportedSnapshot();

    CHECK_FOR_INTERRUPTS();

    cmd_context = AllocSetContextCreate(CurrentMemoryContext,
                                        "Replication command context",
                                        ALLOCSET_DEFAULT_SIZES);
    old_context = MemoryContextSwitchTo(cmd_context);

    replication_scanner_init(cmd_string);
    parse_rc = replication_yyparse();
    if (parse_rc != 0)
        ereport(ERROR,
                (errcode(ERRCODE_SYNTAX_ERROR),
                 errmsg_internal("replication command parser returned %d",
                                 parse_rc)));

    cmd_node = replication_parse_result;

    /*
     * Log replication command if log_replication_commands is enabled. Even
     * when it's disabled, log the command with DEBUG1 level for backward
     * compatibility. Note that SQL commands are not logged here, and will be
     * logged later if log_statement is enabled.
     */
    if (cmd_node->type != T_SQLCmd)
        ereport(log_replication_commands ? LOG : DEBUG1,
                (errmsg("received replication command: %s", cmd_string)));

    /*
     * CREATE_REPLICATION_SLOT ... LOGICAL exports a snapshot. If it was
     * called outside of transaction the snapshot should be cleared here.
     */
    if (!IsTransactionBlock())
        SnapBuildClearExportedSnapshot();

    /*
     * For aborted transactions, don't allow anything except pure SQL, the
     * exec_simple_query() will handle it correctly.
     */
    if (IsAbortedTransactionBlockState() && !IsA(cmd_node, SQLCmd))
        ereport(ERROR,
                (errcode(ERRCODE_IN_FAILED_SQL_TRANSACTION),
                 errmsg("current transaction is aborted, "
                        "commands ignored until end of transaction block")));

    CHECK_FOR_INTERRUPTS();

    /*
     * Allocate buffers that will be used for each outgoing and incoming
     * message.  We do this just once per command to reduce palloc overhead.
     */
    initStringInfo(&output_message);
    initStringInfo(&reply_message);
    initStringInfo(&tmpbuf);

    /* Report to pgstat that this process is running */
    pgstat_report_activity(STATE_RUNNING, NULL);

    switch (cmd_node->type)
    {
        case T_IdentifySystemCmd:
            IdentifySystem();
            break;

        case T_BaseBackupCmd:
            PreventInTransactionBlock(true, "BASE_BACKUP");
            SendBaseBackup((BaseBackupCmd *) cmd_node);
            break;

        case T_CreateReplicationSlotCmd:
            CreateReplicationSlot((CreateReplicationSlotCmd *) cmd_node);
            break;

        case T_DropReplicationSlotCmd:
            DropReplicationSlot((DropReplicationSlotCmd *) cmd_node);
            break;

        case T_StartReplicationCmd:
            {
                StartReplicationCmd *cmd = (StartReplicationCmd *) cmd_node;

                PreventInTransactionBlock(true, "START_REPLICATION");

                if (cmd->kind == REPLICATION_KIND_PHYSICAL)
                    StartReplication(cmd);
                else
                    StartLogicalReplication(cmd);

                Assert(xlogreader != NULL);
                break;
            }

        case T_TimeLineHistoryCmd:
            PreventInTransactionBlock(true, "TIMELINE_HISTORY");
            SendTimeLineHistory((TimeLineHistoryCmd *) cmd_node);
            break;

        case T_VariableShowStmt:
            {
                DestReceiver *dest = CreateDestReceiver(DestRemoteSimple);
                VariableShowStmt *n = (VariableShowStmt *) cmd_node;

                /* syscache access needs a transaction environment */
                StartTransactionCommand();
                GetPGVariable(n->name, dest);
                CommitTransactionCommand();
            }
            break;

        case T_SQLCmd:
            if (MyDatabaseId == InvalidOid)
                ereport(ERROR,
                        (errmsg("cannot execute SQL commands in WAL sender for physical replication")));

            /* Report to pgstat that this process is now idle */
            pgstat_report_activity(STATE_IDLE, NULL);

            /* Tell the caller that this wasn't a WalSender command. */
            return false;

        default:
            elog(ERROR, "unrecognized replication command node tag: %u",
                 cmd_node->type);
    }

    /* done */
    MemoryContextSwitchTo(old_context);
    MemoryContextDelete(cmd_context);

    /* Send CommandComplete message */
    SetQueryCompletion(&qc, CMDTAG_SELECT, 0);
    EndCommand(&qc, DestRemote, true);

    /* Report to pgstat that this process is now idle */
    pgstat_report_activity(STATE_IDLE, NULL);

    return true;
}

HandleWalSndInitStopping()

void HandleWalSndInitStopping

(

void

)

Definition at line 2998 of file walsender.c.

References am_walsender, Assert, got_STOPPING, kill, MyProcPid, and replication_active.

Referenced by procsignal_sigusr1_handler().

{
    Assert(am_walsender);

    /*
     * If replication has not yet started, die like with SIGTERM. If
     * replication is active, only set a flag and wake up the main loop. It
     * will send any outstanding WAL, wait for it to be replicated to the
     * standby, and then exit gracefully.
     */
    if (!replication_active)
        kill(MyProcPid, SIGTERM);
    else
        got_STOPPING = true;
}

InitWalSender()

void InitWalSender

(

void

)

Referenced by PostgresMain().

WalSndErrorCleanup()

void WalSndErrorCleanup

(

void

)

Definition at line 295 of file walsender.c.

References ConditionVariableCancelSleep(), got_SIGUSR2, got_STOPPING, IsTransactionOrTransactionBlock(), LWLockReleaseAll(), MyReplicationSlot, pgstat_report_wait_end(), proc_exit(), replication_active, ReplicationSlotCleanup(), ReplicationSlotRelease(), XLogReaderState::seg, wal_segment_close(), WalSndResourceCleanup(), WalSndSetState(), WALSNDSTATE_STARTUP, and WALOpenSegment::ws_file.

Referenced by PostgresMain().

{
    LWLockReleaseAll();
    ConditionVariableCancelSleep();
    pgstat_report_wait_end();

    if (xlogreader != NULL && xlogreader->seg.ws_file >= 0)
        wal_segment_close(xlogreader);

    if (MyReplicationSlot != NULL)
        ReplicationSlotRelease();

    ReplicationSlotCleanup();

    replication_active = false;

    /*
     * If there is a transaction in progress, it will clean up our
     * ResourceOwner, but if a replication command set up a resource owner
     * without a transaction, we've got to clean that up now.
     */
    if (!IsTransactionOrTransactionBlock())
        WalSndResourceCleanup(false);

    if (got_STOPPING || got_SIGUSR2)
        proc_exit(0);

    /* Revert back to startup state */
    WalSndSetState(WALSNDSTATE_STARTUP);
}

WalSndInitStopping()

void WalSndInitStopping

(

void

)

Definition at line 3124 of file walsender.c.

References i, InvalidBackendId, max_wal_senders, WalSnd::mutex, WalSnd::pid, PROCSIG_WALSND_INIT_STOPPING, SendProcSignal(), SpinLockAcquire, SpinLockRelease, and WalSndCtlData::walsnds.

Referenced by ShutdownXLOG().

{
    int         i;

    for (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;

        SendProcSignal(pid, PROCSIG_WALSND_INIT_STOPPING, InvalidBackendId);
    }
}

WalSndResourceCleanup()

void WalSndResourceCleanup

(

bool

isCommit

)

Definition at line 330 of file walsender.c.

References CurrentResourceOwner, DestNone, DestRemote, proc_exit(), RESOURCE_RELEASE_AFTER_LOCKS, RESOURCE_RELEASE_BEFORE_LOCKS, RESOURCE_RELEASE_LOCKS, ResourceOwnerDelete(), ResourceOwnerRelease(), WalSndShutdown(), and whereToSendOutput.

Referenced by perform_base_backup(), and WalSndErrorCleanup().

{
    ResourceOwner resowner;

    if (CurrentResourceOwner == NULL)
        return;

    /*
     * Deleting CurrentResourceOwner is not allowed, so we must save a pointer
     * in a local variable and clear it first.
     */
    resowner = CurrentResourceOwner;
    CurrentResourceOwner = NULL;

    /* Now we can release resources and delete it. */
    ResourceOwnerRelease(resowner,
                         RESOURCE_RELEASE_BEFORE_LOCKS, isCommit, true);
    ResourceOwnerRelease(resowner,
                         RESOURCE_RELEASE_LOCKS, isCommit, true);
    ResourceOwnerRelease(resowner,
                         RESOURCE_RELEASE_AFTER_LOCKS, isCommit, true);
    ResourceOwnerDelete(resowner);
}

WalSndRqstFileReload()

void WalSndRqstFileReload

(

void

)

Definition at line 2975 of file walsender.c.

References i, max_wal_senders, WalSnd::mutex, WalSnd::needreload, WalSnd::pid, SpinLockAcquire, SpinLockRelease, and WalSndCtlData::walsnds.

Referenced by KeepFileRestoredFromArchive().

{
    int         i;

    for (i = 0; i < max_wal_senders; i++)
    {
        WalSnd     *walsnd = &WalSndCtl->walsnds[i];

        SpinLockAcquire(&walsnd->mutex);
        if (walsnd->pid == 0)
        {
            SpinLockRelease(&walsnd->mutex);
            continue;
        }
        walsnd->needreload = true;
        SpinLockRelease(&walsnd->mutex);
    }
}

WalSndShmemInit()

void WalSndShmemInit

(

void

)

Definition at line 3063 of file walsender.c.

References i, max_wal_senders, MemSet, WalSnd::mutex, NUM_SYNC_REP_WAIT_MODE, ShmemInitStruct(), SHMQueueInit(), SpinLockInit, WalSndCtlData::SyncRepQueue, WalSndCtlData::walsnds, and WalSndShmemSize().

Referenced by CreateSharedMemoryAndSemaphores().

{
    bool        found;
    int         i;

    WalSndCtl = (WalSndCtlData *)
        ShmemInitStruct("Wal Sender Ctl", WalSndShmemSize(), &found);

    if (!found)
    {
        /* First time through, so initialize */
        MemSet(WalSndCtl, 0, WalSndShmemSize());

        for (i = 0; i < NUM_SYNC_REP_WAIT_MODE; i++)
            SHMQueueInit(&(WalSndCtl->SyncRepQueue[i]));

        for (i = 0; i < max_wal_senders; i++)
        {
            WalSnd     *walsnd = &WalSndCtl->walsnds[i];

            SpinLockInit(&walsnd->mutex);
        }
    }
}

WalSndShmemSize()

Size WalSndShmemSize

(

void

)

Definition at line 3051 of file walsender.c.

References add_size(), max_wal_senders, mul_size(), and offsetof.

Referenced by CreateSharedMemoryAndSemaphores(), and WalSndShmemInit().

{
    Size        size = 0;

    size = offsetof(WalSndCtlData, walsnds);
    size = add_size(size, mul_size(max_wal_senders, sizeof(WalSnd)));

    return size;
}

WalSndSignals()

void WalSndSignals

(

void

)

Definition at line 3032 of file walsender.c.

References die, InitializeTimeouts(), pqsignal(), procsignal_sigusr1_handler(), quickdie(), SIG_DFL, SIG_IGN, SIGCHLD, SIGHUP, SignalHandlerForConfigReload(), SIGPIPE, SIGQUIT, SIGUSR1, SIGUSR2, StatementCancelHandler(), and WalSndLastCycleHandler().

Referenced by PostgresMain().

{
    /* Set up signal handlers */
    pqsignal(SIGHUP, SignalHandlerForConfigReload);
    pqsignal(SIGINT, StatementCancelHandler);   /* query cancel */
    pqsignal(SIGTERM, die);     /* request shutdown */
    pqsignal(SIGQUIT, quickdie);    /* hard crash time */
    InitializeTimeouts();       /* establishes SIGALRM handler */
    pqsignal(SIGPIPE, SIG_IGN);
    pqsignal(SIGUSR1, procsignal_sigusr1_handler);
    pqsignal(SIGUSR2, WalSndLastCycleHandler);  /* request a last cycle and
                                                 * shutdown */

    /* Reset some signals that are accepted by postmaster but not here */
    pqsignal(SIGCHLD, SIG_DFL);
}

WalSndWaitStopping()

void WalSndWaitStopping

(

void

)

Definition at line 3150 of file walsender.c.

References i, max_wal_senders, WalSnd::mutex, pg_usleep(), WalSnd::pid, SpinLockAcquire, SpinLockRelease, WalSnd::state, WalSndCtlData::walsnds, and WALSNDSTATE_STOPPING.

Referenced by ShutdownXLOG().

{
    for (;;)
    {
        int         i;
        bool        all_stopped = true;

        for (i = 0; i < max_wal_senders; i++)
        {
            WalSnd     *walsnd = &WalSndCtl->walsnds[i];

            SpinLockAcquire(&walsnd->mutex);

            if (walsnd->pid == 0)
            {
                SpinLockRelease(&walsnd->mutex);
                continue;
            }

            if (walsnd->state != WALSNDSTATE_STOPPING)
            {
                all_stopped = false;
                SpinLockRelease(&walsnd->mutex);
                break;
            }
            SpinLockRelease(&walsnd->mutex);
        }

        /* safe to leave if confirmation is done for all WAL senders */
        if (all_stopped)
            return;

        pg_usleep(10000L);      /* wait for 10 msec */
    }
}

WalSndWakeup()

void WalSndWakeup

(

void

)

Definition at line 3095 of file walsender.c.

References i, WalSnd::latch, max_wal_senders, WalSnd::mutex, SetLatch(), SpinLockAcquire, SpinLockRelease, and WalSndCtlData::walsnds.

Referenced by KeepFileRestoredFromArchive(), StartupXLOG(), and XLogWalRcvFlush().

{
    int         i;

    for (i = 0; i < max_wal_senders; i++)
    {
        Latch      *latch;
        WalSnd     *walsnd = &WalSndCtl->walsnds[i];

        /*
         * Get latch pointer with spinlock held, for the unlikely case that
         * pointer reads aren't atomic (as they're 8 bytes).
         */
        SpinLockAcquire(&walsnd->mutex);
        latch = walsnd->latch;
        SpinLockRelease(&walsnd->mutex);

        if (latch != NULL)
            SetLatch(latch);
    }
}

Variable Documentation

am_cascading_walsender

bool am_cascading_walsender

Definition at line 116 of file walsender.c.

Referenced by IdentifySystem(), ProcessStandbyReplyMessage(), StartLogicalReplication(), StartReplication(), SyncRepGetStandbyPriority(), and XLogSendPhysical().

am_db_walsender

bool am_db_walsender

Definition at line 118 of file walsender.c.

Referenced by check_db(), InitPostgres(), and ProcessStartupPacket().

am_walsender

bool am_walsender

Definition at line 115 of file walsender.c.

Referenced by BackendInitialize(), check_db(), ClientAuthentication(), forbidden_in_wal_sender(), HandleWalSndInitStopping(), InitPostgres(), InitProcess(), LagTrackerWrite(), MarkPostmasterChildWalSender(), PerformAuthentication(), PostgresMain(), ProcessStartupPacket(), and WalSndSetState().

log_replication_commands

bool log_replication_commands

Definition at line 125 of file walsender.c.

Referenced by exec_replication_command().

max_wal_senders

int max_wal_senders

Definition at line 121 of file walsender.c.

Referenced by check_autovacuum_max_workers(), check_max_worker_processes(), check_maxconnections(), CheckRequiredParameterValues(), InitControlFile(), InitializeMaxBackends(), InitProcess(), InitWalSenderSlot(), MaxLivePostmasterChildren(), pg_stat_get_wal_senders(), PostmasterMain(), SyncRepGetCandidateStandbys(), WalSndInitStopping(), WalSndRqstFileReload(), WalSndShmemInit(), WalSndShmemSize(), WalSndWaitStopping(), WalSndWakeup(), and XLogReportParameters().

wake_wal_senders

bool wake_wal_senders

Definition at line 130 of file walsender.c.

wal_sender_timeout

int wal_sender_timeout

Definition at line 123 of file walsender.c.

Referenced by WalSndCheckTimeOut(), WalSndComputeSleeptime(), WalSndKeepaliveIfNecessary(), and WalSndWriteData().