1      内核中的并发与竞态

1.1   并发与竞态概述

当在同一时间段出现两个或更多进程并且这些进程彼此交互时，就存在并发现象，此时必须采取互斥方法。那么什么是内核并发的场景呢？如何防止内核并发？有那些同步方法？以及这些方法的行为有何特点和如何使用它们？

对并发的管理是操作系统编程中核心的问题之一。 并发产生竞态，竞态导致共享数据的非法访问。因为竞态是一种极端低可能性的事件，因此程序员往往会忽视竞态。但是在计算机世界中，百万分之一的事件可能每几秒就会发生，而其结果是灾难性的。

竞态通常是作为对资源的共享访问结果而产生的。在设计自己的驱动程序时，第一个要记住的规则是：只要可能，就应该避免资源的共享。若没有并发访问，就不会有竞态。这种思想的最明显的应用是避免使用全局变量。但是，资源的共享是不可避免的，如硬件资源本质上就是共享、指针传递等等。资源共享的硬性规则：

²      在单个执行线程之外共享硬件或软件资源的任何时候，因为另外一个线程可能产生对该资源的不一致观察，因此必须显示地管理对该资源的访问。访问管理的常见技术成为“锁定”或者“互斥”：确保一次只有一个执行线程可操作共享资源。

²      当内核代码创建了一个可能和其他内核部分共享的对象时，该对象必须在还有其他组件引用自己时保持存在（并正确工作）。对象尚不能正确工作时，不能将其对内核可用。

1.2   内核并发原因

1)       单处理机器

Ø       2.6版本内核之前

对于单处理机器来说情况相对简单一些。在2.6版本内核之前，Linux内核是非抢占式的——在内核任务没有执行完之前不能被打断，这样的话，内核中程序并发执行的情况很少，准确地讲只有两种可能：

中断发生，因为中断执行是异步的，而且中断是在非抢占式内核中打断当前运行内核代码的唯一方法，所以中断显然是可以和其它内核代码并发执行的。因此如果中断操作和被中断的那内核代码都访问同样的内核数据，那么就会发生竞争。

睡眠和再调度, 处于进程上下文的内核任务可以睡眠（睡眠意味放弃处理器），这时调度程序会调度其它程序去执行（首先执行调度任务队列中的内核任务，然后执行软中断等，最后从运行队列中选择一个高优先级的用户进程运行）。显然这里也会造成内核并发访问，当睡眠的内核任务和新投入运行的内核任务访问同一共享数据时，就发生了竞争。

Ø       2.6版本及以后的内核

2.6版本的内核变成了抢占式内核——内核可能在任何时刻抢占正在运行的内核代码。所以内核中发生并发执行的情况大大增加了。内核抢占成为了内核程序并发的又一种可能，所以在开发抢占式内核代码时需要时刻警惕抢占产生的竞争。

2)       多处理器SMP

单处理器上的并发是逻辑上的伪并发，事实上所谓并发的内核程序其实是交错地占用处理器。真正的并发执行程序，必须依靠对称多处理器。但无论是逻辑上的并发还是真正的并发，都会产生竞争，而且它们的处理也是相同的。但是对于对称多处理器来说，由于两个或多个处理器可以在同一时刻执行代码，所以会不可避免地给内核带来并发可能，如果分别在不同处理器上执行的内核代码同时访问同一共享数据，竞争就产生了。因此，不用说对称多处理是内核并发的又一种可能。

可以看到随着Linux内核不断演化，在内核对系统支持更加全面，对任务调度更加高效的同时，也给内核带来了更多的并发可能，更容易引起竞争。上面提到的各种并发情况在内核中都必须得到有效的处理，才能确保内核有高稳定性。

无论是中断产生的并发或是睡眠引起的并发，还是内核抢占引起的并发，要想在内核开发中很好地避免，就必须从本质上了解它们的并发原因。只有在掌握内核任务的调度机制后，才可以真正的达到对并发可能的预测，进而能够采取合适的同步方法——锁——来避免并发。

1.3   内核中的任务调度类型

我们这里所说的任务调度不同于常说的进程调度。进程调度是：内核中的调度程序在进程运行队列中选择合适的（优先级高的）进程执行。而我们所说的内核任务调度指的是，内核中的任务获得执行机会。对于内核并发来说，内核任务之间的关系尤为重要。首先我们来看看内核有那些任务，各有什么特点。

1)       硬中断操作

硬中断是指那些由处理器以外的外设产生的中断，这些中断被处理器接收后交给内核中的中断处理程序处理。要注意的是：第一，硬中断是异步产生的，中断发生后立刻得到处理，也就是说中断操作可以抢占内核中正在运行的代码。这点非常重要。第二，中断操作是发生在中断上下文中的（所谓中断上下文指的是和任何进程无关的上下文环境）。中断上下文中，不可以使用进程相关的资源，也不能够进行调度。

2)       软中断操作

软中断是Linux中为了执行一些硬中断操作来不及完成的任务而采取的推后执行机制。因为硬中断操作期间的中断会被抛弃，所以硬中断是在不安全时间运行的。不安全时间应该尽量短，所以采用软中断来执行大部分任务，它会把硬中断做不完的耗时任务推后到安全时间执行（软中断期间不会丢弃中断信号）。

软中断不象硬中断那样时随时都能够被执行，笼统来讲软中断会在内核处理任务处理完毕后返回用户级程序前得到处理机会。具体的讲有三个时刻它将被执行(do_softirq())：硬件中断操作完成后（包括时钟tick中断）；内核schedule调度程序中；系统调用返回时。需要说明的是软中断的执行也处于中断上下文中，所以中断上下文对它的限制是和硬中断一样的。

3)       系统调用

