Linux并发与竞争（一）

Linux 并发与竞争

在讲 Linux 并发操作之前先了解一下并发和并行区别，这两个说法都是指多个操作同时被执行，不过这两个概念具有很大的差别，很多时候会混淆这两个概念。

并发强调执行多个操作的对象只能有一个，并行则不强调，多个操作可以由多个对象执行。

并发是指一个处理器同时处理多个任务，并行是指多个处理器或者是多核的处理器同时处理多个不同的任务。可以说并发是一个人同时搬三块砖头，而并行是三个人同时搬三块砖头。

并发是逻辑上的同时发生，而并行是物理上的同时发生。并发在单处理器系统上表现为线程在微观串行执行，而在宏观并发执行，即多线程交织执行。

1. 并发优缺点

由于并发机制的支持使得多条程序可以同时被执行，在不支持并发机制的情况下多条程序需要按先后次序依次被执行。如果当前执行耗时程序，那在这段时间内下一条程序将无法得到执行，如果下一条程序属于紧急处理程序那可能造成无法响应紧急异常的情况，而并发机制则可以解决这个问题。并发可以提高 CPU 的使用率即减少 CPU 空闲时间。

由于并发机制的支持所以多个线程（子程序）可同时被 CPU 执行，这也意味着一个资源（软件或硬件）被多个线程同时访问的可能性更高。但是很多时候一个资源同时只能响应一个程序的访问，比如硬盘，比如一个程序读取硬盘数据同时另一个程序向硬盘写入数据，那么就会导致写入或读出的数据被干扰损坏，所以并发的缺点就是存在资源竞争，所以我们要做的就是协商好资源的访问。

2. 原子操作

对于一条程序操作语句来说，这个操作可能由多条更小粒度的操作语句组成，而小粒度操作语句可能由更小粒度的操作组成（这类似于我们使用函数封装出复杂操作一样），而原子操作就是最小的且不可再拆分的操作。

不仅是函数，看似最简单的数值 +1 操作实际上也是可以再拆分的，所以判断原子操作不能简单观察是否为单语句，对于 C 程序来说不可再分的单语句，在汇编的世界里依旧可再拆分，一般要汇编化后才能够判断。

而单语句汇编后可再拆分这会导致什么问题呢？会导致即使一个单语句也会因为并发竞争时被不同程序同时操作了不同部分而导致单语句发生异常。

但是对于源码层面单语句确实不可再分，那怎么保证单语句汇编化后可再分的部分不会被并发程序抢占？要解决这个问题就要保证这些可再分的部分被作为一个整体运行，所以原子操作就是将汇编可拆分的非原子操作让其变为原子操作（作为一个整体不可拆分），Linux 内核提供了一组原子操作 API 函数来完成此功能。

2.1 原子整形操作

Linux 内核定义了 atomic_t 的结构体来完成整形数据的原子操作，用原子变量来代替编程用的整形变量，原子变量定义在 “include/linux/types.h” 文件中，如下。

typedef struct {int counter;
} atomic_t;

使用原子操作 API 函数，首先要定义一个 atomic_t 的变量（对象），定义原子变量的时候给原子变量赋初值，定义原子变量 hello 并赋初值为 0，例如这样。

atomic_t hello = ATOMIC_INIT(0);

操作原子变量，比如读写改变原子变量的值等等，不能直接访问而是需要使用Linux 内核提供的 API 函数。这些操作函数被定义在 “tools\include\asm-generic\bitops\atomic.h” 文件中。

