操作系统导论——第28章 锁

        通过对并发的介绍，看到了并发编程的一个最基本问题：希望原子式执行一系列指令，但由于单处理器上的中断（或者多个线程在多处理器上并发执行），做不到。本章介绍了锁（lock），直接解决这一问题。程序员在源代码中加锁，放在临界区周围，保证临界区能够像单条原子指令一样执行。

一、锁的基本思想

        举个例子，假设临界区像这样，典型的更新共享变量：

balance = balance+1;

        当然，其他临界区也是可能的，比如为链表增加一个元素，或对共享结构的复杂更新操作。为了使用锁，给临界区增加了这样一些代码：

lock_t mutex; // 定义一个锁变量

...

lock(&mutex);

balance = balance + 1;

unlock(&mutex);

        锁就是一个变量，因此我们需要声明一个某种类型的锁变量（lock variable，如上面的mutex），才能使用。这个锁变量（简称锁）保存了锁在某一时刻的状态。它要么是可用的 （available，或 unlocked，或 free），表示没有线程持有锁，要么是被占用的（acquired，或locked，或 held），表示有一个线程持有锁，正处于临界区。我们也可以保存其他的信息，比如持有锁的线程，或请求获取锁的线程队列，但这些信息会隐藏起来，锁的使用者不会发现。

         lock() 和 unlock() 函数的语义很简单。调用 lock() 尝试获取锁，如果没有其他线程持有锁 （即它是可用的），该线程会获得锁，进入临界区。这个线程有时被称为锁的持有者（owner）。 如果另外一个线程对相同的锁变量（本例中的mutex）调用 lock()，因为锁被另一线程持有， 该调用不会返回。这样，当持有锁的线程在临界区时，其他线程就无法进入临界区。

         锁的持有者一旦调用 unlock()，锁就变成可用了。如果没有其他等待线程（即没有其他线程调用过 lock() 并卡在那里），锁的状态就变成可用了。如果有等待线程（卡在 lock() 里），其中一个会（最终）注意到（或收到通知）锁状态的变化，获取该锁，进入临界区。

         锁为程序员提供了最小程度的调度控制。我们把线程视为程序员创建的实体，但是被操作系统调度，具体方式由操作系统选择。锁让程序员获得一些控制权。通过给临界区加锁，可以保证临界区内只有一个线程活跃。锁将原本由操作系统调度的混乱状态变得更为可控。

二、Pthread 锁

        POSIX 库将锁称为互斥量（mutex），因为它被用来提供线程之间的互斥。即当一个线程在临界区，它能够阻止其他线程进入直到本线程离开临界区。因此，如果你看到下面的 POSIX 线程代码，应该理解它和上面的代码段执行相同的任务（再次使用了包装函数来检查获取锁和释放锁时的错误）。

pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER; // 初始化锁

Pthread_mutex_lock(&lock); // wrapper for pthread_mutex_lock()

balance = balance + 1;

Pthread_mutex_unlock(&lock);

        POSIX 的 lock 和 unlock 函数会传入一个变量，我们可能用不同的锁来保护不同的变量。这样可以增加并发：不同于任何临界区都使用同一个大锁（粗粒度的锁策略），通常大家会用不同的锁保护不同的数据和结构，从而允许更多的线程进入临界区（细粒度的方案）。

三、实现一个锁 

                                                     关键问题：怎样实现一个锁

        如何构建一个高效的锁？高效的锁能够以低成本提供互斥，同时能够实现一些特性，我们下面会讨论。需要什么硬件支持？什么操作系统支持？

        需要硬件和操作系统的帮助来实现一个可用的锁。近些年来，各种计算机体系结构的指令集都增加了一些不同的硬件原语，研究如何使用它们来实现像锁这样的互斥原语。我们也会研究操作系统如何发展完善，支持实现成熟复杂的锁库。

