PostgreSQL Source Code git master
Loading...
Searching...
No Matches
syncrep.c File Reference
#include "postgres.h"
#include <unistd.h>
#include "access/xact.h"
#include "common/int.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "replication/syncrep.h"
#include "replication/walsender.h"
#include "replication/walsender_private.h"
#include "storage/proc.h"
#include "tcop/tcopprot.h"
#include "utils/guc_hooks.h"
#include "utils/ps_status.h"
#include "utils/wait_event.h"
Include dependency graph for syncrep.c:

Go to the source code of this file.

Macros

#define SyncStandbysDefined()    (SyncRepStandbyNames != NULL && SyncRepStandbyNames[0] != '\0')
 

Functions

static void SyncRepQueueInsert (int mode)
 
static void SyncRepCancelWait (void)
 
static int SyncRepWakeQueue (bool all, int mode)
 
static bool SyncRepGetSyncRecPtr (XLogRecPtr *writePtr, XLogRecPtr *flushPtr, XLogRecPtr *applyPtr, bool *am_sync)
 
static void SyncRepGetOldestSyncRecPtr (XLogRecPtr *writePtr, XLogRecPtr *flushPtr, XLogRecPtr *applyPtr, SyncRepStandbyData *sync_standbys, int num_standbys)
 
static void SyncRepGetNthLatestSyncRecPtr (XLogRecPtr *writePtr, XLogRecPtr *flushPtr, XLogRecPtr *applyPtr, SyncRepStandbyData *sync_standbys, int num_standbys, uint8 nth)
 
static int SyncRepGetStandbyPriority (void)
 
static int standby_priority_comparator (const void *a, const void *b)
 
static int cmp_lsn (const void *a, const void *b)
 
void SyncRepWaitForLSN (XLogRecPtr lsn, bool commit)
 
void SyncRepCleanupAtProcExit (void)
 
void SyncRepInitConfig (void)
 
void SyncRepReleaseWaiters (void)
 
int SyncRepGetCandidateStandbys (SyncRepStandbyData **standbys)
 
void SyncRepUpdateSyncStandbysDefined (void)
 
bool check_synchronous_standby_names (char **newval, void **extra, GucSource source)
 
void assign_synchronous_standby_names (const char *newval, void *extra)
 
void assign_synchronous_commit (int newval, void *extra)
 

Variables

charSyncRepStandbyNames
 
static bool announce_next_takeover = true
 
SyncRepConfigDataSyncRepConfig = NULL
 
static int SyncRepWaitMode = SYNC_REP_NO_WAIT
 

Macro Definition Documentation

◆ SyncStandbysDefined

#define SyncStandbysDefined ( )     (SyncRepStandbyNames != NULL && SyncRepStandbyNames[0] != '\0')

