PostgreSQL Source Code  git master
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()
 
int StartSlotSyncWorker (void)
 
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 1035 of file slotsync.c.

1036 {
1037  char *dbname;
1038 
1039  /*
1040  * The slot synchronization needs a database connection for walrcv_exec to
1041  * work.
1042  */
1044  if (dbname == NULL)
1045  ereport(ERROR,
1046 
1047  /*
1048  * translator: dbname is a specific option; %s is a GUC variable name
1049  */
1050  errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1051  errmsg("slot synchronization requires dbname to be specified in %s",
1052  "primary_conninfo"));
1053  return dbname;
1054 }
int errcode(int sqlerrcode)
Definition: elog.c:859
int errmsg(const char *fmt,...)
Definition: elog.c:1072
#define ERROR
Definition: elog.h:39
#define ereport(elevel,...)
Definition: elog.h:149
char * dbname
Definition: streamutil.c:52
#define walrcv_get_dbname_from_conninfo(conninfo)
Definition: walreceiver.h:442
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 1569 of file slotsync.c.

1570 {
1571  return syncing_slots;
1572 }
static bool syncing_slots
Definition: slotsync.c:131

References syncing_slots.

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

◆ ReplSlotSyncWorkerMain()

void ReplSlotSyncWorkerMain ( char *  startup_data,
size_t  startup_data_len 
)

Definition at line 1252 of file slotsync.c.

