Initial MVCC code.

New code for locking buffer' context.

src/backend/storage/buffer/buf_init.c

@@ -7,7 +7,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/storage/buffer/buf_init.c,v 1.19 1998/09/01 04:31:39 momjian Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/storage/buffer/buf_init.c,v 1.20 1998/12/15 12:46:18 vadim Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -62,13 +62,13 @@ BufferBlock BufferBlocks;
 
 #ifndef HAS_TEST_AND_SET
 long	   *NWaitIOBackendP;
-
 #endif
 
 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 */
 
@@ -146,21 +146,6 @@ InitBufferPool(IPCKey key)
 				foundDescs;
 	int			i;
 
-	/* check padding of BufferDesc and BufferHdr */
-
-	/*
-	 * we need both checks because a sbufdesc_padded >
-	 * PADDED_SBUFDESC_SIZE will shrink sbufdesc to the required size,
-	 * which is bad
-	 */
-	if (sizeof(struct sbufdesc) != PADDED_SBUFDESC_SIZE ||
-		sizeof(struct sbufdesc_unpadded) > PADDED_SBUFDESC_SIZE)
-		elog(ERROR, "Internal error:  sbufdesc does not have the proper size, "
-			 "contact the Postgres developers");
-	if (sizeof(struct sbufdesc_unpadded) <= PADDED_SBUFDESC_SIZE / 2)
-		elog(ERROR, "Internal error:  sbufdesc is greatly over-sized, "
-			 "contact the Postgres developers");
-
 	Data_Descriptors = NBuffers;
 	Free_List_Descriptor = Data_Descriptors;
 	Lookup_List_Descriptor = Data_Descriptors + 1;
@@ -232,6 +217,7 @@ InitBufferPool(IPCKey key)
 			buf->buf_id = i;
 #ifdef HAS_TEST_AND_SET
 			S_INIT_LOCK(&(buf->io_in_progress_lock));
+			S_INIT_LOCK(&(buf->cntx_lock));
 #endif
 		}
 
@@ -252,10 +238,15 @@ InitBufferPool(IPCKey key)
 
 		WaitIOSemId = IpcSemaphoreCreate(IPCKeyGetWaitIOSemaphoreKey(key),
 										 1, IPCProtection, 0, 1, &status);
+		WaitCLSemId = IpcSemaphoreCreate(IPCKeyGetWaitCLSemaphoreKey(key),
+										 1, IPCProtection, 
+										 IpcSemaphoreDefaultStartValue, 
+										 1, &status);
 	}
 #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));
 }
 

src/backend/storage/buffer/bufmgr.c

@@ -7,7 +7,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/storage/buffer/bufmgr.c,v 1.44 1998/10/08 18:29:54 momjian Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/storage/buffer/bufmgr.c,v 1.45 1998/12/15 12:46:19 vadim Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -86,7 +86,6 @@ static void WaitIO(BufferDesc *buf, SPINLOCK spinlock);
 #ifndef HAS_TEST_AND_SET
 static void SignalIO(BufferDesc *buf);
 extern long *NWaitIOBackendP;	/* defined in buf_init.c */