四、评价锁

        如何评价一种锁实现的效果。评价锁是否能工作（并工作的好），先设立一些标准。

        ① 锁是否能完成其基本任务，即提供互斥。最基本的，锁是否有效，能够阻止多个线程进入临界区。

        ② 公平性。当锁可用时，是否给一个竞争线程有公平的机会抢到锁？用另一个方式看这个问题是检查更极端的情况：是否有竞争锁的线程会饿死，一直无法获得锁。

        ③ 性能。具体来说，是使用锁之后增加的时间开销。有几种场景需要考虑。一种是没有竞争的情况，即只有一个线程抢锁、释放锁的开支如何？另外一种是一个CPU上多个线程竞争，性能如何？最后一种是多个CPU、多个线程竞争时的性能。通过比较不同的场景，能够更好地理解不同的锁技术对性能的影响，下面会进行介绍。

 五、控制中断

        最早提供的互斥解决方案之一，就是在临界区关闭中断。这个解决方案是为单处理器系统开发的。代码如下：

void lock() { DisableInterrupts(); 
} void unlock() { EnableInterrupts(); 
}

        假设我们运行在这样一个单处理器系统上。通过在进入临界区之前关闭中断（使用特殊的硬件指令），可以保证临界区的代码不会被中断，从而原子地执行。结束之后，我们重新打开中断（同样通过硬件指令），程序正常运行。

        这个方法的主要优点就是简单。没有中断，线程可以确信它的代码会继续执行下去，不会被其他线程干扰。

         遗憾的是，缺点很多。首先，这种方法要求我们允许所有调用线程执行特权操作（打开关闭中断），即信任这种机制不会被滥用。众所周知，如果我们必须信任任意一个程序，可能就有麻烦了。这里，麻烦表现为多种形式：第一，一个贪婪的程序可能在它开始时就调用lock()，从而独占处理器。更糟的情况是，恶意程序调用 lock() 后，一直死循环。后一种情况，系统无法重新获得控制，只能重启系统。关闭中断对应用要求太多，不太适合作为通用的同步解决方案。

         第二，这种方案不支持多处理器。如果多个线程运行在不同的CPU上，每个线程都试图进入同一个临界区，关闭中断也没有作用。线程可以运行在其他处理器上，因此能够进入临界区。

        第三，关闭中断导致中断丢失，可能会导致严重的系统问题。假如磁盘设备完成了读取请求，但 CPU 错失了这一事实，那么，操作系统如何知道去唤醒等待读取的进程？

         最后一个原因是效率低。与正常指令执行相比，现代 CPU 对于关闭和打开中断的代码执行得较慢。

基于以上原因，只在很有限的情况下用关闭中断来实现互斥原语。例如，在某些情况下操作系统本身会采用屏蔽中断的方式，保证访问自己数据结构的原子性，或至少避免某些复杂的中断处理情况。这种用法是可行的，因为在操作系统内部不存在信任问题，它总是信任自己可以执行特权操作

六、测试并设置指令（原子交换）

        因为关闭中断的方法无法工作在多处理器上，因此系统设计者开始让硬件支持锁。

        最简单的硬件支持是测试并设置指令（test-and-set instruction），也叫作原子交换（atomic exchange）。为了理解 test-and-set 如何工作，首先实现一个不依赖它的锁，用一个变量标记锁是否被持有。

        在第一次尝试中（见图 28.1），想法很简单：用一个变量来标志锁是否被某些线程占用。第一个线程进入临界区，调用 lock()，检查标志是否为 1（这里不是 1 ），然后设置标志位 1 ，表明线程持有该锁。结束临界区时，线程调用 unlock，清楚标志，表示锁未被持有。 

         当第一个线程正处于临界区时，如果另一个线程调用 lock()，它会在 while 循环中自旋等待（spin-wait)，直到第一个线程调用 unlock() 清空标志。然后等待的线程会退出 while 循环，设置标志，执行临界区代码。

         遗憾的是，这段代码有两个问题：正确性和性能。这个正确性问题在并发编程中很常见。假设代码按照表 28.1 执行，开始时 flag=0。

         从这种交替执行可以看出，通过适时的（不合时宜的？）中断，我们很容易构造出两个线程都将标志设置为 1，都能进入临界区的场景。这种行为就是专家所说的“不好”，显然没有满足最基本的要求：互斥。

        性能问题（稍后会有更多讨论）主要是线程在等待已经被持有的锁时，采用了自旋等待（spin-waiting）的技术，就是不停地检查标志的值。自旋等待在等待其他线程释放锁的时候会浪费时间。尤其是在单处理器上，一个等待线程等待的目标线程甚至无法运行（至少在上下文切换之前）！

