walsender.h File Reference

#include "access/xlogdefs.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)

Enumerations
enum	CRSSnapshotAction { CRS_EXPORT_SNAPSHOT , CRS_NOEXPORT_SNAPSHOT , CRS_USE_SNAPSHOT }

Functions
void	InitWalSender (void)
bool	exec_replication_command (const char *cmd_string)
void	WalSndErrorCleanup (void)
void	PhysicalWakeupLogicalWalSnd (void)
XLogRecPtr	GetStandbyFlushRecPtr (TimeLineID *tli)
void	WalSndSignals (void)
Size	WalSndShmemSize (void)
void	WalSndShmemInit (void)
void	WalSndWakeup (bool physical, bool logical)
void	WalSndInitStopping (void)
void	WalSndWaitStopping (void)
void	HandleWalSndInitStopping (void)
void	WalSndRqstFileReload (void)
static void	WalSndWakeupProcessRequests (bool physical, bool logical)

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

Macro Definition Documentation

WalSndWakeupRequest

#define WalSndWakeupRequest

(

)

do { wake_wal_senders = true; } while (0)

Definition at line 58 of file walsender.h.

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 *

cmd_string

)

Definition at line 1970 of file walsender.c.

{
    yyscan_t    scanner;
    int         parse_rc;
    Node       *cmd_node;
    const char *cmdtag;
    MemoryContext old_context = CurrentMemoryContext;
 
    /* We save and re-use the cmd_context across calls */
    static MemoryContext cmd_context = NULL;
 
    /*
     * 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,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 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();
 
    /*
     * Prepare to parse and execute the command.
     *
     * Because replication command execution can involve beginning or ending
     * transactions, we need a working context that will survive that, so we
     * make it a child of TopMemoryContext.  That in turn creates a hazard of
     * long-lived memory leaks if we lose track of the working context.  We
     * deal with that by creating it only once per walsender, and resetting it
     * for each new command.  (Normally this reset is a no-op, but if the
     * prior exec_replication_command call failed with an error, it won't be.)
     *
     * This is subtler than it looks.  The transactions we manage can extend
     * across replication commands, indeed SnapBuildClearExportedSnapshot
     * might have just ended one.  Because transaction exit will revert to the
     * memory context that was current at transaction start, we need to be
     * sure that that context is still valid.  That motivates re-using the
     * same cmd_context rather than making a new one each time.
     */
    if (cmd_context == NULL)
        cmd_context = AllocSetContextCreate(TopMemoryContext,
                                            "Replication command context",
                                            ALLOCSET_DEFAULT_SIZES);
    else
        MemoryContextReset(cmd_context);
 
    MemoryContextSwitchTo(cmd_context);
 
    replication_scanner_init(cmd_string, &scanner);
 
    /*
     * Is it a WalSender command?
     */
    if (!replication_scanner_is_replication_command(scanner))
    {
        /* Nope; clean up and get out. */
        replication_scanner_finish(scanner);
 
        MemoryContextSwitchTo(old_context);
        MemoryContextReset(cmd_context);
 
        /* XXX this is a pretty random place to make this check */
        if (MyDatabaseId == InvalidOid)
            ereport(ERROR,
                    (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                     errmsg("cannot execute SQL commands in WAL sender for physical replication")));
 
        /* Tell the caller that this wasn't a WalSender command. */
        return false;
    }
 
    /*
     * Looks like a WalSender command, so parse it.
     */
    parse_rc = replication_yyparse(&cmd_node, scanner);
    if (parse_rc != 0)
        ereport(ERROR,
                (errcode(ERRCODE_SYNTAX_ERROR),
                 errmsg_internal("replication command parser returned %d",
                                 parse_rc)));
    replication_scanner_finish(scanner);
 
    /*
     * Report query to various monitoring facilities.  For this purpose, we
     * report replication commands just like SQL commands.
     */
    debug_query_string = cmd_string;
 
    pgstat_report_activity(STATE_RUNNING, cmd_string);
 
    /*
     * Log replication command if log_replication_commands is enabled. Even
     * when it's disabled, log the command with DEBUG1 level for backward
     * compatibility.
     */
    ereport(log_replication_commands ? LOG : DEBUG1,
            (errmsg("received replication command: %s", cmd_string)));
 
    /*
     * Disallow replication commands in aborted transaction blocks.
     */
    if (IsAbortedTransactionBlockState())
        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);
 
    switch (cmd_node->type)
    {
        case T_IdentifySystemCmd:
            cmdtag = "IDENTIFY_SYSTEM";
            set_ps_display(cmdtag);
            IdentifySystem();
            EndReplicationCommand(cmdtag);
            break;
 
        case T_ReadReplicationSlotCmd:
            cmdtag = "READ_REPLICATION_SLOT";
            set_ps_display(cmdtag);
            ReadReplicationSlot((ReadReplicationSlotCmd *) cmd_node);
            EndReplicationCommand(cmdtag);
            break;
 
        case T_BaseBackupCmd:
            cmdtag = "BASE_BACKUP";
            set_ps_display(cmdtag);
            PreventInTransactionBlock(true, cmdtag);
            SendBaseBackup((BaseBackupCmd *) cmd_node, uploaded_manifest);
            EndReplicationCommand(cmdtag);
            break;
 
        case T_CreateReplicationSlotCmd:
            cmdtag = "CREATE_REPLICATION_SLOT";
            set_ps_display(cmdtag);
            CreateReplicationSlot((CreateReplicationSlotCmd *) cmd_node);
            EndReplicationCommand(cmdtag);
            break;
 
        case T_DropReplicationSlotCmd:
            cmdtag = "DROP_REPLICATION_SLOT";
            set_ps_display(cmdtag);
            DropReplicationSlot((DropReplicationSlotCmd *) cmd_node);
            EndReplicationCommand(cmdtag);
            break;
 
        case T_AlterReplicationSlotCmd:
            cmdtag = "ALTER_REPLICATION_SLOT";
            set_ps_display(cmdtag);
            AlterReplicationSlot((AlterReplicationSlotCmd *) cmd_node);
            EndReplicationCommand(cmdtag);
            break;
 
        case T_StartReplicationCmd:
            {
                StartReplicationCmd *cmd = (StartReplicationCmd *) cmd_node;
 
                cmdtag = "START_REPLICATION";
                set_ps_display(cmdtag);
                PreventInTransactionBlock(true, cmdtag);
 
                if (cmd->kind == REPLICATION_KIND_PHYSICAL)
                    StartReplication(cmd);
                else
                    StartLogicalReplication(cmd);
 
                /* dupe, but necessary per libpqrcv_endstreaming */
                EndReplicationCommand(cmdtag);
 
                Assert(xlogreader != NULL);
                break;
            }
 
        case T_TimeLineHistoryCmd:
            cmdtag = "TIMELINE_HISTORY";
            set_ps_display(cmdtag);
            PreventInTransactionBlock(true, cmdtag);
            SendTimeLineHistory((TimeLineHistoryCmd *) cmd_node);
            EndReplicationCommand(cmdtag);
            break;
 
        case T_VariableShowStmt:
            {
                DestReceiver *dest = CreateDestReceiver(DestRemoteSimple);
                VariableShowStmt *n = (VariableShowStmt *) cmd_node;
 
                cmdtag = "SHOW";
                set_ps_display(cmdtag);
 
                /* syscache access needs a transaction environment */
                StartTransactionCommand();
                GetPGVariable(n->name, dest);
                CommitTransactionCommand();
                EndReplicationCommand(cmdtag);
            }
            break;
 
        case T_UploadManifestCmd:
            cmdtag = "UPLOAD_MANIFEST";
            set_ps_display(cmdtag);
            PreventInTransactionBlock(true, cmdtag);
            UploadManifest();
            EndReplicationCommand(cmdtag);
            break;
 
        default:
            elog(ERROR, "unrecognized replication command node tag: %u",
                 cmd_node->type);
    }
 
    /*
     * Done.  Revert to caller's memory context, and clean out the cmd_context
     * to recover memory right away.
     */
    MemoryContextSwitchTo(old_context);
    MemoryContextReset(cmd_context);
 
    /*
     * We need not update ps display or pg_stat_activity, because PostgresMain
     * will reset those to "idle".  But we must reset debug_query_string to
     * ensure it doesn't become a dangling pointer.
     */
    debug_query_string = NULL;
 
    return true;
}

References ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, AlterReplicationSlot(), Assert(), CHECK_FOR_INTERRUPTS, CommitTransactionCommand(), CreateDestReceiver(), CreateReplicationSlot(), CurrentMemoryContext, DEBUG1, debug_query_string, generate_unaccent_rules::dest, DestRemoteSimple, DropReplicationSlot(), elog, EndReplicationCommand(), ereport, errcode(), errmsg(), errmsg_internal(), ERROR, GetPGVariable(), got_STOPPING, IdentifySystem(), initStringInfo(), InvalidOid, IsAbortedTransactionBlockState(), StartReplicationCmd::kind, LOG, log_replication_commands, MemoryContextReset(), MemoryContextSwitchTo(), MyDatabaseId, MyWalSnd, VariableShowStmt::name, output_message, pgstat_report_activity(), PreventInTransactionBlock(), ReadReplicationSlot(), REPLICATION_KIND_PHYSICAL, replication_scanner_finish(), replication_scanner_init(), replication_scanner_is_replication_command(), replication_yyparse(), reply_message, SendBaseBackup(), SendTimeLineHistory(), set_ps_display(), SnapBuildClearExportedSnapshot(), StartLogicalReplication(), StartReplication(), StartTransactionCommand(), WalSnd::state, STATE_RUNNING, tmpbuf, TopMemoryContext, Node::type, ReadReplicationSlotCmd::type, uploaded_manifest, UploadManifest(), WalSndSetState(), WALSNDSTATE_STOPPING, and xlogreader.

