DRY code for RecoverPreparedFromFiles and RecoverPreparedFromXLOG

kelvich · kelvich · commit 62d662f84f45 · 2016-03-29T20:04:35.000+03:00
diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
@@ -98,64 +98,6 @@
 /* GUC variable, can't be changed after startup */
 int			max_prepared_xacts = 0;
 
-/*
- * This struct describes one global transaction that is in prepared state
- * or attempting to become prepared.
- *
- * The lifecycle of a global transaction is:
- *
- * 1. After checking that the requested GID is not in use, set up an entry in
- * the TwoPhaseState->prepXacts array with the correct GID and valid = false,
- * and mark it as locked by my backend.
- *
- * 2. After successfully completing prepare, set valid = true and enter the
- * referenced PGPROC into the global ProcArray.
- *
- * 3. To begin COMMIT PREPARED or ROLLBACK PREPARED, check that the entry is
- * valid and not locked, then mark the entry as locked by storing my current
- * backend ID into locking_backend.  This prevents concurrent attempts to
- * commit or rollback the same prepared xact.
- *
- * 4. On completion of COMMIT PREPARED or ROLLBACK PREPARED, remove the entry
- * from the ProcArray and the TwoPhaseState->prepXacts array and return it to
- * the freelist.
- *
- * Note that if the preparing transaction fails between steps 1 and 2, the
- * entry must be removed so that the GID and the GlobalTransaction struct
- * can be reused.  See AtAbort_Twophase().
- *
- * typedef struct GlobalTransactionData *GlobalTransaction appears in
- * twophase.h
- *
- * Note that the max value of GIDSIZE must fit in the uint16 gidlen,
- * specified in TwoPhaseFileHeader.
- */
-#define GIDSIZE 200
-
-typedef struct GlobalTransactionData
-{
-	GlobalTransaction next;		/* list link for free list */
-	int			pgprocno;		/* ID of associated dummy PGPROC */
-	BackendId	dummyBackendId; /* similar to backend id for backends */
-	TimestampTz prepared_at;	/* time of preparation */
-
-	/*
-	 * Note that we need to keep track of two LSNs for each GXACT.
-	 * We keep track of the start LSN because this is the address we must
-	 * use to read state data back from WAL when committing a prepared GXACT.
-	 * We keep track of the end LSN because that is the LSN we need to wait
-	 * for prior to commit.
-	 */
-	XLogRecPtr	prepare_start_lsn;	/* XLOG offset of prepare record start */
-	XLogRecPtr	prepare_end_lsn;	/* XLOG offset of prepare record end */
-
-	Oid			owner;			/* ID of user that executed the xact */
-	BackendId	locking_backend;	/* backend currently working on the xact */
-	bool		valid;			/* TRUE if PGPROC entry is in proc array */
-	bool		ondisk;			/* TRUE if prepare state file is on disk */
-	char		gid[GIDSIZE];	/* The GID assigned to the prepared xact */
-}	GlobalTransactionData;
-
 /*
  * Two Phase Commit shared state.  Access to this struct is protected
  * by TwoPhaseStateLock.
@@ -487,7 +429,7 @@ GXactLoadSubxactData(GlobalTransaction gxact, int nsubxacts,
  * MarkAsPrepared
  *		Mark the GXACT as fully valid, and enter it into the global ProcArray.
  */
-static void
+void
 MarkAsPrepared(GlobalTransaction gxact)
 {
 	/* Lock here may be overkill, but I'm not convinced of that ... */
@@ -1967,6 +1909,93 @@ PrescanPreparedTransactions(TransactionId **xids_p, int *nxids_p)
 	return result;
 }
 
