mirrors/postgres
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Replace known_assigned_xids_lck with memory barriers.

Browse Source 
This lock was introduced before memory barrier support was added,
and it is only used to guarantee proper memory ordering when
KnownAssignedXidsAdd() appends to the array without a lock.  Now
that such memory barrier support exists, we can remove the lock and
use barriers instead.

Suggested-by: Tom Lane
Author: Michail Nikolaev
Reviewed-by: Robert Haas
Discussion: https://postgr.es/m/CANtu0oh0si%3DjG5z_fLeFtmYcETssQ08kLEa8b6TQqDm_cinroA%40mail.gmail.com
This commit is contained in:
Nathan Bossart 2023-09-05 13:59:06 -07:00
parent a899d07b24
commit 119c23eb98
1 changed files with 28 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

70
src/backend/storage/ipc/procarray.c
View File

@ -61,7 +61,6 @@
#include "port/pg_lfind.h" #include "port/pg_lfind.h"
#include "storage/proc.h" #include "storage/proc.h"
#include "storage/procarray.h" #include "storage/procarray.h"
#include "storage/spin.h"
#include "utils/acl.h" #include "utils/acl.h"
#include "utils/builtins.h" #include "utils/builtins.h"
#include "utils/rel.h" #include "utils/rel.h"
@ -82,7 +81,6 @@ typedef struct ProcArrayStruct
int numKnownAssignedXids; /* current # of valid entries */ int numKnownAssignedXids; /* current # of valid entries */
int tailKnownAssignedXids; /* index of oldest valid element */ int tailKnownAssignedXids; /* index of oldest valid element */
int headKnownAssignedXids; /* index of newest element, + 1 */ int headKnownAssignedXids; /* index of newest element, + 1 */
slock_t known_assigned_xids_lck; /* protects head/tail pointers */
/* /*
* Highest subxid that has been removed from KnownAssignedXids array to * Highest subxid that has been removed from KnownAssignedXids array to
@ -441,7 +439,6 @@ CreateSharedProcArray(void)
procArray->numKnownAssignedXids = 0; procArray->numKnownAssignedXids = 0;
procArray->tailKnownAssignedXids = 0; procArray->tailKnownAssignedXids = 0;
procArray->headKnownAssignedXids = 0; procArray->headKnownAssignedXids = 0;
SpinLockInit(&procArray->known_assigned_xids_lck);
procArray->lastOverflowedXid = InvalidTransactionId; procArray->lastOverflowedXid = InvalidTransactionId;
procArray->replication_slot_xmin = InvalidTransactionId; procArray->replication_slot_xmin = InvalidTransactionId;
procArray->replication_slot_catalog_xmin = InvalidTransactionId; procArray->replication_slot_catalog_xmin = InvalidTransactionId;
@ -4533,22 +4530,19 @@ KnownAssignedTransactionIdsIdleMaintenance(void)
* during normal running). Compressing unused entries out of the array * during normal running). Compressing unused entries out of the array
* likewise requires exclusive lock. To add XIDs to the array, we just insert * likewise requires exclusive lock. To add XIDs to the array, we just insert
* them into slots to the right of the head pointer and then advance the head * them into slots to the right of the head pointer and then advance the head
* pointer. This wouldn't require any lock at all, except that on machines * pointer. This doesn't require any lock at all, but on machines with weak
* with weak memory ordering we need to be careful that other processors * memory ordering, we need to be careful that other processors see the array
* see the array element changes before they see the head pointer change. * element changes before they see the head pointer change. We handle this by
* We handle this by using a spinlock to protect reads and writes of the * using memory barriers when reading or writing the head/tail pointers (unless
* head/tail pointers. (We could dispense with the spinlock if we were to * the caller holds ProcArrayLock exclusively).
* create suitable memory access barrier primitives and use those instead.)
* The spinlock must be taken to read or write the head/tail pointers unless
* the caller holds ProcArrayLock exclusively.
* *
* Algorithmic analysis: * Algorithmic analysis:
* *
* If we have a maximum of M slots, with N XIDs currently spread across * If we have a maximum of M slots, with N XIDs currently spread across
* S elements then we have N <= S <= M always. * S elements then we have N <= S <= M always.
* *
* * Adding a new XID is O(1) and needs little locking (unless compression * * Adding a new XID is O(1) and needs no lock (unless compression must
* must happen) * happen)
* * Compressing the array is O(S) and requires exclusive lock * * Compressing the array is O(S) and requires exclusive lock
* * Removing an XID is O(logS) and requires exclusive lock * * Removing an XID is O(logS) and requires exclusive lock
* * Taking a snapshot is O(S) and requires shared lock * * Taking a snapshot is O(S) and requires shared lock
@ -4778,22 +4772,15 @@ KnownAssignedXidsAdd(TransactionId from_xid, TransactionId to_xid,
pArray->numKnownAssignedXids += nxids; pArray->numKnownAssignedXids += nxids;
/* /*
* Now update the head pointer. We use a spinlock to protect this * Now update the head pointer. We use a write barrier to ensure that
* pointer, not because the update is likely to be non-atomic, but to * other processors see the above array updates before they see the head
* ensure that other processors see the above array updates before they * pointer change. The barrier isn't required if we're holding
* see the head pointer change. * ProcArrayLock exclusively.
*
* If we're holding ProcArrayLock exclusively, there's no need to take the
* spinlock.
*/ */
if (exclusive_lock) if (!exclusive_lock)
pg_write_barrier();
pArray->headKnownAssignedXids = head; pArray->headKnownAssignedXids = head;
else
{
SpinLockAcquire(&pArray->known_assigned_xids_lck);
pArray->headKnownAssignedXids = head;
SpinLockRelease(&pArray->known_assigned_xids_lck);
}
} }
/* /*
@ -4815,20 +4802,15 @@ KnownAssignedXidsSearch(TransactionId xid, bool remove)
int tail; int tail;
int result_index = -1; int result_index = -1;
if (remove)
{
/* we hold ProcArrayLock exclusively, so no need for spinlock */
tail = pArray->tailKnownAssignedXids; tail = pArray->tailKnownAssignedXids;
head = pArray->headKnownAssignedXids; head = pArray->headKnownAssignedXids;
}
else /*
{ * Only the startup process removes entries, so we don't need the read
/* take spinlock to ensure we see up-to-date array contents */ * barrier in that case.
SpinLockAcquire(&pArray->known_assigned_xids_lck); */
tail = pArray->tailKnownAssignedXids; if (!remove)
head = pArray->headKnownAssignedXids; pg_read_barrier(); /* pairs with KnownAssignedXidsAdd */
SpinLockRelease(&pArray->known_assigned_xids_lck);
}
/* /*
* Standard binary search. Note we can ignore the KnownAssignedXidsValid * Standard binary search. Note we can ignore the KnownAssignedXidsValid
@ -5066,13 +5048,11 @@ KnownAssignedXidsGetAndSetXmin(TransactionId *xarray, TransactionId *xmin,
* cannot enter and then leave the array while we hold ProcArrayLock. We * cannot enter and then leave the array while we hold ProcArrayLock. We
* might miss newly-added xids, but they should be >= xmax so irrelevant * might miss newly-added xids, but they should be >= xmax so irrelevant
* anyway. * anyway.
*
* Must take spinlock to ensure we see up-to-date array contents.
*/ */
SpinLockAcquire(&procArray->known_assigned_xids_lck);
tail = procArray->tailKnownAssignedXids; tail = procArray->tailKnownAssignedXids;
head = procArray->headKnownAssignedXids; head = procArray->headKnownAssignedXids;
SpinLockRelease(&procArray->known_assigned_xids_lck);
pg_read_barrier(); /* pairs with KnownAssignedXidsAdd */
for (i = tail; i < head; i++) for (i = tail; i < head; i++)
{ {
@ -5119,10 +5099,10 @@ KnownAssignedXidsGetOldestXmin(void)
/* /*
* Fetch head just once, since it may change while we loop. * Fetch head just once, since it may change while we loop.
*/ */
SpinLockAcquire(&procArray->known_assigned_xids_lck);
tail = procArray->tailKnownAssignedXids; tail = procArray->tailKnownAssignedXids;
head = procArray->headKnownAssignedXids; head = procArray->headKnownAssignedXids;
SpinLockRelease(&procArray->known_assigned_xids_lck);
pg_read_barrier(); /* pairs with KnownAssignedXidsAdd */
for (i = tail; i < head; i++) for (i = tail; i < head; i++)
{ {