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 */