Phase 1 of fix for 'SMgrRelation hashtable corrupted' problem. This

is the minimum required fix.  I want to look next at taking advantage of
it by simplifying the message semantics in the shared inval message queue,
but that part can be held over for 8.1 if it turns out too ugly.

Author: tglsfdc

src/backend/access/transam/xlogutils.c

@@ -11,7 +11,7 @@
  * Portions Copyright (c) 1996-2005, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/backend/access/transam/xlogutils.c,v 1.35 2004/12/31 21:59:30 pgsql Exp $
+ * $PostgreSQL: pgsql/src/backend/access/transam/xlogutils.c,v 1.36 2005/01/10 20:02:19 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -125,8 +125,7 @@ _xl_remove_hash_entry(XLogRelDesc *rdesc)
 	if (hentry == NULL)
 		elog(PANIC, "_xl_remove_hash_entry: file was not found in cache");
 
-	if (rdesc->reldata.rd_smgr != NULL)
-		smgrclose(rdesc->reldata.rd_smgr);
+	RelationCloseSmgr(&(rdesc->reldata));
 
 	memset(rdesc, 0, sizeof(XLogRelDesc));
 	memset(tpgc, 0, sizeof(FormData_pg_class));
@@ -233,7 +232,8 @@ XLogOpenRelation(bool redo, RmgrId rmid, RelFileNode rnode)
 		hentry->rdesc = res;
 
 		res->reldata.rd_targblock = InvalidBlockNumber;
-		res->reldata.rd_smgr = smgropen(res->reldata.rd_node);
+		res->reldata.rd_smgr = NULL;
+		RelationOpenSmgr(&(res->reldata));
 
 		/*
 		 * Create the target file if it doesn't already exist.  This lets
@@ -278,7 +278,5 @@ XLogCloseRelation(RelFileNode rnode)
 
 	rdesc = hentry->rdesc;
 
-	if (rdesc->reldata.rd_smgr != NULL)
-		smgrclose(rdesc->reldata.rd_smgr);
-	rdesc->reldata.rd_smgr = NULL;
+	RelationCloseSmgr(&(rdesc->reldata));
 }

src/backend/catalog/heap.c

@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/catalog/heap.c,v 1.278 2004/12/31 21:59:38 pgsql Exp $
+ *	  $PostgreSQL: pgsql/src/backend/catalog/heap.c,v 1.279 2005/01/10 20:02:19 tgl Exp $
  *
  *
  * INTERFACE ROUTINES
@@ -322,7 +322,7 @@ heap_create(const char *relname,
 	if (create_storage)
 	{
 		Assert(rel->rd_smgr == NULL);
-		rel->rd_smgr = smgropen(rel->rd_node);
+		RelationOpenSmgr(rel);
 		smgrcreate(rel->rd_smgr, rel->rd_istemp, false);
 	}
 
@@ -1186,10 +1186,8 @@ heap_drop_with_catalog(Oid relid)
 	if (rel->rd_rel->relkind != RELKIND_VIEW &&
 		rel->rd_rel->relkind != RELKIND_COMPOSITE_TYPE)
 	{
-		if (rel->rd_smgr == NULL)
-			rel->rd_smgr = smgropen(rel->rd_node);
+		RelationOpenSmgr(rel);
 		smgrscheduleunlink(rel->rd_smgr, rel->rd_istemp);
-		rel->rd_smgr = NULL;
 	}
 
 	/*

src/backend/catalog/index.c

@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/catalog/index.c,v 1.243 2004/12/31 21:59:38 pgsql Exp $
+ *	  $PostgreSQL: pgsql/src/backend/catalog/index.c,v 1.244 2005/01/10 20:02:19 tgl Exp $
  *
  *
  * INTERFACE ROUTINES
@@ -780,11 +780,9 @@ index_drop(Oid indexId)
 	 */
 	FlushRelationBuffers(userIndexRelation, (BlockNumber) 0);
 