1253 {
1254  WalReceiverConn *wrconn = NULL;
1255  char *dbname;
1256  char *err;
1257  sigjmp_buf local_sigjmp_buf;
1258  StringInfoData app_name;
1259 
1260  Assert(startup_data_len == 0);
1261 
1263 
1264  init_ps_display(NULL);
1265 
1267 
1268  /*
1269  * Create a per-backend PGPROC struct in shared memory. We must do this
1270  * before we access any shared memory.
1271  */
1272  InitProcess();
1273 
1274  /*
1275  * Early initialization.
1276  */
1277  BaseInit();
1278 
1279  Assert(SlotSyncCtx != NULL);
1280 
1283 
1284  /*
1285  * Startup process signaled the slot sync worker to stop, so if meanwhile
1286  * postmaster ended up starting the worker again, exit.
1287  */
1289  {
1291  proc_exit(0);
1292  }
1293 
1294  /* Advertise our PID so that the startup process can kill us on promotion */
1297 
1298  ereport(LOG, errmsg("slot sync worker started"));
1299 
1300  /* Register it as soon as SlotSyncCtx->pid is initialized. */
1302 
1303  /* Setup signal handling */
1306  pqsignal(SIGTERM, die);
1312 
1313  /*
1314  * Establishes SIGALRM handler and initialize timeout module. It is needed
1315  * by InitPostgres to register different timeouts.
1316  */
1318 
1319  /* Load the libpq-specific functions */
1320  load_file("libpqwalreceiver", false);
1321 
1322  /*
1323  * If an exception is encountered, processing resumes here.
1324  *
1325  * We just need to clean up, report the error, and go away.
1326  *
1327  * If we do not have this handling here, then since this worker process
1328  * operates at the bottom of the exception stack, ERRORs turn into FATALs.
1329  * Therefore, we create our own exception handler to catch ERRORs.
1330  */
1331  if (sigsetjmp(local_sigjmp_buf, 1) != 0)
1332  {
1333  /* since not using PG_TRY, must reset error stack by hand */
1334  error_context_stack = NULL;
1335 
1336  /* Prevents interrupts while cleaning up */
1337  HOLD_INTERRUPTS();
1338 
1339  /* Report the error to the server log */
1340  EmitErrorReport();
1341 
1342  /*
1343  * We can now go away. Note that because we called InitProcess, a
1344  * callback was registered to do ProcKill, which will clean up
1345  * necessary state.
1346  */
1347  proc_exit(0);
1348  }
1349 
1350  /* We can now handle ereport(ERROR) */
1351  PG_exception_stack = &local_sigjmp_buf;
1352 
1353  /*
1354  * Unblock signals (they were blocked when the postmaster forked us)
1355  */
1356  sigprocmask(SIG_SETMASK, &UnBlockSig, NULL);
1357 
1358  /*
1359  * Set always-secure search path, so malicious users can't redirect user
1360  * code (e.g. operators).
1361  *
1362  * It's not strictly necessary since we won't be scanning or writing to
1363  * any user table locally, but it's good to retain it here for added
1364  * precaution.
1365  */
1366  SetConfigOption("search_path", "", PGC_SUSET, PGC_S_OVERRIDE);
1367 
1369 
1370  /*
1371  * Connect to the database specified by the user in primary_conninfo. We
1372  * need a database connection for walrcv_exec to work which we use to
1373  * fetch slot information from the remote node. See comments atop
1374  * libpqrcv_exec.
1375  *
1376  * We do not specify a specific user here since the slot sync worker will
1377  * operate as a superuser. This is safe because the slot sync worker does
1378  * not interact with user tables, eliminating the risk of executing
1379  * arbitrary code within triggers.
1380  */
1381  InitPostgres(dbname, InvalidOid, NULL, InvalidOid, 0, NULL);
1382 
1384 
1385  initStringInfo(&app_name);
1386  if (cluster_name[0])
1387  appendStringInfo(&app_name, "%s_%s", cluster_name, "slotsync worker");
1388  else
1389  appendStringInfoString(&app_name, "slotsync worker");
1390 
1391  /*
1392  * Establish the connection to the primary server for slot
1393  * synchronization.
1394  */
1395  wrconn = walrcv_connect(PrimaryConnInfo, false, false, false,
1396  app_name.data, &err);
1397  pfree(app_name.data);
1398 
1399  if (!wrconn)
1400  ereport(ERROR,
1401  errcode(ERRCODE_CONNECTION_FAILURE),
1402  errmsg("could not connect to the primary server: %s", err));
1403 
1404  /*
1405  * Register the failure callback once we have the connection.
1406  *
1407  * XXX: This can be combined with previous such cleanup registration of
1408  * slotsync_worker_onexit() but that will need the connection to be made
1409  * global and we want to avoid introducing global for this purpose.
1410  */
1412 
1413  /*
1414  * Using the specified primary server connection, check that we are not a
1415  * cascading standby and slot configured in 'primary_slot_name' exists on
1416  * the primary server.
1417  */
1419 
1420  /* Main loop to synchronize slots */
1421  for (;;)
1422  {
1423  bool some_slot_updated = false;
1424 
1426 
1427  some_slot_updated = synchronize_slots(wrconn);
1428 
1429  wait_for_slot_activity(some_slot_updated);
1430  }
1431 
1432  /*
1433  * The slot sync worker can't get here because it will only stop when it
1434  * receives a SIGINT from the startup process, or when there is an error.
1435  */
1436  Assert(false);
1437 }
sigset_t UnBlockSig
Definition: pqsignal.c:22
#define Assert(condition)
Definition: c.h:858
void load_file(const char *filename, bool restricted)
Definition: dfmgr.c:144
void EmitErrorReport(void)
Definition: elog.c:1672
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:45
void SetConfigOption(const char *name, const char *value, GucContext context, GucSource source)
Definition: guc.c:4275
@ PGC_S_OVERRIDE
Definition: guc.h:119
@ PGC_SUSET
Definition: guc.h:74
char * cluster_name
Definition: guc_tables.c:540
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:1520
@ NormalProcessing
Definition: miscadmin.h:449
@ InitProcessing
Definition: miscadmin.h:448
#define HOLD_INTERRUPTS()
Definition: miscadmin.h:133
#define SetProcessingMode(mode)
Definition: miscadmin.h:460
@ B_SLOTSYNC_WORKER
Definition: miscadmin.h:343
#define InvalidPid
Definition: miscadmin.h:32
BackendType MyBackendType
Definition: miscinit.c:63
#define die(msg)
pqsigfunc pqsignal(int signo, pqsigfunc func)
void FloatExceptionHandler(SIGNAL_ARGS)
Definition: postgres.c:3019
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:645
void InitPostgres(const char *in_dbname, Oid dboid, const char *username, Oid useroid, bits32 flags, char *out_dbname)
Definition: postinit.c:736
void procsignal_sigusr1_handler(SIGNAL_ARGS)
Definition: procsignal.c:635
void init_ps_display(const char *fixed_part)
Definition: ps_status.c:267
SlotSyncCtxStruct * SlotSyncCtx
Definition: slotsync.c:108
static void slotsync_failure_callback(int code, Datum arg)
Definition: slotsync.c:1607
static bool synchronize_slots(WalReceiverConn *wrconn)
Definition: slotsync.c:793
static void wait_for_slot_activity(bool some_slot_updated)
Definition: slotsync.c:1215
char * CheckAndGetDbnameFromConninfo(void)
Definition: slotsync.c:1035
static void slotsync_worker_onexit(int code, Datum arg)
Definition: slotsync.c:1199
static void validate_remote_info(WalReceiverConn *wrconn)
Definition: slotsync.c:956
static void ProcessSlotSyncInterrupts(WalReceiverConn *wrconn)
Definition: slotsync.c:1177
#define SpinLockRelease(lock)
Definition: spin.h:64
#define SpinLockAcquire(lock)
Definition: spin.h:62
void InitProcess(void)
Definition: proc.c:296
void appendStringInfo(StringInfo str, const char *fmt,...)
Definition: stringinfo.c:97
void appendStringInfoString(StringInfo str, const char *s)
Definition: stringinfo.c:182
void initStringInfo(StringInfo str)
Definition: stringinfo.c:59
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:432
#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(), CheckAndGetDbnameFromConninfo(), cluster_name, StringInfoData::data, dbname, die, EmitErrorReport(), ereport, err(), errcode(), errmsg(), ERROR, error_context_stack, FloatExceptionHandler(), HOLD_INTERRUPTS, init_ps_display(), InitializeTimeouts(), InitPostgres(), InitProcess(), InitProcessing, initStringInfo(), InvalidOid, InvalidPid, load_file(), LOG, SlotSyncCtxStruct::mutex, MyBackendType, MyProcPid, NormalProcessing, pfree(), PG_exception_stack, PGC_S_OVERRIDE, PGC_SUSET, SlotSyncCtxStruct::pid, PointerGetDatum(), pqsignal(), PrimaryConnInfo, proc_exit(), ProcessSlotSyncInterrupts(), procsignal_sigusr1_handler(), SetConfigOption(), SetProcessingMode, SIG_DFL, SIG_IGN, SIGCHLD, SIGHUP, SignalHandlerForConfigReload(), SignalHandlerForShutdownRequest(), SIGPIPE, SIGUSR1, SIGUSR2, slotsync_failure_callback(), slotsync_worker_onexit(), SlotSyncCtx, SpinLockAcquire, SpinLockRelease, SlotSyncCtxStruct::stopSignaled, synchronize_slots(), UnBlockSig, validate_remote_info(), wait_for_slot_activity(), walrcv_connect, and wrconn.

