PostgreSQL Source Code git master
All Data Structures Namespaces Files Functions Variables Typedefs Enumerations Enumerator Macros Pages
slotsync.h File Reference
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)
 
void ReplSlotSyncWorkerMain (char *startup_data, size_t startup_data_len) pg_attribute_noreturn()
 
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 1010 of file slotsync.c.

1011{
1012 char *dbname;
1013
1014 /*
1015 * The slot synchronization needs a database connection for walrcv_exec to
1016 * work.
1017 */
1019 if (dbname == NULL)
1020 ereport(ERROR,
1021 errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1022
1023 /*
1024 * translator: first %s is a connection option; second %s is a GUC
1025 * variable name
1026 */
1027 errmsg("replication slot synchronization requires \"%s\" to be specified in \"%s\"",
1028 "dbname", "primary_conninfo"));
1029 return dbname;
1030}
int errcode(int sqlerrcode)
Definition: elog.c:853
int errmsg(const char *fmt,...)
Definition: elog.c:1070
#define ERROR
Definition: elog.h:39
#define ereport(elevel,...)
Definition: elog.h:149
char * dbname
Definition: streamutil.c:50
#define walrcv_get_dbname_from_conninfo(conninfo)
Definition: walreceiver.h:445
char * PrimaryConnInfo
Definition: xlogrecovery.c:96

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 1649 of file slotsync.c.

1650{
1651 return syncing_slots;
1652}
static bool syncing_slots
Definition: slotsync.c:127

References syncing_slots.

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

◆ ReplSlotSyncWorkerMain()

void ReplSlotSyncWorkerMain ( char *  startup_data,
size_t  startup_data_len 
)

Definition at line 1329 of file slotsync.c.

