#include "postgres.h"
#include <unistd.h>
#include "access/xact.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "replication/syncrep.h"
#include "replication/walsender.h"
#include "replication/walsender_private.h"
#include "storage/pmsignal.h"
#include "storage/proc.h"
#include "tcop/tcopprot.h"
#include "utils/builtins.h"
#include "utils/ps_status.h"

Include dependency graph for syncrep.c:

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, List sync_standbys)

static void	SyncRepGetNthLatestSyncRecPtr (XLogRecPtr writePtr, XLogRecPtr flushPtr, XLogRecPtr applyPtr, List sync_standbys, uint8 nth)

static int	SyncRepGetStandbyPriority (void)

static List *	SyncRepGetSyncStandbysPriority (bool *am_sync)

static List *	SyncRepGetSyncStandbysQuorum (bool *am_sync)

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)

List *	SyncRepGetSyncStandbys (bool *am_sync)

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
char *	SyncRepStandbyNames

static bool	announce_next_takeover = true

SyncRepConfigData *	SyncRepConfig = NULL

static int	SyncRepWaitMode = SYNC_REP_NO_WAIT

Macro Definition Documentation

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

Definition at line 92 of file syncrep.c.

Referenced by SyncRepGetStandbyPriority(), SyncRepUpdateSyncStandbysDefined(), and SyncRepWaitForLSN().

Function Documentation

void assign_synchronous_commit	(	int	newval,
		void *	extra
	)

Definition at line 1173 of file syncrep.c.

References SYNC_REP_NO_WAIT, SYNC_REP_WAIT_APPLY, SYNC_REP_WAIT_FLUSH, SYNC_REP_WAIT_WRITE, SYNCHRONOUS_COMMIT_REMOTE_APPLY, SYNCHRONOUS_COMMIT_REMOTE_FLUSH, SYNCHRONOUS_COMMIT_REMOTE_WRITE, and SyncRepWaitMode.

 {
     switch (newval)
     {
         case SYNCHRONOUS_COMMIT_REMOTE_WRITE:
             SyncRepWaitMode = SYNC_REP_WAIT_WRITE;
             break;
         case SYNCHRONOUS_COMMIT_REMOTE_FLUSH:
             SyncRepWaitMode = SYNC_REP_WAIT_FLUSH;
             break;
         case SYNCHRONOUS_COMMIT_REMOTE_APPLY:
             SyncRepWaitMode = SYNC_REP_WAIT_APPLY;
             break;
         default:
             SyncRepWaitMode = SYNC_REP_NO_WAIT;
             break;
     }
 }

void assign_synchronous_standby_names	(	const char *	newval,
		void *	extra
	)

Definition at line 1167 of file syncrep.c.

 {
     SyncRepConfig = (SyncRepConfigData *) extra;
 }

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

Definition at line 1110 of file syncrep.c.

References SyncRepConfigData::config_size, GUC_check_errcode(), GUC_check_errdetail, GUC_check_errmsg, malloc, NULL, SyncRepConfigData::num_sync, syncrep_parse_error_msg, syncrep_parse_result, syncrep_scanner_finish(), syncrep_scanner_init(), and syncrep_yyparse().

 {
     if (*newval != NULL && (*newval)[0] != '\0')
     {
         int         parse_rc;
         SyncRepConfigData *pconf;
 
         /* Reset communication variables to ensure a fresh start */
         syncrep_parse_result = NULL;
         syncrep_parse_error_msg = NULL;
 
         /* Parse the synchronous_standby_names string */
         syncrep_scanner_init(*newval);
         parse_rc = syncrep_yyparse();
         syncrep_scanner_finish();
 
         if (parse_rc != 0 || syncrep_parse_result == NULL)
         {
             GUC_check_errcode(ERRCODE_SYNTAX_ERROR);
             if (syncrep_parse_error_msg)
                 GUC_check_errdetail("%s", syncrep_parse_error_msg);
             else
                 GUC_check_errdetail("synchronous_standby_names parser failed");
             return false;
         }
 
         if (syncrep_parse_result->num_sync <= 0)
         {
             GUC_check_errmsg("number of synchronous standbys (%d) must be greater than zero",
                              syncrep_parse_result->num_sync);
             return false;
         }
 
         /* GUC extra value must be malloc'd, not palloc'd */
         pconf = (SyncRepConfigData *)
             malloc(syncrep_parse_result->config_size);
         if (pconf == NULL)
             return false;
         memcpy(pconf, syncrep_parse_result, syncrep_parse_result->config_size);
 
         *extra = (void *) pconf;
 
         /*
          * We need not explicitly clean up syncrep_parse_result.  It, and any
          * other cruft generated during parsing, will be freed when the
          * current memory context is deleted.  (This code is generally run in
          * a short-lived context used for config file processing, so that will
          * not be very long.)
          */
     }
     else
         *extra = NULL;
 
     return true;
 }

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

