6.s081 Lab7 locks 实现
虽然锁能够解决多进程同步问题,但是多核计算机在高度锁竞争的情况下会出现这种并行性差的现象。在这个 Lab 中,将重新设计代码来提高并行性,涉及到修改数据结构和锁策略来减少争用。
此 Lab 包含两个任务,分别是修改内存分配和缓冲区缓存代码。
xv6-book Chapter 6 & 8 部分内容
一、并发和锁的本质
1. 并发
在操作系统内核中,并发是指多个指令流在时间上相互交叠执行的状态,xv6中会因为以下几种情况产生并发情况:
- 多处理器并行: 多个 CPU 独立运行,同时访问共享的物理内存。
- 线程切换: 即便在单核上,内核通过时分复用在多个线程间切换,造成执行流交替。
- 硬件中断: 异步产生的中断处理程序会打断当前正在执行的内核代码,如果两者操作同一数据,就会发生灾难。
2. 共享内存
- 收益:XV6 利用多核共享 RAM 来维护全局数据结构(如
kmem空闲链表、进程表等),实现了高效的内核协作。 - 风险: 中间状态读取: 当 CPU A 正在更新数据结构到一半时,CPU B 读取了不完整的中间状态。
- 竞争更新: 多个 CPU 同时写同一个变量,导致更新丢失或结构被破坏。
- 结果: 产生不可预知的错误(Incorrect results)或崩溃的结构(Broken data structure)。
3. 并发控制技术
为了在追求高性能(并行性)的同时保证正确性,内核开发者必须设计一套策略和抽象,统称为并发控制技术。
4. 锁
xv6 重点讨论的技术是锁。其核心特征如下:
- 互斥性(Mutual Exclusion): 锁保证了在任何给定时间内,只有一个 CPU 能够持有它。
- 保护机制: 程序员将锁与特定的共享数据项关联。规定:只有持有锁,才能操作数据。
- 代价: 锁会将并行的操作串行化(Serialize)。如果过度使用或设计不当,锁会成为性能瓶颈,限制系统的扩展性。
二、竞态(Races)
书中以 kalloc.c 中的空闲链表(freeList)为例,展示了无保护并发的错误后果。
1. 错误的 push 实现