◆ ShutDownSlotSync()

void ShutDownSlotSync ( void  )

Definition at line 1491 of file slotsync.c.

1492 {
1494 
1495  SlotSyncCtx->stopSignaled = true;
1496 
1497  if (SlotSyncCtx->pid == InvalidPid)
1498  {
1501  return;
1502  }
1504 
1505  kill(SlotSyncCtx->pid, SIGINT);
1506 
1507  /* Wait for it to die */
1508  for (;;)
1509  {
1510  int rc;
1511 
1512  /* Wait a bit, we don't expect to have to wait long */
1513  rc = WaitLatch(MyLatch,
1515  10L, WAIT_EVENT_REPLICATION_SLOTSYNC_SHUTDOWN);
1516 
1517  if (rc & WL_LATCH_SET)
1518  {
1521  }
1522 
1524 
1525  /* Is it gone? */
1526  if (SlotSyncCtx->pid == InvalidPid)
1527  break;
1528 
1530  }
1531 
1533 
1535 }
struct Latch * MyLatch
Definition: globals.c:60
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
static void update_synced_slots_inactive_since(void)
Definition: slotsync.c:1446
#define kill(pid, sig)
Definition: win32_port.h:485

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

1588 {
1590  bool found;
1591 
1593  ShmemInitStruct("Slot Sync Data", size, &found);
1594 
1595  if (!found)
1596  {
1597  memset(SlotSyncCtx, 0, size);
1600  }
1601 }
size_t Size
Definition: c.h:605
void * ShmemInitStruct(const char *name, Size size, bool *foundPtr)
Definition: shmem.c:387
static pg_noinline void Size size
Definition: slab.c:607
Size SlotSyncShmemSize(void)
Definition: slotsync.c:1578
#define SpinLockInit(lock)
Definition: spin.h:60

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

Referenced by CreateOrAttachShmemStructs().

◆ SlotSyncShmemSize()

Size SlotSyncShmemSize ( void  )

Definition at line 1578 of file slotsync.c.