static

Definition at line 655 of file syncrep.c.

Referenced by SyncRepGetNthLatestSyncRecPtr().

 {
     XLogRecPtr  lsn1 = *((const XLogRecPtr *) a);
     XLogRecPtr  lsn2 = *((const XLogRecPtr *) b);
 
     if (lsn1 > lsn2)
         return -1;
     else if (lsn1 == lsn2)
         return 0;
     else
         return 1;
 }

static void SyncRepCancelWait ( void )

static

Definition at line 350 of file syncrep.c.

References LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MyProc, SHMQueueDelete(), SHMQueueIsDetached(), SYNC_REP_NOT_WAITING, PGPROC::syncRepLinks, and PGPROC::syncRepState.

Referenced by SyncRepWaitForLSN().

 {
     LWLockAcquire(SyncRepLock, LW_EXCLUSIVE);
     if (!SHMQueueIsDetached(&(MyProc->syncRepLinks)))
         SHMQueueDelete(&(MyProc->syncRepLinks));
     MyProc->syncRepState = SYNC_REP_NOT_WAITING;
     LWLockRelease(SyncRepLock);
 }

void SyncRepCleanupAtProcExit ( void )

Definition at line 360 of file syncrep.c.

References LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MyProc, SHMQueueDelete(), SHMQueueIsDetached(), and PGPROC::syncRepLinks.

Referenced by ProcKill().

 {
     if (!SHMQueueIsDetached(&(MyProc->syncRepLinks)))
     {
         LWLockAcquire(SyncRepLock, LW_EXCLUSIVE);
         SHMQueueDelete(&(MyProc->syncRepLinks));
         LWLockRelease(SyncRepLock);
     }
 }

static void SyncRepGetNthLatestSyncRecPtr	(	XLogRecPtr *	writePtr,
		XLogRecPtr *	flushPtr,
		XLogRecPtr *	applyPtr,
		List *	sync_standbys,
		uint8	nth
	)

static

Definition at line 608 of file syncrep.c.

References WalSnd::apply, cmp_lsn(), WalSnd::flush, i, lfirst_int, list_length(), WalSnd::mutex, palloc(), pfree(), qsort, SpinLockAcquire, SpinLockRelease, WalSndCtl, WalSndCtlData::walsnds, and WalSnd::write.

Referenced by SyncRepGetSyncRecPtr().

 {
     ListCell   *cell;
     XLogRecPtr *write_array;
     XLogRecPtr *flush_array;
     XLogRecPtr *apply_array;
     int         len;
     int         i = 0;
 
     len = list_length(sync_standbys);
     write_array = (XLogRecPtr *) palloc(sizeof(XLogRecPtr) * len);
     flush_array = (XLogRecPtr *) palloc(sizeof(XLogRecPtr) * len);
     apply_array = (XLogRecPtr *) palloc(sizeof(XLogRecPtr) * len);
 
     foreach(cell, sync_standbys)
     {
         WalSnd     *walsnd = &WalSndCtl->walsnds[lfirst_int(cell)];
 
         SpinLockAcquire(&walsnd->mutex);
         write_array[i] = walsnd->write;
         flush_array[i] = walsnd->flush;
         apply_array[i] = walsnd->apply;
         SpinLockRelease(&walsnd->mutex);
 
         i++;
     }
 
     /* Sort each array in descending order */
     qsort(write_array, len, sizeof(XLogRecPtr), cmp_lsn);
     qsort(flush_array, len, sizeof(XLogRecPtr), cmp_lsn);
     qsort(apply_array, len, sizeof(XLogRecPtr), cmp_lsn);
 
     /* Get Nth latest Write, Flush, Apply positions */
     *writePtr = write_array[nth - 1];
     *flushPtr = flush_array[nth - 1];
     *applyPtr = apply_array[nth - 1];
 
     pfree(write_array);
     pfree(flush_array);
     pfree(apply_array);
 }

static void SyncRepGetOldestSyncRecPtr	(	XLogRecPtr *	writePtr,
		XLogRecPtr *	flushPtr,
		XLogRecPtr *	applyPtr,
		List *	sync_standbys
	)

static

Definition at line 572 of file syncrep.c.

References WalSnd::apply, WalSnd::flush, lfirst_int, WalSnd::mutex, SpinLockAcquire, SpinLockRelease, WalSndCtl, WalSndCtlData::walsnds, write, WalSnd::write, and XLogRecPtrIsInvalid.

