Several changes here, not very related but touching some of the same files.

* Buffer refcount cleanup (per my "progress report" to pghackers, 9/22). * Add links to backend PROC structs to sinval's array of per-backend info, and use these links for routines that need to check the state of all backends (rather than the slow, complicated search of the ShmemIndex hashtable that was used before). Add databaseOID to PROC structs. * Use this to implement an interlock that prevents DESTROY DATABASE of a database containing running backends. (It's a little tricky to prevent a concurrently-starting backend from getting in there, since the new backend is not able to lock anything at the time it tries to look up its database in pg_database. My solution is to recheck that the DB is OK at the end of InitPostgres. It may not be a 100% solution, but it's a lot better than no interlock at all...) * In ALTER TABLE RENAME, flush buffers for the relation before doing the rename of the physical files, to ensure we don't get failures later from mdblindwrt(). * Update TRUNCATE patch so that it actually compiles against current sources :-(. You should do "make clean all" after pulling these changes.
2025-11-12 05:01:15 +03:00 · 1999-09-24 00:25:33 +00:00
parent ad791c1d14
commit e812458b27
43 changed files with 1194 additions and 931 deletions
--- a/src/backend/storage/buffer/buf_init.c
+++ b/src/backend/storage/buffer/buf_init.c
@@ -7,7 +7,7 @@
 *
 *
 * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/storage/buffer/buf_init.c,v 1.29 1999/07/17 20:17:40 momjian Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/storage/buffer/buf_init.c,v 1.30 1999/09/24 00:24:29 tgl Exp $
 *
 *-------------------------------------------------------------------------
 */
@@ -64,7 +64,6 @@ long	   *NWaitIOBackendP;
 extern IpcSemaphoreId WaitIOSemId;

 long	   *PrivateRefCount;	/* also used in freelist.c */
-long	   *LastRefCount;		/* refcounts of last ExecMain level */
 bits8	   *BufferLocks;		/* */
 long	   *CommitInfoNeedsSave;/* to write buffers where we have filled
 								 * in t_infomask */
@@ -244,7 +243,6 @@ InitBufferPool(IPCKey key)
 	}
 #endif
 	PrivateRefCount = (long *) calloc(NBuffers, sizeof(long));
-	LastRefCount = (long *) calloc(NBuffers, sizeof(long));
 	BufferLocks = (bits8 *) calloc(NBuffers, sizeof(bits8));
 	CommitInfoNeedsSave = (long *) calloc(NBuffers, sizeof(long));
 }
--- a/src/backend/storage/buffer/bufmgr.c
+++ b/src/backend/storage/buffer/bufmgr.c
@@ -7,7 +7,7 @@
 *
 *
 * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/storage/buffer/bufmgr.c,v 1.62 1999/09/18 19:07:26 tgl Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/storage/buffer/bufmgr.c,v 1.63 1999/09/24 00:24:29 tgl Exp $
 *
 *-------------------------------------------------------------------------
 */
@@ -89,9 +89,6 @@ static void BufferSync(void);
 static int	BufferReplace(BufferDesc *bufHdr, bool bufferLockHeld);
 void		PrintBufferDescs(void);

-/* not static but used by vacuum only ... */
-int			BlowawayRelationBuffers(Relation rel, BlockNumber block);
-
 /* ---------------------------------------------------
 * RelationGetBufferWithBuffer
 *		see if the given buffer is what we want
@@ -146,9 +143,6 @@ RelationGetBufferWithBuffer(Relation relation,
 *		opened already.
 */

-extern int	ShowPinTrace;
-
-
 #undef ReadBuffer				/* conflicts with macro when BUFMGR_DEBUG
 								 * defined */

@@ -499,6 +493,7 @@ BufferAlloc(Relation reln,
 					SignalIO(buf);
 #endif	 /* !HAS_TEST_AND_SET */
 				PrivateRefCount[BufferDescriptorGetBuffer(buf) - 1] = 0;
+				Assert(buf->refcount > 0);
 				buf->refcount--;
 				if (buf->refcount == 0)
 				{
@@ -575,10 +570,14 @@ BufferAlloc(Relation reln,
 						SignalIO(buf);
 #endif	 /* !HAS_TEST_AND_SET */
 					/* give up the buffer since we don't need it any more */
-					buf->refcount--;
 					PrivateRefCount[BufferDescriptorGetBuffer(buf) - 1] = 0;
-					AddBufferToFreelist(buf);
-					buf->flags |= BM_FREE;
+					Assert(buf->refcount > 0);
+					buf->refcount--;
+					if (buf->refcount == 0)
+					{
+						AddBufferToFreelist(buf);
+						buf->flags |= BM_FREE;
+					}
 					buf->flags &= ~BM_IO_IN_PROGRESS;
 				}

@@ -791,7 +790,7 @@ FlushBuffer(Buffer buffer, bool release)
 	int			status;

 	if (BufferIsLocal(buffer))
-		return FlushLocalBuffer(buffer, release);
+		return FlushLocalBuffer(buffer, release) ? STATUS_OK : STATUS_ERROR;

 	if (BAD_BUFFER_ID(buffer))
 		return STATUS_ERROR;
@@ -813,7 +812,7 @@ FlushBuffer(Buffer buffer, bool release)
 	status = smgrflush(DEFAULT_SMGR, bufrel, bufHdr->tag.blockNum,
 					   (char *) MAKE_PTR(bufHdr->data));

-	/* drop relcache refcount incremented by RelationIdCacheGetRelation */
+	/* drop relcache refcnt incremented by RelationIdCacheGetRelation */
 	RelationDecrementReferenceCount(bufrel);

 	if (status == SM_FAIL)
@@ -908,15 +907,10 @@ ReleaseAndReadBuffer(Buffer buffer,
 			bufHdr = &BufferDescriptors[buffer - 1];
 			Assert(PrivateRefCount[buffer - 1] > 0);
 			PrivateRefCount[buffer - 1]--;
-			if (PrivateRefCount[buffer - 1] == 0 &&
-				LastRefCount[buffer - 1] == 0)
+			if (PrivateRefCount[buffer - 1] == 0)
 			{
-
-				/*
-				 * only release buffer if it is not pinned in previous
-				 * ExecMain level
-				 */
 				SpinAcquire(BufMgrLock);
+				Assert(bufHdr->refcount > 0);
 				bufHdr->refcount--;
 				if (bufHdr->refcount == 0)
 				{
@@ -994,7 +988,7 @@ BufferSync()
 						elog(ERROR, "BufferSync: write error %u for %s",
 							 bufHdr->tag.blockNum, bufHdr->sb_relname);
 					}
-					/* drop refcount from RelationIdCacheGetRelation */
+					/* drop refcnt from RelationIdCacheGetRelation */
 					if (reln != (Relation) NULL)
 						RelationDecrementReferenceCount(reln);
 					continue;
@@ -1049,7 +1043,7 @@ BufferSync()
 				 */
 				if (!(bufHdr->flags & BM_JUST_DIRTIED))
 					bufHdr->flags &= ~BM_DIRTY;
-				/* drop refcount from RelationIdCacheGetRelation */
+				/* drop refcnt from RelationIdCacheGetRelation */
 				if (reln != (Relation) NULL)
 					RelationDecrementReferenceCount(reln);
 			}
@@ -1175,7 +1169,7 @@ ResetBufferUsage()
 *		ResetBufferPool
 *
 *		this routine is supposed to be called when a transaction aborts.
- *		it will release all the buffer pins held by the transaciton.
+ *		it will release all the buffer pins held by the transaction.
 *
 * ----------------------------------------------
 */
@@ -1184,15 +1178,24 @@ ResetBufferPool()
 {
 	int			i;

-	for (i = 1; i <= NBuffers; i++)
+	for (i = 0; i < NBuffers; i++)
 	{
-		CommitInfoNeedsSave[i - 1] = 0;
-		if (BufferIsValid(i))
+		if (PrivateRefCount[i] != 0)
 		{
-			while (PrivateRefCount[i - 1] > 0)
-				ReleaseBuffer(i);
+			BufferDesc *buf = &BufferDescriptors[i];
+
+			SpinAcquire(BufMgrLock);
+			Assert(buf->refcount > 0);
+			buf->refcount--;
+			if (buf->refcount == 0)
+			{
+				AddBufferToFreelist(buf);
+				buf->flags |= BM_FREE;
+			}
+			SpinRelease(BufMgrLock);
 		}
-		LastRefCount[i - 1] = 0;
+		PrivateRefCount[i] = 0;
+		CommitInfoNeedsSave[i] = 0;
 	}

 	ResetLocalBufferPool();
@@ -1213,7 +1216,7 @@ BufferPoolCheckLeak()

 	for (i = 1; i <= NBuffers; i++)
 	{
-		if (BufferIsValid(i))
+		if (PrivateRefCount[i - 1] != 0)
 		{
 			BufferDesc *buf = &(BufferDescriptors[i - 1]);

@@ -1226,7 +1229,7 @@ relname=%s, blockNum=%d, flags=0x%x, refcount=%d %d)",
 			result = 1;
 		}
 	}
-	return (result);
+	return result;
 }

 /* ------------------------------------------------
@@ -1287,7 +1290,7 @@ BufferGetRelation(Buffer buffer)
 	relation = RelationIdGetRelation(relid);
 	Assert(relation);

-	/* drop relcache refcount incremented by RelationIdGetRelation */
+	/* drop relcache refcnt incremented by RelationIdGetRelation */
 	RelationDecrementReferenceCount(relation);

 	if (RelationHasReferenceCountZero(relation))
@@ -1354,7 +1357,7 @@ BufferReplace(BufferDesc *bufHdr, bool bufferLockHeld)
 							  (char *) MAKE_PTR(bufHdr->data));
 	}

