Track the current XID wrap limit (or more accurately, the oldest unfrozen

XID) in checkpoint records.  This eliminates the need to recompute the value
from scratch during database startup, which is one of the two remaining
reasons for the flatfile code to exist.  It should also simplify life for
hot-standby operation.

To avoid bloating the checkpoint records unreasonably, I switched from
tracking the oldest database by name to tracking it by OID.  This turns
out to save cycles in general (everywhere but the warning-generating
paths, which we hardly care about) and also helps us deal with the case
that the oldest database got dropped instead of being vacuumed.  The prior
coding might go for a long time without updating the wrap limit in that case,
which is bad because it might result in a lot of useless autovacuum activity.

1 files changed, 125 insertions, 31 deletions

diff --git a/src/backend/access/transam/varsup.c b/src/backend/access/transam/varsup.c
index 029b2f2deb7..e44cf0d4505 100644
--- a/src/backend/access/transam/varsup.c
+++ b/src/backend/access/transam/varsup.c
@@ -6,7 +6,7 @@
  * Copyright (c) 2000-2009, PostgreSQL Global Development Group
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/access/transam/varsup.c,v 1.84 2009/04/23 00:23:45 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/access/transam/varsup.c,v 1.85 2009/08/31 02:23:21 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -16,11 +16,13 @@
 #include "access/clog.h"
 #include "access/subtrans.h"
 #include "access/transam.h"
+#include "commands/dbcommands.h"
 #include "miscadmin.h"
 #include "postmaster/autovacuum.h"
 #include "storage/pmsignal.h"
 #include "storage/proc.h"
 #include "utils/builtins.h"
+#include "utils/syscache.h"
 
 
 /* Number of OIDs to prefetch (preallocate) per XLOG write */
@@ -31,9 +33,14 @@ VariableCache ShmemVariableCache = NULL;
 
 
 /*
- * Allocate the next XID for my new transaction or subtransaction.
+ * Allocate the next XID for a new transaction or subtransaction.
  *
  * The new XID is also stored into MyProc before returning.
+ *
+ * Note: when this is called, we are actually already inside a valid
+ * transaction, since XIDs are now not allocated until the transaction
+ * does something.  So it is safe to do a database lookup if we want to
+ * issue a warning about XID wrap.
  */
 TransactionId
 GetNewTransactionId(bool isSubXact)
@@ -73,6 +80,20 @@ GetNewTransactionId(bool isSubXact)
 		TransactionIdIsValid(ShmemVariableCache->xidVacLimit))
 	{
 		/*
+		 * For safety's sake, we release XidGenLock while sending signals,
+		 * warnings, etc.  This is not so much because we care about
+		 * preserving concurrency in this situation, as to avoid any
+		 * possibility of deadlock while doing get_database_name().
+		 * First, copy all the shared values we'll need in this path.
+		 */
+		TransactionId xidWarnLimit = ShmemVariableCache->xidWarnLimit;
+		TransactionId xidStopLimit = ShmemVariableCache->xidStopLimit;
+		TransactionId xidWrapLimit = ShmemVariableCache->xidWrapLimit;
+		Oid		oldest_datoid = ShmemVariableCache->oldestXidDB;
+
+		LWLockRelease(XidGenLock);
+
+		/*
 		 * To avoid swamping the postmaster with signals, we issue the autovac
 		 * request only once per 64K transaction starts.  This still gives
 		 * plenty of chances before we get into real trouble.
@@ -81,22 +102,50 @@ GetNewTransactionId(bool isSubXact)
 			SendPostmasterSignal(PMSIGNAL_START_AUTOVAC_LAUNCHER);
 
 		if (IsUnderPostmaster &&
-		 TransactionIdFollowsOrEquals(xid, ShmemVariableCache->xidStopLimit))
-			ereport(ERROR,
-					(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
-					 errmsg("database is not accepting commands to avoid wraparound data loss in database \"%s\"",
-							NameStr(ShmemVariableCache->limit_datname)),
-					 errhint("Stop the postmaster and use a standalone backend to vacuum database \"%s\".\n"
-							 "You might also need to commit or roll back old prepared transactions.",
-							 NameStr(ShmemVariableCache->limit_datname))));
-		else if (TransactionIdFollowsOrEquals(xid, ShmemVariableCache->xidWarnLimit))
-			ereport(WARNING,
-			(errmsg("database \"%s\" must be vacuumed within %u transactions",
-					NameStr(ShmemVariableCache->limit_datname),
-					ShmemVariableCache->xidWrapLimit - xid),
-			 errhint("To avoid a database shutdown, execute a database-wide VACUUM in \"%s\".\n"
-					 "You might also need to commit or roll back old prepared transactions.",
-					 NameStr(ShmemVariableCache->limit_datname))));
+			TransactionIdFollowsOrEquals(xid, xidStopLimit))
+		{
+			char   *oldest_datname = get_database_name(oldest_datoid);
+
+			/* complain even if that DB has disappeared */
+			if (oldest_datname)
+				ereport(ERROR,
+						(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
+						 errmsg("database is not accepting commands to avoid wraparound data loss in database \"%s\"",
+								oldest_datname),
+						 errhint("Stop the postmaster and use a standalone backend to vacuum that database.\n"
+								 "You might also need to commit or roll back old prepared transactions.")));
+			else
+				ereport(ERROR,
+						(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
+						 errmsg("database is not accepting commands to avoid wraparound data loss in database with OID %u",
+								oldest_datoid),
+						 errhint("Stop the postmaster and use a standalone backend to vacuum that database.\n"
+								 "You might also need to commit or roll back old prepared transactions.")));
+		}
+		else if (TransactionIdFollowsOrEquals(xid, xidWarnLimit))
+		{
+			char   *oldest_datname = get_database_name(oldest_datoid);
+
+			/* complain even if that DB has disappeared */
+			if (oldest_datname)
+				ereport(WARNING,
+						(errmsg("database \"%s\" must be vacuumed within %u transactions",
+								oldest_datname,
+								xidWrapLimit - xid),
+						 errhint("To avoid a database shutdown, execute a database-wide VACUUM in that database.\n"
+								 "You might also need to commit or roll back old prepared transactions.")));
+			else
+				ereport(WARNING,
+						(errmsg("database with OID %u must be vacuumed within %u transactions",
+								oldest_datoid,
+								xidWrapLimit - xid),
+						 errhint("To avoid a database shutdown, execute a database-wide VACUUM in that database.\n"
+								 "You might also need to commit or roll back old prepared transactions.")));
+		}
+
+		/* Re-acquire lock and start over */
+		LWLockAcquire(XidGenLock, LW_EXCLUSIVE);
+		xid = ShmemVariableCache->nextXid;
 	}
 
 	/*
@@ -199,11 +248,10 @@ ReadNewTransactionId(void)
 /*
  * Determine the last safe XID to allocate given the currently oldest
  * datfrozenxid (ie, the oldest XID that might exist in any database
- * of our cluster).
+ * of our cluster), and the OID of the (or a) database with that value.
  */
 void
