Linux的几种并发机制

一.并发机制的作用

Linux设备驱动中必须解决的一个问题就是多个进程对共享资源的并发访问，并发的访问会导致竞态。Linux内核中，主要的竞态发生在以下三种情况：

1.对称多处理器(SMP)的多个CPU

对称多处理器是一种紧耦合共享存储的系统模型，多个cpu使用共同的系统总线，因此可以访问共同的外设和存储器。

2.单CPU内进程间

Linux2.6内核支持抢占调度，一个进程在内核执行的时候可能被另一个高优先级的进程打断，进程与抢占它的进程访问共享资源的情况类似于SMP的多个cpu。

3.中断与进程间

当进程被中断打断时，如果中断处理程序访问进程正在访问的资源，也会产生竞态。此外中断也能被更高优先级的中断打断，因此多个中断之间也能导致竞态。

上述三种并发的情况，除了SMP是真正的并行，其他的都是“宏观并行，微观串行”，但实质差不多。

通常解决竞态问题的途径是让共享资源互斥访问，即一个执行单元在访问共享资源的时候，其他的执行单元静止访问。共享资源的代码称作临界区，有以下几种互斥机制：
1.中断屏蔽
2.原子操作
3.自旋锁
4.信号量
5.互斥体

二.几种机制的分析

1.中断屏蔽

由于Linux的异步I/O、进程调度等重要操作都要依赖于中断，在屏蔽中断期间所有的中断都无法得到处理，因此长时间屏蔽中断是很危险的。这就要求在屏蔽了中断后尽快地执行完临界区的代码。一般不采用中断屏蔽，不仅危险，而且严重浪费cpu资源。

在单cpu范围内：中断屏蔽使得中断与进程间的并发不再发生，而且因为进程间的调度都依赖于中断来实现，因此进程与抢占它的进程间的并发也不会发生。从而防止了竞态的发生。
单cpu的中断屏蔽：
local_irq_disable() /*屏蔽中断*/
...
critical section /*临界区*/
...
local_irq_enable /*开中断*/
注意：local_irq_disable()和local_irq_enable()都只能静止和使能本cpu内的中断，因此不能解决SMP多cpu引发的竞态。local_irq_save(flags)和local_irq_restore(flags)除了完成上述之外，还具有对当前中断信息位进行保存和恢复。如果只想禁止中断的底半部，使用local_bh_disable()，使能被local_bh_disable()禁止的底半部使用local_bh_enable()。

SMP多cpu的中断屏蔽：
irq_disable() /*屏蔽中断*/
...
critical section /*临界区*/
...
irq_enable /*开中断*/

2.原子操作

原子操作在执行过程中不会被中断。分为整形原子操作和位原子操作两类，都依赖于cpu的原子操作来实现。

2.1整形原子操作

原子变量（对象）声明：使用atomic_t数据类型来声明。

typedef struct{

int counter;

} atomic_t;

1）设置原子变量的值

void atomic_set(atomic_t *v, int i);/*设置原子变量的值为i，其实原子操作的函数内部使用的是中断屏蔽。因为临界区的指令很短，如atomic_set(atomic_t *v, int i)即将i赋值给v.counter，所以安全*/

atomic_t v =ATOMIC_INIT(0);/*定义原子变量v并初始化为0*/

2）获取原子变量的值

atomic_read(atomic_t *v);/*返回原子变量的值*/

3）原子变量加减

void atomic_add(int i, atomic_t *v);/*原子变量增加i*/

void atomic_sub(int i, atomic_t *v);/*原子变量增加i*/

4）原子变量自增自减

void atomic_inc(atomic_t *v);/*原子变量增加1*/

void atomic_dec(atomic_t *v);/*原子变量减少1*/

5）操作并测试,对原子变量执行操作后测试其是否为0，为0返回true，否则返回false

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);

6）操作并返回新的值

int atomic_add_return(int i, atomic_t *v);

int atomic_sub_return(int i, atomic_t *v);

int atomic_inc_return(atomic_t *v);

int atomic_dec_return(atomic_t *v);

示例代码如下：

static atomic_t atomic_flag = ATOMIC_INIT(1);

If(!atomic_dec_and_test(&atomic_flag))

{

atomic_inc(&atomic_flag);

//退出，共享数据被使用。

}

else