void push(int data) {
struct element *l;
l = malloc(sizeof *l); // 第 13 行:分配内存
l->data = data; // 第 14 行:初始化数据
l->next = list; // 第 15 行:读取当前链表头
list = l; // 第 16 行:更新链表头
}
假设两个 CPU 同时尝试向链表 push 一个新节点,如图 7-1:
并发失效路径推演:
- 步骤 A:CPU 1 执行第 15 行,记录下
list指向节点A。- 步骤 B:CPU 2 在 CPU 1 执行下条指令前,也执行第 15 行,同样记录下
list指向节点A。- 步骤 C:CPU 1 执行第 16 行,设置
list = l_cpu1。此时l_cpu1->next指向A。- 步骤 D:CPU 2 执行第 16 行,设置
list = l_cpu2。此时l_cpu2->next也指向A。
后果:CPU 1 插入的节点 l_cpu1 被 CPU 2 的写入操作直接**覆盖(Overwrite)**了。l_cpu1 变成了没有任何指针指向的“孤儿节点”,造成了内存泄漏。
2. 锁实现原子性
锁通过提供**互斥(Mutual Exclusion)**来消除竞态。它确保同一时间只有一个 CPU 能进入临界代码段。
- 临界区 (Critical Section):acquire 和 release 之间的代码被称为临界区。锁保护了这段代码所操作的数据。
- 原子性 (Atomicity):通过锁,我们可以让一系列操作相对于其他 CPU 而言变得“不可分割”。其他 CPU 要么看到操作前的状态,要么看到操作完成后的状态,永远看不到中间的混乱状态。
3. 锁保护不变量
这是本节最深刻的内容。锁不仅仅保护数据,它保护的是数据结构的一致性逻辑。
-
什么是“不变量”: 它是数据结构必须始终满足的性质。
- 链表不变量:list 必须指向第一个元素,且每个元素的 next 必须指向其后继。
-
不变量的暂时违反: 在
push函数中,执行完l->next = list后,不变量被暂时破坏了(因为l已经指向了后继,但list还没指向l)。 -
锁的作用: 它确保了当一个 CPU 正在“暂时破坏”不变量时,没有其他 CPU 能够介入。锁将多个并发的临界区串行化(Serialization)。
4. 锁竞争 (Contention)
锁虽然解决了正确性问题,但它会严重影响并行。
-
锁竞争 (Contention):当多个进程同时申请同一把锁时,就发生了竞争。
-
串行化代价:如果
kfree在所有 CPU 上都是并行的,性能是 。如果因为锁而串行化,性能就跌回了 。 -
内核设计需要解决的问题:如何在保证正确性的前提下,尽可能避开锁竞争。
策略 1:细粒度锁 (Fine-grained Locking)。
策略 2:Per-CPU 数据结构。例如:每个 CPU 维护自己的空闲链表(这正是本 Lab 要做的核心任务)。
5. 锁的放置原则
锁的位置直接影响性能。
-
过早加锁:如果在
malloc之前就acquire,会导致所有 CPU 的malloc也被串行化,极大降低性能。 -
准则:只在必须保护不变量的代码段加锁。
三、可重入锁 (Re-entrant locks)
如果一个进程已经拿到了锁,再次申请时直接让它通过不就没那么多死锁了吗?这种机制被称为 可重入锁(Re-entrant Locks) 或 递归锁(Recursive Locks)。然而,xv6 坚决地拒绝了它。
1. 什么是可重入锁?
定义:如果一个锁被某个进程持有,当该进程再次尝试获取该锁时,内核不会让其等待或报错,而是增加一个嵌套计数并允许其继续执行。
目的:看起来它可以避免一些由于函数嵌套调用而产生的死锁,减少程序员对加锁顺序的心理负担。
xv6 认为,可重入锁最大的问题在于它破坏了程序员对“原子性(Atomicity)”的直觉判断。
死锁虽然糟糕,但它比静默错误更好。
-
静默错误(Silent Error):在可重入锁环境下,
call_once()被调用两次可能不会立刻崩溃,但会导致后续数据被破坏。这种 Bug 极其难以追踪和复现。 -
确定性 Panic:非重入锁导致的死锁会立刻让内核挂掉。开发者通过调用栈(Call Stack)能迅速定位到是哪里的嵌套加锁出了问题,并修复代码逻辑。
虽然 xv6 不使用它,但书中也提到了实现它的技术要求:
-
持有者追踪:
acquire必须能够识别当前尝试拿锁的线程是否已经是该锁的持有者。 -
嵌套计数:
struct spinlock需要增加一个计数器(Counter),记录嵌套加锁的层数。这与 xv6 处理中断嵌套的push_off机制非常相似。
四、锁和中断处理
1. 单核自我死锁
在多核内核开发中,最危险的死锁往往不是发生在两个 CPU 之间,而是发生在同一个 CPU 的线程与中断处理程序之间。xv6 通过一套严密的“中断管理协议”化解了这一危机。
考虑 sys_sleep(内核线程)和 clockintr(时钟中断处理程序):
- 数据共享:两者都需要访问全局变量
ticks。 - 锁保护:为了同步,它们都必须获取
tickslock。
假设发生以下序列:
a. CPU 0 上的线程执行 sys_sleep,成功获取了 tickslock。
b. 就在此时,CPU 0 产生了一个时钟中断。
c. 中断处理程序 clockintr 强行打断了 sys_sleep,开始在 CPU 0 上运行。
d. clockintr 尝试获取 tickslock。
e. 僵局:clockintr 发现锁被占用,于是开始“自旋”等待释放。但唯一能释放这把锁的 sys_sleep 已经被它打断了,且只要 clockintr 不返回,sys_sleep 就永远没机会运行。
发生以上情况后,该 CPU 陷入永久死锁,系统彻底卡死。
为了彻底根除此类死锁,xv6 制定了一个硬性规定:
实际上,xv6 采取了更保守(但也更简单稳健)的策略:只要 CPU 获取了任何锁,内核都会立即关闭该 CPU 的本地中断。
注:其他 CPU 的中断依然可以开启。如果 CPU 1 的中断处理程序想要 CPU 0 手里的锁,它只需要在 CPU 1 上自旋等待,这不会导致死锁,因为 CPU 0 可以继续运行直到释放锁。
2. 嵌套保护机制
在复杂的内核逻辑中,临界区经常发生嵌套(例如:函数 A 拿了锁 1,又调用函数 B 拿了锁 2)。简单的 cli/sti(关/开中断)指令无法处理这种情况。
xv6 引入了计数机制来管理中断状态:
push_off:在acquire中调用。它不仅关闭中断,还会递增一个“嵌套计数(Nesting Level)”。如果是第一层嵌套,它会记录进入临界区前的中断开启状态。pop_off:在release中调用。它递减嵌套计数。只有当计数归零时,它才会根据记录的原始状态重新开启中断。
3.绝对的时序要求 (Strict Ordering)
在实现 acquire 和 release 时,代码的执行顺序有着近乎苛刻的要求:
a. 获取锁时:必须先执行 push_off(关中断),然后再设置 lk->locked = 1。
- _理由_:如果先设锁后关中断,中间存在一个极小的“空窗期”。如果中断在此瞬间发生,上述的单核死锁依然会重演。
b. 释放锁时:必须先设置 lk->locked = 0,然后再执行 pop_off(开中断)。
- _理由_:如果先开中断后释放锁,中断处理程序(如时钟中断)可能在锁还没释放时就介入并尝试拿锁。
五、指令和内存顺序
在单线程环境下,程序员通常认为代码是按顺序执行的。然而在多核并发环境下,这个直觉是错误的。为了极致的性能,编译器和处理器都会对指令进行重排。
对于单线程代码,指令 A 在指令 B 之前,我们就认为 A 先于 B 发生。但在共享内存的多线程环境中,由于以下两个原因,这种顺序无法保证:
- 编译器优化: 编译器为了减少内存访问或利用寄存器,可能会改变指令生成的顺序,甚至完全省略某些读写操作(例如将数据缓存在寄存器中而不写回内存)。
- CPU 乱序执行 (Out-of-order Execution): 现代 CPU 会分析指令流。如果指令 A 和 B 互不依赖(例如 A 操作寄存器 X,B 操作寄存器 Y),CPU 可能会先启动 B,或者让 A 和 B 的执行重叠,以提高流水线效率。
书中给出了一个关于 push 操作的经典反面教材。假设重排发生了:
1 l = malloc(sizeof *l);
2 l->data = data;
3 acquire(&listlock);
4 l->next = list;
5 list = l;
6 release(&listlock);
如果 CPU 或编译器将第 4 行的存储操作移到了第 6 行(release)之后:
后果:会产生一个恐怖的“空窗期”。另一个 CPU 获取了锁并看到了更新后的 list(因为第 5 行已执行),但它看到的 list->next 却是未初始化的垃圾值(因为第 4 行被推迟到了锁外面)。
锁的互斥性虽然还在,但由于指令“越界”,锁保护的不变量被破坏了。
内存模型与内存屏障 (Memory Barriers)
为了解决这个问题,硬件和编译器提供了一套规则,称为内存模型 (Memory Model),以及一种特殊的指令——内存屏障 (Memory Barrier)。
- __sync_synchronize()
这是 xv6 使用的原语(由 GCC 提供)。它告诉编译器和 CPU:严禁将屏障前后的读写指令进行越界重排。
屏障之前的指令必须在屏障之后的指令开始前完成。
它强迫 CPU 将缓存中的数据写回内存,或使缓存失效,确保多核之间的数据可见性。
- xv6 中的应用
xv6 在 acquire 和 release 中都使用了内存屏障:
在 acquire 中:屏障确保了在锁被真正拿到之后,临界区内的读写操作才开始。
在 release 中:屏障确保了临界区内所有的修改都在锁释放之前已经全局可见。
xv6 的锁机制通过 __sync_synchronize() 强制建立了执行顺序。在绝大多数情况下,只要你正确地在共享数据周围使用 acquire 和 release,内存屏障就会帮你处理好所有复杂的硬件同步问题。
六、睡眠锁 (Sleep locks)
在内核开发中,并非所有的临界区都是几条指令就能完成的。当我们需要进行磁盘读写等耗时操作(可能长达数十毫秒)时,传统的自旋锁(Spinlock)就会引发严重的性能问题。xv6 引入了睡眠锁(Sleep-lock)来应对这种长任务。
自旋锁的设计初衷是保护短小的临界区。如果将自旋锁用于长时间操作(如等待磁盘),会产生两个致命缺陷:
- CPU 资源的极大浪费: 如果进程 A 拿着自旋锁去读磁盘(耗时 10ms),进程 B 尝试获取该锁时会一直自旋。在这 10ms 内,进程 B 白白浪费了数百万次 CPU 指令周期,却什么也没做。
- 禁止让出 CPU 的死锁风险:
在持有自旋锁时禁止调用
yield()让出 CPU。- 为什么? 假设线程 T1 持有自旋锁并 Yield 了。调度器运行了线程 T2,此时 T2 尝试获取同一把自旋锁。由于
acquire不会主动让出 CPU,T2 会一直在 CPU 上自旋。这可能导致 T1 永远无法被重新调度回来释放锁,系统彻底卡死。 - 中断冲突:持有自旋锁要求关闭中断,而 Yield 操作(涉及调度和时钟中断)与此矛盾。
- 为什么? 假设线程 T1 持有自旋锁并 Yield 了。调度器运行了线程 T2,此时 T2 尝试获取同一把自旋锁。由于
为了允许在持有锁时进行长时间等待,并允许其他进程在此期间使用 CPU,xv6 提供了睡眠锁来进行优雅的等待。
核心机制
- 主动让出:当一个进程无法获取睡眠锁时,它不会自旋,而是调用
sleep释放 CPU,进入等待队列。 - 中断友好:睡眠锁在持有期间保持中断开启。
- 低层支撑:睡眠锁内部使用了一个
locked字段,而这个字段本身是由一个自旋锁保护的。这体现了内核设计中“小锁保护大锁”的思想。
acquiresleep 的工作逻辑
调用 acquiresleep 时,它会:
- 获取内部的自旋锁以检查
locked状态。 - 如果锁已被占用,调用
sleep原子性地释放自旋锁并让出 CPU。 - 当锁被释放时,进程被唤醒,重新获取自旋锁,设置
locked,最后释放内部自旋锁。
理解两者的边界是内核编程的基本功:
| 特性 | 自旋锁 (Spinlock) | 睡眠锁 (Sleep-lock) |
|---|---|---|
| 等待方式 | 忙等待 (Spinning),浪费 CPU | 让出 CPU (Yielding),允许他人运行 |
| 中断状态 | 必须关闭中断 | 保持中断开启 |
| 适用场景 | 短时间、高频操作 | 长时间(如磁盘 I/O)操作 |
| 中断处理程序 | 可以使用 | 绝对禁止(因为中断处理程序不能 sleep) |
| 嵌套规则 | 不能在持有期间调用 yield | 可以在持有期间获取自旋锁 |
睡眠锁的引入,本质上是操作系统在正确性、响应速度和 CPU 利用率之间做的又一次权衡:
- 自旋锁:它不容忍任何延迟,适用于多核之间极短的同步。
- 睡眠锁:它允许内核在等待缓慢的硬件时,去做更有意义的事情。
6.s081 Lab7: locks 实现
一、Memory allocator (moderate)
在 kernel/kalloc.c 中,进行内存分配时,所有cpu在运行 kalloc 和 kfree 会获取 kmem.lock 这个全局内存锁,因此当多个 CPU 同时进行内存分配时会产生竞态。
这个任务需要让每个 CPU 有自己的 freelist,用粒度更小的锁来进行同步,可以很大程度上提高并行性。此外还要实现 steal 机制,当一个 CPU 的 freelist 中没有页表可以分配时需要去其他 CPU 的 freeList 中窃取,而不是原地等待自己的其他进程释放页表,这样可以防止内存资源的浪费。
根据上面的思路还需要进行一点补充:对每个 CPU 中的 freelist 各用一把锁去进行同步。
我的实现会先将所有的页表全都分配到 CPU 0 的 freelist 上,虽然这会造成在 xv6 开机分配页表时会频繁发生 steal,但经过测试,这个策略可以有更优的开机性能,因为一定程度上消除了进程切换产生的竞态。
void
kinit()
{
for (int i = 0; i < NCPU; i++) {
char lock_name[8];
snprintf(lock_name, 8, "kmem%d", i);
initlock(&kmem[i].lock, lock_name);
}
freerange(end, (void*)PHYSTOP);
}
void
freerange(void *pa_start, void *pa_end)
{
char *p;
p = (char*)PGROUNDUP((uint64)pa_start);
for(; p + PGSIZE <= (char*)pa_end; p += PGSIZE) {
kfree(p);
}
}
void *
kalloc(void)
{
struct run *r;
push_off();
int hart = cpuid();
acquire(&kmem[hart].lock);
r = kmem[hart].freelist;
if(r) {
kmem[hart].freelist = r->next;
release(&kmem[hart].lock);
}
else {
release(&kmem[hart].lock);
for (int i = 0; i < NCPU; i++) {
if (i != hart) {
acquire(&kmem[i].lock);
r = kmem[i].freelist;
if (r) {
kmem[i].freelist = r->next;
release(&kmem[i].lock);
break;
}
release(&kmem[i].lock);
}
}
}
pop_off();
if(r)
memset((char*)r, 1, PGSIZE); // fill with junk
return (void*)r;
}
void
kfree(void *pa)
{
struct run *r;
if(((uint64)pa % PGSIZE) != 0 || (char*)pa < end || (uint64)pa >= PHYSTOP)
panic("kfree");
// Fill with junk to catch dangling refs.
memset(pa, 1, PGSIZE);
r = (struct run*)pa;
push_off();
int hart = cpuid();
acquire(&kmem[hart].lock);
r->next = kmem[hart].freelist;
kmem[hart].freelist = r;
release(&kmem[hart].lock);
pop_off();
}
这个任务也用到了 xv6 book 6.6 中提到的 push_off 和 pop_off 保证调用 cpuid() 获取 hartid 时的安全。
二、Buffer cache (hard)
在 xv6 原生的实现中,所有的块缓存(Buffer Cache)都受一把全局大锁 bcache.lock 保护。这在单核时代相安无事,但在多核(Multiprocessor)环境下,它很大程度上成为系统性能的枷锁。
当 CPU 0 在读磁盘块 A,CPU 1 想读磁盘块 B 时,明明两者互不干扰,CPU 1 却必须在 bcache.lock 上疯狂自旋等待。我们的任务就是:打破大锁,通过哈希表和细粒度锁(Fine-grained Locking)重塑 bio.c。
为了降低竞争,我们将原来单一的双向链表结构拆解为带桶锁的哈希表。
我选择了 13 个桶(NBUCKET=13,质数有助于减少哈希冲突)。
每个桶拥有独立的自旋锁:这样访问不同桶的进程可以完全并行。
时间戳取代全局链表:原版靠 head 链表位置维护 LRU(最久未使用),但在高并发下,频繁移动链表节点会引发复杂的锁竞争。我们改为在 struct buf 中记录 timestamp,通过扫描时间戳来寻找驱逐对象。
struct {
struct spinlock lock;
struct buf buf[NBUF];
struct spinlock bucket_locks[NBUCKET];
struct buf buckets[NBUCKET];
} bcache;
bget 是整个实验逻辑最复杂的地方,我将其拆分为快路径(命中)与慢路径(驱逐)。
1. 快路径
此路径下,不同桶之间的访问是零竞争的,这是性能提升的核心。
-
计算块号的哈希值:
-
仅获取当前桶的锁
bucket_locks[id]。 -
遍历桶内链表,若命中,
refcnt++,解锁并返回。
2. 慢路径
当缓存未命中时,我们需要找到一个 refcnt == 0 且 timestamp 最小的旧块进行替换。
-
全局串行化:为了避免多个 CPU 同时驱逐同一个块导致数据混乱,我们引入一把全局锁 bcache.lock(或称 eviction_lock)来串行化驱逐过程。
-
重新验证:在拿到全局锁后,必须重新检查目标块是否已被其他 CPU 载入,防止重复分配(在放弃原本桶的锁后,会有短暂的空窗期,这个期间可能要寻找的块已经被调入缓存中)。
-
跨桶搬家:若选中的旧块在桶 A,新块在桶 B,需要处理复杂的指针迁移。
在处理“跨桶搬家”时,我们可能需要同时持有两把桶锁。根据 xv6 Book 6.3 的教导,不按顺序加锁必死锁。
我的策略:
-
持有者校验:如果旧桶和新桶是同一个(
old_id == id),绝对不能重复 acquire 同一把锁,否则会触发 panic。 -
乐观并发控制:在持有全局锁扫描 LRU 时,不应长时间持有桶锁。我采用了“扫描 -> 记录 -> 重新加锁验证”的策略。如果在验证阶段发现该块已被使用(
refcnt != 0),则放弃并重试。
void
binit(void)
{
struct buf *b;
initlock(&bcache.lock, "bcache");
for (int i = 0; i < NBUCKET; i++) {
char lock_name[16];
snprintf(lock_name, 16, "bcache_%d", i);
initlock(&bcache.bucket_locks[i], lock_name);
bcache.buckets[i].next = &bcache.buckets[i];
bcache.buckets[i].prev = &bcache.buckets[i];
}
// Create linked list of buffers
for(b = bcache.buf; b < bcache.buf+NBUF; b++){
b->next = bcache.buckets[0].next;
b->prev = &bcache.buckets[0];
initsleeplock(&b->lock, "buffer");
bcache.buckets[0].next->prev = b;
bcache.buckets[0].next = b;
}
}
static struct buf*
bget(uint dev, uint blockno)
{
struct buf *b;
// Is the block already cached?
int id = blockno % NBUCKET;
acquire(&bcache.bucket_locks[id]);
for(b = bcache.buckets[id].next; b != &bcache.buckets[id]; b = b->next){
if(b->dev == dev && b->blockno == blockno){
b->refcnt++;
release(&bcache.bucket_locks[id]);
acquiresleep(&b->lock);
return b;
}
}
release(&bcache.bucket_locks[id]);
// Not cached.
// Recycle the least recently used (LRU) unused buffer.
acquire(&bcache.lock);
acquire(&bcache.bucket_locks[id]);
for(b = bcache.buckets[id].next; b != &bcache.buckets[id]; b = b->next){
if(b->dev == dev && b->blockno == blockno){
b->refcnt++;
release(&bcache.lock);
release(&bcache.bucket_locks[id]);
acquiresleep(&b->lock);
return b;
}
}
release(&bcache.bucket_locks[id]);
uint min_timestamp = ~0;
int old_id = -1;
struct buf *evict_b = 0;
retry_eviction:
min_timestamp = ~0;
old_id = -1;
evict_b = 0;
for (int i = 0; i < NBUCKET; i++) {
acquire(&bcache.bucket_locks[i]);
for(b = bcache.buckets[i].next; b != &bcache.buckets[i]; b = b->next){
if(b->refcnt == 0) {
if (min_timestamp > b->timestamp) {
evict_b = b;
min_timestamp = b->timestamp;
old_id = i;
}
}
}
release(&bcache.bucket_locks[i]);
}
if (!evict_b)
panic("bget: no buffers");
acquire(&bcache.bucket_locks[old_id]);
if (evict_b->refcnt) {
release(&bcache.bucket_locks[old_id]);
goto retry_eviction;
}
evict_b->dev = dev;
evict_b->blockno = blockno;
evict_b->valid = 0;
evict_b->refcnt = 1;
if (old_id != id) {
evict_b->next->prev = evict_b->prev;
evict_b->prev->next = evict_b->next;
acquire(&bcache.bucket_locks[id]);
evict_b->next = bcache.buckets[id].next;
evict_b->prev = &bcache.buckets[id];
bcache.buckets[id].next->prev = evict_b;
bcache.buckets[id].next = evict_b;
release(&bcache.bucket_locks[id]);
}
release(&bcache.lock);
release(&bcache.bucket_locks[old_id]);
acquiresleep(&evict_b->lock);
return evict_b;
}
在做了以上修改之后,我发现依然无法通过压力测试,后来发现我在 brelse() 中依然获取了 bcache.lock 全局锁,因此会与 bget() 的驱逐过程发生竞争,修改后就通过测试,此外也把 bpin() 与 bunpin() 中同样的问题解决(此 Lab 的结果不会被这两个函数影响)。
void
brelse(struct buf *b)
{
if(!holdingsleep(&b->lock))
panic("brelse");
releasesleep(&b->lock);
int id = b->blockno % NBUCKET;
acquire(&bcache.bucket_locks[id]);
b->refcnt--;
if (b->refcnt == 0) {
// no one is waiting for it.
b->timestamp = ticks;
}
release(&bcache.bucket_locks[id]);
}
void
bpin(struct buf *b) {
int id = b->blockno % NBUCKET;
acquire(&bcache.bucket_locks[id]);
b->refcnt++;
release(&bcache.bucket_locks[id]);
}
void
bunpin(struct buf *b) {
int id = b->blockno % NBUCKET;
acquire(&bcache.bucket_locks[id]);
b->refcnt--;
release(&bcache.bucket_locks[id]);
}
细粒度锁虽然提高了并行度,但也带来了代码复杂度的飙升和潜在的死锁风险。不变量(Invariants)是核心:无论锁怎么拆,必须保证“一个磁盘块在缓存中至多只有一个副本”这一不变量不被打破。
这次 Lab 之后终于理解了乐观锁和悲观锁,也学会了一点使用它们的小 trick;也意识到了还可以使用以前从未使用过的视角去看待和编写程序:不变量。收获颇丰!