Rethink the locking mechanisms used for CREATE/DROP/RENAME DATABASE.

The former approach used ExclusiveLock on pg_database, which being a
cluster-wide lock meant only one of these operations could proceed at
a time; worse, it also blocked all incoming connections in ReverifyMyDatabase.
Now that we have LockSharedObject(), we can use locks of different types
applied to databases considered as objects.  This allows much more
flexible management of the interlocking: two CREATE DATABASEs need not
block each other, and need not block connections except to the template
database being used.  Similarly DROP DATABASE doesn't block unrelated
operations.  The locking used in flatfiles.c is also much narrower in
scope than before.  Per recent proposal.

1 files changed, 110 insertions, 119 deletions

diff --git a/src/backend/utils/init/postinit.c b/src/backend/utils/init/postinit.c
index 4ddc7f712af..0be0fd85438 100644
--- a/src/backend/utils/init/postinit.c
+++ b/src/backend/utils/init/postinit.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/utils/init/postinit.c,v 1.165 2006/05/03 22:45:26 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/utils/init/postinit.c,v 1.166 2006/05/04 16:07:29 tgl Exp $
  *
  *
  *-------------------------------------------------------------------------
@@ -16,14 +16,10 @@
 #include "postgres.h"
 
 #include <fcntl.h>
-#include <sys/file.h>
-#include <math.h>
 #include <unistd.h>
 
-#include "access/genam.h"
 #include "access/heapam.h"
 #include "catalog/catalog.h"
-#include "catalog/indexing.h"
 #include "catalog/namespace.h"
 #include "catalog/pg_authid.h"
 #include "catalog/pg_database.h"
@@ -42,7 +38,6 @@
 #include "storage/smgr.h"
 #include "utils/acl.h"
 #include "utils/flatfiles.h"
-#include "utils/fmgroids.h"
 #include "utils/guc.h"
 #include "utils/portal.h"
 #include "utils/relcache.h"
@@ -51,7 +46,7 @@
 
 
 static bool FindMyDatabase(const char *name, Oid *db_id, Oid *db_tablespace);
-static void ReverifyMyDatabase(const char *name, bool am_superuser);
+static void CheckMyDatabase(const char *name, bool am_superuser);
 static void InitCommunication(void);
 static void ShutdownPostgres(int code, Datum arg);
 static bool ThereIsAtLeastOneRole(void);
@@ -72,7 +67,7 @@ static bool ThereIsAtLeastOneRole(void);
  * file" copy of pg_database that is helpfully maintained by flatfiles.c.
  * This is subject to various race conditions, so after we have the
  * transaction infrastructure started, we have to recheck the information;
- * see ReverifyMyDatabase.
+ * see InitPostgres.
  */
 static bool
 FindMyDatabase(const char *name, Oid *db_id, Oid *db_tablespace)
@@ -108,76 +103,35 @@ FindMyDatabase(const char *name, Oid *db_id, Oid *db_tablespace)
 }
 
 /*
- * ReverifyMyDatabase -- recheck info obtained by FindMyDatabase
- *
- * Since FindMyDatabase cannot lock pg_database, the information it read
- * could be stale; for example we might have attached to a database that's in
- * process of being destroyed by dropdb().	This routine is called after
- * we have all the locking and other infrastructure running --- now we can
- * check that we are really attached to a valid database.
- *
- * In reality, if dropdb() is running in parallel with our startup,
- * it's pretty likely that we will have failed before now, due to being
- * unable to read some of the system tables within the doomed database.
- * This routine just exists to make *sure* we have not started up in an
- * invalid database.  If we quit now, we should have managed to avoid
- * creating any serious problems.
- *
- * This is also a handy place to fetch the database encoding info out
- * of pg_database.
- *
- * To avoid having to read pg_database more times than necessary
- * during session startup, this place is also fitting to check CONNECT
- * privilege and set up any database-specific configuration variables.
+ * CheckMyDatabase -- fetch information from the pg_database entry for our DB
  */
 static void