{

//临界区代码。每次只能被一个进程执行。

atomic_inc(&atomic_flag);

}

2.2位原子操作

1）设置位

void set_bit(nr, void *addr);/*设置addr地址的第nr位，即将位写为1*/

2）清除位

void clear_bit(nr, void *addr);/*清除addr地址的第nr位，即将位写0*/

3）改变位

void change_bit(nr, void *addr);/*将addr地址的第nr位进行反置*/

4）测试位

test_bit(nr, void *addr);/*返回addr地址的第nr位*/

5）测试并操作位

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)，顾名思义。为了获得一个自旋锁，在cpu上运行的代码先执行一个原子操作，测试并设置某个内存变量，该操作不能被中断，因此该变量在完成之前不能被访问。若测试得出该锁空闲，则程序获得该锁继续执行下面的代码；若锁被占用，程序将在一个小循环里重复测试和设置某个内存变量，直至自旋锁的持有者通过重置该变量释放这个自旋锁，这个小循环才得以退出，从而获得自旋锁。这个小循环也就是所谓的“自旋”，即原地打转，注意，自旋时是不会休眠的。与原子操作类似，可以理解为变量的获得与释放的过程，但是它作了一系列的封装，使用起来更加方便。

3.1 基本自旋锁

1.定义自旋锁

spinlock_t lock;

2.初始化自旋锁

spin_lock_init(lock); /*动态初始化自旋锁*/

3.获得自旋锁

spin_lock(lock); /*若能立即获得锁则获得锁并马上返回，否则将自旋，直到自旋锁的保持者释放*/

spin_trylock(lock); /*若能立即获得锁则获得锁并返回真，否则立即返回假，不会自旋*/

4.释放自旋锁

spin_unlock(lock); /*与spin_lock或spin_trylock配对使用*/

代码示例:

spinlock_t lock; /*定义一个自旋锁*/

spin_lock_init(&lock);

spin_lock(&lock); /*获取自旋锁，保护临界区*/

... /*临界区，此时抢占被禁止(即进程间切换)，但是中断仍存在*/

spin_unlock(&lock); /*解锁*/

自旋锁适用于SMP和单cpu抢占机制系统(即任务调度)，在单cpu抢占机制系统中，自旋锁持有期间内核的抢占将被禁止，因单cpu抢占系统的行为类似于SMP系统，所以这两种系统使用自旋锁十分必要。对于单cpu无抢占机制的系统，自旋锁则退化为空操作。但是使用自旋锁在执行临界区代码时，仍允许中断(和底半部中断)的发生，为了防止这种现象的发生，就需要用到自旋锁的衍生，即添加一些开关中断的功能。

spin_lock_irq() = spin_lock() + local_irq_disable()/*添加了 关中断*/

spin_unlock_irq() = spin_unlock() + local_irq_enable() /*添加了 开中断*/

spin_lock_irqsave() = spin_lock() + local_irq_save() /*添加了 关中断并恢复状态*/

spin_unlock_irqrestore() = spin_unlock() + local_irq_restore() /*添加了 开中断并恢复状态*/

spin_lock_bh() = spin_lock() + local_bh_disable() /*添加了 关底半部中断*/

spin_unlock_bh() = spin_unlock() + local_bh_enable() /*添加了 开底半部中断*/

使用自旋锁的注意事项:

1.由于自旋锁是忙等锁，会耗很多资源，所以只有在占用锁时间极短的情况下(临界区很短)，才合理，否则会降低系统性能。

2.自旋锁可导致系统死锁。产生这种情况的原因最常见的是递归的使用了一个自旋锁，即一个已经拥有某个自旋锁的cpu想第二次获得该自旋锁，则cpu将死锁。

3.自旋锁锁定期间不能调用可能引起进程调度的函数。如果进程获得自旋锁后再阻塞，如调用copy_from_user()、copy_to_user()、kmalloc()、msleep()等函数，则可能导致内核的崩溃。

3.2读写自旋锁

基本自旋锁对于读或写都是一视同仁的，其实访问共享资源时，多个执行单元同时读取它是不会有问题的，读写自旋锁可以允许读的并发。最多可以有一个写进程，可以有多个读进程，但是写和读不能同时进行。

1.定义和初始化读写自旋锁

rwlock_t my_rwlock = RW_LOCK_UNLOCKED; /*静态初始化*/

