xv6 Lab7 locks - MIT 6.1810 Fall 2025 Operating System

本文最后更新于 2026年8月18日 晚上

阅读

简单读一下书吧，让 GPT 翻译并提炼重点，然后读对应的部分。

并发的来源：

  • 多核 CPU
  • 线程切换
  • 中断

回顾概念：

  • 竞态条件
  • 临界区

减少锁的竞争：这个 ostep 也有阐述。给整个数据结构加一把大锁固然能解决问题，但是效率低。

关于锁的源码：kernel/spinlock.h kernel/spinlock.c

关于 spinlock 自旋锁

struct spinlock {
  uint locked; // 0 表示没被持有
  char *name;
  struct cpu *cpu;
};

关于得锁，RISC-V 提供原子指令 amoswap，原子地完成读取内存-寄存器写入内存-返回旧值，也就是 acquire() 的实现。释放锁是类似的思想和原理。

Lab 的重点是对 kalloc() 进行优化，把锁的颗粒度拆细。

需要防止死锁。

xv6 的实现里面，只要有任何自旋锁被持有，就关闭 CPU 中断。

acquire() 内部调用：

push_off();

关闭中断。

release() 最后调用：

pop_off();

恢复中断。

关于 sleeplock：因为 spinlock 不能中断，才引入。

使用 acquiresleep() 让出 CPU，内部使用一个 spinlock 进行保护，之后可以 wakeup。

Memory allocator

大致是有一个 kalloctest.c，里面会高强度进行锁竞争（反复调用 kalloc() free()）。

根本原因是kalloc() 只有一个 free list，并且这个 free list 由一个全局锁保护。

减少锁竞争的基本思路是重新设计内存分配器，因此本 lab 的任务是为每个 CPU 单独维护一个 free list，并给每个 list 配一个自己的锁。

所以这个 lab 必须在有多核 CPU 的机器上做。

一个主要难点是某个 CPU 的 free list 已经空了，但另一个 CPU 的 free list 中还有空闲内存，这个时候，缺少内存的 CPU 需要去其他 CPU 的 free list 里面获取一些空闲页面，也就是所谓的 stealing。

现在着手实现：

首先是需要修改 kmem，就是原来空闲链表的结构体（一个 freelist 里面带一个锁）：

原来：

struct {
  struct spinlock lock;
  struct run *freelist;
} kmem;

修改成支持多 CPU 的形式：

struct {
  struct spinlock lock[NCPU];
  struct run *freelist[NCPU];
} kmem;

然后修改 kinit() 等，总之要使用到当前运行的 CPU。

这里首先需要知道获取当前运行的 CPU 的 API，这里 hint 3 也有说，不过还是问 LLM ：

引用核心 API 只有三个：cpuid()、push_off()、pop_off()。 cpuid() 做什么 kernel/proc.c 中： i...

核心 API 只有三个：cpuid()、push_off()、pop_off()。

  1. cpuid() 做什么

kernel/proc.c 中：

int
cpuid()
{
  int id = r_tp();
  return id;
}

每个 CPU 都有一个编号，范围是：

0 到 NCPU - 1

cpuid() 从 RISC-V 的 tp 寄存器读取这个编号。因此，如果当前代码运行在 CPU 3 上：

int id = cpuid();

得到的 id 就是 3，之后可以访问：

kmem.lock[id]
kmem.freelist[id]
  1. 为什么不能直接调用 cpuid()

问题在于：进程可能被调度到另一个 CPU。

设想下面的情况：

int id = cpuid();  // 此时得到 2
// 这里发生时钟中断和重新调度
// 进程后来在 CPU 5 上继续运行

此时变量 id 还是 2，但代码已经运行在 CPU 5 上了。“当前 CPU”与保存的编号不再一致。

因此，cpuid() 的使用约定是：读取和使用 CPU 编号期间必须关闭中断。

源码注释也明确写了这一点：

// Must be called with interrupts disabled,
// to prevent race with process being moved
// to a different CPU.
  1. push_off() 和 pop_off()

安全的基本形式是：

push_off();
int id = cpuid();

// 在这里使用 id

pop_off();
  • push_off()：关闭当前 CPU 的中断。
  • pop_off()：恢复进入这段代码之前的中断状态。