+/*
+ * RecoverPreparedFromBuffer
+ *
+ * Parse data in given buffer (that can be a pointer to WAL record or file)
+ * and load shared-memory state for that prepared transaction.
+ *
+ * It's caller responsibility to call MarkAsPrepared() on returned gxact.
+ *
+ */
+GlobalTransaction
+RecoverPreparedFromBuffer(char *buf, bool forceOverwriteOK)
+{
+	char			*bufptr;
+	const char		*gid;
+	TransactionId	*subxids;
+	bool			overwriteOK = false;
+	int				i;
+	GlobalTransaction gxact;
+	TwoPhaseFileHeader	*hdr;
+
+	/* Deconstruct header */
+	hdr = (TwoPhaseFileHeader *) buf;
+	bufptr = buf + MAXALIGN(sizeof(TwoPhaseFileHeader));
+	gid = (const char *) bufptr;
+	bufptr += MAXALIGN(hdr->gidlen);
+	subxids = (TransactionId *) bufptr;
+	bufptr += MAXALIGN(hdr->nsubxacts * sizeof(TransactionId));
+	bufptr += MAXALIGN(hdr->ncommitrels * sizeof(RelFileNode));
+	bufptr += MAXALIGN(hdr->nabortrels * sizeof(RelFileNode));
+	bufptr += MAXALIGN(hdr->ninvalmsgs * sizeof(SharedInvalidationMessage));
+
+	/*
+	 * It's possible that SubTransSetParent has been set before, if
+	 * the prepared transaction generated xid assignment records. Test
+	 * here must match one used in AssignTransactionId().
+	 */
+	if (InHotStandby && (hdr->nsubxacts >= PGPROC_MAX_CACHED_SUBXIDS ||
+						 XLogLogicalInfoActive()))
+		overwriteOK = true;
+
+	/*
+	 * Caller can also force overwriteOK.
+	 */
+	if (forceOverwriteOK)
+		overwriteOK = true;
+
+	/*
+	 * Reconstruct subtrans state for the transaction --- needed
+	 * because pg_subtrans is not preserved over a restart.  Note that
+	 * we are linking all the subtransactions directly to the
+	 * top-level XID; there may originally have been a more complex
+	 * hierarchy, but there's no need to restore that exactly.
+	 */
+	for (i = 0; i < hdr->nsubxacts; i++)
+		SubTransSetParent(subxids[i], hdr->xid, overwriteOK);
+
+	/*
+	 * Recreate its GXACT and dummy PGPROC
+	 */
+	gxact = MarkAsPreparing(hdr->xid, gid,
+							hdr->prepared_at,
+							hdr->owner, hdr->database);
+	GXactLoadSubxactData(gxact, hdr->nsubxacts, subxids);
+
+	/*
+	 * Recover other state (notably locks) using resource managers
+	 */
+	ProcessRecords(bufptr, hdr->xid, twophase_recover_callbacks);
+
+	/*
+	 * Release locks held by the standby process after we process each
+	 * prepared transaction. As a result, we don't need too many
+	 * additional locks at any one time.
+	 */
+	if (InHotStandby)
+		StandbyReleaseLockTree(hdr->xid, hdr->nsubxacts, subxids);
+
+	/*
+	 * We're done with recovering this transaction. Clear
+	 * MyLockedGxact, like we do in PrepareTransaction() during normal
+	 * operation.
+	 */
+	PostPrepare_Twophase();
+
+	return gxact;
+}
+
 /*
  * RecoverPreparedFromFiles
  *
@@ -1980,7 +2009,6 @@ RecoverPreparedFromFiles(bool forceOverwriteOK)
 	char		dir[MAXPGPATH];
 	DIR		   *cldir;
 	struct dirent *clde;
-	bool		overwriteOK = false;
 
 	snprintf(dir, MAXPGPATH, "%s", TWOPHASE_DIR);
 
@@ -1992,11 +2020,7 @@ RecoverPreparedFromFiles(bool forceOverwriteOK)
 		{
 			TransactionId xid;
 			char	   *buf;
-			char	   *bufptr;
-			TwoPhaseFileHeader *hdr;
-			TransactionId *subxids;
 			GlobalTransaction gxact;
-			const char	*gid;
 			int			i;
 			PGXACT	   *pgxact;
 
@@ -2041,73 +2065,10 @@ RecoverPreparedFromFiles(bool forceOverwriteOK)
 			ereport(LOG,
 					(errmsg("recovering prepared transaction %u", xid)));
 
-			/* Deconstruct header */
-			hdr = (TwoPhaseFileHeader *) buf;
-			Assert(TransactionIdEquals(hdr->xid, xid));
-			bufptr = buf + MAXALIGN(sizeof(TwoPhaseFileHeader));
-			gid = (const char *) bufptr;
-			bufptr += MAXALIGN(hdr->gidlen);
-			subxids = (TransactionId *) bufptr;
-			bufptr += MAXALIGN(hdr->nsubxacts * sizeof(TransactionId));
-			bufptr += MAXALIGN(hdr->ncommitrels * sizeof(RelFileNode));
-			bufptr += MAXALIGN(hdr->nabortrels * sizeof(RelFileNode));
-			bufptr += MAXALIGN(hdr->ninvalmsgs * sizeof(SharedInvalidationMessage));
-
-			/*
-			 * It's possible that SubTransSetParent has been set before, if
-			 * the prepared transaction generated xid assignment records. Test
-			 * here must match one used in AssignTransactionId().
-			 */
-			if (InHotStandby && (hdr->nsubxacts >= PGPROC_MAX_CACHED_SUBXIDS ||
-								 XLogLogicalInfoActive()))
-				overwriteOK = true;
-
-			/*
-			 * Caller can also force overwriteOK.
-			 */
-			if (forceOverwriteOK)
-				overwriteOK = true;
-
-			/*
-			 * Reconstruct subtrans state for the transaction --- needed
-			 * because pg_subtrans is not preserved over a restart.  Note that
-			 * we are linking all the subtransactions directly to the
-			 * top-level XID; there may originally have been a more complex
-			 * hierarchy, but there's no need to restore that exactly.
-			 */
-			for (i = 0; i < hdr->nsubxacts; i++)
-				SubTransSetParent(subxids[i], xid, overwriteOK);
-
-			/*
-			 * Recreate its GXACT and dummy PGPROC
-			 */
-			gxact = MarkAsPreparing(xid, gid,
-									hdr->prepared_at,
-									hdr->owner, hdr->database);
+			gxact = RecoverPreparedFromBuffer(buf, forceOverwriteOK);
 			gxact->ondisk = true;
