PostgreSQL 并发控制 -- 锁体系(spinlock,lwlock,regular lock)实现原理

文章目录

前言
锁类型
Spinlocks
- 信号量方式实现的Spinlock
- - 初始化 SpinLockInit(lock)
  - 加锁 SpinLockAcquire(lock)
  - 解锁 SpinLockRelease(lock)
- TAS 指令方式实现的 Spinlock
- - 初始化 SpinLockInit(lock)
  - 加锁 SpinLockAcquire(lock)
- Linux kernel Spinlock 实现
LWLocks
- InitializeLWLocks 初始化轻量锁
- LWLockAcquire 加锁
- LWLockRelease 解锁
Regular Locks
- 锁模式以及相容性
- 数据结构
- 初始化常规锁
- 加锁 LockAcquire
- 解锁 LockRelease
总结

前言

本节将在之前PG 事务体系实现的基础上记录 PostgreSQL 实现事务过程的一个非常重要的子系统：锁。它是 PG 实现事务的核心系统，为了更好得提升并发场景下的事务可靠性以及性能而存在。

本节的PG代码版本是：REL_12_2，篇幅会比较长，可能会对比不同系统的一些锁实现细节，希望大家能够对锁体系的实现有广度以及深度的系统理解（当然也是自我学习的过程），有一些代码细节理解有问题的情况也希望熟悉的同学及时指出。

锁类型

PG 内部使用了四种类型的锁，自低向上分别是 Spinlocks, LightWeigt locks (LWLocks),Regular locks,SIReadLock predicate locks。

Spinlocks 自旋锁。是用来保护一个非常小的临界区资源，所以它本身适用的持续周期是非常短，而这种锁的实现方式就是占有一个CPU核心持续自旋（自我轮询）。目前来看，自旋锁的实现以及应用遍布整个互联网体系，尤其是底层的基础架构 — 数据库系统，存储系统以及 OS 系统。PG 因为起源于上世纪70年代，当时的互联网也才刚兴起没多久，所以PG的 spinlock 也都是自己实现的（使用了硬件的 atomic-test-and-set 指令），并没有使用 os 现在提供的 spinlock。

在PG内部，spinlock 被用来实现 LWLocks，保护一些原子变量。但是 spinlock 并不会提供死锁检测、持锁期间抛异常无法自动释放锁等机制。但是还是实现了超时机制，即长时间得不到锁，就让出CPU。
LightWeight locks 轻量锁。用来保护一些存储在共享内存中的数据结构。支持两种模式：排他(读/写)和共享(读读)。当然，LWLocks 也不支持死锁检测和超时（底层的spinlock 已经支持了），但是能够在持锁期间抛错误时自动释放锁资源。

它底层的实现机制也是spinlock，等待锁是通过让进程等待在一个 OS 信号量上（不会消耗CPU），这个信号量被持有进程释放之后，则会根据等待顺序来让进程按照顺序加锁。
Regular locks 常规锁。这是PG 非常重要的锁类型，用来保护用户驱动产生的 PG 对象的访问安全。其支持了非常多的锁模式，分别用来保护 PG 对象：表(Relation)，页面(page)，元组(row) 等。

其完整支持了死锁检测以及事务结束时的自动释放锁功能，所以常规锁是 PG事务并发控制中最复杂的一部分，也是我们下文展开锁细节中描述最多的一部分。
SIReadLock predicate locks预测锁。是PG 为了实现 SSI (Serializable and Snapshot Transaction Isolation Level) 隔离级别时使用的锁，这一部分的实现以及 SSI 的介绍会单独放在一个小章节进行描述，因为这个隔离级别并不是 ANSI 标准中的，而是PG 为用户使用方便单独做的一个隔离级别，用来避免写偏序问题。

以上就是PG内部的四种基本的锁类型，接下来我们一起自低向上仔细看看 PG 这样的拥有40多年历史的数据库是如何实现这一些锁的（最后一种本节不会描述）。

Spinlocks

在 PostgreSQL 中 spinlock 的实现有两种方式，一种是依赖信号量，另一种是根据系统是TAS(Test-And-Set 根据系统是否支持来决定是否使用)。

信号量方式实现的Spinlock

所有内核用到的spinlock 相关的接口都在 spin.h中，通过宏定义实现，实际的实现是在 s_lock.h中。

实际使用中，PostgreSQL 内核会建议不要使用信号量实现的 spinlock，因为大多数的 os 会对创建的信号量的个数有限制，除非有必要，建议还是使用 TAS 方式实现的spinlock。

因为信号量有个数限制，

初始化 SpinLockInit(lock)

信号量本身会在 postgres 进程启动的时候预先创建好 spinlock 要使用的固定个数的信号量，方便进程运行过程中快速使用，这一些预先创建好的信号量会保存到SpinlockSemaArray 全局变量中。

预创建指定个数的信号量会在如下逻辑中进行：

PostmasterMainreset_shared()CreateSharedMemoryAndSemaphoresSpinlockSemaInit

预先通过 sem_init系统调用初始化的信号量的个数为 : NUM_SPINLOCK_SEMAPHORES + NUM_ATOMICS_SEMAPHORES = 128 + 64 = 192个。

PG 单机允许的最大的连接数是 100个 backend，很少了，所以目前支持的这么多的信号量肯定是够用，当然逻辑上肯定是要支持超过最大信号量个数的。

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;
}

其中 PGSemaphoreCreate 内部会调用sem_init(sem, 1, 1) 初始化一个新的信号量，并且将这个信号量的值赋值为1 并且设置可以跨线程以及进程可见的标记（第二个参数 pshared）。

启动过程中创建好的这么多信号量在后续有Spinlock 的使用需求时会先初始化spinlock，即通过函数 #define S_INIT_LOCK(lock) s_init_lock_sema(lock, false)进行，这个初始化的目的是标识当前调用者使用的是 SpinlockSemaArray 信号量数组中的哪一个信号量，将index 放在lock中，需要注意的是虽然有192个信号量，但实际让使用的只有 NUM_SPINLOCK_SEMAPHORES 128个。

void
s_init_lock_sema(volatile slock_t *lock, bool nested)
{/* 静态变量，标识当前调用者要使用 SpinlockSemaArray 数组中信号量的 index. */static int    counter = 0;/* 不使用0 号信号量 */*lock = ((++counter) % NUM_SPINLOCK_SEMAPHORES) + 1;
}

加锁 SpinLockAcquire(lock)

调用者进行加锁，会通过逻辑:

#define SpinLockAcquire(lock) S_LOCK(lock)
#define S_LOCK(lock) \(TAS(lock) ? s_lock((lock), __FILE__, __LINE__, PG_FUNCNAME_MACRO) : 0)

宏定义 S_LOCK 主要的作用是加锁，在信号量场景下其加锁步骤为：

通过信号量实现的 TAS(lock) 函数 tas_sema(lock)去等待 lock值对应的信号量数组中的对应信号量可被访问。

操作方式是通过 sem_trywait 将这个从信号量数组中拿到的信号量的值减1 变为0，这样后来的调用者想要对同一个信号量进行访问，因为其value 已经是 0 ，则会返回失败状态。
如果加锁成功(信号初始值大于0，减1之后会返回成功，标识加锁成功)，则会进入s_lock逻辑，从而进行自旋（busy-loop状态）。
这个函数实现也是前面介绍 PG Spinlock 时提到的超时机制的实现，它也是 TAS 方式的spinlock 需要进入的逻辑。

第一步中的 tas_sema函数实现如下：

主要是信号量的系统调用操作，并且要求保证最后执行完 sem_trywait 之后信号量的值为0 才行，这样才能保证后续的其他操作当前信号量的调用者除了解锁操作即调用 sem_post 之外无法持有信号量。