1579 {
1580  return sizeof(SlotSyncCtxStruct);
1581 }
struct SlotSyncCtxStruct SlotSyncCtxStruct

Referenced by CalculateShmemSize(), and SlotSyncShmemInit().

◆ SlotSyncWorkerCanRestart()

bool SlotSyncWorkerCanRestart ( void  )

Definition at line 1549 of file slotsync.c.

1550 {
1551  time_t curtime = time(NULL);
1552 
1553  /* Return false if too soon since last start. */
1554  if ((unsigned int) (curtime - SlotSyncCtx->last_start_time) <
1555  (unsigned int) SLOTSYNC_RESTART_INTERVAL_SEC)
1556  return false;
1557 
1558  SlotSyncCtx->last_start_time = curtime;
1559 
1560  return true;
1561 }
#define SLOTSYNC_RESTART_INTERVAL_SEC
Definition: slotsync.c:124
time_t last_start_time
Definition: slotsync.c:104

References SlotSyncCtxStruct::last_start_time, SLOTSYNC_RESTART_INTERVAL_SEC, and SlotSyncCtx.

Referenced by MaybeStartSlotSyncWorker().

◆ StartSlotSyncWorker()

int StartSlotSyncWorker ( void  )

◆ SyncReplicationSlots()

void SyncReplicationSlots ( WalReceiverConn wrconn)

Definition at line 1633 of file slotsync.c.

1634 {
1636  {
1638 
1640  }
1642 }
#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

References PG_END_ENSURE_ERROR_CLEANUP, PG_ENSURE_ERROR_CLEANUP, PointerGetDatum(), slotsync_failure_callback(), synchronize_slots(), validate_remote_info(), and wrconn.

Referenced by pg_sync_replication_slots().

◆ ValidateSlotSyncParams()

bool ValidateSlotSyncParams ( int  elevel)

Definition at line 1061 of file slotsync.c.

1062 {
1063  /*
1064  * Logical slot sync/creation requires wal_level >= logical.
1065  *
1066  * Sincle altering the wal_level requires a server restart, so error out
1067  * in this case regardless of elevel provided by caller.
1068  */
1070  ereport(ERROR,
1071  errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1072  errmsg("slot synchronization requires wal_level >= \"logical\""));
1073 
1074  /*
1075  * A physical replication slot(primary_slot_name) is required on the
1076  * primary to ensure that the rows needed by the standby are not removed
1077  * after restarting, so that the synchronized slot on the standby will not
1078  * be invalidated.
1079  */
1080  if (PrimarySlotName == NULL || *PrimarySlotName == '\0')
1081  {
1082  ereport(elevel,
1083  /* translator: %s is a GUC variable name */
1084  errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1085  errmsg("slot synchronization requires %s to be defined", "primary_slot_name"));
1086  return false;
1087  }
1088 
1089  /*
1090  * hot_standby_feedback must be enabled to cooperate with the physical
1091  * replication slot, which allows informing the primary about the xmin and
1092  * catalog_xmin values on the standby.
1093  */
1094  if (!hot_standby_feedback)
1095  {
1096  ereport(elevel,
1097  /* translator: %s is a GUC variable name */
1098  errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1099  errmsg("slot synchronization requires %s to be enabled",
1100  "hot_standby_feedback"));
1101  return false;
1102  }
1103 
1104  /*
1105  * The primary_conninfo is required to make connection to primary for
1106  * getting slots information.
1107  */
1108  if (PrimaryConnInfo == NULL || *PrimaryConnInfo == '\0')
1109  {
1110  ereport(elevel,
1111  /* translator: %s is a GUC variable name */
1112  errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1113  errmsg("slot synchronization requires %s to be defined",
1114  "primary_conninfo"));
1115  return false;
1116  }
1117 
1118  return true;
1119 }
bool hot_standby_feedback
Definition: walreceiver.c:89
int wal_level
Definition: xlog.c:131
@ WAL_LEVEL_LOGICAL
Definition: xlog.h:74
char * PrimarySlotName
Definition: xlogrecovery.c:97

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

Referenced by MaybeStartSlotSyncWorker(), and pg_sync_replication_slots().

Variable Documentation

◆ PrimaryConnInfo

◆ PrimarySlotName

◆ sync_replication_slots

PGDLLIMPORT bool sync_replication_slots
extern

Definition at line 111 of file slotsync.c.

Referenced by MaybeStartSlotSyncWorker(), and slotsync_reread_config().