Referenced by SyncRepGetSyncRecPtr().

 {
     ListCell   *cell;
 
     /*
      * Scan through all sync standbys and calculate the oldest Write, Flush
      * and Apply positions.
      */
     foreach(cell, sync_standbys)
     {
         WalSnd     *walsnd = &WalSndCtl->walsnds[lfirst_int(cell)];
         XLogRecPtr  write;
         XLogRecPtr  flush;
         XLogRecPtr  apply;
 
         SpinLockAcquire(&walsnd->mutex);
         write = walsnd->write;
         flush = walsnd->flush;
         apply = walsnd->apply;
         SpinLockRelease(&walsnd->mutex);
 
         if (XLogRecPtrIsInvalid(*writePtr) || *writePtr > write)
             *writePtr = write;
         if (XLogRecPtrIsInvalid(*flushPtr) || *flushPtr > flush)
             *flushPtr = flush;
         if (XLogRecPtrIsInvalid(*applyPtr) || *applyPtr > apply)
             *applyPtr = apply;
     }
 }

static int SyncRepGetStandbyPriority ( void )

static

Definition at line 926 of file syncrep.c.

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

Referenced by SyncRepInitConfig().

 {
     const char *standby_name;
     int         priority;
     bool        found = false;
 
     /*
      * Since synchronous cascade replication is not allowed, we always set the
      * priority of cascading walsender to zero.
      */
     if (am_cascading_walsender)
         return 0;
 
     if (!SyncStandbysDefined() || SyncRepConfig == NULL)
         return 0;
 
     standby_name = SyncRepConfig->member_names;
     for (priority = 1; priority <= SyncRepConfig->nmembers; priority++)
     {
         if (pg_strcasecmp(standby_name, application_name) == 0 ||
             strcmp(standby_name, "*") == 0)
         {
             found = true;
             break;
         }
         standby_name += strlen(standby_name) + 1;
     }
 
     if (!found)
         return 0;
 
     /*
      * In quorum-based sync replication, all the standbys in the list have the
      * same priority, one.
      */
     return (SyncRepConfig->syncrep_method == SYNC_REP_PRIORITY) ? priority : 1;
 }

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

static

Definition at line 515 of file syncrep.c.

References InvalidXLogRecPtr, list_free(), list_length(), NULL, SyncRepConfigData::num_sync, SYNC_REP_PRIORITY, SyncRepConfigData::syncrep_method, SyncRepGetNthLatestSyncRecPtr(), SyncRepGetOldestSyncRecPtr(), and SyncRepGetSyncStandbys().

Referenced by SyncRepReleaseWaiters().

 {
     List       *sync_standbys;
 
     *writePtr = InvalidXLogRecPtr;
     *flushPtr = InvalidXLogRecPtr;
     *applyPtr = InvalidXLogRecPtr;
     *am_sync = false;
 
     /* Get standbys that are considered as synchronous at this moment */
     sync_standbys = SyncRepGetSyncStandbys(am_sync);
 
     /*
      * Quick exit if we are not managing a sync standby or there are not
      * enough synchronous standbys.
      */
     if (!(*am_sync) ||
         SyncRepConfig == NULL ||
         list_length(sync_standbys) < SyncRepConfig->num_sync)
     {
         list_free(sync_standbys);
         return false;
     }
 
     /*
      * In a priority-based sync replication, the synced positions are the
      * oldest ones among sync standbys. In a quorum-based, they are the Nth
      * latest ones.
      *
      * SyncRepGetNthLatestSyncRecPtr() also can calculate the oldest
      * positions. But we use SyncRepGetOldestSyncRecPtr() for that calculation
      * because it's a bit more efficient.
      *
      * XXX If the numbers of current and requested sync standbys are the same,
      * we can use SyncRepGetOldestSyncRecPtr() to calculate the synced
      * positions even in a quorum-based sync replication.
      */
     if (SyncRepConfig->syncrep_method == SYNC_REP_PRIORITY)
     {
         SyncRepGetOldestSyncRecPtr(writePtr, flushPtr, applyPtr,
                                    sync_standbys);
     }
     else
     {
         SyncRepGetNthLatestSyncRecPtr(writePtr, flushPtr, applyPtr,
                                       sync_standbys, SyncRepConfig->num_sync);
     }
 
     list_free(sync_standbys);
     return true;
 }

List* SyncRepGetSyncStandbys ( bool * am_sync )

Definition at line 677 of file syncrep.c.

References NIL, NULL, SYNC_REP_PRIORITY, SyncRepConfigData::syncrep_method, SyncRepGetSyncStandbysPriority(), and SyncRepGetSyncStandbysQuorum().

Referenced by pg_stat_get_wal_senders(), and SyncRepGetSyncRecPtr().

 {
     /* Set default result */
     if (am_sync != NULL)
         *am_sync = false;
 
     /* Quick exit if sync replication is not requested */
     if (SyncRepConfig == NULL)
         return NIL;
 
     return (SyncRepConfig->syncrep_method == SYNC_REP_PRIORITY) ?
         SyncRepGetSyncStandbysPriority(am_sync) :
         SyncRepGetSyncStandbysQuorum(am_sync);
 }