/*定义原子变量的时候对其初始化*/
ATOMIC_INIT(int i);
/*读取 v 的值，并且返回*/
int atomic_read(atomic_t *v);
/*向 v 写入 i 值*/
void atomic_set(atomic_t *v, int i);
/*给 v 加上 i 值*/
void atomic_add(int i, atomic_t *v);
/*从 v 减去 i 值*/
void atomic_sub(int i, atomic_t *v);
/*给 v 加 1，也就是自增*/
void atomic_inc(atomic_t *v);
/*从 v 减 1，也就是自减*/
void atomic_dec(atomic_t *v);
/*从 v 减 1，并且返回 v 的值*/
int atomic_dec_return(atomic_t *v);
/*给 v 加 1，并且返回 v 的值*/
int atomic_inc_return(atomic_t *v);
/*从 v 减 i，如果结果为 0 就返回真，否则返回假*/
int atomic_sub_and_test(int i, atomic_t *v);
/*从 v 减 1，如果结果为 0 就返回真，否则返回假*/
int atomic_dec_and_test(atomic_t *v);
/*给 v 加 1，如果结果为 0 就返回真，否则返回假*/
int atomic_inc_and_test(atomic_t *v);
/*给 v 加 i，如果结果为负就返回真，否则返回假*/
int atomic_add_negative(int i, atomic_t *v);

2.2 原子位操作

除了整形之外 Linux 内核还提供了的原子位操作 API 函数，只是原子位操作没有类似原子整形变量那样的 atomic_t 数据结构，原子位操作属于直接操作内存。这些 API 函数被定义在 ”arch\alpha\include\asm" 目录下。

/*将 p 地址的第 nr 位置 1*/
void set_bit(int nr, void *p);
/*将 p 地址的第 nr 位清零*/
void clear_bit(int nr,void *p);
/*将 p 地址的第 nr 位进行翻转*/
void change_bit(int nr, void *p);
/*获取 p 地址的第 nr 位的值*/
int test_bit(int nr, void *p);
/*将 p 地址的第 nr 位置 1，并且返回 nr 位原来的值*/
int test_and_set_bit(int nr, void *p);
/*将 p 地址的第 nr 位清零，并且返回 nr 位原来的值*/
int test_and_clear_bit(int nr, void *p);
/*将 p 地址的第 nr 位翻转，并且返回 nr 位原来的值*/
int test_and_change_bit(int nr, void *p);

3. 自旋锁

原子操作只能保护整形变量或者位，在实际的编程环境中不可能只有整形变量或位这种简单的临界区。有限共享资源资源可能需要互斥访问 (mutual exclusion) ，这就需要使用另外一种机制，自旋锁。

只有获取了锁的线程才能够对资源进行访问，所以对于互斥访问同一时刻只能有一个线程获取到锁，只要此线程不释放持有的锁，那么其他的线程都不能获取此锁。

那没有获取到锁的线程应该怎么办？很简单，无法获取到锁的线程，该线程将会等待，间隔一段时间后会再次尝试获取，并且会重复这个过程，所以这种会循环的机制就叫做自旋锁（spinlock）。

特性：

(1) 忙等待重复循环获取锁的特性会消耗 CPU 的性能，所以注定自旋锁使用场景上的限制即自旋锁不适合被线程长时间的持有，遇到需要长时间持有锁的场景需要换其他的方法。

(2) 自旋锁不可递归，自己等待自己已经获取的锁，会导致死锁。

(3) 得到自旋锁的线程会暂时禁止内核抢占。

(4) 由于第 (1) 点特性就可以很明确知道自旋锁是 不会休眠的，所以这样的特性可以用在不能睡眠的场景下加锁，比如在中断上下文（中断要求快进快出不能在中断过程睡眠的）。

3.1 方法浅析

Linux 内核使用结构体 spinlock_t 表示自旋锁，和原子操作同样的套路，使用之前先定义类型，类型定义在 “include\linux\spinlock_types.h” 文件中。