-ReverifyMyDatabase(const char *name, bool am_superuser)
+CheckMyDatabase(const char *name, bool am_superuser)
 {
-	Relation	pgdbrel;
-	SysScanDesc pgdbscan;
-	ScanKeyData key;
 	HeapTuple	tup;
 	Form_pg_database dbform;
 
-	/*
-	 * Because we grab RowShareLock here, we can be sure that dropdb() is not
-	 * running in parallel with us (any more).
-	 */
-	pgdbrel = heap_open(DatabaseRelationId, RowShareLock);
-
-	ScanKeyInit(&key,
-				Anum_pg_database_datname,
-				BTEqualStrategyNumber, F_NAMEEQ,
-				NameGetDatum(name));
-
-	pgdbscan = systable_beginscan(pgdbrel, DatabaseNameIndexId, true,
-								  SnapshotNow, 1, &key);
-
-	tup = systable_getnext(pgdbscan);
-	if (!HeapTupleIsValid(tup) ||
-		HeapTupleGetOid(tup) != MyDatabaseId)
-	{
-		/* OOPS */
-		heap_close(pgdbrel, RowShareLock);
+	/* Fetch our real pg_database row */
+	tup = SearchSysCache(DATABASEOID,
+						 ObjectIdGetDatum(MyDatabaseId),
+						 0, 0, 0);
+	if (!HeapTupleIsValid(tup))
+		elog(ERROR, "cache lookup failed for database %u", MyDatabaseId);
+	dbform = (Form_pg_database) GETSTRUCT(tup);
 
-		/*
-		 * The only real problem I could have created is to load dirty buffers
-		 * for the dead database into shared buffer cache; if I did, some
-		 * other backend will eventually try to write them and die in
-		 * mdblindwrt.	Flush any such pages to forestall trouble.
-		 */
-		DropDatabaseBuffers(MyDatabaseId);
-		/* Now I can commit hara-kiri with a clear conscience... */
+	/* This recheck is strictly paranoia */
+	if (strcmp(name, NameStr(dbform->datname)) != 0)
 		ereport(FATAL,
 				(errcode(ERRCODE_UNDEFINED_DATABASE),
-		  errmsg("database \"%s\", OID %u, has disappeared from pg_database",
-				 name, MyDatabaseId)));
-	}
-
-	dbform = (Form_pg_database) GETSTRUCT(tup);
+				 errmsg("database \"%s\" has disappeared from pg_database",
+						name),
+				 errdetail("Database OID %u now seems to belong to \"%s\".",
+						   MyDatabaseId, NameStr(dbform->datname))));
 
 	/*
-	 * These next checks are not enforced when in standalone mode, so that
+	 * Check permissions to connect to the database.
+	 *
+	 * These checks are not enforced when in standalone mode, so that
 	 * there is a way to recover from disabling all access to all databases,
 	 * for example "UPDATE pg_database SET datallowconn = false;".
 	 *
@@ -246,8 +200,8 @@ ReverifyMyDatabase(const char *name, bool am_superuser)
 		Datum		datum;
 		bool		isnull;
 
-		datum = heap_getattr(tup, Anum_pg_database_datconfig,
-							 RelationGetDescr(pgdbrel), &isnull);
+		datum = SysCacheGetAttr(DATABASEOID, tup, Anum_pg_database_datconfig,
+								&isnull);
 		if (!isnull)
 		{
 			ArrayType  *a = DatumGetArrayTypeP(datum);
@@ -256,8 +210,7 @@ ReverifyMyDatabase(const char *name, bool am_superuser)
 		}
 	}
 
-	systable_endscan(pgdbscan);
-	heap_close(pgdbrel, RowShareLock);
+	ReleaseSysCache(tup);
 }
 
 
@@ -337,9 +290,11 @@ InitPostgres(const char *dbname, const char *username)
 	bool		bootstrap = IsBootstrapProcessingMode();
 	bool		autovacuum = IsAutoVacuumProcess();
 	bool		am_superuser;
+	char	   *fullpath;
 
 	/*
-	 * Set up the global variables holding database id and path.
+	 * Set up the global variables holding database id and path.  But note
+	 * we won't actually try to touch the database just yet.
 	 *
 	 * We take a shortcut in the bootstrap case, otherwise we have to look up
 	 * the db name in pg_database.
@@ -348,55 +303,24 @@ InitPostgres(const char *dbname, const char *username)
 	{
 		MyDatabaseId = TemplateDbOid;
 		MyDatabaseTableSpace = DEFAULTTABLESPACE_OID;
-		SetDatabasePath(GetDatabasePath(MyDatabaseId, MyDatabaseTableSpace));
 	}
 	else
 	{
-		char	   *fullpath;
-
-		/*
-		 * Formerly we validated DataDir here, but now that's done earlier.
-		 */
-
 		/*
 		 * Find oid and tablespace of the database we're about to open. Since
 		 * we're not yet up and running we have to use the hackish
-		 * FindMyDatabase.
+		 * FindMyDatabase, which looks in the flat-file copy of pg_database.
 		 */
 		if (!FindMyDatabase(dbname, &MyDatabaseId, &MyDatabaseTableSpace))
 			ereport(FATAL,
 					(errcode(ERRCODE_UNDEFINED_DATABASE),
 					 errmsg("database \"%s\" does not exist",
 							dbname)));
