Locking

很多概念在看Java并发的时候都学习过，这些重复的地方就不做赘述了。

Code: spinlock

spinlock 使用介绍

一、spinlock 简介
自旋锁（spinlock）：是指当一个线程在获取锁的时候，如果锁已经被其它线程获取，那么该线程将循环等待，不断尝试获取锁，直到获取到锁才会退出循环

二、自旋锁与互斥锁的区别
自旋锁与互斥锁类似，它们都是为了解决对某项资源的互斥使用，在任何时刻最多只能有一个线程获得锁
对于互斥锁，如果资源已经被占用，调用者将进入睡眠状态
对于自旋锁，如果资源已经被占用，调用者就一直循环在那里，看是否自旋锁的保持者已经释放了锁

三、自旋锁的优缺点
自旋锁不会发生进程切换，不会使进程进入阻塞状态，减少了不必要的上下文切换，执行速度快。非自旋锁在获取不到锁的时候会进入阻塞状态，从而进入内核态，当获取到锁的时候需要从内核态恢复，需要线程上下文切换，影响性能
如果某个线程持有锁的时间过长，就会导致其它等待获取锁的线程长时间循环等待消耗CPU，造成CPU使用率极高

spinlock

// Mutual exclusion lock.
struct spinlock {
  uint locked;       // Is the lock held?

  // For debugging:
  char *name;        // Name of lock.
  struct cpu *cpu;   // The cpu holding the lock.
};

acquire

大概是这么个原理：

当然这有竞态条件。xv6用的是CPU提供的amoswap原子指令来消除竞态条件的。

// in kernel/spinlock.c
// Acquire the lock.
// Loops (spins) until the lock is acquired.
void
acquire(struct spinlock *lk)
{
  // 关中断
  // xv6允许禁止中断。但是由于xv6是一个多核系统,单个core被禁止中断并不会影响其他core。
  push_off(); // disable interrupts to avoid deadlock.
  
  // holding(): Check whether this cpu is holding the lock.
  if(holding(lk))
    panic("acquire");

  // On RISC-V, sync_lock_test_and_set turns into an atomic swap:
  //   a5 = 1
  //   s1 = &lk->locked
  //   amoswap.w.aq a5, a5, (s1)
  // amoswap: 交换a5和(s1)的值，返回(s1)原来的值
  // 也即是如图所示的竞态条件的原子指令
  while(__sync_lock_test_and_set(&lk->locked, 1) != 0)
    ;

  __sync_synchronize();

  // Record info about lock acquisition for holding() and debugging.
  lk->cpu = mycpu();
}

__sync_synchronize();

代码里的官方注释：

// Tell the C compiler and the processor to not move loads or stores
// past this point, to ensure that the critical section's memory
// references happen strictly after the lock is acquired.
// On RISC-V, this emits a fence instruction.

这个注释其实没太看明白。我去翻了一下asm代码，发现这句话正如它最后一句所说的被翻译成fence指令：

处理器中的存储系统（一）：RISC-V的FENCE、FENCE.I指令

顾名思义,FENCE指令犹如一道屏障,把前面的存储操作和后面的存储操作隔离开来,前面的决不能到后面再执行,后面的决不能先于FENCE前的指令执行。

这个就好明白多了。

这样一来，acquire和release的两个fence就形成了两道屏障：

acquire();
l->nexy = list;
list = l;
release();

中间那部分的指令可以重排，但是中间的指令就绝不会跑到临界区外。

push_off和pop_off

当CPU未持有自旋锁时，xv6重新启用中断；它必须做一些记录来处理嵌套的临界区域。acquire调用push_off (*kernel/spinlock.c*:89) 并且release调用pop_off (*kernel/spinlock.c*:100)来跟踪当前CPU上锁的嵌套级别。当计数达到零时，pop_off恢复最外层临界区域开始时存在的中断使能状态。intr_off和intr_on函数执行RISC-V指令分别用来禁用和启用中断。

release

// in kernel/spinlock.c
// Release the lock.
void
release(struct spinlock *lk)
{
  if(!holding(lk))
    panic("release");

  lk->cpu = 0;

  __sync_synchronize();

  // Release the lock, equivalent to lk->locked = 0.
  // This code doesn't use a C assignment, since the C standard
  // implies that an assignment might be implemented with
  // multiple store instructions.
  // On RISC-V, sync_lock_release turns into an atomic swap:
  //   s1 = &lk->locked
  //   amoswap.w zero, zero, (s1)
  __sync_lock_release(&lk->locked);

  // 开中断
  pop_off();
}

Code: Using locks