typedef struct spinlock {union {struct raw_spinlock rlock;#ifdef CONFIG_DEBUG_LOCK_ALLOC
# define LOCK_PADSIZE (offsetof(struct raw_spinlock, dep_map))struct {u8 __padding[LOCK_PADSIZE];struct lockdep_map dep_map;};
#endif};
} spinlock_t;

这里讲解一个查看 Linux 源码的小技巧，在 Linux 内核中按照功能区分把源码和头文件采用分离的方式组织，所以关于一类功能类型的定义一律在源码根目录下的 “include/linux” 目录下查找，然后根据功能定位到相应名称的文件夹或头文件，在相应的文件夹，或文件中即可找到你要的类型定义。

在使用自旋锁之前，肯定要先定义一个自旋锁变量（对象），定义好自旋锁变量以后就可以使用相应的 API 函数来操作自旋锁。

spinlock_t lock; //定义自旋锁

3.2 普通 API 函数

下面是一些常用的自旋锁变量（对象）操作函数。

/*定义并初始化一个自选变量*/
DEFINE_SPINLOCK(spinlock_t lock);
/*初始化自旋锁*/
int spin_lock_init(spinlock_t *lock);
/*获取指定的自旋锁，也叫做加锁*/
void spin_lock(spinlock_t *lock);
/*释放指定的自旋锁*/
void spin_unlock(spinlock_t *lock);
/*尝试获取指定的自旋锁，如果没有获取到就返回 0*/
int spin_trylock(spinlock_t *lock);
/*检查指定的自旋锁是否被获取，如果没有被获取就返回非 0，否则返回 0*/
int spin_is_locked(spinlock_t *lock);

自旋锁 API 函数适用于 SMP(Symmetric Multi Processing，对称多处理系统) 或支持抢占的单 CPU 下线程之间的并发访问，也就是用于线程与线程之间，被自旋锁保护的临界区一定不能调用任何能够引起睡眠和阻塞的 API 函数，线程在尝试获取锁之前必须先关闭硬件中断，否则的话会可能会导致死锁现象的发生。

线程睡眠影响

如果线程 1 在持有锁期间进入了休眠状态，那么线程 1 会自动放弃 CPU 使用权。线程 2 开始运行，线程 2 也想要获取锁，但是此时锁被 1 线程持有，而且内核抢占还被禁止了！线程 2 无法被调度出去，那么线程 1 就无法运行，线程 1 无法重新运行锁也就无法释放，这样两个线程就发生死锁。

所以自旋锁保护的临界区一定不能调用任何能够引起睡眠和阻塞的 API 函数。

中断带来影响

线程 A 先运行，并且获取到了锁，当线程 A 运行到某个函数的时候中断发生了，中断抢走了 CPU 使用权。中断也想访问共享资源，中断服务函数也要获取这个锁，但是这个锁被线程 A 占有着，中断就会一直自旋，等待锁有效。但是在中断属于硬件级最高优先级，中断服务函数执行完之前，线程 A 是不可能执行的，线程 A 无法重新运行锁也就无法释放，这样中断和线程就发生死锁。

所以最好的解决方法就是普通线程获取锁之前关闭本地中断，释放锁后恢复中断。

local_irq_enable();
local_irq_disable();

3.3 关中断 API 函数

对于避免中断和线程之间的死锁，最好的解决方法就是普通线程获取锁之前关闭本地中断，但是在编程阶段经常被遗忘而导致程序发生死锁，所以 Linux 内核提供了相应的 API 函数，可以在获取锁时自动开启或关闭中断。

/*禁止本地中断，并获取自旋锁*/
void spin_lock_irq(spinlock_t *lock);
/*激活本地中断，并释放自旋锁*/
void spin_unlock_irq(spinlock_t *lock);
/*保存中断状态，禁止本地中断，并获取自旋锁*/
void spin_lock_irqsave(spinlock_t *lock, unsigned long flags);
/*将中断状态恢复到以前的状态，并且激活本地中断，释放自旋锁*/
void spin_unlock_irqrestore(spinlock_t*lock, unsigned long flags);

注意： 可以看到和普通的相比多了 irq 后缀，这个后缀指的是能自动开关中断，不是说在中断函数中使用的。

使用 “spin_lock_irq/spin_unlock_irq” API 的时候要求用户能够确定加锁之前的中断状态，但实际上内核很庞大，运行过程复杂多变，难以确定某个时刻的中断状态，因此不推荐使用 “spin_lock_irq/spin_unlock_irq”。建议使用 “spin_lock_irqsave/ spin_unlock_irqrestore”，区别在于这里又多了后缀，指的是这组函数会保存中断状态，在释放锁的时候会恢复中断状态。

3.4 总结

在线程中使用 “spin_lock_irqsave/spin_unlock_irqrestore”，在中断中使用 “spin_lock/spin_unlock”。

4. 读写自旋锁

自旋锁只有一个，即某一段时刻锁只能由一个线程持有，表现就是同一时刻只能有一个线程写入或读取文件，如果一种共享文件以这种方式进行保护的话，显而易见这种保护机制会让共享文件的访问效率低下。

显然这种形式无法满足共享文件的初衷，即共享文件是可以并发读取的，只需要保证在修改共享文件的时候没人读取，或者在其他人读取共享文件的时候没有人修改共享文件即可。即共享文件的读和写不能同时进行，但是可以多人并发的读取。

基于这种理念呢设计出了读写自旋锁，从字面意思也可以知道它由 读锁 和写锁 两部分构成，如果只读取共享资源用 读锁 加锁，如果要修改共享资源则用 写锁加锁。

4.1 工作原理

当 写锁 没有被线程持有时，多个线程能够并发地持有读锁，这大大提高了共享资源的访问效率，因为 读锁 是用于读取共享资源的场景，所以多个线程同时持有读锁也不会破坏共享资源的数据。
但是，一旦 写锁 被线程持有后，读线程的获取读锁的操作会被阻塞，而且其他写线程的获取写锁的操作也会被阻塞。

4.2 方法浅析

Linux 内核使用 rwlock_t 结构体表示读写锁，同样的套路，使用之前先定义类型，类型定义在 “include\linux\rwlock_types.h” 文件中。

typedef struct {struct rwbase_rt    rwbase;atomic_t        readers;
#ifdef CONFIG_DEBUG_LOCK_ALLOCstruct lockdep_map    dep_map;
#endif
} rwlock_t;

同样的套路在使用之前，肯定要先定义一个读写自旋锁变量（对象），定义好读写自旋锁变量以后就可以使用相应的 API 函数来操作自旋锁。

rwlock_t rw_lock;

4.3 API 函数

读写锁的操作 API 函数分为两部分，一种是给读操作使用的，另一种是给写操作使用的，常用的 API 函数具体如下。

定义/*定义并初始化读写锁*/
DEFINE_RWLOCK(rwlock_t lock);
/*初始化读写锁*/
void rwlock_init(rwlock_t *lock);读锁/*获取读锁*/
void read_lock(rwlock_t *lock);
/*释放读锁*/
void read_unlock(rwlock_t *lock);
/*禁止本地中断，并且获取读锁*/
void read_lock_irq(rwlock_t *lock);
/*打开本地中断，并且释放读锁*/
void read_unlock_irq(rwlock_t *lock);
/*保存中断状态，禁止本地中断，并获取读锁*/
void read_lock_irqsave(rwlock_t *lock, unsigned long flags);
/*将中断状态恢复到以前的状态，并且激活本地中断，释放读锁*/
void read_unlock_irqrestore(rwlock_t *lock, unsigned long flags);
/*关闭下半部，并获取读锁*/
void read_lock_bh(rwlock_t *lock);
/*打开下半部，并释放读锁*/
void read_unlock_bh(rwlock_t *lock);写锁/*获取写锁*/
void write_lock(rwlock_t *lock);
/*释放写锁*/
void write_unlock(rwlock_t *lock);
/*禁止本地中断，并且获取写锁*/
void write_lock_irq(rwlock_t *lock);
/*打开本地中断，并且释放写锁*/
void write_unlock_irq(rwlock_t *lock);
/*保存中断状态，禁止本地中断，并获取写锁*/
void write_lock_irqsave(rwlock_t *lock, unsigned long flags);
/*将中断状态恢复到以前的状态，并且激活本地中断，释放读锁*/
void write_unlock_irqrestore(rwlock_t *lock, unsigned long flags);
/*关闭下半部，并获取读锁*/
void write_lock_bh(rwlock_t *lock);
/*打开下半部，并释放读锁*/
void write_unlock_bh(rwlock_t *lock);

如果使用 read_lock 和 write_lock 成功获取读写自旋锁，read_lock 和 write_lock会立即返回，如果未获取读写自旋锁，read_lock 和 write_lock 宏会被阻塞（在那自旋)，直到可以获取读写自旋锁才返回。