static List * SyncRepGetSyncStandbysPriority ( bool * am_sync )

static

Definition at line 758 of file syncrep.c.

References Assert, WalSnd::flush, i, lappend_int(), lfirst_int, list_concat(), list_delete_cell(), list_free(), list_head(), list_length(), lnext, max_wal_senders, MyWalSnd, next, NIL, SyncRepConfigData::nmembers, NULL, SyncRepConfigData::num_sync, pfree(), WalSnd::pid, result, WalSnd::state, SYNC_REP_PRIORITY, WalSnd::sync_standby_priority, SyncRepConfigData::syncrep_method, WalSndCtl, WalSndCtlData::walsnds, WALSNDSTATE_STREAMING, and XLogRecPtrIsInvalid.

Referenced by SyncRepGetSyncStandbys().

 {
     List       *result = NIL;
     List       *pending = NIL;
     int         lowest_priority;
     int         next_highest_priority;
     int         this_priority;
     int         priority;
     int         i;
     bool        am_in_pending = false;
     volatile WalSnd *walsnd;    /* Use volatile pointer to prevent code
                                  * rearrangement */
 
     Assert(SyncRepConfig->syncrep_method == SYNC_REP_PRIORITY);
 
     lowest_priority = SyncRepConfig->nmembers;
     next_highest_priority = lowest_priority + 1;
 
     /*
      * Find the sync standbys which have the highest priority (i.e, 1). Also
      * store all the other potential sync standbys into the pending list, in
      * order to scan it later and find other sync standbys from it quickly.
      */
     for (i = 0; i < max_wal_senders; i++)
     {
         walsnd = &WalSndCtl->walsnds[i];
 
         /* Must be active */
         if (walsnd->pid == 0)
             continue;
 
         /* Must be streaming */
         if (walsnd->state != WALSNDSTATE_STREAMING)
             continue;
 
         /* Must be synchronous */
         this_priority = walsnd->sync_standby_priority;
         if (this_priority == 0)
             continue;
 
         /* Must have a valid flush position */
         if (XLogRecPtrIsInvalid(walsnd->flush))
             continue;
 
         /*
          * If the priority is equal to 1, consider this standby as sync and
          * append it to the result. Otherwise append this standby to the
          * pending list to check if it's actually sync or not later.
          */
         if (this_priority == 1)
         {
             result = lappend_int(result, i);
             if (am_sync != NULL && walsnd == MyWalSnd)
                 *am_sync = true;
             if (list_length(result) == SyncRepConfig->num_sync)
             {
                 list_free(pending);
                 return result;  /* Exit if got enough sync standbys */
             }
         }
         else
         {
             pending = lappend_int(pending, i);
             if (am_sync != NULL && walsnd == MyWalSnd)
                 am_in_pending = true;
 
             /*
              * Track the highest priority among the standbys in the pending
              * list, in order to use it as the starting priority for later
              * scan of the list. This is useful to find quickly the sync
              * standbys from the pending list later because we can skip
              * unnecessary scans for the unused priorities.
              */
             if (this_priority < next_highest_priority)
                 next_highest_priority = this_priority;
         }
     }
 
     /*
      * Consider all pending standbys as sync if the number of them plus
      * already-found sync ones is lower than the configuration requests.
      */
     if (list_length(result) + list_length(pending) <= SyncRepConfig->num_sync)
     {
         bool        needfree = (result != NIL && pending != NIL);
 
         /*
          * Set *am_sync to true if this walsender is in the pending list
          * because all pending standbys are considered as sync.
          */
         if (am_sync != NULL && !(*am_sync))
             *am_sync = am_in_pending;
 
         result = list_concat(result, pending);
         if (needfree)
             pfree(pending);
         return result;
     }
 
     /*
      * Find the sync standbys from the pending list.
      */
     priority = next_highest_priority;
     while (priority <= lowest_priority)
     {
         ListCell   *cell;
         ListCell   *prev = NULL;
         ListCell   *next;
 
         next_highest_priority = lowest_priority + 1;
 
         for (cell = list_head(pending); cell != NULL; cell = next)
         {
             i = lfirst_int(cell);
             walsnd = &WalSndCtl->walsnds[i];
 
             next = lnext(cell);
 
             this_priority = walsnd->sync_standby_priority;
             if (this_priority == priority)
             {
                 result = lappend_int(result, i);
                 if (am_sync != NULL && walsnd == MyWalSnd)
                     *am_sync = true;
 
                 /*
                  * We should always exit here after the scan of pending list
                  * starts because we know that the list has enough elements to
                  * reach SyncRepConfig->num_sync.
                  */
                 if (list_length(result) == SyncRepConfig->num_sync)
                 {
                     list_free(pending);
                     return result;  /* Exit if got enough sync standbys */
                 }
 
                 /*
                  * Remove the entry for this sync standby from the list to
                  * prevent us from looking at the same entry again.
                  */
                 pending = list_delete_cell(pending, cell, prev);
 
                 continue;
             }
 
             if (this_priority < next_highest_priority)
                 next_highest_priority = this_priority;
 
             prev = cell;
         }
 
         priority = next_highest_priority;
     }
 
     /* never reached, but keep compiler quiet */
     Assert(false);
     return result;
 }