-	if (userIndexRelation->rd_smgr == NULL)
-		userIndexRelation->rd_smgr = smgropen(userIndexRelation->rd_node);
+	RelationOpenSmgr(userIndexRelation);
 	smgrscheduleunlink(userIndexRelation->rd_smgr,
 					   userIndexRelation->rd_istemp);
-	userIndexRelation->rd_smgr = NULL;
 
 	/*
 	 * Close and flush the index's relcache entry, to ensure relcache
@@ -1133,10 +1131,8 @@ setNewRelfilenode(Relation relation)
 	smgrclose(srel);
 
 	/* schedule unlinking old relfilenode */
-	if (relation->rd_smgr == NULL)
-		relation->rd_smgr = smgropen(relation->rd_node);
+	RelationOpenSmgr(relation);
 	smgrscheduleunlink(relation->rd_smgr, relation->rd_istemp);
-	relation->rd_smgr = NULL;
 
 	/* update the pg_class row */
 	rd_rel->relfilenode = newrelfilenode;

src/backend/commands/tablecmds.c

@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/commands/tablecmds.c,v 1.141 2004/12/31 21:59:41 pgsql Exp $
+ *	  $PostgreSQL: pgsql/src/backend/commands/tablecmds.c,v 1.142 2005/01/10 20:02:20 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -5521,10 +5521,8 @@ ATExecSetTableSpace(Oid tableOid, Oid newTableSpace)
 	copy_relation_data(rel, dstrel);
 
 	/* schedule unlinking old physical file */
-	if (rel->rd_smgr == NULL)
-		rel->rd_smgr = smgropen(rel->rd_node);
+	RelationOpenSmgr(rel);
 	smgrscheduleunlink(rel->rd_smgr, rel->rd_istemp);
-	rel->rd_smgr = NULL;
 
 	/*
 	 * Now drop smgr references.  The source was already dropped by

src/backend/postmaster/bgwriter.c

@@ -37,7 +37,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/postmaster/bgwriter.c,v 1.12 2004/12/31 22:00:40 pgsql Exp $
+ *	  $PostgreSQL: pgsql/src/backend/postmaster/bgwriter.c,v 1.13 2005/01/10 20:02:20 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -349,9 +349,7 @@ BackgroundWriterMain(void)
 			/*
 			 * After any checkpoint, close all smgr files.	This is so we
 			 * won't hang onto smgr references to deleted files
-			 * indefinitely. (It is safe to do this because this process
-			 * does not have a relcache, and so no dangling references
-			 * could remain.)
+			 * indefinitely.
 			 */
 			smgrcloseall();
 

src/backend/rewrite/rewriteDefine.c

@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/rewrite/rewriteDefine.c,v 1.99 2004/12/31 22:00:45 pgsql Exp $
+ *	  $PostgreSQL: pgsql/src/backend/rewrite/rewriteDefine.c,v 1.100 2005/01/10 20:02:21 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -484,10 +484,8 @@ DefineQueryRewrite(RuleStmt *stmt)
 	 */
 	if (RelisBecomingView)
 	{
-		if (event_relation->rd_smgr == NULL)
-			event_relation->rd_smgr = smgropen(event_relation->rd_node);
+		RelationOpenSmgr(event_relation);
 		smgrscheduleunlink(event_relation->rd_smgr, event_relation->rd_istemp);
-		event_relation->rd_smgr = NULL;
 	}
 
 	/* Close rel, but keep lock till commit... */

src/backend/storage/buffer/bufmgr.c

