一、原子操作

1.1 产生起源

我们的程序逻辑经常遇到这样的操作序列:

  1. 读一个位于memory中的变量的值到寄存器中
  2. 修改该变量的值(也就是修改寄存器中的值)
  3. 将寄存器中的数值写回memory中的变量值

如果这些操作是一个串行化的操作(在一个thread中串行执行),那么一切是OK的,但是世界总不能如我们所愿,在现在的多CPU架构和支持抢占的内核系统中,总会出现各种奇怪的现象。比如要实现同一时刻只有一个进程打开驱动文件,可以想到用标志位来简单的互斥,但因为CPU的调度是随机的,就会有问题如下:

0.竞争实例1

0.竞争实例2

0.竞争实例3

所以处理器在访问共享资源时,必须对临界区进行同步,即保证同一时间内,只有一个对临界区的访问者。当共享资源为一内存地址时,原子操作是对该类型共享资源同步访问的最佳方式。随着应用的日益复杂和SMP的广泛使用,处理器都开始提供硬件同步原语以支持原子地更新内存地址。

从ARMv6架构开始,ARM处理器提供了Exclusive accesses同步原语,包含两条指令:LDREX和STREX,将对一个内存地址的操作拆分成两个步骤,同CPU内置的记录exclusive accesses的exclusive monitors一起,完成对内存的原子操作。

LDREX与LDR指令类似,完成将内存中的数据加载进寄存器的操作。与LDR指令不同的是,该指令也会同时初始化exclusive monitor来记录对该地址的同步访问。例如LDREX R1, [R0]会将R0寄存器中内存地址的数据,加载进R1中并更新exclusive monitor。

STREX的格式为:STREX Rd, Rm, [Rn],STREX会根据exclusive monitor的指示决定是否将寄存器中的值写回内存中。如果exclusive monitor许可这次写入,则STREX会将寄存器Rm的值写回Rn所存储的内存地址中,并将Rd寄存器设置为0表示操作成功。如果exclusive monitor禁止这次写入,则STREX指令会将Rd寄存器的值设置为1表示操作失败并放弃这次写入。应用程序可以根据Rd中的值来判断写回是否成功。

1.2 原子操作的实现原理与使用

所谓“原子操作”就是这个操作不会被打断。Linux有2种原子操作:原子变量、原子位。原子变量类型如下,实际上就是一个结构体,特殊的地方在于它的操作函数,如下(下表中v都是atomic_t指针):

1
2
3
4
5
6
7
8
9
10
11
12
typedef struct {
int counter;
} atomic_t;
函数名 作用
atomic_read(v) 读出原子变量的值,即v->counter
atomic_set(v,i) 设置原子变量的值,即v->counter = i
atomic_inc(v) v->counter++
atomic_dec(v) v->counter--
atomic_add(i,v) v->counter += i
atomic_sub(i,v) v->counter -= i
atomic_inc_and_test(v) 先加1,再判断新值是否等于0;等于0的话,返回值为1
atomic_dec_and_test(v) 先减1,再判断新值是否等于0;等于0的话,返回值为1

1.3 原子变量的内核实现

SMP就是Symmetric Multi-Processors,对称多处理器;UP即Uni-Processor,系统只有一个单核CPU。原子变量的API都在Linux内核文件arch\arm\include\asm\atomic.h中。

atomic_read,atomic_set这些操作都只需要一条汇编指令,所以它们本身就是不可打断的。问题在于atomic_inc这类操作,要读出、修改、写回。以atomic_inc为例,在atomic.h文件中,如下定义:

1
#define atomic_inc(v)		atomic_add(1, v)

atomic_add又是怎样实现的呢?用这个宏:ATOMIC_OPS(add, +=, add),把这个宏展开:

1
2
3
4
#define ATOMIC_OPS(op, c_op, asm_op)					\
ATOMIC_OP(op, c_op, asm_op) \
ATOMIC_OP_RETURN(op, c_op, asm_op) \
ATOMIC_FETCH_OP(op, c_op, asm_op)

从上面的宏可以知道,一个ATOMIC_OPS定义了3个函数。比如“ATOMIC_OPS(add, +=, add)”就定义了这3个函数:

1
2
3
atomic_add
atomic_add_return
atomic_atomic_fetch_add 或 atomic_fetch_add_relaxed

我们以ATOMIC_OP(add, +=, add)为例,看它是如何实现atomic_add函数的,对于UP系统、SMP系统,分别有不同的实现方法。