七、实现可用的自旋锁

        尽管上面例子的想法很好，但没有硬件的支持是无法实现的。幸运的是，一些系统提供了这一指令，支持基于这种概念创建简单的锁。这个更强大的指令有不同的名字：在 SPARC 上，这个指令叫 ldstub（load/store unsigned byte，加载/保存无符号字节）；在 x86上，是xchg （atomic exchange，原子交换）指令。但它们基本上在不同的平台上做同样的事，通常称为测试并设置指令（test-and-set）。我们用如下的 C代码片段来定义测试并设置指令做了什么：

int TestAndSet(int *old_ptr, int new) { int old = *old_ptr; // fetch old value at old_ptr *old_ptr = new;    // store 'new' into old_ptr return old;        // return the old value 
}

        测试并设置指令做了下述事情。它返回 old_ptr 指向的旧值，同时更新为 new 的新值。 当然，关键是这些代码是原子地（atomically）执行。因为既可以测试旧值，又可以设置新值，所以我们把这条指令叫作“测试并设置”。这一条指令完全可以实现一个简单的自旋锁 （spin lock），如图 28.2 所示。

     确保理解为什么这个锁能工作。① 首先假设一个线程在运行，调用 lock() ，没有其他线程持有锁，所以 flag=0. 当调用TestAndSet(flag, 1)方法，返回 0，线程会跳出 while 循环，获取锁。同时也会原子的设置flag为1，标志锁已被持有。当线程离开临界区，调用 unlock() 将 flag 清理为0

 

        ② 第二种场景，某一个线程已经持有锁（flag 为 1）。本线程调用 lock()，然后调用 TestAndSet（flag,1），这一次返回1。 只要另一个线程一直持有锁，TestAndSet() 会重复返回1， 本线程会一直自旋。当 flag 终于被改为 0，本线程会调用 TestAndSet()，返回 0 并且原子地设置为1，从而获得锁，进入临界区。

        将测试（test 旧的锁值）和设置（set 新的值）合并为一个原子操作之后，我们保证了只有一个线程能获取锁。这就实现了一个有效的互斥原语！

         这种锁被称为 自旋锁（spin lock）。这是最简单的一种锁，一直自旋，利用 CPU 周期，直到锁可用。在单处理器上，需要抢占式的调度器（即不断通过时钟中断一个进程，运行其他线程）。否则，自旋锁在单 CPU 上无法使用，因为一个自旋的线程永远不会放弃 CPU。

八、评价自旋锁

        正确性：能够互斥。自旋锁一次只允许一个线程进入临界区。

        公平性：自旋锁不提供任何公平性保证。实际上，自旋的线程在竞争条件下可能会永远自旋。自旋锁没有公平性，可能会导致饿死。

        性能：考虑几种不同的情况，考虑线程在单处理器上竞争锁的情况，然后考虑这些线程跨多个处理器。

        对于自旋锁，在单CPU的情况下，性能开销相当大。假设一个线程持有锁进入临界区时被抢占。调度器可能会运行其他每一个线程（假设有N−1个这种线程）。而其他线程都在竞争锁，都会在放弃CPU之前，自旋一个时间片，浪费 CPU 周期。

         但是，在多CPU上，自旋锁性能不错（如果线程数大致等于CPU数）。假设线程A在 CPU 1，线程 B 在CPU 2竞争同一个锁。线程A（CPU 1）占有锁时，线程B竞争锁就会自旋（在CPU 2上）。然而，临界区一般都很短，因此很快锁就可用，然后线程B获得锁。自旋等待其他处理器上的锁，并没有浪费很多CPU周期，因此效果不错。

