pick new 2pc stuff

Author: alvherre

src/backend/access/transam/twophase.c

@@ -270,7 +270,10 @@ static struct xllist
 static TwoPhaseStateData *TwoPhaseState;
 
 /*
- * Global transaction entry currently locked by us, if any.
+ * Global transaction entry currently locked by us, if any.  Note that any
+ * access to the entry pointed to by this variable must be protected by
+ * TwoPhaseStateLock, though obviously the pointer itself doesn't need to be
+ * (since it's just local memory).
  */
 static GlobalTransaction MyLockedGxact = NULL;
 
@@ -418,18 +421,13 @@ AtAbort_Twophase(void)
 	 * resources held by the transaction yet.  In those cases, the in-memory
 	 * state can be wrong, but it's too late to back out.
 	 */
+	LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
 	if (!MyLockedGxact->valid)
-	{
 		RemoveGXact(MyLockedGxact);
-	}
 	else
-	{
-		LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
-
 		MyLockedGxact->locking_pid = InvalidBackendId;
+	LWLockRelease(TwoPhaseStateLock);
 
-		LWLockRelease(TwoPhaseStateLock);
-	}
 	MyLockedGxact = NULL;
 }
 
@@ -534,6 +532,8 @@ MarkAsPreparingGuts(GlobalTransaction gxact, TransactionId xid, const char *gid,
 	PGXACT	   *pgxact;
 	int			i;
 
+	Assert(LWLockHeldByMeInMode(TwoPhaseStateLock, LW_EXCLUSIVE));
+
 	Assert(gxact != NULL);
 	proc = &ProcGlobal->allProcs[gxact->pgprocno];
 	pgxact = &ProcGlobal->allPgXact[gxact->pgprocno];
@@ -611,15 +611,19 @@ GXactLoadSubxactData(GlobalTransaction gxact, int nsubxacts,
 /*
  * MarkAsPrepared
  *		Mark the GXACT as fully valid, and enter it into the global ProcArray.
+ *
+ * lock_held indicates whether caller already holds TwoPhaseStateLock.
  */
 static void
-MarkAsPrepared(GlobalTransaction gxact)
+MarkAsPrepared(GlobalTransaction gxact, bool lock_held)
 {
 	/* Lock here may be overkill, but I'm not convinced of that ... */
-	LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
+	if (!lock_held)
+		LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
 	Assert(!gxact->valid);
 	gxact->valid = true;
-	LWLockRelease(TwoPhaseStateLock);
+	if (!lock_held)
+		LWLockRelease(TwoPhaseStateLock);
 
 	/*
 	 * Put it into the global ProcArray so TransactionIdIsInProgress considers
@@ -714,7 +718,7 @@ RemoveGXact(GlobalTransaction gxact)
 {
 	int			i;
 
-	LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
+	Assert(LWLockHeldByMeInMode(TwoPhaseStateLock, LW_EXCLUSIVE));
 
 	for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
 	{
@@ -728,15 +732,10 @@ RemoveGXact(GlobalTransaction gxact)
 			gxact->next = TwoPhaseState->freeGXacts;
 			TwoPhaseState->freeGXacts = gxact;
 
-			gxact->locking_pid = InvalidBackendId;
-			LWLockRelease(TwoPhaseStateLock);
-
 			return;
 		}
 	}
 
-	LWLockRelease(TwoPhaseStateLock);
-
 	elog(ERROR, "failed to find %p in GlobalTransaction array", gxact);
 }
 
@@ -1261,7 +1260,7 @@ EndPrepare(GlobalTransaction gxact)
 	 * the xact crashed.  Instead we have a window where the same XID appears
 	 * twice in ProcArray, which is OK.
 	 */
-	MarkAsPrepared(gxact);
+	MarkAsPrepared(gxact, false);
 
 	/*
 	 * Now we can mark ourselves as out of the commit critical section: a
@@ -1688,7 +1687,9 @@ FinishPreparedTransaction(const char *gid, bool isCommit)
 	if (gxact->ondisk)
 		RemoveTwoPhaseFile(xid, true);
 
+	LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
 	RemoveGXact(gxact);
+	LWLockRelease(TwoPhaseStateLock);
 	MyLockedGxact = NULL;
 
 	pfree(buf);
@@ -1917,6 +1918,7 @@ restoreTwoPhaseData(void)
 	struct dirent  *clde;
 
 	cldir = AllocateDir(TWOPHASE_DIR);
+	LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
 	while ((clde = ReadDir(cldir, TWOPHASE_DIR)) != NULL)
 	{
 		if (strlen(clde->d_name) == 2*sizeof(TransactionId) &&
@@ -1936,6 +1938,7 @@ restoreTwoPhaseData(void)
 			PrepareRedoAdd(buf, InvalidXLogRecPtr, InvalidXLogRecPtr);
 		}
 	}
+	LWLockRelease(TwoPhaseStateLock);
 	FreeDir(cldir);
 }
 
@@ -1977,7 +1980,7 @@ PrescanPreparedTransactions(TransactionId **xids_p, int *nxids_p)
 	int			allocsize = 0;
 	int			i;
 
-	LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
+	LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
 	for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
 	{
 		TransactionId xid;
@@ -2055,7 +2058,7 @@ StandbyRecoverPreparedTransactions(bool overwriteOK)
 {
 	int			i;
 
-	LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
+	LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
 	for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
 	{
 		TransactionId xid;
@@ -2082,16 +2085,22 @@ StandbyRecoverPreparedTransactions(bool overwriteOK)
  * Scan the shared memory entries of TwoPhaseState and reload the state for
  * each prepared transaction (reacquire locks, etc).
  *
- * This is run during database startup.
+ * This is run at the end of recovery, but before we allow backends to write
+ * WAL.
+ *
+ * At the end of recovery the way we take snapshots will change. We now need
+ * to mark all running transactions with their full SubTransSetParent() info
+ * to allow normal snapshots to work correctly if snapshots overflow.
+ * We do this here because by definition prepared transactions are the only
+ * type of write transaction still running, so this is necessary and
+ * complete.
  */
 void
 RecoverPreparedTransactions(void)
 {
 	int			i;
 
-	/*
-	 * Don't need a lock in the recovery phase.
-	 */
+	LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
 	for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
 	{
 		TransactionId xid;
@@ -2114,7 +2123,7 @@ RecoverPreparedTransactions(void)
 			continue;
 
 		ereport(LOG,
-				(errmsg("recovering prepared transaction %u from shared memory", xid)));
+				(errmsg("recovering prepared transaction " XID_FMT " from shared memory", xid)));
 
 		hdr = (TwoPhaseFileHeader *) buf;
 		Assert(TransactionIdEquals(hdr->xid, xid));
@@ -2151,21 +2160,20 @@ RecoverPreparedTransactions(void)
 		 * it was added in redo and already has a shmem entry for
 		 * it.
 		 */
-		LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
 		MarkAsPreparingGuts(gxact, xid, gid,
 							hdr->prepared_at,
 							hdr->owner, hdr->database);
 
 		/* recovered, so reset the flag for entries generated by redo */
 		gxact->inredo = false;
 
-		LWLockRelease(TwoPhaseStateLock);
-
 		GXactLoadSubxactData(gxact, hdr->nsubxacts, subxids);
-		MarkAsPrepared(gxact);
+		MarkAsPrepared(gxact, true);
+
+		LWLockRelease(TwoPhaseStateLock);
 
 		/*
-		 * Recover other state (notably locks) using resource managers
+		 * Recover other state (notably locks) using resource managers.
 		 */
 		ProcessRecords(bufptr, xid, twophase_recover_callbacks);
 
@@ -2185,7 +2193,11 @@ RecoverPreparedTransactions(void)
 		PostPrepare_Twophase();
 
 		pfree(buf);
+
+		LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
 	}
+
+	LWLockRelease(TwoPhaseStateLock);
 }
 
 /*
@@ -2216,6 +2228,8 @@ ProcessTwoPhaseBuffer(TransactionId xid,
 	TwoPhaseFileHeader *hdr;
 	int			i;
 
+	Assert(LWLockHeldByMeInMode(TwoPhaseStateLock, LW_EXCLUSIVE));
+
 	if (!fromdisk)
 		Assert(prepare_start_lsn != InvalidXLogRecPtr);
 
@@ -2230,15 +2244,15 @@ ProcessTwoPhaseBuffer(TransactionId xid,
 		if (fromdisk)
 		{
 			ereport(WARNING,
-					(errmsg("removing stale two-phase state file for \"%u\"",
+					(errmsg("removing stale two-phase state file for \"" XID_FMT "\"",
 							xid)));
 			RemoveTwoPhaseFile(xid, true);
 		}
 		else
 		{
 			ereport(WARNING,
-					(errmsg("removing stale two-phase state from"
-							" shared memory for \"%u\"", xid)));
+					(errmsg("removing stale two-phase state from shared memory for \"" XID_FMT "\"",
+							xid)));
 			PrepareRedoRemove(xid, true);
 		}
 		return NULL;
@@ -2250,14 +2264,14 @@ ProcessTwoPhaseBuffer(TransactionId xid,
 		if (fromdisk)
 		{
 			ereport(WARNING,
-					(errmsg("removing future two-phase state file for \"%u\"",
+					(errmsg("removing future two-phase state file for \"" XID_FMT "\"",
 							xid)));
 			RemoveTwoPhaseFile(xid, true);
 		}
 		else
 		{
 			ereport(WARNING,
-					(errmsg("removing future two-phase state from memory for \"%u\"",
+					(errmsg("removing future two-phase state from memory for \"" XID_FMT "\"",
 							xid)));
 			PrepareRedoRemove(xid, true);
 		}
@@ -2271,7 +2285,7 @@ ProcessTwoPhaseBuffer(TransactionId xid,
 		if (buf == NULL)
 		{
 			ereport(WARNING,
-					(errmsg("removing corrupt two-phase state file for \"%u\"",
+					(errmsg("removing corrupt two-phase state file for \"" XID_FMT "\"",
 							xid)));
 			RemoveTwoPhaseFile(xid, true);
 			return NULL;
@@ -2290,14 +2304,14 @@ ProcessTwoPhaseBuffer(TransactionId xid,
 		if (fromdisk)
 		{
 			ereport(WARNING,
-					(errmsg("removing corrupt two-phase state file for \"%u\"",
+					(errmsg("removing corrupt two-phase state file for \"" XID_FMT "\"",
 							xid)));
 			RemoveTwoPhaseFile(xid, true);
 		}
 		else
 		{
 			ereport(WARNING,
-					(errmsg("removing corrupt two-phase state from memory for \"%u\"",
+					(errmsg("removing corrupt two-phase state from memory for \"" XID_FMT "\"",
 							xid)));
 			PrepareRedoRemove(xid, true);
 		}
@@ -2565,6 +2579,7 @@ PrepareRedoAdd(char *buf, XLogRecPtr start_lsn, XLogRecPtr end_lsn)
 	const char		  *gid;
 	GlobalTransaction gxact;
 
+	Assert(LWLockHeldByMeInMode(TwoPhaseStateLock, LW_EXCLUSIVE));
 	Assert(RecoveryInProgress());
 
 	bufptr = buf + MAXALIGN(sizeof(TwoPhaseFileHeader));
@@ -2581,7 +2596,6 @@ PrepareRedoAdd(char *buf, XLogRecPtr start_lsn, XLogRecPtr end_lsn)
 	 * that it got added in the redo phase
 	 */
 
-	LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
 	/* Get a free gxact from the freelist */
 	if (TwoPhaseState->freeGXacts == NULL)
 		ereport(ERROR,
@@ -2607,27 +2621,27 @@ PrepareRedoAdd(char *buf, XLogRecPtr start_lsn, XLogRecPtr end_lsn)
 	Assert(TwoPhaseState->numPrepXacts < max_prepared_xacts);
 	TwoPhaseState->prepXacts[TwoPhaseState->numPrepXacts++] = gxact;
 
-	LWLockRelease(TwoPhaseStateLock);
-
-	elog(DEBUG2, "Adding 2PC data to shared memory %u", gxact->xid);
+	elog(DEBUG2, "added 2PC data in shared memory for transaction " XID_FMT, gxact->xid);
 }
 
 /*
  * PrepareRedoRemove
  *
- * Remove the corresponding gxact entry from TwoPhaseState. Also
- * remove the 2PC file if a prepared transaction was saved via
- * an earlier checkpoint.
+ * Remove the corresponding gxact entry from TwoPhaseState. Also remove
+ * the 2PC file if a prepared transaction was saved via an earlier checkpoint.
+ *
+ * Caller must hold TwoPhaseStateLock in exclusive mode, because TwoPhaseState
+ * is updated.
  */
 void
 PrepareRedoRemove(TransactionId xid, bool giveWarning)
 {
 	GlobalTransaction gxact = NULL;
 	int			i;
 
+	Assert(LWLockHeldByMeInMode(TwoPhaseStateLock, LW_EXCLUSIVE));
 	Assert(RecoveryInProgress());
 
-	LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
 	for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
 	{
 		gxact = TwoPhaseState->prepXacts[i];
@@ -2638,7 +2652,6 @@ PrepareRedoRemove(TransactionId xid, bool giveWarning)
 			break;
 		}
 	}
-	LWLockRelease(TwoPhaseStateLock);
 
 	/*
 	 * Just leave if there is nothing, this is expected during WAL replay.
@@ -2649,7 +2662,7 @@ PrepareRedoRemove(TransactionId xid, bool giveWarning)
 	/*
 	 * And now we can clean up any files we may have left.
 	 */
-	elog(DEBUG2, "Removing 2PC data from shared memory %u", xid);
+	elog(DEBUG2, "removing 2PC data for transaction " XID_FMT, xid);
 	if (gxact->ondisk)
 		RemoveTwoPhaseFile(xid, giveWarning);
 	RemoveGXact(gxact);