Referenced by PostgresMain().

GetStandbyFlushRecPtr()

XLogRecPtr GetStandbyFlushRecPtr

(

TimeLineID *

tli

)

Definition at line 3533 of file walsender.c.

{
    XLogRecPtr  replayPtr;
    TimeLineID  replayTLI;
    XLogRecPtr  receivePtr;
    TimeLineID  receiveTLI;
    XLogRecPtr  result;
 
    Assert(am_cascading_walsender || IsSyncingReplicationSlots());
 
    /*
     * We can safely send what's already been replayed. Also, if walreceiver
     * is streaming WAL from the same timeline, we can send anything that it
     * has streamed, but hasn't been replayed yet.
     */
 
    receivePtr = GetWalRcvFlushRecPtr(NULL, &receiveTLI);
    replayPtr = GetXLogReplayRecPtr(&replayTLI);
 
    if (tli)
        *tli = replayTLI;
 
    result = replayPtr;
    if (receiveTLI == replayTLI && receivePtr > replayPtr)
        result = receivePtr;
 
    return result;
}

References am_cascading_walsender, Assert(), GetWalRcvFlushRecPtr(), GetXLogReplayRecPtr(), IsSyncingReplicationSlots(), and receiveTLI.

Referenced by IdentifySystem(), StartReplication(), synchronize_one_slot(), XLogSendLogical(), and XLogSendPhysical().

HandleWalSndInitStopping()

void HandleWalSndInitStopping

(

void

)

Definition at line 3589 of file walsender.c.

{
    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;
}

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

Referenced by procsignal_sigusr1_handler().

InitWalSender()

void InitWalSender

(

void

)

Definition at line 283 of file walsender.c.

{
    am_cascading_walsender = RecoveryInProgress();
 
    /* Create a per-walsender data structure in shared memory */
    InitWalSenderSlot();
 
    /* need resource owner for e.g. basebackups */
    CreateAuxProcessResourceOwner();
 
    /*
     * Let postmaster know that we're a WAL sender. Once we've declared us as
     * a WAL sender process, postmaster will let us outlive the bgwriter and
     * kill us last in the shutdown sequence, so we get a chance to stream all
     * remaining WAL at shutdown, including the shutdown checkpoint. Note that
     * there's no going back, and we mustn't write any WAL records after this.
     */
    MarkPostmasterChildWalSender();
    SendPostmasterSignal(PMSIGNAL_ADVANCE_STATE_MACHINE);
 
    /*
     * If the client didn't specify a database to connect to, show in PGPROC
     * that our advertised xmin should affect vacuum horizons in all
     * databases.  This allows physical replication clients to send hot
     * standby feedback that will delay vacuum cleanup in all databases.
     */
    if (MyDatabaseId == InvalidOid)
    {
        Assert(MyProc->xmin == InvalidTransactionId);
        LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
        MyProc->statusFlags |= PROC_AFFECTS_ALL_HORIZONS;
        ProcGlobal->statusFlags[MyProc->pgxactoff] = MyProc->statusFlags;
        LWLockRelease(ProcArrayLock);
    }
 
    /* Initialize empty timestamp buffer for lag tracking. */
    lag_tracker = MemoryContextAllocZero(TopMemoryContext, sizeof(LagTracker));
}