int
tas_sema(volatile slock_t *lock)
{int            lockndx = *lock;if (lockndx <= 0 || lockndx > NUM_SPINLOCK_SEMAPHORES)elog(ERROR, "invalid spinlock number: %d", lockndx);/* Note that TAS macros return 0 if *success* */return !PGSemaphoreTryLock(SpinlockSemaArray[lockndx - 1]);
}bool
PGSemaphoreTryLock(PGSemaphore sema)
{int            errStatus;/** Note: if errStatus is -1 and errno == EINTR then it means we returned* from the operation prematurely because we were sent a signal.  So we* try and lock the semaphore again.*/do{/* * 保证执行完之后，信号量的值变为了0（有效的加锁操作）。* 判断信号量的值是否大于0，是则减一，并返回true。 * 如果内部发现信号量已经是0 了，则会返回false，标识* 已经有一个其他的调用者在持有锁了。*/errStatus = sem_trywait(PG_SEM_REF(sema));} while (errStatus < 0 && errno == EINTR);if (errStatus < 0){if (errno == EAGAIN || errno == EDEADLK)return false;       /* failed to lock it *//* Otherwise we got trouble */elog(FATAL, "sem_trywait failed: %m");}return true;
}

第二步 主要是尝试操作完信号量之后发现已经有一个更早的调用者持有了信号量（其实是信号量为0），通会过 s_lock 进行忙等，且有需要则进入超时逻辑。

忙等以及超时处理的主要逻辑实现是通过一个 SpinDelayStatus 结构体。

typedef struct
{int            spins; /* 自旋的次数，即执行TAS_SPIN(lock)为真的次数 */int         delays; /* 执行pg_usleep 的次数 */int            cur_delay; /* 当前要sleep 的时间，单位是us */const char *file; /* 调用者源代码所在的文件名 */int           line; /* 行号 */const char *func; /*函数名，这三个都是为了方便记录日志。*/
} SpinDelayStatus;

s_lock忙等的前提是 TAS_SPIN(lock)返回为真，即没有获得到锁，才会有如下的逻辑，主要实现是在perform_spin_delay函数中：

spins 次数小于 spins_per_delay 多次(默认是100次) 还没有拿到锁，即TAS_SPIN(lock) 返回值一直为真。

那就继续执行，尝试获取锁。

如果超过了 spins_per_delay：
1. delays 次数小于 NUM_DELAYS(1000次) 且上次的睡眠时间没有设置过，那就设置一个最小的睡眠时间(默认是100us) ，执行pg_usleep，并更新下一次的delay时间为当前的1x或者2x（保证每一次触发delay的时间比当前长，这样较长持有的 spinlock 能减少对CPU的持续的占用）。
2. delay次数超过了 NUM_DELAYS，则标识 spinlock 等待的时间太长了，会elog panic。
3. 每一次进入到超过 spins_per_delay 之后会重置一下 spins 为0。

逻辑如下

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;}
}

当然， PG spin_locks 针对 spins_per_delay 的次数设置还是会动态变更，根据每次加锁是否需要delay，如果不用delay 认为现在加锁需求不是很频繁，则每次 finish 之后会增加100次，直到增加到 MAX_SPINS_PER_DELAY 1000次为止。

解锁 SpinLockRelease(lock)

信号量实现的spinlock的解锁逻辑比较简单了，主要通过操作信号量的 sem_post 完成针对指定信号值的加一操作。

#define SpinLockRelease(lock) S_UNLOCK(lock)
#define S_UNLOCK(lock)   s_unlock_sema(lock)
void
s_unlock_sema(volatile slock_t *lock)
{/* lock 为存储在数组中的信号量index */int         lockndx = *lock;if (lockndx <= 0 || lockndx > NUM_SPINLOCK_SEMAPHORES)elog(ERROR, "invalid spinlock number: %d", lockndx);/* 信号量解锁 */PGSemaphoreUnlock(SpinlockSemaArray[lockndx - 1]);
}void
PGSemaphoreUnlock(PGSemaphore sema)
{int            errStatus;/** Note: if errStatus is -1 and errno == EINTR then it means we returned* from the operation prematurely because we were sent a signal.  So we* try and unlock the semaphore again. Not clear this can really happen,* but might as well cope.*/do{/* 信号量加一 */errStatus = sem_post(PG_SEM_REF(sema));} while (errStatus < 0 && errno == EINTR);if (errStatus < 0)elog(FATAL, "sem_post failed: %m");
}

本来还有一个自旋锁的释放SpinLockFree(lock)，因为目前用到的信号量需要长期驻留在内存中，在PG 主进程退出时统一通过 sem_destroy 进行释放。

可以看到使用信号量实现自旋锁，需要非常多次的系统调用的参与，系统调用意味着有用户态到内核态的上下文切换，这其实是对cpu的一种持续消耗，效率并不高。

接下来我们再看看 PG 内核推荐使用的第二种实现自旋锁的方式 – TAS（test-and-set）。

TAS 指令方式实现的 Spinlock

PG 支持了非常多硬件架构中的自旋锁的实现，因为在硬件服务器架构中不同的原子操作的CPU指令不同，本节我们主要关注 intel x86_64 这个数据库系统所使用的较多的服务器架构。

初始化 SpinLockInit(lock)

TAS 的初始化没有其他的操作，主要是将 lock状态设置为 S_UNLOCK(lock) 时的状态就好了，即设置为 0 就好了。

核心主要是看一下加锁所应用到的指令集。

加锁 SpinLockAcquire(lock)

x86_64架构下支持 TAS 的加锁实现还是会通过如下逻辑：

#if !defined(S_LOCK)
#define S_LOCK(lock) \(TAS(lock) ? s_lock((lock), __FILE__, __LINE__, PG_FUNCNAME_MACRO) : 0)
#endif   /* S_LOCK */

和信号量实现的 spinlock 的差异是在 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;
}

以上实现执行了一段汇编代码：

lock 指令， intel x86_64 的官方文档中的描述（ 8.1.4小节）指出当我们的硬件 cpu型号是 intel486 或者 Pentium处理器，lock 指令会锁内存总线，即使这个变量的内存数据在cpu-cache中，也会锁总线。但是这样对性能的影响非常大，所以后来的 P6 以及更新的处理器对于处于cpu-cache中的内存区域只会锁 cache ，通过 CPU 的多核 cache一致性协议 + 内存屏障来保证访问的原子性，这种方式的lock 指令相比于之前锁总线，对CPU性能提升还是挺明显的。
xchg 指令，用来交换两个寄存器中的值，它的执行是在 lock指令之后
+q(res) 表示指令可以对临时变量 _res进行读写，"+m"(*lock)表示可以对内存变量 *lock的内存区域进行读写。
最后返回 (int)_res的值，如果返回值为 1，则表示加锁失败，如果为0，则表示加锁成功（unlock 会设置 *lock的值为0）。

加完锁之后如果加锁失败，还是会进入到 s_lock的逻辑，我们再次看看这个函数的逻辑：

int
s_lock(volatile slock_t *lock, const char *file, int line, const char *func)
{SpinDelayStatus delayStatus;/* 初始化 SpinDelayStatus 结构体的状态 */init_spin_delay(&delayStatus, file, line, func);/* tas 函数返回值不为0，表示加锁失败，会尝试进入忙等。*/while (TAS_SPIN(lock)){perform_spin_delay(&delayStatus);}finish_spin_delay(&delayStatus);return delayStatus.delays;
}

忙等以及超时等待的逻辑前面讲信号量的时候已经描述的比较清楚了，我们需要关注的是在 perform_spin_delay 函数里面对关于 x86_64 的 SPIN_DELAY();宏定义的实现。

它底层执行了一个 rep; nop指令，其实也就是 pause 指令，这个指令核心目的在前面提到的官方文档中也有描述，因为我们实现的是忙等，在一个循环中，这个过程 lock变量的数据被放在了 cpu-cache 中，访问较为高效。但是，其他cpu 释放锁并修改了这个 lock变量的值，由于cpu cache一致性，发现本地cpu这个变量的值和内存中的值不一致，会触发cache失效并去flush 本地cpu的pipeline。 nop即 pause 指令会提示cpu 当前执行的代码序列是一段 spin-wait循环，cpu会根据这个提示来防止内存序失效并且阻止刷新 pipeline。

而且 pause 指令的优势还在于能够减少为 spin-lock 这样的循环等待生成的流水线序列，从而降低了电源的功耗，cpu 官方建议对于忙等过程，需要加入 pause 指令，能够有效提升整体CPU高负载下的性能。

#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");
}

对于释放锁锁资源的过程，这一部分只需要操作指令集，且lock变量一般是属于某一个结构体的整型变量，结构体资源释放的时候会去做释放操作，看了一下 SpinLockFree的宏定义并没有其他人去使用。