-
 #endif	 /* HAS_TEST_AND_SET */
 
 static Buffer ReadBufferWithBufferLock(Relation relation, BlockNumber blockNum,
@@ -583,7 +582,6 @@ BufferAlloc(Relation reln,
 					if (buf->refcount > 1)
 						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;
@@ -1096,6 +1094,7 @@ WaitIO(BufferDesc *buf, SPINLOCK spinlock)
 
 #else							/* HAS_TEST_AND_SET */
 IpcSemaphoreId WaitIOSemId;
+IpcSemaphoreId WaitCLSemId;
 
 static void
 WaitIO(BufferDesc *buf, SPINLOCK spinlock)
@@ -1933,3 +1932,147 @@ SetBufferCommitInfoNeedsSave(Buffer buffer)
 	if (!BufferIsLocal(buffer))
 		CommitInfoNeedsSave[buffer - 1]++;
 }
+
+void
+UnlockBuffers()
+{
+	BufferDesc *buf;
+	int			i;
+
+	for (i = 0; i < NBuffers; i++)
+	{
+		if (BufferLocks[i] == 0)
+			continue;
+		
+		Assert(BufferIsValid(i+1));
+		buf = &(BufferDescriptors[i]);
+
+#ifdef HAS_TEST_AND_SET
+		S_LOCK(&(buf->cntx_lock));
+#else
+		IpcSemaphoreLock(WaitCLSemId, 0, IpcExclusiveLock);
+#endif
+
+		if (BufferLocks[i] & BL_R_LOCK)
+		{
+			Assert(buf->r_locks > 0);
+			(buf->r_locks)--;
+		}
+		if (BufferLocks[i] & BL_RI_LOCK)
+		{
+			Assert(buf->ri_lock);
+			buf->ri_lock = false;
+		}
+		if (BufferLocks[i] & BL_W_LOCK)
+		{
+			Assert(buf->w_lock);
+			buf->w_lock = false;
+		}
+#ifdef HAS_TEST_AND_SET
+		S_UNLOCK(&(buf->cntx_lock));
+#else
+		IpcSemaphoreUnlock(WaitCLSemId, 0, IpcExclusiveLock);
+#endif
+		BufferLocks[i] = 0;
+	}
+}
+
+void
+LockBuffer (Buffer buffer, int mode)
+{
+	BufferDesc *buf;
+
+	Assert(BufferIsValid(buffer));
+	if (BufferIsLocal(buffer))
+		return;
+
+	buf = &(BufferDescriptors[buffer-1]);
+
+#ifdef HAS_TEST_AND_SET
+		S_LOCK(&(buf->cntx_lock));
+#else
+		IpcSemaphoreLock(WaitCLSemId, 0, IpcExclusiveLock);
+#endif
+
+	if (mode == BUFFER_LOCK_UNLOCK)
+	{
+		if (BufferLocks[buffer-1] & BL_R_LOCK)
+		{
+			Assert(buf->r_locks > 0);
+			Assert(!(buf->w_lock));
+			Assert(!(BufferLocks[buffer-1] & (BL_W_LOCK | BL_RI_LOCK)))
+			(buf->r_locks)--;
+			BufferLocks[buffer-1] &= ~BL_R_LOCK;
+		}
+		else if (BufferLocks[buffer-1] & BL_W_LOCK)
+		{
+			Assert(buf->w_lock);
+			Assert(buf->r_locks == 0 && !buf->ri_lock);
+			Assert(!(BufferLocks[buffer-1] & (BL_R_LOCK | BL_RI_LOCK)))
+			buf->w_lock = false;
+			BufferLocks[buffer-1] &= ~BL_W_LOCK;
+		}
+		else
+			elog(ERROR, "UNLockBuffer: buffer %u is not locked", buffer);
+	}
+	else if (mode == BUFFER_LOCK_SHARE)
+	{
+		unsigned	i = 0;
+
+		Assert(!(BufferLocks[buffer-1] & (BL_R_LOCK | BL_W_LOCK | BL_RI_LOCK)));
+		while (buf->ri_lock || buf->w_lock)
+		{
+#ifdef HAS_TEST_AND_SET
+			S_UNLOCK(&(buf->cntx_lock));
+			s_lock_sleep(i++);
+			S_LOCK(&(buf->cntx_lock));
+#else
+			IpcSemaphoreUnlock(WaitCLSemId, 0, IpcExclusiveLock);
+			s_lock_sleep(i++)
+			IpcSemaphoreLock(WaitCLSemId, 0, IpcExclusiveLock);
+#endif
+		}
+		(buf->r_locks)++;
+		BufferLocks[buffer-1] |= BL_R_LOCK;
+	}
+	else if (mode == BUFFER_LOCK_EXCLUSIVE)
+	{
+		unsigned	i = 0;
+		
+		Assert(!(BufferLocks[buffer-1] & (BL_R_LOCK | BL_W_LOCK | BL_RI_LOCK)));
+		while (buf->r_locks > 0 || buf->w_lock)
+		{
+			if (buf->r_locks > 3)
+			{
+				if (!(BufferLocks[buffer-1] & BL_RI_LOCK))
+					BufferLocks[buffer-1] |= BL_RI_LOCK;
+				buf->ri_lock = true;
+			}
+#ifdef HAS_TEST_AND_SET
+			S_UNLOCK(&(buf->cntx_lock));
+			s_lock_sleep(i++);
+			S_LOCK(&(buf->cntx_lock));
+#else
+			IpcSemaphoreUnlock(WaitCLSemId, 0, IpcExclusiveLock);
+			s_lock_sleep(i++)
+			IpcSemaphoreLock(WaitCLSemId, 0, IpcExclusiveLock);
+#endif
+		}
+		buf->w_lock = true;
+		BufferLocks[buffer-1] |= BL_W_LOCK;
+		if (BufferLocks[buffer-1] & BL_RI_LOCK)
+		{
+			buf->ri_lock = false;
+			BufferLocks[buffer-1] &= ~BL_RI_LOCK;
+		}
+	}
+	else
+		elog(ERROR, "LockBuffer: unknown lock mode %d", mode);
+
+#ifdef HAS_TEST_AND_SET
+		S_UNLOCK(&(buf->cntx_lock));
+#else
+		IpcSemaphoreUnlock(WaitCLSemId, 0, IpcExclusiveLock);
+#endif
+
+}

src/backend/storage/buffer/s_lock.c

@@ -7,13 +7,14 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/storage/buffer/Attic/s_lock.c,v 1.12 1998/09/18 17:18:39 momjian Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/storage/buffer/Attic/s_lock.c,v 1.13 1998/12/15 12:46:21 vadim Exp $
  *
  *-------------------------------------------------------------------------
  */
 
 #include <stdio.h>
 #include <sys/time.h>
+#include <unistd.h>
 
 #include "config.h"
 #include "c.h"
@@ -52,6 +53,16 @@ s_lock_stuck(volatile slock_t *lock, const char *file, const int line)
 }
 
 
+void
+s_lock_sleep(unsigned spin)
+{
+	struct timeval delay;
+
+	delay.tv_sec = 0;
+	delay.tv_usec = s_spincycle[spin % S_NSPINCYCLE];
+	(void) select(0, NULL, NULL, NULL, &delay);
+}
+
 
 /*
  * s_lock(lock) - take a spinlock with backoff
@@ -59,15 +70,11 @@ s_lock_stuck(volatile slock_t *lock, const char *file, const int line)
 void
 s_lock(volatile slock_t *lock, const char *file, const int line)
 {
-	int			spins = 0;
+	unsigned	spins = 0;
 
 	while (TAS(lock))
 	{
-		struct timeval delay;
-
-		delay.tv_sec = 0;
-		delay.tv_usec = s_spincycle[spins % S_NSPINCYCLE];
-		(void) select(0, NULL, NULL, NULL, &delay);
+		s_lock_sleep(spins);
 		if (++spins > S_MAX_BUSY)
 		{
 			/* It's been over a minute...  */