References am_cascading_walsender, Assert(), CreateAuxProcessResourceOwner(), InitWalSenderSlot(), InvalidOid, InvalidTransactionId, lag_tracker, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MarkPostmasterChildWalSender(), MemoryContextAllocZero(), MyDatabaseId, MyProc, PGPROC::pgxactoff, PMSIGNAL_ADVANCE_STATE_MACHINE, PROC_AFFECTS_ALL_HORIZONS, ProcGlobal, RecoveryInProgress(), SendPostmasterSignal(), PGPROC::statusFlags, PROC_HDR::statusFlags, TopMemoryContext, and PGPROC::xmin.

Referenced by PostgresMain().

PhysicalWakeupLogicalWalSnd()

void PhysicalWakeupLogicalWalSnd

(

void

)

Definition at line 1714 of file walsender.c.

{
    Assert(MyReplicationSlot && SlotIsPhysical(MyReplicationSlot));
 
    /*
     * If we are running in a standby, there is no need to wake up walsenders.
     * This is because we do not support syncing slots to cascading standbys,
     * so, there are no walsenders waiting for standbys to catch up.
     */
    if (RecoveryInProgress())
        return;
 
    if (SlotExistsInSyncStandbySlots(NameStr(MyReplicationSlot->data.name)))
        ConditionVariableBroadcast(&WalSndCtl->wal_confirm_rcv_cv);
}

References Assert(), ConditionVariableBroadcast(), ReplicationSlot::data, MyReplicationSlot, ReplicationSlotPersistentData::name, NameStr, RecoveryInProgress(), SlotExistsInSyncStandbySlots(), SlotIsPhysical, WalSndCtlData::wal_confirm_rcv_cv, and WalSndCtl.

Referenced by pg_physical_replication_slot_advance(), and PhysicalConfirmReceivedLocation().

WalSndErrorCleanup()

void WalSndErrorCleanup

(

void

)

Definition at line 330 of file walsender.c.

{
    LWLockReleaseAll();
    ConditionVariableCancelSleep();
    pgstat_report_wait_end();
    pgaio_error_cleanup();
 
    if (xlogreader != NULL && xlogreader->seg.ws_file >= 0)
        wal_segment_close(xlogreader);
 
    if (MyReplicationSlot != NULL)
        ReplicationSlotRelease();
 
    ReplicationSlotCleanup(false);
 
    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())
        ReleaseAuxProcessResources(false);
 
    if (got_STOPPING || got_SIGUSR2)
        proc_exit(0);
 
    /* Revert back to startup state */
    WalSndSetState(WALSNDSTATE_STARTUP);
}

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

Referenced by PostgresMain().

WalSndInitStopping()

void WalSndInitStopping

(

void

)

Definition at line 3774 of file walsender.c.

{
    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, INVALID_PROC_NUMBER);
    }
}

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

Referenced by ShutdownXLOG().

WalSndRqstFileReload()

void WalSndRqstFileReload

(

void

)

Definition at line 3566 of file walsender.c.

{
    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);
    }
}

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

Referenced by KeepFileRestoredFromArchive().

WalSndShmemInit()

void WalSndShmemInit

(

void

)

Definition at line 3650 of file walsender.c.

{
    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++)
            dlist_init(&(WalSndCtl->SyncRepQueue[i]));
 
        for (i = 0; i < max_wal_senders; i++)
        {
            WalSnd     *walsnd = &WalSndCtl->walsnds[i];
 
            SpinLockInit(&walsnd->mutex);
        }
 
        ConditionVariableInit(&WalSndCtl->wal_flush_cv);
        ConditionVariableInit(&WalSndCtl->wal_replay_cv);
        ConditionVariableInit(&WalSndCtl->wal_confirm_rcv_cv);
    }
}

References ConditionVariableInit(), dlist_init(), i, max_wal_senders, MemSet, WalSnd::mutex, NUM_SYNC_REP_WAIT_MODE, ShmemInitStruct(), SpinLockInit, WalSndCtlData::SyncRepQueue, WalSndCtlData::wal_confirm_rcv_cv, WalSndCtlData::wal_flush_cv, WalSndCtlData::wal_replay_cv, WalSndCtl, WalSndCtlData::walsnds, and WalSndShmemSize().

