Lab8: Locks

Memory allocator

本实验完成的任务是为每个CPU都维护一个空闲列表，初始时将所有的空闲内存分配到某个CPU，此后各个CPU需要内存时，如果当前CPU的空闲列表上没有，则窃取其他CPU的。例如，所有的空闲内存初始分配到CPU0，当CPU1需要内存时就会窃取CPU0的，而使用完成后就挂在CPU1的空闲列表，此后CPU1再次需要内存时就可以从自己的空闲列表中取。

(1). 将kmem定义为一个数组，包含NCPU个元素，即每个CPU对应一个

  1. struct {
  2.   struct spinlock lock;
  3.   struct run *freelist;
  4. } kmem[NCPU];

(2). 修改kinit，为所有锁初始化以“kmem”开头的名称，该函数只会被一个CPU调用，freerange调用kfree将所有空闲内存挂在该CPU的空闲列表上

  1. void
  2. kinit()
  3. {
  4.   char lockname[8];
  5.   for(int i = 0;i < NCPU; i++) {
  6.     snprintf(lockname, sizeof(lockname), "kmem_%d", i);
  7.     initlock(&kmem[i].lock, lockname);
  8.   }
  9.   freerange(end, (void*)PHYSTOP);
  10. }

(3). 修改kfree，使用cpuid()和它返回的结果时必须关中断，请参考《XV6使用手册》第7.4节

  1. void
  2. kfree(void *pa)
  3. {
  4.   struct run *r;
  6.   if(((uint64)pa % PGSIZE) != 0 || (char*)pa < end || (uint64)pa >= PHYSTOP)
  7.     panic("kfree");
  9.   // Fill with junk to catch dangling refs.
  10.   memset(pa, 1, PGSIZE);
  12.   r = (struct run*)pa;
  14.   push_off();  // 关中断
  15.   int id = cpuid();
  16.   acquire(&kmem[id].lock);
  17.   r->next = kmem[id].freelist;
  18.   kmem[id].freelist = r;
  19.   release(&kmem[id].lock);
  20.   pop_off();  //开中断
  21. }

(4). 修改kalloc，使得在当前CPU的空闲列表没有可分配内存时窃取其他内存的

  1. void *
  2. kalloc(void)
  3. {
  4.   struct run *r;
  6.   push_off();// 关中断
  7.   int id = cpuid();
  8.   acquire(&kmem[id].lock);
  9.   r = kmem[id].freelist;
  10.   if(r)
  11.     kmem[id].freelist = r->next;
  12.   else {
  13.     int antid;  // another id
  14.     // 遍历所有CPU的空闲列表
  15.     for(antid = 0; antid < NCPU; ++antid) {
  16.       if(antid == id)
  17.         continue;
  18.       acquire(&kmem[antid].lock);
  19.       r = kmem[antid].freelist;
  20.       if(r) {
  21.         kmem[antid].freelist = r->next;
  22.         release(&kmem[antid].lock);
  23.         break;
  24.       }
  25.       release(&kmem[antid].lock);
  26.     }
  27.   }
  28.   release(&kmem[id].lock);
  29.   pop_off();  //开中断
  31.   if(r)
  32.     memset((char*)r, 5, PGSIZE); // fill with junk
  33.   return (void*)r;
  34. }

Buffer cache

这个实验的目的是将缓冲区的分配与回收并行化以提高效率，这个实验折腾了一天，有些内容还是比较绕的，

(1). 定义哈希桶结构，并在bcache中删除全局缓冲区链表，改为使用素数个散列桶

  1. #define NBUCKET 13
  2. #define HASH(id) (id % NBUCKET)
  4. struct hashbuf {
  5.   struct buf head;       // 头节点
  6.   struct spinlock lock;  // 锁
  7. };
  9. struct {
  10.   struct buf buf[NBUF];
  11.   struct hashbuf buckets[NBUCKET];  // 散列桶
  12. } bcache;