作为粗粒度锁的一个例子，xv6的kalloc.c有一个由单个锁保护的空闲列表。如果不同CPU上的多个进程试图同时分配页面，每个进程在获得锁之前将必须在acquire中自旋等待。自旋会降低性能，因为它只是无用的等待。如果对锁的争夺浪费了很大一部分CPU时间，也许可以通过改变内存分配的设计来提高性能，使其拥有多个空闲列表，每个列表都有自己的锁，以允许真正的并行分配。【很棒的思路】

作为细粒度锁的一个例子，xv6对每个文件都有一个单独的锁，这样操作不同文件的进程通常可以不需等待彼此的锁而继续进行。文件锁的粒度可以进一步细化，以允许进程同时写入同一个文件的不同区域。最终的锁粒度决策需要由性能测试和复杂性考量来驱动。

锁	描述
`bcache.lock`	保护块缓冲区缓存项（block buffer cache entries）的分配
`cons.lock`	串行化对控制台硬件的访问，避免混合输出
`ftable.lock`	串行化文件表中文件结构体的分配
`icache.lock`	保护索引结点缓存项（inode cache entries）的分配
`vdisk_lock`	串行化对磁盘硬件和DMA描述符队列的访问
`kmem.lock`	串行化内存分配
`log.lock`	串行化事务日志操作
管道的`pi->lock`	串行化每个管道的操作
`pid_lock`	串行化next_pid的增量
进程的`p->lock`	串行化进程状态的改变
`tickslock`	串行化时钟计数操作
索引结点的 `ip->lock`	串行化索引结点及其内容的操作
缓冲区的`b->lock`	串行化每个块缓冲区的操作

Figure 6.3: Locks in xv6

Deadlock and lock ordering

如果在内核中执行的代码路径必须同时持有数个锁，那么所有代码路径以相同的顺序获取这些锁是很重要的。如果它们不这样做，就有死锁的风险。假设xv6中的两个代码路径需要锁A和B，但是代码路径1按照先A后B的顺序获取锁，另一个路径按照先B后A的顺序获取锁。为了避免这种死锁，所有代码路径必须以相同的顺序获取锁。全局锁获取顺序的需求意味着锁实际上是每个函数规范的一部分：调用者必须以一种使锁按照约定顺序被获取的方式调用函数。

由于sleep的工作方式（见第7章），Xv6有许多包含每个进程的锁（每个struct proc中的锁）在内的长度为2的锁顺序链。例如，consoleintr (*kernel/console.c*:138)是处理键入字符的中断例程。当换行符到达时，任何等待控制台输入的进程都应该被唤醒。为此，consoleintr在调用wakeup时持有cons.lock，wakeup获取等待进程的锁以唤醒它。因此，全局避免死锁的锁顺序包括必须在任何进程锁之前获取cons.lock的规则。【这段不怎么能看懂，学完第七章再回来看看】

文件系统代码包含xv6最长的锁链。例如，创建一个文件需要同时持有目录上的锁、新文件inode上的锁、磁盘块缓冲区上的锁、磁盘驱动程序的vdisk_lock和调用进程的p->lock。为了避免死锁，文件系统代码总是按照前一句中提到的顺序获取锁。

Locks and interrupt handlers

Xv6 is more conservative: when a CPU acquires any lock, xv6 always disables interrupts on that CPU. Interrupts may still occur on other CPUs, so an interrupt’s acquire can wait for a thread to release a spinlock; just not on the same CPU.看来是通过开关中断来保护临界区的

acquire调用push_off (*kernel/spinlock.c*:89) 并且release调用pop_off (*kernel/spinlock.c*:100)来跟踪当前CPU上锁的嵌套级别。当计数达到零时，pop_off恢复最外层临界区域开始时存在的中断使能状态。intr_off和intr_on函数执行RISC-V指令分别用来禁用和启用中断。

严格的在设置lk->locked (kernel/spinlock.c:28)之前让acquire调用push_off是很重要的。如果两者颠倒，会存在一个既持有锁又启用了中断的短暂窗口期，不幸的话定时器中断会使系统死锁。同样，只有在释放锁之后，release才调用pop_off也是很重要的(*kernel/spinlock.c*:66)。

一个解决了一半的疑问

问题

Xv6更保守：当CPU获取任何锁时，xv6总是禁用该CPU上的中断。中断仍然可能发生在其他CPU上，此时中断的acquire可以等待线程释放自旋锁；由于不在同一CPU上，不会造成死锁。

进展：似乎书中说到，“sleep atomically yields the CPU and releases the spinlock”。等了解完sleep，也即读完第七章之后再来看看。

在处理时钟中断的trap.c中：

// in kernel/trap.c devintr()
    } else if(scause == 0x8000000000000001L){
        
    // 这里！！
    // in kernel/trap.c devintr()
    if(cpuid() == 0){
      clockintr();
    }
        
    w_sip(r_sip() & ~2);
    return 2;
  } 

