PostgreSQL数据库锁机制——SpinLock

今天药忘吃喽~ 2022-09-11 13:18 171阅读 0赞

SpinLock是最底层的锁，使用**互斥信号量**实现，与操作系统和硬件环境联系紧密。SpinLock分为与机器相关的实现方法（定义在s\_lock.c中）和与机器不相关的实现方法（定义在Spin.c中）。SpinLock的特点是：封锁时间很短，没有等待队列和死锁检测机制，事务结束时不能自动释放SpinLock。  
作为一种最底层的锁，一般不直接使用SpinLock，而是利用它来实现其他锁（LWLock）。毫无疑问，依赖于硬件的SpinLock机制肯定比不依赖于硬件的SpinLock机制速度快。因为不依赖于硬件的SpinLock机制需要使用PG信号量来仿真SpinLock。如果机器拥有TAS(test-and-set)指令集，那么PostgreSQL会使用s\_lock.h和s\_lock.c中定义的SpinLock实现机制。如果机器没有TAS指令集，那么不依赖于硬件的SpinLock的实现定义在Spin.c中。需要用到PostgreSQL定义的信号量PGSemaphore。

## 依赖于硬件的SpinLock的实现定义 ##

src/include/storage/s\_lock.h

### 对于支持的平台需要提供如下的硬件相关的宏 ###

void S_INIT_LOCK(slock_t *lock) // 初始化一个spinlck（unlocked状态）
    int S_LOCK(slock_t *lock) // 等待获取一个spinlock，在合理时间内不能获取lock则超时并abort，返回delays的数量
    void S_UNLOCK(slock_t *lock) // 解锁之前获取的锁
    bool S_LOCK_FREE(slock_t *lock) // Tests if the lock is free. Returns true if free, false if locked
    void SPIN_DELAY(void) // spinlock等待循环使用的延时操作

s\_lock.h

#### i386 ####

TEST\_AND\_SET

typedef unsigned char slock_t;
    #define TAS(lock) tas(lock)
    static __inline__ int tas(volatile slock_t *lock) { 
    	register slock_t _res = 1;
    	/* Use a non-locking test before asserting the bus lock. Note that the * extra test appears to be a small loss on some x86 platforms and a small * win on others; it's by no means clear that we should keep it. * When this was last tested, we didn't have separate TAS() and TAS_SPIN() * macros. Nowadays it probably would be better to do a non-locking test * in TAS_SPIN() but not in TAS(), like on x86_64, but no-one's done the * testing to verify that. Without some empirical evidence, better to * leave it alone. */
    	__asm__ __volatile__(
    		" cmpb $0,%1 \n"
    		" jne 1f \n"
    		" lock \n"
    		" xchgb %0,%1 \n"
    		"1: \n"
    :		"+q"(_res), "+m"(*lock)
    :		/* no inputs */
    :		"memory", "cc");
    	return (int) _res;
    }

spin delay

#define SPIN_DELAY() spin_delay()
    static __inline__ void spin_delay(void) { 
    	/* This sequence is equivalent to the PAUSE instruction ("rep" is * ignored by old IA32 processors if the following instruction is * not a string operation); the IA-32 Architecture Software * Developer's Manual, Vol. 3, Section 7.7.2 describes why using * PAUSE in the inner loop of a spin lock is necessary for good * performance: * The PAUSE instruction improves the performance of IA-32 * processors supporting Hyper-Threading Technology when * executing spin-wait loops and other routines where one * thread is accessing a shared lock or semaphore in a tight * polling loop. When executing a spin-wait loop, the * processor can suffer a severe performance penalty when * exiting the loop because it detects a possible memory order * violation and flushes the core processor's pipeline. The * PAUSE instruction provides a hint to the processor that the * code sequence is a spin-wait loop. The processor uses this * hint to avoid the memory order violation and prevent the * pipeline flush. In addition, the PAUSE instruction * de-pipelines the spin-wait loop to prevent it from * consuming execution resources excessively. */
    	__asm__ __volatile__(
    		" rep; nop \n");
    }

#### x86-64 ####

TEST\_AND\_SET