-SetTransactionIdLimit(TransactionId oldest_datfrozenxid,
-					  Name oldest_datname)
+SetTransactionIdLimit(TransactionId oldest_datfrozenxid, Oid oldest_datoid)
 {
 	TransactionId xidVacLimit;
 	TransactionId xidWarnLimit;
@@ -275,14 +323,14 @@ SetTransactionIdLimit(TransactionId oldest_datfrozenxid,
 	ShmemVariableCache->xidWarnLimit = xidWarnLimit;
 	ShmemVariableCache->xidStopLimit = xidStopLimit;
 	ShmemVariableCache->xidWrapLimit = xidWrapLimit;
-	namecpy(&ShmemVariableCache->limit_datname, oldest_datname);
+	ShmemVariableCache->oldestXidDB = oldest_datoid;
 	curXid = ShmemVariableCache->nextXid;
 	LWLockRelease(XidGenLock);
 
 	/* Log the info */
 	ereport(DEBUG1,
-	   (errmsg("transaction ID wrap limit is %u, limited by database \"%s\"",
-			   xidWrapLimit, NameStr(*oldest_datname))));
+	   (errmsg("transaction ID wrap limit is %u, limited by database with OID %u",
+			   xidWrapLimit, oldest_datoid)));
 
 	/*
 	 * If past the autovacuum force point, immediately signal an autovac
@@ -297,13 +345,59 @@ SetTransactionIdLimit(TransactionId oldest_datfrozenxid,
 
 	/* Give an immediate warning if past the wrap warn point */
 	if (TransactionIdFollowsOrEquals(curXid, xidWarnLimit))
-		ereport(WARNING,
-		   (errmsg("database \"%s\" must be vacuumed within %u transactions",
-				   NameStr(*oldest_datname),
-				   xidWrapLimit - curXid),
-			errhint("To avoid a database shutdown, execute a database-wide VACUUM in \"%s\".\n"
-					"You might also need to commit or roll back old prepared transactions.",
-					NameStr(*oldest_datname))));
+	{
+		char   *oldest_datname = get_database_name(oldest_datoid);
+
+		/*
+		 * Note: it's possible that get_database_name fails and returns NULL,
+		 * for example because the database just got dropped.  We'll still
+		 * warn, even though the warning might now be unnecessary.
+		 */
+		if (oldest_datname)
+			ereport(WARNING,
+					(errmsg("database \"%s\" must be vacuumed within %u transactions",
+							oldest_datname,
+							xidWrapLimit - curXid),
+					 errhint("To avoid a database shutdown, execute a database-wide VACUUM in that database.\n"
+							 "You might also need to commit or roll back old prepared transactions.")));
+		else
+			ereport(WARNING,
+					(errmsg("database with OID %u must be vacuumed within %u transactions",
+							oldest_datoid,
+							xidWrapLimit - curXid),
+					 errhint("To avoid a database shutdown, execute a database-wide VACUUM in that database.\n"
+							 "You might also need to commit or roll back old prepared transactions.")));
+	}
+}
+
+
+/*
+ * TransactionIdLimitIsValid -- is the shared XID wrap-limit data sane?
+ *
+ * We primarily check whether oldestXidDB is valid.  The cases we have in
+ * mind are that that database was dropped, or the field was reset to zero
+ * by pg_resetxlog.  In either case we should force recalculation of the
+ * wrap limit.  In future we might add some more sanity checks here.
+ */
+bool
+TransactionIdLimitIsValid(void)
+{
+	TransactionId oldestXid;
+	Oid			oldestXidDB;
+
+	/* Locking is probably not really necessary, but let's be careful */
+	LWLockAcquire(XidGenLock, LW_SHARED);
+	oldestXid = ShmemVariableCache->oldestXid;
+	oldestXidDB = ShmemVariableCache->oldestXidDB;
+	LWLockRelease(XidGenLock);
+
+	if (!TransactionIdIsNormal(oldestXid))
+		return false;			/* shouldn't happen, but just in case */
+	if (!SearchSysCacheExists(DATABASEOID,
+							  ObjectIdGetDatum(oldestXidDB),
+							  0, 0, 0))
+		return false;			/* could happen, per comment above */
+	return true;
 }