MIT 6.S081 Lab8: locks_综合

介绍

写在前面

最近有点忙，。。拖了一周多的时间才开始写这个实验。。

这次实验的目的是设计锁，主要目的是降低多线程情况下对锁的竞争，实验地址在 Lab: locks。实验的整体思路还是比较简单的：就是降低锁的粒度，将一个大锁更换为一些粒度小的锁，这样可以大幅度降低锁的竞争。我的代码在 github。

关于这次实验，对应的讲义没有太多有帮助的内容，可以直接上手做实验。我是先看了一下一遍对应的课程，虽然跟这次实验关系不是很大，但还是在下面记录一下。

课程内容

线程切换到调度器线程的执行过程

先获取自己的锁；
将自己的状态从 RUNNING 设置为 RUNNABLE；
调用switch函数（其实是调用 sched 函数，在 sched 函数中再调用的 switch 函数）；
switch 函数将当前的线程切换到调度器线程；
调度器线程之前也调用了switch函数，现在恢复执行会从自己的switch函数返回；
返回之后，调度器线程会释放刚刚出让了CPU的进程的锁。

最开始获取锁的时候，是因为防止其他 cpu 遍历进程表单，检测到该线程的状态为 RUNNABLE，进而执行该线程。在调用 switch 函数时也不释放锁。而是在调度器线程中，在进程的线程完全停止使用自己的栈之后，再释放进程的锁。

注意：在执行switch函数的过程中，不能持有任何其他的锁。只能持有一个该进程的锁。（在XV6中，死锁是通过禁止在线程切换的时候加锁来避免的。）

Sleep&Wakeup 接口

当线程利用 sleep 出让 cpu 进行等待时，线程调用sleep函数并等待一个特定的条件。当特定的条件满足时，代码会调用wakeup函数。这里的sleep函数和wakeup函数是成对出现的， sleep函数和wakeup函数都带有一个叫做sleep channel的参数。

Lost wakeup

简单的说，就是只含有一个 sleep channel 参数的sleep是有缺陷的：

在 sleep 的时候是否应该获取共享变量的锁，如果获取的话，其他线程不能获取锁来对其进行改变，永远停留在判断共享变量的循环上；
另一种方案是在 sleep 之前先释放锁，在返回后，再获取锁，这样保护了共享变量；但一旦释放了锁，当前CPU的中断会被重新打开，会导致在该线程还没 sleep 的时候，对应的 wakeup 已经执行了，这时就会使得 sleep 不会被唤醒，导致 Lost wakeup 事件发生。

解决 Lost wakeup 的方法是：让 sleep 函数带有锁作为参数。这样的话，线程会先释放这个锁，同时为了避免在还没sleep的时候wakeup就执行了，先获取对应进程的锁，然后再释放共享锁；在设置好进程的状态后，释放进程的锁，同时再获取共享变量的锁。

关闭一个进程

exit系统调用

exit 会释放进程的内存和page table，关闭已经打开的文件，父进程也会从wait系统调用中唤醒，所以exit最终会导致父进程被唤醒。由父进程调用的freeproc函数，来完成释放进程资源的最后几个步骤。

如果一个父进程要退出，需要设置它的子进程的父进程为 init 进程。

kill系统调用

kill 本身并没有做什么事情，主要是扫描进程表单，找到目标进程。然后只是将进程的proc结构体中killed标志位设置为1。如果进程正在SLEEPING状态，将其设置为RUNNABLE。

Sleep Lock

磁盘块缓存时，使用的锁是不一样的，而是 Sleep Lock 这个结构。sleep lock是基于spinlock实现的，由于 spinlock 加锁时中断必须要关闭。而且磁盘读取数据需要很长时间。sleep lock的优势就是，可以在持有锁的时候不关闭中断。（这块还是有点懵，等以后再瞅瞅，。）

实验内容

任务一（Memory allocator）

第一个任务是解决内存块的竞争问题。目前的内存块的结构是利用一个大锁锁住一个 freelist 的方式。按照实验指导书所说，我们需要将 freelist 进行拆分，给每个 cpu 分配一个freelist，来降低竞争。

第一步，更改内存块结构

NCPU 在 kernel/param.h 中进行了定义。

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

第二步，修改 kinit

将所有空闲内存全部分配给当前 cpu 的 freelist。

