5.1 前言
截止到上一章,dummyxv6实现的操作系统部分均是在单个核心内跑的逻辑。设想一下,printf函数会按格式要求输出参数到显示设备,这个操作在只有一个核心的CPU上运行是没有问题的,但是如果多个核心并行运行,那么比如hart0正在运行printf相关的指令的时候,hart1也开始执行printf相关的指令,那么hart0涉及的输出内容可能会掺杂hart1正在输出的内容,这显然不是我们想要的结果,为了保证输出操作的原子性,就需要引入锁的概念了。
// 只有一个核心的CPU执行printf("CPU%d:hello world", cpuid())时
CPU0:hello world
// 多个核心的CPU执行未加锁的printf("CPU%d:hello world", cpuid())时可能出现的情况
CCPUP1U2:hhelellloo worlworldd此外,由于内存是多个CPU核心共享的,保障操作内存的原子性也是非常关键的。为此,本章将详细讨论自旋锁的设计与实现。
5.2 锁的概念和作用
前言部分介绍了保持输出字符操作原子性的重要性,本节将从内存操作层面来说明锁的作用。为了彻底理解锁的概念和作用,首先要理解CPU发展的历史。摩尔定律[1]表明,CPU的积体晶管每隔18个月就能翻一番,CPU曾经经历过主频快速提升的时代,但是进入到21世纪之后,单个CPU核心的主频(每秒进行的时钟周期)难以继续快速提升(如图1所示),基本上突破4Ghz后就很难再继续提升了,为了进一步加强CPU的处理能力,各个CPU研发商开始将精力放在拓展CPU核心上。
图1
多个核心能够在同一时刻进行并行运算,因此,理论CPU拥有的核心越多,性能就越好。然而天下没有免费的午餐,多核心架构的CPU虽然提升了程序的性能,但是也引发了新的问题。如果在多个核心运行的程序,不会访问相同地址的内存还好,如果它们同时读写了同一地址的内存,那么可能会出现严重的问题。现在来看一个例子,假设两个CPU的核心hart0和hart1同时执行如下代码:
x = x++;此时会发生什么事情呢?现在通过一段伪代码来展示这个流程:
r0 = memory[address_of_x];
r0 = r0 + 1;
memory[address_of_x] = r0 伪代码中,r0表示CPU核心内的寄存器,每个核心都有自己专属的寄存器。memory表示内存,memory[address_of_x]表示从内存的address_of_x地址读取数据,并存到寄存器r0。接着就是对寄存器r0执行加一操作,最后将r0的值存回内存地址address_of_x中。现在假设x的初始值为0,此时CPU内两个核心hart0和hart1同时运行这段代码,此时会发生的情况如图2所示。
图2
最终的结果是x=1,为什么会这样呢?前面也提到过,因为每个核心有独立的寄存器,在执行运算之前,首先都要将内存中的数据读到各自的寄存器中,因此每个处理器内部都有变量x的副本,修改副本并不会直接影响到内存内的数据,需要回写才行,而hart0和hart1又是同时运行的,hart0进行加法运算的时候,hart1读取x到自己的寄存器r0中,此时由于hart0尚未回写,hart1读取的变量x的值还是0。当hart0回写数据的时候,hart1还在运行,由于操作的是各自的副本,所以最后x的值仍然是1,而非我们期待的2。
这就是并行代码引发的问题,那么要解决这个问题,最简单的方式就是保证以上操作的原子性,什么是原子性呢?即是以上的操作,由一个核心执行完,之后才能被其他核心执行,因此需要对这段逻辑上锁,经过修改后的代码如下所示:
lock;
x = x++;
unlock;此时执行的逻辑如图3所示。读者们应该可以观察到,此时hart0完整执行了读取,操作和回写的操作,在解锁之后,hart1才开始操作。这里展示的就是上锁的概念和逻辑,其中介于lock和unlock之间的区域被称为临界区,锁的作用就是保证临界区的原子性。
图3
5.3 实现自旋锁的基础
上一节展示了锁是怎样保护临界区的,现在引入了新的问题,如何实现这个锁?毕竟上一节只是给了一个比较抽象的概念,具体实现还是要落到代码层面。其实锁的逻辑很简单,就是第一个进入临界区的核心,给锁变量标记,其他核心尝试进入该临界区的时候,如果发现锁变量被标记了,就陷入无限循环等待。正因为等待时是陷入无限循环,所以这种锁也被称之为自旋锁。听起来很简单,不就是根据一个变量判断是否无限循环么,然而真的如此么?先来看一个错误示例(版本v1):
// 版本v1
1 func lockinit(spinlock* lk) {
2 lk->locked = 0;
3 }
4
5 func lock(spinlock* lk) {
6 for(;;) {
7 if (lk->locked) {
8 continue;
9 }
10
11 lk->locked = 1;
12 break;
13 }
14 }
15
16 func unlock(spinlock* lk) {
17 lk->locked = 0;
18 } 现在将上述锁逻辑,带回到前面图3关联的例子中(假设自旋锁lk已经初始化过,此时lk->locked为0)。咋一看似乎没问题,实际上这里存在巨大的隐患,为什么呢?lock函数不是根据lk->locked变量来判断是否陷入死循环么?不正是和图3的情况契合吗?这里仍然存在一个问题,因为hart0和hart1可能同时进入lock逻辑,同时读取lk->locked变量的值,同时将lk->locked设置为1,并且同时调用break语句跳出循环,最后同时进入临界区。这里最大的bug就在于,hart0和hart1可以同时进入lock函数的for语句。
回过头来看,为什么会出现锁失败的情况呢?因为上述实现,从判断锁到给锁变量标记为1,这一段逻辑不是原子执行的,也就是说hart0在执行这段逻辑的过程中,hart1也可以同时进入并执行,这显然是软件层面无法实现的,需要CPU的支持。好在大多数现代CPU,均有原子执行获取一个变量的同时,并写入新值的指令,在x86架构中他是xchg指令,在RISC-V架构中它是amoswap指令。下面是amoswap指令的格式:
amoswap.{w/d}.{aqrl} rd, rs2, (rs1)amo是atomic memory operation的缩写,swap是交换的意思,翻译过来就是原子交换。花括号内的字段,表示可以出现0次或1次,‘/’表示左右的字段二选一。下面来看一下它们的具体表示:
- {w/d}:第一个花括号里的w或者d,表示后面寄存器的长度是4个字节还是8个字节的。
- {aqrl}:aq和rl可以分别出现,也可以一起出现,它表示内存序约束的方式,这里先不管它们。
- rs1:被原子操作的变量的内存地址,存放在这个寄存器中,(rs1)表示访问rs1存放地址的值。
- rd:rs1寄存器指向的内存地址的值会存到rd寄存器中。
- rs2:rs2寄存器的值将会存到rs1指向的内存地址。
在C层面,如何翻译成这个指令呢?在GCC中,有个内置函数叫__sync_lock_test_and_set,这个函数的声明如下所示:
unsigned int __sync_lock_test_and_set(volatile void* p, unsigned int v);第一个参数,是要被原子操作的变量的内存地址,第二个参数就是要换进去的值。指针p所指向的值会作为返回值返回,而后指针p所指向的值会被替换成v,这两个操作是原子执行的。此外还有一个内置函数叫做__sync_lock_release,它的声明如下所示:
void __sync_lock_release(volatile void* p);它负责将p指向的值置为0,这个过程也是原子执行的。
接下来将之前v1版本的lock和unlock函数改成新的版本v2:
// 版本v2
1 func lockinit(spinlock* lk) {
2 lk->locked = 0;
3 }
4
5 func lock(spinlock* lk) {
6 while(__sync_lock_test_and_set(&lk->locked, 1) != 0);
7 }
8
9 func unlock(spinlock* lk) {
10 __sync_lock_release(&lk->locked);
11 }第6行代码中的__sync_lock_test_and_set(&lk->locked, 1)会被翻译成如下伪代码的形式:
a5 = 1
s1 = &lk->locked
amoswap.w a5, a5, (s1)a5寄存器的值将作为返回值返回。现在来看一下图4的情况,hart0和hart1同时执行版本v2中的lock函数,图4展示了操作时间序。虽然hart0和hart1是并行执行的,但是执行到amoswap指令的时候,其中一个hart会原子执行从(s1)中取值,并将a5的值存入(s1)的操作。在那之后,另一个hart才能从(s1)中读取数据。图4中,hart0是首个进入amoswap指令的核心,所以在版本v2的lock函数中,它先获得了锁,可以进入图3所示的临界区。而hart1因为是后来才执行amoswap指令,所以得到的值为1,不满足跳出循环的条件,因此hart1会一直执行循环逻辑。
图4
版本v2的第10行代码sync_lock_release编译后得到的指令和sync_lock_test_and_set差不多,只是a5的赋值不是1,而是0。回顾一下图3的情况,只有hart0执行了unlock逻辑之后,lk->locked才是0,hart1此时才能获取锁,并且进入临界区。
本节介绍了实现自旋锁最基础的操作,其实就是依赖CPU的原子交换指令amoswap来替换锁变量的值,最后实现只有一个hart能进入临界区的逻辑。不过实现自旋锁并没有这么简单,还需要考虑很多其他因素,这些因素将会在本章剩余部分里介绍。
5.4 spinlock的数据结构与代码实现
在完成实现spinlock的基础内容的讨论之后,现在来看一下dummyxv6的spinlock实现,在本章配套的代码中,新增了kernel/spinlock.h|c以及kernel/proc.h|c文件。先来看一下spinlock的数据结构:
// kernel/spinlock.h
1 struct spinlock {
2 unsigned int locked;
3 char* name;
4 struct cpu* cpu;
5 }; spinlock结构很简单,只有一个表示锁状态的locked变量,标记锁名称的name字段,还有一个名为cpu的结构体。前面两个比较容易理解,比较费解的是这个cpu类型,它是做什么用的呢?从硬件层面看,在运行内核时,每个核都会有自己的状态数据(比如发生上下文切换时),这些逻辑如果放在硬件层面来执行,那么就过于复杂了。因此这些工作需要软件层面来处理,换句话说,需要内核来进行处理。比方说hart0在运行一个用户进程时,用户按下了键盘按键,此时hart0来响应这个事件,那么此时hart0寄存器内保存的都是该用户进程的数据,此时要将这些寄存器数据保存起来,等hart0处理键盘中断事件后,再恢复这些数据到hart0的寄存器中,接着之前运行的地方执行。那么存取hart0的寄存器数据的内存,就是这个cpu结构体所占据的内存。
每个核心都有一个专属的cpu类型数据,它们被放在一个数组中,这个数组被定义在kernel/proc.c文件中,现在来看一下cpu的定义和数组声明的代码逻辑。
// kernel/proc.h
1 struct cpu {
2 int intena; // Saved interrupt enable bit
3 int noff; // Depth of push_off() nesting
4 };结构体定义很简单,只有一个表示禁止中断前,CPU核心是否允许中断的标记变量–intena,还有一个是记录进行了多少次中断的变量–noff,本章暂时不引入上下文切换相关的结构体,只关心和自旋锁有关系的变量。暂时先把这两个变量是干啥用的放一边,读者们马上就能看到,先来看一下这个cpu类型数组的定义:
// kernel/proc.c
static struct cpu cpus[NCPU];既然前面提到过每个CPU核心都有一个自己专属的cpu结构体变量,并且不会被其他核心访问到,那么内核是怎么关联核心和该结构体变量的呢?不知道读者是否还记得,第三章介绍内核初始化流程时,将机器模式才能访问的mhartid寄存器的值取出,并存入到tp寄存器中了?这个tp寄存器意为thread pointer,是用来记录核心id用的,每个核心有自己唯一的id,比如第0个核心tp值就是0,第一个核心其tp值就是1,以此类推。每次获取tp的值,要通过RISC-V汇编程序中获取,然后将获取的tp值作为cpus数组的索引来获取核心专属的cpu结构体。
// kernel/riscv.c
1 uint64_t r_tp() {
2 uint64_t x;
3 asm volatile("mv %0, tp" : "=r" (x));
4 return x;
5 }
6
7 // kernel/proc.c
8 int cpuid() {
9 int x = r_tp();
10 return x;
11 }
12
13 struct cpu* mycpu() {
14 int id = cpuid();
15 struct cpu* c = &cpus[id];
16 return c;
17 }图5展示了各个CPU核心和cpus数组的关系,这里需要注意的是,内核要管理好硬件核心和cpus数组的关系,不能让一个CPU核心访问到另一个核心专属的cpu结构体变量。一个核心获取自己专属的cpu变量,只需要调用mycpu函数即可,这个函数的定义已经在上述代码中展现出来了。到现在为止就完成了spinlock数据结构的介绍工作了。
图5
接下来要看的是dummyxv6的spinlock实现,其逻辑如下所示:
// kernel/spinlock.c
1 void initlock(struct spinlock* lk, char* name) {
2 lk->name = name;
3 lk->locked = 0;
4 lk->cpu = 0;
5 }
6
7 void acquire(struct spinlock* lk) {
8 push_off();
9
10 if (holding(lk)) {
11 panic("acquire");
12 }
13
14 // atomic exchange
15 while (__sync_lock_test_and_set(&lk->locked, 1));
16
17 // Tell the C compiler and the processor to not move loads or stores
18 // past this point, to ensure that the critical section's memory
19 __sync_synchronize();
20
21 lk->cpu = mycpu();
22 }
23
24 void release(struct spinlock* lk) {
25 if (!holding(lk)) {
26 panic("release");
27 }
28
29 lk->cpu = 0;
30
31 // Tell the C compiler and the processor to not move loads or stores
32 // past this point, to ensure that all the stores in the critical
33 __sync_synchronize();
34
35 __sync_lock_release(&lk->locked);
36
37 pop_off();
38 }图3中的函数名lock被换成了acquire,unlock被换成了release,虽然名字换了,但是意思是一样的。观察代码,读者们应该可以发现,除了initlock函数和之前讨论的一样以外,加锁函数acquire和解锁逻辑release内的实现,比上一节讨论,多了很多逻辑。上一节讨论的锁变量原子交换位于代码的第15行和第35行,它们分别对应加锁和解锁的逻辑。除此之外,其他逻辑将在本章后续小节里陆续介绍。
5.5 中断的禁止与恢复
从本节开始,将会陆续介绍spinlock的acquire和release函数中每一行代码做了的事情,以及为什么要这样写,争取让读者做到读懂每一行代码。首先要看的是第8行和第37行代码,分别调用了push_off函数和pop_off函数,这两个函数是做什么用的呢?
对于获取自旋锁的CPU核心来说,在执行临界区的指令时,最好不要被打断。哪些因素会导致这种现象呢?答案是中断事件,如果一个CPU的核心正在执行临界区的指令时,突然发生中断,那么该核心则不得不暂停继续执行临界区指令,在完成当前上下文保存之后,再去执行中断处理逻辑。如果中断处理逻辑无意中也尝试acquire同一个锁,那么此时会陷入死锁状态。为什么会这样?因为lk->locked此时已经为1了,执行while语句会和其他核心一样陷入无限循环,此时由于原本的临界区指令没有执行完,自然不会执行解锁逻辑,而此时这个核心已经没有机会去执行解锁逻辑了,这种情况就是死锁。
如何避免这个问题呢?最简单的办法就是在acquire的最开始,把中断禁止了,在release锁的时候再恢复。push_off函数就是做了中断处理,而pop_off函数则是做了恢复处理。但是push和pop两个命名看起来是入栈出栈操作,那这和禁止和恢复中断又有什么关系呢?先看一下它们的函数实现:
// kernel/spinlock.c
1 void push_off() {
2 int intr_ena = intr_get();
3 if (mycpu()->noff == 0) {
4 mycpu()->intena = intr_ena;
5 }
6
7 // disable interrupts to avoid deadlock.
8 mycpu()->noff += 1;
9 intr_off();
10 }
11
12 void pop_off() {
13 mycpu()->noff -= 1;
14
15 if (mycpu()->noff == 0 && mycpu()->intena) {
16 intr_on();
17 }
18 } 先看push_off函数,第2行代码从运行push_off函数的核心中获取当前的中断设置数据,一般来说,是否允许一个CPU核心接受中断事件,应该在最初的初始化阶段就设置好。接下来执行第3行代码到第18行代码,通过图6来解释就很清晰了,CPU核心调用push_off的次数,只有和pop_off匹配时,中断才能被恢复(如果该核心在初始化阶段被禁止接收中断事件的话,就完全不会有恢复的行为)。intr_off函数具体的逻辑,就是暂停运行该逻辑的核心进行中断响应,而intr_on则是尝试恢复该核心的中断。为什么这里要通过对noff的引用计数来管理中断是否恢复呢?因为除了acquire函数以外,还有其他地方可能调用push_off和pop_off函数来禁止和恢复中断响应,比如uart模块的uartputc_sync函数,就需要调用push_off来禁止运行自己的核心响应外部中断,在完成字符打印之后再恢复,以保障输出逻辑不被打断。而这样的逻辑也有可能在临界区里执行,观察一下图6,某个CPU核心在多次调用push_off之后,如果只调用一次pop_off就恢复中断了,那么势必会影响到其他逻辑。所以noff变量就好像栈一样,调用一次push_off就加1,调用一次pop_off就减去1,当push_off和pop_off的次数配对时,noff的值为0,也就是栈空了。
图6
以上就是禁止本核心中断和恢复逻辑的阐述,下一节介绍本核心的死锁检测。
5.6 死锁检测
前面提到了死锁的问题,有没有办法检测出来呢?在同一个核心内,如果该核心acquire一个锁之后,在没解锁前,再次acquire同一个锁,是可以被检测到了,检测的逻辑就是在acquire函数里的holding函数里进行,以下是其代码实现:
// kernel/spinlock.c
1 int holding(struct spinlock* lk) {
2 if (lk->locked && lk->cpu == mycpu()) {
3 return 1;
4 }
5 else {
6 return 0;
7 }
8 }代码逻辑非常简单,如果当前的锁被锁住了,并且还是自己锁住的,就返回1否则返回0。回顾acquire和release的实现,当holding为真时,就会调用panic函数,这个panic函数会输出什么操作导致的panic,并且将运行的核心陷入死循环中。死锁检测没办法帮助逻辑跳出死循环,但是能够输出一些有价值的信息,比如是什么原因导致的死锁。
5.7 内存屏障的概念和作用
本节将简单介绍内存屏障的概念和作用。首先第一个要厘清的概念就是,什么是内存屏障?内存屏障英文表示为memory barrier或者memory fence,现在来看一下RISC-V官方文档对于内存屏障的定义[2]。
By default, the FENCE instruction ensures that all memory accesses from instructions preceding the fence in program order (the “predecessor set”) appear earlier in the global memory order than memory accesses from instructions appearing after the fence in program order (the “successor set”).
这句话是什么意思呢?它主要说明了,在一段代码中,fence指令之前的内存操作指令,必须全部早于代码中fence指令之后的内存操作指令执行。通过下面一段伪代码来理解这个概念比较好。
x = 1
y = 2
fence
z = 3
print(x)
print(y)这段伪代码中使用了内存屏障指令fence,那么x = 1和y = 2这两个赋值就不会跑到fence之后。到这里,有的读者可能会感到疑惑,x = 1和y = 2怎么会跑到fence之后?这里就涉及到CPU和编译器对指令的重排了,为什么编译器和CPU会对指令进行重排呢?一般来说,只要不改变程序的运行结果,具体指令谁先执行,谁后执行是没有问题的,相关联的指令重排到一起,往往能获得更高的运行效率。比如上面这段代码,如果没有内存屏障指令,重排后的结果可能如图7所示。调整x,y的赋值语句到z之后,并不会改变程序的运行结果,并且x=1与print(x),y=1与print(y)是关联的,因此重排后理论上性能会更好。
图7
正常情况下,这种重排对运行结果没有任何影响,但是在并发编程中,可能会遇到一些问题,读者们看一下如下伪代码:
...
lock
// do something 1
x = x + 1
// do someting 2
unlock
...上面这段代码中,x = x + 1是临界区,当有两条线程在不同的核心里同时运行这段代码时,只有一条线程,一个核心可以进入临界区运行代码,但是如果这里的lock和unlock没有加内存屏障的话,x = x + 1有可能会被重排到unlock之后,得到如下结果:
...
lock
// do something 1
...
// do someting 2
unlock
x = x + 1
... 遇到这种情况,锁对于x变量的原子操作就失去了作用。因此lock和lock中,均需要内存屏障来保护临界区内的内存访问操作,不会被重排到临界区外。而dummyxv6的实现中,acquire函数和release函数内的__sync_synchronize函数就是干的这件事情,这个函数是GCC内置的函数,它会被转换成RISC-V的fence指令。这样临界区内的内存访问指令(读指令、写指令和读写指令)都不会被重排到临界区之外。
为什么重排会提升程序执行的性能呢?具体原因如下:
- 提高指令级并行性:
- 现代 CPU 是超标量架构(superscalar),可以同时执行多条指令。
- 如果指令之间没有依赖关系,CPU 可以并行执行它们。
- 通过重排指令,CPU 可以更好地利用其执行单元。
- 减少流水线停顿:
- CPU 的流水线(pipeline)需要连续的指令流来保持高效运行。
- 如果某条指令需要等待数据(如内存读取),流水线可能会停顿。
- 通过重排指令,CPU 可以将不依赖该数据的指令提前执行,减少停顿。
- 优化缓存利用率:
- CPU 缓存(cache)是分块的,每次从内存中读取的数据会被加载到缓存行(cache line)中。
- 如果数据和操作数据的指令被安排在一起,可以减少缓存未命中(cache miss),从而提高性能。
到现在为止,acquire和release函数实现的主体流程就完成叙述了。
5.8 printf模块引入自旋锁
在完成自旋锁的实现之后,笔者希望读者们看一下目前需要用到自旋锁的地方有哪些。正如本章开头写的那样,printf的内容应当是不可被中断的,而之前dummyxv6实现的printf模块并没有多线程支持,因此现在要加上,其实现如下所示:
// kernel/printf.c
...
1 struct {
2 struct spinlock lock;
3 int locking;
4 } pr;
5
6 void printfinit(void) {
7 initlock(&pr.lock, "pr");
8 pr.locking = 1;
9 }
10
11 void printf(const char* fmt, ...) {
12 int locking = pr.locking;
13 if (locking) {
14 acquire(&pr.lock);
15 }
16
17 ...
18
19 if (locking) {
20 release(&pr.lock);
21 }
22 }
23
24 void panic(const char* s) {
25 pr.locking = 0;
26 printf("panic: ");
27 printf(s);
28 printf("\n");
29 pr.locking = 1;
30 for(;;);
31 }代码中的printfinit函数,会在main函数中调用,在开始就完成printf模块的初始化工作。printf在开头和结尾加了自旋锁,以此保证中间输出逻辑的原子性,锁变量pr有个locking字段,这个字段平时使用时为1,panic使用时为0,主要原因是为了确保panic的输出能够立即输出。除了printf,uartputc_sync函数也添加了push_off和push_on逻辑,在输出字符时禁止中断。
// kernel/uart.c
1 void uartputc_sync(char c) {
2 push_off();
3 while ((ReadReg(LSR) & LSR_TX_IDLE) == 0);
4 WriteReg(THR, c);
5 pop_off();
6 }最后编译执行的结果为:
start to initialize the kernel.
CPU0 initialize success.
CPU1 starts initializing.
CPU2 starts initializing.
Initialization of CPU2 complete.
Initialization of CPU1 complete.一共有3个核心同时运行,输出内容没有被打乱,验证了自旋锁的正确性。
5.9 结束语
本章详细讨论了dummyxv6的自旋锁的设计与实现,文章首先说明了为什么CPU会朝着多核方向发展,并且提到了多核CPU给开发带来的挑战。接着介绍了锁的概念和实现,最后针对实现自旋锁的逻辑进行逐一解释。相信读者阅读完本章,对自旋锁有新的认识。
Reference
[1] Moore’s law
[2] riscv-spec-20191213 A.3.6 Fence (Rule 4)