PostgreSQL Source Code git master
Data Structures | Macros | Typedefs | Functions | Variables
syncrep.h File Reference
#include "access/xlogdefs.h"
Include dependency graph for syncrep.h:
This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Data Structures

struct   SyncRepStandbyData
 
struct   SyncRepConfigData
 

Macros

#define  SyncRepRequested()    (max_wal_senders > 0 && synchronous_commit > SYNCHRONOUS_COMMIT_LOCAL_FLUSH)
 
#define  SYNC_REP_NO_WAIT   (-1)
 
#define  SYNC_REP_WAIT_WRITE   0
 
#define  SYNC_REP_WAIT_FLUSH   1
 
#define  SYNC_REP_WAIT_APPLY   2
 
#define  NUM_SYNC_REP_WAIT_MODE   3
 
#define  SYNC_REP_NOT_WAITING   0
 
#define  SYNC_REP_WAITING   1
 
#define  SYNC_REP_WAIT_COMPLETE   2
 
#define  SYNC_REP_PRIORITY   0
 
#define  SYNC_REP_QUORUM   1
 

Typedefs

typedef struct SyncRepStandbyData  SyncRepStandbyData
 
typedef struct SyncRepConfigData  SyncRepConfigData
 
typedef void *  yyscan_t
 

Functions

void  SyncRepWaitForLSN (XLogRecPtr lsn, bool commit)
 
void  SyncRepCleanupAtProcExit (void)
 
void  SyncRepInitConfig (void)
 
void  SyncRepReleaseWaiters (void)
 
int  SyncRepGetCandidateStandbys (SyncRepStandbyData **standbys)
 
void  SyncRepUpdateSyncStandbysDefined (void)
 
int  syncrep_yyparse (SyncRepConfigData **syncrep_parse_result_p, char **syncrep_parse_error_msg_p, yyscan_t yyscanner)
 
int  syncrep_yylex (union YYSTYPE *yylval_param, char **syncrep_parse_error_msg_p, yyscan_t yyscanner)
 
void  syncrep_yyerror (SyncRepConfigData **syncrep_parse_result_p, char **syncrep_parse_error_msg_p, yyscan_t yyscanner, const char *str)
 
void  syncrep_scanner_init (const char *str, yyscan_t *yyscannerp)
 
void  syncrep_scanner_finish (yyscan_t yyscanner)
 

Variables

PGDLLIMPORT SyncRepConfigData *  SyncRepConfig
 
PGDLLIMPORT char *  SyncRepStandbyNames
 

Macro Definition Documentation

◆ NUM_SYNC_REP_WAIT_MODE

#define NUM_SYNC_REP_WAIT_MODE   3

Definition at line 27 of file syncrep.h.

◆ SYNC_REP_NO_WAIT

#define SYNC_REP_NO_WAIT   (-1)

Definition at line 22 of file syncrep.h.

◆ SYNC_REP_NOT_WAITING

#define SYNC_REP_NOT_WAITING   0

Definition at line 30 of file syncrep.h.

◆ SYNC_REP_PRIORITY

#define SYNC_REP_PRIORITY   0

Definition at line 35 of file syncrep.h.

◆ SYNC_REP_QUORUM

#define SYNC_REP_QUORUM   1

Definition at line 36 of file syncrep.h.

◆ SYNC_REP_WAIT_APPLY

#define SYNC_REP_WAIT_APPLY   2

Definition at line 25 of file syncrep.h.

◆ SYNC_REP_WAIT_COMPLETE

#define SYNC_REP_WAIT_COMPLETE   2

Definition at line 32 of file syncrep.h.

◆ SYNC_REP_WAIT_FLUSH

#define SYNC_REP_WAIT_FLUSH   1

Definition at line 24 of file syncrep.h.

◆ SYNC_REP_WAIT_WRITE

#define SYNC_REP_WAIT_WRITE   0

Definition at line 23 of file syncrep.h.

◆ SYNC_REP_WAITING

#define SYNC_REP_WAITING   1

Definition at line 31 of file syncrep.h.

◆ SyncRepRequested

#define SyncRepRequested ( )     (max_wal_senders > 0 && synchronous_commit > SYNCHRONOUS_COMMIT_LOCAL_FLUSH)

