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.

3 files changed, 46 insertions, 22 deletions

diff --git a/src/backend/storage/buffer/bufmgr.c b/src/backend/storage/buffer/bufmgr.c
index 5ec006ef58..457d23b0e0 100644
--- a/src/backend/storage/buffer/bufmgr.c
+++ b/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)
 	{
diff --git a/src/backend/storage/buffer/localbuf.c b/src/backend/storage/buffer/localbuf.c
index f433673655..d873742702 100644
--- a/src/backend/storage/buffer/localbuf.c
+++ b/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);
diff --git a/src/backend/storage/smgr/smgr.c b/src/backend/storage/smgr/smgr.c
index 44dd2d5bff..0c88bbfcc7 100644
--- a/src/backend/storage/smgr/smgr.c
+++ b/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)