void
clockintr()
{
  acquire(&tickslock);
  ticks++;
  wakeup(&ticks);
  release(&tickslock);
}

可见只有CPU0才会进入clockintr【因为要求cpuid==0】，锁住ticks引起ticks递增。

而当sys_sleep获得锁之后，其结束循环的条件是ticks - ticks0 < n:

uint64
sys_sleep(void)
{
  int n;
  uint ticks0;
  if(argint(0, &n) < 0)
    return -1;
  acquire(&tickslock);
  ticks0 = ticks;
  while(ticks - ticks0 < n){
    if(myproc()->killed){
      release(&tickslock);
      return -1;
    }
    sleep(&ticks, &tickslock);
  }
  release(&tickslock);
  return 0;
}

我认为，这会导致死锁情况。假设计算机为多CPU，且从零开始依次递增编号。对该死锁情况的讨论，可以分为以下两类：

sys_sleep在CPU2（或者其他编号非零的CPU）运行，且先获取了tickslock的锁。这时候，ticks将会停止增长，sys_sleep结束循环的条件将无法结束。

理由：对于CPU0，它可以进入clockintr的代码段，但是由于锁已经被获取，所以就只能一直在那边死锁等待；对于其他CPU来说，压根执行不了那段增加ticks的代码段，所以ticks压根不会增加。这样一来，CPU2进程等待ticks增加，从而获取结束循环的条件；CPU0等待CPU2进程结束，从而使得ticks增加，就造成了死锁。
sys_sleep在CPU0运行，且先获取了tickslock的锁。这时候，ticks将会停止增长，sys_sleep结束循环的条件将无法结束。

理由：由于xv6会在获取锁和释放锁期间关闭中断，因而CPU0无法进行时钟中断而发生进程的切换，只能一直在sys_sleep中等待，所以ticks更不可能增加，造成了死锁。

暂时没有很充分的理由反驳这两点。。。

解答

学习完下一章的内容后可知，sleep(&ticks, &tickslock);会释放掉tickslock的锁，这样CPU0就可以进入clockintr增加ticks了。

再详细梳理一次，这里的具体机制是这样的：

可以把ticks看做信号量，sys_sleep为消费者，clockintr为生产者。

// in sys_sleep()
  acquire(&tickslock);
  while(ticks < 某个数字){
    sleep(&ticks, &tickslock);
  }
  release(&tickslock);

void
clockintr()
{
  acquire(&tickslock);
  ticks++;
  wakeup(&ticks);
  release(&tickslock);
}

可以看到，这是非常典型的生产者消费者模型。生产者每生产一次ticks，就会唤醒消费者，让消费者检查条件。如果条件错误，则继续sleep等待消费者下一次唤醒，如此循环往复。

只不过，还有一个小疑点，就是clockintr这段只有CPU0可以执行这一点是否为真依然存疑。如果确实只有CPU0可以执行的话，假若sys_sleep在CPU0上执行，那么还是依然会造成死锁。所以我猜想是不是CPU0是无法关中断的？也就是说CPU0是一个后盾一般的保护角色？或者是别的CPU也能进入本段代码？如果别的CPU也能进，那是怎么实现的？因为很明显这段代码确实只有CPU0可以进入。

Sleep locks

关于sleep lock的由来和优点，书里描述得很详细，简单来说就是：

Thus we’d like a type of lock that yields the CPU while waiting to acquire, and allows yields (and interrupts) while the lock is held.

因为等待会浪费CPU时间，所以自旋锁最适合短的临界区域；睡眠锁对于冗长的操作效果很好。

void
acquiresleep(struct sleeplock *lk)
{
  acquire(&lk->lk);
  // 等待
  while (lk->locked) {
    // sleep atomically yields the CPU and releases the spinlock
    sleep(lk, &lk->lk);
  }
  // 占用
  lk->locked = 1;
  lk->pid = myproc()->pid;
  release(&lk->lk);
}

void
releasesleep(struct sleeplock *lk)
{
  acquire(&lk->lk);
  lk->locked = 0;
  lk->pid = 0;
  // 到时候可以留意一下wakeup是会唤醒一个还是多个
  wakeup(lk);
  release(&lk->lk);
}

有一点值得注意：

Because sleep-locks leave interrupts enabled, they cannot be used in interrupt handlers. Because acquiresleep may yield the CPU, sleep-locks cannot be used inside spinlock critical sections (though spinlocks can be used inside sleep-lock critical sections).

这实际上是因为自旋锁内不能sleep，因而也就不能使用sleep lock。