Definition at line 18 of file syncrep.h.

Typedef Documentation

◆ SyncRepConfigData

typedef struct SyncRepConfigData SyncRepConfigData

◆ SyncRepStandbyData

typedef struct SyncRepStandbyData SyncRepStandbyData

◆ yyscan_t

typedef void* yyscan_t

Definition at line 100 of file syncrep.h.

Function Documentation

◆ syncrep_scanner_finish()

void syncrep_scanner_finish ( yyscan_t  yyscanner )

Definition at line 189 of file syncrep_scanner.l.

190{
191 pfree(yyextra);
192 yylex_destroy(yyscanner);
193}
void pfree(void *pointer)
Definition: mcxt.c:1594
#define yyextra

References pfree(), and yyextra.

Referenced by check_synchronous_standby_names().

◆ syncrep_scanner_init()

void syncrep_scanner_init ( const char *  str,
yyscan_t *  yyscannerp 
)

Definition at line 173 of file syncrep_scanner.l.

174{
175 yyscan_t yyscanner;
177
178 if (yylex_init(yyscannerp) != 0)
179 elog(ERROR, "yylex_init() failed: %m");
180
181 yyscanner = *yyscannerp;
182
183 yyset_extra(yyext, yyscanner);
184
185 yy_scan_string(str, yyscanner);
186}
void * yyscan_t
Definition: cubedata.h:65
#define ERROR
Definition: elog.h:39
#define elog(elevel,...)
Definition: elog.h:226
#define palloc0_object(type)
Definition: fe_memutils.h:75
const char * str

References elog, ERROR, palloc0_object, and str.

Referenced by check_synchronous_standby_names().

◆ syncrep_yyerror()

void syncrep_yyerror ( SyncRepConfigData **  syncrep_parse_result_p,
char **  syncrep_parse_error_msg_p,
yyscan_t  yyscanner,
const char *  str 
)

Definition at line 156 of file syncrep_scanner.l.

157{
158 struct yyguts_t *yyg = (struct yyguts_t *) yyscanner; /* needed for yytext
159 * macro */
160
161 /* report only the first error in a parse operation */
162 if (*syncrep_parse_error_msg_p)
163 return;
164 if (yytext[0])
165 *syncrep_parse_error_msg_p = psprintf("%s at or near \"%s\"",
166 message, yytext);
167 else
168 *syncrep_parse_error_msg_p = psprintf("%s at end of input",
169 message);
170}
char * psprintf(const char *fmt,...)
Definition: psprintf.c:43
Definition: pl_scanner.c:141

References psprintf().

◆ syncrep_yylex()

int syncrep_yylex ( union YYSTYPE *  yylval_param,
char **  syncrep_parse_error_msg_p,
yyscan_t  yyscanner 
)

◆ syncrep_yyparse()

int syncrep_yyparse ( SyncRepConfigData **  syncrep_parse_result_p,
char **  syncrep_parse_error_msg_p,
yyscan_t  yyscanner 
)

Referenced by check_synchronous_standby_names().

◆ SyncRepCleanupAtProcExit()

void SyncRepCleanupAtProcExit ( void  )

Definition at line 416 of file syncrep.c.

417{
418 /*
419 * First check if we are removed from the queue without the lock to not
420 * slow down backend exit.
421 */
423 {
424 LWLockAcquire(SyncRepLock, LW_EXCLUSIVE);
425
426 /* maybe we have just been removed, so recheck */
429
430 LWLockRelease(SyncRepLock);
431 }
432}
static void dlist_delete_thoroughly(dlist_node *node)
Definition: ilist.h:416
static bool dlist_node_is_detached(const dlist_node *node)
Definition: ilist.h:525
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1174
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1894
@ LW_EXCLUSIVE
Definition: lwlock.h:112
PGPROC * MyProc
Definition: proc.c:66
dlist_node syncRepLinks
Definition: proc.h:271

References dlist_delete_thoroughly(), dlist_node_is_detached(), LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MyProc, and PGPROC::syncRepLinks.

Referenced by ProcKill().

◆ SyncRepGetCandidateStandbys()

