lockfree-getbuffer.patch

diff --git a/src/backend/storage/buffer/freelist.c b/src/backend/storage/buffer/freelist.c
index 76a4bec..86843ae 100644
--- a/src/backend/storage/buffer/freelist.c
+++ b/src/backend/storage/buffer/freelist.c
@@ -15,10 +15,10 @@
  */
 #include "postgres.h"
 
+#include "storage/atomic.h"
 #include "storage/buf_internals.h"
 #include "storage/bufmgr.h"
 
-
 /*
  * The shared freelist control information.
  */
@@ -28,12 +28,6 @@ typedef struct
 	int			nextVictimBuffer;
 
 	int			firstFreeBuffer;	/* Head of list of unused buffers */
-	int			lastFreeBuffer; /* Tail of list of unused buffers */
-
-	/*
-	 * NOTE: lastFreeBuffer is undefined when firstFreeBuffer is -1 (that is,
-	 * when the list is empty)
-	 */
 
 	/*
 	 * Statistics.	These counters should be wide enough that they can't
@@ -114,7 +108,9 @@ StrategyGetBuffer(BufferAccessStrategy strategy, bool *lock_held)
 	volatile BufferDesc *buf;
 	Latch	   *bgwriterLatch;
 	int			trycounter;
+	int			victim;
 
+	*lock_held = false;
 	/*
 	 * If given a strategy object, see whether it can select a buffer. We
 	 * assume strategy objects don't need the BufFreelistLock.
@@ -124,36 +120,27 @@ StrategyGetBuffer(BufferAccessStrategy strategy, bool *lock_held)
 		buf = GetBufferFromRing(strategy);
 		if (buf != NULL)
 		{
-			*lock_held = false;
 			return buf;
 		}
 	}
 
-	/* 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.
 	 */
-	StrategyControl->numBufferAllocs++;
+	pg_atomic_inc(StrategyControl->numBufferAllocs);
 
 	/*
-	 * If bgwriterLatch is set, we need to waken the bgwriter, but we should
-	 * not do so while holding BufFreelistLock; so release and re-grab.  This
-	 * is annoyingly tedious, but it happens at most once per bgwriter cycle,
-	 * so the performance hit is minimal.
+	 * If bgwriterLatch is set, we need to waken the bgwriter. Bgwriter needs
+	 * to see our allocation to not go to sleep again so we emit a write
+	 * barrier. Because this happens at most once per bgwriter cycle, the
+	 * performance hit is minimal.
 	 */
-	bgwriterLatch = StrategyControl->bgwriterLatch;
+	pg_atomic_fetch_and_set(bgwriterLatch, StrategyControl->bgwriterLatch, NULL);
 	if (bgwriterLatch)
-	{
-		StrategyControl->bgwriterLatch = NULL;
-		LWLockRelease(BufFreelistLock);
 		SetLatch(bgwriterLatch);
-		LWLockAcquire(BufFreelistLock, LW_EXCLUSIVE);
-	}
 
 	/*
 	 * Try to get a buffer from the freelist.  Note that the freeNext fields
@@ -161,6 +148,16 @@ StrategyGetBuffer(BufferAccessStrategy strategy, bool *lock_held)
 	 * individual buffer spinlocks, so it's OK to manipulate them without
 	 * holding the spinlock.
 	 */
