agora inbox for [email protected]
help / color / mirror / Atom feedFrom: Andres Freund <[email protected]>
Subject: [PATCH] Fix initial logical decoding snapshat race condition.
Date: Sun, 30 Apr 2017 18:24:06 -0700
LogStandbySnapshot() has no interlock preventing
RecordTransactionCommit() from logging its commit record (abort
situation is similar), before the xl_running_xacts record is
generated. That can lead to the situation that logical decoding
forever waits for a commit/abort record, that was logged in the past.
To fix, check for such "missed" transactions when xl_running_xacts are
observed, while in SNAPBUILD_FULL_SNAPSHOT.
This also reverts changes made to GetRunningTransactionData() and
LogStandbySnapshot() by b89e151 as the additional locking does not
solve the problem, and is not required anymore.
Author: Petr Jelined and Andres Freund
Reviewed-By: Andres Freund
Discussion: https://postgr.es/m/[email protected]
Backpatch: 9.4, where logical decoding was introduced
---
src/backend/replication/logical/snapbuild.c | 98 +++++++++++++++++++++++------
src/backend/storage/ipc/procarray.c | 5 +-
src/backend/storage/ipc/standby.c | 19 ------
3 files changed, 81 insertions(+), 41 deletions(-)
diff --git a/src/backend/replication/logical/snapbuild.c b/src/backend/replication/logical/snapbuild.c
index 068d214fa1..8605625555 100644
--- a/src/backend/replication/logical/snapbuild.c
+++ b/src/backend/replication/logical/snapbuild.c
@@ -203,6 +203,7 @@ struct SnapBuild
size_t xcnt; /* number of used xip entries */
size_t xcnt_space; /* allocated size of xip */
TransactionId *xip; /* running xacts array, xidComparator-sorted */
+ bool *xip_running; /* xid in ->xip still running? */
} running;
/*
@@ -253,7 +254,7 @@ static bool ExportInProgress = false;
static void SnapBuildEndTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid);
/* ->running manipulation */
-static bool SnapBuildTxnIsRunning(SnapBuild *builder, TransactionId xid);
+static bool SnapBuildTxnIsRunning(SnapBuild *builder, TransactionId xid, int *off);
/* ->committed manipulation */
static void SnapBuildPurgeCommittedTxn(SnapBuild *builder);
@@ -700,7 +701,7 @@ SnapBuildProcessChange(SnapBuild *builder, TransactionId xid, XLogRecPtr lsn)
* we got into the SNAPBUILD_FULL_SNAPSHOT state.
*/
if (builder->state < SNAPBUILD_CONSISTENT &&
- SnapBuildTxnIsRunning(builder, xid))
+ SnapBuildTxnIsRunning(builder, xid, NULL))
return false;
/*
@@ -776,7 +777,7 @@ SnapBuildProcessNewCid(SnapBuild *builder, TransactionId xid,
* only exist after we freshly started from an < CONSISTENT snapshot.
*/
static bool
-SnapBuildTxnIsRunning(SnapBuild *builder, TransactionId xid)
+SnapBuildTxnIsRunning(SnapBuild *builder, TransactionId xid, int *off)
{
Assert(builder->state < SNAPBUILD_CONSISTENT);
Assert(TransactionIdIsNormal(builder->running.xmin));
@@ -793,8 +794,17 @@ SnapBuildTxnIsRunning(SnapBuild *builder, TransactionId xid)
if (search != NULL)
{
Assert(*search == xid);
+
+ if (off)
+ {
+ *off = search - builder->running.xip;
+ Assert(builder->running.xip[*off] == xid);
+ }
+
return true;
}
+
+
}
return false;
@@ -928,6 +938,8 @@ SnapBuildPurgeCommittedTxn(SnapBuild *builder)
static void
SnapBuildEndTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid)
{
+ int off;
+
if (builder->state == SNAPBUILD_CONSISTENT)
return;
@@ -938,10 +950,13 @@ SnapBuildEndTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid)
* subxids and since they end at the same time it's sufficient to deal
* with them here.
*/
- if (SnapBuildTxnIsRunning(builder, xid))
+ if (SnapBuildTxnIsRunning(builder, xid, &off))
{
Assert(builder->running.xcnt > 0);
+ Assert(builder->running.xip_running[off]);
+ builder->running.xip_running[off] = false;
+
if (!--builder->running.xcnt)
{
/*
@@ -1250,9 +1265,15 @@ SnapBuildFindSnapshot(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *runn
*
* a) There were no running transactions when the xl_running_xacts record
* was inserted, jump to CONSISTENT immediately. We might find such a
- * state we were waiting for b) or c).
+ * state while waiting for b) or c).
*
- * b) Wait for all toplevel transactions that were running to end. We
+ * b) This (in a previous run) or another decoding slot serialized a
+ * snapshot to disk that we can use. Can't use this method for the
+ * initial snapshot when slot is being created and needs full snapshot
+ * for export or direct use, as that snapshot will only contain catalog
+ * modifying transactions.
+ *
+ * c) Wait for all toplevel transactions that were running to end. We
* simply track the number of in-progress toplevel transactions and
* lower it whenever one commits or aborts. When that number
* (builder->running.xcnt) reaches zero, we can go from FULL_SNAPSHOT
@@ -1264,11 +1285,13 @@ SnapBuildFindSnapshot(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *runn
* subtransactions - and by extension suboverflowed xl_running_xacts -
* at all.
*
- * c) This (in a previous run) or another decoding slot serialized a
- * snapshot to disk that we can use. Can't use this method for the
- * initial snapshot when slot is being created and needs full snapshot
- * for export or direct use, as that snapshot will only contain catalog
- * modifying transactions.
+ * Unfortunately there's a race condition around LogStandbySnapshot(),
+ * where transactions might have logged their commit record, before
+ * xl_running_xacts itself is logged. In that case the decoding logic
+ * would have missed that fact. Thus
+ *
+ * d) xl_running_xacts shows us that transaction(s) assumed to be still
+ * running have actually already finished. Adjust their status.
* ---
*/
@@ -1323,16 +1346,15 @@ SnapBuildFindSnapshot(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *runn
return false;
}
- /* c) valid on disk state and not building full snapshot */
+ /* b) valid on disk state and not building full snapshot */
else if (!builder->building_full_snapshot &&
SnapBuildRestore(builder, lsn))
{
/* there won't be any state to cleanup */
return false;
}
-
/*
- * b) first encounter of a useable xl_running_xacts record. If we had
+ * c) first encounter of a useable xl_running_xacts record. If we had
* found one earlier we would either track running transactions (i.e.
* builder->running.xcnt != 0) or be consistent (this function wouldn't
* get called).
@@ -1367,6 +1389,11 @@ SnapBuildFindSnapshot(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *runn
builder->running.xcnt * sizeof(TransactionId));
memcpy(builder->running.xip, running->xids,
builder->running.xcnt * sizeof(TransactionId));
+ builder->running.xip_running =
+ MemoryContextAlloc(builder->context,
+ builder->running.xcnt * sizeof(bool));
+ memset(builder->running.xip_running, 1,
+ builder->running.xcnt * sizeof(bool));
/* sort so we can do a binary search */
qsort(builder->running.xip, builder->running.xcnt,
@@ -1414,13 +1441,44 @@ SnapBuildFindSnapshot(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *runn
/* nothing could have built up so far, so don't perform cleanup */
return false;
}
+ /* d) already tracking running xids, check whether any were missed */
+ else
+ {
+ size_t xcnt = running->xcnt;
+ TransactionId *xip =
+ MemoryContextAlloc(builder->context, xcnt * sizeof(TransactionId));
+ int i;
- /*
- * We already started to track running xacts and need to wait for all
- * in-progress ones to finish. We fall through to the normal processing of
- * records so incremental cleanup can be performed.
- */
- return true;
+ memcpy(xip, running->xids, xcnt * sizeof(TransactionId));
+
+ /* sort so we can do a binary search */
+ qsort(xip, xcnt, sizeof(TransactionId), xidComparator);
+
+ /*
+ * Mark all transactions as finished that we assumed were running, but
+ * actually aren't according to the xl_running_xacts record.
+ */
+ for (i = 0; i < builder->running.xcnt_space; i++)
+ {
+ TransactionId still_running = builder->running.xip[i];
+ void *test;
+
+ if (!builder->running.xip_running[i])
+ continue;
+
+ test = bsearch(&still_running, xip, xcnt,
+ sizeof(TransactionId), xidComparator);
+ if (!test)
+ SnapBuildEndTxn(builder, lsn, still_running);
+ }
+
+ /*
+ * We already started to track running xacts and need to wait for all
+ * in-progress ones to finish. We fall through to the normal processing of
+ * records so incremental cleanup can be performed.
+ */
+ return true;
+ }
}
diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c
index 8a71536791..de3ae92dd7 100644
--- a/src/backend/storage/ipc/procarray.c
+++ b/src/backend/storage/ipc/procarray.c
@@ -2060,12 +2060,13 @@ GetRunningTransactionData(void)
CurrentRunningXacts->oldestRunningXid = oldestRunningXid;
CurrentRunningXacts->latestCompletedXid = latestCompletedXid;
+ /* We don't release XidGenLock here, the caller is responsible for that */
+ LWLockRelease(ProcArrayLock);
+
Assert(TransactionIdIsValid(CurrentRunningXacts->nextXid));
Assert(TransactionIdIsValid(CurrentRunningXacts->oldestRunningXid));
Assert(TransactionIdIsNormal(CurrentRunningXacts->latestCompletedXid));
- /* We don't release the locks here, the caller is responsible for that */
-
return CurrentRunningXacts;
}
diff --git a/src/backend/storage/ipc/standby.c b/src/backend/storage/ipc/standby.c
index 8e57f933ca..ddb279e274 100644
--- a/src/backend/storage/ipc/standby.c
+++ b/src/backend/storage/ipc/standby.c
@@ -929,27 +929,8 @@ LogStandbySnapshot(void)
*/
running = GetRunningTransactionData();
- /*
- * GetRunningTransactionData() acquired ProcArrayLock, we must release it.
- * For Hot Standby this can be done before inserting the WAL record
- * because ProcArrayApplyRecoveryInfo() rechecks the commit status using
- * the clog. For logical decoding, though, the lock can't be released
- * early because the clog might be "in the future" from the POV of the
- * historic snapshot. This would allow for situations where we're waiting
- * for the end of a transaction listed in the xl_running_xacts record
- * which, according to the WAL, has committed before the xl_running_xacts
- * record. Fortunately this routine isn't executed frequently, and it's
- * only a shared lock.
- */
- if (wal_level < WAL_LEVEL_LOGICAL)
- LWLockRelease(ProcArrayLock);
-
recptr = LogCurrentRunningXacts(running);
- /* Release lock if we kept it longer ... */
- if (wal_level >= WAL_LEVEL_LOGICAL)
- LWLockRelease(ProcArrayLock);
-
/* GetRunningTransactionData() acquired XidGenLock, we must release it */
LWLockRelease(XidGenLock);
--
2.12.0.264.gd6db3f2165.dirty
--ptyjiezm3fnjgfig
Content-Type: text/plain
Content-Disposition: inline
Content-Transfer-Encoding: 8bit
MIME-Version: 1.0
--
Sent via pgsql-hackers mailing list ([email protected])
To make changes to your subscription:
http://www.postgresql.org/mailpref/pgsql-hackers
--ptyjiezm3fnjgfig--
view thread (4+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected]
Subject: Re: [PATCH] Fix initial logical decoding snapshat race condition.
In-Reply-To: <no-message-id-56730@localhost>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox