freelist.patch

text/x-patch

Download

diff --git a/src/backend/storage/buffer/freelist.c b/src/backend/storage/buffer/freelist.c
index bf9903b..304135e 100644
--- a/src/backend/storage/buffer/freelist.c
+++ b/src/backend/storage/buffer/freelist.c
@@ -28,7 +28,8 @@ typedef struct
 	int			nextVictimBuffer;
 
 	int			firstFreeBuffer;	/* Head of list of unused buffers */
-	int			lastFreeBuffer; /* Tail of list of unused buffers */
+	int			lastFreeBuffer;		 /* Tail of list of unused buffers */
+	slock_t			mutex;			/* Protects all of these, except sometimes not nextVictimBuffer */
 
 	/*
 	 * NOTE: lastFreeBuffer is undefined when firstFreeBuffer is -1 (that is,
@@ -44,7 +45,7 @@ typedef struct
 } BufferStrategyControl;
 
 /* Pointers to shared state */
-static BufferStrategyControl *StrategyControl = NULL;
+static volatile BufferStrategyControl *StrategyControl = NULL;
 
 /*
  * Private (non-shared) state for managing a ring of shared buffers to re-use.
@@ -123,15 +124,13 @@ StrategyGetBuffer(BufferAccessStrategy strategy, bool *lock_held)
 		}
 	}
 
-	/* Nope, so lock the freelist */
-	*lock_held = true;
-	LWLockAcquire(BufFreelistLock, LW_EXCLUSIVE);
-
 	/*
 	 * We count buffer allocation requests so that the bgwriter can estimate
 	 * the rate of buffer consumption.	Note that buffers recycled by a
 	 * strategy object are intentionally not counted here.
+	 * This is now done without a lock, and so updates can be lost
 	 */
+      
 	StrategyControl->numBufferAllocs++;
 
 	/*
@@ -142,6 +141,12 @@ StrategyGetBuffer(BufferAccessStrategy strategy, bool *lock_held)
 	 */
 	while (StrategyControl->firstFreeBuffer >= 0)
 	{
+		SpinLockAcquire(&StrategyControl->mutex);
+		if (StrategyControl->firstFreeBuffer<0)
+		{
+			SpinLockRelease(&StrategyControl->mutex);
+			break;
+		}
 		buf = &BufferDescriptors[StrategyControl->firstFreeBuffer];
 		Assert(buf->freeNext != FREENEXT_NOT_IN_LIST);
 
@@ -149,6 +154,8 @@ StrategyGetBuffer(BufferAccessStrategy strategy, bool *lock_held)
 		StrategyControl->firstFreeBuffer = buf->freeNext;
 		buf->freeNext = FREENEXT_NOT_IN_LIST;
 
+		SpinLockRelease(&StrategyControl->mutex);
+
 		/*
 		 * If the buffer is pinned or has a nonzero usage_count, we cannot use
 		 * it; discard it and retry.  (This can only happen if VACUUM put a
@@ -161,21 +168,32 @@ StrategyGetBuffer(BufferAccessStrategy strategy, bool *lock_held)
 		{
 			if (strategy != NULL)
 				AddBufferToRing(strategy, buf);
+			*lock_held = false;
 			return buf;
 		}
 		UnlockBufHdr(buf);
 	}
+	*lock_held = false;
 
 	/* Nothing on the freelist, so run the "clock sweep" algorithm */
 	trycounter = NBuffers;
 	for (;;)
 	{
-		buf = &BufferDescriptors[StrategyControl->nextVictimBuffer];
+
+		int localVictim=StrategyControl->nextVictimBuffer;
+		if (localVictim >= NBuffers) localVictim=0;
+
+		buf = &BufferDescriptors[localVictim];
 
 		if (++StrategyControl->nextVictimBuffer >= NBuffers)
 		{
-			StrategyControl->nextVictimBuffer = 0;
-			StrategyControl->completePasses++;
+			SpinLockAcquire(&StrategyControl->mutex);
+			if (StrategyControl->nextVictimBuffer >= NBuffers) 
+			{
+				StrategyControl->nextVictimBuffer = 0;
+				StrategyControl->completePasses++;
+			};
+			SpinLockRelease(&StrategyControl->mutex);
 		}
 
 		/*
@@ -223,7 +241,7 @@ StrategyGetBuffer(BufferAccessStrategy strategy, bool *lock_held)
 void
 StrategyFreeBuffer(volatile BufferDesc *buf)
 {
-	LWLockAcquire(BufFreelistLock, LW_EXCLUSIVE);
+	SpinLockAcquire(&StrategyControl->mutex);
 
 	/*
 	 * It is possible that we are told to put something in the freelist that
@@ -237,7 +255,7 @@ StrategyFreeBuffer(volatile BufferDesc *buf)
 		StrategyControl->firstFreeBuffer = buf->buf_id;
 	}
 
-	LWLockRelease(BufFreelistLock);
+	SpinLockRelease(&StrategyControl->mutex);
 }
 
 /*
@@ -256,8 +274,9 @@ StrategySyncStart(uint32 *complete_passes, uint32 *num_buf_alloc)
 {
 	int			result;
 
-	LWLockAcquire(BufFreelistLock, LW_EXCLUSIVE);
+	SpinLockAcquire(&StrategyControl->mutex);
 	result = StrategyControl->nextVictimBuffer;
+	if (result >= NBuffers) result=0;
 	if (complete_passes)
 		*complete_passes = StrategyControl->completePasses;
 	if (num_buf_alloc)
@@ -265,7 +284,7 @@ StrategySyncStart(uint32 *complete_passes, uint32 *num_buf_alloc)
 		*num_buf_alloc = StrategyControl->numBufferAllocs;
 		StrategyControl->numBufferAllocs = 0;
 	}
-	LWLockRelease(BufFreelistLock);
+	SpinLockRelease(&StrategyControl->mutex);
 	return result;
 }
 
@@ -337,6 +356,7 @@ StrategyInitialize(bool init)
 		 */
 		StrategyControl->firstFreeBuffer = 0;
 		StrategyControl->lastFreeBuffer = NBuffers - 1;
+		SpinLockInit(&StrategyControl->mutex);
 
 		/* Initialize the clock sweep pointer */
 		StrategyControl->nextVictimBuffer = 0;