从零开始写 OS 内核 - 锁与多线程同步操作系统c锁多线程

系列目录

序篇
准备工作
BIOS 启动到实模式
GDT 与保护模式
虚拟内存初探
加载并进入 kernel
显示与打印
全局描述符表 GDT
中断处理
虚拟内存完善
实现堆和 malloc
第一个内核线程
多线程运行与切换
锁与多线程同步
进入用户态
进程的实现
系统调用
简单的文件系统
加载可执行程序
键盘驱动
运行 shell

多线程竞争
上一篇我们终于运行起了多线程，并初步建立起了任务调度系统 scheduler，使这个 kernel 终于开始显现出了一个操作系统应有的面貌。在 multi-threads 运行的基础上，接下来我们需要进入用户态运行 threads，并且建立进程 process 的概念，加载用户可执行程序。
然而在此之前，有一个重要且危险的问题已经伴随着 multi-threads 的运行而降临，这就是线程的竞争和同步问题。相信你应该有用户态下多线程相关的编程经验，理解 threads 之间竞争同步的问题和原因，以及 lock 的概念和使用。本篇会从 kernel 的视角来审视和讨论 lock，以及代码实现。需要说明的是，lock 是一个庞大复杂的课题，且在 kernel 和在用户态下的 lock 的使用方式和实现会有很多的不同（当然也有大部分的共通之处），本篇只是我个人粗浅的理解和实现，欢迎讨论和指正。
锁 Lock
threads 之间竞争导致的数据问题，我想不必在这里多解释。经过前两篇的 thread 启动运行后，我们应该能清楚地认识到，interrupt 是随时并且在任何指令处都可能会发生的，任何非原子的操作都会导致 threads 之间的数据竞争（race）。
对于我们现在这个 kernel，其实已经有很多地方需要加 lock，来保护对公共数据结构的访问，例如:

page fault 处理函数，分配 physical frame 的 bitmap，显然需要保护；
kheap，所有 threads 都在里面挖内存；
scheduler 里的各种任务队列，如 ready_tasks；
......

在大多数支持多线程的编程语言里，都会有 lock 相关的概念和工具，而作为一个一穷二白的 kernel 项目，我们需要自己实现之。
lock 是一门复杂的课题，在安全第一的基础上，设计实现的好坏以及使用方式，会极大地影响系统的性能。坏的 lock 设计和使用，可能会导致 threads 的不合理调度和 CPU 时间的大量浪费，降低系统吞吐性能。
接下来我们从 lock 的底层原理出发，讨论几种常见的 lock 的分类和实现方式，以及它们的使用场景。
原子指令操作从逻辑上讲，lock 的实现很简单：