这段时间不会因为时钟中断被抢占，所以当前执行流不会突然迁移到其他 CPU。

不建议直接使用 intr_off() 和 intr_on()，因为它们不能正确处理嵌套。比如自旋锁的 acquire() 本身也会调用 push_off()。

push_off()/pop_off() 会记录嵌套层数，因此下面这样是合法的：

push_off();                 // 第 1 层
int id = cpuid();

acquire(&kmem.lock[id]);    // acquire 内部进入第 2 层
// 修改 freelist
release(&kmem.lock[id]);    // 退出第 2 层

pop_off();                  // 退出第 1 层

在最外层 pop_off() 之前，中断仍然保持关闭。

  1. 为什么不能取得编号后马上 pop_off()

下面这种写法不理想：

push_off();
int id = cpuid();
pop_off();

acquire(&kmem.lock[id]);

因为 pop_off() 之后，代码可能迁移到另一个 CPU，然后操作的就不再是“当前 CPU 的 freelist”。

所以在 kfree() 中，应该把中断关闭范围覆盖到对应 freelist 操作结束：

push_off();
int id = cpuid();

acquire(&kmem.lock[id]);

// 把 r 插入 kmem.freelist[id]

release(&kmem.lock[id]);
pop_off();

这基本就是你下一步需要完成的内容。

  1. freerange() 为什么会把所有页面交给一个 CPU

启动流程大致是：

启动 CPU
  → kinit()
    → 初始化所有 kmem 锁
    → freerange()
      → 对每个物理页调用 kfree()

freerange() 本身不需要知道 CPU 编号：

for(; p + PGSIZE <= (char*)pa_end; p += PGSIZE)
  kfree(p);

真正选择 freelist 的是 kfree()。

启动时，kinit() 只在启动 CPU 上执行，通常就是 CPU 0。因此，每次 kfree(p) 取得的 CPU 编号都是 0，最终效果是：

kmem.freelist[0]：拥有全部初始空闲页
kmem.freelist[1]：空
kmem.freelist[2]：空
...

这就是 hint 所说的：

让 freerange() 把所有空闲内存先交给当前正在运行 freerange() 的 CPU。

这不是要求修改 freerange()，而是要求 kfree() 根据当前 CPU 选择 freelist。之后其他 CPU 第一次分配内存时发现自己的 freelist 为空，再通过 stealing 从 CPU 0 获取页面。

你现在可以先只改 kfree()，暂时不管 kalloc() 和 stealing。改动目标就是把原来的：

kmem.lock
kmem.freelist

换成由安全取得的 id 索引的：

kmem.lock[id]
kmem.freelist[id]

所以 kfree() 和 kalloc() 这样改：

代码C · 37 行
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 id = cpuid();
  acquire(&kmem.lock[id]);
  r->next = kmem.freelist[id];
  kmem.freelist[id] = r;
  release(&kmem.lock[id]);
  pop_off();
}

void *
kalloc(void)
{
  struct run *r;
  push_off();
  int id = cpuid();
  acquire(&kmem.lock[id]);
  r = kmem.freelist;
  if(r)
    kmem.freelist[id] = r->next;
  release(&kmem.lock[id]);
  pop_off();
  if(r)
    memset((char*)r, 5, PGSIZE); // fill with junk
  return (void*)r;
}

但是这样还没有实现 stealing。

修改 kalloc()：

代码C · 33 行
void *
kalloc(void)
{
  struct run *r;
  push_off();
  int id = cpuid();
  acquire(&kmem.lock[id]);
  r = kmem.freelist[id];
  if(r)
    kmem.freelist[id] = r->next;
  release(&kmem.lock[id]);
  if (r == 0) {
    for (int i = 0; i < NCPU; ++i) {
      // 找到序号最小的有空余页的空闲列表
      if (i == id) {
        continue;
      }
      acquire(&kmem.lock[i]);
      if (kmem.freelist[i]) {
        r = kmem.freelist[i];
        kmem.freelist[i] = r->next;
      }
      release(&kmem.lock[i]);
      if (r) {
        break;
      }
    }
  }
  pop_off();
  if(r)
    memset((char*)r, 5, PGSIZE); // fill with junk
  return (void*)r;
}

然后可以通过所有测试。

Read-write lock