-			GXactLoadSubxactData(gxact, hdr->nsubxacts, subxids);
 			MarkAsPrepared(gxact);
 
-			/*
-			 * Recover other state (notably locks) using resource managers
-			 */
-			ProcessRecords(bufptr, xid, twophase_recover_callbacks);
-
-			/*
-			 * Release locks held by the standby process after we process each
-			 * prepared transaction. As a result, we don't need too many
-			 * additional locks at any one time.
-			 */
-			if (InHotStandby)
-				StandbyReleaseLockTree(xid, hdr->nsubxacts, subxids);
-
-			/*
-			 * We're done with recovering this transaction. Clear
-			 * MyLockedGxact, like we do in PrepareTransaction() during normal
-			 * operation.
-			 */
-			PostPrepare_Twophase();
-
 			pfree(buf);
 		}
 
@@ -2119,99 +2080,6 @@ RecoverPreparedFromFiles(bool forceOverwriteOK)
 }
 
 
-/*
- * RecoverPreparedFromXLOG
- *
- * To avoid creation of state files during replay we registering
- * prepare xlog records in shared memory in the same way as it happens
- * while not in recovery. If replay faces commit xlog record before
- * checkpoint/restartpoint happens then we avoid using files at all.
- *
- * We need this behaviour because the speed of the 2PC replay on the replica
- * should be at least the same as the 2PC transaction speed of the master.
- */
-void
-RecoverPreparedFromXLOG(XLogReaderState *record)
-{
-	bool		overwriteOK = false;
-	TransactionId xid = XLogRecGetXid(record);
-	char *buf = (char *) XLogRecGetData(record);
-	char	   *bufptr;
-	const char *gid;
-	TwoPhaseFileHeader *hdr;
-	TransactionId *subxids;
-	GlobalTransaction gxact;
-	int			i;
-
-	/* Deconstruct header */
-	hdr = (TwoPhaseFileHeader *) buf;
-	Assert(TransactionIdEquals(hdr->xid, xid));
-	bufptr = buf + MAXALIGN(sizeof(TwoPhaseFileHeader));
-	gid = (const char *) bufptr;
-	bufptr += MAXALIGN(hdr->gidlen);
-	subxids = (TransactionId *) bufptr;
-	bufptr += MAXALIGN(hdr->nsubxacts * sizeof(TransactionId));
-	bufptr += MAXALIGN(hdr->ncommitrels * sizeof(RelFileNode));
-	bufptr += MAXALIGN(hdr->nabortrels * sizeof(RelFileNode));
-	bufptr += MAXALIGN(hdr->ninvalmsgs * sizeof(SharedInvalidationMessage));
-
-	/*
-	 * It's possible that SubTransSetParent has been set before, if
-	 * the prepared transaction generated xid assignment records. Test
-	 * here must match one used in AssignTransactionId().
-	 */
-	if (InHotStandby && (hdr->nsubxacts >= PGPROC_MAX_CACHED_SUBXIDS ||
-						 XLogLogicalInfoActive()))
-		overwriteOK = true;
-
-	/*
-	 * Reconstruct subtrans state for the transaction --- needed
-	 * because pg_subtrans is not preserved over a restart.  Note that
-	 * we are linking all the subtransactions directly to the
-	 * top-level XID; there may originally have been a more complex
-	 * hierarchy, but there's no need to restore that exactly.
-	 */
-	for (i = 0; i < hdr->nsubxacts; i++)
-		SubTransSetParent(subxids[i], xid, overwriteOK);
-
-	/*
-	 * Recreate its GXACT and dummy PGPROC
-	 *
-	 * MarkAsPreparing sets prepare_start_lsn to InvalidXLogRecPtr
-	 * so next checkpoint will skip that transaction.
-	 */
-	gxact = MarkAsPreparing(xid, gid,
-							hdr->prepared_at,
-							hdr->owner, hdr->database);
-	GXactLoadSubxactData(gxact, hdr->nsubxacts, subxids);
-	MarkAsPrepared(gxact);
-
-	gxact->prepare_start_lsn = record->ReadRecPtr;
-	gxact->prepare_end_lsn = record->EndRecPtr;
-
-	/*
-	 * Recover other state (notably locks) using resource managers
-	 */
-	ProcessRecords(bufptr, xid, twophase_recover_callbacks);
-
-	/*
-	 * Release locks held by the standby process after we process each
-	 * prepared transaction. As a result, we don't need too many
-	 * additional locks at any one time.
-	 */
-	if (InHotStandby)
-		StandbyReleaseLockTree(xid, hdr->nsubxacts, subxids);
-
-	/*
-	 * We're done with recovering this transaction. Clear
-	 * MyLockedGxact, like we do in PrepareTransaction() during normal
-	 * operation.
-	 */
-	PostPrepare_Twophase();
-}
-
-
-
 /*
  *	RecordTransactionCommitPrepared
  *
diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c
@@ -5599,7 +5599,21 @@ xact_redo(XLogReaderState *record)
 	}
 	else if (info == XLOG_XACT_PREPARE)
 	{
-		RecoverPreparedFromXLOG(record);
+		GlobalTransaction gxact;
+
+		/*
+		 * To avoid creation of state files during replay we registering
+		 * prepare xlog records in shared memory in the same way as it happens
+		 * while not in recovery. If replay faces commit xlog record before
+		 * checkpoint/restartpoint happens then we avoid using files at all.
+		 *
+		 * We need this behaviour because the speed of the 2PC replay on the replica
+		 * should be at least the same as the 2PC transaction speed of the master.
+		 */
+		gxact = RecoverPreparedFromBuffer((char *) XLogRecGetData(record), false);
+		gxact->prepare_start_lsn = record->ReadRecPtr;
+		gxact->prepare_end_lsn = record->EndRecPtr;
+		MarkAsPrepared(gxact);
 	}
 	else if (info == XLOG_XACT_ASSIGNMENT)
 	{
diff --git a/src/include/access/twophase.h b/src/include/access/twophase.h

Original file line number	Diff line number	Diff line change
`@@ -5599,7 +5599,21 @@ xact_redo(XLogReaderState *record)`
`5599`	`5599`	`}`
`5600`	`5600`	`else if (info == XLOG_XACT_PREPARE)`
`5601`	`5601`	`{`
`5602`		`- RecoverPreparedFromXLOG(record);`
	`5602`	`+ GlobalTransaction gxact;`
	`5603`	`+`
	`5604`	`+ /*`
	`5605`	`+ * To avoid creation of state files during replay we registering`
	`5606`	`+ * prepare xlog records in shared memory in the same way as it happens`
	`5607`	`+ * while not in recovery. If replay faces commit xlog record before`
	`5608`	`+ * checkpoint/restartpoint happens then we avoid using files at all.`
	`5609`	`+ *`
	`5610`	`+ * We need this behaviour because the speed of the 2PC replay on the replica`
	`5611`	`+ * should be at least the same as the 2PC transaction speed of the master.`
	`5612`	`+ */`
	`5613`	`+ gxact = RecoverPreparedFromBuffer((char *) XLogRecGetData(record), false);`
	`5614`	`+ gxact->prepare_start_lsn = record->ReadRecPtr;`
	`5615`	`+ gxact->prepare_end_lsn = record->EndRecPtr;`
	`5616`	`+ MarkAsPrepared(gxact);`
`5603`	`5617`	`}`
`5604`	`5618`	`else if (info == XLOG_XACT_ASSIGNMENT)`
`5605`	`5619`	`{`