public inbox for [email protected]
help / color / mirror / Atom feedFrom: Tomas Vondra <[email protected]>
To: Tom Lane <[email protected]>
Cc: PostgreSQL Hackers <[email protected]>
Subject: Re: sequences vs. synchronous replication
Date: Sat, 18 Dec 2021 21:45:09 +0100
Message-ID: <[email protected]> (raw)
In-Reply-To: <[email protected]>
References: <[email protected]>
<[email protected]>
<[email protected]>
On 12/18/21 07:00, Tomas Vondra wrote:
>
>
> On 12/18/21 05:52, Tom Lane wrote:
>> Tomas Vondra <[email protected]> writes:
>>> The problem is exactly the same as in [1] - the aborted transaction
>>> generated WAL, but RecordTransactionAbort() ignores that and does not
>>> update LogwrtResult.Write, with the reasoning that aborted transactions
>>> do not matter. But sequences violate that, because we only write WAL
>>> once every 32 increments, so the following nextval() gets "committed"
>>> without waiting for the replica (because it did not produce WAL).
>>
>> Ugh.
>>
>>> I'm not sure this is a clear data corruption bug, but it surely walks
>>> and quacks like one. My proposal is to fix this by tracking the lsn of
>>> the last LSN for a sequence increment, and then check that LSN in
>>> RecordTransactionCommit() before calling XLogFlush().
>>
>> (1) Does that work if the aborted increment was in a different
>> session? I think it is okay but I'm tired enough to not be sure.
>>
>
> Good point - it doesn't :-( At least not by simply storing LSN in a
> global variable or something like that.
>
> The second backend needs to know the LSN of the last WAL-logged sequence
> increment, but only the first backend knows that. So we'd need to share
> that between backends somehow. I doubt we want to track LSN for every
> individual sequence (because for clusters with many dbs / sequences that
> may be a lot).
>
> Perhaps we could track just a fixed number o LSN values in shared memory
> (say, 1024), and update/read just the element determined by hash(oid).
> That is, the backend WAL-logging sequence with given oid would set the
> current LSN to array[hash(oid) % 1024], and backend doing nextval()
> would simply remember the LSN in that slot. Yes, if there are conflicts
> that'll flush more than needed.
>
Here's a PoC demonstrating this idea. I'm not convinced it's the right
way to deal with this - it surely seems more like a duct tape fix than a
clean solution. But it does the trick.
I wonder if storing this in shmem is good enough - we lose the LSN info
on restart, but the checkpoint should trigger FPI which makes it OK.
A bigger question is whether sequences are the only thing affected by
this. If you look at RecordTransactionCommit() then we skip flush/wait
in two cases:
1) !wrote_xlog - if the xact did not produce WAL
2) !markXidCommitted - if the xact does not have a valid XID
Both apply to sequences, and the PoC patch tweaks them. But maybe there
are other places where we don't generate WAL and/or assign XID in some
cases, to save time?
regards
--
Tomas Vondra
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company
Attachments:
[text/x-patch] sequences-track-lsn-poc.patch (8.3K, ../[email protected]/2-sequences-track-lsn-poc.patch)
download | inline diff:
diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c
index e7b0bc804d..c6a0c07846 100644
--- a/src/backend/access/transam/xact.c
+++ b/src/backend/access/transam/xact.c
@@ -35,6 +35,7 @@
#include "catalog/pg_enum.h"
#include "catalog/storage.h"
#include "commands/async.h"
+#include "commands/sequence.h"
#include "commands/tablecmds.h"
#include "commands/trigger.h"
#include "common/pg_prng.h"
@@ -1289,6 +1290,13 @@ RecordTransactionCommit(void)
bool RelcacheInitFileInval = false;
bool wrote_xlog;
+ /*
+ * Force flush and synchronous replication even if no XID was assigned.
+ * This is needed when depending on WAL produced by another transaction
+ * (possibly in a different backend). Sequences need this, for example.
+ */
+ bool force_sync = false;
+
/*
* Log pending invalidations for logical decoding of in-progress
* transactions. Normally for DDLs, we log this at each command end,
@@ -1299,6 +1307,24 @@ RecordTransactionCommit(void)
if (XLogLogicalInfoActive())
LogLogicalInvalidations();
+ /*
+ * Check the LSN of increments for sequences we touched in this transaction.
+ * If it's higher than XactLastRecEnd, we need to force flush/sync.
+ */
+ {
+ /* Separate call, so that we can compare to XactLastRecEnd. */
+ XLogRecPtr tmpptr = SequenceGetLastLSN();
+
+ /*
+ * If higher than XactLastRecEnd, make sure we flush even without
+ * a XID assigned.
+ */
+ force_sync = (tmpptr > XactLastRecEnd);
+
+ /* Override the XactLastRecEnd value. */
+ XactLastRecEnd = Max(XactLastRecEnd, tmpptr);
+ }
+
/* Get data needed for commit record */
nrels = smgrGetPendingDeletes(true, &rels);
nchildren = xactGetCommittedChildren(&children);
@@ -1446,7 +1472,7 @@ RecordTransactionCommit(void)
* if all to-be-deleted tables are temporary though, since they are lost
* anyway if we crash.)
*/
- if ((wrote_xlog && markXidCommitted &&
+ if ((wrote_xlog && (markXidCommitted || force_sync) &&
synchronous_commit > SYNCHRONOUS_COMMIT_OFF) ||
forceSyncCommit || nrels > 0)
{
@@ -1504,7 +1530,7 @@ RecordTransactionCommit(void)
* Note that at this stage we have marked clog, but still show as running
* in the procarray and continue to hold locks.
*/
- if (wrote_xlog && markXidCommitted)
+ if (wrote_xlog && (markXidCommitted || force_sync))
SyncRepWaitForLSN(XactLastRecEnd, true);
/* remember end of last commit record */
@@ -1512,6 +1538,8 @@ RecordTransactionCommit(void)
/* Reset XactLastRecEnd until the next transaction writes something */
XactLastRecEnd = 0;
+ SequenceResetLastLSN();
+
cleanup:
/* Clean up local data */
if (rels)
diff --git a/src/backend/commands/sequence.c b/src/backend/commands/sequence.c
index 72bfdc07a4..43fe19831e 100644
--- a/src/backend/commands/sequence.c
+++ b/src/backend/commands/sequence.c
@@ -109,6 +109,19 @@ static void init_params(ParseState *pstate, List *options, bool for_identity,
static void do_setval(Oid relid, int64 next, bool iscalled);
static void process_owned_by(Relation seqrel, List *owned_by, bool for_identity);
+/* handle flush/sync for sequences */
+#define SEQUENCE_LSN_COUNT 1024
+
+typedef struct SequenceLsnState {
+ slock_t lock;
+ XLogRecPtr lsn[SEQUENCE_LSN_COUNT];
+} SequenceLsnState;
+
+static SequenceLsnState *SequenceShmemInfo = NULL;
+static XLogRecPtr lastSequenceLSN = InvalidXLogRecPtr;
+
+static void SequenceSetLSN(Oid relid, XLogRecPtr recptr);
+static XLogRecPtr SequenceGetLSN(Oid relid);
/*
* DefineSequence
@@ -406,7 +419,12 @@ fill_seq_with_data(Relation rel, HeapTuple tuple)
recptr = XLogInsert(RM_SEQ_ID, XLOG_SEQ_LOG);
PageSetLSN(page, recptr);
+
+ SequenceSetLSN(rel->rd_id, recptr);
+ lastSequenceLSN = recptr;
}
+ else
+ lastSequenceLSN = SequenceGetLSN(rel->rd_id);
END_CRIT_SECTION();
@@ -810,7 +828,12 @@ nextval_internal(Oid relid, bool check_permissions)
recptr = XLogInsert(RM_SEQ_ID, XLOG_SEQ_LOG);
PageSetLSN(page, recptr);
+
+ SequenceSetLSN(relid, recptr);
+ lastSequenceLSN = recptr;
}
+ else
+ lastSequenceLSN = SequenceGetLSN(relid);
/* Now update sequence tuple to the intended final state */
seq->last_value = last; /* last fetched number */
@@ -1005,7 +1028,12 @@ do_setval(Oid relid, int64 next, bool iscalled)
recptr = XLogInsert(RM_SEQ_ID, XLOG_SEQ_LOG);
PageSetLSN(page, recptr);
+
+ SequenceSetLSN(relid, recptr);
+ lastSequenceLSN = recptr;
}
+ else
+ lastSequenceLSN = SequenceGetLSN(relid);
END_CRIT_SECTION();
@@ -1927,3 +1955,98 @@ seq_mask(char *page, BlockNumber blkno)
mask_unused_space(page);
}
+
+
+/*
+ * Initialization of shared memory for sequence LSN positions.
+ *
+ * To track the last LSN for each sequence we use a fixed-length array, and use
+ * OID of the sequence to select the element. This allows us to handle unknown
+ * number of sequences in a limited amount of space, but it may force us to
+ * sync a bit too much if two sequences conflict (pick the same array element).
+ * But that seems like acceptable tradeoff.
+ */
+Size
+SequenceShmemSize(void)
+{
+ return sizeof(SequenceLsnState);
+}
+
+/*
+ * Initialize the LSN array (set it to InvalidXLogRecPtr).
+ */
+void
+SequenceShmemInit(void)
+{
+ bool found;
+ SequenceShmemInfo
+ = (SequenceLsnState *) ShmemInitStruct("Sequences Shmem", SequenceShmemSize(), &found);
+
+ if (!found)
+ {
+ /* First time through, so initialize */
+ SpinLockInit(&(SequenceShmemInfo->lock));
+ MemSet(SequenceShmemInfo->lsn, 0, SEQUENCE_LSN_COUNT * sizeof(XLogRecPtr));
+ }
+}
+
+/*
+ * Update LSN tracked for a sequence with a given OID.
+ *
+ * XXX Maybe hash the OID.
+ */
+static void
+SequenceSetLSN(Oid relid, XLogRecPtr recptr)
+{
+ int idx = (relid % SEQUENCE_LSN_COUNT);
+
+ Assert(SequenceShmemInfo);
+
+ SpinLockAcquire(&(SequenceShmemInfo->lock));
+ SequenceShmemInfo->lsn[idx] = Max(SequenceShmemInfo->lsn[idx], recptr);
+ SpinLockRelease(&(SequenceShmemInfo->lock));
+}
+
+/*
+ * Get LSN tracked for a sequence with a given OID.
+ *
+ * XXX What if the LSN is 0/0? Presumably that means we don't have any value
+ * (e.g. after a restart). Maybe we should use current insert LSN in that case?
+ * Although, shutdown forces checkpoint, and we write FPI or something?
+ */
+static XLogRecPtr
+SequenceGetLSN(Oid relid)
+{
+ int idx = (relid % SEQUENCE_LSN_COUNT);
+ XLogRecPtr lsn;
+
+ Assert(SequenceShmemInfo);
+
+ SpinLockAcquire(&(SequenceShmemInfo->lock));
+ lsn = SequenceShmemInfo->lsn[idx];
+ SpinLockRelease(&(SequenceShmemInfo->lock));
+
+ return lsn;
+}
+
+/*
+ * Return the higher LSN for any sequence we touched in this session (either
+ * the WAL we wrote, or WAL produced by other transaction we depend on).
+ */
+XLogRecPtr
+SequenceGetLastLSN(void)
+{
+ return lastSequenceLSN;
+}
+
+/*
+ * Reset the last LSN after we flush.
+ *
+ * XXX Not sure we actually need to do this. Maybe we don't need tp reset and
+ * keeping the value forever is fine.
+ */
+void
+SequenceResetLastLSN(void)
+{
+ lastSequenceLSN = InvalidXLogRecPtr;
+}
diff --git a/src/backend/storage/ipc/ipci.c b/src/backend/storage/ipc/ipci.c
index 9fa3e0631e..517cca450b 100644
--- a/src/backend/storage/ipc/ipci.c
+++ b/src/backend/storage/ipc/ipci.c
@@ -23,6 +23,7 @@
#include "access/syncscan.h"
#include "access/twophase.h"
#include "commands/async.h"
+#include "commands/sequence.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "postmaster/autovacuum.h"
@@ -143,6 +144,7 @@ CalculateShmemSize(int *num_semaphores)
size = add_size(size, BTreeShmemSize());
size = add_size(size, SyncScanShmemSize());
size = add_size(size, AsyncShmemSize());
+ size = add_size(size, SequenceShmemSize());
#ifdef EXEC_BACKEND
size = add_size(size, ShmemBackendArraySize());
#endif
@@ -293,6 +295,7 @@ CreateSharedMemoryAndSemaphores(void)
BTreeShmemInit();
SyncScanShmemInit();
AsyncShmemInit();
+ SequenceShmemInit();
#ifdef EXEC_BACKEND
diff --git a/src/include/commands/sequence.h b/src/include/commands/sequence.h
index 40544dd4c7..85fbc3789d 100644
--- a/src/include/commands/sequence.h
+++ b/src/include/commands/sequence.h
@@ -66,4 +66,10 @@ extern void seq_desc(StringInfo buf, XLogReaderState *rptr);
extern const char *seq_identify(uint8 info);
extern void seq_mask(char *pagedata, BlockNumber blkno);
+
+extern Size SequenceShmemSize(void);
+extern void SequenceShmemInit(void);
+extern XLogRecPtr SequenceGetLastLSN(void);
+extern void SequenceResetLastLSN(void);
+
#endif /* SEQUENCE_H */
view thread (78+ 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], [email protected], [email protected]
Subject: Re: sequences vs. synchronous replication
In-Reply-To: <[email protected]>
* 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