(2). 在binit中，（1）初始化散列桶的锁，（2）将所有散列桶的head->prev、head->next都指向自身表示为空，（3）将所有的缓冲区挂载到bucket[0]桶上，代码如下

  1. void
  2. binit(void) {
  3.   struct buf* b;
  4.   char lockname[16];
  6.   for(int i = 0; i < NBUCKET; ++i) {
  7.     // 初始化散列桶的自旋锁
  8.     snprintf(lockname, sizeof(lockname), "bcache_%d", i);
  9.     initlock(&bcache.buckets[i].lock, lockname);
  11.     // 初始化散列桶的头节点
  12.     bcache.buckets[i].head.prev = &bcache.buckets[i].head;
  13.     bcache.buckets[i].head.next = &bcache.buckets[i].head;
  14.   }
  16.   // Create linked list of buffers
  17.   for(b = bcache.buf; b < bcache.buf + NBUF; b++) {
  18.     // 利用头插法初始化缓冲区列表,全部放到散列桶0上
  19.     b->next = bcache.buckets[0].head.next;
  20.     b->prev = &bcache.buckets[0].head;
  21.     initsleeplock(&b->lock, "buffer");
  22.     bcache.buckets[0].head.next->prev = b;
  23.     bcache.buckets[0].head.next = b;
  24.   }
  25. }

(3). 在buf.h中增加新字段timestamp，这里来理解一下这个字段的用途：在原始方案中，每次brelse都将被释放的缓冲区挂载到链表头，禀明这个缓冲区最近刚刚被使用过，在bget中分配时从链表尾向前查找，这样符合条件的第一个就是最久未使用的。而在提示中建议使用时间戳作为LRU判定的法则，这样我们就无需在brelse中进行头插法更改结点位置

  1. struct buf {
  2.   ...
  3.   ...
  4.   uint timestamp;  // 时间戳
  5. };

(4). 更改brelse，不再获取全局锁

  1. void
  2. brelse(struct buf* b) {
  3.   if(!holdingsleep(&b->lock))
  4.     panic("brelse");
  6.   int bid = HASH(b->blockno);
  8.   releasesleep(&b->lock);
  10.   acquire(&bcache.buckets[bid].lock);
  11.   b->refcnt--;
  13.   // 更新时间戳
  14.   // 由于LRU改为使用时间戳判定，不再需要头插法
  15.   acquire(&tickslock);
  16.   b->timestamp = ticks;
  17.   release(&tickslock);
  19.   release(&bcache.buckets[bid].lock);
  20. }