系统调用是用户程序通过门机制来进入内核执行的内核例程，它运行在内核态，处于进程上下文中（进程上下文包括进程的堆栈等等环境），所以系统调用代码可以对进程相关数据进行访问，可以执行调度程序，也可以睡眠。

1.4   内核任务之间并发关系

上述内核任务很多情况是可以交错执行的，所以很有可能产生竞争。下面分析这些内核任务之间有那些可能的并发行为。

可以抽象出，程序（用户态和内核态一样）并发执行的总原因无非是正在运行中的程序被其它程序抢占，所以我们必须看看内核任务之间的抢占关系：

²      中断处理程序可以抢占内核中的所有程序（当没有锁保护时），包括软中断，tasklet，bottom half和系统的调用，甚至也包括中断处理程序（Linux内核是否允许中断嵌套？？）。也就是说中断处理程序可以和这些所有的内核任务并发执行，如果被抢占的程序和中断处理程序都要访问同一个资源，就产生了竞争。

²      软件中断可以抢占硬中断处理程序以外的内核程序，所以内核代码（比如，系统调用）中有数据和软中断共享，就有会有竞争。此外要注意的是，软中断即使是同种类型的也可以并发的运行在不同处理器上，所以它们之间共享数据都会产生竞争。（如果在同一个处理器上软中断是不能相互抢占的，因为软中断此时是顺序执行的，其是原子操作，不可中断）。

²      系统调用这种内核代码可能和各种内核代码并发，除了上面提到的中断（软，硬）抢占它产生并发外，它是有可能自发性地主动睡眠（比如在一些阻塞性的操作中），放弃处理器，重新调度其它任务，所以系统调用中并发情况更普遍，尤其当用户空间需要和内核空间共同操作全局数据时，一定要注意保护。

1.5   临界段

临界段是为解决竞态条件问题而产生的。一个临界段是一段不允许多路访问的受保护的代码。这段代码可以操纵共享数据或共享服务（例如硬件外围设备）。临界段操作时坚持互斥锁（mutual exclusion）原则（当一个线程处于临界段中时，其他所有线程都不能进入临界段）。

临界段中需要解决的一个问题是死锁条件。考虑两个独立的临界段，各自保护不同的资源。每个资源拥有一个锁，在本例中称为 A 和 B。假设有两个线程需要访问这些资源，线程 X 获取了锁 A，线程 Y 获取了锁 B。当这些锁都被持有时，每个线程都试图占有其他线程当前持有的锁（线程 X 想要锁 B，线程 Y 想要锁 A）。这时候线程就被死锁了，因为它们都持有一个锁而且还想要其他锁。一个简单的解决方案就是总是按相同次序获取锁，从而使其中一个线程得以完成。

1.6   内核中的锁定机制

Linux使用的锁定机制可以说从2.0到2.6以来不断发展完善，从最初的原子操作，到后来的信号量，从大内核锁到今天的自旋锁。这些机制的发展伴随Linux从单处理器到对称多处理器的过度；伴随着从非抢占内核到抢占内核的过度。锁机制越来越有效，也越来越复杂。

Linux 性能非凡，其锁定方法各有特色。原子操作多用来做计数使用，原子锁不仅提供了一种锁定机制，同时也提供了算术或 bitwise 操作。自旋锁提供了一种锁定机制（主要应用于 SMP）。信号量在提供锁机制的情况下对系统影响较小。最后，针对信号量改进的互斥锁是一种新的锁定机制，提供了一种构建在原子之上的简单 API。还有经常用于免锁算法的生产者/消费者任务的数据结构之一循环缓冲区，它在设备驱动程序中相当普遍。不管你需要什么，Linux 都会提供一种锁定方案保护您的数据。

要用好Linux锁定机制必须深刻理解内核调度原理，要能清楚区分并发原因。另外Linux内核中锁机制的使用非常普遍，理解和使用锁机制是代码能够可靠运行的关键问题之一，并发产生的错误不可再现，调试困难，往往给开发带来很大麻烦，所以使用锁来保护临界区尤为重要。

即使开发环境不存在多处理器，不要求内核抢占，也最好不要放弃使用锁机制，因为使用恰当的锁机制可以方便将开发的代码向其它环境移植。另外要记住，不要指望在代码编写完毕后，再加入锁来保护资源，一定要在设计初期就要考虑竞争问题，设计锁来保护临界区。

2      原子操作

2.1   原子概述

Linux 中最简单的互斥方法就是原子操作。原子意味着临界段被包含在 API 函数中。不需要额外的锁定，因为 API 函数已经包含了锁定。完整的锁机制对一个简单的整数来讲显得浪费。由于 C 不能实现原子操作，因此 Linux 依靠底层架构来提供这项功能。各种底层架构存在很大差异，因此原子函数的实现方法也各不相同。一些方法完全通过汇编语言来实现，而另一些方法依靠 c 语言并且使用 local_irq_save 和 local_irq_restore 禁用中断。

当需要保护的数据非常简单时，例如一个计数器，原子运算符是种理想的方法。尽管原理简单，原子 API 提供了许多针对不同情形的运算符。

2.2   原子API

要声明一个原子变量（atomic variable），首先声明一个 atomic_t 类型的变量。这个结构包含了单个 int 元素。原子变量操作是非常快的，因为它们在任何可能时编译成一条单个机器指令。原子变量使用 ATOMIC_INIT 符号常量在编译时进行初始化，也可以使用 atomic_set在运行时初始化。

atomic_t my_counter = ATOMIC_INIT(0);

atomic_set( &my_counter, 0 );