1330{
1331 WalReceiverConn *wrconn = NULL;
1332 char *dbname;
1333 char *err;
1334 sigjmp_buf local_sigjmp_buf;
1335 StringInfoData app_name;
1336
1337 Assert(startup_data_len == 0);
1338
1340
1341 init_ps_display(NULL);
1342
1344
1345 /*
1346 * Create a per-backend PGPROC struct in shared memory. We must do this
1347 * before we access any shared memory.
1348 */
1349 InitProcess();
1350
1351 /*
1352 * Early initialization.
1353 */
1354 BaseInit();
1355
1356 Assert(SlotSyncCtx != NULL);
1357
1358 /*
1359 * If an exception is encountered, processing resumes here.
1360 *
1361 * We just need to clean up, report the error, and go away.
1362 *
1363 * If we do not have this handling here, then since this worker process
1364 * operates at the bottom of the exception stack, ERRORs turn into FATALs.
1365 * Therefore, we create our own exception handler to catch ERRORs.
1366 */
1367 if (sigsetjmp(local_sigjmp_buf, 1) != 0)
1368 {
1369 /* since not using PG_TRY, must reset error stack by hand */
1370 error_context_stack = NULL;
1371
1372 /* Prevents interrupts while cleaning up */
1374
1375 /* Report the error to the server log */
1377
1378 /*
1379 * We can now go away. Note that because we called InitProcess, a
1380 * callback was registered to do ProcKill, which will clean up
1381 * necessary state.
1382 */
1383 proc_exit(0);
1384 }
1385
1386 /* We can now handle ereport(ERROR) */
1387 PG_exception_stack = &local_sigjmp_buf;
1388
1389 /* Setup signal handling */
1392 pqsignal(SIGTERM, die);
1398
1400
1401 ereport(LOG, errmsg("slot sync worker started"));
1402
1403 /* Register it as soon as SlotSyncCtx->pid is initialized. */
1405
1406 /*
1407 * Establishes SIGALRM handler and initialize timeout module. It is needed
1408 * by InitPostgres to register different timeouts.
1409 */
1411
1412 /* Load the libpq-specific functions */
1413 load_file("libpqwalreceiver", false);
1414
1415 /*
1416 * Unblock signals (they were blocked when the postmaster forked us)
1417 */
1418 sigprocmask(SIG_SETMASK, &UnBlockSig, NULL);
1419
1420 /*
1421 * Set always-secure search path, so malicious users can't redirect user
1422 * code (e.g. operators).
1423 *
1424 * It's not strictly necessary since we won't be scanning or writing to
1425 * any user table locally, but it's good to retain it here for added
1426 * precaution.
1427 */
1428 SetConfigOption("search_path", "", PGC_SUSET, PGC_S_OVERRIDE);
1429
1431
1432 /*
1433 * Connect to the database specified by the user in primary_conninfo. We
1434 * need a database connection for walrcv_exec to work which we use to
1435 * fetch slot information from the remote node. See comments atop
1436 * libpqrcv_exec.
1437 *
1438 * We do not specify a specific user here since the slot sync worker will
1439 * operate as a superuser. This is safe because the slot sync worker does
1440 * not interact with user tables, eliminating the risk of executing
1441 * arbitrary code within triggers.
1442 */
1443 InitPostgres(dbname, InvalidOid, NULL, InvalidOid, 0, NULL);
1444
1446
1447 initStringInfo(&app_name);
1448 if (cluster_name[0])
1449 appendStringInfo(&app_name, "%s_%s", cluster_name, "slotsync worker");
1450 else
1451 appendStringInfoString(&app_name, "slotsync worker");
1452
1453 /*
1454 * Establish the connection to the primary server for slot
1455 * synchronization.
1456 */
1457 wrconn = walrcv_connect(PrimaryConnInfo, false, false, false,
1458 app_name.data, &err);
1459 pfree(app_name.data);
1460
1461 if (!wrconn)
1462 ereport(ERROR,
1463 errcode(ERRCODE_CONNECTION_FAILURE),
1464 errmsg("synchronization worker \"%s\" could not connect to the primary server: %s",
1465 app_name.data, err));
1466
1467 /*
1468 * Register the disconnection callback.
1469 *
1470 * XXX: This can be combined with previous cleanup registration of
1471 * slotsync_worker_onexit() but that will need the connection to be made
1472 * global and we want to avoid introducing global for this purpose.
1473 */
1475
1476 /*
1477 * Using the specified primary server connection, check that we are not a
1478 * cascading standby and slot configured in 'primary_slot_name' exists on
1479 * the primary server.
1480 */
1482
1483 /* Main loop to synchronize slots */
1484 for (;;)
1485 {
1486 bool some_slot_updated = false;
1487
1489
1490 some_slot_updated = synchronize_slots(wrconn);
1491
1492 wait_for_slot_activity(some_slot_updated);
1493 }
1494
1495 /*
1496 * The slot sync worker can't get here because it will only stop when it
1497 * receives a SIGINT from the startup process, or when there is an error.
1498 */
1499 Assert(false);
1500}
sigset_t UnBlockSig
Definition: pqsignal.c:22
#define Assert(condition)
Definition: c.h:812
void load_file(const char *filename, bool restricted)
Definition: dfmgr.c:134
void EmitErrorReport(void)
Definition: elog.c:1687
ErrorContextCallback * error_context_stack
Definition: elog.c:94
sigjmp_buf * PG_exception_stack
Definition: elog.c:96
#define LOG
Definition: elog.h:31
void err(int eval, const char *fmt,...)
Definition: err.c:43
int MyProcPid
Definition: globals.c:46
void SetConfigOption(const char *name, const char *value, GucContext context, GucSource source)
Definition: guc.c:4332
@ PGC_S_OVERRIDE
Definition: guc.h:119
@ PGC_SUSET
Definition: guc.h:74
char * cluster_name
Definition: guc_tables.c:537
void SignalHandlerForShutdownRequest(SIGNAL_ARGS)
Definition: interrupt.c:105
void SignalHandlerForConfigReload(SIGNAL_ARGS)
Definition: interrupt.c:61
void before_shmem_exit(pg_on_exit_callback function, Datum arg)
Definition: ipc.c:337
void proc_exit(int code)
Definition: ipc.c:104
void pfree(void *pointer)
Definition: mcxt.c:1521
@ NormalProcessing
Definition: miscadmin.h:461
@ InitProcessing
Definition: miscadmin.h:460
#define GetProcessingMode()
Definition: miscadmin.h:470
#define HOLD_INTERRUPTS()
Definition: miscadmin.h:133
#define SetProcessingMode(mode)
Definition: miscadmin.h:472
@ B_SLOTSYNC_WORKER
Definition: miscadmin.h:347
BackendType MyBackendType
Definition: miscinit.c:64
#define die(msg)
pqsigfunc pqsignal(int signo, pqsigfunc func)
void FloatExceptionHandler(SIGNAL_ARGS)
Definition: postgres.c:3048
static Datum PointerGetDatum(const void *X)
Definition: postgres.h:322
uintptr_t Datum
Definition: postgres.h:64
#define InvalidOid
Definition: postgres_ext.h:36
void BaseInit(void)
Definition: postinit.c:606
void InitPostgres(const char *in_dbname, Oid dboid, const char *username, Oid useroid, bits32 flags, char *out_dbname)
Definition: postinit.c:700
void procsignal_sigusr1_handler(SIGNAL_ARGS)
Definition: procsignal.c:671
void init_ps_display(const char *fixed_part)
Definition: ps_status.c:269
static void slotsync_worker_disconnect(int code, Datum arg)
Definition: slotsync.c:1175
static SlotSyncCtxStruct * SlotSyncCtx
Definition: slotsync.c:104
char * CheckAndGetDbnameFromConninfo(void)
Definition: slotsync.c:1010
static bool synchronize_slots(WalReceiverConn *wrconn)
Definition: slotsync.c:789
static void wait_for_slot_activity(bool some_slot_updated)
Definition: slotsync.c:1234
static void slotsync_worker_onexit(int code, Datum arg)
Definition: slotsync.c:1188
static void check_and_set_sync_info(pid_t worker_pid)
Definition: slotsync.c:1269
static void validate_remote_info(WalReceiverConn *wrconn)
Definition: slotsync.c:932
static void ProcessSlotSyncInterrupts(WalReceiverConn *wrconn)
Definition: slotsync.c:1153
void InitProcess(void)
Definition: proc.c:341
void appendStringInfo(StringInfo str, const char *fmt,...)
Definition: stringinfo.c:94
void appendStringInfoString(StringInfo str, const char *s)
Definition: stringinfo.c:179
void initStringInfo(StringInfo str)
Definition: stringinfo.c:56
void InitializeTimeouts(void)
Definition: timeout.c:470
static WalReceiverConn * wrconn
Definition: walreceiver.c:92
#define walrcv_connect(conninfo, replication, logical, must_use_password, appname, err)
Definition: walreceiver.h:435
#define SIGCHLD
Definition: win32_port.h:178
#define SIGHUP
Definition: win32_port.h:168
#define SIG_DFL
Definition: win32_port.h:163
#define SIGPIPE
Definition: win32_port.h:173
#define SIGUSR1
Definition: win32_port.h:180
#define SIGUSR2
Definition: win32_port.h:181
#define SIG_IGN
Definition: win32_port.h:165

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