Definition at line 93 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
191
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 {
211 {
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 */
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 */
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;
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 {
302 if (ProcDieSenderPid != 0)
305 errmsg("canceling the wait for synchronous replication and terminating connection due to administrator command"),
306 errdetail("The transaction has already committed locally, but might not have been replicated to the standby."),
307 errdetail_log("The transaction has already committed locally, but might not have been replicated to the standby. Signal sent by PID %d, UID %d.",
308 (int) ProcDieSenderPid,
309 (int) ProcDieSenderUid)));
310 else
313 errmsg("canceling the wait for synchronous replication and terminating connection due to administrator command"),
314 errdetail("The transaction has already committed locally, but might not have been replicated to the standby.")));
317 break;
318 }
319
320 /*
321 * It's unclear what to do if a query cancel interrupt arrives. We
322 * can't actually abort at this point, but ignoring the interrupt
323 * altogether is not helpful, so we just terminate the wait with a
324 * suitable warning.
325 */
327 {
328 QueryCancelPending = false;
330 (errmsg("canceling wait for synchronous replication due to user request"),
331 errdetail("The transaction has already committed locally, but might not have been replicated to the standby.")));
333 break;
334 }
335
336 /*
337 * Wait on latch. Any condition that should wake us up will set the
338 * latch, so no need for timeout.
339 */
342
343 /*
344 * If the postmaster dies, we'll probably never get an acknowledgment,
345 * because all the wal sender processes will exit. So just bail out.
346 */
347 if (rc & WL_POSTMASTER_DEATH)
348 {
349 ProcDiePending = true;
352 break;
353 }
354 }
355
356 /*
357 * WalSender has checked our LSN and has removed us from queue. Clean up
358 * state and leave. It's OK to reset these shared memory fields without
359 * holding SyncRepLock, because any walsenders will ignore us anyway when
360 * we're not on the queue. We need a read barrier to make sure we see the
361 * changes to the queue link (this might be unnecessary without
362 * assertions, but better safe than sorry).
363 */
368
369 /* reset ps display to remove the suffix */
372}
373
374/*
375 * Insert MyProc into the specified SyncRepQueue, maintaining sorted invariant.
376 *
377 * Usually we will go at tail of queue, though it's possible that we arrive
378 * here out of order, so start at tail and work back to insertion point.
379 */
380static void
382{
383 dlist_head *queue;
384 dlist_iter iter;
385
387 queue = &WalSndCtl->SyncRepQueue[mode];
388
389 dlist_reverse_foreach(iter, queue)
390 {
391 PGPROC *proc = dlist_container(PGPROC, syncRepLinks, iter.cur);
392
393 /*
394 * Stop at the queue element that we should insert after to ensure the
395 * queue is ordered by LSN.
396 */
397 if (proc->waitLSN < MyProc->waitLSN)
398 {
400 return;
401 }
402 }
403
404 /*
405 * If we get here, the list was either empty, or this process needs to be
406 * at the head.
407 */
409}
410
411/*
412 * Acquire SyncRepLock and cancel any wait currently in progress.
413 */
414static void
416{
422}
423
424void
426{
427 /*
428 * First check if we are removed from the queue without the lock to not
429 * slow down backend exit.
430 */
432 {
434
435 /* maybe we have just been removed, so recheck */
438
440 }
441}
442
443/*
444 * ===========================================================
445 * Synchronous Replication functions for wal sender processes
446 * ===========================================================
447 */
448
449/*
450 * Take any action required to initialise sync rep state from config
451 * data. Called at WALSender startup and after each SIGHUP.
452 */
453void
455{
456 int priority;
457
458 /*
459 * Determine if we are a potential sync standby and remember the result
460 * for handling replies from standby.
461 */
464 {
468
470 (errmsg_internal("standby \"%s\" now has synchronous standby priority %d",
472 }
473}
474
475/*
476 * Update the LSNs on each queue based upon our latest state. This
477 * implements a simple policy of first-valid-sync-standby-releases-waiter.
478 *
479 * Other policies are possible, which would change what we do here and
480 * perhaps also which information we store as well.
481 */
482void
484{
488 bool got_recptr;
489 bool am_sync;
490 int numwrite = 0;
491 int numflush = 0;
492 int numapply = 0;
493
494 /*
495 * If this WALSender is serving a standby that is not on the list of
496 * potential sync standbys then we have nothing to do. If we are still
497 * starting up, still running base backup or the current flush position is
498 * still invalid, then leave quickly also. Streaming or stopping WAL
499 * senders are allowed to release waiters.
500 */
505 {
507 return;
508 }
509
510 /*
511 * We're a potential sync standby. Release waiters if there are enough
512 * sync standbys and we are considered as sync.
513 */
515
516 /*
517 * Check whether we are a sync standby or not, and calculate the synced
518 * positions among all sync standbys. (Note: although this step does not
519 * of itself require holding SyncRepLock, it seems like a good idea to do
520 * it after acquiring the lock. This ensures that the WAL pointers we use
521 * to release waiters are newer than any previous execution of this
522 * routine used.)
523 */
525
526 /*
527 * If we are managing a sync standby, though we weren't prior to this,
528 * then announce we are now a sync standby.
529 */
531 {
533
535 ereport(LOG,
536 (errmsg("standby \"%s\" is now a synchronous standby with priority %d",
538 else
539 ereport(LOG,
540 (errmsg("standby \"%s\" is now a candidate for quorum synchronous standby",
542 }
543
544 /*
545 * If the number of sync standbys is less than requested or we aren't
546 * managing a sync standby then just leave.
547 */
548 if (!got_recptr || !am_sync)
549 {
552 return;
553 }
554
555 /*
556 * Set the lsn first so that when we wake backends they will release up to
557 * this location.
558 */
560 {
563 }
565 {
568 }
570 {
573 }
574
576
577 elog(DEBUG3, "released %d procs up to write %X/%08X, %d procs up to flush %X/%08X, %d procs up to apply %X/%08X",
581}
582
583/*
584 * Calculate the synced Write, Flush and Apply positions among sync standbys.
585 *
586 * Return false if the number of sync standbys is less than
587 * synchronous_standby_names specifies. Otherwise return true and
588 * store the positions into *writePtr, *flushPtr and *applyPtr.
589 *
590 * On return, *am_sync is set to true if this walsender is connecting to
591 * sync standby. Otherwise it's set to false.
592 */
593static bool
596{
598 int num_standbys;
599 int i;
600
601 /* Initialize default results */
605 *am_sync = false;
606
607 /* Quick out if not even configured to be synchronous */
608 if (SyncRepConfig == NULL)
609 return false;
610
611 /* Get standbys that are considered as synchronous at this moment */
613
614 /* Am I among the candidate sync standbys? */
615 for (i = 0; i < num_standbys; i++)
616 {
617 if (sync_standbys[i].is_me)
618 {
619 *am_sync = true;
620 break;
621 }
622 }
623
624 /*
625 * Nothing more to do if we are not managing a sync standby or there are
626 * not enough synchronous standbys.
627 */
628 if (!(*am_sync) ||
630 {
632 return false;
633 }
634
635 /*
636 * In a priority-based sync replication, the synced positions are the
637 * oldest ones among sync standbys. In a quorum-based, they are the Nth
638 * latest ones.
639 *
640 * SyncRepGetNthLatestSyncRecPtr() also can calculate the oldest
641 * positions. But we use SyncRepGetOldestSyncRecPtr() for that calculation
642 * because it's a bit more efficient.
643 *
644 * XXX If the numbers of current and requested sync standbys are the same,
645 * we can use SyncRepGetOldestSyncRecPtr() to calculate the synced
646 * positions even in a quorum-based sync replication.
647 */
649 {
652 }
653 else
654 {
658 }
659
661 return true;
662}
663
664/*
665 * Calculate the oldest Write, Flush and Apply positions among sync standbys.
666 */
667static void
672 int num_standbys)
673{
674 int i;
675
676 /*
677 * Scan through all sync standbys and calculate the oldest Write, Flush
678 * and Apply positions. We assume *writePtr et al were initialized to
679 * InvalidXLogRecPtr.
680 */
681 for (i = 0; i < num_standbys; i++)
682 {
684 XLogRecPtr flush = sync_standbys[i].flush;
685 XLogRecPtr apply = sync_standbys[i].apply;
686
688 *writePtr = write;
689 if (!XLogRecPtrIsValid(*flushPtr) || *flushPtr > flush)
690 *flushPtr = flush;
691 if (!XLogRecPtrIsValid(*applyPtr) || *applyPtr > apply)
692 *applyPtr = apply;
693 }
694}
695
696/*
697 * Calculate the Nth latest Write, Flush and Apply positions among sync
698 * standbys.
699 */
700static void
705 int num_standbys,
706 uint8 nth)
707{
711 int i;
712
713 /* Should have enough candidates, or somebody messed up */
714 Assert(nth > 0 && nth <= num_standbys);
715
719
720 for (i = 0; i < num_standbys; i++)
721 {
722 write_array[i] = sync_standbys[i].write;
723 flush_array[i] = sync_standbys[i].flush;
724 apply_array[i] = sync_standbys[i].apply;
725 }
726
727 /* Sort each array in descending order */
731
732 /* Get Nth latest Write, Flush, Apply positions */
733 *writePtr = write_array[nth - 1];
734 *flushPtr = flush_array[nth - 1];
735 *applyPtr = apply_array[nth - 1];
736
740}
741
742/*
743 * Compare lsn in order to sort array in descending order.
744 */
745static int
746cmp_lsn(const void *a, const void *b)
747{
748 XLogRecPtr lsn1 = *((const XLogRecPtr *) a);
749 XLogRecPtr lsn2 = *((const XLogRecPtr *) b);
750
751 return pg_cmp_u64(lsn2, lsn1);
752}
753
754/*
755 * Return data about walsenders that are candidates to be sync standbys.
756 *
757 * *standbys is set to a palloc'd array of structs of per-walsender data,
758 * and the number of valid entries (candidate sync senders) is returned.
759 * (This might be more or fewer than num_sync; caller must check.)
760 */
761int
763{
764 int i;
765 int n;
766
767 /* Create result array */
769
770 /* Quick exit if sync replication is not requested */
771 if (SyncRepConfig == NULL)
772 return 0;
773
774 /* Collect raw data from shared memory */
775 n = 0;
776 for (i = 0; i < max_wal_senders; i++)
777 {
778 WalSnd *walsnd;
780 WalSndState state; /* not included in SyncRepStandbyData */
781
783 stby = *standbys + n;
784
785 SpinLockAcquire(&walsnd->mutex);
786 stby->pid = walsnd->pid;
787 state = walsnd->state;
788 stby->write = walsnd->write;
789 stby->flush = walsnd->flush;
790 stby->apply = walsnd->apply;
791 stby->sync_standby_priority = walsnd->sync_standby_priority;
792 SpinLockRelease(&walsnd->mutex);
793
794 /* Must be active */
795 if (stby->pid == 0)
796 continue;
797
798 /* Must be streaming or stopping */
801 continue;
802
803 /* Must be synchronous */
804 if (stby->sync_standby_priority == 0)
805 continue;
806
807 /* Must have a valid flush position */
808 if (!XLogRecPtrIsValid(stby->flush))
809 continue;
810
811 /* OK, it's a candidate */
812 stby->walsnd_index = i;
813 stby->is_me = (walsnd == MyWalSnd);
814 n++;
815 }
816
817 /*
818 * In quorum mode, we return all the candidates. In priority mode, if we
819 * have too many candidates then return only the num_sync ones of highest
820 * priority.
821 */
824 {
825 /* Sort by priority ... */
826 qsort(*standbys, n, sizeof(SyncRepStandbyData),
828 /* ... then report just the first num_sync ones */
830 }
831
832 return n;
833}
834
835/*
836 * qsort comparator to sort SyncRepStandbyData entries by priority
837 */
838static int
839standby_priority_comparator(const void *a, const void *b)
840{
841 const SyncRepStandbyData *sa = (const SyncRepStandbyData *) a;
842 const SyncRepStandbyData *sb = (const SyncRepStandbyData *) b;
843
844 /* First, sort by increasing priority value */
845 if (sa->sync_standby_priority != sb->sync_standby_priority)
846 return sa->sync_standby_priority - sb->sync_standby_priority;
847
848 /*
849 * We might have equal priority values; arbitrarily break ties by position
850 * in the WalSnd array. (This is utterly bogus, since that is arrival
851 * order dependent, but there are regression tests that rely on it.)
852 */
853 return sa->walsnd_index - sb->walsnd_index;
854}
855
856
857/*
858 * Check if we are in the list of sync standbys, and if so, determine
859 * priority sequence. Return priority if set, or zero to indicate that
860 * we are not a potential sync standby.
861 *
862 * Compare the parameter SyncRepStandbyNames against the application_name
863 * for this WALSender, or allow any name if we find a wildcard "*".
864 */
865static int
867{
868 const char *standby_name;
869 int priority;
870 bool found = false;
871
872 /*
873 * Since synchronous cascade replication is not allowed, we always set the
874 * priority of cascading walsender to zero.
875 */
877 return 0;
878
880 return 0;
881
884 {
886 strcmp(standby_name, "*") == 0)
887 {
888 found = true;
889 break;
890 }
892 }
893
894 if (!found)
895 return 0;
896
897 /*
898 * In quorum-based sync replication, all the standbys in the list have the
899 * same priority, one.
900 */
902}
903
904/*
905 * Walk the specified queue from head. Set the state of any backends that
906 * need to be woken, remove them from the queue, and then wake them.
907 * Pass all = true to wake whole queue; otherwise, just wake up to
908 * the walsender's LSN.
909 *
910 * The caller must hold SyncRepLock in exclusive mode.
911 */
912static int
913SyncRepWakeQueue(bool all, int mode)
914{
915 int numprocs = 0;
917
921
923 {
924 PGPROC *proc = dlist_container(PGPROC, syncRepLinks, iter.cur);
925
926 /*
927 * Assume the queue is ordered by LSN
928 */
929 if (!all && WalSndCtl->lsn[mode] < proc->waitLSN)
930 return numprocs;
931
932 /*
933 * Remove from queue.
934 */
936
937 /*
938 * SyncRepWaitForLSN() reads syncRepState without holding the lock, so
939 * make sure that it sees the queue link being removed before the
940 * syncRepState change.
941 */
943
944 /*
945 * Set state to complete; see SyncRepWaitForLSN() for discussion of
946 * the various states.
947 */
949
950 /*
951 * Wake only when we have set state and removed from queue.
952 */
953 SetLatch(&(proc->procLatch));
954
955 numprocs++;
956 }
957
958 return numprocs;
959}
960
961/*
962 * The checkpointer calls this as needed to update the shared
963 * sync_standbys_status flag, so that backends don't remain permanently wedged
964 * if synchronous_standby_names is unset. It's safe to check the current value
965 * without the lock, because it's only ever updated by one process. But we
966 * must take the lock to change it.
967 */
968void
970{
972
975 {
977
978 /*
979 * If synchronous_standby_names has been reset to empty, it's futile
980 * for backends to continue waiting. Since the user no longer wants
981 * synchronous replication, we'd better wake them up.
982 */
984 {
985 int i;
986
987 for (i = 0; i < NUM_SYNC_REP_WAIT_MODE; i++)
988 SyncRepWakeQueue(true, i);
989 }
990
991 /*
992 * Only allow people to join the queue when there are synchronous
993 * standbys defined. Without this interlock, there's a race
994 * condition: we might wake up all the current waiters; then, some
995 * backend that hasn't yet reloaded its config might go to sleep on
996 * the queue (and never wake up). This prevents that.
997 */
1000
1002 }
1004 {
1006
1007 /*
1008 * Note that there is no need to wake up the queues here. We would
1009 * reach this path only if SyncStandbysDefined() returns false, or it
1010 * would mean that some backends are waiting with the GUC set. See
1011 * SyncRepWaitForLSN().
1012 */
1014
1015 /*
1016 * Even if there is no sync standby defined, let the readers of this
1017 * information know that the sync standby data has been initialized.
1018 * This can just be done once, hence the previous check on
1019 * SYNC_STANDBY_INIT to avoid useless work.
1020 */
1022
1024 }
1025}
1026
1027#ifdef USE_ASSERT_CHECKING
1028static bool
1030{
1032 dlist_iter iter;
1033
1035
1037
1039 {
1040 PGPROC *proc = dlist_container(PGPROC, syncRepLinks, iter.cur);
1041
1042 /*
1043 * Check the queue is ordered by LSN and that multiple procs don't
1044 * have matching LSNs
1045 */
1046 if (proc->waitLSN <= lastLSN)
1047 return false;
1048
1049 lastLSN = proc->waitLSN;
1050 }
1051
1052 return true;
1053}
1054#endif
1055
1056/*
1057 * ===========================================================
1058 * Synchronous Replication functions executed by any process
1059 * ===========================================================
1060 */
1061
1062bool
1064{
1065 if (*newval != NULL && (*newval)[0] != '\0')
1066 {
1067 yyscan_t scanner;
1068 int parse_rc;
1070
1071 /* Result of parsing is returned in one of these two variables */
1074
1075 /* Parse the synchronous_standby_names string */
1076 syncrep_scanner_init(*newval, &scanner);
1078 syncrep_scanner_finish(scanner);
1079
1080 if (parse_rc != 0 || syncrep_parse_result == NULL)
1081 {
1085 else
1086 /* translator: %s is a GUC name */
1087 GUC_check_errdetail("\"%s\" parser failed.",
1088 "synchronous_standby_names");
1089 return false;
1090 }
1091
1092 if (syncrep_parse_result->num_sync <= 0)
1093 {
1094 GUC_check_errmsg("number of synchronous standbys (%d) must be greater than zero",
1095 syncrep_parse_result->num_sync);
1096 return false;
1097 }
1098
1099 /* GUC extra value must be guc_malloc'd, not palloc'd */
1101 guc_malloc(LOG, syncrep_parse_result->config_size);
1102 if (pconf == NULL)
1103 return false;
1105
1106 *extra = pconf;
1107
1108 /*
1109 * We need not explicitly clean up syncrep_parse_result. It, and any
1110 * other cruft generated during parsing, will be freed when the
1111 * current memory context is deleted. (This code is generally run in
1112 * a short-lived context used for config file processing, so that will
1113 * not be very long.)
1114 */
1115 }
1116 else
1117 *extra = NULL;
1118
1119 return true;
1120}
1121
1122void
1123assign_synchronous_standby_names(const char *newval, void *extra)
1124{
1125 SyncRepConfig = (SyncRepConfigData *) extra;
1126}
1127
1128void
1129assign_synchronous_commit(int newval, void *extra)
1130{
1131 switch (newval)
1132 {
1135 break;
1138 break;
1141 break;
1142 default:
1144 break;
1145 }
1146}
#define pg_read_barrier()
Definition atomics.h:154
#define pg_write_barrier()
Definition atomics.h:155
#define Min(x, y)
Definition c.h:1131
uint8_t uint8
Definition c.h:681
#define Assert(condition)
Definition c.h:1002
memcpy(sums, checksumBaseOffsets, sizeof(checksumBaseOffsets))
void * yyscan_t
Definition cubedata.h:65
@ DestNone
Definition dest.h:87
int errcode(int sqlerrcode)
Definition elog.c:875
#define LOG
Definition elog.h:32
#define DEBUG3
Definition elog.h:29
int errdetail(const char *fmt,...) pg_attribute_printf(1
int int errmsg_internal(const char *fmt,...) pg_attribute_printf(1
#define WARNING
Definition elog.h:37
#define DEBUG1
Definition elog.h:31
int int int errdetail_log(const char *fmt,...) pg_attribute_printf(1
#define elog(elevel,...)
Definition elog.h:228
#define ereport(elevel,...)
Definition elog.h:152
#define palloc_array(type, count)
Definition fe_memutils.h:91
volatile int ProcDieSenderPid
Definition globals.c:46
volatile uint32 InterruptHoldoffCount
Definition globals.c:43
volatile int ProcDieSenderUid
Definition globals.c:47
volatile sig_atomic_t QueryCancelPending
Definition globals.c:33
struct Latch * MyLatch
Definition globals.c:65
volatile sig_atomic_t ProcDiePending
Definition globals.c:34
void GUC_check_errcode(int sqlerrcode)
Definition guc.c:6666
void * guc_malloc(int elevel, size_t size)
Definition guc.c:637
#define newval
#define GUC_check_errmsg
Definition guc.h:504
#define GUC_check_errdetail
Definition guc.h:508
GucSource
Definition guc.h:112
char * application_name
Definition guc_tables.c:590
static void dlist_insert_after(dlist_node *after, dlist_node *node)
Definition ilist.h:381
#define dlist_foreach(iter, lhead)
Definition ilist.h:623
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
#define dlist_reverse_foreach(iter, lhead)
Definition ilist.h:654
static void dlist_push_head(dlist_head *head, dlist_node *node)
Definition ilist.h:347
#define dlist_foreach_modify(iter, lhead)
Definition ilist.h:640
#define dlist_container(type, membername, ptr)
Definition ilist.h:593
static int pg_cmp_u64(uint64 a, uint64 b)
Definition int.h:731
#define write(a, b, c)
Definition win32.h:14
int b
Definition isn.c:74
int a
Definition isn.c:73
int i
Definition isn.c:77
void SetLatch(Latch *latch)
Definition latch.c:290
void ResetLatch(Latch *latch)
Definition latch.c:374
int WaitLatch(Latch *latch, int wakeEvents, long timeout, uint32 wait_event_info)
Definition latch.c:172
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition lwlock.c:1150
bool LWLockHeldByMeInMode(LWLock *lock, LWLockMode mode)
Definition lwlock.c:1929
void LWLockRelease(LWLock *lock)
Definition lwlock.c:1767
@ LW_EXCLUSIVE
Definition lwlock.h:104
void pfree(void *pointer)
Definition mcxt.c:1619
static char * errmsg
static PgChecksumMode mode
static rewind_source * source
Definition pg_rewind.c:89
int pg_strcasecmp(const char *s1, const char *s2)
#define sprintf
Definition port.h:263
#define qsort(a, b, c, d)
Definition port.h:496
CommandDest whereToSendOutput
Definition postgres.c:97
static int fb(int x)
void set_ps_display_remove_suffix(void)
Definition ps_status.c:440
void set_ps_display_suffix(const char *suffix)
Definition ps_status.c:388
bool update_process_title
Definition ps_status.c:31
static void SpinLockRelease(volatile slock_t *lock)
Definition spin.h:62
static void SpinLockAcquire(volatile slock_t *lock)
Definition spin.h:56
PGPROC * MyProc
Definition proc.c:71
Definition proc.h:179
XLogRecPtr waitLSN
Definition proc.h:341
dlist_node syncRepLinks
Definition proc.h:343
int syncRepState
Definition proc.h:342
Latch procLatch
Definition proc.h:256
uint8 syncrep_method
Definition syncrep.h:68
char member_names[FLEXIBLE_ARRAY_MEMBER]
Definition syncrep.h:71
XLogRecPtr lsn[NUM_SYNC_REP_WAIT_MODE]
WalSnd walsnds[FLEXIBLE_ARRAY_MEMBER]
dlist_head SyncRepQueue[NUM_SYNC_REP_WAIT_MODE]
slock_t mutex
XLogRecPtr flush
WalSndState state
int sync_standby_priority
dlist_node * cur
Definition ilist.h:179
dlist_node * cur
Definition ilist.h:200
static int SyncRepWaitMode
Definition syncrep.c:99
void SyncRepInitConfig(void)
Definition syncrep.c:455
static bool SyncRepGetSyncRecPtr(XLogRecPtr *writePtr, XLogRecPtr *flushPtr, XLogRecPtr *applyPtr, bool *am_sync)
Definition syncrep.c:595
static void SyncRepGetNthLatestSyncRecPtr(XLogRecPtr *writePtr, XLogRecPtr *flushPtr, XLogRecPtr *applyPtr, SyncRepStandbyData *sync_standbys, int num_standbys, uint8 nth)
Definition syncrep.c:702
void assign_synchronous_commit(int newval, void *extra)
Definition syncrep.c:1130
void assign_synchronous_standby_names(const char *newval, void *extra)
Definition syncrep.c:1124
static int standby_priority_comparator(const void *a, const void *b)
Definition syncrep.c:840
static int SyncRepWakeQueue(bool all, int mode)
Definition syncrep.c:914
SyncRepConfigData * SyncRepConfig
Definition syncrep.c:98
int SyncRepGetCandidateStandbys(SyncRepStandbyData **standbys)
Definition syncrep.c:763
void SyncRepReleaseWaiters(void)
Definition syncrep.c:484
void SyncRepUpdateSyncStandbysDefined(void)
Definition syncrep.c:970
static bool announce_next_takeover
Definition syncrep.c:96
static int SyncRepGetStandbyPriority(void)
Definition syncrep.c:867
static void SyncRepQueueInsert(int mode)
Definition syncrep.c:382
static void SyncRepCancelWait(void)
Definition syncrep.c:416
bool check_synchronous_standby_names(char **newval, void **extra, GucSource source)
Definition syncrep.c:1064
static void SyncRepGetOldestSyncRecPtr(XLogRecPtr *writePtr, XLogRecPtr *flushPtr, XLogRecPtr *applyPtr, SyncRepStandbyData *sync_standbys, int num_standbys)
Definition syncrep.c:669
void SyncRepCleanupAtProcExit(void)
Definition syncrep.c:426
static int cmp_lsn(const void *a, const void *b)
Definition syncrep.c:747
#define SyncStandbysDefined()
Definition syncrep.c:93
#define SYNC_REP_PRIORITY
Definition syncrep.h:35
#define NUM_SYNC_REP_WAIT_MODE
Definition syncrep.h:27
#define SyncRepRequested()
Definition syncrep.h:18
#define SYNC_REP_NO_WAIT
Definition syncrep.h:22
#define SYNC_REP_WAIT_WRITE
Definition syncrep.h:23
#define SYNC_REP_WAITING
Definition syncrep.h:31
#define SYNC_REP_WAIT_COMPLETE
Definition syncrep.h:32
#define SYNC_REP_WAIT_FLUSH
Definition syncrep.h:24
#define SYNC_REP_NOT_WAITING
Definition syncrep.h:30
int syncrep_yyparse(SyncRepConfigData **syncrep_parse_result_p, char **syncrep_parse_error_msg_p, yyscan_t yyscanner)
#define SYNC_REP_WAIT_APPLY
Definition syncrep.h:25
void syncrep_scanner_finish(yyscan_t yyscanner)
void syncrep_scanner_init(const char *str, yyscan_t *yyscannerp)
#define WL_LATCH_SET
#define WL_POSTMASTER_DEATH
WalSnd * MyWalSnd
Definition walsender.c:132
int max_wal_senders
Definition walsender.c:141
bool am_cascading_walsender
Definition walsender.c:136
WalSndCtlData * WalSndCtl
Definition walsender.c:121
#define SYNC_STANDBY_DEFINED
WalSndState
@ WALSNDSTATE_STREAMING
@ WALSNDSTATE_STOPPING
#define SYNC_STANDBY_INIT
@ SYNCHRONOUS_COMMIT_REMOTE_WRITE
Definition xact.h:73
@ SYNCHRONOUS_COMMIT_REMOTE_APPLY
Definition xact.h:76
@ SYNCHRONOUS_COMMIT_REMOTE_FLUSH
Definition xact.h:75
#define XLogRecPtrIsValid(r)
Definition xlogdefs.h:29
#define LSN_FORMAT_ARGS(lsn)
Definition xlogdefs.h:47
uint64 XLogRecPtr
Definition xlogdefs.h:21
#define InvalidXLogRecPtr
Definition xlogdefs.h:28

Function Documentation

◆ assign_synchronous_commit()

void assign_synchronous_commit ( int  newval,
void extra 
)

◆ assign_synchronous_standby_names()

void assign_synchronous_standby_names ( const char newval,
void extra 
)

Definition at line 1124 of file syncrep.c.

1125{
1126 SyncRepConfig = (SyncRepConfigData *) extra;
1127}

References SyncRepConfig.

◆ check_synchronous_standby_names()

bool check_synchronous_standby_names ( char **  newval,
void **  extra,
GucSource  source 
)

Definition at line 1064 of file syncrep.c.

1065{
1066 if (*newval != NULL && (*newval)[0] != '\0')
1067 {
1068 yyscan_t scanner;
1069 int parse_rc;
1071
1072 /* Result of parsing is returned in one of these two variables */
1075
1076 /* Parse the synchronous_standby_names string */
1077 syncrep_scanner_init(*newval, &scanner);
1079 syncrep_scanner_finish(scanner);
1080
1081 if (parse_rc != 0 || syncrep_parse_result == NULL)
1082 {
1086 else
1087 /* translator: %s is a GUC name */
1088 GUC_check_errdetail("\"%s\" parser failed.",
1089 "synchronous_standby_names");
1090 return false;
1091 }
1092
1093 if (syncrep_parse_result->num_sync <= 0)
1094 {
1095 GUC_check_errmsg("number of synchronous standbys (%d) must be greater than zero",
1096 syncrep_parse_result->num_sync);
1097 return false;
1098 }
1099
1100 /* GUC extra value must be guc_malloc'd, not palloc'd */
1102 guc_malloc(LOG, syncrep_parse_result->config_size);
1103 if (pconf == NULL)
1104 return false;
1106
1107 *extra = pconf;
1108
1109 /*
1110 * We need not explicitly clean up syncrep_parse_result. It, and any
1111 * other cruft generated during parsing, will be freed when the
1112 * current memory context is deleted. (This code is generally run in
1113 * a short-lived context used for config file processing, so that will
1114 * not be very long.)
1115 */
1116 }
1117 else
1118 *extra = NULL;
1119
1120 return true;
1121}

References fb(), GUC_check_errcode(), GUC_check_errdetail, GUC_check_errmsg, guc_malloc(), LOG, memcpy(), newval, syncrep_scanner_finish(), syncrep_scanner_init(), and syncrep_yyparse().

◆ cmp_lsn()

static int cmp_lsn ( const void a,
const void b 
)
static

Definition at line 747 of file syncrep.c.

748{
749 XLogRecPtr lsn1 = *((const XLogRecPtr *) a);
750 XLogRecPtr lsn2 = *((const XLogRecPtr *) b);
751
752 return pg_cmp_u64(lsn2, lsn1);
753}

References a, b, fb(), and pg_cmp_u64().

Referenced by SyncRepGetNthLatestSyncRecPtr().

◆ standby_priority_comparator()

static int standby_priority_comparator ( const void a,
const void b 
)
static

Definition at line 840 of file syncrep.c.

841{
842 const SyncRepStandbyData *sa = (const SyncRepStandbyData *) a;
843 const SyncRepStandbyData *sb = (const SyncRepStandbyData *) b;
844
845 /* First, sort by increasing priority value */
846 if (sa->sync_standby_priority != sb->sync_standby_priority)
847 return sa->sync_standby_priority - sb->sync_standby_priority;
848
849 /*
850 * We might have equal priority values; arbitrarily break ties by position
851 * in the WalSnd array. (This is utterly bogus, since that is arrival
852 * order dependent, but there are regression tests that rely on it.)
853 */
854 return sa->walsnd_index - sb->walsnd_index;
855}

References a, b, and fb().

Referenced by SyncRepGetCandidateStandbys().

◆ SyncRepCancelWait()

◆ SyncRepCleanupAtProcExit()

void SyncRepCleanupAtProcExit ( void  )

Definition at line 426 of file syncrep.c.

427{
428 /*
429 * First check if we are removed from the queue without the lock to not
430 * slow down backend exit.
431 */
433 {
435
436 /* maybe we have just been removed, so recheck */
439
441 }
442}

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

Referenced by ProcKill().

◆ SyncRepGetCandidateStandbys()

int SyncRepGetCandidateStandbys ( SyncRepStandbyData **  standbys)

Definition at line 763 of file syncrep.c.

764{
765 int i;
766 int n;
767
768 /* Create result array */
770
771 /* Quick exit if sync replication is not requested */
772 if (SyncRepConfig == NULL)
773 return 0;
774
775 /* Collect raw data from shared memory */
776 n = 0;
777 for (i = 0; i < max_wal_senders; i++)
778 {
779 WalSnd *walsnd;
781 WalSndState state; /* not included in SyncRepStandbyData */
782
784 stby = *standbys + n;
785
786 SpinLockAcquire(&walsnd->mutex);
787 stby->pid = walsnd->pid;
788 state = walsnd->state;
789 stby->write = walsnd->write;
790 stby->flush = walsnd->flush;
791 stby->apply = walsnd->apply;
792 stby->sync_standby_priority = walsnd->sync_standby_priority;
793 SpinLockRelease(&walsnd->mutex);
794
795 /* Must be active */
796 if (stby->pid == 0)
797 continue;
798
799 /* Must be streaming or stopping */
802 continue;
803
804 /* Must be synchronous */
805 if (stby->sync_standby_priority == 0)
806 continue;
807
808 /* Must have a valid flush position */
809 if (!XLogRecPtrIsValid(stby->flush))
810 continue;
811
812 /* OK, it's a candidate */
813 stby->walsnd_index = i;
814 stby->is_me = (walsnd == MyWalSnd);
815 n++;
816 }
817
818 /*
819 * In quorum mode, we return all the candidates. In priority mode, if we
820 * have too many candidates then return only the num_sync ones of highest
821 * priority.
822 */
825 {
826 /* Sort by priority ... */
827 qsort(*standbys, n, sizeof(SyncRepStandbyData),
829 /* ... then report just the first num_sync ones */
831 }
832
833 return n;
834}

References fb(), i, max_wal_senders, MyWalSnd, SyncRepConfigData::num_sync, palloc_array, qsort, SpinLockAcquire(), SpinLockRelease(), standby_priority_comparator(), SYNC_REP_PRIORITY, SyncRepConfigData::syncrep_method, SyncRepConfig, WalSndCtl, WalSndCtlData::walsnds, WALSNDSTATE_STOPPING, WALSNDSTATE_STREAMING, and XLogRecPtrIsValid.

Referenced by pg_stat_get_wal_senders(), and SyncRepGetSyncRecPtr().

◆ SyncRepGetNthLatestSyncRecPtr()

static void SyncRepGetNthLatestSyncRecPtr ( XLogRecPtr writePtr,
XLogRecPtr flushPtr,
XLogRecPtr applyPtr,
SyncRepStandbyData sync_standbys,
int  num_standbys,
uint8  nth 
)
static

Definition at line 702 of file syncrep.c.

708{
712 int i;
713
714 /* Should have enough candidates, or somebody messed up */
715 Assert(nth > 0 && nth <= num_standbys);
716
720
721 for (i = 0; i < num_standbys; i++)
722 {
723 write_array[i] = sync_standbys[i].write;
724 flush_array[i] = sync_standbys[i].flush;
725 apply_array[i] = sync_standbys[i].apply;
726 }
727
728 /* Sort each array in descending order */
732
733 /* Get Nth latest Write, Flush, Apply positions */
734 *writePtr = write_array[nth - 1];
735 *flushPtr = flush_array[nth - 1];
736 *applyPtr = apply_array[nth - 1];
737
741}

References Assert, cmp_lsn(), fb(), i, palloc_array, pfree(), and qsort.

Referenced by SyncRepGetSyncRecPtr().

◆ SyncRepGetOldestSyncRecPtr()

static void SyncRepGetOldestSyncRecPtr ( XLogRecPtr writePtr,
XLogRecPtr flushPtr,
XLogRecPtr applyPtr,
SyncRepStandbyData sync_standbys,
int  num_standbys 
)
static

Definition at line 669 of file syncrep.c.

674{
675 int i;
676
677 /*
678 * Scan through all sync standbys and calculate the oldest Write, Flush
679 * and Apply positions. We assume *writePtr et al were initialized to
680 * InvalidXLogRecPtr.
681 */
682 for (i = 0; i < num_standbys; i++)
683 {
685 XLogRecPtr flush = sync_standbys[i].flush;
686 XLogRecPtr apply = sync_standbys[i].apply;
687
689 *writePtr = write;
690 if (!XLogRecPtrIsValid(*flushPtr) || *flushPtr > flush)
691 *flushPtr = flush;
692 if (!XLogRecPtrIsValid(*applyPtr) || *applyPtr > apply)
693 *applyPtr = apply;
694 }
695}

References fb(), i, write, and XLogRecPtrIsValid.

Referenced by SyncRepGetSyncRecPtr().

◆ SyncRepGetStandbyPriority()

static int SyncRepGetStandbyPriority ( void  )
static

Definition at line 867 of file syncrep.c.

868{
869 const char *standby_name;
870 int priority;
871 bool found = false;
872
873 /*
874 * Since synchronous cascade replication is not allowed, we always set the
875 * priority of cascading walsender to zero.
876 */
878 return 0;
879
881 return 0;
882
885 {
887 strcmp(standby_name, "*") == 0)
888 {
889 found = true;
890 break;
891 }
893 }
894
895 if (!found)
896 return 0;
897
898 /*
899 * In quorum-based sync replication, all the standbys in the list have the
900 * same priority, one.
901 */
903}

References am_cascading_walsender, application_name, fb(), SyncRepConfigData::member_names, SyncRepConfigData::nmembers, pg_strcasecmp(), SYNC_REP_PRIORITY, SyncRepConfigData::syncrep_method, SyncRepConfig, and SyncStandbysDefined.

Referenced by SyncRepInitConfig().

◆ SyncRepGetSyncRecPtr()

static bool SyncRepGetSyncRecPtr ( XLogRecPtr writePtr,
XLogRecPtr flushPtr,
XLogRecPtr applyPtr,
bool am_sync 
)
static

Definition at line 595 of file syncrep.c.

597{
599 int num_standbys;
600 int i;
601
602 /* Initialize default results */
606 *am_sync = false;
607
608 /* Quick out if not even configured to be synchronous */
609 if (SyncRepConfig == NULL)
610 return false;
611
612 /* Get standbys that are considered as synchronous at this moment */
614
615 /* Am I among the candidate sync standbys? */
616 for (i = 0; i < num_standbys; i++)
617 {
618 if (sync_standbys[i].is_me)
619 {
620 *am_sync = true;
621 break;
622 }
623 }
624
625 /*
626 * Nothing more to do if we are not managing a sync standby or there are
627 * not enough synchronous standbys.
628 */
629 if (!(*am_sync) ||
631 {
633 return false;
634 }
635
636 /*
637 * In a priority-based sync replication, the synced positions are the
638 * oldest ones among sync standbys. In a quorum-based, they are the Nth
639 * latest ones.
640 *
641 * SyncRepGetNthLatestSyncRecPtr() also can calculate the oldest
642 * positions. But we use SyncRepGetOldestSyncRecPtr() for that calculation
643 * because it's a bit more efficient.
644 *
645 * XXX If the numbers of current and requested sync standbys are the same,
646 * we can use SyncRepGetOldestSyncRecPtr() to calculate the synced
647 * positions even in a quorum-based sync replication.
648 */
650 {
653 }
654 else
655 {
659 }
660
662 return true;
663}

References fb(), i, InvalidXLogRecPtr, SyncRepConfigData::num_sync, pfree(), SYNC_REP_PRIORITY, SyncRepConfigData::syncrep_method, SyncRepConfig, SyncRepGetCandidateStandbys(), SyncRepGetNthLatestSyncRecPtr(), and SyncRepGetOldestSyncRecPtr().

Referenced by SyncRepReleaseWaiters().

◆ SyncRepInitConfig()

void SyncRepInitConfig ( void  )

Definition at line 455 of file syncrep.c.

456{
457 int priority;
458
459 /*
460 * Determine if we are a potential sync standby and remember the result
461 * for handling replies from standby.
462 */
465 {
469
471 (errmsg_internal("standby \"%s\" now has synchronous standby priority %d",
473 }
474}

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

Referenced by StartLogicalReplication(), StartReplication(), and WalSndHandleConfigReload().

◆ SyncRepQueueInsert()

static void SyncRepQueueInsert ( int  mode)
static

Definition at line 382 of file syncrep.c.

383{
384 dlist_head *queue;
385 dlist_iter iter;
386
388 queue = &WalSndCtl->SyncRepQueue[mode];
389
390 dlist_reverse_foreach(iter, queue)
391 {
392 PGPROC *proc = dlist_container(PGPROC, syncRepLinks, iter.cur);
393
394 /*
395 * Stop at the queue element that we should insert after to ensure the
396 * queue is ordered by LSN.
397 */
398 if (proc->waitLSN < MyProc->waitLSN)
399 {
401 return;
402 }
403 }
404
405 /*
406 * If we get here, the list was either empty, or this process needs to be
407 * at the head.
408 */
410}

References Assert, dlist_iter::cur, dlist_container, dlist_insert_after(), dlist_push_head(), dlist_reverse_foreach, mode, MyProc, NUM_SYNC_REP_WAIT_MODE, PGPROC::syncRepLinks, WalSndCtlData::SyncRepQueue, PGPROC::waitLSN, and WalSndCtl.

Referenced by SyncRepWaitForLSN().

◆ SyncRepReleaseWaiters()

void SyncRepReleaseWaiters ( void  )

Definition at line 484 of file syncrep.c.

485{
489 bool got_recptr;
490 bool am_sync;
491 int numwrite = 0;
492 int numflush = 0;
493 int numapply = 0;
494
495 /*
496 * If this WALSender is serving a standby that is not on the list of
497 * potential sync standbys then we have nothing to do. If we are still
498 * starting up, still running base backup or the current flush position is
499 * still invalid, then leave quickly also. Streaming or stopping WAL
500 * senders are allowed to release waiters.
501 */
506 {
508 return;
509 }
510
511 /*
512 * We're a potential sync standby. Release waiters if there are enough
513 * sync standbys and we are considered as sync.
514 */
516
517 /*
518 * Check whether we are a sync standby or not, and calculate the synced
519 * positions among all sync standbys. (Note: although this step does not
520 * of itself require holding SyncRepLock, it seems like a good idea to do
521 * it after acquiring the lock. This ensures that the WAL pointers we use
522 * to release waiters are newer than any previous execution of this
523 * routine used.)
524 */
526
527 /*
528 * If we are managing a sync standby, though we weren't prior to this,
529 * then announce we are now a sync standby.
530 */
532 {
534
536 ereport(LOG,
537 (errmsg("standby \"%s\" is now a synchronous standby with priority %d",
539 else
540 ereport(LOG,
541 (errmsg("standby \"%s\" is now a candidate for quorum synchronous standby",
543 }
544
545 /*
546 * If the number of sync standbys is less than requested or we aren't
547 * managing a sync standby then just leave.
548 */
549 if (!got_recptr || !am_sync)
550 {
553 return;
554 }
555
556 /*
557 * Set the lsn first so that when we wake backends they will release up to
558 * this location.
559 */
561 {
564 }
566 {
569 }
571 {
574 }
575
577
578 elog(DEBUG3, "released %d procs up to write %X/%08X, %d procs up to flush %X/%08X, %d procs up to apply %X/%08X",
582}

References announce_next_takeover, application_name, DEBUG3, elog, ereport, errmsg, fb(), 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 XLogRecPtrIsValid.

Referenced by ProcessStandbyReplyMessage(), and WalSndHandleConfigReload().

◆ SyncRepUpdateSyncStandbysDefined()

void SyncRepUpdateSyncStandbysDefined ( void  )

Definition at line 970 of file syncrep.c.

971{
973
976 {
978
979 /*
980 * If synchronous_standby_names has been reset to empty, it's futile
981 * for backends to continue waiting. Since the user no longer wants
982 * synchronous replication, we'd better wake them up.
983 */
985 {
986 int i;
987
988 for (i = 0; i < NUM_SYNC_REP_WAIT_MODE; i++)
989 SyncRepWakeQueue(true, i);
990 }
991
992 /*
993 * Only allow people to join the queue when there are synchronous
994 * standbys defined. Without this interlock, there's a race
995 * condition: we might wake up all the current waiters; then, some
996 * backend that hasn't yet reloaded its config might go to sleep on
997 * the queue (and never wake up). This prevents that.
998 */
1001
1003 }
1005 {
1007
1008 /*
1009 * Note that there is no need to wake up the queues here. We would
1010 * reach this path only if SyncStandbysDefined() returns false, or it
1011 * would mean that some backends are waiting with the GUC set. See
1012 * SyncRepWaitForLSN().
1013 */
1015
1016 /*
1017 * Even if there is no sync standby defined, let the readers of this
1018 * information know that the sync standby data has been initialized.
1019 * This can just be done once, hence the previous check on
1020 * SYNC_STANDBY_INIT to avoid useless work.
1021 */
1023
1025 }
1026}

References Assert, fb(), 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 149 of file syncrep.c.

150{
151 int mode;
152
153 /*
154 * This should be called while holding interrupts during a transaction
155 * commit to prevent the follow-up shared memory queue cleanups to be
156 * influenced by external interruptions.
157 */
159
160 /*
161 * Fast exit if user has not requested sync replication, or there are no
162 * sync replication standby names defined.
163 *
164 * Since this routine gets called every commit time, it's important to
165 * exit quickly if sync replication is not requested.
166 *
167 * We check WalSndCtl->sync_standbys_status flag without the lock and exit
168 * immediately if SYNC_STANDBY_INIT is set (the checkpointer has
169 * initialized this data) but SYNC_STANDBY_DEFINED is missing (no sync
170 * replication requested).
171 *
172 * If SYNC_STANDBY_DEFINED is set, we need to check the status again later
173 * while holding the lock, to check the flag and operate the sync rep
174 * queue atomically. This is necessary to avoid the race condition
175 * described in SyncRepUpdateSyncStandbysDefined(). On the other hand, if
176 * SYNC_STANDBY_DEFINED is not set, the lock is not necessary because we
177 * don't touch the queue.
178 */
179 if (!SyncRepRequested() ||
180 ((((volatile WalSndCtlData *) WalSndCtl)->sync_standbys_status) &
182 return;
183
184 /* Cap the level for anything other than commit to remote flush only. */
185 if (commit)
187 else
189
192
195
196 /*
197 * We don't wait for sync rep if SYNC_STANDBY_DEFINED is not set. See
198 * SyncRepUpdateSyncStandbysDefined().
199 *
200 * Also check that the standby hasn't already replied. Unlikely race
201 * condition but we'll be fetching that cache line anyway so it's likely
202 * to be a low cost check.
203 *
204 * If the sync standby data has not been initialized yet
205 * (SYNC_STANDBY_INIT is not set), fall back to a check based on the LSN,
206 * then do a direct GUC check.
207 */
209 {
212 {
214 return;
215 }
216 }
217 else if (lsn <= WalSndCtl->lsn[mode])
218 {
219 /*
220 * The LSN is older than what we need to wait for. The sync standby
221 * data has not been initialized yet, but we are OK to not wait
222 * because we know that there is no point in doing so based on the
223 * LSN.
224 */
226 return;
227 }
228 else if (!SyncStandbysDefined())
229 {
230 /*
231 * If we are here, the sync standby data has not been initialized yet,
232 * and the LSN is newer than what need to wait for, so we have fallen
233 * back to the best thing we could do in this case: a check on
234 * SyncStandbysDefined() to see if the GUC is set or not.
235 *
236 * When the GUC has a value, we wait until the checkpointer updates
237 * the status data because we cannot be sure yet if we should wait or
238 * not. Here, the GUC has *no* value, we are sure that there is no
239 * point to wait; this matters for example when initializing a
240 * cluster, where we should never wait, and no sync standbys is the
241 * default behavior.
242 */
244 return;
245 }
246
247 /*
248 * Set our waitLSN so WALSender will know when to wake us, and add
249 * ourselves to the queue.
250 */
251 MyProc->waitLSN = lsn;
256
257 /* Alter ps display to show waiting for sync rep. */
259 {
260 char buffer[32];
261
262 sprintf(buffer, "waiting for %X/%08X", LSN_FORMAT_ARGS(lsn));
263 set_ps_display_suffix(buffer);
264 }
265
266 /*
267 * Wait for specified LSN to be confirmed.
268 *
269 * Each proc has its own wait latch, so we perform a normal latch
270 * check/wait loop here.
271 */
272 for (;;)
273 {
274 int rc;
275
276 /* Must reset the latch before testing state. */
278
279 /*
280 * Acquiring the lock is not needed, the latch ensures proper
281 * barriers. If it looks like we're done, we must really be done,
282 * because once walsender changes the state to SYNC_REP_WAIT_COMPLETE,
283 * it will never update it again, so we can't be seeing a stale value
284 * in that case.
285 */
287 break;
288
289 /*
290 * If a wait for synchronous replication is pending, we can neither
291 * acknowledge the commit nor raise ERROR or FATAL. The latter would
292 * lead the client to believe that the transaction aborted, which is
293 * not true: it's already committed locally. The former is no good
294 * either: the client has requested synchronous replication, and is
295 * entitled to assume that an acknowledged commit is also replicated,
296 * which might not be true. So in this case we issue a WARNING (which
297 * some clients may be able to interpret) and shut off further output.
298 * We do NOT reset ProcDiePending, so that the process will die after
299 * the commit is cleaned up.
300 */
301 if (ProcDiePending)
302 {
303 if (ProcDieSenderPid != 0)
306 errmsg("canceling the wait for synchronous replication and terminating connection due to administrator command"),
307 errdetail("The transaction has already committed locally, but might not have been replicated to the standby."),
308 errdetail_log("The transaction has already committed locally, but might not have been replicated to the standby. Signal sent by PID %d, UID %d.",
309 (int) ProcDieSenderPid,
310 (int) ProcDieSenderUid)));
311 else
314 errmsg("canceling the wait for synchronous replication and terminating connection due to administrator command"),
315 errdetail("The transaction has already committed locally, but might not have been replicated to the standby.")));
318 break;
319 }
320
321 /*
322 * It's unclear what to do if a query cancel interrupt arrives. We
323 * can't actually abort at this point, but ignoring the interrupt
324 * altogether is not helpful, so we just terminate the wait with a
325 * suitable warning.
326 */
328 {
329 QueryCancelPending = false;
331 (errmsg("canceling wait for synchronous replication due to user request"),
332 errdetail("The transaction has already committed locally, but might not have been replicated to the standby.")));
334 break;
335 }
336
337 /*
338 * Wait on latch. Any condition that should wake us up will set the
339 * latch, so no need for timeout.
340 */
343
344 /*
345 * If the postmaster dies, we'll probably never get an acknowledgment,
346 * because all the wal sender processes will exit. So just bail out.
347 */
348 if (rc & WL_POSTMASTER_DEATH)
349 {
350 ProcDiePending = true;
353 break;
354 }
355 }
356
357 /*
358 * WalSender has checked our LSN and has removed us from queue. Clean up
359 * state and leave. It's OK to reset these shared memory fields without
360 * holding SyncRepLock, because any walsenders will ignore us anyway when
361 * we're not on the queue. We need a read barrier to make sure we see the
362 * changes to the queue link (this might be unnecessary without
363 * assertions, but better safe than sorry).
364 */
369
370 /* reset ps display to remove the suffix */
373}

References Assert, DestNone, dlist_node_is_detached(), ereport, errcode(), errdetail(), errdetail_log(), errmsg, fb(), InterruptHoldoffCount, InvalidXLogRecPtr, LSN_FORMAT_ARGS, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), Min, mode, MyLatch, MyProc, pg_read_barrier, ProcDiePending, ProcDieSenderPid, ProcDieSenderUid, 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().

◆ SyncRepWakeQueue()

static int SyncRepWakeQueue ( bool  all,
int  mode 
)
static

Definition at line 914 of file syncrep.c.

915{
916 int numprocs = 0;
918
922
924 {
925 PGPROC *proc = dlist_container(PGPROC, syncRepLinks, iter.cur);
926
927 /*
928 * Assume the queue is ordered by LSN
929 */
930 if (!all && WalSndCtl->lsn[mode] < proc->waitLSN)
931 return numprocs;
932
933 /*
934 * Remove from queue.
935 */
937
938 /*
939 * SyncRepWaitForLSN() reads syncRepState without holding the lock, so
940 * make sure that it sees the queue link being removed before the
941 * syncRepState change.
942 */
944
945 /*
946 * Set state to complete; see SyncRepWaitForLSN() for discussion of
947 * the various states.
948 */
950
951 /*
952 * Wake only when we have set state and removed from queue.
953 */
954 SetLatch(&(proc->procLatch));
955
956 numprocs++;
957 }
958
959 return numprocs;
960}

References Assert, dlist_mutable_iter::cur, dlist_container, dlist_delete_thoroughly(), dlist_foreach_modify, fb(), WalSndCtlData::lsn, LW_EXCLUSIVE, LWLockHeldByMeInMode(), mode, NUM_SYNC_REP_WAIT_MODE, pg_write_barrier, PGPROC::procLatch, SetLatch(), SYNC_REP_WAIT_COMPLETE, PGPROC::syncRepLinks, WalSndCtlData::SyncRepQueue, PGPROC::syncRepState, PGPROC::waitLSN, and WalSndCtl.

Referenced by SyncRepReleaseWaiters(), and SyncRepUpdateSyncStandbysDefined().

Variable Documentation

◆ announce_next_takeover

bool announce_next_takeover = true
static

Definition at line 96 of file syncrep.c.

Referenced by SyncRepReleaseWaiters().

◆ SyncRepConfig

◆ SyncRepStandbyNames

char* SyncRepStandbyNames

Definition at line 91 of file syncrep.c.

◆ SyncRepWaitMode

int SyncRepWaitMode = SYNC_REP_NO_WAIT
static

Definition at line 99 of file syncrep.c.

Referenced by assign_synchronous_commit(), and SyncRepWaitForLSN().