fix_concurrent_slot_xmin_update_version2.patch
application/octet-stream
Filename: fix_concurrent_slot_xmin_update_version2.patch
Type: application/octet-stream
Part: 0
Patch
Same data as JSON:
GET /api/v1/attachments/:id/patch
the parsed metadata as JSON — format, series position, per-file stats; never the diff bytes.
API reference →
Format: unified
| File | + | − |
|---|---|---|
| src/backend/replication/logical/logical.c | 7 | 5 |
| src/backend/replication/logical/slotsync.c | 3 | 1 |
| src/backend/replication/slot.c | 27 | 5 |
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index f1eb798f3e9..e3a2a1ffdd4 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -405,11 +405,11 @@ CreateInitDecodingContext(const char *plugin,
* without further interlock its return value might immediately be out of
* date.
*
- * So we have to acquire the ProcArrayLock to prevent computation of new
- * xmin horizons by other backends, get the safe decoding xid, and inform
- * the slot machinery about the new limit. Once that's done the
- * ProcArrayLock can be released as the slot machinery now is
- * protecting against vacuum.
+ * So we have to acquire both the ReplicationSlotControlLock and the
+ * ProcArrayLock to prevent concurrent computation and update of new xmin
+ * horizons by other backends, get the safe decoding xid, and inform the
+ * slot machinery about the new limit. Once that's done the both locks
+ * can be released as the slot machinery now is protecting against vacuum.
*
* Note that, temporarily, the data, not just the catalog, xmin has to be
* reserved if a data snapshot is to be exported. Otherwise the initial
@@ -422,6 +422,7 @@ CreateInitDecodingContext(const char *plugin,
*
* ----
*/
+ LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
xmin_horizon = GetOldestSafeDecodingTransactionId(!need_full_snapshot);
@@ -436,6 +437,7 @@ CreateInitDecodingContext(const char *plugin,
ReplicationSlotsComputeRequiredXmin(true);
LWLockRelease(ProcArrayLock);
+ LWLockRelease(ReplicationSlotControlLock);
ReplicationSlotMarkDirty();
ReplicationSlotSave();
diff --git a/src/backend/replication/logical/slotsync.c b/src/backend/replication/logical/slotsync.c
index 61b2e9396aa..fee5aad7574 100644
--- a/src/backend/replication/logical/slotsync.c
+++ b/src/backend/replication/logical/slotsync.c
@@ -775,7 +775,8 @@ synchronize_one_slot(RemoteSlot *remote_slot, Oid remote_dbid)
SpinLockRelease(&slot->mutex);
reserve_wal_for_local_slot(remote_slot->restart_lsn);
-
+
+ LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
xmin_horizon = GetOldestSafeDecodingTransactionId(true);
SpinLockAcquire(&slot->mutex);
@@ -784,6 +785,7 @@ synchronize_one_slot(RemoteSlot *remote_slot, Oid remote_dbid)
SpinLockRelease(&slot->mutex);
ReplicationSlotsComputeRequiredXmin(true);
LWLockRelease(ProcArrayLock);
+ LWLockRelease(ReplicationSlotControlLock);
update_and_persist_local_synced_slot(remote_slot, remote_dbid);
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 101157ed8c9..303a40e61ea 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -1129,8 +1129,11 @@ ReplicationSlotPersist(void)
/*
* Compute the oldest xmin across all slots and store it in the ProcArray.
*
- * If already_locked is true, ProcArrayLock has already been acquired
- * exclusively.
+ * If already_locked is true, both the ReplicationSlotControlLock and
+ * the ProcArrayLock have already been acquired exclusively.
+ *
+ * Note that the ReplicationSlotControlLock must be locked first to avoid
+ * deadlocks.
*/
void
ReplicationSlotsComputeRequiredXmin(bool already_locked)
@@ -1140,8 +1143,26 @@ ReplicationSlotsComputeRequiredXmin(bool already_locked)
TransactionId agg_catalog_xmin = InvalidTransactionId;
Assert(ReplicationSlotCtl != NULL);
+ Assert(!already_locked ||
+ (LWLockHeldByMeInMode(ReplicationSlotControlLock, LW_EXCLUSIVE) &&
+ LWLockHeldByMeInMode(ProcArrayLock, LW_EXCLUSIVE)));
- LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
+ /*
+ * Hold the ReplicationSlotControlLock exclusive until after updating
+ * the slot xmin values, so no backend can compute and update the new
+ * value concurrently.
+ *
+ * One might think that we can hold the ProcArrayLock exclusively,
+ * compute the xmin values while holding the ReplicationSlotControlLock
+ * in shared mode, and update the slot xmin values, but it could increase
+ * lock contention on the ProcArrayLock, which is not great since this
+ * function can be called at non-negligible frequency.
+ *
+ * We instead increase lock contention on the ReplicationSlotControlLock
+ * but it would be less harmful.
+ */
+ if (!already_locked)
+ LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
for (i = 0; i < max_replication_slots; i++)
{
@@ -1176,9 +1197,10 @@ ReplicationSlotsComputeRequiredXmin(bool already_locked)
agg_catalog_xmin = effective_catalog_xmin;
}
- LWLockRelease(ReplicationSlotControlLock);
-
ProcArraySetReplicationSlotXmin(agg_xmin, agg_catalog_xmin, already_locked);
+
+ if (!already_locked)
+ LWLockRelease(ReplicationSlotControlLock);
}
/*