锁机制 ========================================= 本节导读 ----------------------------------------- 到目前为止,我们已经实现了进程和线程,也能够理解在一个时间段内,会有多个线程在执行,这就是并发。 而且,由于线程的引入,多个线程可以共享进程中的全局数据。如果多个线程都想读和更新全局数据, 那么谁先更新取决于操作系统内核的抢占式调度和分派策略。在一般情况下,每个线程都有可能先执行, 且可能由于中断等因素,随时被操作系统打断其执行,而切换到另外一个线程运行, 形成在一段时间内,多个线程交替执行的现象。如果没有一些保障机制(比如互斥、同步等), 那么这些对共享数据进行读写的交替执行的线程,其期望的共享数据的正确结果可能无法达到。 所以,我们需要研究一种保障机制 --- 锁 ,确保无论操作系统如何抢占线程,调度和切换线程的执行, 都可以保证对拥有锁的线程,可以独占地对共享数据进行读写,从而能够得到正确的共享数据结果。 这种机制的能力来自于处理器的指令、操作系统系统调用的基本支持,从而能够保证线程间互斥地读写共享数据。 下面各个小节将从为什么需要锁、锁的基本思路、锁的不同实现方式等逐步展开讲解。 为什么需要锁 ----------------------------------------- 上一小节已经提到,没有保障机制的多个线程,在对共享数据进行读写的过程中,可能得不到预期的结果。 我们来看看这个简单的例子: .. code-block:: c :linenos: :emphasize-lines: 4 // 线程的入口函数 int a=0; void f() { a = a + 1; } 对于上述函数中的第 4 行代码,一般人理解处理器会一次就执行完这条简单的语句,但实际情况并不是这样。 我们可以用 GCC 编译出上述函数的汇编码: .. code-block:: shell :linenos: $ riscv64-unknown-elf-gcc -o f.s -S f.c 可以看到生成的汇编代码如下: .. code-block:: asm :linenos: :emphasize-lines: 18-23 //f.s .text .globl a .section .sbss,"aw",@nobits .align 2 .type a, @object .size a, 4 a: .zero 4 .text .align 1 .globl f .type f, @function f: addi sp,sp,-16 sd s0,8(sp) addi s0,sp,16 lui a5,%hi(a) lw a5,%lo(a)(a5) addiw a5,a5,1 sext.w a4,a5 lui a5,%hi(a) sw a4,%lo(a)(a5) nop ld s0,8(sp) addi sp,sp,16 jr ra .. chyyuu 可以给上面的汇编码添加注释??? 从中可以看出,对于高级语言的一条简单语句(C 代码的第 4 行,对全局变量进行读写),很可能是由多条汇编代码 (汇编代码的第 18~23 行)组成。如果这个函数是多个线程要执行的函数,那么在上述汇编代码第 18 行到第 23 行中的各行之间,可能会发生中断,从而导致操作系统执行抢占式的线程调度和切换, 就会得到不一样的结果。由于执行这段汇编代码(第 18~23 行))的多个线程在访问全局变量过程中可能导致竞争状态, 因此我们将此段代码称为临界区(critical section)。临界区是访问共享变量(或共享资源)的代码片段, 不能由多个线程同时执行,即需要保证互斥。 下面是有两个线程T0、T1在一个时间段内的一种可能的执行情况: ===== ===== ======= ======= =========== ========= 时间 T0 T1 OS 共享变量a 寄存器a5 ===== ===== ======= ======= =========== ========= 1 L18 -- -- 0 a的高位地址 2 -- -- 切换 0 0 3 -- L18 -- 0 a的高位地址 4 L20 -- -- 0 1 5 -- -- 切换 0 a的高位地址 6 -- L20 -- 0 1 7 -- -- 切换 0 1 8 L23 -- -- 1 1 9 -- -- 切换 1 1 10 -- L23 -- 1 1 ===== ===== ======= ======= =========== ========= 一般情况下,线程 T0 执行完毕后,再执行线程 T1,那么共享全局变量 ``a`` 的值为 2 。但在上面的执行过程中, 可以看到在线程执行指令的过程中会发生线程切换,这样在时刻 10 的时候,共享全局变量 ``a`` 的值为 1 , 这不是我们预期的结果。出现这种情况的原因是两个线程在操作系统的调度下(在哪个时刻调度具有不确定性), 交错执行 ``a = a + 1`` 的不同汇编指令序列,导致虽然增加全局变量 ``a`` 的代码被执行了两次, 但结果还是只增加了 1 。这种多线程的最终执行结果不确定(indeterminate),取决于由于调度导致的、 不确定指令执行序列的情况就是竞态条件(race condition)。 如果每个线程在执行 ``a = a + 1`` 这个 C 语句所对应多条汇编语句过程中,不会被操作系统切换, 那么就不会出现多个线程交叉读写全局变量的情况,也就不会出现结果不确定的问题了。 所以,访问(特指写操作)共享变量代码片段,不能由多个线程同时执行(即并行)或者在一个时间段内都去执行 (即并发)。要做到这一点,需要互斥机制的保障。从某种角度上看,这种互斥性也是一种原子性, 即线程在临界区的执行过程中,不会出现只执行了一部分,就被打断并切换到其他线程执行的情况。即, 要么线程执行的这一系列操作/指令都完成,要么这一系列操作/指令都不做,不会出现指令序列执行中被打断的情况。 锁的基本思路 ----------------------------------------- 要保证多线程并发执行中的临界区的代码具有互斥性或原子性,我们可以建立一种锁, 只有拿到锁的线程才能在临界区中执行。这里的锁与现实生活中的锁的含义很类似。比如,我们可以写出如下的伪代码: .. code-block:: C :linenos: lock(mutex); // 尝试取锁 a = a + 1; // 临界区,访问临界资源 a unlock(mutex); // 是否锁 ... // 剩余区 对于一个应用程序而言,它的执行是受到其执行环境的管理和限制的,而执行环境的主要组成就是用户态的系统库、 操作系统和更底层的处理器,这说明我们需要有硬件和操作系统来对互斥进行支持。一个自然的想法是,这个 ``lock/unlock`` 互斥操作就是CPU提供的机器指令,那上面这一段程序就很容易在计算机上执行了。 但需要注意,这里互斥的对象是线程的临界区代码,而临界区代码可以访问各种共享变量(简称临界资源)。 只靠两条机器指令,难以识别各种共享变量,不太可能约束可能在临界区的各种指令执行共享变量操作的互斥性。 所以,我们还是需要有一些相对更灵活和复杂一点的方法,能够设置一种所有线程能看到的标记, 在一个能进入临界区的线程设置好这个标记后,其他线程都不能再进入临界区了。总体上看, 对临界区的访问过程分为四个部分: 1. 尝试取锁: 查看锁是否可用,即临界区是否可访问(看占用临界区标志是否被设置),如果可以访问, 则设置占用临界区标志(锁不可用)并转到步骤 2 ,否则线程忙等或被阻塞; 2. 临界区: 访问临界资源的系列操作 3. 释放锁: 清除占用临界区标志(锁可用),如果有线程被阻塞,会唤醒阻塞线程; 4. 剩余区: 与临界区不相关部分的代码 根据上面的步骤,可以看到锁机制有两种:让线程忙等的忙等锁(spin lock),以及让线程阻塞的睡眠锁 (sleep lock)。锁的实现大体上基于三类机制:用户态软件、机器指令硬件、内核态操作系统。 下面我们介绍来 rCore 中基于内核态操作系统级方法实现的支持互斥的锁。 我们还需要知道如何评价各种锁实现的效果。一般我们需要关注锁的三种属性: 1. 互斥性(mutual exclusion),即锁是否能够有效阻止多个线程进入临界区,这是最基本的属性。 2. 公平性(fairness),当锁可用时,每个竞争线程是否有公平的机会抢到锁。 3. 性能(performance),即使用锁的时间开销。 内核态操作系统级方法实现锁 --- mutex 系统调用 ------------------------------------------------------------------ 使用 mutex 系统调用 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ 如何能够实现轻量的可睡眠锁?一个自然的想法就是,让等待锁的线程睡眠,让释放锁的线程显式地唤醒等待锁的线程。 如果有多个等待锁的线程,可以全部释放,让大家再次竞争锁;也可以只释放最早等待的那个线程。 这就需要更多的操作系统支持,特别是需要一个等待队列来保存等待锁的线程。 我们先看看多线程应用程序如何使用mutex系统调用的: .. code-block:: C :linenos: :emphasize-lines: 10,16,24,35,45,50 // user/src/ch8b_mut_race.c ... int mutex_id; int a; int threads[thread_count]; void fun(long i) { int t = i + 1; for (int i = 0; i < per_thread; i++) { assert_eq(mutex_lock(mutex_id), 0); int old_a = a; for (int i = 0; i < 500; i++) { t = t * t % 10007; } a = old_a + 1; assert_eq(mutex_unlock(mutex_id), 0); } exit(t); } int main() { int64 start = get_mtime(); assert((mutex_id = mutex_blocking_create()) >= 0); for (int i = 0; i < thread_count; i++) { threads[i] = thread_create(fun, (void *)i); assert(threads[i] > 0); } ... } // usr/lib/syscall.c int mutex_create() { return syscall(SYS_mutex_create, 0); } int mutex_blocking_create() { return syscall(SYS_mutex_create, 1); } int mutex_lock(int mid) { return syscall(SYS_mutex_lock, mid); } int mutex_unlock(int mid) { return syscall(SYS_mutex_unlock, mid); } - 第24行,创建了一个ID为 ``mutex_id`` 的互斥锁,对应的是第35行 ``SYS_mutex_create`` 系统调用; - 第10行,尝试获取锁(对应的是第45行 ``SYS_mutex_lock`` 系统调用),如果取得锁, 将继续向下执行临界区代码;如果没有取得锁,将阻塞; - 第16行,释放锁(对应的是第50行 ``SYS_mutex_unlock`` 系统调用),如果有等待在该锁上的线程, 则唤醒这些等待线程。 mutex 系统调用的实现 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ 操作系统如何实现这些系统调用呢?首先考虑一下与此相关的核心数据结构, 然后考虑与数据结构相关的相关函数/方法的实现。 .. note:: 互斥锁根据 lock 一个已被占用的锁时的行为也分阻塞互斥锁(blocking mutex)与自旋互斥锁(spinning mutex), 下文我们主要聚焦阻塞互斥锁,但是 ucore 里两者都通过结构体 ``struct mutex`` 与几个函数的不同分支实现。 在线程的眼里, **互斥** 是一种每个线程能看到的资源,且在一个进程中,可以存在多个不同互斥资源, 所以我们可以把所有的互斥资源放在一起让进程来管理,如下面代码第 4 行所示。这里需要注意的是: ``struct mutex mutex_pool[LOCK_POOL_SIZE]`` 表示的是实现了 ``mutex`` 的一个“互斥资源”的内存池。而 ``struct mutex`` 是会实现 ``mutex`` 的内核数据结构,它就是我们提到的 **互斥资源** 即 **互斥锁** 。操作系统需要显式地施加某种控制,来确定当一个线程释放锁时,等待的线程谁将能抢到锁。 为了做到这一点,操作系统需要有一个等待队列来保存等待锁的线程,如下面代码的第 10 行所示。 .. code-block:: C :linenos: :emphasize-lines: 4,10 struct proc { ... uint next_mutex_id; struct mutex mutex_pool[LOCK_POOL_SIZE]; }; struct mutex { uint blocking; uint locked; struct queue wait_queue; // "alloc" data for wait queue int _wait_queue_data[WAIT_QUEUE_MAX_LENGTH]; }; 这样,在操作系统中,需要设计实现几个核心成员变量。互斥锁的成员变量有四个:表示是阻塞锁还是自旋锁的 ``blocking``, 是否锁上的 ``locked``,和(阻塞锁会用到的)管理等待线程的等待队列 ``wait_queue`` 及其内存单元 ``_wait_queue_data``; 进程的成员变量:锁内存池 ``mutex_pool`` 及记录互斥锁分配情况的 ``next_mutex_id``。 首先需要创建一个互斥锁,下面是应对 ``SYSCALL_MUTEX_CREATE`` 系统调用的创建互斥锁的函数: .. code-block:: C :linenos: :emphasize-lines: 16-19 // os/syscall.c int sys_mutex_create(int blocking) { struct mutex *m = mutex_create(blocking); if (m == NULL) { return -1; } int mutex_id = m - curr_proc()->mutex_pool; return mutex_id; } // os/sync.c struct mutex *mutex_create(int blocking) { struct proc *p = curr_proc(); if (p->next_mutex_id >= LOCK_POOL_SIZE) { return NULL; } struct mutex *m = &p->mutex_pool[p->next_mutex_id]; p->next_mutex_id++; m->blocking = blocking; m->locked = 0; if (blocking) { // blocking mutex need wait queue but spinning mutex not init_queue(&m->wait_queue, WAIT_QUEUE_MAX_LENGTH, m->_wait_queue_data); } return m; } - 第16~19行,互斥锁池还没用完,就找下一个可行的锁,否则返回NULL。 有了互斥锁,接下来就是实现 ``Mutex`` 的内核函数:对应 ``SYSCALL_MUTEX_LOCK`` 系统调用的 ``sys_mutex_lock`` 。操作系统主要工作是,在锁已被其他线程获取的情况下,把当前线程放到等待队列中, 并调度一个新线程执行。主要代码如下: .. code-block:: C :linenos: :emphasize-lines: 7,14,15-16,18,23,25-26 // os/syscall.c int sys_mutex_lock(int mutex_id) { if (mutex_id < 0 || mutex_id >= curr_proc()->next_mutex_id) { return -1; } mutex_lock(&curr_proc()->mutex_pool[mutex_id]); return 0; } // os/sync.c void mutex_lock(struct mutex *m) { if (!m->locked) { m->locked = 1; return; } if (!m->blocking) { ... } // blocking mutex will wait in the queue struct thread *t = curr_thread(); push_queue(&m->wait_queue, task_to_id(t)); // don't forget to change thread state to SLEEPING t->state = SLEEPING; sched(); // here lock is released (with locked = 1) and passed to me, so just do nothing } - 第 7 行,以 ID 为 ``mutex_id`` 的互斥锁指针 ``m`` 为参数调用 ``mutex_lock`` 方法,具体工作由该方法来完成。 - 第 18 行,分类讨论了自旋互斥锁的实现细节,这里我们聚焦阻塞互斥锁,省略了这一分支的细节; - 第 14 行,如果互斥锁 ``m`` 已经被其他线程获取了,那么在第 23 行,将把当前线程放入等待队列中; 在第 25~26 行,让当前线程处于等待状态,并调度其他线程执行。 - 第 15~16 行,如果互斥锁 ``m`` 还没被获取,那么当前线程会获取给互斥锁,并返回系统调用。 最后是实现 ``Mutex`` 的内核函数:对应 ``SYSCALL_MUTEX_UNLOCK`` 系统调用的 ``sys_mutex_unlock`` 。 操作系统的主要工作是,如果有等待在这个互斥锁上的线程,需要唤醒最早等待的线程。主要代码如下: .. code-block:: C :linenos: :emphasize-lines: 7,18,21-22 // os/syscall.c int sys_mutex_unlock(int mutex_id) { if (mutex_id < 0 || mutex_id >= curr_proc()->next_mutex_id) { return -1; } mutex_unlock(&curr_proc()->mutex_pool[mutex_id]); return 0; } // os/sync.c void mutex_unlock(struct mutex *m) { if (m->blocking) { struct thread *t = id_to_task(pop_queue(&m->wait_queue)); if (t == NULL) { // Without waiting thread, just release the lock m->locked = 0; } else { // Or we should give lock to next thread t->state = RUNNABLE; add_task(t); } } else { m->locked = 0; } } - 第 7 行,以 ID 为 ``mutex_id`` 的互斥锁 ``m`` 为参数调用 ``mutex_unlock`` 方法,具体工作由该方法来完成的。 - 第 18 行,如果等待队列为空,直接释放锁。 - 第 21-22 行,如果等待队列非空,则唤醒等待最久的线程 ``t``,注意这里不改变 ``m->locked``,可以想想为什么(实验问答题之一)。