if (lock_hold == 0) { // Not locked, I get the lock! lock_hold = 1; } else { // Already locked :( }

这里 lock_hold 保存了当前状态是否上锁，值为 true / false。试图拿锁的人，首先检查它是否为 0，是 0 则表示 lock 还未被其它人持有，则我可以拿到 lock，并设为 1，表示锁上，防止后面再有人得到 lock。
然而上面实现的错误之处在于，if 条件的判断和下面的设置 lock_hold = 1 两步操作并非是原子的，两个 threads 可能都在 lock_hold 是 0 且对方还未来得及修改 lock_hold = 1 的情况下执行 if 判断并且都通过，一起拿到 lock 并进入。
这里面的核心问题是对 lock 的判断和修改这两步操作不是原子的，也就是说它们不是一条指令完成的，那么两个 threads 在这里面的交叉运行可能会导致数据 race。
所以任何 lock，最底层实现必须是一条原子指令，即使用一条指令，完成对数据的检验和变更，它保证了只有一个 thread 能顺利通过该指令，而将其它挡在门外，例如：

uint32 compare_and_exchange(volatile uint32* dst, uint32 src);

它必须由汇编实现：

compare_and_exchange: mov edx, [esp + 4] mov ecx, [esp + 8] mov eax, 0 lock cmpxchg [edx], ecx ret

cmpxchg 指令是一条 compare and exchange 指令，它的作用是比较第一个操作数和 eax 的值：

如果相同，则将第二个操作数加载到第一个操作数中；
如果不同，则将第一个操作数的值，赋值给 eax；

（cmpxchg 前面加上 lock 前缀，使该指令在多核 CPU 上执行时保证对内存的访问是独占（exclusive）的且能被其它 core 感知（visible）的，这里涉及到了多核 CPU 的缓存一致性等问题，你可以暂时跳过。对于我们的项目实验用的单核 CPU 而言，lock 前缀不是必须的。）
实际上这条指令就实现了对于检查并修改操作的原子合并。我们用它实现 lock 的逻辑，以操作数 dst 来标志 lock 是否已经被锁上，将它和 eax = 0 比较：

如果相等，那么就是第一种情况，0 说明没有锁上，那么将 1 赋值给 dst，表示拿到 lock 并且上锁，返回值是 eax = 0；
如果不等，那么说明 dst 已经等于 1，lock 已经被别人上锁，那么就是第二种情况，将 dst = 1 值赋值给 eax，返回值是 eax，它已经被修改为 1；

int lock_hold = 0; void acquire_lock() { if (compare_and_exchange(&lock_hold, 1) == 0) { // Get lock! } else { // NOT get lock. } }

除了cmpxchg 指令，还有一种实现方式是 xchg 指令，我个人感觉更好理解：

atomic_exchange: mov ecx, [esp + 4] mov eax, [esp + 8] xchg [ecx], eax ret

xchg 指令，有两个操作数，它表示交换它们的值，然后 atomic_exchange 函数会返回交换后的第二个操作数的值，实际上也就是第一个参数交换前的旧值。
那如何用 atomic_exchange 来实现 lock 的功能？一样的代码：

int lock_hold = 0; void acquire_lock() { if (atomic_exchange(&lock_hold, 1) == 0) { // Get lock! } else { // NOT get lock. } }

试图拿锁的人，总是用 1（上锁）这个值，去和 lock_hold 交换，所以 atomic_exchange 函数，永远返回 lock_hold 的旧值，即将 lock_hold 的旧值交换出来并返回，只有当 lock_hold 旧值为 0 时，上面的判断才能通过，表示锁之前没有被人持有，继而成功拿锁。
可以看到这里也只用了一条指令，完成了对 lock_hold 的变更和检查，有趣的是它是先变更，然后检查，但这丝毫不影响它的正确性。
自旋锁 spinlock 上面讨论了 lock 的 acquire 操作的底层实现，然而这只是 lock 相关问题的冰山一角。lock 真正的复杂的地方在于 acquire 失败后的处理，这也是一种很重要的用来对 lock 的分类方式，它极大地影响 lock 的性能和使用场景。
这里首先要讨论的一种最简单的 lock 类型就是自旋锁（spinlock），它在 acquire 失败后会不断重试直到成功：

#define LOCKED_YES1 #define LOCKED_NO0void spin_lock() { while (atomic_exchange(&lock_hold , LOCKED_YES) != LOCKED_NO) {} }

这是一种 busy wait 的方式，当前 thread 继续持有 CPU 运行，不停地重试 acquire，简单粗暴；
首先需要明确一点，spinlock 不可以在单核 CPU 上使用。单核 CPU 上同一时刻只有一个 thread 被执行，如果拿不到 lock，这样的 spin 空转没有任何意义，因为持有 lock 的 thread 不可能在此期间释放 lock。
然而如果在一个多核 CPU 上，spinlock 却有用武之地。如果 acquire lock 失败，继续重试一段时间，就可能等来持有 lock 的 thread 释放 lock，因为它此时很有可能正在另一个 core 上运行，而且它必定是在 critical section 之内：

文章图片

这对于 critical section 很小，lock 竞争不是很激烈的使用场景就非常适合，因为在这种情况下，spin wait 的时间大概率不会很长。而如果当前 thread 因为拿不到 lock 而放弃了 CPU，则可能会付出更大的代价，这一点后面会详细讨论。
然而如果是 critical section 比较大或者 lock 竞争很激烈的情况，即使在多核 CPU 上，spin lock 也是不适用的，不停地等待和 spin 空转会大量浪费 CPU 时间，这是不明智的。
yield lock 上面说了，spinlock 对于单核 CPU 是没有任何意义的，不过我们这个 kernel 恰好是在单核 CPU 模拟器上运行，所以需要实现一种类似 spinlock 的轻量级 lock，我姑且称它为 yield lock。
顾名思义，yield lock 是指在 acquire 失败后，主动让出 CPU。也就是说我暂时拿不到 lock，我去休息一会儿，让其它 threads 先运行，等它们轮转完一圈，我再回来试试看。
它的行为本质上也是一种 spin，但不同于原地空转，它没有浪费任何 CPU 时间，而是立即将 CPU 让给了别人，这样持有 lock 的 thread 就可能得到运行，待下一轮时间片运行完后，它就很有可能已经释放了 lock：

文章图片

void yield_lock() { while (atomic_exchange(&lock_hold , LOCKED_YES) != LOCKED_NO) { schedule_thread_yield(); } }

注意这里必须将 schedule_thread_yield 放在 while 循环里，因为即使持有 lock 的 thread 释放了锁，也不代表当前 thread 等会儿一定能拿到 lock，因为可能还有别的竞争者，所以在 yield 回来后，必须再次竞争 acquire lock；
和 spinlock 类似，yield lock 也适合于 critical section 比较小，竞争不是很激烈的情形，否则很多 threads 一次次地空等，也是对 CPU 资源的浪费。
阻塞锁上面两种锁，都是非阻塞锁，也就是说 thread 在 acquire 失败的情况下不会 block，而是不停重试，或者过一段时间重试，本质上都是在重试。然而在 critical section 比较大或者 lock 的竞争比较激烈的情况下，不断重试很可能是徒劳的，这是对 CPU 资源的浪费。
为了解决这个问题就有了阻塞锁（blocking lock），它的内部会维护一个队列，如果 thread 拿不到锁，会将自己加入到这个队列里睡眠，让出 CPU，在睡眠期间它不会再被调度运行，即进入了阻塞态；等到持有 lock 的 thread 释放 lock 时，会从队列里拿出一个 thread 重新唤醒。
例如，我们定义以下的阻塞锁，命名为 mutex：

struct mutex { volatile uint32 hold; linked_list_t waiting_task_queue; yieldlock_t ydlock; };

上锁的实现：

void mutex_lock(mutex_t* mp) { yieldlock_lock(&mp->ydlock); while (atomic_exchange(&mp->hold, LOCKED_YES) != LOCKED_NO) { // Add current thread to wait queue. thread_node_t* thread_node = get_crt_thread_node(); linked_list_append(&mp->waiting_task_queue, thread_node); // Mark this task status TASK_WAITING so that // it will not be put into ready_tasks queue // by scheduler. schedule_mark_thread_block(); yieldlock_unlock(&mp->ydlock); schedule_thread_yield(); // Waken up, and try acquire lock again. yieldlock_lock(&mp->ydlock); } yieldlock_unlock(&mp->ydlock); }

这里上锁（lock）的实现已经比较复杂了，这其实是标准的 conditional wait 的实现方法。所谓 conditional wait，即条件等待，就是以阻塞的方式，等待一个期望的条件得到满足。这里所要等待的期望条件就是：lock 被释放，我可以去尝试获得 lock。
在尝试获取 lock 失败后，当前 thread 将自己加入到 mutex 的 waiting_task_queue，并且标记自己为 TASK_WAITING 状态，然后让出了 CPU；这里的让出 CPU 和上面的 yield lock 里一样，都调用了 schedule_thread_yield 函数，然而它们有着本质区别：

yield lock 里的 thread 经过 yield，仍然会被放入 ready_tasks 队列，后面仍然会被 scheduler 调度；
而这里 thread 先把自己标记为了TASK_WAITING，这样在 schedule_thread_yield 的实现里，该 thread 不会被添加到 ready_tasks 队列里，它因而实际上进入了阻塞状态，不会被再次调度，直到持有 lock 的 thread 下次 unlock 时，将它从 mutex 的 waiting_task_queue 里取出唤醒，重新放到 ready_tasks 队列；

void mutex_unlock(mutex_t* mp) { yieldlock_lock(&mp->ydlock); mp->hold = LOCKED_NO; if (mp->waiting_task_queue.size > 0) { // Wake up a waiting thread from queue. thread_node_t* head = mp->waiting_task_queue.head; linked_list_remove(&mp->waiting_task_queue, head); // Put waken up thread back to ready_tasks queue. add_thread_node_to_schedule(head); } yieldlock_unlock(&mp->ydlock); }

上面的 lock 和 unlock 的代码里都有个关键元素，那就是定义在 mutex 内部的一个 yieldlock。这看上去似乎比较奇怪，因为 mutex 的本质功能就是锁，结果这个锁的内部数据居然还需要另一把锁去保护它自己，这岂不是套娃？
从实现上来说，mutex 已经是一种比较复杂的锁了，它内部维护了 waiting 队列，而这个队列显然需要被保护起来，所以就有了上面的套娃悖论。这里关键点在于，这两层锁的类型和存在目的是有本质区别的：

mutex 是一把重型锁，它是提供给外部使用的，它的用途和保护对象是是不确定的，一般来说是 critical section 比较大，竞争比较激烈的区域；
内部 yield lock 是轻量级的锁，这把锁的用途和保护对象是是确定的，它保护的是 mutex 内部的操作，这个 critical section 可以控制得非常小，所以这把锁的引入是必须而且合理的；

【从零开始写 OS 内核 - 锁与多线程同步】内部 yield lock 的代价在于，它引入了新的竞争，使整个 mutex 上的 threads 竞争更加激烈。然而这样的额外 cost 从 mutex 的设计和使用情形上来说是不可避免的，某种意义上来说也是可以承受和忽略的：因为通常认为 mutex 保护的外部 critical section 是比较大的，相比于其内部 yield lock 保护的区域而言。
kernel 和用户态的 lock
以上讨论的是几种锁的原理和实现方式，以及它们的使用场景的不同。一个很重要的区分原则是，critical section 的大小和竞争的激烈程度，这本质上反映的是一个 thread 每次尝试得到这把锁的容易程度（或者机率），以此为依据，我们可以分成两类使用情形：

critical section 很小，且竞争不激烈，那么使用 spin 类型的 lock（包括 spinlock 和 yieldlock），它们是非阻塞的；
critical section 很大，或者竞争激烈，那么使用阻塞锁；

实际上在 kernel 态下使用哪种锁的选择远不止那么简单，它还涉及到使用锁的地方，例如是中断上下文还是线程上下文，以及其它的很多考虑，都会带来很多限制和不同。关于这些问题，以后有时间我会尝试单独写一篇文章讨论，先挖个坑。
而如果到了用户态，lock 的使用和 kernel 态又会有很大的不同，一个被讨论的最多就是关于阻塞锁和 spinlock 的选择。上面已经说了，阻塞锁常被用于 critical section 很大，或者竞争激烈的情形，因为它不会导致 CPU 大量空转，看似是节约了 CPU 资源，然而用户态下的线程进入阻塞睡眠需要陷入 kernel 态，这对于 CPU 的性能影响是非常大的，可能还不如原地 spin 等待来的划算（前提是多核 CPU），所以这里的考量和 kernel 态下锁的使用又会有很大不同。
总结
本篇讨论了 lock 的原理和实现，限于自身水平，只是我个人粗浅的理解，希望能对读者有所帮助，也欢迎一起讨论评述。在这个 scroll 项目中，性能暂时并不是我们考虑的问题，为了简单安全起见，我大量使用了 yieldlock 作为 kernel 下的主要用锁。