int SyncRepGetCandidateStandbys ( SyncRepStandbyData **  standbys )

Definition at line 754 of file syncrep.c.

755{
756 int i;
757 int n;
758
759 /* Create result array */
760 *standbys = (SyncRepStandbyData *)
762
763 /* Quick exit if sync replication is not requested */
764 if (SyncRepConfig == NULL)
765 return 0;
766
767 /* Collect raw data from shared memory */
768 n = 0;
769 for (i = 0; i < max_wal_senders; i++)
770 {
771 volatile WalSnd *walsnd; /* Use volatile pointer to prevent code
772 * rearrangement */
773 SyncRepStandbyData *stby;
774 WalSndState state; /* not included in SyncRepStandbyData */
775
776 walsnd = &WalSndCtl->walsnds[i];
777 stby = *standbys + n;
778
779 SpinLockAcquire(&walsnd->mutex);
780 stby->pid = walsnd->pid;
781 state = walsnd->state;
782 stby->write = walsnd->write;
783 stby->flush = walsnd->flush;
784 stby->apply = walsnd->apply;
786 SpinLockRelease(&walsnd->mutex);
787
788 /* Must be active */
789 if (stby->pid == 0)
790 continue;
791
792 /* Must be streaming or stopping */
795 continue;
796
797 /* Must be synchronous */
798 if (stby->sync_standby_priority == 0)
799 continue;
800
801 /* Must have a valid flush position */
802 if (XLogRecPtrIsInvalid(stby->flush))
803 continue;
804
805 /* OK, it's a candidate */
806 stby->walsnd_index = i;
807 stby->is_me = (walsnd == MyWalSnd);
808 n++;
809 }
810
811 /*
812 * In quorum mode, we return all the candidates. In priority mode, if we
813 * have too many candidates then return only the num_sync ones of highest
814 * priority.
815 */
818 {
819 /* Sort by priority ... */
820 qsort(*standbys, n, sizeof(SyncRepStandbyData),
822 /* ... then report just the first num_sync ones */
824 }
825
826 return n;
827}
i
int i
Definition: isn.c:77
void * palloc(Size size)
Definition: mcxt.c:1365
#define qsort(a, b, c, d)
Definition: port.h:479
#define SpinLockRelease(lock)
Definition: spin.h:61
#define SpinLockAcquire(lock)
Definition: spin.h:59
uint8 syncrep_method
Definition: syncrep.h:68
int num_sync
Definition: syncrep.h:66
int walsnd_index
Definition: syncrep.h:51
int sync_standby_priority
Definition: syncrep.h:49
XLogRecPtr apply
Definition: syncrep.h:48
XLogRecPtr write
Definition: syncrep.h:46
XLogRecPtr flush
Definition: syncrep.h:47
WalSnd walsnds[FLEXIBLE_ARRAY_MEMBER]
slock_t mutex
XLogRecPtr flush
pid_t pid
WalSndState state
XLogRecPtr write
int sync_standby_priority
XLogRecPtr apply
Definition: regguts.h:323
static int standby_priority_comparator(const void *a, const void *b)
Definition: syncrep.c:833
SyncRepConfigData * SyncRepConfig
Definition: syncrep.c:97
#define SYNC_REP_PRIORITY
Definition: syncrep.h:35
WalSnd * MyWalSnd
Definition: walsender.c:120
int max_wal_senders
Definition: walsender.c:129
WalSndCtlData * WalSndCtl
Definition: walsender.c:117
WalSndState
@ WALSNDSTATE_STREAMING
@ WALSNDSTATE_STOPPING
#define XLogRecPtrIsInvalid(r)
Definition: xlogdefs.h:29

References SyncRepStandbyData::apply, WalSnd::apply, SyncRepStandbyData::flush, WalSnd::flush, i, SyncRepStandbyData::is_me, max_wal_senders, WalSnd::mutex, MyWalSnd, SyncRepConfigData::num_sync, palloc(), SyncRepStandbyData::pid, WalSnd::pid, qsort, SpinLockAcquire, SpinLockRelease, standby_priority_comparator(), WalSnd::state, SYNC_REP_PRIORITY, SyncRepStandbyData::sync_standby_priority, WalSnd::sync_standby_priority, SyncRepConfigData::syncrep_method, SyncRepConfig, SyncRepStandbyData::walsnd_index, WalSndCtl, WalSndCtlData::walsnds, WALSNDSTATE_STOPPING, WALSNDSTATE_STREAMING, SyncRepStandbyData::write, WalSnd::write, and XLogRecPtrIsInvalid.

Referenced by pg_stat_get_wal_senders(), and SyncRepGetSyncRecPtr().

◆ SyncRepInitConfig()

void SyncRepInitConfig ( void  )

Definition at line 445 of file syncrep.c.

446{
447 int priority;
448
449 /*
450 * Determine if we are a potential sync standby and remember the result
451 * for handling replies from standby.
452 */
453 priority = SyncRepGetStandbyPriority();
454 if (MyWalSnd->sync_standby_priority != priority)
455 {
459
461 (errmsg_internal("standby \"%s\" now has synchronous standby priority %d",
462 application_name, priority)));
463 }
464}
int errmsg_internal(const char *fmt,...)
Definition: elog.c:1161
#define DEBUG1
Definition: elog.h:30
#define ereport(elevel,...)
Definition: elog.h:150
char * application_name
Definition: guc_tables.c:561
static int SyncRepGetStandbyPriority(void)
Definition: syncrep.c:860

