Fix failure in CreateCheckPoint on some Alpha boxes --- it's not OK to

assume that TAS() will always succeed the first time, even if the lock
is known to be free.  Also, make sure that code will eventually time out
and report a stuck spinlock, rather than looping forever.  Small cleanups
in s_lock.h, too.

Author: tglsfdc

src/backend/access/transam/xlog.c

@@ -6,7 +6,7 @@
  * Portions Copyright (c) 1996-2000, PostgreSQL, Inc
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $Header: /cvsroot/pgsql/src/backend/access/transam/xlog.c,v 1.45 2000/12/28 13:00:08 vadim Exp $
+ * $Header: /cvsroot/pgsql/src/backend/access/transam/xlog.c,v 1.46 2000/12/29 21:31:21 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -411,7 +411,7 @@ begin:;
 					}
 				}
 			}
-			s_lock_sleep(i++);
+			S_LOCK_SLEEP(&(XLogCtl->insert_lck), i++);
 			if (!TAS(&(XLogCtl->insert_lck)))
 				break;
 		}
@@ -599,17 +599,10 @@ begin:;
 
 	if (updrqst)
 	{
-		for (;;)
-		{
-			if (!TAS(&(XLogCtl->info_lck)))
-			{
-				if (XLByteLT(XLogCtl->LgwrRqst.Write, LgwrRqst.Write))
-					XLogCtl->LgwrRqst.Write = LgwrRqst.Write;
-				S_UNLOCK(&(XLogCtl->info_lck));
-				break;
-			}
-			s_lock_sleep(i++);
-		}
+		S_LOCK(&(XLogCtl->info_lck));
+		if (XLByteLT(XLogCtl->LgwrRqst.Write, LgwrRqst.Write))
+			XLogCtl->LgwrRqst.Write = LgwrRqst.Write;
+		S_UNLOCK(&(XLogCtl->info_lck));
 	}
 
 	END_CRIT_CODE;
@@ -622,7 +615,7 @@ XLogFlush(XLogRecPtr record)
 	XLogRecPtr	WriteRqst;
 	char		buffer[BLCKSZ];
 	char	   *usebuf = NULL;
-	unsigned	i = 0;
+	unsigned	spins = 0;
 	bool		force_lgwr = false;
 
 	if (XLOG_DEBUG)
@@ -715,7 +708,7 @@ XLogFlush(XLogRecPtr record)
 				break;
 			}
 		}
-		s_lock_sleep(i++);
+		S_LOCK_SLEEP(&(XLogCtl->lgwr_lck), spins++);
 	}
 
 	if (logFile >= 0 && (LgwrResult.Write.xlogid != logId ||
@@ -740,18 +733,12 @@ XLogFlush(XLogRecPtr record)
 			 logId, logSeg);
 	LgwrResult.Flush = LgwrResult.Write;
 
-	for (i = 0;;)
-	{
-		if (!TAS(&(XLogCtl->info_lck)))
-		{
-			XLogCtl->LgwrResult = LgwrResult;
-			if (XLByteLT(XLogCtl->LgwrRqst.Write, LgwrResult.Write))
-				XLogCtl->LgwrRqst.Write = LgwrResult.Write;
-			S_UNLOCK(&(XLogCtl->info_lck));
-			break;
-		}
-		s_lock_sleep(i++);
-	}
+	S_LOCK(&(XLogCtl->info_lck));
+	XLogCtl->LgwrResult = LgwrResult;
+	if (XLByteLT(XLogCtl->LgwrRqst.Write, LgwrResult.Write))
+		XLogCtl->LgwrRqst.Write = LgwrResult.Write;
+	S_UNLOCK(&(XLogCtl->info_lck));
+
 	XLogCtl->Write.LgwrResult = LgwrResult;
 
 	S_UNLOCK(&(XLogCtl->lgwr_lck));
@@ -767,6 +754,7 @@ GetFreeXLBuffer()
 	XLogCtlInsert *Insert = &XLogCtl->Insert;
 	XLogCtlWrite *Write = &XLogCtl->Write;
 	uint16		curridx = NextBufIdx(Insert->curridx);
+	unsigned	spins = 0;
 
 	LgwrRqst.Write = XLogCtl->xlblocks[Insert->curridx];
 	for (;;)
@@ -809,9 +797,8 @@ GetFreeXLBuffer()
 			InitXLBuffer(curridx);
 			return;
 		}