九、比较并交换

         某些系统提供了另一个硬件原语，即比较并交换指令（SPARC系统中是compare-and-swap， x86 系统是compare-and-exchange）。图 28.3 是这条指令的C语言伪代码。

int CompareAndSwap(int *ptr, int expected, int new) { int actual = *ptr; if (actual == expected)      *ptr = new; return actual; 
} 

        比较并交换的基本思路是检测 ptr 指向的值是否和 expected 相等；如果是，更新 ptr 所指的值为新值。否则，什么也不做，不论哪种情况，都返回该内存地址的实际值，让调用者能够知道执行是否成功。

        有了比较交换指令，就可以实现一个锁，类似于用测试并设置那样。例如，只要用下面的代码替换 lock() 函数： 

void lock(lock_t *lock) { while (CompareAndSwap(&lock->flag, 0, 1) == 1) ; // spin 
} 

        其余代码和上面测试并设置的例子完全一样。代码工作的方式很类似，检查标志是否为0，如果是，原子地交换为 1，从而获得锁。锁被持有时，竞争锁的线程会自旋。

十、链接的加载和条件式存储指令

        一些平台提供了实现临界区的一对指令。例如 MIPS 架构[H93]中，链接的加载 （load-linked）和条件式存储（store-conditional）可以用来配合使用，实现其他并发结构。 图28.4 是这些指令的C语言伪代码。