◆ ShutDownSlotSync()

void ShutDownSlotSync ( void  )

Definition at line 1561 of file slotsync.c.

1562{
1563 pid_t worker_pid;
1564
1566
1567 SlotSyncCtx->stopSignaled = true;
1568
1569 /*
1570 * Return if neither the slot sync worker is running nor the function
1571 * pg_sync_replication_slots() is executing.
1572 */
1573 if (!SlotSyncCtx->syncing)
1574 {
1577 return;
1578 }
1579
1580 worker_pid = SlotSyncCtx->pid;
1581
1583
1584 if (worker_pid != InvalidPid)
1585 kill(worker_pid, SIGINT);
1586
1587 /* Wait for slot sync to end */
1588 for (;;)
1589 {
1590 int rc;
1591
1592 /* Wait a bit, we don't expect to have to wait long */
1593 rc = WaitLatch(MyLatch,
1595 10L, WAIT_EVENT_REPLICATION_SLOTSYNC_SHUTDOWN);
1596
1597 if (rc & WL_LATCH_SET)
1598 {
1601 }
1602
1604
1605 /* Ensure that no process is syncing the slots. */
1606 if (!SlotSyncCtx->syncing)
1607 break;
1608
1610 }
1611
1613
1615}
struct Latch * MyLatch
Definition: globals.c:62
void ResetLatch(Latch *latch)
Definition: latch.c:724
int WaitLatch(Latch *latch, int wakeEvents, long timeout, uint32 wait_event_info)
Definition: latch.c:517
#define WL_TIMEOUT
Definition: latch.h:130
#define WL_EXIT_ON_PM_DEATH
Definition: latch.h:132
#define WL_LATCH_SET
Definition: latch.h:127
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:122
#define InvalidPid
Definition: miscadmin.h:32
static void update_synced_slots_inactive_since(void)
Definition: slotsync.c:1509
#define SpinLockRelease(lock)
Definition: spin.h:61
#define SpinLockAcquire(lock)
Definition: spin.h:59
#define kill(pid, sig)
Definition: win32_port.h:503