+		S_LOCK_SLEEP(&(XLogCtl->lgwr_lck), spins++);
 	}
-
-	return;
 }
 
 static void
@@ -820,7 +807,6 @@ XLogWrite(char *buffer)
 	XLogCtlWrite *Write = &XLogCtl->Write;
 	char	   *from;
 	uint32		wcnt = 0;
-	int			i = 0;
 	bool		usexistent;
 
 	for (; XLByteLT(LgwrResult.Write, LgwrRqst.Write);)
@@ -919,18 +905,12 @@ XLogWrite(char *buffer)
 		LgwrResult.Flush = LgwrResult.Write;
 	}
 
-	for (;;)
-	{
-		if (!TAS(&(XLogCtl->info_lck)))
-		{
-			XLogCtl->LgwrResult = LgwrResult;
-			if (XLByteLT(XLogCtl->LgwrRqst.Write, LgwrResult.Write))
-				XLogCtl->LgwrRqst.Write = LgwrResult.Write;
-			S_UNLOCK(&(XLogCtl->info_lck));
-			break;
-		}
-		s_lock_sleep(i++);
-	}
+	S_LOCK(&(XLogCtl->info_lck));
+	XLogCtl->LgwrResult = LgwrResult;
+	if (XLByteLT(XLogCtl->LgwrRqst.Write, LgwrResult.Write))
+		XLogCtl->LgwrRqst.Write = LgwrResult.Write;
+	S_UNLOCK(&(XLogCtl->info_lck));
+
 	Write->LgwrResult = LgwrResult;
 }
 
@@ -2062,18 +2042,17 @@ CreateCheckPoint(bool shutdown)
 	uint32		_logId;
 	uint32		_logSeg;
 	char		archdir[MAXPGPATH];
+	unsigned	spins = 0;
 
 	if (MyLastRecPtr.xrecoff != 0)
 		elog(ERROR, "CreateCheckPoint: cannot be called inside transaction block");
 
 	START_CRIT_CODE;
+
+	/* Grab lock, using larger than normal sleep between tries (1 sec) */
 	while (TAS(&(XLogCtl->chkp_lck)))
 	{
-		struct timeval delay = {2, 0};
-
-		if (shutdown)
-			elog(STOP, "Checkpoint lock is busy while data base is shutting down");
-		(void) select(0, NULL, NULL, NULL, &delay);
+		S_LOCK_SLEEP_INTERVAL(&(XLogCtl->chkp_lck), spins++, 1000000);
 	}
 
 	memset(&checkPoint, 0, sizeof(checkPoint));
@@ -2087,14 +2066,7 @@ CreateCheckPoint(bool shutdown)
 	checkPoint.Shutdown = shutdown;
 
 	/* Get REDO record ptr */
-	while (TAS(&(XLogCtl->insert_lck)))
-	{
-		struct timeval delay = {1, 0};
-
-		if (shutdown)
-			elog(STOP, "XLog insert lock is busy while data base is shutting down");
-		(void) select(0, NULL, NULL, NULL, &delay);
-	}
+	S_LOCK(&(XLogCtl->insert_lck));
 	freespace = ((char *) Insert->currpage) + BLCKSZ - Insert->currpos;
 	if (freespace < SizeOfXLogRecord)
 	{

src/backend/storage/buffer/bufmgr.c

@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/storage/buffer/bufmgr.c,v 1.100 2000/12/28 13:00:21 vadim Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/storage/buffer/bufmgr.c,v 1.101 2000/12/29 21:31:21 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -1987,7 +1987,7 @@ LockBuffer(Buffer buffer, int mode)
 		while (buf->ri_lock || buf->w_lock)
 		{
 			S_UNLOCK(&(buf->cntx_lock));
-			s_lock_sleep(i++);
+			S_LOCK_SLEEP(&(buf->cntx_lock), i++);
 			S_LOCK(&(buf->cntx_lock));
 		}
 		(buf->r_locks)++;
@@ -2013,7 +2013,7 @@ LockBuffer(Buffer buffer, int mode)
 				buf->ri_lock = true;
 			}
 			S_UNLOCK(&(buf->cntx_lock));