原子 API 支持一个涵盖许多用例的函数集。可以使用 atomic_read 读取原子变量中的内容，也可以使用 atomic_add 为一个变量添加指定值。最常用的操作是使用 atomic_inc 使变量递增。也可用减号运算符，它的作用与相加和递增操作相反。

int atomic_read(atomic_t *v); /*返回 v 的当前值.*/ void atomic_add(int i, atomic_t *v);/*由 v 指向的原子变量加 i. 返回值是 void*/ void atomic_sub(int i, atomic_t *v); /*从 *v 减去 i.*/ void atomic_inc(atomic_t *v); void atomic_dec(atomic_t *v); /*递增或递减一个原子变量.*/

该 API 也支持许多其他常用用例，包括 operate-and-test 例程。这些例程允许对原子变量进行操纵和测试（作为一个原子操作来执行）。

int atomic_inc_and_test(atomic_t *v); int atomic_dec_and_test(atomic_t *v); int atomic_sub_and_test(int i, atomic_t *v); /*进行一个特定的操作并且测试结果; 如果, 在操作后, 原子值是 0, 那么返回值是真; 否则, 它是假. 注意没有 atomic_add_and_test.*/ int atomic_add_negative(int i, atomic_t *v); /*加整数变量 i 到 v. 如果结果是负值返回值是真, 否则为假。这被内核中一些依赖于架构的信号量函数使用。*/

atomic_t 数据项必须通过这些函数存取。 如果你传递一个原子项给一个期望一个整数参数的函数， 你会得到一个编译错误。需要多个 atomic_t 变量的操作仍然需要某种其他种类的加锁。

2.3   位操作

内核提供了一套函数来原子地修改或测试单个位。原子位操作非常快, 因为它们使用单个机器指令来进行操作, 而在任何时候低层平台做的时候不用禁止中断。函数是体系依赖的并且在 <asm/bitops.h> 中声明。以下函数中的数据是体系依赖的。 nr 参数(描述要操作哪个位)在ARM体系中定义为unsigned int：

void set_bit(nr, void *addr); /*设置第 nr 位在 addr 指向的数据项中。*/ void clear_bit(nr, void *addr); /*清除指定位在 addr 处的无符号长型数据.*/

许多驱动程序使用这些原子操作，使用这些操作前，需要提供一个值和将要进行操作的位掩码。

void change_bit(nr, void *addr);/*翻转nr位.*/ test_bit(nr, void *addr); /*这个函数是唯一一个不需要是原子的位操作; 它简单地返回这个位的当前值.*/

/*以下原子操作如同前面列出的, 除了它们还返回这个位以前的值.*/

int test_and_set_bit(nr, void *addr); int test_and_clear_bit(nr, void *addr); int test_and_change_bit(nr, void *addr);

3      自旋锁spin_lock

3.1   自旋锁概述

自旋锁是专为防止对称多处理器SMP并发而引入的一种锁，它在内核中大量应用于中断处理等部分。在单处理器上，对于2.6的抢占式内核，自旋锁仅仅当作一个设置内核抢占的开关。如果2.4内核抢占也不存在，那么自旋锁会在编译时被完全剔除出内核，即实现为空。对于单处理器UP来说，防止中断处理中的并发可简单采用关闭中断的方式，不需要自旋锁。

自旋锁只有在内核可抢占或SMP的情况下才真正需要，在单CPU且不可抢占的内核下，自旋锁的所有操作都是空操作。但是因为您的代码最终可能会在 SMP 系统上运行，将它们添加到 UP 系统是个明智的做法。

假如运行在其中的时候该进程的时间片用完了的话，是该如何来处理呢？

在单处理机上，spin_lock()就退化成了Preemp_disable()，它就是禁止抢占，即虽然时间片用完了，但是仍然是不能切换到其它进程去的。因此每个进程有一个preemp_count这个变量，如果这个变量为0的话表示其可以被其它进程所抢占，如果大于0则不能够被抢占，而在调用preemp_disable()的时候就会将该进程的preemp_count值加1。使得其不能够被抢占。而继续运行该进程，直到调用preemp_enable()将其设为可抢占，由于时间一般不长，因此不会有影响。

自旋锁最多只能被一个内核任务持有，自旋锁不会引起调用者休眠。如果一个内核任务试图请求一个已被争用(已经被持有)的自旋锁，那么这个任务就会一直进行忙循环——旋转——等待锁重新可用。"自旋"一词就是因此而得名。要是锁未被争用，请求它的内核任务便能立刻得到它并且继续进行。自旋锁可以在任何时刻防止多于一个的内核任务同时进入临界区，因此这种锁可有效地避免多处理器上并发运行的内核任务竞争共享资源。  

事实上，自旋锁的初衷就是：在短时间内进行轻量级的锁定。自旋锁适合于保持时间非常短的情况，因为一个被争用的自旋锁使得请求它的线程在等待重新可用时自旋，特别浪费处理时间，这是自旋锁的要害之处，所以自旋锁不应该被长时间持有。忙等待看起来效率低下，但它实际上比将线程休眠然后当锁可用时将其唤醒要快得多。如果需要长时间锁定的话, 最好使用信号量。

在实际应用中自旋锁代码只有几行，而持有自旋锁的时间也一般不会超过两次上下文切换，因线程一旦要进行切换，就至少花费切出切入两次，自旋锁的占用时间如果远远长于两次上下文切换，我们就可以让线程睡眠，这就失去了设计自旋锁的意义。

自旋锁是一个互斥设备，他只能会两个值：“锁定”和“解锁”。使用自旋锁的核心规则：

²      持有锁时，对应处理器的内核抢占被禁止；