-	/* drop relcache refcount incremented by RelationIdCacheGetRelation */
+	/* drop relcache refcnt incremented by RelationIdCacheGetRelation */
 	if (reln != (Relation) NULL)
 		RelationDecrementReferenceCount(reln);

@@ -1549,10 +1552,27 @@ BufferPoolBlowaway()
 #endif

 /* ---------------------------------------------------------------------
- *		BlowawayRelationBuffers
+ *		FlushRelationBuffers
 *
- *		This function blowaway all the pages with blocknumber >= passed
- *		of a relation in the buffer pool. Used by vacuum before truncation...
+ *		This function removes from the buffer pool all pages of a relation
+ *		that have blocknumber >= specified block.  If doFlush is true,
+ *		dirty buffers are written out --- otherwise it's an error for any
+ *		of the buffers to be dirty.
+ *
+ *		This is used by VACUUM before truncating the relation to the given
+ *		number of blocks.  For VACUUM, we pass doFlush = false since it would
+ *		mean a bug in VACUUM if any of the unwanted pages were still dirty.
+ *		(TRUNCATE TABLE also uses it in the same way.)
+ *
+ *		This is also used by RENAME TABLE (with block = 0 and doFlush = true)
+ *		to clear out the buffer cache before renaming the physical files of
+ *		a relation.  Without that, some other backend might try to do a
+ *		blind write of a buffer page (relying on the sb_relname of the buffer)
+ *		and fail because it's not got the right filename anymore.
+ *
+ *		In both cases, the caller should be holding AccessExclusiveLock on
+ *		the target relation to ensure that no other backend is busy reading
+ *		more blocks of the relation...
 *
 *		Returns: 0 - Ok, -1 - DIRTY, -2 - PINNED
 *
@@ -1561,7 +1581,7 @@ BufferPoolBlowaway()
 * --------------------------------------------------------------------
 */
 int