+	if (StrategyControl->firstFreeBuffer >= 0)
+	{
+		/*
+		 * There might be something in the freelist, so lock and check
+		 * again */
+		*lock_held = true;
+		LWLockAcquire(BufFreelistLock, LW_EXCLUSIVE);
+		/*
+		 * XXX Need to avoid reusing value fetched before lock here,
+		 * volatile access required? Output from GCC 4.6 says no. */
 	while (StrategyControl->firstFreeBuffer >= 0)
 	{
 		buf = &BufferDescriptors[StrategyControl->firstFreeBuffer];
@@ -187,17 +184,33 @@ StrategyGetBuffer(BufferAccessStrategy strategy, bool *lock_held)
 		UnlockBufHdr(buf);
 	}
 
+		LWLockRelease(BufFreelistLock);
+		*lock_held = false;
+	}
+
 	/* Nothing on the freelist, so run the "clock sweep" algorithm */
 	trycounter = NBuffers;
 	for (;;)
 	{
-		buf = &BufferDescriptors[StrategyControl->nextVictimBuffer];
-
-		if (++StrategyControl->nextVictimBuffer >= NBuffers)
+		pg_atomic_fetch_and_add(victim, StrategyControl->nextVictimBuffer, 1);
+		/*
+		 * If we just incremented the value past the wrapping point
+		 * emit a correcting subtraction and account for the pass.
+		 */
+		if (victim == NBuffers-1)
 		{
-			StrategyControl->nextVictimBuffer = 0;
-			StrategyControl->completePasses++;
+			pg_atomic_sub(StrategyControl->nextVictimBuffer, NBuffers);
+			pg_atomic_inc(StrategyControl->completePasses);
 		}
+		/*
+		 * Do the modulus operation asynchronously in case the value had
+		 * overflowed. */
+		while (victim >= NBuffers)
+			victim -= NBuffers;
+
+		Assert(victim < NBuffers && victim >= 0);
+
+		buf = &BufferDescriptors[victim];
 
 		/*
 		 * If the buffer is pinned or has a nonzero usage_count, we cannot use
@@ -253,8 +266,6 @@ StrategyFreeBuffer(volatile BufferDesc *buf)
 	if (buf->freeNext == FREENEXT_NOT_IN_LIST)
 	{
 		buf->freeNext = StrategyControl->firstFreeBuffer;
-		if (buf->freeNext < 0)
-			StrategyControl->lastFreeBuffer = buf->buf_id;
 		StrategyControl->firstFreeBuffer = buf->buf_id;
 	}
 
@@ -277,16 +288,17 @@ StrategySyncStart(uint32 *complete_passes, uint32 *num_buf_alloc)
 {
 	int			result;
 
-	LWLockAcquire(BufFreelistLock, LW_EXCLUSIVE);
 	result = StrategyControl->nextVictimBuffer;
+	while (result >= NBuffers)
+		result -= NBuffers;
+
 	if (complete_passes)
 		*complete_passes = StrategyControl->completePasses;
 	if (num_buf_alloc)
 	{
-		*num_buf_alloc = StrategyControl->numBufferAllocs;
-		StrategyControl->numBufferAllocs = 0;
+		pg_atomic_fetch_and_set(*num_buf_alloc,
+			StrategyControl->numBufferAllocs, 0);
 	}
-	LWLockRelease(BufFreelistLock);
 	return result;
 }
 
@@ -306,9 +318,7 @@ StrategyNotifyBgWriter(Latch *bgwriterLatch)
 	 * atomic to StrategyGetBuffer.  The bgwriter should call this rather
 	 * infrequently, so there's no performance penalty from being safe.
 	 */
-	LWLockAcquire(BufFreelistLock, LW_EXCLUSIVE);
 	StrategyControl->bgwriterLatch = bgwriterLatch;
-	LWLockRelease(BufFreelistLock);
 }
 
 
@@ -378,7 +388,6 @@ StrategyInitialize(bool init)
 		 * assume it was previously set up by InitBufferPool().
 		 */
 		StrategyControl->firstFreeBuffer = 0;
-		StrategyControl->lastFreeBuffer = NBuffers - 1;
 
 		/* Initialize the clock sweep pointer */
 		StrategyControl->nextVictimBuffer = 0;
diff --git a/src/include/storage/atomic.h b/src/include/storage/atomic.h
new file mode 100644
index 0000000..359d940
--- /dev/null
+++ b/src/include/storage/atomic.h
@@ -0,0 +1,63 @@
+/*-------------------------------------------------------------------------
+ *
+ * atomic.h
+ *	  Atomic memory operations.
+ *
+ * Portions Copyright (c) 1996-2012, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/storage/atomic.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef ATOMIC_H
+#define ATOMIC_H
+
+#include "storage/barrier.h"
+
+#if defined(__GNUC__)
+
+#define pg_atomic_fetch_and_add(dst, src, value) \
+	do { dst = __sync_fetch_and_add(&(src), (value)); } while(0)
+
+#define pg_atomic_sub(var, value) \
+	do { __sync_fetch_and_sub(&(var), (value)); } while(0)
+
+#define pg_atomic_inc(var) do { __sync_fetch_and_add(&(var), 1); } while(0)
+
+#define pg_atomic_fetch_and_set(dst, src, value) \
+	do { dst = __sync_lock_test_and_set(&(src), (value)); } while (0)
+
+#endif
+
+#if !defined(pg_atomic_fetch_and_add)
+#define pg_atomic_fetch_and_add(dst, src, value) \
+	do { S_LOCK(&dummy_spinlock); \
+	dst = src; \
+	src += value; \
+	S_UNLOCK(&dummy_spinlock); } while (0)
+#endif
+
+#if !defined(pg_atomic_sub)
+#define pg_atomic_sub(var, value) \
+	do { S_LOCK(&dummy_spinlock); \
+	var -= value; \
+	S_UNLOCK(&dummy_spinlock); } while (0)
+#endif
+
+#if !defined(pg_atomic_inc)
+#define pg_atomic_inc(var) \
+	do { S_LOCK(&dummy_spinlock); \
+	var++; \
+	S_UNLOCK(&dummy_spinlock); } while (0)
+#endif
+
+#if !defined(pg_atomic_fetch_and_set)
+#define pg_atomic_fetch_and_set(dst, src, value) \
+	do { S_LOCK(&dummy_spinlock); \
+	dst = src; \
+	src = value; \
+	S_UNLOCK(&dummy_spinlock); } while (0)
+#endif
+
+#endif   /* ATOMIC_H */