其中一些带 irq，irqsave 后缀的 API 函数中用法和普通自旋锁一样是支持线程获取锁前关中断，和保存中断状态的，用法一样这里不重复说明（看看上一节）。

5. 顺序锁

读写锁的读操作和写操作不能同时进行，同时读写锁的读操作锁可以并发获取，所以可以认为读写锁更侧重提高读操作的性能，以至于因为写操作的优先级很低，非常影响写进程的执行效率，极端情况下甚至可能出现写进程饿死的情况。

而顺序锁和读写锁的区别就在于顺序锁对读写线程的优先级做了调整，让写操作的优先级始终高于读操作，从而实现写操作可以任意时刻打断读操作，因此顺序锁更适用于写操作优先级更高的应用场景（注意不等同于读操作频繁）。

特性：

(1) 读写锁对读和写进程都互斥，而顺序锁只对写进程互斥，可以允许在写的时候进行读操作，也就是实现同时读写。

(2) 不允许同时进行并发的写操作。

(3) 为了保证数据完整性，如果在读的过程中发生了写操作，最好重新进行读取。

(4) 顺序锁不是为了替代读写锁，只是让写操作的优先级更高，适用条件是读多写少的情况，如果写频繁的话，会导致读进程被频繁打断。

工作原理

(1) 使用一个锁变量来记录锁的状态，该锁变量是循环递增的，读者只读取锁变量，不对锁变量进行任何操作。