static List * SyncRepGetSyncStandbysQuorum ( bool * am_sync )

static

Definition at line 703 of file syncrep.c.

References Assert, WalSnd::flush, i, lappend_int(), max_wal_senders, MyWalSnd, NIL, NULL, WalSnd::pid, result, WalSnd::state, SYNC_REP_QUORUM, WalSnd::sync_standby_priority, SyncRepConfigData::syncrep_method, WalSndCtl, WalSndCtlData::walsnds, WALSNDSTATE_STREAMING, and XLogRecPtrIsInvalid.

Referenced by SyncRepGetSyncStandbys().

 {
     List       *result = NIL;
     int         i;
     volatile WalSnd *walsnd;    /* Use volatile pointer to prevent code
                                  * rearrangement */
 
     Assert(SyncRepConfig->syncrep_method == SYNC_REP_QUORUM);
 
     for (i = 0; i < max_wal_senders; i++)
     {
         walsnd = &WalSndCtl->walsnds[i];
 
         /* Must be active */
         if (walsnd->pid == 0)
             continue;
 
         /* Must be streaming */
         if (walsnd->state != WALSNDSTATE_STREAMING)
             continue;
 
         /* Must be synchronous */
         if (walsnd->sync_standby_priority == 0)
             continue;
 
         /* Must have a valid flush position */
         if (XLogRecPtrIsInvalid(walsnd->flush))
             continue;
 
         /*
          * Consider this standby as a candidate for quorum sync standbys and
          * append it to the result.
          */
         result = lappend_int(result, i);
         if (am_sync != NULL && walsnd == MyWalSnd)
             *am_sync = true;
     }
 
     return result;
 }

void SyncRepInitConfig ( void )

Definition at line 381 of file syncrep.c.

References application_name, DEBUG1, ereport, errmsg(), LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MyWalSnd, WalSnd::sync_standby_priority, and SyncRepGetStandbyPriority().

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

 {
     int         priority;
 
     /*
      * Determine if we are a potential sync standby and remember the result
      * for handling replies from standby.
      */
     priority = SyncRepGetStandbyPriority();
     if (MyWalSnd->sync_standby_priority != priority)
     {
         LWLockAcquire(SyncRepLock, LW_EXCLUSIVE);
         MyWalSnd->sync_standby_priority = priority;
         LWLockRelease(SyncRepLock);
         ereport(DEBUG1,
                 (errmsg("standby \"%s\" now has synchronous standby priority %u",
                         application_name, priority)));
     }
 }

static void SyncRepQueueInsert ( int mode )

static

Definition at line 317 of file syncrep.c.

References Assert, MyProc, NUM_SYNC_REP_WAIT_MODE, offsetof, SHMQueueInsertAfter(), SHMQueuePrev(), PGPROC::syncRepLinks, WalSndCtlData::SyncRepQueue, PGPROC::waitLSN, and WalSndCtl.

Referenced by SyncRepWaitForLSN().

 {
     PGPROC     *proc;
 
     Assert(mode >= 0 && mode < NUM_SYNC_REP_WAIT_MODE);
     proc = (PGPROC *) SHMQueuePrev(&(WalSndCtl->SyncRepQueue[mode]),
                                    &(WalSndCtl->SyncRepQueue[mode]),
                                    offsetof(PGPROC, syncRepLinks));
 
     while (proc)
     {
         /*
          * Stop at the queue element that we should after to ensure the queue
          * is ordered by LSN.
          */
         if (proc->waitLSN < MyProc->waitLSN)
             break;
 
         proc = (PGPROC *) SHMQueuePrev(&(WalSndCtl->SyncRepQueue[mode]),
                                        &(proc->syncRepLinks),
                                        offsetof(PGPROC, syncRepLinks));
     }
 
     if (proc)
         SHMQueueInsertAfter(&(proc->syncRepLinks), &(MyProc->syncRepLinks));
     else
         SHMQueueInsertAfter(&(WalSndCtl->SyncRepQueue[mode]), &(MyProc->syncRepLinks));
 }

void SyncRepReleaseWaiters ( void )

Definition at line 409 of file syncrep.c.

