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	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 62 of file walsender.h.

Referenced by XLogBackgroundFlush(), and XLogFlush().

WalSndWakeupRequest

#define WalSndWakeupRequest

(

)

do { wake_wal_senders = true; } while (0)

Definition at line 56 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 1438 of file walsender.c.

References ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, 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(), 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;

    /*
     * 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);
                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 */
    EndCommand("SELECT", DestRemote);

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

    return true;
}

HandleWalSndInitStopping()

void HandleWalSndInitStopping

(

void

)

Definition at line 2985 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 298 of file walsender.c.

References close, ConditionVariableCancelSleep(), DestNone, DestRemote, got_SIGUSR2, got_STOPPING, LWLockReleaseAll(), MyReplicationSlot, pgstat_report_wait_end(), proc_exit(), replication_active, ReplicationSlotCleanup(), ReplicationSlotRelease(), WalSndSetState(), WalSndShutdown(), WALSNDSTATE_STARTUP, whereToSendOutput, and WALOpenSegment::ws_file.

Referenced by PostgresMain().

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

    if (sendSeg->ws_file >= 0)
    {
        close(sendSeg->ws_file);
        sendSeg->ws_file = -1;
    }

    if (MyReplicationSlot != NULL)
        ReplicationSlotRelease();

    ReplicationSlotCleanup();

    replication_active = false;

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

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

WalSndInitStopping()

void WalSndInitStopping

(

void

)

Definition at line 3112 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);
    }
}

WalSndRqstFileReload()

void WalSndRqstFileReload

(

void

)

Definition at line 2962 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 3051 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 3039 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 3019 of file walsender.c.

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

Referenced by PostgresMain().

{
    /* Set up signal handlers */
    pqsignal(SIGHUP, PostgresSigHupHandler);    /* set flag to read config
                                                 * file */
    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 3138 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 3083 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 115 of file walsender.c.

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

am_db_walsender

bool am_db_walsender

Definition at line 117 of file walsender.c.

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

am_walsender

bool am_walsender

Definition at line 114 of file walsender.c.

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

log_replication_commands

bool log_replication_commands

Definition at line 124 of file walsender.c.

Referenced by exec_replication_command().

max_wal_senders

int max_wal_senders

Definition at line 120 of file walsender.c.

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

wake_wal_senders

bool wake_wal_senders

Definition at line 129 of file walsender.c.

wal_sender_timeout

int wal_sender_timeout

Definition at line 122 of file walsender.c.

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