(5). 更改bget，当没有找到指定的缓冲区时进行分配，分配方式是优先从当前列表遍历，找到一个没有引用且timestamp最小的缓冲区，如果没有就申请下一个桶的锁，并遍历该桶，找到后将该缓冲区从原来的桶移动到当前桶中，最多将所有桶都遍历完。在代码中要注意锁的释放

  1. static struct buf*
  2. bget(uint dev, uint blockno) {
  3.   struct buf* b;
  5.   int bid = HASH(blockno);
  6.   acquire(&bcache.buckets[bid].lock);
  8.   // Is the block already cached?
  9.   for(b = bcache.buckets[bid].head.next; b != &bcache.buckets[bid].head; b = b->next) {
  10.     if(b->dev == dev && b->blockno == blockno) {
  11.       b->refcnt++;
  13.       // 记录使用时间戳
  14.       acquire(&tickslock);
  15.       b->timestamp = ticks;
  16.       release(&tickslock);
  18.       release(&bcache.buckets[bid].lock);
  19.       acquiresleep(&b->lock);
  20.       return b;
  21.     }
  22.   }
  24.   // Not cached.
  25.   b = 0;
  26.   struct buf* tmp;
  28.   // Recycle the least recently used (LRU) unused buffer.
  29.   // 从当前散列桶开始查找
  30.   for(int i = bid, cycle = 0; cycle != NBUCKET; i = (i + 1) % NBUCKET) {
  31.     ++cycle;
  32.     // 如果遍历到当前散列桶，则不重新获取锁
  33.     if(i != bid) {
  34.       if(!holding(&bcache.buckets[i].lock))
  35.         acquire(&bcache.buckets[i].lock);
  36.       else
  37.         continue;
  38.     }
  40.     for(tmp = bcache.buckets[i].head.next; tmp != &bcache.buckets[i].head; tmp = tmp->next)
  41.       // 使用时间戳进行LRU算法，而不是根据结点在链表中的位置
  42.       if(tmp->refcnt == 0 && (b == 0 || tmp->timestamp < b->timestamp))
  43.         b = tmp;
  45.     if(b) {
  46.       // 如果是从其他散列桶窃取的，则将其以头插法插入到当前桶
  47.       if(i != bid) {
  48.         b->next->prev = b->prev;
  49.         b->prev->next = b->next;
  50.         release(&bcache.buckets[i].lock);
  52.         b->next = bcache.buckets[bid].head.next;
  53.         b->prev = &bcache.buckets[bid].head;
  54.         bcache.buckets[bid].head.next->prev = b;
  55.         bcache.buckets[bid].head.next = b;
  56.       }
  58.       b->dev = dev;
  59.       b->blockno = blockno;
  60.       b->valid = 0;
  61.       b->refcnt = 1;
  63.       acquire(&tickslock);
  64.       b->timestamp = ticks;
  65.       release(&tickslock);
  67.       release(&bcache.buckets[bid].lock);
  68.       acquiresleep(&b->lock);
  69.       return b;
  70.     } else {
  71.       // 在当前散列桶中未找到，则直接释放锁
  72.       if(i != bid)
  73.         release(&bcache.buckets[i].lock);
  74.     }
  75.   }
  77.   panic("bget: no buffers");
  78. }

(6). 最后将末尾的两个小函数也改一下

  1. void
  2. bpin(struct buf* b) {
  3.   int bid = HASH(b->blockno);
  4.   acquire(&bcache.buckets[bid].lock);
  5.   b->refcnt++;
  6.   release(&bcache.buckets[bid].lock);
  7. }
  9. void
  10. bunpin(struct buf* b) {
  11.   int bid = HASH(b->blockno);
  12.   acquire(&bcache.buckets[bid].lock);
  13.   b->refcnt--;
  14.   release(&bcache.buckets[bid].lock);
  15. }

踩过的坑：

  1. bget中重新分配可能要持有两个锁，如果桶a持有自己的锁，再申请桶b的锁，与此同时如果桶b持有自己的锁，再申请桶a的锁就会造成死锁！因此代码中使用了if(!holding(&bcache.bucket[i].lock))来进行检查。此外，代码优先从自己的桶中获取缓冲区，如果自身没有依次向后查找这样的方式也尽可能地避免了前面的情况。
  2. 在bget中搜索缓冲区并在找不到缓冲区时为该缓冲区分配条目必须是原子的！在提示中说bget如果未找到而进行分配的操作可以是串行化的，也就是说多个CPU中未找到，应当串行的执行分配，同时还应当避免死锁。于是在发现未命中（Not cached）后，我写了如下的代码（此时未删除bcache.lock）
  1. // 前半部分查找缓冲区的代码
  2. // Not cached
  3. release(&bcache.buckets[bid].lock);
  4. acquire(&bcache.lock);
  5. acquire(&bcache.buckets[bid].lock);
  6. // 后半部分分配缓冲区的代码

这段代码中先释放了散列桶的锁之后再重新获取，之所以这样做是为了让所有代码都保证申请锁的顺序：先获取整个缓冲区的大锁再获取散列桶的小锁，这样才能避免死锁。但是这样做却破坏了程序执行的原子性。

在release桶的锁并重新acquire的这段时间，另一个CPU可能也以相同的参数调用了bget，也发现没有该缓冲区并想要执行分配。最终的结果是一个磁盘块对应了两个缓冲区，破坏了最重要的不变量，即每个块最多缓存一个副本。这样会导致usertests中的manywrites测试报错：panic: freeing free block