-BlowawayRelationBuffers(Relation rel, BlockNumber block)
+FlushRelationBuffers(Relation rel, BlockNumber block, bool doFlush)
 {
 	int			i;
 	BufferDesc *buf;
@@ -1576,13 +1596,25 @@ BlowawayRelationBuffers(Relation rel, BlockNumber block)
 			{
 				if (buf->flags & BM_DIRTY)
 				{
-					elog(NOTICE, "BlowawayRelationBuffers(%s (local), %u): block %u is dirty",
-					rel->rd_rel->relname.data, block, buf->tag.blockNum);
-					return -1;
+					if (doFlush)
+					{
+						if (FlushBuffer(-i-1, false) != STATUS_OK)
+						{
+							elog(NOTICE, "FlushRelationBuffers(%s (local), %u): block %u is dirty, could not flush it",
+								 rel->rd_rel->relname.data, block, buf->tag.blockNum);
+							return -1;
+						}
+					}
+					else
+					{
+						elog(NOTICE, "FlushRelationBuffers(%s (local), %u): block %u is dirty",
+							 rel->rd_rel->relname.data, block, buf->tag.blockNum);
+						return -1;
+					}
 				}
 				if (LocalRefCount[i] > 0)
 				{
-					elog(NOTICE, "BlowawayRelationBuffers(%s (local), %u): block %u is referenced (%d)",
+					elog(NOTICE, "FlushRelationBuffers(%s (local), %u): block %u is referenced (%d)",
 						 rel->rd_rel->relname.data, block,
 						 buf->tag.blockNum, LocalRefCount[i]);
 					return -2;
@@ -1603,18 +1635,33 @@ BlowawayRelationBuffers(Relation rel, BlockNumber block)
 		{
 			if (buf->flags & BM_DIRTY)
 			{
-				elog(NOTICE, "BlowawayRelationBuffers(%s, %u): block %u is dirty (private %d, last %d, global %d)",
-					 buf->sb_relname, block, buf->tag.blockNum,
-					 PrivateRefCount[i], LastRefCount[i], buf->refcount);
-				SpinRelease(BufMgrLock);
-				return -1;
+				if (doFlush)
+				{
+					SpinRelease(BufMgrLock);
+					if (FlushBuffer(i+1, false) != STATUS_OK)
+					{
+						elog(NOTICE, "FlushRelationBuffers(%s, %u): block %u is dirty (private %d, global %d), could not flush it",
+							 buf->sb_relname, block, buf->tag.blockNum,
+							 PrivateRefCount[i], buf->refcount);
+						return -1;
+					}
+					SpinAcquire(BufMgrLock);
+				}
+				else
+				{
+					SpinRelease(BufMgrLock);
+					elog(NOTICE, "FlushRelationBuffers(%s, %u): block %u is dirty (private %d, global %d)",
+						 buf->sb_relname, block, buf->tag.blockNum,
+						 PrivateRefCount[i], buf->refcount);
+					return -1;
+				}
 			}
 			if (!(buf->flags & BM_FREE))
 			{
-				elog(NOTICE, "BlowawayRelationBuffers(%s, %u): block %u is referenced (private %d, last %d, global %d)",
-					 buf->sb_relname, block, buf->tag.blockNum,
-					 PrivateRefCount[i], LastRefCount[i], buf->refcount);
 				SpinRelease(BufMgrLock);
+				elog(NOTICE, "FlushRelationBuffers(%s, %u): block %u is referenced (private %d, global %d)",
+					 buf->sb_relname, block, buf->tag.blockNum,
+					 PrivateRefCount[i], buf->refcount);
 				return -2;
 			}
 			BufTableDelete(buf);
@@ -1650,14 +1697,10 @@ ReleaseBuffer(Buffer buffer)

 	Assert(PrivateRefCount[buffer - 1] > 0);
 	PrivateRefCount[buffer - 1]--;
-	if (PrivateRefCount[buffer - 1] == 0 && LastRefCount[buffer - 1] == 0)
+	if (PrivateRefCount[buffer - 1] == 0)
 	{
-
-		/*
-		 * only release buffer if it is not pinned in previous ExecMain
-		 * levels
-		 */
 		SpinAcquire(BufMgrLock);
+		Assert(bufHdr->refcount > 0);
 		bufHdr->refcount--;
 		if (bufHdr->refcount == 0)
 		{
@@ -1892,32 +1935,6 @@ _bm_die(Oid dbId, Oid relId, int blkNo, int bufNo,

 #endif	 /* BMTRACE */

-void
-BufferRefCountReset(int *refcountsave)
-{
-	int			i;
-
-	for (i = 0; i < NBuffers; i++)
-	{
-		refcountsave[i] = PrivateRefCount[i];
-		LastRefCount[i] += PrivateRefCount[i];
-		PrivateRefCount[i] = 0;
-	}
-}
-
-void
-BufferRefCountRestore(int *refcountsave)
-{
-	int			i;
-
-	for (i = 0; i < NBuffers; i++)
-	{
-		PrivateRefCount[i] = refcountsave[i];
-		LastRefCount[i] -= refcountsave[i];
-		refcountsave[i] = 0;
-	}
-}
-
 int
 SetBufferWriteMode(int mode)
 {
--- a/src/backend/storage/buffer/freelist.c
+++ b/src/backend/storage/buffer/freelist.c
@@ -8,7 +8,7 @@
 *
 *
 * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/storage/buffer/freelist.c,v 1.18 1999/07/17 20:17:41 momjian Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/storage/buffer/freelist.c,v 1.19 1999/09/24 00:24:29 tgl Exp $
 *
 *-------------------------------------------------------------------------
 */
@@ -105,7 +105,7 @@ PinBuffer(BufferDesc *buf)

 	b = BufferDescriptorGetBuffer(buf) - 1;
 	Assert(PrivateRefCount[b] >= 0);
-	if (PrivateRefCount[b] == 0 && LastRefCount[b] == 0)
+	if (PrivateRefCount[b] == 0)
 		buf->refcount++;
 	PrivateRefCount[b]++;
 }
@@ -138,10 +138,10 @@ UnpinBuffer(BufferDesc *buf)
 {
 	long		b = BufferDescriptorGetBuffer(buf) - 1;

-	Assert(buf->refcount);
+	Assert(buf->refcount > 0);
 	Assert(PrivateRefCount[b] > 0);
 	PrivateRefCount[b]--;
-	if (PrivateRefCount[b] == 0 && LastRefCount[b] == 0)
+	if (PrivateRefCount[b] == 0)
 		buf->refcount--;
 	NotInQueue(buf);

--- a/src/backend/storage/buffer/localbuf.c
+++ b/src/backend/storage/buffer/localbuf.c
@@ -15,7 +15,7 @@
 *
 *
 * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/storage/buffer/localbuf.c,v 1.27 1999/09/18 19:07:26 tgl Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/storage/buffer/localbuf.c,v 1.28 1999/09/24 00:24:29 tgl Exp $
 *
 *-------------------------------------------------------------------------
 */
@@ -193,9 +193,11 @@ FlushLocalBuffer(Buffer buffer, bool release)
 	/* drop relcache refcount incremented by RelationIdCacheGetRelation */
 	RelationDecrementReferenceCount(bufrel);

-	Assert(LocalRefCount[bufid] > 0);
 	if (release)
+	{
+		Assert(LocalRefCount[bufid] > 0);
 		LocalRefCount[bufid]--;
+	}

 	return true;
 }
--- a/src/backend/storage/ipc/shmem.c
+++ b/src/backend/storage/ipc/shmem.c
@@ -7,7 +7,7 @@
 *
 *
 * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/storage/ipc/shmem.c,v 1.45 1999/07/17 20:17:44 momjian Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/storage/ipc/shmem.c,v 1.46 1999/09/24 00:24:35 tgl Exp $
 *
 *-------------------------------------------------------------------------
 */
@@ -579,174 +579,3 @@ ShmemInitStruct(char *name, unsigned long size, bool *foundPtr)
 	SpinRelease(ShmemIndexLock);
 	return structPtr;
 }
-
-/*
- * TransactionIdIsInProgress -- is given transaction running by some backend
- *
- * Strange place for this func, but we have to lookup process data structures
- * for all running backends. - vadim 11/26/96
- *
- * We should keep all PROC structs not in ShmemIndex - this is too
- * general hash table...
- *
- */
-bool
-TransactionIdIsInProgress(TransactionId xid)
-{
-	ShmemIndexEnt *result;
-	PROC	   *proc;
-
-	Assert(ShmemIndex);
-
-	SpinAcquire(ShmemIndexLock);
-
-	hash_seq((HTAB *) NULL);
-	while ((result = (ShmemIndexEnt *) hash_seq(ShmemIndex)) != NULL)
-	{
-		if (result == (ShmemIndexEnt *) TRUE)
-		{
-			SpinRelease(ShmemIndexLock);
-			return false;
-		}
-		if (result->location == INVALID_OFFSET ||
-			strncmp(result->key, "PID ", 4) != 0)
-			continue;
-		proc = (PROC *) MAKE_PTR(result->location);
-		if (proc->xid == xid)
-		{
-			SpinRelease(ShmemIndexLock);
-			return true;
-		}
-	}
-
-	SpinRelease(ShmemIndexLock);
-	elog(ERROR, "TransactionIdIsInProgress: ShmemIndex corrupted");
-	return false;
-}
-
-/*
- * GetSnapshotData -- returns information about running transactions.
- *
- * Yet another strange func for this place...	- vadim 07/21/98
- */
-Snapshot
-GetSnapshotData(bool serializable)
-{
-	Snapshot	snapshot = (Snapshot) malloc(sizeof(SnapshotData));
-	ShmemIndexEnt *result;
-	PROC	   *proc;
-	TransactionId cid = GetCurrentTransactionId();
-	TransactionId xid;
-	uint32		count = 0;
-	uint32		have = 32;
-
-	Assert(ShmemIndex);
-
-	snapshot->xip = (TransactionId *) malloc(have * sizeof(TransactionId));
-	snapshot->xmin = cid;
-
-	SpinAcquire(ShmemIndexLock);
-	/*
-	 * Unfortunately, we have to call ReadNewTransactionId()
-	 * after acquiring ShmemIndexLock above. It's not good because of
-	 * ReadNewTransactionId() does SpinAcquire(OidGenLockId) but
-	 * _necessary_.
-	 */
-	ReadNewTransactionId(&(snapshot->xmax));
-
-	hash_seq((HTAB *) NULL);
-	while ((result = (ShmemIndexEnt *) hash_seq(ShmemIndex)) != NULL)
-	{
-		if (result == (ShmemIndexEnt *) TRUE)
-		{
-			if (serializable)
-				MyProc->xmin = snapshot->xmin;
-			/* Serializable snapshot must be computed before any other... */
-			Assert(MyProc->xmin != InvalidTransactionId);
-			SpinRelease(ShmemIndexLock);
-			snapshot->xcnt = count;
-			return snapshot;
-		}
-		if (result->location == INVALID_OFFSET ||
-			strncmp(result->key, "PID ", 4) != 0)
-			continue;
-		proc = (PROC *) MAKE_PTR(result->location);
-		/* 
-		 * We don't use spin-locking when changing proc->xid 
-		 * in GetNewTransactionId() and in AbortTransaction() !..
-		 */
-		xid = proc->xid;
-		if (proc == MyProc || 
-			xid < FirstTransactionId || xid >= snapshot->xmax)
-		{
-			/*
-			 * Seems that there is no sense to store xid >= snapshot->xmax
-			 * (what we got from ReadNewTransactionId above) in snapshot->xip 
-			 * - we just assume that all xacts with such xid-s are running 
-			 * and may be ignored.
-			 */
-			continue;
-		}
-		if (xid < snapshot->xmin)
-			snapshot->xmin = xid;
-		if (have == 0)
-		{
-			snapshot->xip = (TransactionId *) realloc(snapshot->xip,
-								   (count + 32) * sizeof(TransactionId));
-			have = 32;
-		}
-		snapshot->xip[count] = xid;
-		have--;
-		count++;
-	}
-
-	SpinRelease(ShmemIndexLock);
-	free(snapshot->xip);
-	free(snapshot);
-	elog(ERROR, "GetSnapshotData: ShmemIndex corrupted");
-	return NULL;
-}
-
-/*
- * GetXmaxRecent -- returns oldest transaction that was running
- *					when all current transaction was started.
- *					It's used by vacuum to decide what deleted
- *					tuples must be preserved in a table.
- *
- * And yet another strange func for this place...	- vadim 03/18/99
- */
-void
-GetXmaxRecent(TransactionId *XmaxRecent)
-{
-	ShmemIndexEnt *result;
-	PROC	   *proc;
-	TransactionId xmin;
-
-	Assert(ShmemIndex);
-
-	*XmaxRecent = GetCurrentTransactionId();
-
-	SpinAcquire(ShmemIndexLock);
-
-	hash_seq((HTAB *) NULL);
-	while ((result = (ShmemIndexEnt *) hash_seq(ShmemIndex)) != NULL)
-	{
-		if (result == (ShmemIndexEnt *) TRUE)
-		{
-			SpinRelease(ShmemIndexLock);
-			return;
-		}
-		if (result->location == INVALID_OFFSET ||
-			strncmp(result->key, "PID ", 4) != 0)
-			continue;
-		proc = (PROC *) MAKE_PTR(result->location);
-		xmin = proc->xmin;	/* we don't use spin-locking in AbortTransaction() ! */
-		if (proc == MyProc || xmin < FirstTransactionId)
-			continue;
-		if (xmin < *XmaxRecent)
-			*XmaxRecent = xmin;
-	}
-
-	SpinRelease(ShmemIndexLock);
-	elog(ERROR, "GetXmaxRecent: ShmemIndex corrupted");
-}
--- a/src/backend/storage/ipc/sinval.c
+++ b/src/backend/storage/ipc/sinval.c
@@ -7,7 +7,7 @@
 *
 *
 * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/storage/ipc/sinval.c,v 1.18 1999/09/06 19:37:38 tgl Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/storage/ipc/sinval.c,v 1.19 1999/09/24 00:24:35 tgl Exp $
 *
 *-------------------------------------------------------------------------
 */
@@ -18,8 +18,10 @@
 #include "postgres.h"

 #include "storage/backendid.h"
+#include "storage/proc.h"
 #include "storage/sinval.h"
 #include "storage/sinvaladt.h"
+#include "utils/tqual.h"

 SPINLOCK	SInvalLock = (SPINLOCK) NULL;

@@ -165,3 +167,201 @@ InvalidateSharedInvalid(void (*invalFunction) (),
 		SpinRelease(SInvalLock);
 	}
 }
+
+
+/****************************************************************************/
+/* Functions that need to scan the PROC structures of all running backends.	*/
+/* It's a bit strange to keep these in sinval.c, since they don't have any	*/
+/* direct relationship to shared-cache invalidation.  But the procState		*/
+/* array in the SI segment is the only place in the system where we have	*/
+/* an array of per-backend data, so it is the most convenient place to keep	*/
+/* pointers to the backends' PROC structures.  We used to implement these	*/
+/* functions with a slow, ugly search through the ShmemIndex hash table ---	*/
+/* now they are simple loops over the SI ProcState array.					*/
+/****************************************************************************/
+
+
+/*
+ * DatabaseHasActiveBackends -- are there any backends running in the given DB
+ *
+ * This function is used to interlock DROP DATABASE against there being
+ * any active backends in the target DB --- dropping the DB while active
+ * backends remain would be a Bad Thing.  Note that we cannot detect here
+ * the possibility of a newly-started backend that is trying to connect
+ * to the doomed database, so additional interlocking is needed during
+ * backend startup.
+ */
+
+bool
+DatabaseHasActiveBackends(Oid databaseId)
+{
+	bool		result = false;
+	SISeg	   *segP = shmInvalBuffer;
+	ProcState  *stateP = segP->procState;
+	int			index;
+
+	SpinAcquire(SInvalLock);
+
+	for (index = 0; index < segP->maxBackends; index++)
+	{
+		SHMEM_OFFSET	pOffset = stateP[index].procStruct;
+
+		if (pOffset != INVALID_OFFSET)
+		{
+			PROC	   *proc = (PROC *) MAKE_PTR(pOffset);
+
+			if (proc->databaseId == databaseId)
+			{
+				result = true;
+				break;
+			}
+		}
+	}
+
+	SpinRelease(SInvalLock);
+
+	return result;
+}
+
+/*
+ * TransactionIdIsInProgress -- is given transaction running by some backend
+ */
+bool
+TransactionIdIsInProgress(TransactionId xid)
+{
+	bool		result = false;
+	SISeg	   *segP = shmInvalBuffer;
+	ProcState  *stateP = segP->procState;
+	int			index;
+
+	SpinAcquire(SInvalLock);
+
+	for (index = 0; index < segP->maxBackends; index++)
+	{
+		SHMEM_OFFSET	pOffset = stateP[index].procStruct;
+
+		if (pOffset != INVALID_OFFSET)
+		{
+			PROC	   *proc = (PROC *) MAKE_PTR(pOffset);
+
+			if (proc->xid == xid)
+			{
+				result = true;
+				break;
+			}
+		}
+	}
+
+	SpinRelease(SInvalLock);
+
+	return result;
+}
+
+/*
+ * GetXmaxRecent -- returns oldest transaction that was running
+ *					when all current transaction was started.
+ *					It's used by vacuum to decide what deleted
+ *					tuples must be preserved in a table.
+ */
+void
+GetXmaxRecent(TransactionId *XmaxRecent)
+{
+	SISeg	   *segP = shmInvalBuffer;
+	ProcState  *stateP = segP->procState;
+	int			index;
+
+	*XmaxRecent = GetCurrentTransactionId();
+
+	SpinAcquire(SInvalLock);
+
+	for (index = 0; index < segP->maxBackends; index++)
+	{
+		SHMEM_OFFSET	pOffset = stateP[index].procStruct;
+
+		if (pOffset != INVALID_OFFSET)
+		{
+			PROC	   *proc = (PROC *) MAKE_PTR(pOffset);
+			TransactionId xmin;
+
+			xmin = proc->xmin;	/* we don't use spin-locking in AbortTransaction() ! */
+			if (proc == MyProc || xmin < FirstTransactionId)
+				continue;
+			if (xmin < *XmaxRecent)
+				*XmaxRecent = xmin;
+		}
+	}
+
+	SpinRelease(SInvalLock);
+}
+
+/*
+ * GetSnapshotData -- returns information about running transactions.
+ */
+Snapshot
+GetSnapshotData(bool serializable)
+{
+	Snapshot	snapshot = (Snapshot) malloc(sizeof(SnapshotData));
+	SISeg	   *segP = shmInvalBuffer;
+	ProcState  *stateP = segP->procState;
+	int			index;
+	int			count = 0;
+
+	/* There can be no more than maxBackends active transactions,
+	 * so this is enough space:
+	 */
+	snapshot->xip = (TransactionId *)
+		malloc(segP->maxBackends * sizeof(TransactionId));
+	snapshot->xmin = GetCurrentTransactionId();
+
+	SpinAcquire(SInvalLock);
+
+	/*
+	 * Unfortunately, we have to call ReadNewTransactionId()
+	 * after acquiring SInvalLock above. It's not good because
+	 * ReadNewTransactionId() does SpinAcquire(OidGenLockId) but
+	 * _necessary_.
+	 */
+	ReadNewTransactionId(&(snapshot->xmax));
+
+	for (index = 0; index < segP->maxBackends; index++)
+	{
+		SHMEM_OFFSET	pOffset = stateP[index].procStruct;
+
+		if (pOffset != INVALID_OFFSET)
+		{
+			PROC	   *proc = (PROC *) MAKE_PTR(pOffset);
+			TransactionId xid;
+
+			/* 
+			 * We don't use spin-locking when changing proc->xid 
+			 * in GetNewTransactionId() and in AbortTransaction() !..
+			 */
+			xid = proc->xid;
+			if (proc == MyProc || 
+				xid < FirstTransactionId || xid >= snapshot->xmax)
+			{
+				/*
+				 * Seems that there is no sense to store xid >= snapshot->xmax
+				 * (what we got from ReadNewTransactionId above) in
+				 * snapshot->xip - we just assume that all xacts with such
+				 * xid-s are running and may be ignored.
+				 */
+				continue;
+			}
+			if (xid < snapshot->xmin)
+				snapshot->xmin = xid;
+			snapshot->xip[count] = xid;
+			count++;
+		}
+	}
+
+	if (serializable)
+		MyProc->xmin = snapshot->xmin;
+	/* Serializable snapshot must be computed before any other... */
+	Assert(MyProc->xmin != InvalidTransactionId);
+
+	SpinRelease(SInvalLock);
+
+	snapshot->xcnt = count;
+	return snapshot;
+}
--- a/src/backend/storage/ipc/sinvaladt.c
+++ b/src/backend/storage/ipc/sinvaladt.c
@@ -7,7 +7,7 @@
 *
 *
 * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/storage/ipc/sinvaladt.c,v 1.26 1999/09/09 14:56:06 tgl Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/storage/ipc/sinvaladt.c,v 1.27 1999/09/24 00:24:35 tgl Exp $
 *
 *-------------------------------------------------------------------------
 */
@@ -19,6 +19,8 @@
 #include "miscadmin.h"
 #include "storage/backendid.h"
 #include "storage/lmgr.h"
+#include "storage/proc.h"
+#include "storage/sinval.h"
 #include "storage/sinvaladt.h"
 #include "utils/trace.h"

@@ -125,6 +127,7 @@ SISegInit(SISeg *segP, int maxBackends)
 		segP->procState[i].nextMsgNum = -1;	/* inactive */
 		segP->procState[i].resetState = false;
 		segP->procState[i].tag = InvalidBackendTag;
+		segP->procState[i].procStruct = INVALID_OFFSET;
 	}
 }

@@ -161,8 +164,8 @@ SIBackendInit(SISeg *segP)
 		}
 	}

-	/* elog() with spinlock held is probably not too cool, but these
-	 * conditions should never happen anyway.
+	/* elog() with spinlock held is probably not too cool, but this
+	 * condition should never happen anyway.
 	 */
 	if (stateP == NULL)
 	{
@@ -179,9 +182,10 @@ SIBackendInit(SISeg *segP)
 #endif	 /* INVALIDDEBUG */

 	/* mark myself active, with all extant messages already read */
-	stateP->tag = MyBackendTag;
-	stateP->resetState = false;
 	stateP->nextMsgNum = segP->maxMsgNum;
+	stateP->resetState = false;
+	stateP->tag = MyBackendTag;
+	stateP->procStruct = MAKE_OFFSET(MyProc);

 	/* register exit routine to mark my entry inactive at exit */
 	on_shmem_exit(CleanupInvalidationState, (caddr_t) segP);
@@ -193,7 +197,8 @@ SIBackendInit(SISeg *segP)
 * CleanupInvalidationState
 *		Mark the current backend as no longer active.
 *
- * This function is called via on_shmem_exit() during backend shutdown.
+ * This function is called via on_shmem_exit() during backend shutdown,
+ * so the caller has NOT acquired the lock for us.
 */
 static void
 CleanupInvalidationState(int status,
@@ -201,13 +206,14 @@ CleanupInvalidationState(int status,
 {
 	Assert(PointerIsValid(segP));

-	/* XXX we probably oughta grab the SInval spinlock for this...
-	 * but I think it is safe not to.
-	 */
+	SpinAcquire(SInvalLock);

 	segP->procState[MyBackendId - 1].nextMsgNum = -1;
 	segP->procState[MyBackendId - 1].resetState = false;
 	segP->procState[MyBackendId - 1].tag = InvalidBackendTag;
+	segP->procState[MyBackendId - 1].procStruct = INVALID_OFFSET;
+
+	SpinRelease(SInvalLock);
 }

 /*
--- a/src/backend/storage/lmgr/proc.c
+++ b/src/backend/storage/lmgr/proc.c
@@ -7,7 +7,7 @@
 *
 *
 * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/storage/lmgr/proc.c,v 1.60 1999/07/17 20:17:47 momjian Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/storage/lmgr/proc.c,v 1.61 1999/09/24 00:24:41 tgl Exp $
 *
 *-------------------------------------------------------------------------
 */
@@ -46,7 +46,7 @@
 *		This is so that we can support more backends. (system-wide semaphore
 *		sets run out pretty fast.)				  -ay 4/95
 *
- * $Header: /cvsroot/pgsql/src/backend/storage/lmgr/proc.c,v 1.60 1999/07/17 20:17:47 momjian Exp $
+ * $Header: /cvsroot/pgsql/src/backend/storage/lmgr/proc.c,v 1.61 1999/09/24 00:24:41 tgl Exp $
 */
 #include <sys/time.h>
 #include <unistd.h>
@@ -296,6 +296,7 @@ InitProcess(IPCKey key)
 	SpinRelease(ProcStructLock);

 	MyProc->pid = MyProcPid;
+	MyProc->databaseId = MyDatabaseId;
 	MyProc->xid = InvalidTransactionId;
 	MyProc->xmin = InvalidTransactionId;

--- a/src/backend/storage/smgr/md.c
+++ b/src/backend/storage/smgr/md.c
@@ -7,7 +7,7 @@
 *
 *
 * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/storage/smgr/md.c,v 1.53 1999/09/05 23:24:53 tgl Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/storage/smgr/md.c,v 1.54 1999/09/24 00:24:47 tgl Exp $
 *
 *-------------------------------------------------------------------------
 */
@@ -576,12 +576,10 @@ mdblindwrt(char *dbstr,
 /* this is work arround only !!! */
 	{
 		char		dbpath[MAXPGPATH + 1];
-		int4		owner;
 		Oid			id;
 		char	   *tmpPath;
-		int			tmpEncoding;

-		GetRawDatabaseInfo(dbstr, &owner, &id, dbpath, &tmpEncoding);
+		GetRawDatabaseInfo(dbstr, &id, dbpath);

 		if (id != dbid)
 			elog(FATAL, "mdblindwrt: oid of db %s is not %u", dbstr, dbid);
@@ -615,12 +613,10 @@ mdblindwrt(char *dbstr,
 /* this is work arround only !!! */
 	{
 		char		dbpath[MAXPGPATH + 1];
-		int4		owner;
 		Oid			id;
 		char	   *tmpPath;
-		int			tmpEncoding;

-		GetRawDatabaseInfo(dbstr, &owner, &id, dbpath, &tmpEncoding);
+		GetRawDatabaseInfo(dbstr, &id, dbpath);

 		if (id != dbid)
 			elog(FATAL, "mdblindwrt: oid of db %s is not %u", dbstr, dbid);