(2) 读加锁不使用同步机制，甚至不需要禁止内核抢占，因为读者只读取锁变量。

(3) 写者在写之前将锁变量加 1，执行完写之后将锁变量再加 1，意味着当锁变量为偶数时，表示没有写者正在执行，当变量为奇数时，表示写者正在执行操作。

(4) 读者实现同步的方式为：当锁变量为奇数时，读者自旋等待，只有当锁变量为偶数时，读者才执行读操作，同时在读之前记录锁变量的值，在读完成之后再对比锁变量的值，如果不一致，表示在读的过程中有写者更新，返回特定值。

(5) 写操作直接使用 spinlock 进行互斥，以保证多进程或者多CPU之间不存在同步地写操作。

5.1 方法浅析

Linux 内核使用 seqlock_t 结构体表示顺序锁，同样的套路，使用之前先定义类型，类型定义在 “include\linux\seqlock.h” 文件中。

typedef struct {/** Make sure that readers don't starve writers on PREEMPT_RT: use* seqcount_spinlock_t instead of seqcount_t. Check __SEQ_LOCK().*/seqcount_spinlock_t seqcount;spinlock_t lock;
} seqlock_t;

5.2 API 函数

顺序锁的操作 API 函数分为两部分，一种是给读操作使用的，另一种是给写操作使用的，常用的 API 函数具体如下。

定义/*定义并初始化顺序锁*/
DEFINE_SEQLOCK(seqlock_t sl);
/*初始化顺序锁*/
void seqlock_init(seqlock_t *sl);顺序锁写操作/*获取写顺序锁。*/
void write_seqlock(seqlock_t *sl);
/*释放写顺序锁。*/
void write_sequnlock(seqlock_t *sl);
/*禁止本地中断，并且获取写顺序锁*/
void write_seqlock_irq(seqlock_t *sl);
/*打开本地中断，并且释放写顺序锁。*/
void write_sequnlock_irq(seqlock_t *sl);
/*保存中断状态，禁止本地中断，并获取写顺序锁*/
void write_seqlock_irqsave(seqlock_t *sl, unsigned long flags);
/*将中断状态恢复到以前的状态，并且激活本地中断，释放写顺序锁*/
void write_sequnlock_irqrestore(seqlock_t *sl, unsigned long flags);
/*关闭下半部，并获取写读锁*/
void write_seqlock_bh(seqlock_t *sl);
/*打开下半部，并释放写读锁*/
void write_sequnlock_bh(seqlock_t *sl);顺序锁读操作/*读单元访问共享资源的时候调用此函数，此函数会返回顺序锁的顺序号。*/
unsigned read_seqbegin(const seqlock_t *sl);
/*读结束以后调用此函数检查在读的过程中有没有对资源进行写操作，如果有的话就要重读*/
unsigned read_seqretry(const seqlock_t *sl, unsigned start);