1.3.1 ATOMIC_OP在UP系统中的实现

对于ARMv6以下的CPU系统,不支持SMP。原子变量的操作简单粗暴:关中断,中断都关了,谁能来打断我?代码如下(arch\arm\include\asm\atomic.h):

1.原子操作1.

1.3.2 ATOMIC_OP在SMP系统中的实现

对于ARMv6及以上的CPU,有一些特殊的汇编指令来实现原子操作,不再需要关中断,代码如下(arch\arm\include\asm\atomic.h):

1.原子操作2

在ARMv6及以上的架构中,有ldrex、strex指令,ex表示exclude,意为独占地。这2条指令要配合使用,举例如下:

  1. 读出:ldrex r0, [r1];读取r1所指内存的数据,存入r0,并且标记r1所指内存为“独占访问”。如果有其他程序再次执行“ldrex r0, [r1]”,一样会成功,一样会标记r1所指内存为“独占访问”。
  2. 修改r0的值
  3. 写入:strex r2, r0, [r1];如果r1的“独占访问”标记还存在,则把r0的新值写入r1所指内存,并且清除“独占访问”的标记,把r2设为0表示成功。如果r1的“独占访问”标记不存在了,就不会更新内存,并且把r2设为1表示失败。

假设这样的抢占场景:

  1. 程序A在读出、修改某个变量时,被程序B抢占了;
  2. 程序B先完成了操作,程序B的strex操作会清除“独占访问”的标记;
  3. 轮到程序A执行剩下的写入操作时,它发现独占访问”标记不存在了,于是取消写入操作。这就避免了这样的事情发生:程序A、B同时修改这个变量,并且都自认为成功了。

举个例子,比如atomic_dec,假设一开始变量值为1,程序A本想把值从1变为0;但是中途被程序B先把值从1变成0了;但是没关系,程序A里会再次读出新值、修改、写入,最终这个值被程序A从0改为-1。

在ARMv6及以上的架构中,原子操作不再需要关闭中断,关中断的花销太大了。并且关中断并不适合SMP多CPU系统,你关了CPU0的中断,CPU1也可能会来执行些操作啊。

在ARMv6及以上的架构中,原子操作的执行过程是可以被打断的,但是它的效果符合“原子”的定义:一个完整的“读、修改、写入”是原子的,不会被别的程序打断。它的思路很简单:如果被别的程序打断了,那就重来一遍“读、修改、写入”,最后总会成功的。

1.3.3 原子变量使用案例

使用原子变量实现:只能有一个APP访问驱动程序。代码如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
static atomic_t valid = ATOMIC_INIT(1);
static ssize_t gpio_key_drv_open (struct inode *node, struct file *file)
{
if (atomic_dec_and_test(&valid))
return 0;
atomic_inc(&valid);
return -EBUSY;
}
static int gpio_key_drv_close (struct inode *node, struct file *file)
{
atomic_inc(&valid);
return 0;
}

1.原子操作3

1.4 原子位的内核实现

在ARMv6以下的架构里,不支持SMP系统,原子位的操作函数也是简单粗暴:关中断。以set_bit函数为例,代码在内核文件arch\arm\include\asm\bitops.h中,如下:

1.原子位操作4

在ARMv6及以上的架构中,不需要关中断,有ldrex、strex等指令,这些指令的作用在前面介绍过,还是以set_bit函数为例,代码如下:

1.原子位操作5

二、Linux锁的介绍与使用

2.1 自旋锁

Linux内核中最常见的锁是自旋锁,自旋锁最多只能被一个可执行线程持有。如果一个线程试图获取一个已被持有的自旋锁,这个线程会进行忙循环——旋转等待锁重新可用,要是锁未被争用,请求锁的执行线程会立马得到它,继续执行。

  1. 持有自旋锁时会关闭内核抢占,他内部会调用调用“preempt_disable()”,防止当前CPU的其他进程抢占当前持有锁的进程,然后又去获取自旋锁导致死锁。
  2. 持有自旋锁时不能睡眠,不然会有死锁风险,由于不睡眠,可以在中断上下文中使用。
  3. 自旋锁最初的设计就是为了SMP系统设计,实现多处理器情况下保护临界区。如果UP不支持抢占,spin_lock是空函数,若UP支持抢占,这时spin_lock的实现就是调用“preempt_disable()”。即:单核没有“取锁动作”,只有多核时自旋锁才存在“取锁动作”