References CHECK_FOR_INTERRUPTS, InvalidPid, kill, SlotSyncCtxStruct::mutex, MyLatch, SlotSyncCtxStruct::pid, ResetLatch(), 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 1667 of file slotsync.c.

1668{
1670 bool found;
1671
1673 ShmemInitStruct("Slot Sync Data", size, &found);
1674
1675 if (!found)
1676 {
1677 memset(SlotSyncCtx, 0, size);
1680 }
1681}
size_t Size
Definition: c.h:559
void * ShmemInitStruct(const char *name, Size size, bool *foundPtr)
Definition: shmem.c:382
static pg_noinline void Size size
Definition: slab.c:607
Size SlotSyncShmemSize(void)
Definition: slotsync.c:1658
#define SpinLockInit(lock)
Definition: spin.h:57

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

Referenced by CreateOrAttachShmemStructs().

◆ SlotSyncShmemSize()

Size SlotSyncShmemSize ( void  )

Definition at line 1658 of file slotsync.c.

1659{
1660 return sizeof(SlotSyncCtxStruct);
1661}
struct SlotSyncCtxStruct SlotSyncCtxStruct

Referenced by CalculateShmemSize(), and SlotSyncShmemInit().

◆ SlotSyncWorkerCanRestart()

bool SlotSyncWorkerCanRestart ( void  )

Definition at line 1629 of file slotsync.c.

1630{
1631 time_t curtime = time(NULL);
1632
1633 /* Return false if too soon since last start. */
1634 if ((unsigned int) (curtime - SlotSyncCtx->last_start_time) <
1635 (unsigned int) SLOTSYNC_RESTART_INTERVAL_SEC)
1636 return false;
1637
1638 SlotSyncCtx->last_start_time = curtime;
1639
1640 return true;
1641}
#define SLOTSYNC_RESTART_INTERVAL_SEC
Definition: slotsync.c:120
time_t last_start_time
Definition: slotsync.c:100

References SlotSyncCtxStruct::last_start_time, SLOTSYNC_RESTART_INTERVAL_SEC, and SlotSyncCtx.

Referenced by LaunchMissingBackgroundProcesses().

◆ SyncReplicationSlots()

void SyncReplicationSlots ( WalReceiverConn wrconn)

Definition at line 1724 of file slotsync.c.

