睡眠与唤醒
上面解法中的 Peterson 、TSL 和 XCHG 解法都是正确的,但是它们都有忙等待的缺点。这些解法的本质上都是一样的,先检查是否能够进入临界区,若不允许,则该进程将原地等待,直到允许为止。
这种方式不但浪费了 CPU 时间,而且还可能引起意想不到的结果。考虑一台计算机上有两个进程,这两个进程具有不同的优先级,H
是属于优先级比较高的进程,L
是属于优先级比较低的进程。进程调度的规则是不论何时只要 H 进程处于就绪态 H 就开始运行。在某一时刻,L 处于临界区中,此时 H 变为就绪态,准备运行(例如,一条 I/O 操作结束)。现在 H 要开始忙等,但由于当 H 就绪时 L 就不会被调度,L 从来不会有机会离开关键区域,所以 H 会变成死循环,有时将这种情况称为优先级反转问题(priority inversion problem)
。
现在让我们看一下进程间的通信原语,这些原语在不允许它们进入关键区域之前会阻塞而不是浪费 CPU 时间,最简单的是 sleep
和 wakeup
。Sleep 是一个能够造成调用者阻塞的系统调用,也就是说,这个系统调用会暂停直到其他进程唤醒它。wakeup 调用有一个参数,即要唤醒的进程。还有一种方式是 wakeup 和 sleep 都有一个参数,即 sleep 和 wakeup 需要匹配的内存地址。
生产者-消费者问题
作为这些私有原语的例子,让我们考虑生产者-消费者(producer-consumer)
问题,也称作 有界缓冲区(bounded-buffer)
问题。两个进程共享一个公共的固定大小的缓冲区。其中一个是生产者(producer)
,将信息放入缓冲区, 另一个是消费者(consumer)
,会从缓冲区中取出。也可以把这个问题一般化为 m 个生产者和 n 个消费者的问题,但是我们这里只讨论一个生产者和一个消费者的情况,这样可以简化实现方案。
如果缓冲队列已满,那么当生产者仍想要将数据写入缓冲区的时候,会出现问题。它的解决办法是让生产者睡眠,也就是阻塞生产者。等到消费者从缓冲区中取出一个或多个数据项时再唤醒它。同样的,当消费者试图从缓冲区中取数据,但是发现缓冲区为空时,消费者也会睡眠,阻塞。直到生产者向其中放入一个新的数据。
这个逻辑听起来比较简单,而且这种方式也需要一种称作 监听
的变量,这个变量用于监视缓冲区的数据,我们暂定为 count,如果缓冲区最多存放 N 个数据项,生产者会每次判断 count 是否达到 N,否则生产者向缓冲区放入一个数据项并增量 count 的值。消费者的逻辑也很相似:首先测试 count 的值是否为 0 ,如果为 0 则消费者睡眠、阻塞,否则会从缓冲区取出数据并使 count 数量递减。每个进程也会检查检查是否其他线程是否应该被唤醒,如果应该被唤醒,那么就唤醒该线程。下面是生产者消费者的代码
#define N 100 /* 缓冲区 slot 槽的数量 */ int count = 0 /* 缓冲区数据的数量 */ // 生产者 void producer(void){ int item; while(TRUE){ /* 无限循环 */ item = produce_item() /* 生成下一项数据 */ if(count == N){ sleep(); /* 如果缓存区是满的,就会阻塞 */ } insert_item(item); /* 把当前数据放在缓冲区中 */ count = count + 1; /* 增加缓冲区 count 的数量 */ if(count == 1){ wakeup(consumer); /* 缓冲区是否为空?*/ } } } // 消费者 void consumer(void){ int item; while(TRUE){ /* 无限循环 */ if(count == 0){ /* 如果缓冲区是空的,就会进行阻塞 */ sleep(); } item = remove_item(); /* 从缓冲区中取出一个数据 */ count = count - 1 /* 将缓冲区的 count 数量减一 */ if(count == N - 1){ /* 缓冲区满嘛?*/ wakeup(producer); } consumer_item(item); /* 打印数据项 */ } }
为了在 C 语言中描述像是 sleep
和 wakeup
的系统调用,我们将以库函数调用的形式来表示。它们不是 C 标准库的一部分,但可以在实际具有这些系统调用的任何系统上使用。代码中未实现的 insert_item
和 remove_item
用来记录将数据项放入缓冲区和从缓冲区取出数据等。
现在让我们回到生产者-消费者问题上来,上面代码中会产生竞争条件,因为 count 这个变量是暴露在大众视野下的。有可能出现下面这种情况:缓冲区为空,此时消费者刚好读取 count 的值发现它为 0 。此时调度程序决定暂停消费者并启动运行生产者。生产者生产了一条数据并把它放在缓冲区中,然后增加 count 的值,并注意到它的值是 1 。由于 count 为 0,消费者必须处于睡眠状态,因此生产者调用 wakeup
来唤醒消费者。但是,消费者此时在逻辑上并没有睡眠,所以 wakeup 信号会丢失。当消费者下次启动后,它会查看之前读取的 count 值,发现它的值是 0 ,然后在此进行睡眠。不久之后生产者会填满整个缓冲区,在这之后会阻塞,这样一来两个进程将永远睡眠下去。
引起上面问题的本质是 唤醒尚未进行睡眠状态的进程会导致唤醒丢失。如果它没有丢失,则一切都很正常。一种快速解决上面问题的方式是增加一个唤醒等待位(wakeup waiting bit)
。当一个 wakeup 信号发送给仍在清醒的进程后,该位置为 1 。之后,当进程尝试睡眠的时候,如果唤醒等待位为 1 ,则该位清除,而进程仍然保持清醒。
然而,当进程数量有许多的时候,这时你可以说通过增加唤醒等待位的数量来唤醒等待位,于是就有了 2、4、6、8 个唤醒等待位,但是并没有从根本上解决问题。
信号量
信号量是 E.W.Dijkstra 在 1965 年提出的一种方法,它使用一个整形变量来累计唤醒次数,以供之后使用。在他的观点中,有一个新的变量类型称作 信号量(semaphore)
。一个信号量的取值可以是 0 ,或任意正数。0 表示的是不需要任何唤醒,任意的正数表示的就是唤醒次数。
Dijkstra 提出了信号量有两个操作,现在通常使用 down
和 up
(分别可以用 sleep 和 wakeup 来表示)。down 这个指令的操作会检查值是否大于 0 。如果大于 0 ,则将其值减 1 ;若该值为 0 ,则进程将睡眠,而且此时 down 操作将会继续执行。检查数值、修改变量值以及可能发生的睡眠操作均为一个单一的、不可分割的 原子操作(atomic action)
完成。这会保证一旦信号量操作开始,没有其他的进程能够访问信号量,直到操作完成或者阻塞。这种原子性对于解决同步问题和避免竞争绝对必不可少。
“原子性操作指的是在计算机科学的许多其他领域中,一组相关操作全部执行而没有中断或根本不执行。
up 操作会使信号量的值 + 1。如果一个或者多个进程在信号量上睡眠,无法完成一个先前的 down 操作,则由系统选择其中一个并允许该程完成 down 操作。因此,对一个进程在其上睡眠的信号量执行一次 up 操作之后,该信号量的值仍然是 0 ,但在其上睡眠的进程却少了一个。信号量的值增 1 和唤醒一个进程同样也是不可分割的。不会有某个进程因执行 up 而阻塞,正如在前面的模型中不会有进程因执行 wakeup 而阻塞是一样的道理。
用信号量解决生产者 - 消费者问题
用信号量解决丢失的 wakeup 问题,代码如下
#define N 100 /* 定义缓冲区槽的数量 */ typedef int semaphore; /* 信号量是一种特殊的 int */ semaphore mutex = 1; /* 控制关键区域的访问 */ semaphore empty = N; /* 统计 buffer 空槽的数量 */ semaphore full = 0; /* 统计 buffer 满槽的数量 */ void producer(void){ int item; while(TRUE){ /* TRUE 的常量是 1 */ item = producer_item(); /* 产生放在缓冲区的一些数据 */ down(&empty); /* 将空槽数量减 1 */ down(&mutex); /* 进入关键区域 */ insert_item(item); /* 把数据放入缓冲区中 */ up(&mutex); /* 离开临界区 */ up(&full); /* 将 buffer 满槽数量 + 1 */ } } void consumer(void){ int item; while(TRUE){ /* 无限循环 */ down(&full); /* 缓存区满槽数量 - 1 */ down(&mutex); /* 进入缓冲区 */ item = remove_item(); /* 从缓冲区取出数据 */ up(&mutex); /* 离开临界区 */ up(&empty); /* 将空槽数目 + 1 */ consume_item(item); /* 处理数据 */ } }
为了确保信号量能正确工作,最重要的是要采用一种不可分割的方式来实现它。通常是将 up 和 down 作为系统调用来实现。而且操作系统只需在执行以下操作时暂时屏蔽全部中断:检查信号量、更新、必要时使进程睡眠。由于这些操作仅需要非常少的指令,因此中断不会造成影响。如果使用多个 CPU,那么信号量应该被锁进行保护。使用 TSL 或者 XCHG 指令用来确保同一时刻只有一个 CPU 对信号量进行操作。
使用 TSL 或者 XCHG 来防止几个 CPU 同时访问一个信号量,与生产者或消费者使用忙等待来等待其他腾出或填充缓冲区是完全不一样的。前者的操作仅需要几个毫秒,而生产者或消费者可能需要任意长的时间。
上面这个解决方案使用了三种信号量:一个称为 full,用来记录充满的缓冲槽数目;一个称为 empty,记录空的缓冲槽数目;一个称为 mutex,用来确保生产者和消费者不会同时进入缓冲区。Full
被初始化为 0 ,empty 初始化为缓冲区中插槽数,mutex 初始化为 1。信号量初始化为 1 并且由两个或多个进程使用,以确保它们中同时只有一个可以进入关键区域的信号被称为 二进制信号量(binary semaphores)
。如果每个进程都在进入关键区域之前执行 down 操作,而在离开关键区域之后执行 up 操作,则可以确保相互互斥。
现在我们有了一个好的进程间原语的保证。然后我们再来看一下中断的顺序保证
- 硬件压入堆栈程序计数器等
- 硬件从中断向量装入新的程序计数器
- 汇编语言过程保存寄存器的值
- 汇编语言过程设置新的堆栈
- C 中断服务器运行(典型的读和缓存写入)
- 调度器决定下面哪个程序先运行
- C 过程返回至汇编代码
- 汇编语言过程开始运行新的当前进程
在使用信号量
的系统中,隐藏中断的自然方法是让每个 I/O 设备都配备一个信号量,该信号量最初设置为0。在 I/O 设备启动后,中断处理程序立刻对相关联的信号执行一个 down
操作,于是进程立即被阻塞。当中断进入时,中断处理程序随后对相关的信号量执行一个 up
操作,能够使已经阻止的进程恢复运行。在上面的中断处理步骤中,其中的第 5 步 C 中断服务器运行
就是中断处理程序在信号量上执行的一个 up 操作,所以在第 6 步中,操作系统能够执行设备驱动程序。当然,如果有几个进程已经处于就绪状态,调度程序可能会选择接下来运行一个更重要的进程,我们会在后面讨论调度的算法。
上面的代码实际上是通过两种不同的方式来使用信号量的,而这两种信号量之间的区别也是很重要的。mutex
信号量用于互斥。它用于确保任意时刻只有一个进程能够对缓冲区和相关变量进行读写。互斥是用于避免进程混乱所必须的一种操作。
另外一个信号量是关于同步(synchronization)
的。full
和 empty
信号量用于确保事件的发生或者不发生。在这个事例中,它们确保了缓冲区满时生产者停止运行;缓冲区为空时消费者停止运行。这两个信号量的使用与 mutex 不同。
互斥量
如果不需要信号量的计数能力时,可以使用信号量的一个简单版本,称为 mutex(互斥量)
。互斥量的优势就在于在一些共享资源和一段代码中保持互斥。由于互斥的实现既简单又有效,这使得互斥量在实现用户空间线程包时非常有用。
互斥量是一个处于两种状态之一的共享变量:解锁(unlocked)
和 加锁(locked)
。这样,只需要一个二进制位来表示它,不过一般情况下,通常会用一个 整形(integer)
来表示。0 表示解锁,其他所有的值表示加锁,比 1 大的值表示加锁的次数。
mutex 使用两个过程,当一个线程(或者进程)需要访问关键区域时,会调用 mutex_lock
进行加锁。如果互斥锁当前处于解锁状态(表示关键区域可用),则调用成功,并且调用线程可以自由进入关键区域。
另一方面,如果 mutex 互斥量已经锁定的话,调用线程会阻塞直到关键区域内的线程执行完毕并且调用了 mutex_unlock
。如果多个线程在 mutex 互斥量上阻塞,将随机选择一个线程并允许它获得锁。
由于 mutex 互斥量非常简单,所以只要有 TSL 或者是 XCHG 指令,就可以很容易地在用户空间实现它们。用于用户级线程包的 mutex_lock
和 mutex_unlock
代码如下,XCHG 的本质也一样。
mutex_lock: TSL REGISTER,MUTEX | 将互斥信号量复制到寄存器,并将互斥信号量置为1 CMP REGISTER,#0 | 互斥信号量是 0 吗? JZE ok | 如果互斥信号量为0,它被解锁,所以返回 CALL thread_yield | 互斥信号正在使用;调度其他线程 JMP mutex_lock | 再试一次 ok: RET | 返回调用者,进入临界区 mutex_unlcok: MOVE MUTEX,#0 | 将 mutex 置为 0 RET | 返回调用者
mutex_lock 的代码和上面 enter_region 的代码很相似,我们可以对比着看一下
上面代码最大的区别你看出来了吗?
- 根据上面我们对 TSL 的分析,我们知道,如果 TSL 判断没有进入临界区的进程会进行无限循环获取锁,而在 TSL 的处理中,如果 mutex 正在使用,那么就调度其他线程进行处理。所以上面最大的区别其实就是在判断 mutex/TSL 之后的处理。
- 在(用户)线程中,情况有所不同,因为没有时钟来停止运行时间过长的线程。结果是通过忙等待的方式来试图获得锁的线程将永远循环下去,决不会得到锁,因为这个运行的线程不会让其他线程运行从而释放锁,其他线程根本没有获得锁的机会。在后者获取锁失败时,它会调用
thread_yield
将 CPU 放弃给另外一个线程。结果就不会进行忙等待。在该线程下次运行时,它再一次对锁进行测试。
上面就是 enter_region 和 mutex_lock 的差别所在。由于 thread_yield 仅仅是一个用户空间的线程调度,所以它的运行非常快捷。这样,mutex_lock
和 mutex_unlock
都不需要任何内核调用。通过使用这些过程,用户线程完全可以实现在用户空间中的同步,这个过程仅仅需要少量的同步。
我们上面描述的互斥量其实是一套调用框架中的指令。从软件角度来说,总是需要更多的特性和同步原语。例如,有时线程包提供一个调用 mutex_trylock
,这个调用尝试获取锁或者返回错误码,但是不会进行加锁操作。这就给了调用线程一个灵活性,以决定下一步做什么,是使用替代方法还是等候下去。
Futexes
随着并行的增加,有效的同步(synchronization)
和锁定(locking)
对于性能来说是非常重要的。如果进程等待时间很短,那么自旋锁(Spin lock)
是非常有效;但是如果等待时间比较长,那么这会浪费 CPU 周期。如果进程很多,那么阻塞此进程,并仅当锁被释放的时候让内核解除阻塞是更有效的方式。不幸的是,这种方式也会导致另外的问题:它可以在进程竞争频繁的时候运行良好,但是在竞争不是很激烈的情况下内核切换的消耗会非常大,而且更困难的是,预测锁的竞争数量更不容易。
有一种有趣的解决方案是把两者的优点结合起来,提出一种新的思想,称为 futex
,或者是 快速用户空间互斥(fast user space mutex)
,是不是听起来很有意思?
futex 是 Linux
中的特性实现了基本的锁定(很像是互斥锁)而且避免了陷入内核中,因为内核的切换的开销非常大,这样做可以大大提高性能。futex 由两部分组成:内核服务和用户库。内核服务提供了了一个 等待队列(wait queue)
允许多个进程在锁上排队等待。除非内核明确的对他们解除阻塞,否则它们不会运行。
对于一个进程来说,把它放到等待队列需要昂贵的系统调用,这种方式应该被避免。在没有竞争的情况下,futex 可以直接在用户空间中工作。这些进程共享一个 32 位整数(integer)
作为公共锁变量。假设锁的初始化为 1,我们认为这时锁已经被释放了。线程通过执行原子性的操作减少并测试(decrement and test)
来抢占锁。decrement and set 是 Linux 中的原子功能,由包裹在 C 函数中的内联汇编组成,并在头文件中进行定义。下一步,线程会检查结果来查看锁是否已经被释放。如果锁现在不是锁定状态,那么刚好我们的线程可以成功抢占该锁。然而,如果锁被其他线程持有,抢占锁的线程不得不等待。在这种情况下,futex 库不会自旋
,但是会使用一个系统调用来把线程放在内核中的等待队列中。这样一来,切换到内核的开销已经是合情合理的了,因为线程可以在任何时候阻塞。当线程完成了锁的工作时,它会使用原子性的 增加并测试(increment and test)
释放锁,并检查结果以查看内核等待队列上是否仍阻止任何进程。如果有的话,它会通知内核可以对等待队列中的一个或多个进程解除阻塞。如果没有锁竞争,内核则不需要参与竞争。