-			s_lock_sleep(i++);
+			S_LOCK_SLEEP(&(buf->cntx_lock), i++);
 			S_LOCK(&(buf->cntx_lock));
 		}
 		buf->w_lock = true;

src/backend/storage/buffer/s_lock.c

@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/storage/buffer/Attic/s_lock.c,v 1.27 2000/12/11 00:49:51 tgl Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/storage/buffer/Attic/s_lock.c,v 1.28 2000/12/29 21:31:20 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -25,19 +25,21 @@
  * number of microseconds to wait. This accomplishes pseudo random back-off.
  * Values are not critical but 10 milliseconds is a common platform
  * granularity.
- * note: total time to cycle through all 16 entries might be about .07 sec.
+ *
+ * Total time to cycle through all 20 entries might be about .07 sec,
+ * so the given value of S_MAX_BUSY results in timeout after ~70 sec.
  */
 #define S_NSPINCYCLE	20
 #define S_MAX_BUSY		1000 * S_NSPINCYCLE
 
 int			s_spincycle[S_NSPINCYCLE] =
-{0, 0, 0, 0, 10000, 0, 0, 0, 10000, 0,
+{	0, 0, 0, 0, 10000, 0, 0, 0, 10000, 0,
 	0, 10000, 0, 0, 10000, 0, 10000, 0, 10000, 10000
 };
 
 
 /*
- * s_lock_stuck(lock) - complain about a stuck spinlock
+ * s_lock_stuck() - complain about a stuck spinlock
  */
 static void
 s_lock_stuck(volatile slock_t *lock, const char *file, const int line)
@@ -52,13 +54,38 @@ s_lock_stuck(volatile slock_t *lock, const char *file, const int line)
 }
 
 
+/*
+ * s_lock_sleep() - sleep a pseudo-random amount of time, check for timeout
+ *
+ * Normally 'microsec' is 0, specifying to use the next s_spincycle[] value.
+ * Some callers may pass a nonzero interval, specifying to use exactly that
+ * delay value rather than a pseudo-random delay.
+ */
 void
-s_lock_sleep(unsigned spin)
+s_lock_sleep(unsigned spins, int microsec,
+			 volatile slock_t *lock,
+			 const char *file, const int line)
 {
 	struct timeval delay;
+	unsigned	max_spins;
+
+	if (microsec > 0)
+	{
+		delay.tv_sec = 0;
+		delay.tv_usec = microsec;
+		/* two-minute timeout in this case */
+		max_spins = 120000000 / microsec;
+	}
+	else
+	{
+		delay.tv_sec = 0;
+		delay.tv_usec = s_spincycle[spins % S_NSPINCYCLE];
+		max_spins = S_MAX_BUSY;
+	}
+
+	if (spins > max_spins)
+		s_lock_stuck(lock, file, line);
 
-	delay.tv_sec = 0;
-	delay.tv_usec = s_spincycle[spin % S_NSPINCYCLE];
 	(void) select(0, NULL, NULL, NULL, &delay);
 }
 
@@ -71,14 +98,13 @@ s_lock(volatile slock_t *lock, const char *file, const int line)
 {
 	unsigned	spins = 0;
 
+	/*
+	 * If you are thinking of changing this code, be careful.  This same
+	 * loop logic is used in other places that call TAS() directly.
+	 */
 	while (TAS(lock))
 	{
-		s_lock_sleep(spins);
-		if (++spins > S_MAX_BUSY)
-		{
-			/* It's been over a minute...  */
-			s_lock_stuck(lock, file, line);
-		}
+		s_lock_sleep(spins++, 0, lock, file, line);
 	}
 }