void
kinit()
{
    for (int i = 0; i < NCPU; i++)initlock(&kmem[i].lock, "kmem");freerange(end, (void*)PHYSTOP);
}

第三步，修改 kfree

获取 cpuid 的时候需要关闭中断。

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();pop_off();acquire(&kmem[id].lock);r->next = kmem[id].freelist;kmem[id].freelist = r;release(&kmem[id].lock);
}

第四步，修改 kalloc

如果当前 cpu 有空闲内存块，就直接返回；没有的话，从其他 cpu 对应的 freelist 中“偷”一块。

void *
kalloc(void)
{
    struct run *r;push_off();int id = cpuid();pop_off();acquire(&kmem[id].lock);r = kmem[id].freelist;if(r)kmem[id].freelist = r->next;else {
    for (int i = 0; i < NCPU; i++) {
    if (i == id) continue;acquire(&kmem[i].lock);r = kmem[i].freelist;if(r)kmem[i].freelist = r->next;release(&kmem[i].lock);if(r) break;}}release(&kmem[id].lock);if(r)memset((char*)r, 5, PGSIZE); // fill with junkreturn (void*)r;
}

任务二（Buffer cache）

任务二的目的是解决 cache 缓存竞争问题。可以参考任务一，设计小粒度的锁。这里缓存区大小是固定的，不好随意更改。而且，由于题目要求使用 ticks 的方式代替现有的 LRU 机制，因此设计时复杂一些。

第一步，修改结构

也是同样，先修改数据结构，将 buf 分成 13 份（实验指导书上建议使用的质数），同时获取 trap.c 中的 ticks 变量。

这里不仅需要所有 bucket 的小锁，还需要一个大的锁防止死锁。不能像任务一那样直接使用 bcache 数组，这会改变 buf 的大小。

extern uint ticks;struct {
    struct spinlock biglock;struct spinlock lock[NBUCKET];struct buf buf[NBUF];// Linked list of all buffers, through prev/next.// Sorted by how recently the buffer was used.// head.next is most recent, head.prev is least.struct buf head[NBUCKET];
} bcache;

为 kernel/buf.h 中添加 lastuse 字段，便于使用 LRU 机制：

struct buf {
    int valid;   // has data been read from disk?int disk;    // does disk "own" buf?uint dev;uint blockno;struct sleeplock lock;uint refcnt;struct buf *prev; // LRU cache liststruct buf *next;uchar data[BSIZE];uint lastuse;
};

第二步，初始化

原本的 LRU 机制是使用双向链表来实现的，在这里由于使用了 ticks，因此可以不用这种方式实现，直接单向链表即可。但我后面写的时候，发现单向链表不方便删除 block，因此这里依然使用了双向的结构。

void
binit(void)
{
    struct buf *b;initlock(&bcache.biglock, "bcache_biglock");for (int i = 0; i < NBUCKET; i++)initlock(&bcache.lock[i], "bcache");// Create linked list of buffers//bcache.head.prev = &bcache.head;//bcache.head.next = &bcache.head;for (int i = 0; i < NBUCKET; i++) {
    bcache.head[i].next = &bcache.head[i];bcache.head[i].prev = &bcache.head[i];}for(b = bcache.buf; b < bcache.buf+NBUF; b++){
    b->next = bcache.head[0].next;b->prev = &bcache.head[0];initsleeplock(&b->lock, "buffer");bcache.head[0].next->prev = b;bcache.head[0].next = b;}
}

第三步，修改 brels，bpin和bunpin

这些函数修改起来比较简单，直接加锁即可。brelse 中，由于不使用之前的方式实现 LRU，因此当遇到空闲块的时候，直接设置它的使用时间即可。

int
hash(int blockno)
{
    return blockno % NBUCKET;
}
void
brelse(struct buf *b)
{
    if(!holdingsleep(&b->lock))panic("brelse");releasesleep(&b->lock);int i = hash(b->blockno);acquire(&bcache.lock[i]);b->refcnt--;if (b->refcnt == 0) {
    // no one is waiting for it.//b->next->prev = b->prev;//b->prev->next = b->next;//b->next = bcache.head[i].next;//b->prev = &bcache.head[i];//bcache.head[i].next->prev = b;//bcache.head[i].next = b;b->lastuse = ticks;}release(&bcache.lock[i]);
}void
bpin(struct buf *b) {
    int i = hash(b->blockno);acquire(&bcache.lock[i]);b->refcnt++;release(&bcache.lock[i]);
}void
bunpin(struct buf *b) {
    int i = hash(b->blockno);acquire(&bcache.lock[i]);b->refcnt--;release(&bcache.lock[i]);
}

第四步，bget

这是最关键、最核心的一步。这里的步骤：

首先还是判断是否命中，如果已经缓存好，皆大欢喜，直接返回；
如果没找到，释放锁，按顺序先获取大锁，再获取小锁，避免死锁；这时由于可能释放锁后，又可能会有缓存，因此再遍历一遍；
如果还没命中，就去寻找当前 bucket 对应的 LRU 的空闲块，使用 ticks 的方式寻找，如果找到了，就返回；
如果还没找到，需要向其他 bucket 中拿内存块。

代码如下：

static struct buf*
bget(uint dev, uint blockno)
{
    struct buf *b, *b2 = 0;int i = hash(blockno), min_ticks = 0;acquire(&bcache.lock[i]);// 1. Is the block already cached?for(b = bcache.head[i].next; b != &bcache.head[i]; b = b->next){
    if(b->dev == dev && b->blockno == blockno){
    b->refcnt++;release(&bcache.lock[i]);acquiresleep(&b->lock);return b;}}release(&bcache.lock[i]);// 2. Not cached.acquire(&bcache.biglock);acquire(&bcache.lock[i]);// 2.1 find from current bucket.for (b = bcache.head[i].next; b != &bcache.head[i]; b = b->next) {
    if(b->dev == dev && b->blockno == blockno) {
    b->refcnt++;release(&bcache.lock[i]);release(&bcache.biglock);acquiresleep(&b->lock);return b;}}// 2.2 find a LRU block from current bucket.for (b = bcache.head[i].next; b != &bcache.head[i]; b = b->next) {
    if (b->refcnt == 0 && (b2 == 0 || b->lastuse < min_ticks)) {
    min_ticks = b->lastuse;b2 = b;}}if (b2) {
    b2->dev = dev;b2->blockno = blockno;b2->refcnt++;b2->valid = 0;//acquiresleep(&b2->lock);release(&bcache.lock[i]);release(&bcache.biglock);acquiresleep(&b2->lock);return b2;}// 2.3 find block from the other buckets.for (int j = hash(i + 1); j != i; j = hash(j + 1)) {
    acquire(&bcache.lock[j]);for (b = bcache.head[j].next; b != &bcache.head[j]; b = b->next) {
    if (b->refcnt == 0 && (b2 == 0 || b->lastuse < min_ticks)) {
    min_ticks = b->lastuse;b2 = b;}}if(b2) {
    b2->dev = dev;b2->refcnt++;b2->valid = 0;b2->blockno = blockno;// remove block from its original bucket.b2->next->prev = b2->prev;b2->prev->next = b2->next;release(&bcache.lock[j]);// add blockb2->next = bcache.head[i].next;b2->prev = &bcache.head[i];bcache.head[i].next->prev = b2;bcache.head[i].next = b2;release(&bcache.lock[i]);release(&bcache.biglock);acquiresleep(&b2->lock);return b2;}release(&bcache.lock[j]);}release(&bcache.lock[i]);release(&bcache.biglock);panic("bget: no buffers");
}

过程有点复杂，但知道思路后，写起来还挺快的。

总结

任务一思路很简单，做起来很快。

任务二，其实也可以简单地做，直接像任务一一样，如果未缓存，就从其他 bucket 中直接拿一个未使用的。但题目要求我们需要利用 LRU 机制，而且不使用原来的方式，而是用 ticks 的方式来替换，因此设计起来复杂了一些。

最后想吐槽的是，有一部分直接使用 ./make-grade-lock 命令是不行的，，这里并没有清楚上一次的错误，导致我改了一晚上。。。一直出现 panic:freeing free block 错误。。。需要先 make clean 才行，。

感觉可以优化一下查找速度，改 bug 改得我很烦躁，不想优化了。。

文章同步在知乎。

参考文章

MIT 6.S081 2020 LAB8记录
MIT6.S081学习总结-lab8:Lock