自旋锁最应该防止的就是死锁,不同情况下可能需要禁止本地中断,或者禁止本地下半部:

2.锁的介绍1

举例介绍一下,上表中第一列“IRQ Handler A”和第三行“Softirq A”的交叉点是“spin_lock_irq()”,意思就是说如果“IRQ Handler A”和“Softirq A”要竞争临界资源,那么需要使用“spin_lock_irq()”函数。为什么不能用spin_lock而要用spin_lock_irq?也就是为什么要把中断给关掉?假设在Softirq A中获得了临界资源,这时发生了IRQ A中断,IRQ Handler A去尝试获得自旋锁,这就会导致死锁:所以需要关中断。

我的总结是:锁是用来防SMP,禁止中断和下部是防止本地死锁。注意,需要禁止的只是当前处理器上的中断和下半部,如果中断和下半部发生在不同的处理器上,即使在同一把锁上自旋,让它等待当前处理器释放锁即可,不会导致死锁。

2.2 信号量

Linux中的信号量是一种睡眠锁,如果一个任务试图获得一个不可用(已经被占用)的信号量时,信号量会将其推进一个等待队列,然后让其睡眠。这是处理器能重获自由,从而去执行其他代码。当持有的信号量可用(被释放)后,处于等待队列中的那个任务将被唤醒,并获得信号量。

  1. 信号量适用于锁会被长时间持有的情况。因为睡眠和维护等待队列以及唤醒有一定的开销。
  2. 占用信号量的同时不能占用自旋锁,因为在等待信号量时可能会睡眠。
  3. 信号量不会禁止内核抢占,所有持有信号量的代码可以被抢占。
  4. 信号量可以允许任意数量的锁持有者,而自旋锁和互斥体一个时刻只允许一个任务持有。

2.3 互斥体

mutex的使用场景相对而言更严格,它是计数为1的信号量(有区别),注意事项如下:

  1. 同一时刻只有一个线程可以持有mutex,当进程持有mutex时,进程不可以退出(有争议)。
  2. 只有锁持有者可以解锁,不能在一个进程中持有mutex,在另外一个进程中释放。
  3. 不允许递归地加锁和解锁,不允许在中断处理程序或者中断下半部中使用

2.4 选择合适的锁

  1. 低开销、短期锁定优先使用自旋锁。
  2. 中断上下文中只能使用自旋锁。
  3. 长期加锁使用信号量,优先使用互斥体。
  4. 持有锁需要睡眠,优先使用互斥体。
  5. 信号量和互斥体:除非互斥体的某个约束妨碍你的使用,否则优先使用mutex。

三、内核中锁的实现

3.1 spinlock在UP系统中的实现

对于单CPU系统,没有“其他CPU”;如果内核不支持preempt,当前在内核态执行的线程也不可能被其他线程抢占,也就“没有其他进程/线程”。所以,对于不支持preempt的单CPU系统,spin_lock是空函数,不需要做其他事情。

如果单CPU系统的内核支持preempt,即当前线程正在执行内核态函数时,它是有可能被别的线程抢占的。这时spin_lock的实现就是调用“preempt_disable()”:你想抢我,我干脆禁止你运行。

在UP系统中spin_lock()退化为preempt_disable(),如果用的内核不支持preempt,则spin_lock()什么事都不用做,spin_lock函数定义如下:

3.内核中锁的实现1

对于spin_lock_irq(),在UP系统中就退化为local_irq_disable()和preempt_disable(),假设程序A要访问临界资源,可能会有中断也来访问临界资源,可能会有程序B也来访问临界资源,那么使用spin_lock_irq()来保护临界资源:先禁止中断防止中断来抢,再禁止preempt防止其他进程来抢。如下图所示:

3.内核中锁的实现2

对于spin_lock_bh(),在UP系统中就退化为禁止软中断和preempt_disable(),如下图所示:

3.内核中锁的实现3

3.2 spinlock在SMP系统中的实现

要让多CPU中只能有一个获得临界资源,使用原子变量就可以实现。但是还要保证公平,先到先得。比如有CPU0、CPU1、CPU2都调用spin_lock想获得临界资源,谁先申请谁先获得。