rwlock_t my_rwlock;

rwlock_init(&my_rwlock); /*动态初始化*/

2.读锁定

void read_lock(rwlock_t *lock);

void read_lock_irqsave(rwlock_t *lock, unsigned long flags);

void read_lock_irq(rwlock_t *lock);

void read_lock_bh(rwlock_t *lock);

3.读解锁

void read_unlock(rwlock_t *lock);

void read_unlock_irqsave(rwlock_t *lock, unsigned long flags);

void read_unlock_irq(rwlock_t *lock);

void read_unlock_bh(rwlock_t *lock);

4.写锁定

void write_lock(rwlock_t *lock);

void write_lock_irqsave(rwlock_t *lock, unsigned long flags);

void write_lock_irq(rwlock_t *lock);

void write_lock_bh(rwlock_t *lock);

void write_trylock(rwlock_t *lock);

5.写解锁

void write_unlock(rwlock_t *lock);

void write_unlock_irqsave(rwlock_t *lock, unsigned long flags);

void write_unlock_irq(rwlock_t *lock);

void write_unlock_bh(rwlock_t *lock);

示例代码：
rwlock_t lock; /*定义rwlock*/

rwlock_init(&lock); /*初始化rwlock*/

read_lock(&lock); /*读时获取锁，没有写进程在进行时，不会等待，总能获得锁*/

... /*临界区*/

read_unlock(&lock);

write_lock_irqsave(&lock,flags); /*写时获得锁，独占锁，一旦占有，其他进程的写或读都会被自旋。如果无法获得锁，则会一直自旋这儿，直到获得锁，即这里可能会写阻塞*/

...

write_unlock_irqrestore(&lock,flags);

3.3顺序锁

顺序锁(seqlock)是对读写锁的一种优化，它将使读进程不会被写进程阻塞。但是写进程之间是互斥的，读与写之间不互斥，即允许读与写同时进行，值得注意的是如果读操作期间发生了写操作，那么读进程必须重新读取数据，以确保得到的数据是完整的。顺序锁有一个限制，被保护的临界区不能含有指针，因为写操作可能使指针失效，如果读操作访问该指针则会导致oops。

1.获得写顺序锁

void write_seqlock(seqlock_t *sl);

int write_tryseqlock(seqlock_t *sl);

write_seqlock_irqsave(lock,flags)

write_seqlock_irq(lock)

write_seqlock_bh(lock)

2.释放写顺序锁

void write_sequnlock(seqlock_t *sl);

write_sequnlock_irqrestore(lock,flags)

write_sequnlock_irq(lock)

write_sequnlock_bh(lock)

3.读开始

unsigned read_seqbegin(const seqlock_t *sl); /*读操作在对被顺序锁sl保护的共享资源进行访问前调用该函数，函数返回顺序锁sl当前的顺序号*/

read_seqbegin_irqsave(lock,flags)

4.重读

int read_seqretry(const seqlock_t *sl, unsigned iv); /*读操作在访问完顺序锁sl 保护的共享资源后需要调用该函数来检查，在读访问期间是否有写操作，如果有写操作就得进行重读操作*/

read_seqretry_irqrestore(lock,iv,flags)

示例代码如下:

seqlock_t seqlock_a;

write_seqlock(&seqlock_a);

... /*写操作代码*/

write_sequnlock(&seqlock_a);

do {

seqnum = read_seqbegin(&seqlock_a);

... /*读操作代码*/

} while (read_seqretry(&seqlock_a, seqnum)); /*判等，若不等则重读*/

四.信号量

信号量与自旋锁一样都是一种数据完整性的保护机制，使用方法类似，与自旋锁相同，只有得到信号量的进程才能执行临界区代码，不同之处是当它不能获取信号量（操作权限）时，不像自旋锁那样原地等待，而是将进程的本次CPU使用权限让给其它进程（或其它程序代码）执行，类似于休眠。只有当信号释放时，系统会自动唤醒（或下一次进程切换时）将其载入CPU中继续运行。

所以信号量有类似休眠功能，其临界区代码要求比自旋锁要宽松。

4.1基本信号量

1.定义信号量
struct semaphore *sem;

2.初始化信号量
void sema_init(struct semaphore *sem, int val); /*初始化信号量sem为val*/
init_MUTEX(struct semaphore *sem) /*初始化信号量sem为1*/
init_MUTEX_LOCKED(struct semaphore *sem) /*初始化信号量sem为0*/