int LoadLinked(int *ptr) { return *ptr; 
} int StoreConditional(int *ptr, int value) { if (no one has updated *ptr since the LoadLinked to this address) { *ptr = value; return 1; // success! } else { return 0; // failed to update } 
}

        链接的加载指令和典型加载指令类似，都是从内存中取出值存入一个寄存器。关键区别来自条件式存储（store-conditional）指令，只有上一次加载的地址在期间都没有更新时， 才会成功，（同时更新刚才链接的加载的地址的值）。成功时，条件存储返回1，并将ptr指的值更新为value。失败时，返回 0，并且不会更新值。

         使用链接的加载和条件式存储来实现一个锁。解决方案如图 28.5 所示。

        lock() 代码：首先，一个线程自旋等待标志被设置为 0（因此表明锁没有被保持）。一旦如此，线程尝试通过条件存储获取锁。如果成功，则线程自动将标志值更改为1，从而可以进入临界区。

        条件式存储失败是如何发生的。一个线程调用 lock()，执行了链接的加载指令，返回 0。在执行条件式存储之前，中断产生了，另一个线程进入 lock 的代码，也执行链接式加载指令，同样返回 0。现在，两个线程都执行了链接式加载指令，将要执行条件存储。重点是只有一个线程能够成功更新标志为 1，从而获得锁；第二个执行条件存储的线程会失败（因为另一个线程已经成功执行了条件更新），必须重新尝试获取锁。

有更加简短代码的实现：

void lock(lock_t *lock) { while (LoadLinked(&lock->flag)||!StoreConditional(&lock->flag, 1)) ; // spin 
}

十一、获取并增加

        最后一个硬件原语是获取并增加（fetch and add）指令，它能原子地返回特定地址的旧值，并且让该值自增一。获取并增加的 C 语言伪代码如下：

        在这个例子中，我们会用获取并增加指令，实现一个更有趣的 ticket 锁，图 28.6 是lock 和unlock 的代码。

 

      不是用一个值，使用了 ticket 和 turn 变量来构建锁。基本操作也很简单：如果线程希望获取锁，首先对一个 ticket 值执行一个原子的获取并相加指令。这个值作为该线程的“turn”（顺位，即myturn）。根据全局共享的 lock->turn 变量，当某一个线程的（myturn == turn）时，则轮到这个线程进入临界区。unlock则是增加 turn，从而下一个等待线程可以进入临界区。

        不同于之前的方法：本方法能够保证所有线程都能抢到锁。只要一个线程获得了 ticket 值，它最终会被调度。之前的方法则不会保证。比如基于测试并设置的方法，一个线程有可能一直自旋，即使其他线程在获取和释放锁。 

十二、自旋过多

        基于硬件的锁简单（只有几行代码）而且有效，这也是任何好的系统或者代码的特点。但是，某些场景下，这些解决方案会效率低下。以两个线程运行在单处理器上为例，当一个线程（线程0）持有锁时，被中断。第二个线程（线程1）去获取锁，发现锁已经被持有。因此，它就开始自旋。接着自旋。

        然后它继续自旋。最后，时钟中断产生，线程 0 重新运行，它释放锁。最后（比如下次它运行时），线程1 不需要继续自旋了，它获取了锁。因此，类似的场景下，一个线程会一直自旋检查一个不会改变的值，浪费掉整个时间片！如果有 N 个线程去竞争一个锁，情况会更糟糕。同样的场景下，会浪费N−1个时间片，只是自旋并等待一个线程释放该锁。 因此，下一个问题是： 

                                             关键问题：怎样避免自旋

        如何让锁不会不必要地自旋，浪费CPU时间？

        只有硬件支持是不够的，还需要操作系统支持。

 十三、简单方法

        问题存在：如果临界区的线程发生上下文切换，其他线程只能一直自旋，等待被中断的（持有锁的）进程重新运行。有什么好办法？

         第一种简单友好的方法就是，在要自旋的时候，放弃CPU。图 28.7 展示了这种方法：

        在这种方法中，假定操作系统提供原语 yield()，线程可以调用它主动放弃 CPU，让其他线程运行。线程可以处于 3 种状态之一（运行、就绪和阻塞）。yield()系统调用能够让运行（running）态变为就绪（ready）态，从而允许其他线程运行。因此，让出线程本质上取消调度（deschedules）了它自己。

         考虑在 单CPU 上运行两个线程。在这个例子中，基于 yield 的方法十分有效。一个线程调用lock()，发现锁被占用时，让出 CPU，另外一个线程运行，完成临界区。在这个简单的例子中，让出方法工作得非常好。

         现在来考虑许多线程（例如 100 个）反复竞争一把锁的情况。在这种情况下，一个线程持有锁，在释放锁之前被抢占，其他 99 个线程分别调用 lock()，发现锁被抢占，然后让出CPU。假定采用某种轮转调度程序，这99个线程会一直处于运行—让出这种模式，直到持有锁的线程再次运行。虽然比原来的浪费 99 个时间片的自旋方案要好，但这种方法仍然成本很高，上下文切换的成本是实实在在的，因此浪费很大。

         更糟的是，还没有考虑饿死的问题。一个线程可能一直处于让出的循环，而其他线程反复进出临界区。

十四、使用队列：休眠替代自旋

        显式地施加某种控制，决定锁释放时，谁能抢到锁。为了做到这一点，需要操作系统的更多支持，并需要一个队列来保存等待锁的线程。

         提供了两个调用：park()能够让调用线程休眠，unpark(threadID)则会唤醒 threadID 标识的线程。可以用这两个调用来实现锁，让调用者在获取不到锁时睡眠，在锁可用时被唤醒。我们来看看图28.8中的代码，理解这组原语的一种可能用法。

typedef struct  lock_t { int flag; int guard; // 辅助锁，确保对锁的状态和队列的操作是原子的queue_t *q;  // 指向队列的指针
} lock_t; void lock_init(lock_t *m) { m->flag = 0; m->guard = 0;  // 辅助锁未持有queue_init(m->q);  // 对队列进行初始化
} void lock(lock_t *m) { while (TestAndSet(&m->guard, 1) == 1)  // 尝试获取辅助锁; //acquire guard lock by spinning if (m->flag == 0) { m->flag = 1; // lock is acquired m->guard = 0;  // 释放辅助锁} else { queue_add(m->q, gettid());  // 把当前线程 ID 加入等待队列qm->guard = 0; park();  // 调用park使得当前线程进入阻塞状态} 
} void unlock(lock_t *m) { while (TestAndSet(&m->guard, 1) == 1)  // 尝试获取辅助锁，如果持有辅助锁，则保持自旋; //acquire guard lock by spinning if (queue_empty(m->q)) // 判断队列是否为空m->flag = 0; // let go of lock; no one wants it else  unpark(queue_remove(m->q)); // hold lock (for next thread!) m->guard = 0;  // 释放辅助锁
} 

         首先，将之前的测试并设置和等待队列结合，实现了一个更高性能的锁。其次，通过队列来控制谁会获得锁，避免饿死。

         guard 基本上起到了自旋锁的作用，围绕着 flag 和 队列 操作。因此，这个方法并没有完全避免自旋等待。线程在获取锁或者释放锁时可能被中断，从而导致其他线程自旋等待。但是，这个自旋等待时间是很有限的（不是用户定义的临界区，只是在 lock 和 unlock 代码中的几个指令），因此，这种方法也许是合理的。

         第二点，可能注意到在 lock() 函数中，如果线程不能获取锁（它已被持有），线程会把自己加入队列（通过调用 gettid()获得当前的线程ID），将 guard 设置为 0，然后让出CPU。 留给读者一个问题：如果我们在 park() 之后，才把 guard 设置为 0 释放锁，会发生什么呢？ 提示一下，这是有问题的。

        还可能注意到了很有趣一点，当要唤醒另一个线程时，flag并没有设置为 0。为什么呢？其实这不是错，而是必须的！线程被唤醒时，就像是从park()调用返回。但是，此时它没有持有guard，所以也不能将 flag 设置为1。因此，我们就直接把锁从释放的线程传递给下一个获得锁的线程，期间flag不必设置为0。 

         最后，你可能注意到解决方案中的竞争条件，就在park()调用之前。如果不凑巧，一个线程将要 park，假定它应该睡到锁可用时。这时切换到另一个线程（比如持有锁的线程），这可能会导致麻烦。比如，如果该线程随后释放了锁。接下来第一个线程的park会永远睡下去（可能）。这种问题有时称为唤醒/等待竞争（wakeup/waiting race）。为了避免这种情况，我们需要额外的工作。

        Solaris 通过增加了第三个系统调用 separk() 来解决这一问题。通过 setpark()，一个线程表明自己马上要 park。如果刚好另一个线程被调度，并且调用了 unpark，那么后续的 park 调用就会直接返回，而不是一直睡眠。 lock() 调用做一个小修改： 

queue_add(m->q, gettid());  

setpark(); // new code

m->guard = 0; 

         另外一种方案就是将 guard 传入内核。在这种情况下，内核能够采取预防措施，保证原子地释放锁，把运行线程移出队列。

十五、不同操作系统，不同实现

         为了构建更有效率的锁，一个操作系统提供的一种支持。其他操作系统也提供了类似的支持，但细节不同。

         例如，Linux 提供了 futex，它类似于 Solaris 的接口，但提供了更多内核功能。具体来说， 每个 futex 都关联一个特定的物理内存位置，也有一个事先建好的内核队列。调用者通过 futex 调用（见下面的描述）来睡眠或者唤醒。

        具体来说，有两个调用。调用 futex_wait(address, expected) 时，如果 address 处的值等于 expected，就会让调线程睡眠。否则，调用立刻返回。调用 futex_wake(address) 唤醒等待队 列中的一个线程。图28.9是Linux环境下的例子。

 

        这段代码来自 nptl 库（gnu libc 库的一部分）[L09]中 lowlevellock.h，它很有趣。基本上，它利用一个整数，同时记录锁是否被持有（整数的最高位），以及等待者的个数（整数的其余所有位）。因此，如果锁是负的，它就被持有（因为最高位被设置，该位决定了整数的符号）。这段代码的有趣之处还在于，它展示了如何优化常见的情况，即没有竞争时：只有一个线程获取和释放锁，所做的工作很少（获取锁时测试和设置的原子位运算，释放锁时原子的加法）。你可以看看这个“真实世界”的锁的其余部分，是否能理解其工作原理。

十六、两阶段锁