XV6中的锁:MIT6.s081/6.828 lectrue10:Locking 以及 Lab8 locks Part1 心得
阅读原文时间:2023年08月31日阅读:3

这节课程的内容是锁(本节只讨论最基础的锁)。其实锁本身就是一个很简单的概念,这里的简单包括 3 点:

  1. 概念简单,和实际生活中的锁可以类比,不像学习虚拟内存时,现实世界中几乎没有可以类比的对象,所以即使这节课偏向于理论介绍,也一点不会感觉晦涩。
  2. 使用简单,几乎所有的锁都实现了非常简单的api,acquire 就是获取锁,release 就是释放锁,作为用户,用起来锁来非常简单(甚至比你现实中拿钥匙开一把锁还要简单)
  3. 实现简单,本节展示了如何实现一个最基础的 spin lock,实现起来甚至一句话就可以概括:将软件锁转换为硬件锁,利用CPU的低层指令atomic swap 实现锁。

至于CPU底层如何保证 atomic,这个话题已经超出了本节的讨论范围,甚至我的看法更加激进:如果不是CPU的设计者,压根没必要了解这一点,因为即使详尽如 CPU 的 data sheet,也不会和用户说明 atomic swap 是如何实现的,不过想要了解的童鞋看这里:atomic的底层实现

这里顺便说一下,Lab8本来是后面的lab,但是和这一节相关度较高,所以拿到这里讲解,Lab8有两个part,其中part1要求重新设计内存分配器,而part涉及到文件系统,所以讲到文件系统时再来讲解。

首先考虑一个问题,单线程的性能是由什么决定:是CPU的时钟频率,频率越快,执行一条指令所需的时间越短,从下图可以看出,大概从2000年开始:

  • CPU的时钟频率就没有再增加过了(绿线)。
  • 这样的结果是,CPU的单线程性能达到了一个极限并且也没有再增加过(蓝线)。
  • 但是另一方面,CPU中的晶体管数量在持续的增加 (深红色线)。
  • 所以现在不能通过使用单核来让代码运行的更快,要想运行的更快,唯一的选择就是使用多个CPU核。所以从2000年开始,处理器上核的数量开始在增加(黑线)。

但是多核带来的问题就是会有多个进程访问共享的数据结构,所以需要锁,锁可以保证共享数据的正确性

锁就是一个对象,有一个结构体叫做 lock,它包含了一些字段,这些字段中维护了锁的状态,最典型也是最基本的一个锁应该有以下三个字段:是否加锁?锁的名字?哪个cpu核持有锁?

// 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.
};

锁应该有非常直观的API

  • acquire,接收指向lock的指针作为参数,表示要获取这把锁。acquire确保了在任何时间,只会有一个进程能够成功的获取锁。
  • release,也接收指向lock的指针作为参数,表示要释放这把锁。在同一时间尝试获取锁的其他进程需要等待,直到持有锁的进程对锁调用release。

锁的acquire和release之间的代码,通常被称为critical section,称为临界区,而锁就是保护这部分代码的原子性的

这里的关键问题是锁到底是怎么做到原子性的?即锁为什么不能被两个进程同时 acquire?要解答这个问题。就需要深入源码,看一看 acquire是如何实现的

// Acquire the lock.
// Loops (spins) until the lock is acquired.
void
acquire(struct spinlock *lk)
{
  push_off(); // disable interrupts to avoid deadlock.
  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)
  while(__sync_lock_test_and_set(&lk->locked, 1) != 0)
    ;

  // 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.
  __sync_synchronize();

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

这里的关键是:while(__sync_lock_test_and_set(&lk->locked, 1) != 0),注释中已经给出了提示,__sync_lock_test_and_set 是一个C 标准库中的函数,用来实现 atomic swap,或者说用来实现原子性的 test and set。

这里是需要重点解释的地方:

  • 首先解释什么是 test and set,就和他的名字一样,test locked 是否为1,是 1 则说明该锁已经被获取,是 0 则说明该锁没有被获取,就可以获取到锁,所谓获取锁就是 set locked 的值为 1,这样其他进程就 test 到 locked 值为1,从而无法获取锁。

  • 然后解释 atomic swap 是什么,这是一种底层硬件指令,几乎所有真实的CPU都会支持这个指令,可以在硬件层面保证“原子交换”。在RISC-V中,这个特殊的指令叫 amoswap(atomic memory swap),这个指令接收3个参数,分别是address,寄存器r1,寄存器r2。这条指令可以会先锁定住address,将address中的数据保存在一个临时变量中(tmp),之后将r1中的数据写入到address中,之后再将tmp变量中的数据写入到r2中,最后再对于地址解锁。

    相当于原子性地实现了 address->r2, r1->address

  • test and set 和 atomic swap 什么关系?答案就是 atomic swap 就可以实现 test and set

    看下面的 test and set 函数,做的工作就是 *ptr->old, new->*ptr,然后返回old,这个模式是不是和上面的 address->r2, r1->address 一模一样?test 就是将locked的值取出赋值到old并返回,set就是将new值set到locked中

    //test-and-set的C代码模拟
    int TestAndSet(int *ptr, int new) {
        int old = *ptr; //抓取旧值
        *ptr = new; //设置新值
        return old; //返回旧值
    }
    
    typedef struct __lock_t {
        int locked;
    } lock_t;
    
    void init (lock_t *lock) {
        lock->flag = 0;
    }
    
    void lock(lock_t *lock) {
        // 如果为 1,说明锁被其他进程获取
        // 如果为 0,说明该进程可以获取锁
        while (TestAndSet(&lock->flag, 1) == 1)
            ; //spin-wait (do noting)
    }
    
    void unlock (lock_t *lock) {
        lock->flag = 0;
    }