解锁则在TAS中就是变更 lock状态，目前的实现是标识 *lock = 0，因为lock不是原子变量，所以会有安全问题：

*lock并不是一个原子变量，会导致内存重新派列内存序列，有可能访问不安全，且会有cpu 和内存数据同步较慢的问题，间接影响性能
不同的硬件平台可能定义自己的 S_UNLOCK，如果将这个简单得定义为内联宏定义，编译器可能会重新排列临界区内部的执行指令，从而可能出现部分临界区内部的指令在临界区外部锁释放之后执行。所以这里保持所有平台都用相同的逻辑 *lock = 0。

Linux kernel Spinlock 实现

PG 这里对自己应用的系统资源占用情况的把控较为精确，不太会有 spinlock 频繁竞争的情况，所以也就没有排队问题的处理了，仅仅支持忙等以及 timeout 或者 delay limit次数这样的限制就足够了。

而在 Linux kernel中为了提供一个较为通用的 spinlock 接口或者内部代码对 spinlock的需求都提出了排队的需求（多进程以及多线程的支持，尤其多多线程），不能所有的 caller 都等待一个变量的状态，这样可能会出现某一个caller 长期饥饿的情况。

所以 Linux kernel 实现的自旋锁是需要排队机制的（当然，linux 内核支持了很多不同的硬件架构，不同架构下的spinlock的实现因为各自的架构原因都是有差异的，这里我们介绍一个比较有代表性的 arm 架构）

如下图是 arm 架构下的 spinlock 实现的形态，需要利用 tickets 维护一个排队机制：

先看看 spinlock 的锁结构体内容：