References announce_next_takeover, application_name, DEBUG3, elog, ereport, errmsg(), WalSnd::flush, LOG, WalSndCtlData::lsn, 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, SyncRepGetSyncRecPtr(), SyncRepWakeQueue(), WalSndCtl, WALSNDSTATE_STREAMING, and XLogRecPtrIsInvalid.

Referenced by ProcessStandbyReplyMessage().

 {
     volatile WalSndCtlData *walsndctl = WalSndCtl;
     XLogRecPtr  writePtr;
     XLogRecPtr  flushPtr;
     XLogRecPtr  applyPtr;
     bool        got_recptr;
     bool        am_sync;
     int         numwrite = 0;
     int         numflush = 0;
     int         numapply = 0;
 
     /*
      * If this WALSender is serving a standby that is not on the list of
      * potential sync standbys then we have nothing to do. If we are still
      * starting up, still running base backup or the current flush position is
      * still invalid, then leave quickly also.
      */
     if (MyWalSnd->sync_standby_priority == 0 ||
         MyWalSnd->state < WALSNDSTATE_STREAMING ||
         XLogRecPtrIsInvalid(MyWalSnd->flush))
     {
         announce_next_takeover = true;
         return;
     }
 
     /*
      * We're a potential sync standby. Release waiters if there are enough
      * sync standbys and we are considered as sync.
      */
     LWLockAcquire(SyncRepLock, LW_EXCLUSIVE);
 
     /*
      * Check whether we are a sync standby or not, and calculate the synced
      * positions among all sync standbys.
      */
     got_recptr = SyncRepGetSyncRecPtr(&writePtr, &flushPtr, &applyPtr, &am_sync);
 
     /*
      * If we are managing a sync standby, though we weren't prior to this,
      * then announce we are now a sync standby.
      */
     if (announce_next_takeover && am_sync)
     {
         announce_next_takeover = false;
 
         if (SyncRepConfig->syncrep_method == SYNC_REP_PRIORITY)
             ereport(LOG,
                     (errmsg("standby \"%s\" is now a synchronous standby with priority %u",
                             application_name, MyWalSnd->sync_standby_priority)));
         else
             ereport(LOG,
                     (errmsg("standby \"%s\" is now a candidate for quorum synchronous standby",
                             application_name)));
     }
 
     /*
      * If the number of sync standbys is less than requested or we aren't
      * managing a sync standby then just leave.
      */
     if (!got_recptr || !am_sync)
     {
         LWLockRelease(SyncRepLock);
         announce_next_takeover = !am_sync;
         return;
     }
 
     /*
      * Set the lsn first so that when we wake backends they will release up to
      * this location.
      */
     if (walsndctl->lsn[SYNC_REP_WAIT_WRITE] < writePtr)
     {
         walsndctl->lsn[SYNC_REP_WAIT_WRITE] = writePtr;
         numwrite = SyncRepWakeQueue(false, SYNC_REP_WAIT_WRITE);
     }
     if (walsndctl->lsn[SYNC_REP_WAIT_FLUSH] < flushPtr)
     {
         walsndctl->lsn[SYNC_REP_WAIT_FLUSH] = flushPtr;
         numflush = SyncRepWakeQueue(false, SYNC_REP_WAIT_FLUSH);
     }
     if (walsndctl->lsn[SYNC_REP_WAIT_APPLY] < applyPtr)
     {
         walsndctl->lsn[SYNC_REP_WAIT_APPLY] = applyPtr;
         numapply = SyncRepWakeQueue(false, SYNC_REP_WAIT_APPLY);
     }
 
     LWLockRelease(SyncRepLock);
 
     elog(DEBUG3, "released %d procs up to write %X/%X, %d procs up to flush %X/%X, %d procs up to apply %X/%X",
          numwrite, (uint32) (writePtr >> 32), (uint32) writePtr,
          numflush, (uint32) (flushPtr >> 32), (uint32) flushPtr,
          numapply, (uint32) (applyPtr >> 32), (uint32) applyPtr);
 }

void SyncRepUpdateSyncStandbysDefined ( void )

Definition at line 1034 of file syncrep.c.

References i, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), NUM_SYNC_REP_WAIT_MODE, WalSndCtlData::sync_standbys_defined, SyncRepWakeQueue(), SyncStandbysDefined, and WalSndCtl.

Referenced by UpdateSharedMemoryConfig().

 {
     bool        sync_standbys_defined = SyncStandbysDefined();
 
     if (sync_standbys_defined != WalSndCtl->sync_standbys_defined)
     {
         LWLockAcquire(SyncRepLock, LW_EXCLUSIVE);
 
         /*
          * If synchronous_standby_names has been reset to empty, it's futile
          * for backends to continue to waiting.  Since the user no longer
          * wants synchronous replication, we'd better wake them up.
          */
         if (!sync_standbys_defined)
         {
             int         i;
 
             for (i = 0; i < NUM_SYNC_REP_WAIT_MODE; i++)
                 SyncRepWakeQueue(true, i);
         }
 
         /*
          * Only allow people to join the queue when there are synchronous
          * standbys defined.  Without this interlock, there's a race
          * condition: we might wake up all the current waiters; then, some
          * backend that hasn't yet reloaded its config might go to sleep on
          * the queue (and never wake up).  This prevents that.
          */
         WalSndCtl->sync_standbys_defined = sync_standbys_defined;
 
         LWLockRelease(SyncRepLock);
     }
 }