typedef unsigned char slock_t;
    #define TAS(lock) tas(lock)
    /* On Intel EM64T, it's a win to use a non-locking test before the xchg proper, * but only when spinning. * See also Implementing Scalable Atomic Locks for Multi-Core Intel(tm) EM64T * and IA32, by Michael Chynoweth and Mary R. Lee. As of this writing, it is * available at: * http://software.intel.com/en-us/articles/implementing-scalable-atomic-locks-for-multi-core-intel-em64t-and-ia32-architectures */
    #define TAS_SPIN(lock) (*(lock) ? 1 : TAS(lock))
    static __inline__ int tas(volatile slock_t *lock) { 
    	register slock_t _res = 1;
    	__asm__ __volatile__(
    		" lock \n"
    		" xchgb %0,%1 \n"
    :		"+q"(_res), "+m"(*lock)
    :		/* no inputs */
    :		"memory", "cc");
    	return (int) _res;
    }

spin delay

#define SPIN_DELAY() spin_delay()
    static __inline__ void spin_delay(void) { 
    	/* Adding a PAUSE in the spin delay loop is demonstrably a no-op on * Opteron, but it may be of some use on EM64T, so we keep it. */
    	__asm__ __volatile__(" rep; nop \n");
    }

#### ia64 ####

#define HAS_TEST_AND_SET
    typedef unsigned int slock_t;
    #define TAS(lock) tas(lock)
    /* On IA64, it's a win to use a non-locking test before the xchg proper */
    #define TAS_SPIN(lock) (*(lock) ? 1 : TAS(lock))
    #ifndef __INTEL_COMPILER
    static __inline__ int tas(volatile slock_t *lock) { 
    	long int	ret;
    	__asm__ __volatile__(
    		" xchg4 %0=%1,%2 \n"
    :		"=r"(ret), "+m"(*lock)
    :		"r"(1)
    :		"memory");
    	return (int) ret;
    }
    #else /* __INTEL_COMPILER */
    static __inline__ int tas(volatile slock_t *lock) { 
    	int		ret;
    	ret = _InterlockedExchange(lock,1);	/* this is a xchg asm macro */
    	return ret;
    }
    /* icc can't use the regular gcc S_UNLOCK() macro either in this case */
    #define S_UNLOCK(lock) do { __memory_barrier(); *(lock) = 0; } while (0)
    #endif /* __INTEL_COMPILER */

#### ARM ####

参考代码

#### S/390 and S/390x Linux (32- and 64-bit zSeries) ####

参考代码

#### Sparc ####

参考代码

#### PowerPC ####

参考代码

### 使用信号量实现的spinlock ###

typedef int slock_t;
    extern bool s_lock_free_sema(volatile slock_t *lock);
    extern void s_unlock_sema(volatile slock_t *lock);
    extern void s_init_lock_sema(volatile slock_t *lock, bool nested);
    extern int	tas_sema(volatile slock_t *lock);
    #define S_LOCK_FREE(lock) s_lock_free_sema(lock)
    #define S_UNLOCK(lock) s_unlock_sema(lock)
    #define S_INIT_LOCK(lock) s_init_lock_sema(lock, false)
    #define TAS(lock) tas_sema(lock)

### 其他函数 ###

extern slock_t dummy_spinlock;
    /* Platform-independent out-of-line support routines */
    extern int s_lock(volatile slock_t *lock, const char *file, int line, const char *func);
    /* Support for dynamic adjustment of spins_per_delay */
    #define DEFAULT_SPINS_PER_DELAY 100
    extern void set_spins_per_delay(int shared_spins_per_delay);
    extern int	update_spins_per_delay(int shared_spins_per_delay);
    /* Support for spin delay which is useful in various places where spinlock-like procedures take place. */
    typedef struct { 
    	int			spins;
    	int			delays;
    	int			cur_delay;
    	const char *file;
    	int			line;
    	const char *func;
    } SpinDelayStatus;
    static inline void init_spin_delay(SpinDelayStatus *status, const char *file, int line, const char *func) { 
    	status->spins = 0;
    	status->delays = 0;
    	status->cur_delay = 0;
    	status->file = file;
    	status->line = line;
    	status->func = func;
    }
    #define init_local_spin_delay(status) init_spin_delay(status, __FILE__, __LINE__, PG_FUNCNAME_MACRO)
    void perform_spin_delay(SpinDelayStatus *status);
    void finish_spin_delay(SpinDelayStatus *status);

### s\_lock.c ###