-
-		fullpath = GetDatabasePath(MyDatabaseId, MyDatabaseTableSpace);
-
-		/* Verify the database path */
-
-		if (access(fullpath, F_OK) == -1)
-		{
-			if (errno == ENOENT)
-				ereport(FATAL,
-						(errcode(ERRCODE_UNDEFINED_DATABASE),
-						 errmsg("database \"%s\" does not exist",
-								dbname),
-					errdetail("The database subdirectory \"%s\" is missing.",
-							  fullpath)));
-			else
-				ereport(FATAL,
-						(errcode_for_file_access(),
-						 errmsg("could not access directory \"%s\": %m",
-								fullpath)));
-		}
-
-		ValidatePgVersion(fullpath);
-
-		SetDatabasePath(fullpath);
 	}
 
-	/*
-	 * Code after this point assumes we are in the proper directory!
-	 */
+	fullpath = GetDatabasePath(MyDatabaseId, MyDatabaseTableSpace);
+
+	SetDatabasePath(fullpath);
 
 	/*
 	 * Finish filling in the PGPROC struct, and add it to the ProcArray.
@@ -459,11 +383,79 @@ InitPostgres(const char *dbname, const char *username)
 	 */
 	on_shmem_exit(ShutdownPostgres, 0);
 
-	/* start a new transaction here before access to db */
+	/*
+	 * Start a new transaction here before first access to db
+	 */
 	if (!bootstrap)
 		StartTransactionCommand();
 
 	/*
+	 * Now that we have a transaction, we can take locks.  Take a writer's
+	 * lock on the database we are trying to connect to.  If there is
+	 * a concurrently running DROP DATABASE on that database, this will
+	 * block us until it finishes (and has updated the flat file copy
+	 * of pg_database).
+	 *
+	 * Note that the lock is not held long, only until the end of this
+	 * startup transaction.  This is OK since we are already advertising
+	 * our use of the database in the PGPROC array; anyone trying a DROP
+	 * DATABASE after this point will see us there.
+	 *
+	 * Note: use of RowExclusiveLock here is reasonable because we envision
+	 * our session as being a concurrent writer of the database.  If we had
+	 * a way of declaring a session as being guaranteed-read-only, we could
+	 * use AccessShareLock for such sessions and thereby not conflict against
+	 * CREATE DATABASE.
+	 */
+	if (!bootstrap)
+		LockSharedObject(DatabaseRelationId, MyDatabaseId, 0,
+						 RowExclusiveLock);
+
+	/*
+	 * Recheck the flat file copy of pg_database to make sure the target
+	 * database hasn't gone away.  If there was a concurrent DROP DATABASE,
+	 * this ensures we will die cleanly without creating a mess.
+	 */
+	if (!bootstrap)
+	{
+		Oid		dbid2;
+		Oid		tsid2;
+
+		if (!FindMyDatabase(dbname, &dbid2, &tsid2) ||
+			dbid2 != MyDatabaseId || tsid2 != MyDatabaseTableSpace)
+			ereport(FATAL,
+					(errcode(ERRCODE_UNDEFINED_DATABASE),
+					 errmsg("database \"%s\" does not exist",
+							dbname),
+				errdetail("It seems to have just been dropped or renamed.")));
+	}
+
+	/*
+	 * Now we should be able to access the database directory safely.
+	 * Verify it's there and looks reasonable.
+	 */
+	if (!bootstrap)
+	{
+		if (access(fullpath, F_OK) == -1)
+		{
+			if (errno == ENOENT)
+				ereport(FATAL,
+						(errcode(ERRCODE_UNDEFINED_DATABASE),
+						 errmsg("database \"%s\" does not exist",
+								dbname),
+						 errdetail("The database subdirectory \"%s\" is missing.",
+								   fullpath)));
+			else
+				ereport(FATAL,
+						(errcode_for_file_access(),
+						 errmsg("could not access directory \"%s\": %m",
+								fullpath)));
+		}
+
+		ValidatePgVersion(fullpath);
+	}
+
+	/*
 	 * It's now possible to do real access to the system catalogs.
 	 *
 	 * Replace faked-up relcache entries with correct info.
@@ -499,22 +491,21 @@ InitPostgres(const char *dbname, const char *username)
 		am_superuser = superuser();
 	}
 
-	/* set up ACL framework (so ReverifyMyDatabase can check permissions) */
+	/* set up ACL framework (so CheckMyDatabase can check permissions) */
 	initialize_acl();
 
 	/*
-	 * Unless we are bootstrapping, double-check that InitMyDatabaseInfo() got
-	 * a correct result.  We can't do this until all the database-access
-	 * infrastructure is up.  (Also, it wants to know if the user is a
-	 * superuser, so the above stuff has to happen first.)
+	 * Read the real pg_database row for our database, check permissions
+	 * and set up database-specific GUC settings.  We can't do this until all
+	 * the database-access infrastructure is up.  (Also, it wants to know if
+	 * the user is a superuser, so the above stuff has to happen first.)
 	 */
 	if (!bootstrap)