²      任何拥有自旋锁的代码都必须是原子的，不能放弃CPU（如休眠，休眠可发生在许多无法预期的地方）。否则CPU将有可能永远自旋下去（死机）。因为Linux内核实现的自旋锁是不可递归的，如果你想得到一个你正持有的锁，你必须自旋，等待你自己释放这个锁，但是你处于自旋忙等待中，所以永远没有机会释放锁，于是你就被自己锁死了！

²      如果某个获得锁的函数要调用其他同样试图获取这个锁的函数，代码就会锁死。（不允许锁的拥有者第二次获得同个锁。）

²      自旋锁可用于硬中断及软中断中。因此与中断共享数据时，获得自旋锁前要关闭本地中断，否则中断服务程序可能访问持有内核锁的相关代码，这时中断会自旋，但在中断返回前，原有持有锁的代码无法运行去释放锁。

²      拥有自旋锁的时间越短越好。自旋锁被占有时会导致其它处理器上的空等待，因此应越短越好。

²      在必须获取多个锁时，应始终以相同顺序获取，这样可以防止互锁。

²      若我们拥有信号量和自旋锁的组合，必须先获得信号量。

3.2   自旋锁的API

自旋锁原语所需包含的文件是<linux/spinlock.h> ,以下是自旋锁的内核API:

创建和初始化自旋锁 

#define SPIN_LOCK_UNLOCKED (spinlock_t) { 1 }

spinlock_t my_lock = SPIN_LOCK_UNLOCKED;/* 编译时静态初始化spinlock*/

DEFINE_SPINLOCK( my_spinlock ); void spin_lock_init(spinlock_t *lock);/* 运行时初始化指定的spinlock*/

获得锁和解锁

定义了自旋锁之后，就可以使用大量的锁定函数了。每个函数用于不同的上下文。

void spin_lock(spinlock_t *lock);/* 获得spinlock*/

void spin_unlock(spinlock_t *lock);

这是一个最简单的函数，它不会执行中断禁用，但是包含全部的内存壁垒（memory barrier）。这个变量假定中断处理程序和该锁之间没有交互。

void spin_lock_irqsave(spinlock_t *lock, unsigned long flags);

函数获得自旋锁，禁止内核抢占并且在本地处理器（在 SMP 情形中）上禁用中断，一切其他活动停止。保存中断标志于flags。

自旋锁可以用在中断处理程序中。若与中断共享受保护的资源，则在使用时一定要在获取锁之前，首先禁止本地中断（当前处理器上的中断），否则中断处理程序就可能打断正持有锁的内核代码，有可能会试图争用这个已经被持有的自旋锁。这样一来，中断处理程序就会自旋，等待该锁重新可用，但是锁的持有者在这个中断处理程序执行完毕之前不可能运行，这就会造成双重请求死锁。

void spin_unlock_irqrestore(spinlock_t *lock, unsigned long flags);

spin_unlock_irqrestore 函数释放自旋锁，并且（通过 flags 参数）恢复原始中断设置情况。这类函数有最高的互斥性，但要注意不能在获得锁的情况下休眠。适用范围最广。

void spin_lock_irq(spinlock_t *lock);

获得spinlock,禁止本地cpu中断

void spin_unlock_irq(spinlock_t *lock);

是spin_lock_irqsave/spin_unlock_irqrestore 的一个不太安全的变体，因为它会假设spin_unlock_irq后中断状态是打开的。

void spin_lock_bh(spinlock_t *lock)

获得spinlock，禁止软件中断，保持硬件中断打开

void spin_unlock_bh(spinlock_t *lock);

bottom half 方法可以将设备驱动程序中的工作延迟到中断处理后执行。由于下半部（中断程序下半部）可以抢占进程上下文中的代码，所以当下半部和进程上下文共享数据时，必须对进程上下文中的共享数据进行保护，所以需要加锁的同时还要禁止下半部执行。这种自旋锁禁用了本地 CPU 上的软中断。这可以阻止 softirq、tasklet 和 bottom half 在本地 CPU 上运行。同样，由于中断处理程序可以抢占下半部，所以如果中断处理程序和下半部共享数据，那么就必须在获取恰当的锁的同时还要禁止中断。

int spin_trylock(spinlock_t *lock); int spin_trylock_bh(spinlock_t *lock);

该宏尽力获得自旋锁lock，如果能立即获得锁，它获得锁并返回真，否则不能立即获得锁，立即返回假。它不会自旋等待lock被释放。

spin_trylock_irqsave(lock, flags)

该宏如果获得自旋锁lock，它也将保存标志寄存器的值到变量flags中，并且禁止本地中断，如果没有获得锁，它什么也不做。 因此如果能够立即获得锁，它等同于spin_lock_irqsave，如果不能获得锁，它等同于spin_trylock。如果该宏获得自旋锁lock，那需要使用spin_unlock_irqrestore来释放。

spin_is_locked(x)

该宏用于判断自旋锁x是否已经被某执行单元保持（即被锁），如果是，返回真，否则返回假。

spin_can_lock(lock)

该宏用于判断自旋锁lock是否能够被锁，它实际是spin_is_locked取反。如果lock没有被锁，它返回真，否则，返回假。

spin_unlock_wait(x)

该宏用于等待自旋锁x变得没有被任何执行单元保持，如果没有任何执行单元保持该自旋锁，该宏立即返回，否则将循环在那里，直到该自旋锁被保持者释放。

3.3   自旋锁的实现细节

在SMP中，spinlock_t定义如下：

typedef struct {

volatile unsigned int lock;

} spinlock_t;

自旋锁的实现和体系结构密切相关，代码一般通过汇编实现。

static inline void spin_lock(spinlock_t *lock)

