#include <signal.h>
#include "fmgr.h"

Include dependency graph for walsender.h:

This graph shows which files directly or indirectly include 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

#define WalSndWakeupProcessRequests ( )

Value:

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

Definition at line 64 of file walsender.h.

Referenced by XLogBackgroundFlush(), and XLogFlush().

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

Definition at line 58 of file walsender.h.

Referenced by XLogWrite().

Enumeration Type Documentation

enum CRSSnapshotAction

Enumerator
CRS_EXPORT_SNAPSHOT
CRS_NOEXPORT_SNAPSHOT
CRS_USE_SNAPSHOT

Definition at line 22 of file walsender.h.

 {
     CRS_EXPORT_SNAPSHOT,
     CRS_NOEXPORT_SNAPSHOT,
     CRS_USE_SNAPSHOT
 } CRSSnapshotAction;

Function Documentation

bool exec_replication_command ( const char * query_string )

Definition at line 1416 of file walsender.c.

References ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate(), CHECK_FOR_INTERRUPTS, CreateDestReceiver(), CreateReplicationSlot(), CurrentMemoryContext, DEBUG1, 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, PreventTransactionChain(), REPLICATION_KIND_PHYSICAL, replication_parse_result, replication_scanner_init(), replication_yyparse(), SendBaseBackup(), SendTimeLineHistory(), SnapBuildClearExportedSnapshot(), StartLogicalReplication(), StartReplication(), WalSnd::state, T_BaseBackupCmd, T_CreateReplicationSlotCmd, T_DropReplicationSlotCmd, T_IdentifySystemCmd, T_SQLCmd, T_StartReplicationCmd, T_TimeLineHistoryCmd, T_VariableShowStmt, 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);
 
     switch (cmd_node->type)
     {
         case T_IdentifySystemCmd:
             IdentifySystem();
             break;
 
         case T_BaseBackupCmd:
             PreventTransactionChain(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;
 
                 PreventTransactionChain(true, "START_REPLICATION");
 
                 if (cmd->kind == REPLICATION_KIND_PHYSICAL)
                     StartReplication(cmd);
                 else
                     StartLogicalReplication(cmd);
                 break;
             }
 
         case T_TimeLineHistoryCmd:
             PreventTransactionChain(true, "TIMELINE_HISTORY");
             SendTimeLineHistory((TimeLineHistoryCmd *) cmd_node);
             break;
 
         case T_VariableShowStmt:
             {
                 DestReceiver *dest = CreateDestReceiver(DestRemoteSimple);
                 VariableShowStmt *n = (VariableShowStmt *) cmd_node;
 
                 GetPGVariable(n->name, dest);
             }
             break;
 
         case T_SQLCmd:
             if (MyDatabaseId == InvalidOid)
                 ereport(ERROR,
                         (errmsg("cannot execute SQL commands in WAL sender for physical replication")));
 
             /* 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);
 
     return true;
 }

void HandleWalSndInitStopping ( void )

Definition at line 2902 of file walsender.c.

References am_walsender, Assert, got_STOPPING, 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;
 }

void InitWalSender ( void )

Referenced by PostgresMain().

void WalSndErrorCleanup ( void )

Definition at line 291 of file walsender.c.

References close, ConditionVariableCancelSleep(), got_SIGUSR2, got_STOPPING, LWLockReleaseAll(), MyReplicationSlot, pgstat_report_wait_end(), proc_exit(), replication_active, ReplicationSlotCleanup(), ReplicationSlotRelease(), sendFile, WalSndSetState(), and WALSNDSTATE_STARTUP.

Referenced by PostgresMain().

 {
     LWLockReleaseAll();
     ConditionVariableCancelSleep();
     pgstat_report_wait_end();
 
     if (sendFile >= 0)
     {
         close(sendFile);
         sendFile = -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);
 }

void WalSndInitStopping ( void )

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

void WalSndRqstFileReload ( void )

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

void WalSndShmemInit ( void )

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

Size WalSndShmemSize ( void )

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

void WalSndSignals ( void )

Definition at line 2936 of file walsender.c.

References die(), InitializeTimeouts(), PostgresSigHupHandler(), pqsignal(), procsignal_sigusr1_handler(), quickdie(), SIG_DFL, SIG_IGN, SIGCHLD, SIGCONT, SIGHUP, SIGPIPE, SIGQUIT, SIGTTIN, SIGTTOU, SIGUSR1, SIGUSR2, SIGWINCH, 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);
     pqsignal(SIGTTIN, SIG_DFL);
     pqsignal(SIGTTOU, SIG_DFL);
     pqsignal(SIGCONT, SIG_DFL);
     pqsignal(SIGWINCH, SIG_DFL);
 }

void WalSndWaitStopping ( void )

Definition at line 3059 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 */
     }
 }

void WalSndWakeup ( void )

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