@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/storage/buffer/bufmgr.c,v 1.184 2005/01/03 18:49:41 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/storage/buffer/bufmgr.c,v 1.185 2005/01/10 20:02:21 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -131,8 +131,7 @@ ReadBufferInternal(Relation reln, BlockNumber blockNum,
 	isLocalBuf = reln->rd_istemp;
 
 	/* Open it at the smgr level if not already done */
-	if (reln->rd_smgr == NULL)
-		reln->rd_smgr = smgropen(reln->rd_node);
+	RelationOpenSmgr(reln);
 
 	/* Substitute proper block number if caller asked for P_NEW */
 	if (isExtend)
@@ -1130,8 +1129,7 @@ BlockNumber
 RelationGetNumberOfBlocks(Relation relation)
 {
 	/* Open it at the smgr level if not already done */
-	if (relation->rd_smgr == NULL)
-		relation->rd_smgr = smgropen(relation->rd_node);
+	RelationOpenSmgr(relation);
 
 	return smgrnblocks(relation->rd_smgr);
 }
@@ -1147,8 +1145,7 @@ void
 RelationTruncate(Relation rel, BlockNumber nblocks)
 {
 	/* Open it at the smgr level if not already done */
-	if (rel->rd_smgr == NULL)
-		rel->rd_smgr = smgropen(rel->rd_node);
+	RelationOpenSmgr(rel);
 
 	/* Make sure rd_targblock isn't pointing somewhere past end */
 	rel->rd_targblock = InvalidBlockNumber;
@@ -1445,8 +1442,7 @@ FlushRelationBuffers(Relation rel, BlockNumber firstDelBlock)
 	BufferDesc *bufHdr;
 
 	/* Open rel at the smgr level if not already done */
-	if (rel->rd_smgr == NULL)
-		rel->rd_smgr = smgropen(rel->rd_node);
+	RelationOpenSmgr(rel);
 
 	if (rel->rd_istemp)
 	{

src/backend/storage/buffer/localbuf.c

@@ -9,7 +9,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/storage/buffer/localbuf.c,v 1.61 2004/12/31 22:00:49 pgsql Exp $
+ *	  $PostgreSQL: pgsql/src/backend/storage/buffer/localbuf.c,v 1.62 2005/01/10 20:02:21 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -108,13 +108,13 @@ LocalBufferAlloc(Relation reln, BlockNumber blockNum, bool *foundPtr)
 	 */
 	if (bufHdr->flags & BM_DIRTY || bufHdr->cntxDirty)
 	{
-		SMgrRelation reln;
+		SMgrRelation oreln;
 
 		/* Find smgr relation for buffer */
-		reln = smgropen(bufHdr->tag.rnode);
+		oreln = smgropen(bufHdr->tag.rnode);
 
 		/* And write... */
-		smgrwrite(reln,
+		smgrwrite(oreln,
 				  bufHdr->tag.blockNum,
 				  (char *) MAKE_PTR(bufHdr->data),
 				  true);

src/backend/storage/smgr/smgr.c

@@ -11,7 +11,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/storage/smgr/smgr.c,v 1.84 2004/12/31 22:01:13 pgsql Exp $
+ *	  $PostgreSQL: pgsql/src/backend/storage/smgr/smgr.c,v 1.85 2005/01/10 20:02:22 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -216,6 +216,7 @@ smgropen(RelFileNode rnode)
 	if (!found)
 	{
 		/* hash_search already filled in the lookup key */
+		reln->smgr_owner = NULL;
 		reln->smgr_which = 0;	/* we only have md.c at present */
 		reln->md_fd = NULL;		/* mark it not open */
 	}
@@ -224,15 +225,36 @@ smgropen(RelFileNode rnode)
 }
 
 /*
- *	smgrclose() -- Close and delete an SMgrRelation object.
+ * smgrsetowner() -- Establish a long-lived reference to an SMgrRelation object
  *
- * It is the caller's responsibility not to leave any dangling references
- * to the object.  (Pointers should be cleared after successful return;
- * on the off chance of failure, the SMgrRelation object will still exist.)
+ * There can be only one owner at a time; this is sufficient since currently
+ * the only such owners exist in the relcache.
+ */
+void
+smgrsetowner(SMgrRelation *owner, SMgrRelation reln)
+{
+	/*
+	 * First, unhook any old owner.  (Normally there shouldn't be any, but
+	 * it seems possible that this can happen during swap_relation_files()
+	 * depending on the order of processing.  It's ok to close the old
+	 * relcache entry early in that case.)
+	 */
+	if (reln->smgr_owner)
+		*(reln->smgr_owner) = NULL;
+
+	/* Now establish the ownership relationship. */
+	reln->smgr_owner = owner;
+	*owner = reln;
+}
+
+/*
+ *	smgrclose() -- Close and delete an SMgrRelation object.
  */
 void
 smgrclose(SMgrRelation reln)
 {
+	SMgrRelation *owner;
+
 	if (!(*(smgrsw[reln->smgr_which].smgr_close)) (reln))
 		ereport(ERROR,
 				(errcode_for_file_access(),
@@ -241,16 +263,24 @@ smgrclose(SMgrRelation reln)
 						reln->smgr_rnode.dbNode,
 						reln->smgr_rnode.relNode)));
 
+	owner = reln->smgr_owner;
+
 	if (hash_search(SMgrRelationHash,
 					(void *) &(reln->smgr_rnode),
 					HASH_REMOVE, NULL) == NULL)
 		elog(ERROR, "SMgrRelation hashtable corrupted");
+
+	/*
+	 * Unhook the owner pointer, if any.  We do this last since in the
+	 * remote possibility of failure above, the SMgrRelation object will still
+	 * exist.
+	 */
+	if (owner)
+		*owner = NULL;
 }
 
 /*
  *	smgrcloseall() -- Close all existing SMgrRelation objects.
- *
- * It is the caller's responsibility not to leave any dangling references.
  */
 void
 smgrcloseall(void)
@@ -275,8 +305,6 @@ smgrcloseall(void)
  * This has the same effects as smgrclose(smgropen(rnode)), but it avoids
  * uselessly creating a hashtable entry only to drop it again when no
  * such entry exists already.
- *
- * It is the caller's responsibility not to leave any dangling references.
  */
 void
 smgrclosenode(RelFileNode rnode)

src/backend/utils/cache/inval.c

@@ -80,7 +80,7 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/utils/cache/inval.c,v 1.68 2004/12/31 22:01:25 pgsql Exp $
+ *	  $PostgreSQL: pgsql/src/backend/utils/cache/inval.c,v 1.69 2005/01/10 20:02:23 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -414,17 +414,9 @@ LocalExecuteInvalidationMessage(SharedInvalidationMessage *msg)
 	}
 	else if (msg->id == SHAREDINVALRELCACHE_ID)
 	{
-		/*
-		 * If the message includes a valid relfilenode, we must ensure
-		 * that smgr cache entry gets zapped.  The relcache will handle
-		 * this if called, otherwise we must do it directly.
-		 */
 		if (msg->rc.dbId == MyDatabaseId || msg->rc.dbId == InvalidOid)
 		{
-			if (OidIsValid(msg->rc.physId.relNode))
-				RelationCacheInvalidateEntry(msg->rc.relId, &msg->rc.physId);
-			else
-				RelationCacheInvalidateEntry(msg->rc.relId, NULL);
+			RelationCacheInvalidateEntry(msg->rc.relId);
 
 			for (i = 0; i < cache_callback_count; i++)
 			{
@@ -434,12 +426,17 @@ LocalExecuteInvalidationMessage(SharedInvalidationMessage *msg)
 					(*ccitem->function) (ccitem->arg, msg->rc.relId);
 			}
 		}
-		else
-		{
-			/* might have smgr entry even if not in our database */
-			if (OidIsValid(msg->rc.physId.relNode))
-				smgrclosenode(msg->rc.physId);
-		}
+		/*
+		 * If the message includes a valid relfilenode, we must ensure
+		 * the smgr cache entry gets zapped.  This might not have happened
+		 * above since the relcache entry might not have existed or might
+		 * have been associated with a different relfilenode.
+		 *
+		 * XXX there is no real good reason for rnode inval to be in the
+		 * same message at all.  FIXME in 8.1.
+		 */
+		if (OidIsValid(msg->rc.physId.relNode))
+			smgrclosenode(msg->rc.physId);
 	}
 	else
 		elog(FATAL, "unrecognized SI message id: %d", msg->id);