openGauss LWLock 相关代码走读

openGauss 2024-12-04

205

一、LWLock 简介

LWLock 在 openGauss 数据库中起着至关重要的作用，它通过提供高效的互斥访问机制，确保了数据库在高并发环境下的稳定性和数据一致性。

本文讲述在 openGauss 中 LWLock 的一些基本的函数和代码细节。

二、LWLock 相关结构

LWLock 的数据结构及定义

typedef struct LWLock {
    uint16 tranche;            * tranche ID */
    pg_atomic_uint64 state;    * state of exlusive/nonexclusive lockers */
    dlist_head waiters;        * list of waiting PGPROCs */
    int tag;                   * information about the target object we protect, decode by LWLockExplainTag. */
#ifdef LOCK_DEBUG
    pg_atomic_uint32 nwaiters; * number of waiters */
    struct PGPROC* owner;      * last exlusive owner of the lock */
#endif
#ifdef ENABLE_THREAD_CHECK
    pg_atomic_uint32 rwlock;
    pg_atomic_uint32 listlock;
#endif
} LWLock;

tranche 相当于是 LWLock 的 id，唯一标记一个 LWLock，可以用来查找某个 LWLock 去查看其状态；
state 是锁的状态值，包含多个标志位；
waiters 是用来记录等待获取 LWLock 的进程号；
nwaiters 是等待的进程数量，调试相关的；
owner 是上一次获取 LWLock 的进程，调试相关的；
rwlock 读写锁，线程调试与检查相关；
listlock 列表锁，用于保护等待列表，线程调试与检查相关。

state 的标志位

#define LW_FLAG_HAS_WAITERS ((uint64)1LU << 30 << 32)
#define LW_FLAG_RELEASE_OK ((uint64)1LU << 29 << 32)
#define LW_FLAG_LOCKED ((uint64)1LU << 28 << 32)


#define LW_VAL_EXCLUSIVE (((uint64)1LU << 24 << 32) + (1LU << 47) + (1LU << 39) + (1LU << 31) + (1LU << 23) + (1LU << 15) + (1LU << 7))
#define LW_VAL_SHARED 1

LW_FLAG_HAS_WAITERS 标记是否有进程在等待，用于快速判断等待列表是否为空；
LW_FLAG_RELEASE_OK 标记是否可以释放锁，即是否可以进行唤醒操作；
LW_FLAG_LOCKED 标记锁已被锁定，用于保护进程列表的并发操作；
LW_VAL_EXCLUSIVE 标记独占锁是否被占用；
LW_VAL_SHARED 标记共享锁是否已获取。

LWLock 的种类/模式

typedef enum LWLockMode {
    LW_EXCLUSIVE,
    LW_SHARED,
    LW_WAIT_UNTIL_FREE /* A special mode used in PGPROC->lwlockMode,
                        * when waiting for lock to become free. Not
                        * to be used as LWLockAcquire argument */
} LWLockMode;

LW_EXCLUSIVE 表示锁为独占模式，当一个进程以独占模式获取锁时，它会阻止其他所有进程（包括需要共享锁的进程）获取同一锁。这种模式通常用于写操作，因为它需要对共享资源进行修改，而这些修改可能会破坏数据的一致性；
LW_SHARED 表示锁为共享模式，在共享模式下，多个进程可以同时获取同一锁，只要它们不与任何需要独占锁的进程冲突。这种模式通常用于读操作，因为多个读操作可以同时进行，而不会相互干扰；
LW_WAIT_UNTIL_FREE 准确来说不是锁的模式，而是一种等待的状态直到锁变为可用状态，当一个进程想要获取锁但锁当前被其他进程持有时，它会设置这个模式，并等待直到锁被释放。

三、LWLock 加锁实现流程

在代码里 LWLockAcquire 函数负责整个加锁过程，这里将过程分为五步：

判断持锁数量有没有达到上限、判断获取的锁模式符不符合要求等加锁前的准备步骤；
尝试获取锁，如果获取成功，直接返回，否则执行第 3 步；
将自身进程添加到等待队列，然后再一次尝试获取锁，如果获取锁成功，则将自身从等待队列中删除并直接返回，否则执行第 4 步；
通过阻塞式获取信号量，若获取到信号量便被唤醒或等待其他进程唤醒，否则继续循环尝试获取信号量；
当因为锁释放被唤醒之后（该进程已经被唤醒进程从等待队列里删除了），回到第 2 步继续执行，直到加锁成功后返回。

加锁前的检查

AssertArg(mode == LW_SHARED || mode == LW_EXCLUSIVE);


Assert(!(proc == NULL && IsUnderPostmaster));