References application_name, DEBUG1, ereport, errmsg_internal(), WalSnd::mutex, MyWalSnd, SpinLockAcquire, SpinLockRelease, WalSnd::sync_standby_priority, and SyncRepGetStandbyPriority().

Referenced by ProcessPendingWrites(), StartLogicalReplication(), StartReplication(), WalSndLoop(), and WalSndWaitForWal().

◆ SyncRepReleaseWaiters()

void SyncRepReleaseWaiters ( void  )

Definition at line 474 of file syncrep.c.

475{
476 volatile WalSndCtlData *walsndctl = WalSndCtl;
477 XLogRecPtr writePtr;
478 XLogRecPtr flushPtr;
479 XLogRecPtr applyPtr;
480 bool got_recptr;
481 bool am_sync;
482 int numwrite = 0;
483 int numflush = 0;
484 int numapply = 0;
485
486 /*
487 * If this WALSender is serving a standby that is not on the list of
488 * potential sync standbys then we have nothing to do. If we are still
489 * starting up, still running base backup or the current flush position is
490 * still invalid, then leave quickly also. Streaming or stopping WAL
491 * senders are allowed to release waiters.
492 */
497 {
499 return;
500 }
501
502 /*
503 * We're a potential sync standby. Release waiters if there are enough
504 * sync standbys and we are considered as sync.
505 */
506 LWLockAcquire(SyncRepLock, LW_EXCLUSIVE);
507
508 /*
509 * Check whether we are a sync standby or not, and calculate the synced
510 * positions among all sync standbys. (Note: although this step does not
511 * of itself require holding SyncRepLock, it seems like a good idea to do
512 * it after acquiring the lock. This ensures that the WAL pointers we use
513 * to release waiters are newer than any previous execution of this
514 * routine used.)
515 */
516 got_recptr = SyncRepGetSyncRecPtr(&writePtr, &flushPtr, &applyPtr, &am_sync);
517
518 /*
519 * If we are managing a sync standby, though we weren't prior to this,
520 * then announce we are now a sync standby.
521 */
522 if (announce_next_takeover && am_sync)
523 {
525
527 ereport(LOG,
528 (errmsg("standby \"%s\" is now a synchronous standby with priority %d",
530 else
531 ereport(LOG,
532 (errmsg("standby \"%s\" is now a candidate for quorum synchronous standby",
534 }
535
536 /*
537 * If the number of sync standbys is less than requested or we aren't
538 * managing a sync standby then just leave.
539 */
540 if (!got_recptr || !am_sync)
541 {
542 LWLockRelease(SyncRepLock);
543 announce_next_takeover = !am_sync;
544 return;
545 }
546
547 /*
548 * Set the lsn first so that when we wake backends they will release up to
549 * this location.
550 */
551 if (walsndctl->lsn[SYNC_REP_WAIT_WRITE] < writePtr)
552 {
553 walsndctl->lsn[SYNC_REP_WAIT_WRITE] = writePtr;
554 numwrite = SyncRepWakeQueue(false, SYNC_REP_WAIT_WRITE);
555 }
556 if (walsndctl->lsn[SYNC_REP_WAIT_FLUSH] < flushPtr)
557 {
558 walsndctl->lsn[SYNC_REP_WAIT_FLUSH] = flushPtr;
559 numflush = SyncRepWakeQueue(false, SYNC_REP_WAIT_FLUSH);
560 }
561 if (walsndctl->lsn[SYNC_REP_WAIT_APPLY] < applyPtr)
562 {
563 walsndctl->lsn[SYNC_REP_WAIT_APPLY] = applyPtr;
564 numapply = SyncRepWakeQueue(false, SYNC_REP_WAIT_APPLY);
565 }
566
567 LWLockRelease(SyncRepLock);
568
569 elog(DEBUG3, "released %d procs up to write %X/%08X, %d procs up to flush %X/%08X, %d procs up to apply %X/%08X",
570 numwrite, LSN_FORMAT_ARGS(writePtr),
571 numflush, LSN_FORMAT_ARGS(flushPtr),
572 numapply, LSN_FORMAT_ARGS(applyPtr));
573}
int errmsg(const char *fmt,...)
Definition: elog.c:1071
#define LOG
Definition: elog.h:31
#define DEBUG3
Definition: elog.h:28
XLogRecPtr lsn[NUM_SYNC_REP_WAIT_MODE]
static bool SyncRepGetSyncRecPtr(XLogRecPtr *writePtr, XLogRecPtr *flushPtr, XLogRecPtr *applyPtr, bool *am_sync)
Definition: syncrep.c:586
static int SyncRepWakeQueue(bool all, int mode)
Definition: syncrep.c:907
static bool announce_next_takeover
Definition: syncrep.c:95
#define SYNC_REP_WAIT_WRITE
Definition: syncrep.h:23
#define SYNC_REP_WAIT_FLUSH
Definition: syncrep.h:24
#define SYNC_REP_WAIT_APPLY
Definition: syncrep.h:25
#define LSN_FORMAT_ARGS(lsn)
Definition: xlogdefs.h:46
uint64 XLogRecPtr
Definition: xlogdefs.h:21

References announce_next_takeover, application_name, DEBUG3, elog, ereport, errmsg(), WalSnd::flush, LOG, WalSndCtlData::lsn, LSN_FORMAT_ARGS, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MyWalSnd, WalSnd::state, SYNC_REP_PRIORITY, SYNC_REP_WAIT_APPLY, SYNC_REP_WAIT_FLUSH, SYNC_REP_WAIT_WRITE, WalSnd::sync_standby_priority, SyncRepConfigData::syncrep_method, SyncRepConfig, SyncRepGetSyncRecPtr(), SyncRepWakeQueue(), WalSndCtl, WALSNDSTATE_STOPPING, WALSNDSTATE_STREAMING, and XLogRecPtrIsInvalid.

Referenced by ProcessStandbyReplyMessage().

◆ SyncRepUpdateSyncStandbysDefined()

void SyncRepUpdateSyncStandbysDefined ( void  )

Definition at line 964 of file syncrep.c.

965{
966 bool sync_standbys_defined = SyncStandbysDefined();
967
968 if (sync_standbys_defined !=
970 {
971 LWLockAcquire(SyncRepLock, LW_EXCLUSIVE);
972
973 /*
974 * If synchronous_standby_names has been reset to empty, it's futile
975 * for backends to continue waiting. Since the user no longer wants
976 * synchronous replication, we'd better wake them up.
977 */
978 if (!sync_standbys_defined)
979 {
980 int i;
981
982 for (i = 0; i < NUM_SYNC_REP_WAIT_MODE; i++)
983 SyncRepWakeQueue(true, i);
984 }
985
986 /*
987 * Only allow people to join the queue when there are synchronous
988 * standbys defined. Without this interlock, there's a race
989 * condition: we might wake up all the current waiters; then, some
990 * backend that hasn't yet reloaded its config might go to sleep on
991 * the queue (and never wake up). This prevents that.
992 */
994 (sync_standbys_defined ? SYNC_STANDBY_DEFINED : 0);
995
996 LWLockRelease(SyncRepLock);
997 }
999 {
1000 LWLockAcquire(SyncRepLock, LW_EXCLUSIVE);
1001
1002 /*
1003 * Note that there is no need to wake up the queues here. We would
1004 * reach this path only if SyncStandbysDefined() returns false, or it
1005 * would mean that some backends are waiting with the GUC set. See
1006 * SyncRepWaitForLSN().
1007 */
1009
1010 /*
1011 * Even if there is no sync standby defined, let the readers of this
1012 * information know that the sync standby data has been initialized.
1013 * This can just be done once, hence the previous check on
1014 * SYNC_STANDBY_INIT to avoid useless work.
1015 */
1017
1018 LWLockRelease(SyncRepLock);
1019 }
1020}
Assert(PointerIsAligned(start, uint64))
bits8 sync_standbys_status
#define SyncStandbysDefined()
Definition: syncrep.c:92
#define NUM_SYNC_REP_WAIT_MODE
Definition: syncrep.h:27
#define SYNC_STANDBY_DEFINED
#define SYNC_STANDBY_INIT

References Assert(), i, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), NUM_SYNC_REP_WAIT_MODE, SYNC_STANDBY_DEFINED, SYNC_STANDBY_INIT, WalSndCtlData::sync_standbys_status, SyncRepWakeQueue(), SyncStandbysDefined, and WalSndCtl.

Referenced by UpdateSharedMemoryConfig().

◆ SyncRepWaitForLSN()

void SyncRepWaitForLSN ( XLogRecPtr  lsn,
bool  commit 
)

Definition at line 148 of file syncrep.c.

149{
150 int mode;
151
152 /*
153 * This should be called while holding interrupts during a transaction
154 * commit to prevent the follow-up shared memory queue cleanups to be
155 * influenced by external interruptions.
156 */
158
159 /*
160 * Fast exit if user has not requested sync replication, or there are no
161 * sync replication standby names defined.
162 *
163 * Since this routine gets called every commit time, it's important to
164 * exit quickly if sync replication is not requested.
165 *
166 * We check WalSndCtl->sync_standbys_status flag without the lock and exit
167 * immediately if SYNC_STANDBY_INIT is set (the checkpointer has
168 * initialized this data) but SYNC_STANDBY_DEFINED is missing (no sync
169 * replication requested).
170 *
171 * If SYNC_STANDBY_DEFINED is set, we need to check the status again later
172 * while holding the lock, to check the flag and operate the sync rep
173 * queue atomically. This is necessary to avoid the race condition
174 * described in SyncRepUpdateSyncStandbysDefined(). On the other hand, if
175 * SYNC_STANDBY_DEFINED is not set, the lock is not necessary because we
176 * don't touch the queue.
177 */
178 if (!SyncRepRequested() ||
179 ((((volatile WalSndCtlData *) WalSndCtl)->sync_standbys_status) &
181 return;
182
183 /* Cap the level for anything other than commit to remote flush only. */
184 if (commit)
186 else
188
190 Assert(WalSndCtl != NULL);
191
192 LWLockAcquire(SyncRepLock, LW_EXCLUSIVE);
194
195 /*
196 * We don't wait for sync rep if SYNC_STANDBY_DEFINED is not set. See
197 * SyncRepUpdateSyncStandbysDefined().
198 *
199 * Also check that the standby hasn't already replied. Unlikely race
200 * condition but we'll be fetching that cache line anyway so it's likely
201 * to be a low cost check.
202 *
203 * If the sync standby data has not been initialized yet
204 * (SYNC_STANDBY_INIT is not set), fall back to a check based on the LSN,
205 * then do a direct GUC check.
206 */
208 {
210 lsn <= WalSndCtl->lsn[mode])
211 {
212 LWLockRelease(SyncRepLock);
213 return;
214 }
215 }
216 else if (lsn <= WalSndCtl->lsn[mode])
217 {
218 /*
219 * The LSN is older than what we need to wait for. The sync standby
220 * data has not been initialized yet, but we are OK to not wait
221 * because we know that there is no point in doing so based on the
222 * LSN.
223 */
224 LWLockRelease(SyncRepLock);
225 return;
226 }
227 else if (!SyncStandbysDefined())
228 {
229 /*
230 * If we are here, the sync standby data has not been initialized yet,
231 * and the LSN is newer than what need to wait for, so we have fallen
232 * back to the best thing we could do in this case: a check on
233 * SyncStandbysDefined() to see if the GUC is set or not.
234 *
235 * When the GUC has a value, we wait until the checkpointer updates
236 * the status data because we cannot be sure yet if we should wait or
237 * not. Here, the GUC has *no* value, we are sure that there is no
238 * point to wait; this matters for example when initializing a
239 * cluster, where we should never wait, and no sync standbys is the
240 * default behavior.
241 */
242 LWLockRelease(SyncRepLock);
243 return;
244 }
245
246 /*
247 * Set our waitLSN so WALSender will know when to wake us, and add
248 * ourselves to the queue.
249 */
250 MyProc->waitLSN = lsn;
253 Assert(SyncRepQueueIsOrderedByLSN(mode));
254 LWLockRelease(SyncRepLock);
255
256 /* Alter ps display to show waiting for sync rep. */
258 {
259 char buffer[32];
260
261 sprintf(buffer, "waiting for %X/%08X", LSN_FORMAT_ARGS(lsn));
262 set_ps_display_suffix(buffer);
263 }
264
265 /*
266 * Wait for specified LSN to be confirmed.
267 *
268 * Each proc has its own wait latch, so we perform a normal latch
269 * check/wait loop here.
270 */
271 for (;;)
272 {
273 int rc;
274
275 /* Must reset the latch before testing state. */
277
278 /*
279 * Acquiring the lock is not needed, the latch ensures proper
280 * barriers. If it looks like we're done, we must really be done,
281 * because once walsender changes the state to SYNC_REP_WAIT_COMPLETE,
282 * it will never update it again, so we can't be seeing a stale value
283 * in that case.
284 */
286 break;
287
288 /*
289 * If a wait for synchronous replication is pending, we can neither
290 * acknowledge the commit nor raise ERROR or FATAL. The latter would
291 * lead the client to believe that the transaction aborted, which is
292 * not true: it's already committed locally. The former is no good
293 * either: the client has requested synchronous replication, and is
294 * entitled to assume that an acknowledged commit is also replicated,
295 * which might not be true. So in this case we issue a WARNING (which
296 * some clients may be able to interpret) and shut off further output.
297 * We do NOT reset ProcDiePending, so that the process will die after
298 * the commit is cleaned up.
299 */
300 if (ProcDiePending)
301 {
303 (errcode(ERRCODE_ADMIN_SHUTDOWN),
304 errmsg("canceling the wait for synchronous replication and terminating connection due to administrator command"),
305 errdetail("The transaction has already committed locally, but might not have been replicated to the standby.")));
308 break;
309 }
310
311 /*
312 * It's unclear what to do if a query cancel interrupt arrives. We
313 * can't actually abort at this point, but ignoring the interrupt
314 * altogether is not helpful, so we just terminate the wait with a
315 * suitable warning.
316 */
318 {
319 QueryCancelPending = false;
321 (errmsg("canceling wait for synchronous replication due to user request"),
322 errdetail("The transaction has already committed locally, but might not have been replicated to the standby.")));
324 break;
325 }
326
327 /*
328 * Wait on latch. Any condition that should wake us up will set the
329 * latch, so no need for timeout.
330 */
332 WAIT_EVENT_SYNC_REP);
333
334 /*
335 * If the postmaster dies, we'll probably never get an acknowledgment,
336 * because all the wal sender processes will exit. So just bail out.
337 */
338 if (rc & WL_POSTMASTER_DEATH)
339 {
340 ProcDiePending = true;
343 break;
344 }
345 }
346
347 /*
348 * WalSender has checked our LSN and has removed us from queue. Clean up
349 * state and leave. It's OK to reset these shared memory fields without
350 * holding SyncRepLock, because any walsenders will ignore us anyway when
351 * we're not on the queue. We need a read barrier to make sure we see the
352 * changes to the queue link (this might be unnecessary without
353 * assertions, but better safe than sorry).
354 */
358 MyProc->waitLSN = 0;
359
360 /* reset ps display to remove the suffix */
363}
#define pg_read_barrier()
Definition: atomics.h:154
#define Min(x, y)
Definition: c.h:1003
@ DestNone
Definition: dest.h:87
int errdetail(const char *fmt,...)
Definition: elog.c:1207
int errcode(int sqlerrcode)
Definition: elog.c:854
#define WARNING
Definition: elog.h:36
volatile uint32 InterruptHoldoffCount
Definition: globals.c:43
volatile sig_atomic_t QueryCancelPending
Definition: globals.c:33
struct Latch * MyLatch
Definition: globals.c:63
volatile sig_atomic_t ProcDiePending
Definition: globals.c:34
void ResetLatch(Latch *latch)
Definition: latch.c:374
int WaitLatch(Latch *latch, int wakeEvents, long timeout, uint32 wait_event_info)
Definition: latch.c:172
static PgChecksumMode mode
Definition: pg_checksums.c:55
#define sprintf
Definition: port.h:241
CommandDest whereToSendOutput
Definition: postgres.c:91
void set_ps_display_remove_suffix(void)
Definition: ps_status.c:439
void set_ps_display_suffix(const char *suffix)
Definition: ps_status.c:387
bool update_process_title
Definition: ps_status.c:31
XLogRecPtr waitLSN
Definition: proc.h:269
int syncRepState
Definition: proc.h:270
static int SyncRepWaitMode
Definition: syncrep.c:98
static void SyncRepQueueInsert(int mode)
Definition: syncrep.c:372
static void SyncRepCancelWait(void)
Definition: syncrep.c:406
#define SyncRepRequested()
Definition: syncrep.h:18
#define SYNC_REP_WAITING
Definition: syncrep.h:31
#define SYNC_REP_WAIT_COMPLETE
Definition: syncrep.h:32
#define SYNC_REP_NOT_WAITING
Definition: syncrep.h:30
#define WL_LATCH_SET
Definition: waiteventset.h:34
#define WL_POSTMASTER_DEATH
Definition: waiteventset.h:38

References Assert(), DestNone, dlist_node_is_detached(), ereport, errcode(), errdetail(), errmsg(), InterruptHoldoffCount, LSN_FORMAT_ARGS, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), Min, mode, MyLatch, MyProc, pg_read_barrier, ProcDiePending, QueryCancelPending, ResetLatch(), set_ps_display_remove_suffix(), set_ps_display_suffix(), sprintf, SYNC_REP_NOT_WAITING, SYNC_REP_WAIT_COMPLETE, SYNC_REP_WAIT_FLUSH, SYNC_REP_WAITING, SYNC_STANDBY_DEFINED, SYNC_STANDBY_INIT, WalSndCtlData::sync_standbys_status, SyncRepCancelWait(), PGPROC::syncRepLinks, SyncRepQueueInsert(), SyncRepRequested, PGPROC::syncRepState, SyncRepWaitMode, SyncStandbysDefined, update_process_title, WaitLatch(), PGPROC::waitLSN, WalSndCtl, WARNING, whereToSendOutput, WL_LATCH_SET, and WL_POSTMASTER_DEATH.

Referenced by EndPrepare(), RecordTransactionAbortPrepared(), RecordTransactionCommit(), and RecordTransactionCommitPrepared().

Variable Documentation

◆ SyncRepConfig

PGDLLIMPORT SyncRepConfigData* SyncRepConfig
extern

Definition at line 97 of file syncrep.c.

Referenced by assign_synchronous_standby_names(), pg_stat_get_wal_senders(), SyncRepGetCandidateStandbys(), SyncRepGetStandbyPriority(), SyncRepGetSyncRecPtr(), and SyncRepReleaseWaiters().

◆ SyncRepStandbyNames

PGDLLIMPORT char* SyncRepStandbyNames
extern

Definition at line 90 of file syncrep.c.

AltStyle によって変換されたページ (->オリジナル) /