/* s_lock(lock) - platform-independent portion of waiting for a spinlock. */
    int s_lock(volatile slock_t *lock, const char *file, int line, const char *func) { 
    	SpinDelayStatus delayStatus;
    	init_spin_delay(&delayStatus, file, line, func);
    	while (TAS_SPIN(lock)) { 
    		perform_spin_delay(&delayStatus);
    	}
    	finish_spin_delay(&delayStatus);
    	return delayStatus.delays;
    }
    
    /* Wait while spinning on a contended spinlock. */
    void perform_spin_delay(SpinDelayStatus *status) { 
    	/* CPU-specific delay each time through the loop */
    	SPIN_DELAY();
    	/* Block the process every spins_per_delay tries */
    	if (++(status->spins) >= spins_per_delay) { 
    		if (++(status->delays) > NUM_DELAYS)
    			s_lock_stuck(status->file, status->line, status->func);
    		if (status->cur_delay == 0) /* first time to delay? */
    			status->cur_delay = MIN_DELAY_USEC;
    		pg_usleep(status->cur_delay);
    #if defined(S_LOCK_TEST)
    		fprintf(stdout, "*");
    		fflush(stdout);
    #endif
    		/* increase delay by a random fraction between 1X and 2X */
    		status->cur_delay += (int) (status->cur_delay * ((double) random() / (double) MAX_RANDOM_VALUE) + 0.5);
    		/* wrap back to minimum delay when max is exceeded */
    		if (status->cur_delay > MAX_DELAY_USEC) status->cur_delay = MIN_DELAY_USEC;
    		status->spins = 0;
    	}
    }

## 不依赖于硬件的SpinLock的实现定义(Hardware-independent implementation of spinlocks) ##

使用信号量PGSemaphore实现 src/backend/storage/lmgr/spin.c  
如下为最重要的宏定义：

#define SpinLockInit(lock) S_INIT_LOCK(lock)
    #define SpinLockAcquire(lock) S_LOCK(lock)
    #define SpinLockRelease(lock) S_UNLOCK(lock)
    #define SpinLockFree(lock) S_LOCK_FREE(lock)

申请的共享内存的大小

#ifndef HAVE_ATOMICS
    #define NUM_EMULATION_SEMAPHORES (NUM_SPINLOCK_SEMAPHORES + NUM_ATOMICS_SEMAPHORES)
    #else
    #define NUM_EMULATION_SEMAPHORES (NUM_SPINLOCK_SEMAPHORES)
    #endif /* DISABLE_ATOMICS */
    PGSemaphore *SpinlockSemaArray;
    #else /* !HAVE_SPINLOCKS */
    #define NUM_EMULATION_SEMAPHORES 0
    #endif
    Size SpinlockSemaSize(void) { 
    	return NUM_EMULATION_SEMAPHORES * sizeof(PGSemaphore);
    }

信号量相关函数

void SpinlockSemaInit(void) { 
    	PGSemaphore *spinsemas;
    	int			nsemas = SpinlockSemas();
    	int			i;
    	/* We must use ShmemAllocUnlocked(), since the spinlock protecting ShmemAlloc() obviously can't be ready yet. */
    	spinsemas = (PGSemaphore *) ShmemAllocUnlocked(SpinlockSemaSize());
    	for (i = 0; i < nsemas; ++i) spinsemas[i] = PGSemaphoreCreate();
    	SpinlockSemaArray = spinsemas;
    }
    void s_init_lock_sema(volatile slock_t *lock, bool nested) { 
    	static uint32 counter = 0;
    	uint32		offset;
    	uint32		sema_total;
    	uint32		idx;
    	if (nested) { 
    		/* To allow nesting atomics inside spinlocked sections, use a different spinlock. See comment above. */
    		offset = 1 + NUM_SPINLOCK_SEMAPHORES;
    		sema_total = NUM_ATOMICS_SEMAPHORES;
    	}else{ 
    		offset = 1;
    		sema_total = NUM_SPINLOCK_SEMAPHORES;
    	}
    	idx = (counter++ % sema_total) + offset;
    	/* double check we did things correctly */
    	s_check_valid(idx);
    	*lock = idx;
    }
    
    void s_unlock_sema(volatile slock_t *lock) { 
    	int			lockndx = *lock;
    	s_check_valid(lockndx);
    	PGSemaphoreUnlock(SpinlockSemaArray[lockndx - 1]);
    }
    
    bool s_lock_free_sema(volatile slock_t *lock) { 
    	/* We don't currently use S_LOCK_FREE anyway */
    	elog(ERROR, "spin.c does not support S_LOCK_FREE()");
    	return false;
    }
    
    int tas_sema(volatile slock_t *lock) { 	
        int			lockndx = *lock;
    	s_check_valid(lockndx);
    	/* Note that TAS macros return 0 if *success* */
    	return !PGSemaphoreTryLock(SpinlockSemaArray[lockndx - 1]);
    }

不依赖于硬件的SpinLock的实现定义相关函数