diff --git a/doc/src/sgml/func.sgml b/doc/src/sgml/func.sgml index 4c3e232..d7d2fbe 100644 *** a/doc/src/sgml/func.sgml --- b/doc/src/sgml/func.sgml *************** FOR EACH ROW EXECUTE PROCEDURE suppress_ *** 15012,15015 **** --- 15012,15089 ---- . + + + Snapshot Synchronization Functions + + + pg_export_snapshot + + + + PostgreSQL allows different sessions to synchronize their + snapshots. A database snapshot determines which data is visible to + the client that is using this snapshot. Synchronized snapshots are necessary when + two clients need to see the same content in the database. If these two clients + just connected to the database and opened their transactions, then they could + never be sure that there was no data modification right between both + connections. + + + As a solution, PostgreSQL offers the function + pg_export_snapshot which saves the snapshot internally and + from then on until the end of the saving transaction, the snapshot can be + used on a (or ) command to open a second transaction with the + exact same snapshot. Now both transactions are guaranteed to see the exact same + data even though they might have connected at different times. + + + Note that a snapshot can only be used to start a new transaction as long + as the transaction that originally saved it is held open. Also note that even + after the synchronization both clients still run their own independent + transactions. As a consequence, even though synchronized with respect to + reading pre-existing data, both transactions won't be able to see each other's + uncommitted data. + + + Snapshot Synchronization Functions + + + Name Return Type Description + + + + + + + pg_export_snapshot() + + text + Save the snapshot and return its identifier + + + +
+ + + The function pg_export_snapshot does not take an argument + and returns the snapshot's identifier as text data. Internally the + function will save the snapshot data to a file so that it can be retrieved + from a different backend process later on. Note that as soon as the + transaction ends, any saved snapshots become invalid and their + identifiers cannot be used to start other transactions anymore. If the function + has been executed, the transaction cannot be prepared anymore with . + + + SELECT pg_export_snapshot(); + + pg_export_snapshot + -------------------- + 000003A1-1 + (1 row) + + + diff --git a/doc/src/sgml/ref/begin.sgml b/doc/src/sgml/ref/begin.sgml index acd8232..d5fb728 100644 *** a/doc/src/sgml/ref/begin.sgml --- b/doc/src/sgml/ref/begin.sgml *************** BEGIN [ WORK | TRANSACTION ] [ *************** BEGIN [ WORK | TRANSACTION ] [ + + + BEGIN ... (SNAPSHOT = snapshot_id ) + + + Use the (SNAPSHOT = snapshot_id) parameter to start a + new transaction with the same snapshot as an already running transaction. + A call to pg_export_snapshot (see ) returns a snapshot id which must be + passed to the BEGIN command to create a second + transaction running with the same snapshot. You also need to make the + transaction ISOLATION LEVEL SERIALIZABLE or + ISOLATION LEVEL REPEATABLE READ. + + + *************** BEGIN [ WORK | TRANSACTION ] [ BEGIN; + + + To begin a new transaction block with the same snapshot as an already + existing transaction, first export the snapshot from the existing + transaction. This will return the snapshot id: + + + # SELECT pg_export_snapshot(); + pg_export_snapshot + -------------------- + 000003A1-1 + (1 row) + + + Then reference this snapshot id on the BEGIN TRANSACTION command: + + + BEGIN TRANSACTION ISOLATION LEVEL REPEATABLE READ (SNAPSHOT = '000003A1-1'); + + + + diff --git a/doc/src/sgml/ref/start_transaction.sgml b/doc/src/sgml/ref/start_transaction.sgml index f25a3e9..0576faf 100644 *** a/doc/src/sgml/ref/start_transaction.sgml --- b/doc/src/sgml/ref/start_transaction.sgml *************** START TRANSACTION [ Parameters ! Refer to for information on the meaning ! of the parameters to this statement. --- 47,54 ---- Parameters ! Refer to and ! for information on the meaning of the parameters to this statement. diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c index e8821f7..e372c80 100644 *** a/src/backend/access/transam/xact.c --- b/src/backend/access/transam/xact.c *************** CommitTransaction(void) *** 1855,1860 **** --- 1855,1866 ---- */ PreCommit_Notify(); + /* + * Cleans up exported snapshots (this needs to happen before we update + * our MyProc entry). + */ + PreCommit_Snapshot(); + /* Prevent cancel/die interrupt while cleaning up */ HOLD_INTERRUPTS(); *************** PrepareTransaction(void) *** 2073,2078 **** --- 2079,2089 ---- (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("cannot PREPARE a transaction that has operated on temporary tables"))); + if (exportedSnapshots) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("cannot PREPARE a transaction that has exported snapshots"))); + /* Prevent cancel/die interrupt while cleaning up */ HOLD_INTERRUPTS(); diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index 11035e6..611ac81 100644 *** a/src/backend/access/transam/xlog.c --- b/src/backend/access/transam/xlog.c *************** *** 58,63 **** --- 58,64 ---- #include "utils/guc.h" #include "utils/ps_status.h" #include "utils/relmapper.h" + #include "utils/snapmgr.h" #include "pg_trace.h" *************** StartupXLOG(void) *** 6368,6373 **** --- 6369,6379 ---- CheckRequiredParameterValues(); /* + * We can delete any saved transaction snapshots that still exist + */ + DeleteAllExportedSnapshotFiles(); + + /* * We're in recovery, so unlogged relations relations may be trashed * and must be reset. This should be done BEFORE allowing Hot Standby * connections, so that read-only backends don't try to read whatever diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y index e9f3896..8bdf096 100644 *** a/src/backend/parser/gram.y --- b/src/backend/parser/gram.y *************** transaction_mode_item: *** 7252,7257 **** --- 7252,7260 ---- | NOT DEFERRABLE { $$ = makeDefElem("transaction_deferrable", makeIntConst(FALSE, @1)); } + | '(' ColId '=' Sconst ')' + { $$ = makeDefElem($2, + makeStringConst($4, @4)); } ; /* Syntax with commas is SQL-spec, without commas is Postgres historical */ diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c index e7593fa..8b98e4e 100644 *** a/src/backend/storage/ipc/procarray.c --- b/src/backend/storage/ipc/procarray.c *************** ProcArrayShmemSize(void) *** 167,175 **** { Size size; - /* Size of the ProcArray structure itself */ - #define PROCARRAY_MAXPROCS (MaxBackends + max_prepared_xacts) - size = offsetof(ProcArrayStruct, procs); size = add_size(size, mul_size(sizeof(PGPROC *), PROCARRAY_MAXPROCS)); --- 167,172 ---- *************** ProcArrayShmemSize(void) *** 180,193 **** * TransactionIdIsInProgress() and GetRunningTransactionData(). All of the * main structures created in those functions must be identically sized, * since we may at times copy the whole of the data structures around. We ! * refer to this size as TOTAL_MAX_CACHED_SUBXIDS. * * Ideally we'd only create this structure if we were actually doing hot * standby in the current run, but we don't know that yet at the time * shared memory is being set up. */ - #define TOTAL_MAX_CACHED_SUBXIDS \ - ((PGPROC_MAX_CACHED_SUBXIDS + 1) * PROCARRAY_MAXPROCS) if (EnableHotStandby) { --- 177,188 ---- * TransactionIdIsInProgress() and GetRunningTransactionData(). All of the * main structures created in those functions must be identically sized, * since we may at times copy the whole of the data structures around. We ! * refer to this size as TOTAL_MAX_CACHED_SUBXIDS, defined in procarray.h. * * Ideally we'd only create this structure if we were actually doing hot * standby in the current run, but we don't know that yet at the time * shared memory is being set up. */ if (EnableHotStandby) { *************** GetOldestXmin(bool allDbs, bool ignoreVa *** 1145,1151 **** * not statically allocated (see xip allocation below). */ Snapshot ! GetSnapshotData(Snapshot snapshot) { ProcArrayStruct *arrayP = procArray; TransactionId xmin; --- 1140,1146 ---- * not statically allocated (see xip allocation below). */ Snapshot ! GetSnapshotData(Snapshot snapshot, Snapshot stemplate) { ProcArrayStruct *arrayP = procArray; TransactionId xmin; *************** GetSnapshotData(Snapshot snapshot) *** 1159,1164 **** --- 1154,1182 ---- Assert(snapshot != NULL); /* + * We only get a valid snapshot in stemplate if the snapshot + * synchronization feature used. In that case we just need to copy the + * values that we get onto the snapshot we return. + * Note that in this case we always duplicate an existing snapshot, that is + * currently held by another active transaction. That's why we do not need + * to update any { RecentGlobalXmin, RecentXmin, globalxmin }. + */ + if (stemplate != InvalidSnapshot) + { + /* + * 'stemplate' is only read and its values are copied onto 'snapshot'. + */ + CopySnapshotOnto(stemplate, snapshot); + + /* + * We can use the result of the copy except for that this snapshot + * should look like new and not copied. + */ + snapshot->copied = false; + return snapshot; + } + + /* * Allocating space for maxProcs xids is usually overkill; numProcs would * be sufficient. But it seems better to do the malloc while not holding * the lock, so we can't look at numProcs. Likewise, we allocate much diff --git a/src/backend/storage/lmgr/predicate.c b/src/backend/storage/lmgr/predicate.c index 8e7a7f0..cca0bef 100644 *** a/src/backend/storage/lmgr/predicate.c --- b/src/backend/storage/lmgr/predicate.c *************** static void OldSerXidSetActiveSerXmin(Tr *** 416,423 **** static uint32 predicatelock_hash(const void *key, Size keysize); static void SummarizeOldestCommittedSxact(void); ! static Snapshot GetSafeSnapshot(Snapshot snapshot); ! static Snapshot RegisterSerializableTransactionInt(Snapshot snapshot); static bool PredicateLockExists(const PREDICATELOCKTARGETTAG *targettag); static bool GetParentPredicateLockTag(const PREDICATELOCKTARGETTAG *tag, PREDICATELOCKTARGETTAG *parent); --- 416,423 ---- static uint32 predicatelock_hash(const void *key, Size keysize); static void SummarizeOldestCommittedSxact(void); ! static Snapshot GetSafeSnapshot(Snapshot snapshot, Snapshot stemplate); ! static Snapshot RegisterSerializableTransactionInt(Snapshot snapshot, Snapshot stemplate); static bool PredicateLockExists(const PREDICATELOCKTARGETTAG *targettag); static bool GetParentPredicateLockTag(const PREDICATELOCKTARGETTAG *tag, PREDICATELOCKTARGETTAG *parent); *************** SummarizeOldestCommittedSxact(void) *** 1487,1493 **** * one of them could possibly create a conflict. */ static Snapshot ! GetSafeSnapshot(Snapshot origSnapshot) { Snapshot snapshot; --- 1487,1493 ---- * one of them could possibly create a conflict. */ static Snapshot ! GetSafeSnapshot(Snapshot origSnapshot, Snapshot stemplate) { Snapshot snapshot; *************** GetSafeSnapshot(Snapshot origSnapshot) *** 1501,1507 **** * caller passed to us. It returns a copy of that snapshot and * registers it on TopTransactionResourceOwner. */ ! snapshot = RegisterSerializableTransactionInt(origSnapshot); if (MySerializableXact == InvalidSerializableXact) return snapshot; /* no concurrent r/w xacts; it's safe */ --- 1501,1507 ---- * caller passed to us. It returns a copy of that snapshot and * registers it on TopTransactionResourceOwner. */ ! snapshot = RegisterSerializableTransactionInt(origSnapshot, stemplate); if (MySerializableXact == InvalidSerializableXact) return snapshot; /* no concurrent r/w xacts; it's safe */ *************** GetSafeSnapshot(Snapshot origSnapshot) *** 1554,1560 **** * It should be current for this process and be contained in PredXact. */ Snapshot ! RegisterSerializableTransaction(Snapshot snapshot) { Assert(IsolationIsSerializable()); --- 1554,1560 ---- * It should be current for this process and be contained in PredXact. */ Snapshot ! RegisterSerializableTransaction(Snapshot snapshot, Snapshot stemplate) { Assert(IsolationIsSerializable()); *************** RegisterSerializableTransaction(Snapshot *** 1564,1576 **** * thereby avoid all SSI overhead once it's running.. */ if (XactReadOnly && XactDeferrable) ! return GetSafeSnapshot(snapshot); ! return RegisterSerializableTransactionInt(snapshot); } static Snapshot ! RegisterSerializableTransactionInt(Snapshot snapshot) { PGPROC *proc; VirtualTransactionId vxid; --- 1564,1576 ---- * thereby avoid all SSI overhead once it's running.. */ if (XactReadOnly && XactDeferrable) ! return GetSafeSnapshot(snapshot, stemplate); ! return RegisterSerializableTransactionInt(snapshot, stemplate); } static Snapshot ! RegisterSerializableTransactionInt(Snapshot snapshot, Snapshot stemplate) { PGPROC *proc; VirtualTransactionId vxid; *************** RegisterSerializableTransactionInt(Snaps *** 1608,1614 **** } while (!sxact); /* Get and register a snapshot */ ! snapshot = GetSnapshotData(snapshot); snapshot = RegisterSnapshotOnOwner(snapshot, TopTransactionResourceOwner); /* --- 1608,1614 ---- } while (!sxact); /* Get and register a snapshot */ ! snapshot = GetSnapshotData(snapshot, stemplate); snapshot = RegisterSnapshotOnOwner(snapshot, TopTransactionResourceOwner); /* diff --git a/src/backend/tcop/utility.c b/src/backend/tcop/utility.c index 0749227..8dbcc40 100644 *** a/src/backend/tcop/utility.c --- b/src/backend/tcop/utility.c *************** *** 58,63 **** --- 58,64 ---- #include "tcop/utility.h" #include "utils/acl.h" #include "utils/guc.h" + #include "utils/snapmgr.h" #include "utils/syscache.h" *************** standard_ProcessUtility(Node *parsetree, *** 375,380 **** --- 376,382 ---- case TRANS_STMT_START: { ListCell *lc; + char *snapshotId = NULL; BeginTransactionBlock(); foreach(lc, stmt->options) *************** standard_ProcessUtility(Node *parsetree, *** 393,398 **** --- 395,411 ---- SetPGVariable("transaction_deferrable", list_make1(item->arg), true); + else if (strcmp(item->defname, "snapshot") == 0) + /* + * Only save the snapshot's id for now, so that we + * process any other modifiers first. + */ + snapshotId = ((A_Const *) item->arg)->val.val.str; + } + if (snapshotId) + { + if (!ImportSnapshot(snapshotId)) + elog(ERROR, "Could not import the requested snapshot"); } } break; diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c index 6670997..6c583bb 100644 *** a/src/backend/utils/misc/guc.c --- b/src/backend/utils/misc/guc.c *************** ExecSetVariableStmt(VariableSetStmt *stm *** 6115,6120 **** --- 6115,6125 ---- else if (strcmp(item->defname, "transaction_deferrable") == 0) SetPGVariable("transaction_deferrable", list_make1(item->arg), stmt->is_local); + else if (strcmp(item->defname, "snapshot") == 0) + ereport(ERROR, + (errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("cannot set a snapshot with SET TRANSACTION"), + errhint("use BEGIN/START TRANSACTION instead"))); else elog(ERROR, "unexpected SET TRANSACTION element: %s", item->defname); diff --git a/src/backend/utils/time/snapmgr.c b/src/backend/utils/time/snapmgr.c index ef66466..52ded72 100644 *** a/src/backend/utils/time/snapmgr.c --- b/src/backend/utils/time/snapmgr.c *************** *** 25,36 **** */ #include "postgres.h" #include "access/transam.h" #include "access/xact.h" #include "storage/predicate.h" #include "storage/proc.h" #include "storage/procarray.h" ! #include "utils/memutils.h" #include "utils/memutils.h" #include "utils/resowner.h" #include "utils/snapmgr.h" --- 25,42 ---- */ #include "postgres.h" + #include + #include + #include + #include "access/transam.h" #include "access/xact.h" + #include "miscadmin.h" + #include "storage/fd.h" #include "storage/predicate.h" #include "storage/proc.h" #include "storage/procarray.h" ! #include "utils/builtins.h" #include "utils/memutils.h" #include "utils/resowner.h" #include "utils/snapmgr.h" *************** static Snapshot CopySnapshot(Snapshot sn *** 109,126 **** static void FreeSnapshot(Snapshot snapshot); static void SnapshotResetXmin(void); /* ! * GetTransactionSnapshot * Get the appropriate snapshot for a new query in a transaction. * ! * Note that the return value may point at static storage that will be modified ! * by future calls and by CommandCounterIncrement(). Callers should call ! * RegisterSnapshot or PushActiveSnapshot on the returned snap if it is to be ! * used very long. */ ! Snapshot ! GetTransactionSnapshot(void) { /* First call in transaction? */ if (!FirstSnapshotSet) --- 115,138 ---- static void FreeSnapshot(Snapshot snapshot); static void SnapshotResetXmin(void); + /* What we need for exporting snapshots */ + #define SNAPSHOT_EXPORT_DIR "pg_snapshots" + #define XactExportFilePath(path, xid, num, suffix) \ + snprintf(path, MAXPGPATH, SNAPSHOT_EXPORT_DIR "/%08X-%d%s", xid, num, suffix) + + List *exportedSnapshots = NIL; /* ! * GetTransactionSnapshotFromTemplate * Get the appropriate snapshot for a new query in a transaction. * ! * A template snapshot is passed for the synchronized snapshots feature. ! * In that case we want to have a snapshot back that has the template's ! * values. We just pass it along and the lower level functions take care ! * of it. */ ! static Snapshot ! GetTransactionSnapshotFromTemplate(Snapshot stemplate) { /* First call in transaction? */ if (!FirstSnapshotSet) *************** GetTransactionSnapshot(void) *** 135,151 **** if (IsolationUsesXactSnapshot()) { if (IsolationIsSerializable()) ! CurrentSnapshot = RegisterSerializableTransaction(&CurrentSnapshotData); else { ! CurrentSnapshot = GetSnapshotData(&CurrentSnapshotData); CurrentSnapshot = RegisterSnapshotOnOwner(CurrentSnapshot, TopTransactionResourceOwner); } registered_xact_snapshot = true; } else ! CurrentSnapshot = GetSnapshotData(&CurrentSnapshotData); FirstSnapshotSet = true; return CurrentSnapshot; --- 147,171 ---- if (IsolationUsesXactSnapshot()) { if (IsolationIsSerializable()) ! CurrentSnapshot = RegisterSerializableTransaction(&CurrentSnapshotData, ! stemplate); else { ! CurrentSnapshot = GetSnapshotData(&CurrentSnapshotData, stemplate); CurrentSnapshot = RegisterSnapshotOnOwner(CurrentSnapshot, TopTransactionResourceOwner); } registered_xact_snapshot = true; } else ! { ! /* ! * template is only used for the synchronized snapshot feature. Which in ! * turn is only allowed for IsolationUsesXactSnapshot() == true transactions ! */ ! Assert(stemplate == InvalidSnapshot); ! CurrentSnapshot = GetSnapshotData(&CurrentSnapshotData, InvalidSnapshot); ! } FirstSnapshotSet = true; return CurrentSnapshot; *************** GetTransactionSnapshot(void) *** 154,165 **** if (IsolationUsesXactSnapshot()) return CurrentSnapshot; ! CurrentSnapshot = GetSnapshotData(&CurrentSnapshotData); return CurrentSnapshot; } /* * GetLatestSnapshot * Get a snapshot that is up-to-date as of the current instant, * even if we are executing in transaction-snapshot mode. --- 174,206 ---- if (IsolationUsesXactSnapshot()) return CurrentSnapshot; ! /* see comment above */ ! Assert(stemplate == InvalidSnapshot); ! CurrentSnapshot = GetSnapshotData(&CurrentSnapshotData, InvalidSnapshot); return CurrentSnapshot; } /* + * GetTransactionSnapshot + * Get the appropriate snapshot for a new query in a transaction. + * + * This is the public interface for anything different than the snapshot + * synchronization feature. + * + * Note that the return value may point at static storage that will be modified + * by future calls and by CommandCounterIncrement(). Callers should call + * RegisterSnapshot or PushActiveSnapshot on the returned snap if it is to be + * used very long. + */ + Snapshot + GetTransactionSnapshot(void) + { + return GetTransactionSnapshotFromTemplate(InvalidSnapshot); + } + + + /* * GetLatestSnapshot * Get a snapshot that is up-to-date as of the current instant, * even if we are executing in transaction-snapshot mode. *************** GetLatestSnapshot(void) *** 171,177 **** if (!FirstSnapshotSet) elog(ERROR, "no snapshot has been set"); ! SecondarySnapshot = GetSnapshotData(&SecondarySnapshotData); return SecondarySnapshot; } --- 212,218 ---- if (!FirstSnapshotSet) elog(ERROR, "no snapshot has been set"); ! SecondarySnapshot = GetSnapshotData(&SecondarySnapshotData, InvalidSnapshot); return SecondarySnapshot; } *************** SnapshotSetCommandId(CommandId curcid) *** 193,235 **** } /* ! * CopySnapshot ! * Copy the given snapshot. * ! * The copy is palloc'd in TopTransactionContext and has initial refcounts set ! * to 0. The returned snapshot has the copied flag set. */ ! static Snapshot ! CopySnapshot(Snapshot snapshot) { - Snapshot newsnap; Size subxipoff; - Size size; - - Assert(snapshot != InvalidSnapshot); ! /* We allocate any XID arrays needed in the same palloc block. */ ! size = subxipoff = sizeof(SnapshotData) + ! snapshot->xcnt * sizeof(TransactionId); ! if (snapshot->subxcnt > 0) ! size += snapshot->subxcnt * sizeof(TransactionId); ! ! newsnap = (Snapshot) MemoryContextAlloc(TopTransactionContext, size); ! memcpy(newsnap, snapshot, sizeof(SnapshotData)); ! newsnap->regd_count = 0; ! newsnap->active_count = 0; ! newsnap->copied = true; /* setup XID array */ if (snapshot->xcnt > 0) { ! newsnap->xip = (TransactionId *) (newsnap + 1); ! memcpy(newsnap->xip, snapshot->xip, snapshot->xcnt * sizeof(TransactionId)); } else ! newsnap->xip = NULL; /* * Setup subXID array. Don't bother to copy it if it had overflowed, --- 234,266 ---- } /* ! * CopySnapshotOnto ! * Copy the given snapshot onto an already sufficiently allocated other ! * snapshot. * ! * Return the modified snapshot (onto). */ ! Snapshot ! CopySnapshotOnto(Snapshot snapshot, Snapshot onto) { Size subxipoff; ! subxipoff = sizeof(SnapshotData) + snapshot->xcnt * sizeof(TransactionId); ! memcpy(onto, snapshot, sizeof(SnapshotData)); ! onto->regd_count = 0; ! onto->active_count = 0; ! onto->copied = true; /* setup XID array */ if (snapshot->xcnt > 0) { ! onto->xip = (TransactionId *) (onto + 1); ! memcpy(onto->xip, snapshot->xip, snapshot->xcnt * sizeof(TransactionId)); } else ! onto->xip = NULL; /* * Setup subXID array. Don't bother to copy it if it had overflowed, *************** CopySnapshot(Snapshot snapshot) *** 240,253 **** if (snapshot->subxcnt > 0 && (!snapshot->suboverflowed || snapshot->takenDuringRecovery)) { ! newsnap->subxip = (TransactionId *) ((char *) newsnap + subxipoff); ! memcpy(newsnap->subxip, snapshot->subxip, snapshot->subxcnt * sizeof(TransactionId)); } else ! newsnap->subxip = NULL; ! return newsnap; } /* --- 271,309 ---- if (snapshot->subxcnt > 0 && (!snapshot->suboverflowed || snapshot->takenDuringRecovery)) { ! onto->subxip = (TransactionId *) ((char *) onto + subxipoff); ! memcpy(onto->subxip, snapshot->subxip, snapshot->subxcnt * sizeof(TransactionId)); } else ! onto->subxip = NULL; ! return onto; ! } ! ! /* ! * CopySnapshot ! * Copy the given snapshot. ! * ! * The copy is palloc'd in TopTransactionContext and has initial refcounts set ! * to 0. The returned snapshot has the copied flag set. ! */ ! static Snapshot ! CopySnapshot(Snapshot snapshot) ! { ! Snapshot newsnap; ! Size size; ! ! Assert(snapshot != InvalidSnapshot); ! ! /* We allocate any XID arrays needed in the same palloc block. */ ! size = sizeof(SnapshotData) + snapshot->xcnt * sizeof(TransactionId); ! if (snapshot->subxcnt > 0) ! size += snapshot->subxcnt * sizeof(TransactionId); ! ! newsnap = (Snapshot) MemoryContextAlloc(TopTransactionContext, size); ! ! return CopySnapshotOnto(snapshot, newsnap); } /* *************** AtEOXact_Snapshot(bool isCommit) *** 577,579 **** --- 633,1021 ---- FirstSnapshotSet = false; registered_xact_snapshot = false; } + + /* + * PreCommit_Snapshot + * Cleans up exported snapshots (this needs to happen before we update + * our MyProc entry, hence it is in PreCommit). + */ + void + PreCommit_Snapshot(void) + { + ListCell *snapshot; + int i; + char buf[MAXPGPATH]; + + if (exportedSnapshots == NIL) + return; + + Assert(list_length(exportedSnapshots) > 0); + Assert(TransactionIdIsValid(GetTopTransactionIdIfAny())); + + for(i = 1; i <= list_length(exportedSnapshots); i++) + { + XactExportFilePath(buf, GetTopTransactionId(), i, ""); + unlink(buf); + } + + foreach(snapshot, exportedSnapshots) + UnregisterSnapshotFromOwner(lfirst(snapshot), TopTransactionResourceOwner); + + exportedSnapshots = NIL; + } + + /* + * DeleteAllExportedSnapshotFiles + * Cleans up any files that have been left behind by a crashed backend + * that had exported snapshots before it died. + */ + void + DeleteAllExportedSnapshotFiles(void) + { + char buf[MAXPGPATH]; + DIR *s_dir; + struct dirent *s_de; + + if (!(s_dir = AllocateDir(SNAPSHOT_EXPORT_DIR))) + { + /* + * We really should have that directory in a sane cluster setup. But + * then again if we don't it's not fatal enough to make it FATAL. + */ + elog(WARNING, + "could not open directory \"%s\": %m", + SNAPSHOT_EXPORT_DIR); + return; + } + + while ((s_de = ReadDir(s_dir, SNAPSHOT_EXPORT_DIR)) != NULL) + { + if (strcmp(s_de->d_name, ".") == 0 || + strcmp(s_de->d_name, "..") == 0) + continue; + + snprintf(buf, MAXPGPATH, SNAPSHOT_EXPORT_DIR "/%s", s_de->d_name); + unlink(buf); + } + FreeDir(s_dir); + } + + /* + * ExportSnapshot + * Export the snapshot to a file so that other backends can import the same + * snapshot. + * Returns the token (the file name) that can be used to import this + * snapshot. + */ + static char * + ExportSnapshot(Snapshot snapshot) + { + #define SNAPSHOT_APPEND(x, y) (appendStringInfo(&buf, (x), (y))) + TransactionId *children, topXid; + FILE *f; + int i; + int nchildren; + MemoryContext oldcxt; + char path[MAXPGPATH]; + char pathtmp[MAXPGPATH]; + StringInfoData buf; + + Assert(IsTransactionState()); + + /* + * This will also assign a transaction id if we do not yet have one. + */ + topXid = GetTopTransactionId(); + + Assert(TransactionIdIsValid(GetTopTransactionIdIfAny())); + + /* + * We cannot export a snapshot from a subtransaction because in a + * subtransaction we don't see our open subxip values in the snapshot so + * they would be missing in the backend applying it. + */ + if (GetCurrentTransactionNestLevel() != 1) + ereport(ERROR, + (errcode(ERRCODE_ACTIVE_SQL_TRANSACTION), + errmsg("cannot export a snapshot from a subtransaction"))); + + /* + * We do however see our already committed subxip values and add them to + * the subxip array. + */ + nchildren = xactGetCommittedChildren(&children); + + initStringInfo(&buf); + + /* Write up all the data that we return */ + SNAPSHOT_APPEND("xid:%d ", topXid); + SNAPSHOT_APPEND("xmi:%d ", snapshot->xmin); + SNAPSHOT_APPEND("xma:%d ", snapshot->xmax); + /* Include our own transaction ID into the count. */ + SNAPSHOT_APPEND("xcnt:%d ", snapshot->xcnt + 1); + for (i = 0; i < snapshot->xcnt; i++) + SNAPSHOT_APPEND("xip:%d ", snapshot->xip[i]); + /* + * Finally add our own XID, since by definition we will still be running + * when the other transaction takes over the snapshot. + */ + SNAPSHOT_APPEND("xip:%d ", topXid); + if (snapshot->suboverflowed || snapshot->subxcnt + nchildren > TOTAL_MAX_CACHED_SUBXIDS) + SNAPSHOT_APPEND("sof:%d ", 1); + else + { + SNAPSHOT_APPEND("sxcnt:%d ", snapshot->subxcnt + nchildren); + for (i = 0; i < snapshot->subxcnt; i++) + SNAPSHOT_APPEND("sxp:%d ", snapshot->subxip[i]); + /* Add already committed subtransactions. */ + for (i = 0; i < nchildren; i++) + SNAPSHOT_APPEND("sxp:%d ", children[i]); + } + + /* + * buf ends with a trailing space but we leave it in for simplicity. The + * parsing routines also depend on it. + */ + + /* Register the snapshot and add it to the list of exported snapshots */ + snapshot = RegisterSnapshotOnOwner(snapshot, TopTransactionResourceOwner); + + oldcxt = MemoryContextSwitchTo(TopTransactionContext); + exportedSnapshots = lappend(exportedSnapshots, snapshot); + MemoryContextSwitchTo(oldcxt); + + XactExportFilePath(pathtmp, topXid, list_length(exportedSnapshots), ".tmp"); + if (!(f = AllocateFile(pathtmp, PG_BINARY_W))) + ereport(ERROR, + (errcode_for_file_access(), + errmsg("could not create file \"%s\": %m", pathtmp))); + + if (fwrite(buf.data, buf.len, 1, f) != 1) + /* Aborting the transaction will also call FreeFile() */ + ereport(ERROR, + (errcode_for_file_access(), + errmsg("could not write to file \"%s\": %m", pathtmp))); + + if (FreeFile(f)) + ereport(ERROR, + (errcode_for_file_access(), + errmsg("could not write to file \"%s\": %m", pathtmp))); + + /* + * Now that we have written everything into a .tmp file we rename the file + * and remove the .tmp suffix. Our filename is predictable and we're + * paranoid enough to not let us read a partially written file (we can't + * read a .tmp file because this would fail the valid characters check in + * ImportSnapshot). + */ + XactExportFilePath(path, topXid, list_length(exportedSnapshots), ""); + + if (rename(pathtmp, path) < 0) + ereport(WARNING, + (errcode_for_file_access(), + errmsg("could not rename file \"%s\" to \"%s\": %m", + pathtmp, path))); + + /* + * The basename of the file is what we return from pg_export_snapshot(). + * It's already in path in a textual format and we know that the path + * starts with SNAPSHOT_EXPORT_DIR. Skip over the prefix and over the + * slash and pstrdup it to not return a local variable. + */ + return pstrdup(path + strlen(SNAPSHOT_EXPORT_DIR) + 1); + #undef SNAPSHOT_APPEND + } + + /* + * Poor man's type independent parser. We only use it in the three functions + * below so there's no need to get ambitious about putting extra (x) around the + * arguments. + */ + #define SNAPSHOT_PARSE(valFunc, inFunc, type, strpp, prfx, notfound) \ + do { \ + char *n, *p = strstr(*strpp, prfx); \ + type v; \ + \ + if (!p) \ + return notfound; \ + p += strlen(prfx); \ + n = strchr(p, ' '); \ + if (!n) \ + return notfound; \ + *n = '\0'; \ + v = valFunc(DirectFunctionCall1(inFunc, CStringGetDatum(p))); \ + *strpp = n + 1; \ + return v; \ + } while (0); + + static int + parseIntFromText(char **s, const char *prefix) + { + SNAPSHOT_PARSE(DatumGetInt32, int4in, int, s, prefix, 0); + } + + static bool + parseBoolFromText(char **s, const char *prefix) + { + SNAPSHOT_PARSE(DatumGetInt32, int4in, bool, s, prefix, false); + } + + static TransactionId + parseXactFromText(char **s, const char *prefix) + { + SNAPSHOT_PARSE(DatumGetTransactionId, xidin, TransactionId, + s, prefix, InvalidTransactionId); + } + + #undef SNAPSHOT_PARSE + + /* + * ImportSnapshot + * Import a previously exported snapshot. We expect that whatever we get + * is a filename in SNAPSHOT_EXPORT_DIR. Load the snapshot from that file. + * This is called from "START TRANSACTION (SNAPSHOT = 'foo')" so we always + * start fresh from zero with respect to the transaction state that we + * work on. Returns true on success and false on failure. + */ + bool + ImportSnapshot(char *idstr) + { + char path[MAXPGPATH]; + FILE *f; + int i; + char *s; + struct stat stat_buf; + int sxcnt, xcnt; + TransactionId xid, origXid, myXid; + SnapshotData snapshot = {HeapTupleSatisfiesMVCC}; + + /* + * If we were in read committed mode then the next query would execute with a + * new snapshot thus making this function call quite useless. + */ + if (!IsolationUsesXactSnapshot()) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("A snapshot importing transaction must have ISOLATION " + "LEVEL SERIALIZABLE or ISOLATION LEVEL REPEATABLE READ"))); + + /* We're lucky to always start off from a pretty clean state */ + Assert(IsTransactionState()); + Assert(GetCurrentTransactionNestLevel() == 1); + Assert(GetTopTransactionIdIfAny() == InvalidTransactionId); + Assert(CurrentSnapshot == NULL); + Assert(SecondarySnapshot == NULL); + + /* verify the identifier, only 0-9,A-F and a hyphen are allowed... */ + s = idstr; + while (*s) + { + if (!isdigit(*s) && !(*s >= 'A' && *s <= 'F') && *s != '-') + return false; + s++; + } + + /* + * Assign a transaction id. We only do this to detect a possible + * transaction id wraparound which is somewhere between unlikely + * and impossible... + */ + myXid = GetTopTransactionId(); + + snprintf(path, MAXPGPATH, SNAPSHOT_EXPORT_DIR "/%s", idstr); + + /* get the size of the file so that we know how much memory we need */ + if (stat(path, &stat_buf) != 0) + return false; + + if (!(f = AllocateFile(path, PG_BINARY_R))) + return false; + + s = palloc(stat_buf.st_size + 1); + if (fread(s, stat_buf.st_size, 1, f) != 1) + return false; + + s[stat_buf.st_size] = '\0'; + + FreeFile(f); + + origXid = parseXactFromText(&s, "xid:"); + + snapshot.xmin = parseXactFromText(&s, "xmi:"); + Assert(snapshot.xmin != InvalidTransactionId); + snapshot.xmax = parseXactFromText(&s, "xma:"); + Assert(snapshot.xmax != InvalidTransactionId); + + xcnt = parseIntFromText(&s, "xcnt:"); + /* + * This snapshot only serves as a template, there is no need for it to have + * maxProcs entries, so let's make it just as large as we need it. + */ + snapshot.xip = palloc(xcnt * sizeof(TransactionId)); + + i = 0; + while ((xid = parseXactFromText(&s, "xip:")) != InvalidTransactionId) + snapshot.xip[i++] = xid; + snapshot.xcnt = i; + Assert(snapshot.xcnt == xcnt); + + /* + * We only write "sof:1" if the snapshot overflowed. If not, then there is + * no "sof:x" entry at all and parseBoolFromText() will return false. + */ + snapshot.suboverflowed = parseBoolFromText(&s, "sof:"); + + if (!snapshot.suboverflowed) + { + sxcnt = parseIntFromText(&s, "sxcnt:"); + snapshot.subxip = palloc(sxcnt * sizeof(TransactionId)); + + i = 0; + while ((xid = parseXactFromText(&s, "sxp:")) != InvalidTransactionId) + snapshot.subxip[i++] = xid; + snapshot.subxcnt = i; + Assert(snapshot.subxcnt == sxcnt); + } else { + snapshot.subxip = NULL; + snapshot.subxcnt = 0; + } + + /* complete the snapshot data structure */ + snapshot.curcid = 0; + snapshot.takenDuringRecovery = RecoveryInProgress(); + + /* + * Note that MyProc->xmin can go backwards here. However this is safe + * because the xmin we set here is the same as in the backend's proc->xmin + * whose snapshot we are copying. At this very moment, anybody computing a + * minimum will calculate at least this xmin as the overall xmin with or + * without us setting MyProc->xmin to this value. + */ + LWLockAcquire(ProcArrayLock, LW_SHARED); + MyProc->xmin = snapshot.xmin; + LWLockRelease(ProcArrayLock); + + /* bail out if the original transaction is not running anymore... */ + if (!TransactionIdIsInProgress(origXid) || TransactionIdPrecedes(myXid, origXid)) + return false; + + /* + * Install the snapshot as if we got it through GetTransactionSnapshot(). + * This will set up CurrentSnapshot and also set up the predicate locks for a + * serializable transaction. + */ + GetTransactionSnapshotFromTemplate(&snapshot); + return true; + } + + Datum + pg_export_snapshot(PG_FUNCTION_ARGS) + { + char *snapshotData; + + RequireTransactionChain(true, "pg_export_snapshot()"); + + snapshotData = ExportSnapshot(GetTransactionSnapshot()); + PG_RETURN_TEXT_P(cstring_to_text(snapshotData)); + } + diff --git a/src/bin/initdb/initdb.c b/src/bin/initdb/initdb.c index 399c734..c1b80da 100644 *** a/src/bin/initdb/initdb.c --- b/src/bin/initdb/initdb.c *************** main(int argc, char *argv[]) *** 2598,2603 **** --- 2598,2604 ---- "pg_serial", "pg_subtrans", "pg_twophase", + "pg_snapshots", "pg_multixact/members", "pg_multixact/offsets", "base", diff --git a/src/include/access/twophase.h b/src/include/access/twophase.h index 799bf8b..bd2ae80 100644 *** a/src/include/access/twophase.h --- b/src/include/access/twophase.h *************** *** 25,33 **** */ typedef struct GlobalTransactionData *GlobalTransaction; - /* GUC variable */ - extern int max_prepared_xacts; - extern Size TwoPhaseShmemSize(void); extern void TwoPhaseShmemInit(void); --- 25,30 ---- diff --git a/src/include/catalog/pg_proc.h b/src/include/catalog/pg_proc.h index 96f43fe..a4e0387 100644 *** a/src/include/catalog/pg_proc.h --- b/src/include/catalog/pg_proc.h *************** DATA(insert OID = 2171 ( pg_cancel_backe *** 2853,2858 **** --- 2853,2860 ---- DESCR("cancel a server process' current query"); DATA(insert OID = 2096 ( pg_terminate_backend PGNSP PGUID 12 1 0 0 0 f f f t f v 1 0 16 "23" _null_ _null_ _null_ _null_ pg_terminate_backend _null_ _null_ _null_ )); DESCR("terminate a server process"); + DATA(insert OID = 3122 ( pg_export_snapshot PGNSP PGUID 12 1 0 0 0 f f f t f v 0 0 25 "" _null_ _null_ _null_ _null_ pg_export_snapshot _null_ _null_ _null_ )); + DESCR("export a snapshot"); DATA(insert OID = 2172 ( pg_start_backup PGNSP PGUID 12 1 0 0 0 f f f t f v 2 0 25 "25 16" _null_ _null_ _null_ _null_ pg_start_backup _null_ _null_ _null_ )); DESCR("prepare for taking an online backup"); DATA(insert OID = 2173 ( pg_stop_backup PGNSP PGUID 12 1 0 0 0 f f f t f v 0 0 25 "" _null_ _null_ _null_ _null_ pg_stop_backup _null_ _null_ _null_ )); diff --git a/src/include/miscadmin.h b/src/include/miscadmin.h index 9d19417..15326cf 100644 *** a/src/include/miscadmin.h --- b/src/include/miscadmin.h *************** extern PGDLLIMPORT int NBuffers; *** 134,139 **** --- 134,142 ---- extern int MaxBackends; extern int MaxConnections; + /* GUC variable */ + extern int max_prepared_xacts; + extern PGDLLIMPORT int MyProcPid; extern PGDLLIMPORT pg_time_t MyStartTime; extern PGDLLIMPORT struct Port *MyProcPort; diff --git a/src/include/storage/predicate.h b/src/include/storage/predicate.h index 5ddbc1d..f4f0303 100644 *** a/src/include/storage/predicate.h --- b/src/include/storage/predicate.h *************** extern void CheckPointPredicate(void); *** 42,48 **** extern bool PageIsPredicateLocked(Relation relation, BlockNumber blkno); /* predicate lock maintenance */ ! extern Snapshot RegisterSerializableTransaction(Snapshot snapshot); extern void RegisterPredicateLockingXid(TransactionId xid); extern void PredicateLockRelation(Relation relation, Snapshot snapshot); extern void PredicateLockPage(Relation relation, BlockNumber blkno, Snapshot snapshot); --- 42,48 ---- extern bool PageIsPredicateLocked(Relation relation, BlockNumber blkno); /* predicate lock maintenance */ ! extern Snapshot RegisterSerializableTransaction(Snapshot snapshot, Snapshot stemplate); extern void RegisterPredicateLockingXid(TransactionId xid); extern void PredicateLockRelation(Relation relation, Snapshot snapshot); extern void PredicateLockPage(Relation relation, BlockNumber blkno, Snapshot snapshot); diff --git a/src/include/storage/procarray.h b/src/include/storage/procarray.h index 3c20fc4..a2440e4 100644 *** a/src/include/storage/procarray.h --- b/src/include/storage/procarray.h *************** extern void ExpireOldKnownAssignedTransa *** 41,47 **** extern RunningTransactions GetRunningTransactionData(void); ! extern Snapshot GetSnapshotData(Snapshot snapshot); extern bool TransactionIdIsInProgress(TransactionId xid); extern bool TransactionIdIsActive(TransactionId xid); --- 41,47 ---- extern RunningTransactions GetRunningTransactionData(void); ! extern Snapshot GetSnapshotData(Snapshot snapshot, Snapshot stemplate); extern bool TransactionIdIsInProgress(TransactionId xid); extern bool TransactionIdIsActive(TransactionId xid); *************** extern void XidCacheRemoveRunningXids(Tr *** 71,74 **** --- 71,80 ---- int nxids, const TransactionId *xids, TransactionId latestXid); + /* Size of the ProcArray structure itself */ + #define PROCARRAY_MAXPROCS (MaxBackends + max_prepared_xacts) + + #define TOTAL_MAX_CACHED_SUBXIDS \ + ((PGPROC_MAX_CACHED_SUBXIDS + 1) * PROCARRAY_MAXPROCS) + #endif /* PROCARRAY_H */ diff --git a/src/include/utils/snapmgr.h b/src/include/utils/snapmgr.h index a7e7d3d..47f62ab 100644 *** a/src/include/utils/snapmgr.h --- b/src/include/utils/snapmgr.h *************** extern TransactionId TransactionXmin; *** 23,28 **** --- 23,30 ---- extern TransactionId RecentXmin; extern TransactionId RecentGlobalXmin; + extern List *exportedSnapshots; + extern Snapshot GetTransactionSnapshot(void); extern Snapshot GetLatestSnapshot(void); extern void SnapshotSetCommandId(CommandId curcid); *************** extern void UpdateActiveSnapshotCommandI *** 33,47 **** --- 35,55 ---- extern void PopActiveSnapshot(void); extern Snapshot GetActiveSnapshot(void); extern bool ActiveSnapshotSet(void); + extern Snapshot CopySnapshotOnto(Snapshot onto, Snapshot snapshot); extern Snapshot RegisterSnapshot(Snapshot snapshot); extern void UnregisterSnapshot(Snapshot snapshot); extern Snapshot RegisterSnapshotOnOwner(Snapshot snapshot, ResourceOwner owner); extern void UnregisterSnapshotFromOwner(Snapshot snapshot, ResourceOwner owner); + extern void PreCommit_Snapshot(void); extern void AtSubCommit_Snapshot(int level); extern void AtSubAbort_Snapshot(int level); extern void AtEarlyCommit_Snapshot(void); extern void AtEOXact_Snapshot(bool isCommit); + extern Datum pg_export_snapshot(PG_FUNCTION_ARGS); + extern bool ImportSnapshot(char *idstr); + extern void DeleteAllExportedSnapshotFiles(void); + #endif /* SNAPMGR_H */