Received: from malur.postgresql.org ([217.196.149.56]) by arkaria.postgresql.org with esmtps (TLS1.3:ECDHE_RSA_AES_256_GCM_SHA384:256) (Exim 4.92) (envelope-from ) id 1jOSEo-0006QC-NJ for pgsql-hackers@arkaria.postgresql.org; Tue, 14 Apr 2020 20:32:55 +0000 Received: from localhost ([127.0.0.1] helo=malur.postgresql.org) by malur.postgresql.org with esmtp (Exim 4.92) (envelope-from ) id 1jOSEm-00027D-M9 for pgsql-hackers@arkaria.postgresql.org; Tue, 14 Apr 2020 20:32:52 +0000 Received: from makus.postgresql.org ([2001:4800:3e1:1::229]) by malur.postgresql.org with esmtps (TLS1.3:ECDHE_RSA_AES_256_GCM_SHA384:256) (Exim 4.92) (envelope-from ) id 1jOSEl-000276-IM for pgsql-hackers@lists.postgresql.org; Tue, 14 Apr 2020 20:32:51 +0000 Received: from sss.pgh.pa.us ([66.207.139.130]) by makus.postgresql.org with esmtps (TLS1.2:ECDHE_RSA_AES_256_GCM_SHA384:256) (Exim 4.92) (envelope-from ) id 1jOSEe-0003mH-Di for pgsql-hackers@lists.postgresql.org; Tue, 14 Apr 2020 20:32:50 +0000 Received: from sss1.sss.pgh.pa.us (localhost [127.0.0.1]) by sss.pgh.pa.us (8.14.4/8.14.4) with ESMTP id 03EKWe7t013060; Tue, 14 Apr 2020 16:32:40 -0400 From: Tom Lane To: Kyotaro Horiguchi cc: masahiko.sawada@2ndquadrant.com, masao.fujii@oss.nttdata.com, pgsql-hackers@lists.postgresql.org Subject: Re: Race condition in SyncRepGetSyncStandbysPriority In-reply-to: <15616.1586872362@sss.pgh.pa.us> References: <20200413.153101.688623432137444694.horikyota.ntt@gmail.com> <8325.1586828060@sss.pgh.pa.us> <20200414.183538.344904915964423903.horikyota.ntt@gmail.com> <15616.1586872362@sss.pgh.pa.us> Comments: In-reply-to Tom Lane message dated "Tue, 14 Apr 2020 09:52:42 -0400" MIME-Version: 1.0 Content-Type: multipart/mixed; boundary="----- =_aaaaaaaaaa0" Content-ID: <13035.1586896319.0@sss.pgh.pa.us> Date: Tue, 14 Apr 2020 16:32:40 -0400 Message-ID: <13059.1586896360@sss.pgh.pa.us> List-Id: List-Help: List-Subscribe: List-Post: List-Owner: List-Archive: Precedence: bulk ------- =_aaaaaaaaaa0 Content-Type: text/plain; charset="us-ascii" Content-ID: <13035.1586896319.1@sss.pgh.pa.us> I wrote: > It doesn't seem to me to be that hard to implement the desired > semantics for synchronous_standby_names with inconsistent info. > In FIRST mode you basically just need to take the N smallest > priorities you see in the array, but without assuming there are no > duplicates or holes. It might be a good idea to include ties at the > end, that is if you see 1,2,2,4 or 1,3,3,4 and you want 2 sync > standbys, include the first three of them in the calculation until > the inconsistency is resolved. In ANY mode I don't see that > inconsistent priorities matter at all. Concretely, I think we ought to do the attached, or something pretty close to it. I'm not really happy about breaking ties based on walsnd_index, but I see that there are several TAP test cases that fail if we do something else. I'm inclined to think those tests are bogus ... but I won't argue to change them right now. regards, tom lane ------- =_aaaaaaaaaa0 Content-Type: text/x-diff; name="syncrep-fixes-1.patch"; charset="us-ascii" Content-ID: <13035.1586896319.2@sss.pgh.pa.us> Content-Description: syncrep-fixes-1.patch Content-Transfer-Encoding: quoted-printable diff --git a/src/backend/replication/syncrep.c b/src/backend/replication/s= yncrep.c index ffd5b31..c66c371 100644 --- a/src/backend/replication/syncrep.c +++ b/src/backend/replication/syncrep.c @@ -108,14 +108,17 @@ static bool SyncRepGetSyncRecPtr(XLogRecPtr *writePt= r, static void SyncRepGetOldestSyncRecPtr(XLogRecPtr *writePtr, XLogRecPtr *flushPtr, XLogRecPtr *applyPtr, - List *sync_standbys); + SyncRepStandbyData *sync_standbys, + int num_standbys); static void SyncRepGetNthLatestSyncRecPtr(XLogRecPtr *writePtr, XLogRecPtr *flushPtr, XLogRecPtr *applyPtr, - List *sync_standbys, uint8 nth); + SyncRepStandbyData *sync_standbys, + int num_standbys, + uint8 nth); static int SyncRepGetStandbyPriority(void); -static List *SyncRepGetSyncStandbysPriority(bool *am_sync); -static List *SyncRepGetSyncStandbysQuorum(bool *am_sync); +static void SyncRepGetSyncStandbysPriority(SyncRepStandbyData *standbys, = int n); +static int standby_priority_comparator(const void *a, const void *b); static int cmp_lsn(const void *a, const void *b); = #ifdef USE_ASSERT_CHECKING @@ -406,9 +409,10 @@ SyncRepInitConfig(void) priority =3D SyncRepGetStandbyPriority(); if (MyWalSnd->sync_standby_priority !=3D priority) { - LWLockAcquire(SyncRepLock, LW_EXCLUSIVE); + SpinLockAcquire(&MyWalSnd->mutex); MyWalSnd->sync_standby_priority =3D priority; - LWLockRelease(SyncRepLock); + SpinLockRelease(&MyWalSnd->mutex); + ereport(DEBUG1, (errmsg("standby \"%s\" now has synchronous standby priority %u", application_name, priority))); @@ -523,8 +527,6 @@ SyncRepReleaseWaiters(void) /* * Calculate the synced Write, Flush and Apply positions among sync stand= bys. * - * The caller must hold SyncRepLock. - * * Return false if the number of sync standbys is less than * synchronous_standby_names specifies. Otherwise return true and * store the positions into *writePtr, *flushPtr and *applyPtr. @@ -536,27 +538,43 @@ static bool SyncRepGetSyncRecPtr(XLogRecPtr *writePtr, XLogRecPtr *flushPtr, XLogRecPtr *applyPtr, bool *am_sync) { - List *sync_standbys; - - Assert(LWLockHeldByMe(SyncRepLock)); + SyncRepStandbyData *sync_standbys; + int num_standbys; + int i; = + /* Initialize default results */ *writePtr =3D InvalidXLogRecPtr; *flushPtr =3D InvalidXLogRecPtr; *applyPtr =3D InvalidXLogRecPtr; *am_sync =3D false; = + /* Quick out if not even configured to be synchronous */ + if (SyncRepConfig =3D=3D NULL) + return false; + /* Get standbys that are considered as synchronous at this moment */ - sync_standbys =3D SyncRepGetSyncStandbys(am_sync); + num_standbys =3D SyncRepGetSyncStandbys(&sync_standbys); + + /* Am I among the candidate sync standbys? */ + for (i =3D 0; i < num_standbys; i++) + { + if (sync_standbys[i].is_me) + { + *am_sync =3D sync_standbys[i].is_sync_standby; + break; + } + } = /* - * Quick exit if we are not managing a sync standby or there are not - * enough synchronous standbys. + * Nothing more to do if we are not managing a sync standby or there are + * not enough synchronous standbys. (Note: if there are least num_sync + * candidates, then at least num_sync of them will be marked as + * is_sync_standby; we don't need to count them here.) */ if (!(*am_sync) || - SyncRepConfig =3D=3D NULL || - list_length(sync_standbys) < SyncRepConfig->num_sync) + num_standbys < SyncRepConfig->num_sync) { - list_free(sync_standbys); + pfree(sync_standbys); return false; } = @@ -576,15 +594,16 @@ SyncRepGetSyncRecPtr(XLogRecPtr *writePtr, XLogRecPt= r *flushPtr, if (SyncRepConfig->syncrep_method =3D=3D SYNC_REP_PRIORITY) { SyncRepGetOldestSyncRecPtr(writePtr, flushPtr, applyPtr, - sync_standbys); + sync_standbys, num_standbys); } else { SyncRepGetNthLatestSyncRecPtr(writePtr, flushPtr, applyPtr, - sync_standbys, SyncRepConfig->num_sync); + sync_standbys, num_standbys, + SyncRepConfig->num_sync); } = - list_free(sync_standbys); + pfree(sync_standbys); return true; } = @@ -592,27 +611,28 @@ SyncRepGetSyncRecPtr(XLogRecPtr *writePtr, XLogRecPt= r *flushPtr, * Calculate the oldest Write, Flush and Apply positions among sync stand= bys. */ static void -SyncRepGetOldestSyncRecPtr(XLogRecPtr *writePtr, XLogRecPtr *flushPtr, - XLogRecPtr *applyPtr, List *sync_standbys) +SyncRepGetOldestSyncRecPtr(XLogRecPtr *writePtr, + XLogRecPtr *flushPtr, + XLogRecPtr *applyPtr, + SyncRepStandbyData *sync_standbys, + int num_standbys) { - ListCell *cell; + int i; = /* * Scan through all sync standbys and calculate the oldest Write, Flush - * and Apply positions. + * and Apply positions. We assume *writePtr et al were initialized to + * InvalidXLogRecPtr. */ - foreach(cell, sync_standbys) + for (i =3D 0; i < num_standbys; i++) { - WalSnd *walsnd =3D &WalSndCtl->walsnds[lfirst_int(cell)]; - XLogRecPtr write; - XLogRecPtr flush; - XLogRecPtr apply; + XLogRecPtr write =3D sync_standbys[i].write; + XLogRecPtr flush =3D sync_standbys[i].flush; + XLogRecPtr apply =3D sync_standbys[i].apply; = - SpinLockAcquire(&walsnd->mutex); - write =3D walsnd->write; - flush =3D walsnd->flush; - apply =3D walsnd->apply; - SpinLockRelease(&walsnd->mutex); + /* Ignore candidates that aren't considered synchronous */ + if (!sync_standbys[i].is_sync_standby) + continue; = if (XLogRecPtrIsInvalid(*writePtr) || *writePtr > write) *writePtr =3D write; @@ -628,38 +648,43 @@ SyncRepGetOldestSyncRecPtr(XLogRecPtr *writePtr, XLo= gRecPtr *flushPtr, * standbys. */ static void -SyncRepGetNthLatestSyncRecPtr(XLogRecPtr *writePtr, XLogRecPtr *flushPtr, - XLogRecPtr *applyPtr, List *sync_standbys, uint8 nth) +SyncRepGetNthLatestSyncRecPtr(XLogRecPtr *writePtr, + XLogRecPtr *flushPtr, + XLogRecPtr *applyPtr, + SyncRepStandbyData *sync_standbys, + int num_standbys, + uint8 nth) { - ListCell *cell; XLogRecPtr *write_array; XLogRecPtr *flush_array; XLogRecPtr *apply_array; - int len; - int i =3D 0; + int i; + int n; = - len =3D list_length(sync_standbys); - write_array =3D (XLogRecPtr *) palloc(sizeof(XLogRecPtr) * len); - flush_array =3D (XLogRecPtr *) palloc(sizeof(XLogRecPtr) * len); - apply_array =3D (XLogRecPtr *) palloc(sizeof(XLogRecPtr) * len); + write_array =3D (XLogRecPtr *) palloc(sizeof(XLogRecPtr) * num_standbys)= ; + flush_array =3D (XLogRecPtr *) palloc(sizeof(XLogRecPtr) * num_standbys)= ; + apply_array =3D (XLogRecPtr *) palloc(sizeof(XLogRecPtr) * num_standbys)= ; = - foreach(cell, sync_standbys) + n =3D 0; + for (i =3D 0; i < num_standbys; i++) { - WalSnd *walsnd =3D &WalSndCtl->walsnds[lfirst_int(cell)]; - - SpinLockAcquire(&walsnd->mutex); - write_array[i] =3D walsnd->write; - flush_array[i] =3D walsnd->flush; - apply_array[i] =3D walsnd->apply; - SpinLockRelease(&walsnd->mutex); + /* Ignore candidates that aren't considered synchronous */ + if (!sync_standbys[i].is_sync_standby) + continue; = - i++; + write_array[n] =3D sync_standbys[i].write; + flush_array[n] =3D sync_standbys[i].flush; + apply_array[n] =3D sync_standbys[i].apply; + n++; } = + /* Should have enough, or somebody messed up */ + Assert(n >=3D nth); + /* 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); + qsort(write_array, n, sizeof(XLogRecPtr), cmp_lsn); + qsort(flush_array, n, sizeof(XLogRecPtr), cmp_lsn); + qsort(apply_array, n, sizeof(XLogRecPtr), cmp_lsn); = /* Get Nth latest Write, Flush, Apply positions */ *writePtr =3D write_array[nth - 1]; @@ -689,67 +714,48 @@ cmp_lsn(const void *a, const void *b) } = /* - * Return the list of sync standbys, or NIL if no sync standby is connect= ed. - * - * The caller must hold SyncRepLock. + * Return data about walsenders that are candidates to be sync standbys. * - * On return, *am_sync is set to true if this walsender is connecting to - * sync standby. Otherwise it's set to false. + * *standbys is set to a palloc'd array of structs of per-walsender data, + * and the number of valid entries (candidate sync senders) is returned. */ -List * -SyncRepGetSyncStandbys(bool *am_sync) +int +SyncRepGetSyncStandbys(SyncRepStandbyData **standbys) { - Assert(LWLockHeldByMe(SyncRepLock)); + int i; + int n; = - /* Set default result */ - if (am_sync !=3D NULL) - *am_sync =3D false; + /* Create result array */ + *standbys =3D (SyncRepStandbyData *) + palloc(max_wal_senders * sizeof(SyncRepStandbyData)); = /* Quick exit if sync replication is not requested */ if (SyncRepConfig =3D=3D NULL) - return NIL; - - return (SyncRepConfig->syncrep_method =3D=3D SYNC_REP_PRIORITY) ? - SyncRepGetSyncStandbysPriority(am_sync) : - SyncRepGetSyncStandbysQuorum(am_sync); -} - -/* - * Return the list of all the candidates for quorum sync standbys, - * or NIL if no such standby is connected. - * - * The caller must hold SyncRepLock. This function must be called only in - * a quorum-based sync replication. - * - * On return, *am_sync is set to true if this walsender is connecting to - * sync standby. Otherwise it's set to false. - */ -static List * -SyncRepGetSyncStandbysQuorum(bool *am_sync) -{ - List *result =3D NIL; - int i; - volatile WalSnd *walsnd; /* Use volatile pointer to prevent code - * rearrangement */ - - Assert(SyncRepConfig->syncrep_method =3D=3D SYNC_REP_QUORUM); + return false; = + /* Collect raw data from shared memory */ + n =3D 0; for (i =3D 0; i < max_wal_senders; i++) { - XLogRecPtr flush; - WalSndState state; - int pid; + volatile WalSnd *walsnd; /* Use volatile pointer to prevent code + * rearrangement */ + SyncRepStandbyData *stby; + WalSndState state; /* not included in SyncRepStandbyData */ = walsnd =3D &WalSndCtl->walsnds[i]; + stby =3D *standbys + n; = SpinLockAcquire(&walsnd->mutex); - pid =3D walsnd->pid; - flush =3D walsnd->flush; + stby->pid =3D walsnd->pid; state =3D walsnd->state; + stby->write =3D walsnd->write; + stby->flush =3D walsnd->flush; + stby->apply =3D walsnd->apply; + stby->sync_standby_priority =3D walsnd->sync_standby_priority; SpinLockRelease(&walsnd->mutex); = /* Must be active */ - if (pid =3D=3D 0) + if (stby->pid =3D=3D 0) continue; = /* Must be streaming or stopping */ @@ -758,200 +764,79 @@ SyncRepGetSyncStandbysQuorum(bool *am_sync) continue; = /* Must be synchronous */ - if (walsnd->sync_standby_priority =3D=3D 0) + if (stby->sync_standby_priority =3D=3D 0) continue; = /* Must have a valid flush position */ - if (XLogRecPtrIsInvalid(flush)) + if (XLogRecPtrIsInvalid(stby->flush)) continue; = - /* - * Consider this standby as a candidate for quorum sync standbys and - * append it to the result. - */ - result =3D lappend_int(result, i); - if (am_sync !=3D NULL && walsnd =3D=3D MyWalSnd) - *am_sync =3D true; + /* OK, it's a candidate */ + stby->walsnd_index =3D i; + stby->is_me =3D (walsnd =3D=3D MyWalSnd); + stby->is_sync_standby =3D true; /* might change below */ + n++; } = - return result; + /* + * In quorum mode, that's all we have to do. In priority mode, decide + * which ones are high enough priority to consider sync standbys. + */ + if (SyncRepConfig->syncrep_method =3D=3D SYNC_REP_PRIORITY) + SyncRepGetSyncStandbysPriority(*standbys, n); + + return n; } = /* - * Return the list of sync standbys chosen based on their priorities, - * or NIL if no sync standby is connected. - * - * If there are multiple standbys with the same priority, - * the first one found is selected preferentially. - * - * The caller must hold SyncRepLock. This function must be called only in - * a priority-based sync replication. + * Decide which standbys to consider synchronous. * - * On return, *am_sync is set to true if this walsender is connecting to - * sync standby. Otherwise it's set to false. + * This function must be called only in priority-based sync replication. */ -static List * -SyncRepGetSyncStandbysPriority(bool *am_sync) +static void +SyncRepGetSyncStandbysPriority(SyncRepStandbyData *standbys, int n) { - List *result =3D NIL; - List *pending =3D NIL; - int lowest_priority; - int next_highest_priority; - int this_priority; - int priority; int i; - bool am_in_pending =3D false; - volatile WalSnd *walsnd; /* Use volatile pointer to prevent code - * rearrangement */ = Assert(SyncRepConfig->syncrep_method =3D=3D SYNC_REP_PRIORITY); = - lowest_priority =3D SyncRepConfig->nmembers; - next_highest_priority =3D lowest_priority + 1; + /* Nothing to do here unless there are too many candidates. */ + if (n <=3D SyncRepConfig->num_sync) + return; = /* - * 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. + * Sort the candidates by priority; then the first num_sync ones are + * synchronous, and the rest aren't. */ - for (i =3D 0; i < max_wal_senders; i++) - { - XLogRecPtr flush; - WalSndState state; - int pid; - - walsnd =3D &WalSndCtl->walsnds[i]; - - SpinLockAcquire(&walsnd->mutex); - pid =3D walsnd->pid; - flush =3D walsnd->flush; - state =3D walsnd->state; - SpinLockRelease(&walsnd->mutex); - - /* Must be active */ - if (pid =3D=3D 0) - continue; + qsort(standbys, n, sizeof(SyncRepStandbyData), + standby_priority_comparator); = - /* Must be streaming or stopping */ - if (state !=3D WALSNDSTATE_STREAMING && - state !=3D WALSNDSTATE_STOPPING) - continue; - - /* Must be synchronous */ - this_priority =3D walsnd->sync_standby_priority; - if (this_priority =3D=3D 0) - continue; - - /* Must have a valid flush position */ - if (XLogRecPtrIsInvalid(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 =3D=3D 1) - { - result =3D lappend_int(result, i); - if (am_sync !=3D NULL && walsnd =3D=3D MyWalSnd) - *am_sync =3D true; - if (list_length(result) =3D=3D SyncRepConfig->num_sync) - { - list_free(pending); - return result; /* Exit if got enough sync standbys */ - } - } - else - { - pending =3D lappend_int(pending, i); - if (am_sync !=3D NULL && walsnd =3D=3D MyWalSnd) - am_in_pending =3D 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 =3D this_priority; - } - } + for (i =3D SyncRepConfig->num_sync; i < n; i++) + standbys[i].is_sync_standby =3D false; +} = - /* - * 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) <=3D SyncRepConfig->num_s= ync) - { - /* - * Set *am_sync to true if this walsender is in the pending list - * because all pending standbys are considered as sync. - */ - if (am_sync !=3D NULL && !(*am_sync)) - *am_sync =3D am_in_pending; +/* + * qsort comparator to sort SyncRepStandbyData entries by priority + */ +static int +standby_priority_comparator(const void *a, const void *b) +{ + const SyncRepStandbyData *sa =3D (const SyncRepStandbyData *) a; + const SyncRepStandbyData *sb =3D (const SyncRepStandbyData *) b; = - result =3D list_concat(result, pending); - list_free(pending); - return result; - } + /* First, sort by increasing priority value */ + if (sa->sync_standby_priority !=3D sb->sync_standby_priority) + return sa->sync_standby_priority - sb->sync_standby_priority; = /* - * Find the sync standbys from the pending list. + * We might have equal priority values; arbitrarily break ties by positi= on + * in the WALSnd array. (This is utterly bogus, since that is arrival + * order dependent, but there are regression tests that rely on it.) */ - priority =3D next_highest_priority; - while (priority <=3D lowest_priority) - { - ListCell *cell; - - next_highest_priority =3D lowest_priority + 1; - - foreach(cell, pending) - { - i =3D lfirst_int(cell); - walsnd =3D &WalSndCtl->walsnds[i]; - - this_priority =3D walsnd->sync_standby_priority; - if (this_priority =3D=3D priority) - { - result =3D lappend_int(result, i); - if (am_sync !=3D NULL && walsnd =3D=3D MyWalSnd) - *am_sync =3D 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) =3D=3D 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 =3D foreach_delete_current(pending, cell); - - continue; /* don't adjust next_highest_priority */ - } - - if (this_priority < next_highest_priority) - next_highest_priority =3D this_priority; - } - - priority =3D next_highest_priority; - } - - /* never reached, but keep compiler quiet */ - Assert(false); - return result; + return sa->walsnd_index - sb->walsnd_index; } = + /* * Check if we are in the list of sync standbys, and if so, determine * priority sequence. Return priority if set, or zero to indicate that diff --git a/src/backend/replication/walsender.c b/src/backend/replication= /walsender.c index fc475d1..859ca60 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -2375,14 +2375,16 @@ InitWalSenderSlot(void) * Found a free slot. Reserve it for us. */ walsnd->pid =3D MyProcPid; + walsnd->state =3D WALSNDSTATE_STARTUP; walsnd->sentPtr =3D InvalidXLogRecPtr; + walsnd->needreload =3D false; walsnd->write =3D InvalidXLogRecPtr; walsnd->flush =3D InvalidXLogRecPtr; walsnd->apply =3D InvalidXLogRecPtr; walsnd->writeLag =3D -1; walsnd->flushLag =3D -1; walsnd->applyLag =3D -1; - walsnd->state =3D WALSNDSTATE_STARTUP; + walsnd->sync_standby_priority =3D 0; walsnd->latch =3D &MyProc->procLatch; walsnd->replyTime =3D 0; walsnd->spillTxns =3D 0; @@ -3235,7 +3237,8 @@ pg_stat_get_wal_senders(PG_FUNCTION_ARGS) Tuplestorestate *tupstore; MemoryContext per_query_ctx; MemoryContext oldcontext; - List *sync_standbys; + SyncRepStandbyData *sync_standbys; + int num_standbys; int i; = /* check to see if caller supports us returning a tuplestore */ @@ -3263,11 +3266,10 @@ pg_stat_get_wal_senders(PG_FUNCTION_ARGS) MemoryContextSwitchTo(oldcontext); = /* - * Get the currently active synchronous standbys. + * Get the currently active synchronous standbys. This could be out of + * date before we're done, but we'll use the data anyway. */ - LWLockAcquire(SyncRepLock, LW_SHARED); - sync_standbys =3D SyncRepGetSyncStandbys(NULL); - LWLockRelease(SyncRepLock); + num_standbys =3D SyncRepGetSyncStandbys(&sync_standbys); = for (i =3D 0; i < max_wal_senders; i++) { @@ -3286,9 +3288,12 @@ pg_stat_get_wal_senders(PG_FUNCTION_ARGS) int64 spillTxns; int64 spillCount; int64 spillBytes; + bool is_sync_standby; Datum values[PG_STAT_GET_WAL_SENDERS_COLS]; bool nulls[PG_STAT_GET_WAL_SENDERS_COLS]; + int j; = + /* Collect data from shared memory */ SpinLockAcquire(&walsnd->mutex); if (walsnd->pid =3D=3D 0) { @@ -3311,6 +3316,17 @@ pg_stat_get_wal_senders(PG_FUNCTION_ARGS) spillBytes =3D walsnd->spillBytes; SpinLockRelease(&walsnd->mutex); = + /* Detect whether walsender is/was considered synchronous */ + is_sync_standby =3D false; + for (j =3D 0; j < num_standbys; j++) + { + if (sync_standbys[j].walsnd_index =3D=3D i) + { + is_sync_standby =3D sync_standbys[j].is_sync_standby; + break; + } + } + memset(nulls, 0, sizeof(nulls)); values[0] =3D Int32GetDatum(pid); = @@ -3380,7 +3396,7 @@ pg_stat_get_wal_senders(PG_FUNCTION_ARGS) */ if (priority =3D=3D 0) values[10] =3D CStringGetTextDatum("async"); - else if (list_member_int(sync_standbys, i)) + else if (is_sync_standby) values[10] =3D SyncRepConfig->syncrep_method =3D=3D SYNC_REP_PRIORITY= ? CStringGetTextDatum("sync") : CStringGetTextDatum("quorum"); else diff --git a/src/include/replication/syncrep.h b/src/include/replication/s= yncrep.h index c5f0e91..1308ada 100644 --- a/src/include/replication/syncrep.h +++ b/src/include/replication/syncrep.h @@ -37,6 +37,26 @@ #define SYNC_REP_QUORUM 1 = /* + * SyncRepGetSyncStandbys returns an array of these structs, + * one per candidate synchronous walsender. + */ +typedef struct SyncRepStandbyData +{ + /* Copies of relevant fields from WalSnd shared-memory struct */ + pid_t pid; + XLogRecPtr write; + XLogRecPtr flush; + XLogRecPtr apply; + int sync_standby_priority; + /* Index of this walsender in the WalSnd shared-memory array */ + int walsnd_index; + /* This flag indicates whether this struct is about our own process */ + bool is_me; + /* After-the-fact conclusion about whether this is a sync standby */ + bool is_sync_standby; +} SyncRepStandbyData; + +/* * Struct for the configuration of synchronous replication. * * Note: this must be a flat representation that can be held in a single @@ -74,7 +94,7 @@ extern void SyncRepInitConfig(void); extern void SyncRepReleaseWaiters(void); = /* called by wal sender and user backend */ -extern List *SyncRepGetSyncStandbys(bool *am_sync); +extern int SyncRepGetSyncStandbys(SyncRepStandbyData **standbys); = /* called by checkpointer */ extern void SyncRepUpdateSyncStandbysDefined(void); diff --git a/src/include/replication/walsender_private.h b/src/include/rep= lication/walsender_private.h index 366828f..734acec 100644 --- a/src/include/replication/walsender_private.h +++ b/src/include/replication/walsender_private.h @@ -31,8 +31,7 @@ typedef enum WalSndState /* * Each walsender has a WalSnd struct in shared memory. * - * This struct is protected by 'mutex', with two exceptions: one is - * sync_standby_priority as noted below. The other exception is that som= e + * This struct is protected by its 'mutex' spinlock field, except that so= me * members are only written by the walsender process itself, and thus tha= t * process is free to read those members without holding spinlock. pid a= nd * needreload always require the spinlock to be held for all accesses. @@ -60,6 +59,12 @@ typedef struct WalSnd TimeOffset flushLag; TimeOffset applyLag; = + /* + * The priority order of the standby managed by this WALSender, as liste= d + * in synchronous_standby_names, or 0 if not-listed. + */ + int sync_standby_priority; + /* Protects shared variables shown above. */ slock_t mutex; = @@ -70,13 +75,6 @@ typedef struct WalSnd Latch *latch; = /* - * The priority order of the standby managed by this WALSender, as liste= d - * in synchronous_standby_names, or 0 if not-listed. Protected by - * SyncRepLock. - */ - int sync_standby_priority; - - /* * Timestamp of the last message received from standby. */ TimestampTz replyTime; ------- =_aaaaaaaaaa0--