为什么不能sleep？我猜测应该是因为sleep中会释放自旋锁然后再调度别的进程。此时，临界区就不受保护了很危险，不符合spinlock在临界区结束才能释放的规范。

在查阅别人的说法的时候，我还看到了这个讨论：

中断中为什么不能sleep | Linux内核的评论区

在中断服务程序中，无法sleep的原因应该是sleep后，调度程序将CPU窃走，由于调度的基本单位是线程（中断服务程序不是线程），因此中断服务程序无法再被调度回来，即中断程序中sleep后的部分永远无法得到执行。

Real world

大多数操作系统都支持POSIX线程（Pthread），它允许一个用户进程在不同的CPU上同时运行几个线程。Pthread支持用户级锁（user-level locks）、障碍（barriers）等。支持Pthread需要操作系统的支持。例如，如果一个Pthread在系统调用中阻塞，同一进程的另一个Pthread应当能够在该CPU上运行。另一个例子是，如果一个线程改变了其进程的地址空间（例如，映射或取消映射内存），内核必须安排运行同一进程下的线程的其他CPU更新其硬件页表，以反映地址空间的变化。

Lab: locks

In this lab you’ll gain experience in re-designing code to increase parallelism. You’ll do this for the xv6 memory allocator and block cache.

Memory allocator