1725{
1727 {
1729
1731
1733
1734 /* Cleanup the synced temporary slots */
1736
1737 /* We are done with sync, so reset sync flag */
1739 }
1741}
#define PG_ENSURE_ERROR_CLEANUP(cleanup_function, arg)
Definition: ipc.h:47
#define PG_END_ENSURE_ERROR_CLEANUP(cleanup_function, arg)
Definition: ipc.h:52
void ReplicationSlotCleanup(bool synced_only)
Definition: slot.c:745
static void slotsync_failure_callback(int code, Datum arg)
Definition: slotsync.c:1687
static void reset_syncing_flag()
Definition: slotsync.c:1313

References check_and_set_sync_info(), InvalidPid, PG_END_ENSURE_ERROR_CLEANUP, PG_ENSURE_ERROR_CLEANUP, PointerGetDatum(), ReplicationSlotCleanup(), reset_syncing_flag(), slotsync_failure_callback(), synchronize_slots(), validate_remote_info(), and wrconn.

Referenced by pg_sync_replication_slots().

◆ ValidateSlotSyncParams()

bool ValidateSlotSyncParams ( int  elevel)

Definition at line 1037 of file slotsync.c.

1038{
1039 /*
1040 * Logical slot sync/creation requires wal_level >= logical.
1041 *
1042 * Since altering the wal_level requires a server restart, so error out in
1043 * this case regardless of elevel provided by caller.
1044 */
1046 ereport(ERROR,
1047 errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1048 errmsg("replication slot synchronization requires \"wal_level\" >= \"logical\""));
1049
1050 /*
1051 * A physical replication slot(primary_slot_name) is required on the
1052 * primary to ensure that the rows needed by the standby are not removed
1053 * after restarting, so that the synchronized slot on the standby will not
1054 * be invalidated.
1055 */
1056 if (PrimarySlotName == NULL || *PrimarySlotName == '\0')
1057 {
1058 ereport(elevel,
1059 errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1060 /* translator: %s is a GUC variable name */
1061 errmsg("replication slot synchronization requires \"%s\" to be set", "primary_slot_name"));
1062 return false;
1063 }
1064
1065 /*
1066 * hot_standby_feedback must be enabled to cooperate with the physical
1067 * replication slot, which allows informing the primary about the xmin and
1068 * catalog_xmin values on the standby.
1069 */
1071 {
1072 ereport(elevel,
1073 errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1074 /* translator: %s is a GUC variable name */
1075 errmsg("replication slot synchronization requires \"%s\" to be enabled",
1076 "hot_standby_feedback"));
1077 return false;
1078 }
1079
1080 /*
1081 * The primary_conninfo is required to make connection to primary for
1082 * getting slots information.
1083 */
1084 if (PrimaryConnInfo == NULL || *PrimaryConnInfo == '\0')
1085 {
1086 ereport(elevel,
1087 errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1088 /* translator: %s is a GUC variable name */
1089 errmsg("replication slot synchronization requires \"%s\" to be set",
1090 "primary_conninfo"));
1091 return false;
1092 }
1093
1094 return true;
1095}
bool hot_standby_feedback
Definition: walreceiver.c:89
int wal_level
Definition: xlog.c:131
@ WAL_LEVEL_LOGICAL
Definition: xlog.h:76
char * PrimarySlotName
Definition: xlogrecovery.c:97

References ereport, errcode(), errmsg(), ERROR, hot_standby_feedback, PrimaryConnInfo, PrimarySlotName, wal_level, and WAL_LEVEL_LOGICAL.

Referenced by LaunchMissingBackgroundProcesses(), and pg_sync_replication_slots().

Variable Documentation

◆ PrimaryConnInfo

◆ PrimarySlotName

◆ sync_replication_slots

PGDLLIMPORT bool sync_replication_slots
extern

Definition at line 107 of file slotsync.c.

Referenced by LaunchMissingBackgroundProcesses(), and slotsync_reread_config().