-		ReverifyMyDatabase(dbname, am_superuser);
+		CheckMyDatabase(dbname, am_superuser);
 
 	/*
 	 * Final phase of relation cache startup: write a new cache file if
-	 * necessary.  This is done after ReverifyMyDatabase to avoid writing a
-	 * cache file into a dead database.
+	 * necessary.  (XXX this could be folded back into Phase2)
 	 */
 	RelationCacheInitializePhase3();
 
@@ -530,7 +521,7 @@ InitPostgres(const char *dbname, const char *username)
 
 	/*
 	 * Initialize various default states that can't be set up until we've
-	 * selected the active user and done ReverifyMyDatabase.
+	 * selected the active user and gotten the right GUC settings.
 	 */
 
 	/* set default namespace search path */
@@ -587,13 +578,13 @@ ThereIsAtLeastOneRole(void)
 	HeapScanDesc scan;
 	bool		result;
 
-	pg_authid_rel = heap_open(AuthIdRelationId, AccessExclusiveLock);
+	pg_authid_rel = heap_open(AuthIdRelationId, AccessShareLock);
 
 	scan = heap_beginscan(pg_authid_rel, SnapshotNow, 0, NULL);
 	result = (heap_getnext(scan, ForwardScanDirection) != NULL);
 
 	heap_endscan(scan);
-	heap_close(pg_authid_rel, AccessExclusiveLock);
+	heap_close(pg_authid_rel, AccessShareLock);
 
 	return result;
 }