Referenced by CreateOrAttachShmemStructs().

WalSndShmemSize()

Size WalSndShmemSize

(

void

)

Definition at line 3638 of file walsender.c.

{
    Size        size = 0;
 
    size = offsetof(WalSndCtlData, walsnds);
    size = add_size(size, mul_size(max_wal_senders, sizeof(WalSnd)));
 
    return size;
}

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

Referenced by CalculateShmemSize(), and WalSndShmemInit().

WalSndSignals()

void WalSndSignals

(

void

)

Definition at line 3619 of file walsender.c.

{
    /* Set up signal handlers */
    pqsignal(SIGHUP, SignalHandlerForConfigReload);
    pqsignal(SIGINT, StatementCancelHandler);   /* query cancel */
    pqsignal(SIGTERM, die);     /* request shutdown */
    /* SIGQUIT handler was already set up by InitPostmasterChild */
    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);
}

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

Referenced by PostgresMain().

WalSndWaitStopping()

void WalSndWaitStopping

(

void

)

Definition at line 3800 of file walsender.c.

{
    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 */
    }
}

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

Referenced by ShutdownXLOG().

WalSndWakeup()

void WalSndWakeup	(	bool	physical,
		bool	logical
	)

Definition at line 3695 of file walsender.c.

{
    /*
     * Wake up all the walsenders waiting on WAL being flushed or replayed
     * respectively.  Note that waiting walsender would have prepared to sleep
     * on the CV (i.e., added itself to the CV's waitlist) in WalSndWait()
     * before actually waiting.
     */
    if (physical)
        ConditionVariableBroadcast(&WalSndCtl->wal_flush_cv);
 
    if (logical)
        ConditionVariableBroadcast(&WalSndCtl->wal_replay_cv);
}

References ConditionVariableBroadcast(), WalSndCtlData::wal_flush_cv, WalSndCtlData::wal_replay_cv, and WalSndCtl.

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

WalSndWakeupProcessRequests()

static void WalSndWakeupProcessRequests ( bool  physical, bool  logical  )

inlinestatic

Definition at line 65 of file walsender.h.

{
    if (wake_wal_senders)
    {
        wake_wal_senders = false;
        if (max_wal_senders > 0)
            WalSndWakeup(physical, logical);
    }
}

References max_wal_senders, wake_wal_senders, and WalSndWakeup().

Referenced by XLogBackgroundFlush(), and XLogFlush().

Variable Documentation

am_cascading_walsender

PGDLLIMPORT bool am_cascading_walsender

extern

Definition at line 121 of file walsender.c.

Referenced by GetStandbyFlushRecPtr(), IdentifySystem(), InitWalSender(), logical_read_xlog_page(), ProcessStandbyReplyMessage(), StartLogicalReplication(), StartReplication(), SyncRepGetStandbyPriority(), XLogSendLogical(), and XLogSendPhysical().

am_db_walsender

PGDLLIMPORT bool am_db_walsender

extern

Definition at line 123 of file walsender.c.

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

am_walsender

PGDLLIMPORT bool am_walsender

extern

Definition at line 120 of file walsender.c.

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

log_replication_commands

PGDLLIMPORT bool log_replication_commands

extern

Definition at line 130 of file walsender.c.

Referenced by exec_replication_command(), ReplicationSlotAcquire(), and ReplicationSlotRelease().

max_wal_senders

PGDLLIMPORT int max_wal_senders

extern

Definition at line 126 of file walsender.c.

Referenced by CheckRequiredParameterValues(), InitControlFile(), InitializeMaxBackends(), InitPostmasterChildSlots(), InitProcess(), InitWalSenderSlot(), pg_stat_get_wal_senders(), PostmasterMain(), SyncRepGetCandidateStandbys(), WalSndInitStopping(), WalSndRqstFileReload(), WalSndShmemInit(), WalSndShmemSize(), WalSndWaitStopping(), WalSndWakeupProcessRequests(), and XLogReportParameters().

wake_wal_senders

PGDLLIMPORT bool wake_wal_senders

extern

Definition at line 135 of file walsender.c.

Referenced by WalSndWakeupProcessRequests().

wal_sender_timeout

PGDLLIMPORT int wal_sender_timeout

extern

Definition at line 128 of file walsender.c.

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