typedef struct {union {u32 slock;struct __raw_tickets {#ifdef __ARMEB__u16 next;u16 owner;
#elseu16 owner; /* 标识当前进程 认为是谁正在持有锁 */u16 next; /* 唯一标识一个 无法抢占spinlock 的进程  */
#endif} tickets;};
} arch_spinlock_t;

tickets 内部的两个 uint16 类型的变量用来维护不同spinlock 之间的等待队列。

再看看以上流程图，加锁以及解锁时的排队部分逻辑如下：

初始时 spinlock 的核心字段 next = owner = 0
进程1 首次拿到spin_lock，会本地保存下next的值，即 next = 0，并将 spinlock 的 next 字段原子加1。
获得锁的前提是，进程本地的 next 和 owner相等，该进程才能获得锁。
接下来进程2 尝试抢占spin_lock，发现此时 spin_lock 的next值为1（进程1 抢占之后原子自增了1），先保存到进程本地，然后再次执行spinlock 本身的next 值原子加一。此时 spinlock的 owner 仍然为0。因为对于进程2，next = 1 , owner = 0，两者不想等，所以进行自旋。
同理进程3 和进程 4 也想要抢占 spinlock 锁，进程3 先到，将next = 2 保存到进程本地让spinlock 加一，并进入自旋。进程4 将next = 3 保存到本地，让spinlock 加一进入自旋。
进程1 处理完临界区释放锁，会让 spinlock 的 owner 加1，因为进程2 自旋时一直 check 进程本地的 next 和 spinlock 的 owner字段是否相等，此时 lock.owner = 1 , 进程2 的 next = 1，已经相等了。所以进程2 持有了锁。
进程2 处理完临界区释放锁，会让 spinlock 的 owner 再次加一，变成了2，就和进程3 的 next 字段相等了，这样进程3就持有了锁。依次，进程4也会持有锁。

代码中 arm架构的加锁流程如下：

static inline void arch_spin_lock(arch_spinlock_t *lock)
{unsigned long tmp;u32 newval;arch_spinlock_t lockval; // 进程/线程 本地保存的 spinlock 锁信息,主要维护next 和 ownerprefetchw(&lock->slock); // 从内存中读取 不同进程/线程 间传递的 spinlock 信息到L2 cache__asm__ __volatile__(
"1:    ldrex   %0, [%3]\n" // 原子读取 lock信息，因为已经prefetch到cpu-cache了，所以不需要锁内存总线
"  add %1, %0, %4\n" // 自增
"  strex   %2, %1, [%3]\n" // 自增的结果原子更新到 spinlock变量
"  teq %2, #0\n"
"  bne 1b": "=&r" (lockval), "=&r" (newval), "=&r" (tmp): "r" (&lock->slock), "I" (1 << TICKET_SHIFT): "cc");/* 进程 check 本地的next 和 读取到的spinlock 的 owner字段是否相等，不等则自旋 */while (lockval.tickets.next != lockval.tickets.owner) {wfe(); // 类似 x86_64 的 pause 指令，防止cpu 流水线失效造成的性能损耗// 尝试从内存中读取 spinlock 本身的 owner字段lockval.tickets.owner = READ_ONCE(lock->tickets.owner);}smp_mb(); // 内存屏障
}

解锁代码就比较简单了，spinlock 的 owner 字段加一就好了。

static inline void arch_spin_unlock(arch_spinlock_t *lock)
{smp_mb();lock->tickets.owner++;dsb_sev();
}

到此，整个 PG spinlock的基本体系就描述完了，当然内核的spinlock 实现比较多，也有一些依赖排队论实现的 qspinlock ,有更多的细节。

对于我们来说，PG 本身实现的spinlock 在通用架构下已经是对硬件非常友好的实现方式了，而且提供了 timeout 和 retry limit 这样的限制，最大程度得降低了 spinlock 本身不当使用的场景下对整体性能的影响。

对于 PG 内核的开发者们来说，spinlock 本身的使用原则肯定是 临界区执行时间非常短的场景。

接下来我们一起看看 LWLock 的实现方式，它主要用来保护 PG 内部非常多的存储在共享内存中的数据结构。

LWLocks

LWLocks 在 PG 内部起着非常重要的作用，保护了对众多共享内存变量的访问以及修改。

对于轻量锁来说本身的使用形态可能比较接近读写锁（读读共享，读写加锁，写写加锁），但是PG 为了保证加锁过程中的性能以及减少锁冲突会有一些额外的优化（维护了等待队列）。

InitializeLWLocks 初始化轻量锁

关于轻量锁实现部分的主要代码是在 lwlock.c中，轻量锁本身底层实现（加锁/解锁）都是一样的，但是为了降低锁冲突，PG 为每一个子系统/子结构分配了一个轻量锁变量，也做了一些锁变量的分类。

INDIVIDUAL_LWLOCKS，顾名思义，这种轻量锁变量只为一些特定的共享内存变量提供保护。这一些锁变量的数量都是比较固定的，所处的位置以及保护的临界区也是 PG 内部最为重要的一部分。

比如下图 async 中保护 asyncCtl 共享内存变量的就只用 AsyncCtlLock 变量，保护 autovacuum 进程正常运行的就只用 AutvacuumLock变量：

PG 在 lwlocknames.txt 中定义了很多 individual 轻量锁变量，这一些变量都会被初始化放在 MainLWLockArray 全局数组里，方便随时取用。

初始化代码 InitializeLWLocks 中的下面这一部分逻辑就是主要初始化他们：

Name tranches locks

这一种锁主要是一些不在特定场景使用的轻量锁，或者用户自定义新的轻量锁，自己使用。

typedef enum BuiltinTrancheIds
{LWTRANCHE_CLOG_BUFFERS = NUM_INDIVIDUAL_LWLOCKS,LWTRANCHE_COMMITTS_BUFFERS,LWTRANCHE_SUBTRANS_BUFFERS,LWTRANCHE_MXACTOFFSET_BUFFERS,LWTRANCHE_MXACTMEMBER_BUFFERS,LWTRANCHE_ASYNC_BUFFERS,LWTRANCHE_OLDSERXID_BUFFERS,LWTRANCHE_WAL_INSERT,LWTRANCHE_BUFFER_CONTENT,LWTRANCHE_BUFFER_IO_IN_PROGRESS,LWTRANCHE_REPLICATION_ORIGIN,LWTRANCHE_REPLICATION_SLOT_IO_IN_PROGRESS,LWTRANCHE_PROC,LWTRANCHE_BUFFER_MAPPING,LWTRANCHE_LOCK_MANAGER,LWTRANCHE_PREDICATE_LOCK_MANAGER,LWTRANCHE_PARALLEL_HASH_JOIN,LWTRANCHE_PARALLEL_QUERY_DSA,LWTRANCHE_SESSION_DSA,LWTRANCHE_SESSION_RECORD_TABLE,LWTRANCHE_SESSION_TYPMOD_TABLE,LWTRANCHE_SHARED_TUPLESTORE,LWTRANCHE_TBM,LWTRANCHE_PARALLEL_APPEND,LWTRANCHE_SXACT,LWTRANCHE_FIRST_USER_DEFINED
}           BuiltinTrancheIds;

每一个这种类型的轻量锁变量会被单独注册，有属于自己的 id 标识以及名字，开发者需要加，也需要通过如下方式注册

当然，这一些轻量锁变量会单独放在 &NamedLWLockTrancheArray 全局数组中，初始化的时候会从 individual 个锁之后开始初始化接下来的拥有命名的轻量锁。

static void
InitializeLWLocks(void)
{int            numNamedLocks = NumLWLocksByNamedTranches();.../* Initialize named tranches. */if (NamedLWLockTrancheRequests > 0){char    *trancheNames;NamedLWLockTrancheArray = (NamedLWLockTranche *)&MainLWLockArray[NUM_FIXED_LWLOCKS + numNamedLocks];trancheNames = (char *) NamedLWLockTrancheArray +(NamedLWLockTrancheRequests * sizeof(NamedLWLockTranche));lock = &MainLWLockArray[NUM_FIXED_LWLOCKS];...}
}

这一些轻量锁变量的所有具体使用场景，是我们本节没有精力关注的，主要还是关注在锁本身的实现上，关注其实现细节，有我们可以学习借鉴的地方。

LWLockAcquire 加锁

在 PG 7.0 版本以及之前的版本，LWLock 的加锁实现是直接通过 spinlock 来实现的。但是轻量锁本身的应用场景对锁本身提出了两个比较重要的需求：

对于读读场景，我们希望可以无锁方式去读，而读写或者写写这样有修改的场景则需要有锁来进行保护
在有修改的场景，不希望有部分操作很难甚至永远抢不到锁的现象。

对于这样的需求，仅仅只用spinlock 显然无法实现。

对于第一个需求，在 LWLock 的实现中提出了两种主要的锁模式 LWLockMode, LW_EXCLUSIVE 和 LW_SHARED。当然，实际实现中还增加了一种额外的锁模式 LW_WAIT_UNTIL_FREE，这个模式下所有的操作都是需要等待的。

先看一下锁数据结构：

typedef struct LWLock
{uint16     tranche;        /* tranche ID */pg_atomic_uint32 state;     /* state of exclusive/nonexclusive lockers */proclist_head waiters;     /* list of waiting PGPROCs */
#ifdef LOCK_DEBUGpg_atomic_uint32 nwaiters; /* number of waiters */struct PGPROC *owner;        /* last exclusive owner of the lock */
#endif
} LWLock;

其中：

tranche, 用来唯一标识一个轻量锁变量，tranche id 以及 trance name 的类型和注册方式前面已经讲过了。
state，标识当前锁的mode，是 exclusive 还是 noexclusive 的。
Waiters，这是一个按照加锁顺序来保存的等待锁队列（双向链表实现的，其中的节点元素是pgprocno）。

第一需求的实现方式是在上一个锁解锁的时对等待队列中的加锁操作进行唤醒，如果第一个锁操作是一个排他锁(exclusive)，那么只会唤醒一个锁操作；如果第一个锁操作是一个共享锁(shared)，那么会唤醒当前等待队列中的所有的共享锁操作，如下图：

唤醒锁(解锁)的代码实现待会再细说，先对解锁的大体流程有一个初步的了解，可以发现对于第一个需求这样的解锁肯定是满足的，对于第二个需求，也就是加锁过程中会构造一个锁操作的队列（同一个锁变量被并发使用是）。

加锁的主要步骤如下：

尝试加锁，如果成功就返回
如果尝试失败，则需要等待。
1. 这里 PG 为了考虑部分轻量锁操作的临界区比较小，这里本应做的等待并没有直接去执行。而是先将当前锁加入等待者队列中。紧接着再次尝试加锁，此时对于那一些操作较小的临界区可能已经释放锁了，此时尝试加锁可能会成功，如果成功了的话则需要回退添加到队列中的锁。
2. 如果前面的步骤还是没有加锁成功，意味着需要有较长时间的等待。则会让这把锁等待在一个信号量上，直到这个信号量被释放。

bool
LWLockAcquire(LWLock *lock, LWLockMode mode)
{PGPROC    *proc = MyProc;bool     result = true;int          extraWaits = 0;...for (;;){bool        mustwait;/** 尝试加锁，如果成功，则直接返回。*/mustwait = LWLockAttemptLock(lock, mode);if (!mustwait){LOG_LWDEBUG("LWLockAcquire", lock, "immediately acquired lock");break;                /* got the lock */}.../* 先加入到锁队列中。 */LWLockQueueSelf(lock, mode); /* 再次尝试加锁，如果成功，则回退等待队列。 */mustwait = LWLockAttemptLock(lock, mode);if (!mustwait){LOG_LWDEBUG("LWLockAcquire", lock, "acquired, undoing queue");LWLockDequeueSelf(lock);break;}.../* 如果失败，则等待在一个信号量上。 */for (;;){PGSemaphoreLock(proc->sem);if (!proc->lwWaiting)break;extraWaits++;}...}
}

中间过程中的一些锁入队或者尝试加锁操作会统一由 spinlock 实现的一系列原子操作来保障。

比如锁对象加入双向链表构造的队列之前会有一个 LWLockWaitListLock 进行链表操作的保护，这个函数内部是pg_atomic_fetch_or_u32 实现的 CAS 操作，底层是通过spinlock 实现的。
还有一系列的 spinlock 实现的原子操作（原子写，原子TAS 操作，原子Fetch等）都在 fallback.h中。

LWLockRelease 解锁

解锁过程需要解决的需求前面加锁部分介绍的时候已经提到了。轻量锁的核心是在保证不被饿死的情况下尽可能降低锁的冲突，最直接的方式就是利用共享和排他模式来解决。

而这两种模式的实现就是在解锁的过程。

大体步骤如下：

创建要唤醒的临时锁队列 wakeup。
使用 LWLockWaitListLock 进行加锁，后续会操作当前的lock 等待队列waiters。
遍历当前锁的 waiters，将每一个锁操作添加到 wakeup 临时锁队列。
如果第一个等待的锁操作模式是 EXCLUSIVE，则break 遍历，否则将非 EXCLUSIVE的锁模式添加到 wakeup 队列中。
遍历 wakeup队列，执行唤醒操作，主要是通过设置信号量（让每一个锁操作 – proc 的waiter 的信号量加一），恢复其可以消费的状态。

对应的源代码逻辑如下：

LWLockReleaseLWLockWakeupstatic void
LWLockWakeup(LWLock *lock)
{bool       new_release_ok;bool     wokeup_somebody = false;proclist_head wakeup;proclist_mutable_iter iter;/* 初始化 临时的等待队列。 */proclist_init(&wakeup);/* 遍历所有的 lock-waiters. */proclist_foreach_modify(iter, &lock->waiters, lwWaitLink){PGPROC     *waiter = GetPGProcByNumber(iter.cur);/* 已经添加到唤醒队列的，且waiter->lwWaitMode 是排他模式就跳过。  */if (wokeup_somebody && waiter->lwWaitMode == LW_EXCLUSIVE)continue;/* 否则，就冲原来的wiaters 中删除，添加到 wakeup临时队列的尾部。 */proclist_delete(&lock->waiters, iter.cur, lwWaitLink);proclist_push_tail(&wakeup, iter.cur, lwWaitLink);if (waiter->lwWaitMode != LW_WAIT_UNTIL_FREE){/** Prevent additional wakeups until retryer gets to run. Backends* that are just waiting for the lock to become free don't retry* automatically.*/new_release_ok = false;/** Don't wakeup (further) exclusive locks.*/wokeup_somebody = true;}/** 对于第一个是排他锁的，就直接break. */if (waiter->lwWaitMode == LW_EXCLUSIVE)break;}.../* 唤醒 缓存到 wakeup临时队列中的锁操作. */proclist_foreach_modify(iter, &wakeup, lwWaitLink){PGPROC    *waiter = GetPGProcByNumber(iter.cur);LOG_LWDEBUG("LWLockRelease", lock, "release waiter");proclist_delete(&wakeup, iter.cur, lwWaitLink);/** Guarantee that lwWaiting being unset only becomes visible once the* unlink from the link has completed. Otherwise the target backend* could be woken up for other reason and enqueue for a new lock - if* that happens before the list unlink happens, the list would end up* being corrupted.** The barrier pairs with the LWLockWaitListLock() when enqueuing for* another lock.*/pg_write_barrier();waiter->lwWaiting = false;/* 主要通过操作这个waiter等待的信号量，会对信号量的值+1，来恢复信号量可以在加锁部分继续消费的能力。 */PGSemaphoreUnlock(waiter->sem);}

轻量锁利用spinlock 实现的一系列原子变量，保障了内部数据结构并发访问的安全性。

为了满足上层应用/调用者需求（降低锁冲突，减少长期无法加锁/饿死的情况），分别实现了共享模式和排他模式来减少部分场景的锁冲突以及等待队列来减少锁操作饿死的情况，提供了按照锁操作的顺序来加锁的能力。

当然，LWLock 本身为了性能的考虑，内部也有一些代码细节（比如第一次尝试加锁之后不成功，会放入等待队列，再一次尝试加锁，对于临界区较小的场景，这样能够快速得到锁，不需要再次陷入信号量的长期等待中）。

接下来再一起看看 PG 内部最为复杂的锁类型 – 常规锁。

Regular Locks

常规锁它保护的临界区是数据库对象的操作，而不是单纯的共享内存变量或者某一个原子变量。在PG内部数据库对象包括表、页面、元组等，Regular lock 在这一些对象的保护性质中就像是读写锁，在这里听起来并没有什么复杂度，和 LWLock 的作用大同小异？

Regular locks 复杂度的体现在于用户上层较多的 DML/DDL 语句对不同的数据库对象的操作保证安全的情况下尽最大可能提升性能，这才是核心。比如 select 一个表的一个元组，总不能因为有其他对该表的 delete/update 事务操作就像读写锁一样阻塞等待吧？并发得针对一个表进行修改 (alter-table) + update /delete 不能出现互相等待的情况吧，还需要有死锁检测。

所以，如果在众多的DML/DCL 语句中操作众多的数据库对象来定制一套通用的规则保证性能最大化且不会出现类似死锁的安全问题，那复杂度就上来了。

锁模式以及相容性

Regular locks 总共提供了8种锁模式，锁模式这里的区分主要是为了对 DML 和 DDL 操作进行区分，保证安全性的情况下让不同的操作一起执行的时候拥有最大的性能。

8种锁模式以及加锁对应的主要操作语句类型如下：

#define AccessShareLock          1   /* SELECT 最低级别的锁 */
#define RowShareLock            2   /* SELECT FOR UPDATE/FOR SHARE */
#define RowExclusiveLock        3   /* INSERT, UPDATE, DELETE */
#define ShareUpdateExclusiveLock 4  /* VACUUM (non-FULL),ANALYZE, CREATE INDEX* CONCURRENTLY */
#define ShareLock               5   /* CREATE INDEX (WITHOUT CONCURRENTLY) */
#define ShareRowExclusiveLock   6   /* like EXCLUSIVE MODE, but allows ROW* SHARE */
#define ExclusiveLock           7   /* blocks ROW SHARE/SELECT...FOR UPDATE */
#define AccessExclusiveLock     8   /* ALTER TABLE, DROP TABLE, VACUUM FULL,* and unqualified LOCK TABLE ，对系统表进行操作时会申请该锁*/

这8种锁之间的相容性如下表：

Mode	(1)	(2)	(3)	(4)	(5)	(6)	(7)
AccessShareLock (1)							❌
RowShareLock (2)						❌	❌
RowExclusiveLock (3)					❌	❌	❌
ShareUpdateExclusiveLock (4)				❌	❌	❌	❌
ShareLock (5)			❌		❌	❌	❌
ShareRowExclusiveLock (6)			❌	❌	❌	❌	❌
ExclusiveLock (7)		❌	❌	❌	❌	❌	❌
AccessExclusiveLock (8)	❌	❌	❌	❌	❌	❌	❌

锁模式相容的意思是当我们加锁的时候，相容的锁类型之间不会互相阻塞；不相容的模式对于后加锁的事务操作会阻塞等锁，具体如何等待获取锁的逻辑后续会详细描述。

接下来看几个相容性相关的简单例子：

AccessExclusiveLock 和 AccessShareLock

AccessExclusiveLock 主要是保护对系统表的操作，也就是 DDL 操作，这种锁模式与所有的锁模式都不相容。

s1 : begin;
s1 : alter table a add c3 int;
s2 : select relation::regclass, pid, mode,granted,fastpath from pg_locks where relation = 'a'::regclass;relation |   pid   |        mode         | granted | fastpath
----------+---------+---------------------+---------+----------a        | 4016693 | AccessExclusiveLock | t       | f
(1 row)
s3 : select * from a; # 卡住，因为不相容，需要等锁，直到 s1 的事务提交。
s2 : select relation::regclass, pid, mode,granted,fastpath from pg_locks where relation = 'a'::regclass;relation |   pid   |        mode         | granted | fastpath
----------+---------+---------------------+---------+----------a        | 4017039 | AccessShareLock     | f       | fa        | 4016693 | AccessExclusiveLock | t       | f# 此时 s3 还没有获取到锁，可以看到 granted 是false
(2 rows)
s1 : commit;
s3 : # 恢复执行c1 | c2  | c3
----+-----+----11 | 333 |3 | 333 |
(2 rows)
s2 : select relation::regclass, pid, mode,granted,fastpath from pg_locks where relation = 'a'::regclass;relation |   pid   |      mode       | granted | fastpath
----------+---------+-----------------+---------+----------a        | 4017039 | AccessShareLock | t       | f
(1 row)
# 此时 s3 加锁成功， granted 为 true. 因为s3 还没有提交，所以还持有锁。

由上面的案例可以很明显的发现 AccessExclusiveLock 的相容性是最差的，因为它是表锁，涉及到一些元数据(系统表 catalog)的操作，所以持锁期间需要保证这段时间内不会有其他任何操作能够访问或者操作这个表。

pg_locks 系统表能够看到整个数据库内部的所有锁模式的视图，上面提到的几列含义分别如下：

relation, 当前数据库内部唯一标识一张表，这里会展示表名。

pid, 当前操作所属的backend 进程id

mode , 这个操作要加锁的模式

granted, 是否获得了当前模式的锁，是则为 t – true, 否则为 f – false.

fastpath, 是否进入了快速路径（后续会细说，就是提升性能的一种方式）, t 或者 f 同上。

ExclusiveLock , RowExclusiveLock , RowShareLock, AccessShareLock

ExclusiveLock 模式相容性也比较差，它会阻塞除了 AccessShareLock 之外的所有的锁。

RowExclusiveLock 行锁和 RowExclusiveLock 以及 RowShareLock 说的是兼容的，但是前提是不会操作到同一行即同一个元组，如果操作的是同一个元组的话那还是会有不兼容的问题（额外加一个 ExclusiveLock），如下案例：

s1 : begin;
s1 : select * from a for update;s2 : begin;
s2 : update a SET c3 = 2; # 阻塞，因为s1 select for update了，需要提交之后 s2才能继续加锁s3 : select relation::regclass, pid, mode,granted,fastpath from pg_locks where relation = 'a'::regclass;relation |   pid   |       mode       | granted | fastpath
----------+---------+------------------+---------+----------a        | 4017039 | RowExclusiveLock | t       | t        # s2 ,update 操作要加 RowExclusiveLock 锁a        | 4016693 | RowShareLock     | t       | t        # s1, select for update 加 `RowShareLock` 模式a        | 4017039 | ExclusiveLock    | t       | f        # s2
(3 rows)s1 : commit;
s3 : select relation::regclass, pid, mode,granted,fastpath from pg_locks where relation = 'a'::regclass;relation |   pid   |       mode       | granted | fastpath
----------+---------+------------------+---------+----------a        | 4017039 | RowExclusiveLock | t       | t         # s1 提交了，仅剩下 s2 的update操作了，只需要保留`RowExclusiveLock`即可
(1 row)

可以看到不同事物之间当 update 操作和 select ... for update 会操作同一行数据时对于 update 操作的进程除了正常要加的 RowExclusiveLock 模式之外，还需要加一个 ExclusiveLock，保证 s1 事务提交之后当前的 update 才会有效。

此时，仍然保留 s2 的事务不提交，如果我们尝试 update 一个其他行的数据，那就不会有冲突了:

s4 : begin;
s4 : update a set c3 = 3 where c1 = 3;s3 : select relation::regclass, pid, mode,granted,fastpath from pg_locks where relation = 'a'::regclass;relation |   pid   |       mode       | granted | fastpath
----------+---------+------------------+---------+----------a        | 4017039 | RowExclusiveLock | t       | t         # s2 的 未提交的 update.a        | 4016693 | RowExclusiveLock | t       | t         # s4 的 update 操作
(2 rows)

当然，PG 还提供了 LOCK 语句来直接对一个事务加指定模式的锁，做一些锁模式的正确性测试。

s5 : begin;
s5 : lock TABLE a in share mode;
s5 : select relation::regclass, pid, mode,granted,fastpath from pg_locks where relation ='a'::regclass;relation |   pid   |   mode    | granted | fastpath
----------+---------+-----------+---------+----------a        | 4016988 | ShareLock | t       | f
(1 row)

其他像share 相关的lock 模式就是创建索引时会加的。

从上面简单的例子能够较为清楚得看到一些兼容性的基本情况，对于不兼容的锁模式也很明显，会阻塞直到其他先持有锁的事务提交/终止（完成释放锁）。

数据结构

介绍详细的数据结构内容之前先粗略理清楚实现 Regular 过程中的几种主要的锁数据结构。

LOCK 存在于共享内存中，用来保存当前数据库中所有事物的锁对象，被称为 主锁表(LockMethodLockHash)
PROCLOCK 存在于共享内存中，用来保存当前进程（会话 – backend）的事务锁状态，用于建立锁和进程会话的关系。
LOCALLOCK 存储到会话本地 – 非共享，因为实际操作过程中会话可能会在一个锁对象上申请多次相同类型的锁，这样就没有必要做冲突检测，直接从会话本地拿就好了。

在描述这三个主要的常规锁数据结构细节之前，先通过如下流程图整体看看这三种锁之间的关系，就能有一个整体的认识了:

以上流程图详细描述了三种主要的锁数据结构之间的互相指向关系，需要注意的是：

LOCALLOCK 是保存在会话本地(backend 进程本地)，对其他的会话不可见，所以从上面的 LOCALLOCK 数据结构中能看到其保存的 LOCK 以及 PROCLOCK 只有属于自己会话的。同时维护了一个可以同台扩容的 LOCALLOCKOWNER 数组来让自己能够管理同一个会话内部多个不同的锁之间的归属关系。因为同一个会话内部可能会加同一种类型的锁多次，这样在释放资源的时候就能够按照归属关系有序释放。

当然，LOCALLOCK 存在和核心目的还是为了一个会话避免频繁加同一种模式锁的时候去访问共享内存中的 LOCK和PROCLOCK，从而降低加锁性能。

关于 ResourceOwnerData 数据结构的管理形态如下，本质上是一个单链表，希望能够标识每一个ResourceOwnerData 节点所属的初始资源管理器（执行器的 CreatePortal会标识当前执行语句中所有的加锁操作都有一个可以追溯的起始资源管理器），从而快速得申请释放锁，不同的 ResourceOwnerData 之间也有链表维护的所属关系。
对于 LOCK 和 PROCLOCK 来说，它们本身都存储在共享内存中，会被整个数据库看到，不同的会话都能看到全局的一个锁视图，这个时候对这两种数据结构的访问就需要提供一些能够访问到全局的子变量。比如 LOCK 的 procLocks 变量，是一个双向链表，能够访问到所有的 PROCLOCKS；同时PROCLOCK的 tag 变量，能够访问到自己的 LOCK 结构，也能访问到自己所属的 PROC结构。

PROC 是唯一标识一个会话的内核结构，保存用户访问数据库时启用的一个会话backend进程所有的上下文信息。其内部也存储了一个当前会话访问数据操作所添加的锁模式数组 myProcLocks，能够访问到自己所有的 PROCLOCK 对象。

初始化常规锁

对常规锁的初始化过程主要通过 InitLocks函数来实现，会为以上三个数据结构各创建一个hash表，用来保存锁数据结构。

LockMethodLockHash，数据库级别的锁表，为Lock 数据结构创建的hash表，hash key 用LOCKTAG通过hash函数生成，整个hash表会存储到共享内存中。
LockMethodProcLockHash, 进程级别的锁表，为 ProcLock 数据结构创建的hash 表，hash key用 PROCLOCKTAG 通过hash函数生成，同样会存储到共享内存中。
LockMethodLocalHash 本地锁表，为 LocalLock 数据结构的创建的hash表，LOCALLOCKTAG 通过hash 函数生成hash-key, 存储到本地。

关于 LOCKTAG 或者 PROCLOCKTAG 这样的标记都是为了存储这个锁锁定的对象类型、对象id 以及加锁方法。比如对 LOCKTAG的一个宏定义初始化：

#define SET_LOCKTAG_RELATION(locktag,dboid,reloid) \((locktag).locktag_field1 = (dboid), \(locktag).locktag_field2 = (reloid), \(locktag).locktag_field3 = 0, \(locktag).locktag_field4 = 0, \(locktag).locktag_type = LOCKTAG_RELATION, \(locktag).locktag_lockmethodid = DEFAULT_LOCKMETHOD)

这个锁的对象类型是 relation，也就表锁。在表锁参与的情况下对 LOCKTAG的初始化还需要 dboid (数据库对象id)以及 reloid (表对象id) 的参与。同样的还有 Tuple锁以及 page锁，tuple的话就需要额外保存 blocknum 以及 offnum id信息来在数据表内唯一标识一个tuple。

回到常规锁的初始化，主体逻辑是在 InitLocks 中进行的，它被CreateSharedMemoryAndSemaphores调用，这个函数则是postmaster或者backend进程每次启动的时候都必须执行的逻辑，用来初始化自己进程需要的一些全局内存变量或者共享内存变量。

在 InitLocks 中，对三个数据结构的初始化逻辑如下：

void
InitLocks(void)
{HASHCTL        info;long       init_table_size,max_table_size;bool     found;...MemSet(&info, 0, sizeof(info));info.keysize = sizeof(LOCKTAG);info.entrysize = sizeof(LOCK);info.num_partitions = NUM_LOCK_PARTITIONS;/* 初始化 存储数据库级别锁 的hash表。 */LockMethodLockHash = ShmemInitHash("LOCK hash",init_table_size,max_table_size,&info,HASH_ELEM | HASH_BLOBS | HASH_PARTITION);.../* 初始化 进程级别锁的 hash表。 */LockMethodProcLockHash = ShmemInitHash("PROCLOCK hash",init_table_size,max_table_size,&info,HASH_ELEM | HASH_FUNCTION | HASH_PARTITION);.../* 快速锁机制需要用到的数据结构。 */FastPathStrongRelationLocks =ShmemInitStruct("Fast Path Strong Relation Lock Data",.../* 本地锁 的hash表。 */LockMethodLocalHash = hash_create("LOCALLOCK hash",16,&info,HASH_ELEM | HASH_BLOBS);

因为 LOCK 和 PROCLOCK 本身需要保存到共享内存中，所以会通过 ShmemInitHash 函数进行初始化，同时hash表需要分桶 HASH_PARTITION，而对于LOCKLOCK 则保存在进程本地，并且不需要分桶，所以它的初始化并没有加入HASH_PARTITION标记。

快速锁机制是为了让加多次弱锁（锁模式部分介绍过的8种锁模式中有三种互相兼容，只有这三种是弱锁，其他的都是强锁）的进程保存信息到进程本地而不用让其他进程知道，弱锁场景是比较通用的场景，这样能够提升加锁的性能，不需要每次去共享内存中加载主锁表和进程锁表的信息(不需要走死锁检测流程)。

加锁 LockAcquire

接下来就进入了常规锁的核心链路，加锁的实现了，主要逻辑还是如何利用前面提到的三种数据结构来加速锁的获取或者降低等待者对性能的影响。LockAcquire --> LockAcquireExtended，基本步骤如下：

主要的输入参数为： locktag 锁对象的类型，lockmode 8种锁模式中的一种。

先利用输入的 logtag 作为hash-key，在本地锁表对应的hash表(LockMethodLocalHash)中查找，没有找到则新建一个，存储到hash表中。
如果当前锁对象加锁次数nLocks 大于 0，则这个锁意味着被别人持有了，则直接返回 LOCKACQUIRE_ALREADY_HELD.
如果锁类型是 AccessExclusiveLock 且锁对象是 Relation，则会尝试分配一个 transactionid 来在后续加锁成功之后写一条 WAL record.
如果EligibleForRelationFastPath为真，则表示满足快速锁的条件，申请的是弱锁（弱锁，锁模式 < ShareUpdateExclusiveLock）:

a. 如果强锁的计数不为0，则表示已经有强锁了，因为不兼容，则无法成功获取锁。

b. 否则，通过 FastPathGrantRelationLock进行加锁，成功了就返回加锁成功。
如果申请的是强锁，ConflictsWithRelationFastPath 这个为真，则会先将当前进程持有的强锁的计数自增，通过 FastPathTransferRelationLocks并将快速锁信息从会话本地转移到主锁表中（共享内存中）。
要加的锁不在本地锁表，也不在 FashPath中，则需要访问进程锁表和主锁表。当前申请锁类型是强关系锁，则通过SetupLockInTable 尝试从主锁表中查找或者创建当前的锁项。
如果这种锁模式需要做冲突检测且没有其他会话在等待这把锁，那通过 LockCheckConflicts 直接执行锁模式的冲突检测（并不是死锁检测，只是确认已经持有的锁和当前锁是否冲突）；否则当前会话也需要进入 WaitOnLock 的等锁逻辑，这里WaitOnLock中会有死锁检测的逻辑，防止循环等待问题。
如果第三步执行成功了，则需要写入一条 WAL-record（方便其他的主从进程恢复的时候能够知道这里加了一把表锁），返回加锁成功即可。

这几步的源代码实现如下：

第一步：查找本地锁表

/* 仍然从InitLocks 初始化好的 保存本地锁表的hash表 LockMethodLocalHash 中查找是否有localtag对应的锁 */locallock = (LOCALLOCK *) hash_search(LockMethodLocalHash,(void *) &localtag,HASH_ENTER, &found);
/* 没有则创建一个 */if (!found){locallock->lock = NULL;locallock->proclock = NULL;locallock->hashcode = LockTagHashCode(&(localtag.lock));locallock->nLocks = 0;locallock->holdsStrongLockCount = false;locallock->lockCleared = false;locallock->numLockOwners = 0;locallock->maxLockOwners = 8;locallock->lockOwners = NULL; /* in case next line fails */locallock->lockOwners = (LOCALLOCKOWNER *)MemoryContextAlloc(TopMemoryContext,locallock->maxLockOwners * sizeof(LOCALLOCKOWNER));}

第二步：检查该锁在本地锁表中的加锁次数

因为 nLocks 大于0 就意味着这个锁已经被当前进程持有了，会做一个resource owner的一个更新，将当前owner添加到前面介绍锁数据结构中提到的 ResourceOwner 链表中。

/* nLocks 标识当前locklock 被持有的次数。 */if (locallock->nLocks > 0){/* 继续增加次数，并将本次的resource owner 添加到本地管理的 resource owner节点链表中。 */GrantLockLocal(locallock, owner);if (locallock->lockCleared)return LOCKACQUIRE_ALREADY_CLEAR;else/* 返回已经持有了。 */return LOCKACQUIRE_ALREADY_HELD;}

**第三步：针对表锁且锁模式为AccessExclusiveLock 分配一个transactionid **

主要目的是为了记录一条 wal-log ，对于表锁这种超大粒度的锁需要让其他进程拿到这个wal-log 进行recovery的时候能够看到这里曾经加过表锁。

/* 表锁的check如下，且要求本次加锁不是 recovery期间 */if (lockmode >= AccessExclusiveLock &&locktag->locktag_type == LOCKTAG_RELATION &&!RecoveryInProgress() &&XLogStandbyInfoActive()){LogAccessExclusiveLockPrepare();log_lock = true;}

第四步：进入快速锁模式

快速锁模式是针对会话加弱锁时减少对主锁表/进程锁表等存储在共享内存锁信息的访问，弱锁之间是互相兼容的，所以不需要做死锁检测。当然，快速锁机制添加成功的前提是当前会话之前没有添加过强锁且还有充足的保存弱锁的位置，那就能够加锁成功。

/*EligibleForRelationFastPath 用来检查当前的tag和mode 是否满足进入快速锁机制的条件。 */if (EligibleForRelationFastPath(locktag, lockmode) &&FastPathLocalUseCount < FP_LOCK_SLOTS_PER_BACKEND){.../* 使用轻量锁保护这个加锁过程。 */LWLockAcquire(&MyProc->backendLock, LW_EXCLUSIVE);/* 先检查强锁计数，如果不为0，表示已经加过强锁了，则无法进入快速锁模式。 */if (FastPathStrongRelationLocks->count[fasthashcode] != 0)acquired = false;else/* 否则尝试加锁。 */acquired = FastPathGrantRelationLock(locktag->locktag_field2,lockmode);LWLockRelease(&MyProc->backendLock);/* 如果加锁成功，直接返回成功。*/if (acquired){/** The locallock might contain stale pointers to some old shared* objects; we MUST reset these to null before considering the* lock to be acquired via fast-path.*/locallock->lock = NULL;locallock->proclock = NULL;GrantLockLocal(locallock, owner);return LOCKACQUIRE_OK;}}static bool
FastPathGrantRelationLock(Oid relid, LOCKMODE lockmode)
{uint32     f;uint32        unused_slot = FP_LOCK_SLOTS_PER_BACKEND;/* relid 是唯一标识一个数据库的表，查找已有的保存的快速锁数组(16个)中是否有这个relid，有则直接返回真。 *//* Scan for existing entry for this relid, remembering empty slot. */for (f = 0; f < FP_LOCK_SLOTS_PER_BACKEND; f++){if (FAST_PATH_GET_BITS(MyProc, f) == 0)unused_slot = f;else if (MyProc->fpRelId[f] == relid){Assert(!FAST_PATH_CHECK_LOCKMODE(MyProc, f, lockmode));FAST_PATH_SET_LOCKMODE(MyProc, f, lockmode);return true;}}/* 没有找到，则拿空闲slot保存当前锁，同样直接返回真就好了。 *//* If no existing entry, use any empty slot. */if (unused_slot < FP_LOCK_SLOTS_PER_BACKEND){MyProc->fpRelId[unused_slot] = relid;FAST_PATH_SET_LOCKMODE(MyProc, unused_slot, lockmode);++FastPathLocalUseCount;return true;}/* No existing entry, and no empty slot. */return false;
}

第五步：加强锁的话需要转移快速锁存储的锁信息到主锁表中

弱锁优先会尝试能否进入快速锁模式，也就是第四步。如果是当前会话加的是强锁，则会进入如下逻辑。

/* 对强锁的检查，主要是通过检查mode 是否大于 ShareUpdateExclusiveLock. */if (ConflictsWithRelationFastPath(locktag, lockmode)){uint32       fasthashcode = FastPathStrongLockHashPartition(hashcode);/* 通过spinlock保护，自增强锁的计数。 */BeginStrongLockAcquire(locallock, fasthashcode);/* 尝试将 快速锁机制中的锁信息转移到主锁表中。*/if (!FastPathTransferRelationLocks(lockMethodTable, locktag,hashcode)){/* 失败了，则终止强锁的加锁过程。*/AbortStrongLockAcquire();if (locallock->nLocks == 0)RemoveLocalLock(locallock);if (locallockp)*locallockp = NULL;if (reportMemoryError)ereport(ERROR,(errcode(ERRCODE_OUT_OF_MEMORY),errmsg("out of shared memory"),errhint("You might need to increase max_locks_per_transaction.")));elsereturn LOCKACQUIRE_NOT_AVAIL;}}static bool
FastPathTransferRelationLocks(LockMethod lockMethodTable, const LOCKTAG *locktag,uint32 hashcode)
{.../* 1. 遍历所有的进程项 */for (i = 0; i < ProcGlobal->allProcCount; i++){PGPROC    *proc = &ProcGlobal->allProcs[i];uint32      f;/* 2. 找到当前锁所在的数据库id. */if (proc->databaseId != locktag->locktag_field1){LWLockRelease(&proc->backendLock);continue;}/* 3. 找到了databaseId，则查看当前会话内部的16个slot中 是否有当前表的relid. */for (f = 0; f < FP_LOCK_SLOTS_PER_BACKEND; f++){/* Look for an allocated slot matching the given relid. */if (relid != proc->fpRelId[f] || FAST_PATH_GET_BITS(proc, f) == 0)continue;/** 4. 找到了relid，则进入转移过程：先将快速机制保存的所有锁信息* 通过 SetupLockInTable 添加到主锁表，完成后再从快速锁中清理这个锁信息。*/。。。for (lockmode = FAST_PATH_LOCKNUMBER_OFFSET;lockmode < FAST_PATH_LOCKNUMBER_OFFSET + FAST_PATH_BITS_PER_SLOT;++lockmode){...proclock = SetupLockInTable(lockMethodTable, proc, locktag,hashcode, lockmode);...GrantLock(proclock->tag.myLock, proclock, lockmode);FAST_PATH_CLEAR_LOCKMODE(proc, f, lockmode);}...}...}return true;
}

第六步：在主锁表中查找或者新建当前强锁项

 /* 先从 主锁表 -- LockMethodLockHash 中查找 + 创建；没有找到则从 进程锁表 -- LockMethodProcLockHash查找+创建 */proclock = SetupLockInTable(lockMethodTable, MyProc, locktag,hashcode, lockmode);

第七步：检查添加的锁是否与已有的锁模式冲突，锁模式冲突的情况下进入等锁

 if (lockMethodTable->conflictTab[lockmode] & lock->waitMask)status = STATUS_FOUND;else/* 检查锁模式是否冲突，返回ok 则表示能够获得锁；否则就要进入等锁模式。*/status = LockCheckConflicts(lockMethodTable, lockmode,lock, proclock);

等锁的过程主要通过WaitOnLock函数实现，因为有强锁冲突，所以需要尝试死锁检测。

死锁检测的调用栈如下：

ProcSleepCheckDeadLockDeadLockCheckRecurse // 检查soft-edge是否有死锁FindLockCycle // 检查hard-edge是否有死锁

PG 的死锁检测和大多数数据库的实现都类似，采用的是WFG(Waits-For Graph) 有向图，参与加锁的所有进程作为有向图中的一个个定点，如果进程A 等待进程B释放锁，则生成一条A->B的边。如果存在环，则存储死锁。

有一些差异点的地方是 PG 除了维护进程等待已经被其他进程持有的锁的边(hard-edge) 之外还会维护不同等待进程之间的相互依赖边(soft-edge)，比如进程A在B之后申请锁，A和B 的锁互相冲突，那么PG 的锁唤醒函数ProcLockWakeup会优先唤醒B获取锁，这个时候需要建立 A -->B 的一条边来表示两者的依赖关系，这种边就是soft-edge，它们也需要单独做死锁检测。

在函数 CheckDeadLock中：

利用 DeadLockCheckRecurse 函数先进行 soft-edge 的死锁检测，如果发现有死锁，则后续会尝试调整内部的依赖关系来避免死锁，利用这种方式会避免 abort当前事务，尝试解决死锁。
如果 soft-edge没有死锁，则在FindLockCycle 检测hard-edge 是否有死锁，进程等待已持有锁的其他进程释放锁，这个过程有死锁问题，那PG 这里会直接 elog(FATAL)，来Abort 终止当前事务。

因为死锁检测在PG的实现中还是有比较多的细节，这个就需要后续深入更详细的优化细节之后再来单独分享了。

第八步：记录 AccessExclusiveLock 一条record-log

主要是为了针对相容性最差的锁，需要记录一条WAL日志，防止其他stand-by进程当前WAL recovery 时漏掉这个锁（只有这个锁会与其他只读的锁冲突 – SELECT 语句对应的最弱的锁模式 AccessShareLock ），导致一些数据一致性的问题，standby是只读的 pg 进程。

 if (log_lock){/** Decode the locktag back to the original values, to avoid sending* lots of empty bytes with every message.  See lock.h to check how a* locktag is defined for LOCKTAG_RELATION*//* 获取第三步构造好的 transactionid, 并利用 LogAccessExclusiveLocks 将封装好的log-tag 写入到WAL中。*/LogAccessExclusiveLock(locktag->locktag_field1,locktag->locktag_field2);}

到此完成加锁之后返回 LOCKACQUIRE_OK即可。

解锁 LockRelease

解锁相对来说流程就简单很多了：

从LockMethodLocalHash 查找LOCALLOCK，在它的owner数组里找到与当前进程对应的元素，解除LOCALLOCK与ResourceOwner之间的关联，并减小相应计数值。
将LOCALLOCK的加锁次数(nLocks)减1，如果减为0了，表示当前进程不再持有锁，则后续有两种解锁途径，分别是快速锁和主锁表。
对于快速锁，调用FastPathUnGrantRelationLock清除快速锁的加锁标记。如果解锁成功即可返回。
如果没找到快速锁，且LOCALLOCK.lock为空，说明被转移到了主锁表里(在加锁逻辑中，当申请快速锁成功时，会把LOCALLOCK.lock置空)。
查找主锁表LockMethodLockHash 以及 LockMethodProcLockHash，如果没有找到，则blog-error。找到了调用UnGrantLock更新LOCK和PROCLOCK信息。
调用CleanUpLock：要么删除主表项，要么调用ProcLockWakeup()唤醒等待者，这里会按照 WFG 中构建的 soft-edge 以及 hard-edge 依赖关系进行唤醒。

解锁的代码流程链路整体简单一些，就不展开了。

总结

到此整个PG的几种锁体系就描述完了，自低向上可以发现越来越复杂，代码细节中的性能取舍在 PG 这样拥有接近40年历史的经典数据库中展现的淋漓尽致了。

最底层的 Spinlock 保护的对象是某一个变量，临界区足够小，需要考虑和硬件交互细节，不会很复杂。但是需要对各种体系结构下的CPU-内存架构指令足够精通才能写出对硬件友好的极致代码，而这一部分也是对性能影响最重要的部分（这里的性能比较好解决，因为有现成的解决方案，比如x86_64 架构的 lock + xchg 指令），因为调用次数足够多。

上一层的 LWLock 保护的对象是共享内存的变量，临界区大了不少，因为底层Spinlock 提供了足够好的性能且为了方便实现了一系列原子操作，这样LWLock 就可以轻松的保证部分小的临界区访问的安全性，从而更关注于保护共享内存的临界区。因为共享内存的结构体变量有读也有写，所以不应该让读读之间冲突，也就区分了 shared 类型的lwlock 和 exclusive 类型的 lwlock，并且维护了小型的等待着队列来方便管理同一个共享内存变量类型多次加锁之后解锁的唤醒操作。

更上层的 Regular lock 直接面向的对象是数据库对象（relation/page/tuple 等），因为保护对象的多样性，需要有不同的锁配置才能让在操作不同对象时的性能最大化。这个时候就有了8种锁模式，不同的锁模式对应了不同的数据库对象的操作。那在这样的组合操作场景之下也就意味着实现上需要为性能考虑更多（临界区有大有小，且交织在一起）。所以就有了主锁表，进程锁表，本地锁表以及快速锁机制这几种数据结构的出现，来尽可能提升不同锁操作交织在一起之后的性能。又因为锁交织在了一起，那就有可能产生互相等待的情况，为了稳定性也就有了死锁检测，从而成为PG内部最为复杂的锁机制。虽然spinlock 对性能影响最大，但其简单且没有太多设计融入，而 regular lock 临界区甚至可以大到一张表且因为有需求，所以设计的足够复杂，这里的代码可以说是 PG 1986-2022 37年历史的并发控制精华了。如何在数据库复杂的操作场景下保证访问安全的情况下最大化性能收益，学习 regular lock 就够了:)。