void SyncRepWaitForLSN	(	XLogRecPtr	lsn,
		bool	commit
	)

Definition at line 146 of file syncrep.c.

References Assert, DestNone, ereport, errcode(), errdetail(), errmsg(), get_ps_display(), LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), Min, MyLatch, MyProc, NULL, palloc(), pfree(), PostmasterIsAlive(), ProcDiePending, QueryCancelPending, ResetLatch(), set_ps_display(), SHMQueueIsDetached(), SYNC_REP_NOT_WAITING, SYNC_REP_WAIT_COMPLETE, SYNC_REP_WAIT_FLUSH, SYNC_REP_WAITING, WalSndCtlData::sync_standbys_defined, SyncRepCancelWait(), PGPROC::syncRepLinks, SyncRepQueueInsert(), SyncRepRequested, PGPROC::syncRepState, SyncRepWaitMode, SyncStandbysDefined, update_process_title, WAIT_EVENT_SYNC_REP, WaitLatch(), PGPROC::waitLSN, WalSndCtl, WARNING, whereToSendOutput, WL_LATCH_SET, and WL_POSTMASTER_DEATH.

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

 {
     char       *new_status = NULL;
     const char *old_status;
     int         mode;
 
     /* Cap the level for anything other than commit to remote flush only. */
     if (commit)
         mode = SyncRepWaitMode;
     else
         mode = Min(SyncRepWaitMode, SYNC_REP_WAIT_FLUSH);
 
     /*
      * Fast exit if user has not requested sync replication, or there are no
      * sync replication standby names defined. Note that those standbys don't
      * need to be connected.
      */
     if (!SyncRepRequested() || !SyncStandbysDefined())
         return;
 
     Assert(SHMQueueIsDetached(&(MyProc->syncRepLinks)));
     Assert(WalSndCtl != NULL);
 
     LWLockAcquire(SyncRepLock, LW_EXCLUSIVE);
     Assert(MyProc->syncRepState == SYNC_REP_NOT_WAITING);
 
     /*
      * We don't wait for sync rep if WalSndCtl->sync_standbys_defined is not
      * set.  See SyncRepUpdateSyncStandbysDefined.
      *
      * Also check that the standby hasn't already replied. Unlikely race
      * condition but we'll be fetching that cache line anyway so it's likely
      * to be a low cost check.
      */
     if (!WalSndCtl->sync_standbys_defined ||
         lsn <= WalSndCtl->lsn[mode])
     {
         LWLockRelease(SyncRepLock);
         return;
     }
 
     /*
      * Set our waitLSN so WALSender will know when to wake us, and add
      * ourselves to the queue.
      */
     MyProc->waitLSN = lsn;
     MyProc->syncRepState = SYNC_REP_WAITING;
     SyncRepQueueInsert(mode);
     Assert(SyncRepQueueIsOrderedByLSN(mode));
     LWLockRelease(SyncRepLock);
 
     /* Alter ps display to show waiting for sync rep. */
     if (update_process_title)
     {
         int         len;
 
         old_status = get_ps_display(&len);
         new_status = (char *) palloc(len + 32 + 1);
         memcpy(new_status, old_status, len);
         sprintf(new_status + len, " waiting for %X/%X",
                 (uint32) (lsn >> 32), (uint32) lsn);
         set_ps_display(new_status, false);
         new_status[len] = '\0'; /* truncate off " waiting ..." */
     }
 
     /*
      * Wait for specified LSN to be confirmed.
      *
      * Each proc has its own wait latch, so we perform a normal latch
      * check/wait loop here.
      */
     for (;;)
     {
         /* Must reset the latch before testing state. */
         ResetLatch(MyLatch);
 
         /*
          * Acquiring the lock is not needed, the latch ensures proper
          * barriers. If it looks like we're done, we must really be done,
          * because once walsender changes the state to SYNC_REP_WAIT_COMPLETE,
          * it will never update it again, so we can't be seeing a stale value
          * in that case.
          */
         if (MyProc->syncRepState == SYNC_REP_WAIT_COMPLETE)
             break;
 
         /*
          * If a wait for synchronous replication is pending, we can neither
          * acknowledge the commit nor raise ERROR or FATAL.  The latter would
          * lead the client to believe that the transaction aborted, which is
          * not true: it's already committed locally. The former is no good
          * either: the client has requested synchronous replication, and is
          * entitled to assume that an acknowledged commit is also replicated,
          * which might not be true. So in this case we issue a WARNING (which
          * some clients may be able to interpret) and shut off further output.
          * We do NOT reset ProcDiePending, so that the process will die after
          * the commit is cleaned up.
          */
         if (ProcDiePending)
         {
             ereport(WARNING,
                     (errcode(ERRCODE_ADMIN_SHUTDOWN),
                      errmsg("canceling the wait for synchronous replication and terminating connection due to administrator command"),
                      errdetail("The transaction has already committed locally, but might not have been replicated to the standby.")));
             whereToSendOutput = DestNone;
             SyncRepCancelWait();
             break;
         }
 
         /*
          * It's unclear what to do if a query cancel interrupt arrives.  We
          * can't actually abort at this point, but ignoring the interrupt
          * altogether is not helpful, so we just terminate the wait with a
          * suitable warning.
          */
         if (QueryCancelPending)
         {
             QueryCancelPending = false;
             ereport(WARNING,
                     (errmsg("canceling wait for synchronous replication due to user request"),
                      errdetail("The transaction has already committed locally, but might not have been replicated to the standby.")));
             SyncRepCancelWait();
             break;
         }
 
         /*
          * If the postmaster dies, we'll probably never get an
          * acknowledgement, because all the wal sender processes will exit. So
          * just bail out.
          */
         if (!PostmasterIsAlive())
         {
             ProcDiePending = true;
             whereToSendOutput = DestNone;
             SyncRepCancelWait();
             break;
         }
 
         /*
          * Wait on latch.  Any condition that should wake us up will set the
          * latch, so no need for timeout.
          */
         WaitLatch(MyLatch, WL_LATCH_SET | WL_POSTMASTER_DEATH, -1,
                   WAIT_EVENT_SYNC_REP);
     }
 
     /*
      * WalSender has checked our LSN and has removed us from queue. Clean up
      * state and leave.  It's OK to reset these shared memory fields without
      * holding SyncRepLock, because any walsenders will ignore us anyway when
      * we're not on the queue.
      */
     Assert(SHMQueueIsDetached(&(MyProc->syncRepLinks)));
     MyProc->syncRepState = SYNC_REP_NOT_WAITING;
     MyProc->waitLSN = 0;
 
     if (new_status)
     {
         /* Reset ps display */
         set_ps_display(new_status, false);
         pfree(new_status);
     }
 }