The program user/kalloctest stresses xv6’s memory allocator: three processes grow and shrink their address spaces, resulting in many calls to kalloc and kfree. kalloc and kfree obtain kmem.lock. kalloctest prints (as “#fetch-and-add”) the number of loop iterations in acquire due to attempts to acquire a lock that another core already holds, for the kmem lock and a few other locks. The number of loop iterations in acquire is a rough measure of lock contention.

To remove lock contention, you will have to redesign the memory allocator to avoid a single lock and list. 也就是说要把kalloc中的整个列表上锁，修改为每个CPU有自己的列表The basic idea is to maintain a free list per CPU, each list with its own lock. Allocations and frees on different CPUs can run in parallel, because each CPU will operate on a different list. The main challenge will be to deal with the case in which one CPU’s free list is empty, but another CPU’s list has free memory; in that case, the one CPU must “steal” part of the other CPU’s free list. Stealing may introduce lock contention, but that will hopefully be infrequent.主要挑战将是处理一个 CPU 的空闲列表为空，但另一个 CPU 的列表有空闲内存的情况；在这种情况下，一个 CPU 必须“窃取”另一个 CPU 的空闲列表的一部分。

Your job is to implement per-CPU freelists, and stealing when a CPU’s free list is empty.

You must give all of your locks names that start with “kmem”. That is, you should call initlock for each of your locks, and pass a name that starts with “kmem”.

Run kalloctest to see if your implementation has reduced lock contention. To check that it can still allocate all of memory, run usertests sbrkmuch. Your output will look similar to that shown below, with much-reduced contention in total on kmem locks, although the specific numbers will differ.

感想

总之，意思就是kalloc里面本来是多核CPU共用一个空闲页list，现在要做的就是给每一核的CPU独立分配一个空闲页list。我觉得可以分为如下几步来做：

定义list数组以及对应的锁

cpu的数量是一定的；cpuid可以用来作为数组下标索引
在init时初始化锁，在freelist的时候把空闲页均分给CPU
当kalloc和kfree的时候，获取当前cpuid上锁
当一个CPU的内存不够的时候，去向另一个CPU窃取。窃取之前，首先应该获取另一个CPU的锁。

以上是初见思路。正确思路确实跟上面的一样，编码过程也比较简单，没有很恶心的细节和奇奇怪怪的bug，没什么好说的。

第二步中，hints是推荐把所有空闲页都分给CPU0。

第四步的时候我是一次窃取一页。我看到一个一次窃取多页的做法，我觉得很有想法，在这里附上链接：

MIT6.S081 lab8 locks

代码

定义

struct {
  struct spinlock kmem_locks[NCPU];
  struct run *freelists[NCPU];
} kmem;

初始化

由于kinit仅会由一个cpu执行一次【详情见main.c】，故而我这里在kinit的做法是由一个CPU初始化所有CPU，而没有选择去修改main.c从而使每个CPU都执行一次kinit。

void
kinit()
{
  for(int i=0;i<NCPU;i++){
    char buf[8];
    snprintf(buf,6,"kmem%d",i);
    initlock(&kmem.kmem_locks[i], buf);
  }
  freerange(end, (void*)PHYSTOP);
}

// 多带一个参数表示cpuid，仅在kinit的freerange中使用
void
kfree_init(void *pa,int i)
{
  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;

  r->next = kmem.freelists[i];
  kmem.freelists[i] = r;
}
void
freerange(void *pa_start, void *pa_end)
{
  char *p;
  p = (char*)PGROUNDUP((uint64)pa_start);
   
  // 把空闲内存页均分给每个CPU
  uint64 sz = ((uint64)pa_end - (uint64)pa_start)/NCPU;
  uint64 tmp = PGROUNDDOWN(sz) + (uint64)p;
  for(int i=0;i<NCPU;i++){
    for(; p + PGSIZE <= (char*)tmp; p += PGSIZE)
      kfree_init(p,i);
    tmp += PGROUNDDOWN(sz);
    if(i == NCPU-2){
      tmp = (uint64)pa_end;
    }
  }
}

kfree

void
kfree(void *pa)
{
  // ...

  r = (struct run*)pa;
  // 在这
  push_off();
  int id = cpuid();

  acquire(&kmem.kmem_locks[id]);
  r->next = kmem.freelists[id];
  kmem.freelists[id] = r;
  release(&kmem.kmem_locks[id]);
  pop_off();
}

kalloc

void *
kalloc(void)
{
  struct run *r;

  push_off();
  int id = cpuid();

  acquire(&kmem.kmem_locks[id]);
  r = kmem.freelists[id];
  if(r){
    kmem.freelists[id] = r->next;
  }
  release(&kmem.kmem_locks[id]);
  pop_off();

  // 如果无空闲页，则窃取
  if(!r){
    for(int i=NCPU-1;i>=0;i--){
      acquire(&kmem.kmem_locks[i]);
      r = kmem.freelists[i];
      if(r){
        kmem.freelists[i] = r->next;
        release(&kmem.kmem_locks[i]);
        break;
      }
      release(&kmem.kmem_locks[i]);
    }
  }
  
  if(r)
    memset((char*)r, 5, PGSIZE); // fill with junk
  return (void*)r;
}

Buffer cache

buffer cache的结构其实跟kalloc的内存分配结构有一定的类似之处，都是采用链表管理，但是buffer cache的实现相较于kalloc更为复杂。

Reducing contention in the block cache is more tricky than for kalloc, because bcache buffers are truly shared among processes (and thus CPUs).

For kalloc, one could eliminate most contention by giving each CPU its own allocator; that won’t work for the block cache.

We suggest you look up block numbers in the cache with a hash table that has a lock per hash bucket.

感想

初见思路

我想我们可以这么实现：

首先有一个双向链表，接收着所有空闲无设备分配的buf。然后再有多个双向链表桶，以设备号为索引值。

设备号数量，也即hash table的大小定义在kernel/param.h中：

1	#define NDEV 10

在bget中，第一个循环仅需在设备链表中查找即可，第二个循环需要先看设备链表是否有空闲的对象，如果没有，则去接收所有空闲无设备分配的那个双向链表中窃取一个对象。

在brelse中，则把要释放的buf对象添加在head中即可。

因而，我们要做以下几件事：

修改bcache的定义

添加数量为设备号的head数组，以及对应的锁
初始化bcache
添加工具函数：将一个buf加入一个双向链表；从一个双向链表中得到一个buf
写bget和brelse

看起来确实好像可以实现的样子，但是这个问题在于，这么做就直接破坏了LRU的这个规则。所以还是不能这么写的。但总之先把我的代码放上来。

以下代码是不能正常运行的。比如说在执行ls命令时，会发生如下错误：

会打印出一些乱七八糟的东西，并且这些东西似乎是固定的，每次都会发生，看来应该不是多进程的问题，而是代码有哪里出现逻辑错误了。不过注意到会产生“stopforking”、“bigarg-ok”，这两个似乎是在usertest中的两个文件名，很奇怪。

很遗憾我暂时没有精力debug了。姑且先把错误代码放在这里吧。

struct {
  struct spinlock lock;
  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;
  struct buf dev_heads[NDEV];
  struct spinlock dev_locks[NDEV];
} bcache;

void
binit(void)
{
  struct buf *b;

  initlock(&bcache.lock, "bcache");
  for(int i=0;i<NDEV;i++){
    char buf[10];
    snprintf(buf,9,"bcache%02d",i);
    initlock(&(bcache.dev_locks[i]), buf);
    bcache.dev_heads[i].prev = &(bcache.dev_heads[i]);
    bcache.dev_heads[i].next = &(bcache.dev_heads[i]);
  }

  // 初始时，每一个桶内都有一个buf结点
  b = bcache.buf;
  for(int i=0;i<NDEV;i++){
    b->next = bcache.dev_heads[i].next;
    b->prev = &bcache.dev_heads[i];
    initsleeplock(&b->lock, "buffer");
    bcache.dev_heads[i].next->prev = b;
    bcache.dev_heads[i].next = b;
    b++;
  }

  // Create linked list of buffers
  bcache.head.prev = &bcache.head;
  bcache.head.next = &bcache.head;
  for(; b < bcache.buf+NBUF; b++){
    b->next = bcache.head.next;
    b->prev = &bcache.head;
    initsleeplock(&b->lock, "buffer");
    bcache.head.next->prev = b;
    bcache.head.next = b;
  }
}

void
brelse(struct buf *b)
{
  if(!holdingsleep(&b->lock))
    panic("brelse");

  uint dev = b->dev;
  releasesleep(&b->lock);

  acquire(&(bcache.dev_locks[dev]));
  b->refcnt--;
  if (b->refcnt == 0) {
    b->next->prev = b->prev;
    b->prev->next = b->next;
    release(&(bcache.dev_locks[dev]));

    acquire(&bcache.lock);
    b->next = bcache.head.next;
    b->prev = &bcache.head;
    bcache.head.next->prev = b;
    bcache.head.next = b;
    release(&bcache.lock);
  }else
    release(&(bcache.dev_locks[dev]));
}

static struct buf*
bget(uint dev, uint blockno)
{
  acquire(&(bcache.dev_locks[dev]));

  // Is the block already cached?
  for(struct buf* b = bcache.dev_heads[dev].next; b != &(bcache.dev_heads[dev]); b = b->next){
    if(b->blockno == blockno){
      b->refcnt++;
      release(&(bcache.dev_locks[dev]));
      acquiresleep(&b->lock);
      return b;
    }
  }
  release(&(bcache.dev_locks[dev]));

  // 在head中找
  acquire(&bcache.lock);
  // Recycle the least recently used (LRU) unused buffer.
  for(struct buf* b = bcache.head.prev; b != &(bcache.head); b = b->prev){
    if(b->refcnt == 0) {
      b->dev = dev;
      b->blockno = blockno;
      b->valid = 0;
      b->refcnt = 1;
      release(&bcache.lock);
      acquiresleep(&b->lock);
      return b;
    }
  }
  panic("bget: no buffers");
}

正确思路

首先，大家似乎都是用blockno来hash的，这点就跟我的原始思路不一样了（。其实也很对，因为每个设备的使用频率是不平均的，用blockno来hash比用dev来hash其实会让访问次数更加平均。

然后就是怎么保证LRU依然OK。hints的做法是使用时间戳。我们可以在brelse的时候记录时间戳字段，在bget缺块的时候遍历hash table，找出对应timestamp最小的block即可。

历经了几小时的debug，代码最终正确。正确版本在下面的代码模块处。

debug过程

coding过程其实很短暂，毕竟思路很直观。我一开始是按初见思路写的代码，然后再从初见思路改到正确思路，这个过程，给我埋下了极大的安全隐患【悲】

其实几个小时下来，很多细节都已经忘记了，接下来就说点印象比较深的吧。

首先，我使用了正确思路以来，依然出现了跟初见思路一样的错误，也即xv6正常boot，但是执行ls命令会有错误。但是，当我make clean之后再次make qemu，错误改变了，变成了xv6 boot失败，并且爆出错误panic:ilock:no type。

注：关于此处的make clean，有两点需要解释。一是为什么会做出make clean的行为，二是这个变化的原理是什么。

此处突然做出make clean的行为，是因为参照了该文章：

MIT6.S081 lab8 locks

没想到我make clean之后反而就变成了他这样的问题23333也是感觉蛮惊讶的

这个的原理说实话我不大清楚。猜想可能是make qemu的某段访问磁盘初始化之类的代码只会执行一次，只有make clean之后才会让其执行第二次。所以我们手动完全boot了一遍操作系统，才会导致这个错误爆出来，否则，操作系统就会使用原本的正确boot版本启动，之后再执行命令就当然是错误的了。

我想知乎文章里也应该是这个原因。操作系统本来使用的是错误版本，make clean后才会重新使用正确版本。

我之后写对了又尝试了一下，觉得我的猜想应该是对的。我的执行路线：

make qemu，得到正确结果

将bio.c改回错误版本

再次make qemu，发现xv6正常boot，但是执行ls命令会出以上同样的错误

make clean，然后make qemu，爆出panic:ilock: no type

挺完美地符合了我的猜想。

【来自之后的学习：

in lab file system：

mkfs 程序创建 xv6 文件系统磁盘映像并确定文件系统总共有多少个块；这个大小由 kernel/param.h 中的 FSSIZE 控制。您会看到本实验存储库中的 FSSIZE 设置为 200,000 个块。您应该在 make 输出中看到 mkfs/mkfs 的以下输出：
nmeta 70 (boot, super, log blocks 30 inode blocks 13, bitmap blocks 25) blocks 199930 total 200000
这一行描述了 mkfs/mkfs 构建的文件系统：它有 70 个元数据块（用于描述文件系统的块）和 199,930 个数据块，共计 200,000 个块。
如果在实验期间的任何时候您发现自己必须从头开始重建文件系统，您可以运行 make clean 来强制 make 重建 fs.img。

可以看到，我们上面就是做了强制重构fs.img。】

我想来想去不知道这个错到底怎么爆的，看了下ilock()对应报错点：

// in fs.c ilock()
  if(ip->valid == 0){
    //printf("ilock begin.\n");
    bp = bread(ip->dev, IBLOCK(ip->inum, sb));
    dip = (struct dinode*)bp->data + ip->inum%IPB;
    ip->type = dip->type;
	// ...
    ip->size = dip->size;
    memmove(ip->addrs, dip->addrs, sizeof(ip->addrs));
    brelse(bp);
    ip->valid = 1;
    if(ip->type == 0){
      //print_buf();
      printf("bp->blockno = %d, bp->refcnt = %d\n",bp->blockno,bp->refcnt);
      panic("ilock: no type");
    }
  }

可知大概就是，ip的type为0这个非法数值就报错了，而ip的type来源于dip，dip又指向了bp的data，bp也就是我们在bio.c一直在打交道的buf结构体。所以说，其实问题是出在了buf上，我们的bread返回的是一个错误的buf。

那么，究竟是buf的哪里出错了呢？这个问题想了我很久很久很久，依然没想出来。我一直认为是我的hashtable+双向链表这个数据结构哪里写错了，反反复复看了三四遍，其他地方的逻辑也反反复复研究了好几遍，依然没有结论。当然此过程也抓出了很多bug，但抓完bug后报错仍在，非常坚挺。

快要放弃的时候，我发现了错误。这很大一部分归功于我用于调试的这个函数：

// 打印出hashtable的所有结点
void
print_buf(){
  printf("**********************\n");
  printf("cnt = %d,dec = %d\n",cnt,dec);
  for(int i=0;i<NBUCKET;i++){
  int should = !holding(&(bcache.dev_locks[i]));
   if(should)
     acquire(&(bcache.dev_locks[i]));
   printf("--------------\n");
   int tmp_cnt = 0;
   struct buf* b;
   for(b = bcache.dev_heads[i].next; b != &(bcache.dev_heads[i]); b = b->next){
   //for(b = bcache.dev_heads[i].prev; b != &(bcache.dev_heads[i]); b = b->prev){
     tmp_cnt++;
     printf("b.refcnt = %d,b.dev = %d,b.blockno = %d\n",b->refcnt,b->dev,b->blockno);
   }
   printf("%d:total:%d\n",i,tmp_cnt);
   printf("--------------\n");
   if(should)
     release(&(bcache.dev_locks[i]));
  }
}

我在ilock()的panic前面调用了这个函数，并且打印了出问题的buf的blockno：

if(ip->type == 0){
  print_buf();
  printf("bp->blockno = %d, bp->refcnt = %d\n",bp->blockno,bp->refcnt);
  panic("ilock: no type");
}

可以看到，出问题的这里blockno=33，而在桶7中，首先有两个blockno==33的结点，这已经违反了不变性条件；其次有一个refcnt==1的结点，那个是所需结点，但我们却没有找到那个结点，反而去新申请了一个结点。这显然非常地古怪。

于是随后，我就在bio.c的bget()中添加了这么几句话：

最终结果是会打印出两个blockno==0的结点，但是blockno==33的结点没有访问到。

这就很奇怪了。print_buf中以及bget的这个地方，都是遍历hashtable的某个双向链表，但是，为什么print_buf可以访问到，但是bget不行呢？

我首先对比出来的，是print_buf是逆序遍历，而bget是顺序遍历，所以我就又猜想是因为我的数据结构写错了，然而又看了一遍发现并不是。

这时候，可能我的视力恢复了吧，我猛然发现：：

我超，这里是不是应该用hash。。。。。改完这处之后，果然就非常顺利地pass了所有测试【悲】

可以看到伏笔回收了。我是在旧思路代码基础上改过来的。旧思路代码是用dev作为index的，这个for循环忘记改了。因而，就这样，就这么寄了，看了我三四个小时【悲】

不过这倒是可以解释得通所有的错误了。之所以ilock中buf出错，没有正确找到已经映射在cache中的buf而是自己新建了一个，是因为，我压根就没有在正确的桶里找，而是在别的桶中找，这样自然就找不到了，就会自己新建一个，然后就寄了。

这个故事告诉我们，还是得谨慎写代码（）以及，我在旧代码基础上改的时候，其实可以用更聪明的替换方法：修改dev的变量名为hash->把参数里的dev变量名改为dev。这样就不会出错了。很遗憾，我并没有想到，只是很急很急地手动一个个改了，之后也没有检查，才发生如此错误。忏悔。

本次bug虽然很sb，但确实让我在debug过程中收获了些许，至少毅力变强了（）途中无数次想要放弃，还好我坚持了下来，才能看到如此感动的OK一片：

代码

之后写学校实验时回过头来看，发现之前的实现是不对的，在同时进入bfree函数时有死锁风险。经过修改后虽然粒度大了但是安全了。对了，额外附上一版不知道为啥错了的细粒度版本……看了真感觉没什么问题，但依然是会在bfree时panic两次free。等以后有精力再继续研究吧（泪目）

错误版本的思路就是，使用每个block块自己的锁(b->lock)和每个桶的锁来实现细粒度加锁。我是左看右看感觉每个block从在bget中获取一直到brelse释放的b->lock锁是一直持有的，但确实依然有可能发生两个进程同时获取同一个block的锁的情况。实在不知道怎么办了，想了很久还是没想出细粒度好方法（泪）总之代码先放在这里。

正确版本

请见我的github。

错误版本

static struct buf*
bget(uint dev, uint blockno)
{
  // printf("bget\n");
  uint hash = blockno % 13;

  acquire(&(bcache.dev_locks[hash]));

  // Is the block already cached?
  for(struct buf* b = bcache.dev_heads[hash].next; b != &(bcache.dev_heads[hash]); b = b->next){
    int initial_hold = holdingsleep(&b->lock);
    release(&(bcache.dev_locks[hash]));

    if (!initial_hold)
      acquiresleep(&b->lock);

    if(b->blockno == blockno&&b->dev == dev){ // 找到了
      b->refcnt++;
      b->timestamp = ticks;
      // release(&(bcache.dev_locks[hash]));
      // acquiresleep(&b->lock);
      return b;
    }

    if (!initial_hold)
      releasesleep(&b->lock);
    acquire(&(bcache.dev_locks[hash]));
  }

  release(&(bcache.dev_locks[hash]));

  // 没找到，进行LRU
  // 遍历hash table，找到LRU，也即时间戳最小的且refcnt小于0的那一项

  uint min_time = 4294967295;// uint的最大值。此处不能使用(uint)(-1)
  struct buf* goal = 0;
  for(int i = 0; i < NBUCKET; i++) {
    uint time = 0;
    acquire(&(bcache.dev_locks[i]));
    for(struct buf* b = bcache.dev_heads[i].prev; b != &(bcache.dev_heads[i]); b = b->prev){
      int initial_hold = holdingsleep(&b->lock);
      release(&(bcache.dev_locks[i]));
      if (!initial_hold) 
        acquiresleep(&b->lock);
      
      if(b->refcnt == 0) {
          time = b->timestamp;
          if(time < min_time){
            min_time = time;
            if (goal) releasesleep(&goal->lock);
            goal = b;
          }
      }
      if (!initial_hold && goal != b)  releasesleep(&b->lock);
      acquire(&(bcache.dev_locks[i]));
    }
    release(&(bcache.dev_locks[i]));
  }
  // hashtable中存在着空闲buf
  if(goal != 0){
      // acquiresleep(&goal->lock);
      goal->dev = dev;
      goal->blockno = blockno;
      goal->valid = 0;
      goal->refcnt = 1;

      // 将goal从其所在双向链表中移除
      acquire(&(bcache.dev_locks[hash]));

      goal->prev->next = goal->next;
      goal->next->prev = goal->prev;

      // 在新双向链表中添加goal
      goal->prev = &(bcache.dev_heads[hash]);
      goal->next = bcache.dev_heads[hash].next;

      bcache.dev_heads[hash].next->prev = goal;
      bcache.dev_heads[hash].next = goal;

      release(&(bcache.dev_locks[hash]));

      return goal;
  }
  panic("bget: no buffers");
}

void
brelse(struct buf *b)
{
  if(!holdingsleep(&b->lock))
    panic("brelse");

  uint hash = b->blockno%NBUCKET;

  acquire(&(bcache.dev_locks[hash]));
  b->refcnt--;
  b->timestamp = ticks;
  if (b->refcnt == 0) {
    // no one is waiting for it.
    b->next->prev = b->prev;
    b->prev->next = b->next;
    
    b->next = bcache.dev_heads[hash].next;
    b->prev = &bcache.dev_heads[hash];
    bcache.dev_heads[hash].next->prev = b;
    bcache.dev_heads[hash].next = b;
  }
  release(&(bcache.dev_locks[hash]));
  releasesleep(&b->lock);
}