举个例子:餐厅里只有一个座位,去吃饭的人都得先取号、等叫号。注意,有2个动作:顾客从取号机取号,电子叫号牌叫号。

  1. 一开始取号机待取号码为0。
  2. 顾客A从取号机得到号码0,电子叫号牌显示0,顾客A上座;取号机显示下一个待取号码为1。
  3. 顾客B从取号机得到号码1,电子叫号牌还显示为0,顾客B等待;取号机显示下一个待取号码为2。
  4. 顾客C从取号机得到号码2,电子叫号牌还显示为0,顾客C等待;取号机显示下一个待取号码为3。
  5. 顾客A吃完离座,电子叫号牌显示为1,顾客B的号码等于1,他上座;
  6. 顾客B吃完离座,电子叫号牌显示为2,顾客C的号码等于2,他上座;

在这个例子中有2个号码:取号机显示的“下一个号码”,顾客取号后它会自动加1;电子叫号牌显示“当前号码”,顾客离座后它会自动加1。某个客户手上拿到的号码等于电子叫号牌的号码时,该客户上座。在这个过程中,即使顾客B、C同时到店,只要保证他们从取号机上得到的号码不同,他们就不会打架。

所以,关键点在于:取号机的号码发放,必须互斥,保证客户的号码互不相同。而电子叫号牌上号码的变动不需要保护,只有顾客离开后它才会变化,没人争抢它。

在ARMv6及以上的ARM架构中支持SMP系统。它的spinlock结构体定义如下:

3.内核中锁的实现4

__raw_tickets结构体中有owner、next两个成员,这是在SMP系统中实现spinlock的关键。owner就相当于电子叫号牌,现在谁在吃饭。next就当于于取号机,下一个号码是什么。从取号机上取到的号码保存在spin_lock函数中的局部变量里。

spin_lock函数调用关系如下,核心是arch_spin_lock:

3.内核中锁的实现5

arch_spin_lock代码如下:即使不同的个体去同时取号,也可以保证取到的号码各不相同。

3.内核中锁的实现6

假设第1个程序取到了号码,它访问了临界资源后,调用spin_unlock,代码如下:

3.内核中锁的实现7

释放锁后,假如有其他程序正在spin_lock函数中循环等待,它就会立刻判断自己手上的next是否等于lock->tickets.owner,如果相等就表示轮到它获得了锁。

3.3 信号量semaphore的实现

信号量的定义及操作函数都在Linux内核文件include\linux\semaphore.h中定义,如下:

3.内核中锁的实现8

初始化semaphore之后,就可以使用down函数或其他衍生版本来获取信号量,使用up函数释放信号量。

3.3.1 down函数的实现

如果semaphore中的count大于0,那么down函数就可以获得信号量;否则就休眠。在读取、修改count时,要使用spinlock来实现互斥。休眠时,要把当前进程放在semaphore的wait_list链表中,别的进程释放信号量时去wait_list中把进程取出、唤醒。

3.内核中锁的实现9

3.3.2 up函数的实现

如果有其他进程在等待信号量,则count值无需调整,直接取出第1个等待信号量的进程,把信号量给它,然后把它唤醒。如果没有其他进程在等待信号量,则调整count。整个过程需要使用spinlock来保护,代码如下:

3.内核中锁的实现10

3.4 互斥量mutex的实现

mutex的定义及操作函数都在Linux内核文件include\linux\mutex.h中定义,如下:

3.内核中锁的实现11

这里要注意一下:《linux内核设计与实现》说mutex中的owner是用来记录获得mutex的进程,以后必须由它来释放mutex。但是从上面的代码可知,owner并不一定存在!应该是内核版本不同,书中的版本是2.6,此处的版本是4.x。

owner有2个用途:debug(CONFIG_DEBUG_MUTEXES)或spin_on_owner(CONFIG_MUTEX_SPIN_ON_OWNER)。

什么叫spin on owner?我们使用mutex的目的一般是用来保护一小段代码,这段代码运行的时间很快。这意味着一个获得mutex的进程,可能很快就会释放mutex。针对这点可以进行优化,特别是当前获得mutex的进程是在别的CPU上运行、并且“我”是唯一等待这个mutex的进程。在这种情况下,那“我”就原地spin等待吧:懒得去休眠了,休眠又唤醒就太慢了。

所以,mutex是做了特殊的优化,比semaphore效率更高。但是在代码上,并没有要求“谁获得mutex,就必须由谁释放mutex”,只是在使用惯例上是“谁获得mutex,就必须由谁释放mutex”。

3.4.1 mutex_lock函数的实现

mutex的设计非常精巧,比semaphore复杂,但是更高效。首先要知道mutex的操作函数中有fastpath、slowpath两条路径(快速、慢速):如果fastpath成功,就不必使用slowpath。

比较复杂,未完待续……