3.定义并初始化信号量
DECLARE_MUTEX(struct semaphore *sem) /*定义并初始化sem为1*/
DECLARE_MUTEX_LOCKED(struct semaphore *sem) /*定义并初始化sem 为0*/

4.获得信号量
void down(struct semaphore *sem); /*获得信号量sem，它会导致睡眠，因此不能再中断上下文中使用*/

int down_interruptible(struct semphore *sem); /*与down不同的是因为down而进入的睡眠状态时不能被信号打断的，但是因为down_interruptible而进入的睡眠时可以被信号打断的，信号也会导致该函数返回，这时候返回值非0*/

int down_trylock(struct semaphore *sem); /*获得信号量sem，如果能立即获得则获得信号量并且返回0，否则返回非0，它不会导致睡眠，可以再中断上下文中使用*/

5.释放信号量

void up(struct semaphore *sem); /*释放信号量sem,唤醒等待者*/

示例代码:

DECLARE_MUTEX(mount_sem);

down(&mount_sem);

...

critical section/*临界区可以包含引起阻塞的代码(进程间切换)
*/

...

up(&mount_sem);

4.2读写信号量

与读写自旋锁类似，允许N个读操作同时访问共享资源，最多只能有1个写操作，读和写不能同时进行。
1.定义和初始化读写信号量
struct rw_semaphore my_rws;
void init_rwsem(struct rw_semaphore *sem);
2.读信号量获取
void down_read(struct rw_semaphore *sem);
int down_read_trylock(struct rw_semaphore *sem);
3.读信号量释放
void up_read(struct rw_semaphore *sem);
4.写信号量获取
void down_write(struct rw_semaphore *sem);
int down_write_trylock(struct rw_semaphore *sem);
5.写信号量释放
void up_write(struct rw_semaphore *sem);
示例代码：
rw_semaphore rw_sem; /*定义读写信号量*/
init_rwsem(&rw_sem); /*初始化读写信号量*/
down_read(&rw_sem); /*读时获取信号量*/
... /*临界区*/
up_read(&rw_sem);
down_write(&rw_sem); /*写时获取信号量*/
... /*临界区*/
up_write(&rw_sem);

自旋锁和信号量的说明:

1.获得信号量的过程中可能会引起睡眠，进程上下文的切换，鉴于进程上下文的切换系统开销很大，因此只有临界区访问时间较长时，信号量才是较好的选择。同理临界区访问时间较短时，使用自旋锁。

2.为了保证信号量结构存取的原子性，在多cpu中不会使用信号量，使用自旋锁，同时单cpu抢占机制系统中也使用自旋锁。

3.信号量存在于进程上下文，临界区可以包含可能引起阻塞的代码。若自旋锁的临界区包含阻塞的代码，因为阻塞意味着要进行进程的切换，如果进程切换过去后，另一个进程企图获得本自旋锁，则会产生死锁；如果被保护的资源需要在中断或者软中断中使用，则选择自旋锁。总之，临界区中存在中断使用自旋锁，存在进程间切换或者阻塞使用信号量,例如驱动中的copy_from_user(),copy_to_user()等。

4.在Linux内核中，中断处理程序运行期间是不能发生进程切换的，因此，也就不能够使用睡眠。因为中断的内核控制路径在恢复时需要的所有数据都存放在被中断进程的内核栈中，如果发生了进程切换，那么在恢复时就找不到之前的那个进程，因为也就不能够获得那个进程的内核栈中的数据，使得中断不能正确的退出。

五.互斥体

信号量中已经使用了互斥的功能，取值为0和1时就是一种互斥体，不过内核系统函数提供了专用 互斥体。

1.定义并初始化互斥体、

struct mutex my_mutex;

mutex_init(struct mutex *mutex);

2.获取互斥体

void inline __sched mutex_lock(struct mutex *mutex);

int __sched mutex_lock_interrptible(struct mutex *mutex);

int __sched mutex_trylock(struct mutex *mutex);

3.释放互斥体

void __sched mutex_unlock(struct mutex *mutex);

示例代码：

struct mutex my_mutex;

mutex_init(&my_mutex);

mutex_lock(&my_mutex);

...

mutex_unlock(&my_mutex);