{

__asm__ __volatile__(

"/n1:/t"

"lock ; decb %0/n/t"

"js 2f /n" //lock->lock< 0 ,jmp 2 forward

".section .text.lock,/"ax/"/n"

"2:/t"

"cmpb $0,%0/n/t" //wait lock->lock==1

"rep;nop/n/t"

"jle 2b/n/t"

"jmp 1b/n"

".previous"

:"=m" (lock->lock) : : "memory");

}

static inline void spin_unlock(spinlock_t *lock)

{

__asm__ __volatile__(

"movb $1,%0"

:"=m" (lock->lock) : : "memory"); //lock->lock=1

}

#define local_irq_save(x) __asm__ __volatile__("pushfl ; popl %0 ; 

cli":

"=g" (x): /* no input */ :"memory")

#define local_irq_restore(x) __asm__ __volatile__("pushl %0 ; popfl": /* 

no 

output */ :"g" (x):"memory")

#define local_irq_disable() __asm__ __volatile__("cli": : :"memory")

#define local_irq_enable() __asm__ __volatile__("sti": : :"memory")

#define cpu_bh_disable(cpu) do { local_bh_count(cpu)++; barrier(); } while (0)

#define cpu_bh_enable(cpu) do { barrier(); local_bh_count(cpu)--; } while (0)

#define local_bh_disable() cpu_bh_disable(smp_processor_id())

#define local_bh_enable() cpu_bh_enable(smp_processor_id())

对于UP来说，上面已经是足够了，不过对于SMP来说，还要防止来自其它cpu的影响，这时解决的方法就可以把上面的spin_lock机制也综合进来了。

#define spin_lock_irqsave(lock, flags) do { 

local_irq_save(flags);

spin_lock(lock); } while (0)

#define spin_lock_irq(lock) do {

local_irq_disable(); 

spin_lock(lock); } while (0)

#define spin_lock_bh(lock) do {

local_bh_disable(); 

spin_lock(lock); } while (0)

#define spin_unlock_irqrestore(lock, flags) do {

spin_unlock(lock);

local_irq_restore(flags); } while (0)

#define spin_unlock_irq(lock) do {

spin_unlock(lock); 

local_irq_enable();

} while (0)

#define spin_unlock_bh(lock) do {

spin_unlock(lock); 

local_bh_enable(); 

} while (0)

4      信号量Semaphore

4.1   信号量概述

一个信号量（semaphore: 旗语，信号灯）本质上是一个整数值，它和一对函数联合使用，这一对函数通常称为P和V。希望进入临界区的进程将在相关信号量上调用P；如果信号量的值大于零，则该值会减小一，而进程可以继续。相反，如果信号量的值为零（或更小），进程必须等待知道其他人释放该信号。对信号量的解锁通过调用V完成；该函数增加信号量的值，并在必要时唤醒等待的进程。

当信号量用于互斥时（即避免多个进程同时在一个临界区运行），信号量的值应初始化为1。这种信号量在任何给定时刻只能由单个进程或线程拥有。在这种使用模式下，一个信号量有时也称为一个“互斥体（mutex）”,它是互斥（mutual exclusion）的简称。Linux内核中几乎所有的信号量均用于互斥。

对于互斥模式的信号量，如果有一个任务试图获得一个已被持有的信号量时，信号量会将其推入等待队列（排他性等待），然后让其睡眠。这时处理器获得自由去执行其它代码。当持有信号量的进程将信号量释放后，在等待队列中的第一个任务将被唤醒，从而便可以获得这个信号量。信号量的睡眠特性，使得信号量适用于锁会被长时间持有的情况；只能在进程上下文中使用，因为中断上下文中是不能被调度的；

4.2   信号量和自旋锁的区别

其实上面介绍的几种同步和互斥机制，其底层源码都是使用自旋锁。因此信号量可以理解为自旋锁的再包装。所以从这里就可以理解为什么自旋锁通常可以提供比信号量更高的互斥性能。

信号量和自旋锁的适用场合与区别如下：

²      自旋锁保持期间是抢占失效的，即未获得锁之前不会进行进程切换，而信号量保持期间是可以被抢占的。这意味者信号量不会对影响调度反应时间带来负面影响。如果持有自旋锁的时间过长，那么内核针对任务调度的各种改进将失去意义；

²      信号量适合于保持时间较长的情况，当信号量不可用时会导致调用者睡眠，进而发生进程切换，因此只能在进程上下文使用（_trylock的变种能够在中断上下文使用）；

²      如果被保护的共享资源只在进程上下文访问，使用信号量保护该共享资源非常合适；

²      如果在进程上下文中对共享资源的访问时间非常短，优先使用自旋锁，系统整体开销小；

²      如果被保护的共享资源需要在中断上下文访问（包括底半部即中断处理和顶半部即软中断），就只能使用自旋锁。

4.3   信号量的API

使用信号量，内核代码必须包含<asm/semaphore.h> 。

以下是信号量初始化的方法：

/*通用的初始化函数*/ void sem_init(struct semaphore *sem, int val);

由于信号量通常被用于互斥模式。所以以下是内核提供的一组辅助函数和宏：

/*方法一、声明+初始化宏*/ DECLARE_MUTEX(name); DECLARE_MUTEX_LOCKED(name); /*方法二、初始化函数*/ void init_MUTEX(struct semaphore *sem); void init_MUTEX_LOCKED(struct semaphore *sem); 带有“_LOCKED”的是将信号量初始化为0，即锁定，允许任何线程访问时必须先解锁。没带的为1。

P函数为：

void down(struct semaphore *sem);

不推荐使用，会建立不可杀进程 int down_interruptible(struct semaphore *sem);

推荐使用，使用down_interruptible需要格外小心，若操作被中断，该函数会返回非零值，而调用这不会拥有该信号量。对down_interruptible的正确使用需要始终检查返回值，并做出相应的响应。