考虑 xv6 中的：

sys_pause()
sys_uptime()

这两个函数都会读取全局变量 ticks。

但是 ticks 会同时被 clockintr() 更新，所以两个函数读取 ticks 之前会获取 tickslock。但是其实只需要持写锁就行了，完全没必要设读相关的锁。这个 lab 就是区分读者和写者。

复习一下 ostep 的内容，读写者问题里面，规则是：

  • 同一时间最多只能有 一个 writer；
  • 存在 writer 时，不能存在 reader；
  • 如果没有 writer，则可以同时存在 多个 reader。
  • 为了防止 reader 一直读导致 writer 饿死，设置 writer priority，即一旦有 writer 开始尝试获取锁，那么之后到来的 reader 必须等待，直到这个 writer 成功获得锁并释放它。

read-write lock API (in kernel/defs.h)：

void initrwlock(struct rwspinlock*);
void read_acquire(struct rwspinlock*);
void read_release(struct rwspinlock*);
void write_acquire(struct rwspinlock*);
void write_release(struct rwspinlock*);

本 lab 需要补全 kernel/spinlock.c 中 read-write spinlock API 对应的 stub 函数，以及修改 kernel/spinlock.h 中的 struct rwspinlock 定义。

hints 说要读 sys_rwlktest() 函数了解测试用例，其实大概就是一直验证读和写（上述的四条规则）。

从测试可以反推，这个锁至少需要记录：

  • 读者数
  • 当前是否有 active reader
  • 当前 waiting writer 的数量

同时，需要保证状态的转换都是原子的。

// Reader-writer lock.
struct rwspinlock {
  struct spinlock l; // state lock
  int readers;
  int waiting_writers;
  int is_writer_active;
};

hints 说：

如果你什么都不修改，直接在 xv6 中运行 rwlktest，内核会打印：

panic: acquire

原因是原有 spinlock 的实现只允许某一时刻由一个 thread 持有锁。

你需要替换：

write_acquire_inner
write_release_inner
read_acquire_inner
read_release_inner

中对 acquire() 和 release() 的调用，改为你自己的锁实现。

实现的部分：

代码C · 60 行
static void
read_acquire_inner(struct rwspinlock *rwlk)
{
  while (1) {
    acquire(&rwlk->l);
    if (rwlk->is_writer_active == 0 && rwlk->waiting_writers == 0) {
      rwlk->readers++;
      release(&rwlk->l);
      return;
    }
    release(&rwlk->l);
  }
}

static void
read_release_inner(struct rwspinlock *rwlk)
{
  while (1) {
    acquire(&rwlk->l);
    if (rwlk->readers > 0) {
      rwlk->readers--;
      release(&rwlk->l);
      return;
    }
    release(&rwlk->l);
  }
}

static void
write_acquire_inner(struct rwspinlock *rwlk)
{
  // 首先这个肯定不能放到循环里面，其次这个务必要加锁
  acquire(&rwlk->l);
  rwlk->waiting_writers++;
  release(&rwlk->l);
  while (1) {
    acquire(&rwlk->l);
    if (rwlk->readers == 0 && rwlk->is_writer_active == 0) {
      rwlk->waiting_writers--;
      rwlk->is_writer_active = 1;
      release(&rwlk->l);
      return;
    }
    release(&rwlk->l);
  }
}

static void
write_release_inner(struct rwspinlock *rwlk)
{
  while (1) {
    acquire(&rwlk->l);
    if (rwlk->is_writer_active) {
      rwlk->is_writer_active = 0;
      release(&rwlk->l);
      return;
    }
    release(&rwlk->l);
  }
}

2 个注意点：

  1. 务必要加循环，因为是一直去尝试获得，考虑到这是自旋锁
    • 我开始是读了 ostep 的实现读写锁的部分没看到，我后面才意识到人家是用的信号量机制，而信号量机制底层还是读写锁。我们实现的这些数据，其实就有点像是实现了一个信号量机制。
  2. write_acquire_inner(*struct* rwspinlock **rwlk*) 函数，waiting_writers 变量的添加，肯定是要放在循环外面并且是函数开头的，因为这表示了开始等待，而自旋空转就是等待的过程。其次显然要加锁，这里开始忘记加了，test8过不去。
通过所有测试