static int SyncRepWakeQueue	(	bool	all,
		int	mode
	)

static

Definition at line 973 of file syncrep.c.

References Assert, WalSndCtlData::lsn, NULL, NUM_SYNC_REP_WAIT_MODE, offsetof, PGPROC::procLatch, SetLatch(), SHMQueueDelete(), SHMQueueNext(), SYNC_REP_WAIT_COMPLETE, PGPROC::syncRepLinks, WalSndCtlData::SyncRepQueue, PGPROC::syncRepState, PGPROC::waitLSN, and WalSndCtl.

Referenced by SyncRepReleaseWaiters(), and SyncRepUpdateSyncStandbysDefined().

 {
     volatile WalSndCtlData *walsndctl = WalSndCtl;
     PGPROC     *proc = NULL;
     PGPROC     *thisproc = NULL;
     int         numprocs = 0;
 
     Assert(mode >= 0 && mode < NUM_SYNC_REP_WAIT_MODE);
     Assert(SyncRepQueueIsOrderedByLSN(mode));
 
     proc = (PGPROC *) SHMQueueNext(&(WalSndCtl->SyncRepQueue[mode]),
                                    &(WalSndCtl->SyncRepQueue[mode]),
                                    offsetof(PGPROC, syncRepLinks));
 
     while (proc)
     {
         /*
          * Assume the queue is ordered by LSN
          */
         if (!all && walsndctl->lsn[mode] < proc->waitLSN)
             return numprocs;
 
         /*
          * Move to next proc, so we can delete thisproc from the queue.
          * thisproc is valid, proc may be NULL after this.
          */
         thisproc = proc;
         proc = (PGPROC *) SHMQueueNext(&(WalSndCtl->SyncRepQueue[mode]),
                                        &(proc->syncRepLinks),
                                        offsetof(PGPROC, syncRepLinks));
 
         /*
          * Set state to complete; see SyncRepWaitForLSN() for discussion of
          * the various states.
          */
         thisproc->syncRepState = SYNC_REP_WAIT_COMPLETE;
 
         /*
          * Remove thisproc from queue.
          */
         SHMQueueDelete(&(thisproc->syncRepLinks));
 
         /*
          * Wake only when we have set state and removed from queue.
          */
         SetLatch(&(thisproc->procLatch));
 
         numprocs++;
     }
 
     return numprocs;
 }