什么时候才必须要加锁呢?课程给出了一个非常保守同时也是非常简单的规则:如果两个进程访问了一个共享的数据结构,并且其中一个进程会更新共享的数据结构,那么就需要对于这个共享的数据结构加锁。

死锁的两个常见情形:

  1. 重入导致死锁:即多次 acquire 同一个锁,一个死锁的最简单的场景就是:首先acquire一个锁,然后进入到critical section;在critical section中,再acquire同一个锁;第二个acquire必须要等到第一个acquire状态被release了才能继续执行,但是不继续执行的话又走不到第一个release,所以程序就一直卡在这了。这就是一个死锁。(但如果是可重入锁的话,这种情况就不会死锁)
  2. 相互等待导致死锁:两个进程,两个锁,A进程需要获取锁1和2,B进程需要获取锁2和1,A获取了1,B获取了2;A要获取2,B要获取1,相互等待导致死锁

避免死锁的方法:使用锁定策略(locking strategy),对锁进行排序,所有的操作都必须以相同的顺序获取锁。

Lab8 Part1 Memory allocator

这个lab要求重新设计内存分配器,原来的内存分配器实现如下,可以看到使用了一个链表 freelist 来保存所有的空闲物理 page

// kernel/kalloc.c
void *
kalloc(void)
{
  struct run *r;

  acquire(&kmem.lock);
  r = kmem.freelist;
  if(r)
    kmem.freelist = r->next;
  release(&kmem.lock);

  if(r)
    memset((char*)r, 5, PGSIZE); // fill with junk
  return (void*)r;
}

下图解释了 kalloc 的运行过程,如果有进程需要内存空间,就调用kalloc函数,kalloc就会返回一个page的地址

可以看到 kalloc 加了一把大锁(即在kalloc首尾acquire和release锁),来保护所有进程共享的数据结构 freelist,但是这样做会有效率问题,所以lab要求在保证正确性的前提下优化这把锁的性能。优化方式也给出来了,只需要为每个cpu core维护一个freelist就好了,然后每个freelist都有自己的锁,之所以还要设计锁,是因为在一个 CPU core 的 freelist 中空闲页不足的情况下,仍需要从其他 CPU 的 freelist 中“偷”内存页,所以一个 CPU core 的 freelist 还可能在“偷”内存页的时候被其他 CPU core 访问,故仍然需要使用单独的锁来保护每个 CPU core 的 freelist。

至于怎么偷?雨露均沾地偷?还是全部从一个 other cpu core 中偷?lab就没有要求了,自己设计即可。

这里的一个关键、也是guide中没有提到的就是:要清楚首次初始化时所有的page都被分配给了一个 core,所以其他 core 首次调用 kalloc 时一定会执行 steal 动作,这里给出修改后的kalloc代码:

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

  push_off();
  int cpu_id = cpuid();

  acquire(&kmem[cpu_id].lock);

  //steal pages form other cpu's freelist
  if(!kmem[cpu_id].freelist) {
    int steal_page = 32;
    for(int i = 0; i < NCPU; i++) {
      if(i == cpu_id) continue;
      acquire(&kmem[i].lock);
      struct run *rr = kmem[i].freelist;

      while(rr && steal_page) {//该cpu的freelist有page,且steal_page不为0
        kmem[i].freelist = rr->next;
        rr->next = kmem[cpu_id].freelist;
        kmem[cpu_id].freelist = rr;
        rr = kmem[i].freelist;
        steal_page--;
      }

      release(&kmem[i].lock);

      if(steal_page == 0) break;
    }

  }

  r = kmem[cpu_id].freelist;
  if(r)
    kmem[cpu_id].freelist = r->next;
  release(&kmem[cpu_id].lock);

  if(r)
    memset((char*)r, 5, PGSIZE); // fill with junk

  pop_off();
  return (void*)r;
}

获得更好的阅读体验,这里是我的博客,欢迎访问:byFMH - 博客园

所有代码见:我的GitHub实现(记得切换到相应分支)