if (t_thrd.storage_cxt.num_held_lwlocks >= MAX_SIMUL_LWLOCKS) {
    ereport(ERROR, (errcode(ERRCODE_LOCK_NOT_AVAILABLE), errmsg("too many LWLocks taken")));
}

上述代码依次为：

检查锁模式是否为共享或独占；
检查进程是否在非共享内存初始化阶段为空；
确保有足够的空间记录锁。

尝试获取锁

static bool LWLockAttemptLock(LWLock *lock, LWLockMode mode)

尝试通过CAS操作来设置 LW_VAL_EXCLUSIVE 或 LW_VAL_SHARED 标志位。LWLockAttemptLock 返回 false 表示成功拿到了锁。返回 true 表示拿锁失败。

加入等待队列

static void LWLockQueueSelf(LWLock *lock, LWLockMode mode)

把当前进程加入到锁的等待队列中包含三个步骤：

使用原子操作和自旋锁对等待队列加锁；
将当前进程加入到等待队列中（同时还需要更新自身进程对应的 PGPORC 实例的 lwWaiting 和 lwWaitingMode 成员）；
使用原子操作对等待队列解锁。

当加入到等待队列后，还需要更新 LW_FLAG_HAS_WAITERS 标记位，表示有进程在等待。

信号量

//加锁之前
extraWaits = 0;
for (;;) {
   //阻塞式获取信号量
   PGSemaphoreLock(proc->sem);
   //被唤醒之后检查唤醒条件
   //如果是锁被释放了，那么 proc->lwWaiting 会是 false
   if (!(proc->lwWaiting)) {
       if(!(proc->lwIsVictim)) {
           break;
       }
       //被动成为牺牲线程的后续操作，修改状态，允许释放等待队列中的线程
       pg_atomic_fetch_or_u64(&lock->state, LW_FLAG_RELEASE_OK);
       LWThreadSuicide(proc, extraWaits, lock, mode);
   }
   extraWaits++;
}


//加到锁之后
//因为刚刚占用了多余的唤醒，所以需要进行补偿
while(extraWaits-- > 0) {
   PGSemaphoreUnlock(proc->sem);
}

退出等待队列

static void LWLockDequeueSelf(LWLock *lock, LWLockMode mode)

当加锁成功后，如果自身在等待队列中则将其删除掉，然后若队列为空，则需要清除持锁标记位（LW_FLAG_HAS_WAITERS）。重要的是在执行上述操作时需要对等待队列加锁后进行。

四、LWLock 放锁实现流程

在代码里 LWLockRelease 函数负责整个放锁流程，这里将过程分为：

检查持锁信息，若没有找到要放的锁则报错，否则继续执行；
清除锁标记位，如果之前占用的是独占锁，那么清除 LW_VAL_EXCLUSIVE 标志位，如果是共享锁，那么共享锁数量减一。同时持锁数量也要减一；
检查 LW_FLAG_HAS_WAITERS 和 LW_FLAG_RELEASE_OK 标志位，如果都设置了并且现在独占、共享锁都没有被占用，那么需要执行唤醒操作。

放锁前的检查

for (i = t_thrd.storage_cxt.num_held_lwlocks; --i >= 0;) {
    if (lock == t_thrd.storage_cxt.held_lwlocks[i].lock) {
        mode = t_thrd.storage_cxt.held_lwlocks[i].mode;
        break;
    }
}
if (i < 0) {
    ereport(ERROR, (errcode(ERRCODE_LOCK_NOT_AVAILABLE), errmsg("lock %s is not held", T_NAME(lock))));
}

检查持锁信息，有没有要放的锁，若有则将锁模式赋给 mode ；若没有（i < 0）则停止操作并打印锁没有被持有的报错信息。

放锁

t_thrd.storage_cxt.num_held_lwlocks--;
for (; i < t_thrd.storage_cxt.num_held_lwlocks; i++) {
    t_thrd.storage_cxt.held_lwlocks[i] = t_thrd.storage_cxt.held_lwlocks[i + 1];
    t_thrd.storage_cxt.lwlock_held_times[i] = t_thrd.storage_cxt.lwlock_held_times[i + 1];
}


if (mode == LW_EXCLUSIVE) {
    TsAnnotateRWLockReleased(&lock->rwlock, 1);
    oldstate = pg_atomic_sub_fetch_u64(&lock->state, LW_VAL_EXCLUSIVE);
} else {
    TsAnnotateRWLockReleased(&lock->rwlock, 0);
    oldstate = __sync_sub_and_fetch(&lock->state, LOCK_REFCOUNT_ONE_BY_THREADID);
}