QuanZhou's Wiki

6.s081 Lab7 locks 实现

~/ 6.s081#OS#MIT courses

虽然锁能够解决多进程同步问题,但是多核计算机在高度锁竞争的情况下会出现这种并行性差的现象。在这个 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 实现

图 7-1 Example race

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 上都是并行的,性能是 NN。如果因为锁而串行化,性能就跌回了 11

  • 内核设计需要解决的问题:如何在保证正确性的前提下,尽可能避开锁竞争。

    策略 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)

在实现 acquirerelease 时,代码的执行顺序有着近乎苛刻的要求:

a. 获取锁时:必须先执行 push_off(关中断),然后再设置 lk->locked = 1

- _理由_:如果先设锁后关中断,中间存在一个极小的“空窗期”。如果中断在此瞬间发生,上述的单核死锁依然会重演。

b. 释放锁时:必须先设置 lk->locked = 0,然后再执行 pop_off(开中断)。

- _理由_:如果先开中断后释放锁,中断处理程序(如时钟中断)可能在锁还没释放时就介入并尝试拿锁。

五、指令和内存顺序

在单线程环境下,程序员通常认为代码是按顺序执行的。然而在多核并发环境下,这个直觉是错误的。为了极致的性能,编译器和处理器都会对指令进行重排。

对于单线程代码,指令 A 在指令 B 之前,我们就认为 A 先于 B 发生。但在共享内存的多线程环境中,由于以下两个原因,这种顺序无法保证:

  1. 编译器优化: 编译器为了减少内存访问或利用寄存器,可能会改变指令生成的顺序,甚至完全省略某些读写操作(例如将数据缓存在寄存器中而不写回内存)。
  2. 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)

  1. __sync_synchronize()

这是 xv6 使用的原语(由 GCC 提供)。它告诉编译器和 CPU:严禁将屏障前后的读写指令进行越界重排。

屏障之前的指令必须在屏障之后的指令开始前完成。

它强迫 CPU 将缓存中的数据写回内存,或使缓存失效,确保多核之间的数据可见性。

  1. xv6 中的应用

xv6 在 acquirerelease 中都使用了内存屏障:

acquire 中:屏障确保了在锁被真正拿到之后,临界区内的读写操作才开始。

release 中:屏障确保了临界区内所有的修改都在锁释放之前已经全局可见。

xv6 的锁机制通过 __sync_synchronize() 强制建立了执行顺序。在绝大多数情况下,只要你正确地在共享数据周围使用 acquirerelease,内存屏障就会帮你处理好所有复杂的硬件同步问题。

六、睡眠锁 (Sleep locks)

在内核开发中,并非所有的临界区都是几条指令就能完成的。当我们需要进行磁盘读写等耗时操作(可能长达数十毫秒)时,传统的自旋锁(Spinlock)就会引发严重的性能问题。xv6 引入了睡眠锁(Sleep-lock)来应对这种长任务。

自旋锁的设计初衷是保护短小的临界区。如果将自旋锁用于长时间操作(如等待磁盘),会产生两个致命缺陷:

  1. CPU 资源的极大浪费: 如果进程 A 拿着自旋锁去读磁盘(耗时 10ms),进程 B 尝试获取该锁时会一直自旋。在这 10ms 内,进程 B 白白浪费了数百万次 CPU 指令周期,却什么也没做。
  2. 禁止让出 CPU 的死锁风险: 在持有自旋锁时禁止调用 yield() 让出 CPU
    • 为什么? 假设线程 T1 持有自旋锁并 Yield 了。调度器运行了线程 T2,此时 T2 尝试获取同一把自旋锁。由于 acquire 不会主动让出 CPU,T2 会一直在 CPU 上自旋。这可能导致 T1 永远无法被重新调度回来释放锁,系统彻底卡死。
    • 中断冲突:持有自旋锁要求关闭中断,而 Yield 操作(涉及调度和时钟中断)与此矛盾。

为了允许在持有锁时进行长时间等待,并允许其他进程在此期间使用 CPU,xv6 提供了睡眠锁来进行优雅的等待。

核心机制

  • 主动让出:当一个进程无法获取睡眠锁时,它不会自旋,而是调用 sleep 释放 CPU,进入等待队列。
  • 中断友好:睡眠锁在持有期间保持中断开启
  • 低层支撑:睡眠锁内部使用了一个 locked 字段,而这个字段本身是由一个自旋锁保护的。这体现了内核设计中“小锁保护大锁”的思想。

acquiresleep 的工作逻辑

调用 acquiresleep 时,它会:

  1. 获取内部的自旋锁以检查 locked 状态。
  2. 如果锁已被占用,调用 sleep 原子性地释放自旋锁并让出 CPU。
  3. 当锁被释放时,进程被唤醒,重新获取自旋锁,设置 locked,最后释放内部自旋锁。

理解两者的边界是内核编程的基本功:

特性自旋锁 (Spinlock)睡眠锁 (Sleep-lock)
等待方式忙等待 (Spinning),浪费 CPU让出 CPU (Yielding),允许他人运行
中断状态必须关闭中断保持中断开启
适用场景短时间、高频操作长时间(如磁盘 I/O)操作
中断处理程序可以使用绝对禁止(因为中断处理程序不能 sleep)
嵌套规则不能在持有期间调用 yield可以在持有期间获取自旋锁

睡眠锁的引入,本质上是操作系统在正确性、响应速度和 CPU 利用率之间做的又一次权衡:

  • 自旋锁:它不容忍任何延迟,适用于多核之间极短的同步。
  • 睡眠锁:它允许内核在等待缓慢的硬件时,去做更有意义的事情。

6.s081 Lab7: locks 实现

一、Memory allocator (moderate)

在 kernel/kalloc.c 中,进行内存分配时,所有cpu在运行 kallockfree 会获取 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_offpop_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. 快路径

此路径下,不同桶之间的访问是零竞争的,这是性能提升的核心。

  • 计算块号的哈希值:

    id=blocknomodNBUCKET \mathtt{id} = \mathtt{blockno} \bmod \mathtt{NBUCKET}
  • 仅获取当前桶的锁 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;也意识到了还可以使用以前从未使用过的视角去看待和编写程序:不变量。收获颇丰!