int down_trylock(struct semaphore *sem);

带有“_trylock”的永不休眠，若信号量在调用是不可获得，会返回非零值。可用于中断上下文

V函数为：

void up(struct semaphore *sem);

任何拿到信号量的线程都必须通过有且仅有一次对up的调用而释放该信号量。在出错时，要特别小心；若在拥有一个信号量时发生错误，必须在将错误状态返回前释放信号量。

4.4   信号量的实现细节

struct semaphore {

        atomic_t count;

        int sleepers;

        wait_queue_head_t wait;

};

Count该信号量表示的可用资源数目，其必须支持原子操作。内核用atomic_t的数据结构和在它上面的一系列操作如atomic_add()、atomic_sub()等等实现；

Sleepers，睡眠在该信号量上的进程数目，等于等待队列中的表项数；

Wait内嵌的等待队列头结构，信号量本质上是一种资源，资源不可用时，睡眠的进程通过等待队列和该信号量相关联，以便资源可用时从等待队列中唤醒相应进程。

信号量本身的互斥操作是由wait中的互斥锁提供的。

功能:  获取信号灯进行临界资源操作，如果获得失败则睡眠，睡眠为深度睡眠，不可中断

fastcall void__sched  __down(struct semaphore * sem)

{

     struct task_struct *tsk = current;

     DECLARE_WAITQUEUE(wait，tsk);

//这个wait为等待结点，其和当前进程关联，若资源不可用则将该进程添加到等待队列中

unsigned long flags;

     tsk->state = TASK_UNINTERRUPTIBLE; //将当前进程置为不可中断状态

     spin_lock_irqsave(&sem->wait.lock，flags);

     add_wait_queue_exclusive_locked(&sem->wait，&wait);

//将进程添加到等待队列中， exclusive的，唤醒时只能唤醒一个，从队列头部开始

     sem->sleepers++;

       for (;;) {

          int sleepers = sem->sleepers;

         if(!atomic_add_negative(sleepers – 1，&sem->count)) {

              sem->sleepers = 0;

              break;

           }

           sem->sleepers = 1;

// 休眠前先解开锁，否则无人可以唤醒之

           spin_unlock_irqrestore(&sem->waitlock，flags);

//由于无可用资源，当前进程对应的wait_queue_t{}结点插入该sem信号等待队列中，然后通过schedule()让出cpu，让cpu去调度执行其他进程，而当前进程睡眠去了

         schedule();

// 当schedule返回时，说明当前进程被唤醒了

            spin_lock_irqsave(&sem->wait.lock，flags);

           tsk-<state = TASK_UNINTERRUPTIBLE;

//回到for，一方面对于smp来说再次检查资源是否可用，另一方面和未休眠的处理保持一致性。

           }

// 当前资源可用，将进程从等待队列中移除

           remove_wait_queue_locked(&sem->wait，&wait);

           wake_up_locked(&sem->wait);

           spin_unlock_irqrestore(&sem->wait.lock，flags);  

           tsk->state = TASK_RUNNING;

当该临界资源一直未被释放时，所有等待该临界资源的进程均在等待队列中sleep，不断的用schedule()让出cpu调度其他进程运行直至sem->count增加后至少变为0或正数后，当前进程直接从队头出来进入执行，同时还唤醒了等待队列中第一个等待进程，不过由于sem->sleepers被赋为0，sem->count= =0，此进程刚唤醒，又要去sleep，除非sem->count又增加了，此唤醒进程才能执行

  针对上面示意图分析:  atomic_add_negative(sleepers-1，&sem->count)  

__down( )函数流程分析：

__down()函数架构分析：

当前信号量上睡眠着三个进程。

  count    sleepers  sleepers-1   atomic_add_negative()

     -3        3       2           -1

相应的up()函数为:

void up(struct semaphore*sem)

{

sem->count++; //为原子操作

if(sem->count<=0)

{

//唤醒等待队列中的一个符合条件的进程（因为每个进程都加了TASK_EXCLUSIVE标志） 。

};

当进程A又获得机会运行时，它先执行wake_up(&sem->wait)操作，唤醒等待队列里的一个进程（只是将其状态更改为TASK_RUNNING，并不一定立即得到运行。即使运行也可能会再次休眠），接着它进入临界区，从临界区出来时执行up()操作，使sem->count++,（如果进程A是从down()中直接返回，因为这时等待队列一定为空，所以它不用执行wake_up()操作，直接进入临界区，在从临界区出来时一样执行up()操作，使 sem->count++）。这时如果count的值小于等于0，这表明在它在临界区期间又有一个进程（可能就是它进入临界区时唤醒的那个进程）进入睡眠了，则执行wake_up()操作，反之，如果count的值已经大于0，这表明在它在临界区期间没有别的进程（包括在它进入临界区时被它唤醒过的那个进程）进入睡眠，那么它就可以直接返回了。

从被唤醒的那个进程来看，如果在唤醒它的进程没执行up()之前它就得到了运行机会，这时它又重新计算count=sleepers - 1 + count=-1；从而sleepers被赋值1；这时它又必须进行调度让出运行的机会给别的进程，自己去睡眠。这正是发生在唤醒它的进程在临界区时运行的时候。如果是在唤醒它的进程执行了up()操作后它才得到了运行机会，而且在唤醒它的进程在临界区期间时没别的进程执行down()，则count的值在进程执行up()之前依然为0，这时在up()里面就不必要再执行wake_up()函数了。

可以通过一个例子来说明具体的实现。设开始sem->count=sem->sleepers=0。也就是有锁但无等待队列 （一个进程已经在运行中）。先后分别进行3个down()操作，和3个up()操作，如下：为了阐述方便，只保留了一些会改变sleepers和count值的步骤，并且遵循从左到右依次进行的原则。

down1:

count(0->-1),sleepers(0->1),sleepers-1+count(-1),count(-1),sleepers(1),调度

down2:

count(-1->-2),sleepers(1->2),sleepers-1+count(-1),count(-1),sleepers(1),调度

down3:

count(-1->-2),sleepers(1->2),sleepers-1+count(-1),count(-1),sleepers(1),调度

up1:

count(-1->0),唤醒一个睡眠进程（设为1）,（进程1得到机会运行）

sleepers-1+count(0),count(0),sleepers(0),break,

唤醒另一个睡眠进程（设为2），（进程2得到机会运行）

sleepers-1+count(-1),count(-1),sleepers(1),

调度（没达到条件，又得睡觉）也可能是这样的：

up1`:

count(-1->0),唤醒一个睡眠进程（设为1）,（进程1得到机会运行）

sleepers-1+count(0),count(0),sleepers(0),break,

唤醒另一个睡眠进程（设为2），进程2在以后才得到机会运行）

up2:

count(-1->0),（因为count<=0）唤醒一个睡眠进程（设为2）,进程2得到机会运行）

sleepers-+count(0) , count(0) , sleepers(0) ,break,

唤醒另一个睡眠进程（设为3），进程3得到机会运行）

sleepers-1+count(-1),count(-1),sleepers(1),

调度（没达到条件，又得睡觉）对应上面的1`：

up2`:

count(0->1),（因为count>0,所以直接返回）进程2得到机会运行）

sleepers-1+count(0),count(0),sleepers(0),break,

唤醒另一个睡眠进程，（设为3）

up3:

count(-1->0),（因为count<=0）唤醒一个睡眠进程（设为3）,进程3得到机会运行）

sleepers-1+count(0),count(0),sleepers(0),break,

唤醒另一个睡眠进程（这时队列里没进程了）

进程3运行结束，执行up(), 使count =1 ,这时变成没锁状态 ）

对应上边的2`：

up3`:

count(0->1),（因为count>0,所以直接返回）进程3得到机会运行）

sleepers-1+count(0),count(0),sleepers(0),break,

唤醒另一个睡眠进程（这时队列里没进程了）

进程3运行结束，执行up(), 使count =1 ,这时变成没锁状态 ）

当然，还有另一种情况，就是up()操作和down()操作是交*出现的，一般的规律就是，如果进程在临界区期间又有进程（无论是哪个进程，新来的还是刚被唤醒的那个）进入睡眠，就会令count的值从0变为-1，从而进程在从临界区出来执行up()里就必须执行一次wake_up()，以确保所有的进程都能被唤醒，因为多唤醒几个是没关系的。如果进程在临界区期间没有别的进程进入睡眠，则从临界区出来执行up()时就不会去执行wake_up()了。

而为什么要把wake_up()和count++分开呢，可以从上面的up1看出来，例如，进程2第一次得到机会运行时，本来这时唤醒它的进程还没执行up()的，但有可能其它进程执行了up()了，所以真有可能会发现count==1的情况，这时它就真的不用睡觉了，令count=sleepers=0，就可以接着往下执行了。还可看出一点，一般的，( count ,sleepers)的值的取值范围为(n ,0)[n>0] 和(0 ,0)和 (1 ,-1)。

fastcall int__sched  __down_interruptible(struct semaphore * sem)

{ 

功能: :获得信号灯，成功获得进入临界资源运行，获得失败，则睡眠，但为浅度睡眠，可被唤醒中断，但需一个返回值，指明是否获得信号灯，新建 wait这个wait_queue_t{}等待队列结点，并初始化之，private指针指向当前进程

  int retval = 0;                        

struct task_struct *tsk=current;      

DECLARE_WAITQUEUE(wait，tsk);    

  unsigned long flags;

 tsk->state = TASK_INTERRUPTIBLE;

 spin_lock_irqsave(&sem->wait.lock，flags);

 add_wait_queue_exclusive_locked(&sem->wait，&wait);

  sem->sleepers++;                       

  for(;;) {                                              

int sleepers = sem->sleepers;          　

if(signal_pending(current)){ //当前进程从浅度睡眠中被中断唤醒出循环．

  retval = _EINTR;

  sem->sleepers = 0;

  atomic_add(sleepers，&sem->count);　　

  break;　

}　　　　　　　　　　　　　

 　　　　　 if(!atomic_add_negative(sleepers – 1，&sem->count)) {

    　　　 sem->sleepers = 0;     　　

  　　　　 break;　　//当sem->count> = 1时资源空闲，跳出循环．　　　　

 　　　　 }　　

// sem->count<=0无空闲资源　　　　　　　　　　　    　　

           sem->sleepers = 1; /*us-see-1 above */

  　　　　 spin_unlock_irqrestore(&sem->wait.lock，flags); //休眠前必须释放锁，否则可能死锁

     　　　schedule();　

   　　　  spin_lock_irqsave(&sem->wait.lock，flags);

   　　　  tsk->state = YASK_INTERRUPTIBLE;

 　　　　}

  remove_wait_queue_locked(&sem->wait，&wait);

   wake_up_locked(&sem->wait);

   spin_unlock_irqrestore(&sem->wait.lock，flags);

//在信号量sem的等待队列中，删除指向当前进程wait结点．唤醒该信号灯sem的等待队列中的第一个结点．释放该信号灯的等待队列首部的自旋锁，并恢复EFLAGS

   tsk->state = TASK_RUNNING;

   return retval; //0   获得信号灯

//-EINTR: 没有获得信号灯，中断返回，不再睡眠等待信号灯

  }                         

―――――――――――――――――――――――――――――――――――

  static inline int down_interruptible(struct semaphore * sem)

 {

int result;

might_sleep();

_asm_ volatile_(

 “#atomic interruptible down operation/n/t”

LOCK “decl %1/n/t”   /*- -sem->count */

“js 2f /n/t”

“xorl %0，%0/n”

    “1:/n”

    LOCK_SECTION_START(“”)

“2:/tlea %1，%%eax/n/t”

“call__down _interruptible/n/t”

“gmp lb/n”

    LOCK_SECTION_END

:“=a”(result)，“=m”(sem->count)

:

:“memory”);

return result;

-------------------------------------------------------------------------------------------------------

down_interruptible()架构分析:

fastcall int  __down_trylock(struct semaphore * sem)

//若临界资源现在变为空闲，则唤醒等待队列中第一个结点，自己结束返回。若临界资源仍忙，则结束返回，不阻塞也不唤醒等待队列中的结点进程。

{

  int sleepers;

  unsigned long flags;

 spin_lock_irqsave(&sem->wait.lock，flags);

 sleepers = sem->sleepers + 1;

 sem->sleepers = 0；

  if(!atomic_add_negative(sleepers，&sem->count)){

     wake_up_locked(&sem->wait);

  }

     spin_unlock_irqrestore(&sem->wait.lock，flags);

     return 1;

}

    _ _ down_try_lock()      sem->sleepers

sem->count原始值    sem->count- -    sleepers原始值    sleepeers+ +   atomic_add_negative

                                                                 (sleepers，$sem->count.)

          3              2             0            1            3

          2              1             0            1            2 

          1              0             0            1            1          

          0             --1             0            1            0

         --1             --2             1            2            0

         --2             --3             2            3            0

         --3             --4             3            4            0

+

5      内核互斥锁mutex

5.1   互斥锁数据结构

在内核中可以使用互斥锁来实现信号量行为。内核互斥锁是在原子 API 之上实现的，但这对于内核用户是不可见的。同一时间只能有一个任务持有互斥锁，而且只有这个任务可以对互斥锁进行解锁。互斥锁不能进行递归锁定或解锁。但是互斥锁比当前的内核信号量选项更快，并且更加紧凑。互斥锁是二值信号量的特例，但在执行流上必信号量更有效。

/include/linux/mutex.h

struct mutex {

        /* 1: unlocked, 0: locked, negative: locked, possible waiters */

        atomic_t           count;

        spinlock_t         wait_lock;

        struct list_head wait_list;

};

spinlock_t         wait_lock;

struct list_head wait_list;

二者即构成了wait_queue_head_t结构，相比semaphere而言，mutex少了sleepers成员，因为count只能是0、1，sleepers失去了意义，只能唤醒处于wait_list中的第一个进程。

/*

 * This is the control structure for tasks blocked on mutex,

 * which resides on the blocked task's kernel stack:

 */

struct mutex_waiter {

        struct list_head list;

        struct task_struct      *task;

};

mutex_waiter相当于一个wait_queue_t，添加到wait_list中。

5.2   创建和初始化

# define mutex_init(mutex) /

do {                                                /

        static struct lock_class_key __key;          /

                                                      /

        __mutex_init((mutex), #mutex, &__key);         /

} while (0)

# define mutex_destroy(mutex)                               do { } while (0)

#define __MUTEX_INITIALIZER(lockname) /

                { .count = ATOMIC_INIT(1) /

                , .wait_lock = SPIN_LOCK_UNLOCKED /

                , .wait_list = LIST_HEAD_INIT(lockname.wait_list)

}

#define DEFINE_MUTEX(mutexname) /

        struct mutex mutexname = __MUTEX_INITIALIZER(mutexname)

extern void __mutex_init(struct mutex *lock, const char *name,

                        struct lock_class_key *key);

可以通过 DEFINE_MUTEX 宏使用一个操作创建和初始化互斥锁。这将创建一个新的互斥锁并初始化其结构。可以在 ./linux/include/linux/mutex.h 中查看该实现。

DEFINE_MUTEX( my_mutex );

5.3   互斥锁 API

互斥锁 API 提供了 5 个函数：

其中 3 个用于锁定：

/*

 * See kernel/mutex.c for detailed documentation of these APIs.

 * Also see Documentation/mutex-design.txt.

 */

extern void fastcall mutex_lock(struct mutex *lock);

如果想等待这个锁，可以调用 mutex_lock。这个调用在互斥锁可用时返回，否则，在互斥锁锁可用之前它将休眠。无论在哪种情形中，当控制被返回时，调用者将持有互斥锁。

extern int fastcall mutex_lock_interruptible(struct mutex *lock);

在这种情况下，该函数可能返回 -EINTR。

extern int fastcall mutex_trylock(struct mutex *lock);

在需要立即锁定以及希望在互斥锁不可用时掌握控制的情形下，可以使用第一个函数 mutex_trylock。

一个用于解锁：

extern void fastcall mutex_unlock(struct mutex *lock);

当一个互斥锁被锁定后，它必须被解锁。这是由 mutex_unlock 函数来完成的。这个函数不能从中断上下文调用。

另一个用于测试互斥锁：

static inline int fastcall mutex_is_locked(struct mutex *lock)

{

        return atomic_read(&lock->count) != 1;

}

检查互斥锁的状态。如果互斥锁被持有（锁定），那么就会返回 1；否则，返回 0。

参考资料：

信号灯的相关源代码分析，余旭

《Linux设备驱动程序（第3版）》

内核中的同步与任务调度，康华

Linux内核中的同步和互斥分析报告

Linux系统内核的同步机制“自旋锁”

Linux 同步方法剖析

Linux 同步方法