谈一谈Linux设备驱动中的各种并发控制

• 并发：多个执行单元（进程、线程、SMP对称多处理器）同时被执行，并发的这种情况就有可能会导致竞态（同时操作临界区，访问临界资源）。

• 临界区：能够被多个执行单元同时执行的代码区。

• 临界资源：能够被多个执行单元同时访问操作的资源（软件资源和硬件资源）

• 导致竞态的具体原因：

• 进程和抢占它的进程之间会有可能产生竞态；

• 进程和中断之间有可能产生竞态；

• 对称多处理器

• 解决竞态的方法：

• 同步：强调的是顺序性

• 互斥：强调的是排他性

1、原子变量

• 原子变量就是，在对其进行操作时不会被其它任务或中断打断，最简单的原子操作就是一条条的汇编指令(不包括一些伪指令，伪指令会被汇编器解释成多条汇编指令)。

• 原子操作需要硬件的支持，因此是架构相关的，其API和原子类型的定义在内核include/asm/atomic.h文件中，都是用汇编语言实现的。它的优点是使用简单，但缺点是功能单一，只能做计数操作，保护的东西太少。

• 在Linux中，原子变量对应的数据结构为 atomic_t，其的定义如下：

typedef struct {    volatile int counter;}atomic_t;

• 原子变量本质上是一个整型变量，且因为volatile的修饰，编译器不会对原子变量的值进行访问优化，以确保对它的访问都是对真实内存地址的访问，而非缓存（寄存器）。

在Linux中定义了两种原子变量操作方法，一是原子整型操作，二是原子位操作。

1.1 原子整型操作

• 定义并初始化atomic_t变量atomic_t counter = ATOMIC_INIT(0); //定义并初始化原子变量counter为0.

• 设置atomic_t变量的值atomic_set(&counter, 2); //设置原子变量counter的值为2.

• 读atomic_t变量的值var = atomic_read(&counter); // 读原子变量counter的值到var中。

• 原子变量的加减法atomic_add(2, &counter); //将原子变量counter加2.atomic_sub(2, &counter); //将原子变量counter减2.

• 原子变量的自增/自减atomic_inc(&counter); //将原子变量counter自增1atomic_dec(&counter); //将原子变量counter自减1

• 原子变量的加减测试atomic_inc_and_test(&counter); //将原子变量counter自增1,若结果为0返回真，否则返回假。atomic_dec_and_test(&counter); //将原子变量counter自减1，若结果为0返回真，否则返回假。

• 原子变量的加减返回

  atomic_sub_return(i, &counter); //将原子变量counter减i，并返回减去i后的counter。

  atomic_add_return(i,&counter); //将原子变量counter加i，并返回加上i后的counter。

1.2 原子位操作

函数原型：static inline void set_bit(int nr, volatile unsigned long *addr)

• 设置atomic_t变量的某一位set_bit(nr, &addr); //设置原子变量addr的第nr位.

• 清除atomic_t变量的某一位clear_bit(nr, &addr); //清除原子变量addr的第nr位.

• 取反atomic_t变量的某一位change_bit(nr, &addr); //取反原子变量addr的第nr位.

• 测试及设置atomic_t变量的某一位test_and_set_bit(nr, &addr); //返回原子变量addr的第nr位，然后设置该位.

• 测试及清除atomic_t变量的某一位test_and_clear_bit(nr, &addr); //返回原子变量addr的第nr位，然后清除该位.

• 测试及取反atomic_t变量的某一位test_and_change_bit(nr, &addr); //返回原子变量addr的第nr位，然后取反该位.

• 在linux中还定义了一组与原子位操作函数功能相同的函数，其函数名是在原子位操作函数名前加两个下划线。区别在于他们不会保证是一个原子操作。

2、自旋锁

自旋锁是一种阻塞结构（忙等待），这样会对系统的性能有所影响，所以不应该长时间持有，他是一种适合短时间锁定的轻量级的加锁机制，另外，自旋锁不能递归使用。自旋锁的类型也是一个结构体，即struct spinlock_t。下面对它的操作函数进行介绍：

2.1 定义和初始化自旋锁

• 定义时初始化spinlock_t lock = SPIN_LOCK_UNLOCKED; //大写字母表示的是一个初始化宏

• 动态初始化

spinlock_t  lock;
spin_lock_init(lock);

• 锁定自旋锁spin_lock(lock); //这个宏一直等待，直到获得自旋锁

• 释放自旋锁spin_unlock(lock); //这个宏立刻释放自旋锁

• 自旋锁的使用举例在驱动程序中，有些设备只允许打开一次，那么就需要一个自旋锁保护表示设备打开次数的count变量。如果不对count变量进行保护，当该设备被频繁打开的话，容易出现错误的count计数。

int count = 0;spinlock_t lock;intxxx_init(void){    ...        spin_lock_init(&lock);        ...}/* 设备打开函数 */intxxx_open(structinode*inode, structfile*filp){    ...    spin_lock(&lock);    /* 临界代码 */    if(count)    /*已经被其它程序打开过了*/    {        spin_unlock(&lock);        return-EBUSY;    }    count++;    spin_unlock(&lock);    ...}/* 设备释放函数 */intxxx_open(struct inode *inode, structfile *filp){    ...    spin_lock(&lock);    count--;    spin_unlock(&lock);    ...}

3、信号量

• Linux中实现了两种信号量，一种用于内核程序中，另一种应用于应用程序中，本文仅介绍内核中的信号量。

• 信号量与自旋锁的不同点在于：当一个进程或线程试图去获取一个已经被锁定的信号量时，它不会向自旋锁一样在原地忙等待，而是将自身加入到系统的一个等待队列中去睡眠，直到拥有信号量的进程释放该信号量后，才会被系统唤醒并再次尝试获取该信号量。此处也提醒我们，只有能够睡眠的进程（函数）才能使用信号量，像中断处理函数和可延迟函数那样需要立刻执行的函数是不能使用信号量的。

• 信号量适用于锁会被长时间持有的情况；相反，当锁只需要被短时间持有时，因为睡眠、维护等待队列以及唤醒所花费的开销可能比锁占用的全部时间表还要长，再使用信号量就不合适了。此时可能采用自旋锁更合适！

3.1 信号量的定义

在不同的实现中，信号量的实现可能不同。在Linux中其定义如下：

struct semaphore {    spinlock_t lock;    //用来对count起保护作用    unsigned int count;    struct list_head wait_list;};

• 关于count：等于0时，表示信号量正在被一个进程使用，现在不可以获取，且等待队列wait_list中没有进程在等待信号量；小于0时，代表wait_list中还有-count个进程在等待信号量；大于0时，表示可以获取该信号量。count初始化的值代表该信号量可以同时被多少进程持有。

• 关于wait_list：它是一个链表，将所有等待该信号量的正在睡眠的进程组成一个链表结构。 

3.2 信号量的原理

• 一个任务要想访问共享资源，首先必须得到信号量，获取信号量的操作将把信号量的值减1

• 若当前信号量的值为负数，表明无法获得信号量，该任务必须挂起在该信号量的等待队列等待该信号量可用；

• 若当前信号量的值为非负数，表示能获得信号量，因而能即时访问被该信号量保护的共享资源。

• 当任务访问完被信号量保护的共享资源后，必须释放信号量，释放信号量通过把信号量的值加1实现

• 如果信号量的值为非正数，表明有任务等待当前信号量，因此他也唤醒所有等待该信号量的任务。

3.3 信号量的接口

3.3.1 定义

• 内核信号量是struct semaphore类型的对象，在内核源码中位于include\linux\semaphore.h文件

struct semaphore{    raw_spinlock_t lock;        unsigned int count;        struct list_head wait_list;}

• 大于0，资源空闲；

• 等于0，资源忙，但没有进程等待这个保护的资源；

• 小于0，资源不可用，并至少有一个进程等待资源

• count：相当于信号量的值，

• wait_list：内核链表，当前未获得信号量的任务会被注册到该等待链表中

3.3.2 初始化

• 静态定义并初始化一个信号量DEFINE_SEMAPHORE(name);——定义一个信号量name，并赋初值1

• 动态初始化一个信号量

• sema_init(&sema, val);——初始化信号量sema为val

当sema中的count为1时，我们称为互斥体（同一时间仅有一个进程持有该信号量），他有专门的宏来进行初始化：init_MUTEX(struct semaphore *sema); //初始化sema信号量为1。init_MUTEX_LOCKED(struct semaphore *sema);//初始化sema信号量为0。

3.3.3 获取信号量

• void down(struct semaphore * sem);

• 该函数将把sem的值减1，如果信号量sem的值非负，就直接返回，否则调用者将被挂起，直到别的任务释放该信号量才能继续运行，故不能用于中断上下文中（包括IRQ上下文和softirq上下文）；

• int down_interruptible(struct semaphore * sem);

• 该函数功能和down类似，不同之处为，down不会被信号（signal）打断，但down_interruptible能被信号打断，因此该函数用返回值来区分是正常返回还是被信号中断（如果返回0，表示获得信号量正常返回，如果被信号打断，返回-EINTR）。

• 使用可被中断的信号量版本的意思是，万一出现了semaphore的死锁，还有机会用ctrl+c发出软中断，让等待这个内核驱动返回的用户态进程退出。而不是把整个系统都锁住了。在休眠时，能被中断信号终止，这个进程是可以接受中断信号的！

• int down_trylock(struct semaphore * sem);

• 该函数试着获得信号量sem，如果能够即时获得，他就获得该信号量并返回0，否则，表示不能获得信号量sem，返回值为非0值。因此，他不会导致调用者睡眠，能在中断上下文使用。

• int down_killable(struct semaphore *sem);

• int down_timeout(struct semaphore *sem, long jiffies);

• int down_timeout_interruptible(struct semaphore *sem, long jiffies);

3.3.4 释放信号量

• void up(struct semaphore * sem);

• 该函数释放信号量sem，即把sem的值加1，如果sem的值为非正数，表明有任务等待该信号量，因此唤醒这些等待者。

3.3.5 信号量的使用场景

• 信号量用于保护临界资源

//定义一个信号量struct semaphoresema;//初始化一个二值信号量intxxx_init(void){    ...    init_MUTEX(&sema);    ...}intxxx_open(struct inode *inode, structfile* filp){    ...        //获取一个信号量    if(down_interruptible(&sema))        {                return -ERESATRTSYS;        }    ...    return0;}intxxx_release(struct inode *inode, structfile *filp){    ...    //获取一个信号量    up(&sema);    ...    return0;}

• 信号量用于同步当信号量被初始化为0时，可以用来保证两个进程的执行顺序，如下图所示：

• 总结：信号了用于在多个进程之间互斥，但信号量的执行会引起进程的睡眠，而睡眠需要进程上下文的切换，是很耗费时间的。所以，只有在一个进程对其保护的资源占用时间要比进程切换的时间长很多时，才划算！

4、互斥体

• mutex用于互斥操作。信号量的count初始化为1，down()/up()也可以实现类似mutex的作用。mutex的语义相对于信号量要简单轻便一些，在锁争用激烈的测试场景下，mutex比信号量执行速度更快，可扩展性更好，另外mutex数据结构的定义比信号量小。所以，除非mutex的某个约束妨碍你使用，否则相比信号量要优先使用mutex。

• mutex的使用注意事项

• 同一时刻只有一个线程可以持有mutex。

• 只有锁持有者可以解锁。也就是说不能在进程间同步（A进程中持有mutex，而在B进程中释放mutex），也不能在内核空间和用户空间进行同步，所以一般是在同一上下文中上锁和解锁。

• 不允许递归地加锁和解锁（不能重复持有和重复解锁），也不能被重新初始化，其所使用的内存区域更是不能被释放的。

• 当进程持有mutex时，进程不可以退出。

• mutex必须使用官方API来初始化，而不能手动赋值、使用memset和重复初始化。

• mutex可以睡眠，所以不允许用在中断上下文中，例如tasklet、定时器等。

4.1互斥体的接口

• 初始化

• 静态初始化——DEFINE_MUTEX(name);或DEFINE_MUTEX_LOCKED(name);

• 动态初始化——mutex_init(&mutex);

• 互斥锁的申请和释放

• mutex_lock(struct mutex*)——为指定的mutex上锁，如果不可用则进入不可中断睡眠；

• mutex_unlock(struct mutex*)——为指定的mutex解锁；

• mutex_lock_interruptible(struct mutex*)——为指定的mutex上锁，如果不可用则进入可中断睡眠，如果在睡眠时被信号打断，则返回_EINIR；

• mutex_unlock_interruptible(struct mutex*)——为指定的mutex解锁；

• mutex_trylock(struct mutex*)——非阻塞的获取mutex锁，成功返回1，否则返回0；

• mutex_is_lock(struct mutex*)——判断锁是否可用，不可用返回1，否则返回0。

• 使用示例

struct mutex mutex;    //定义mutex_init(&mutex);    //初始化mutex_lock(&mutex);    //加锁...                    //临界区mutex_unlock(&mutex);  //解锁

4.2互斥体和自旋锁的使用比较

• 在中断上下文中只能使用自旋锁，而在任务可能睡眠时只能使用互斥体或信号量。

• 短期锁定时使用自旋锁，长期锁定使用互斥体或信号量；

• 多进程同步使用信号量。

5、读写自旋锁

• 因为自旋锁毫无线程并发性可言，使得多处理器系统的性能受到限制。通过观察线程在临界区的访问行为，发现有些线程只是简单地读取信息，并不修改任何东西，所以即使它们同时进入临界区也不会有任何危险，反而能大大提高系统的并发性。这种将线程区分为读者和写者、多个读者允许同时访问共享资源、申请线程在等待期内依然使用忙等待方式的锁，我们称之为读写自旋锁（Reader-Writer Spinlock）。

• 读/写自旋锁是在保护SMP体系下的共享数据结构而引入的，只要没有对数据结构进行修改，读/写自旋锁就允许多个线程同时读同一数据结构。但如果一个线程想对这个结构进行写操作，那么它必须首先获取读/写锁的写锁，写锁授权独占访问这个资源。

5.1读写自旋锁的特点

• 互斥：任意时刻读者和写者不能同时访问共享资源（即获得锁），即任意时刻只能有至多一个写者访问共享资源。

• 读者并发：在满足“互斥”的前提下，多个读者可以同时访问共享资源。

• 无死锁（Freedom from Deadlock）：如果线程 A 试图获取锁，那么某个线程必将获得锁，这个线程可能是 A 自己；如果线程 A 试图但是却永远没有获得锁，那么某个或某些线程必定无限次地获得锁。意思就是，从全局来看一定会有申请线程获得锁，但对于某个或某些申请线程而言，它们可能永远无法获得锁，这种现象也称为饥饿（Starvation）。一种原因源于计算机体系结构的特点，一种原因源于设计策略（读者优先、写者优先、）。

• 忙等待：申请锁的线程必须不断地查询是否发生退出等待的事件，而不能进入睡眠状态。

5.2读写锁的接口

• 初始化

• 动态初始化——rwlock_init (&rw_lock);

• 静态初始化——DEFINE_RWLOCK(rwlock);

• 读临界区

• read_lock(rw_lock);

• read_lock_irqsave(rw_lock，flags);= read_lock（） + local_irq_save（）

• read_lock_irq(rw_lock); = read_lock（） + local_irq_disable（）

• read_lock_bh(rw_lock);= read_lock（） + local_bh_disable（）

• read_unlock(rw_lock);

• read_unlock_irqstore(rw_lock，flags);read_unlock（）+ local_irq_restore（）

• read_unlock_irq(rw_lock);= read_unlock（）+ local_irq_enable（）

• read_unlock_bh(rw_lock);= read_unlock（）+ local_bh_enable（）

• 写临界区

• write_lock(rw_lock);

• write_lock_irqsave(rw_lock, flags);= write_lock（） + local_irq_save（）

• write_lock_irq(rw_lock);= write_lock（） + local_irq_disable（）

• write_lock_bh(rw_lock);= write_lock（） + local_bh_disable（）

• write_trylock(rw_lock);

• write_unlock(rw_lock);

• write_unlock_irqsave(rw_lock, flags);= write_unlock（）+ local_irq_restore（）

• write_unlock_irq(rw_lock);= write_unlock（）+ local_irq_enable（）

• write_unlock_bh(rw_lock);= write_unlock（）+ local_bh_enable（）

可通过如下函数禁用所有处理器上的中断:

disable_irq(void);——在2.6内核之后才有这个函数,之前是没有方法全局禁用整个系统的所有中断。

可通过如下函数禁用当前处理器上的中断:

local_irq_disable(void);—— 仅仅是禁用当前CPU的中断;

local_irq_save(unsigned long flags);——把当前中断状态保存到flags中，然后禁用当前处理器上的中断；

可通过如下函数打开当前处理器上的中断:

void local_irq_enable(void);——将local_irq_save保存的flags状态值恢复后，打开中断。

void local_irq_restore(unsigned long flags);——无条件打开中断。

可通过如下函数判断当前进程是否在硬件中断上下文中:

in_irq(void);

可通过如下函数判断当前进程是否处于中断上下文(包括中断底半部和硬件中断处理过程):

in_interrupt(void);

5.3读写锁的使用示例

rwlock_t lock; /* 定义 rwlock */rwlock_init(&lock); /* 初始化 rwlock *//* 读时获取锁 */read_lock(&lock);... /* 临界资源 */read_unlock(&lock);/* 写时获取锁 */write_lock_irqsave(&lock, flags);... /* 临界资源 */    write_unlock_irqrestore(&lock, flags);

• 更多关于读写锁的内核实现细节参考读写自旋锁详解DarkMagician_Potter的博客-CSDN博客读写自旋锁

6、完成量

上节中讲的进程间的同步（一个线程等待另一个线程完成某操作后才能继续执行），在Linux中有专门的机制（虽然使用信号量也可以实现）叫完成量，即一个线程发送一个信号通知另一个线程开始完成某个任务。

6.1完成量的定义

struct completion{    unsigned int done;    wait_queue_head_t wait;};

done：维护一个计数，其被初始化为1。当done为0时，会将拥有完成量的线程置于等待状态；当其值大于1时，表示等待完成量的函数可以立刻执行！wait：存放所有等待该完成量的正在睡眠的进程组成的链表

6.2完成量的接口

• 定义一个完成量struct completion com;

• 初始化一个完成量init_completion(&com);//将done设置为0

• 定义并初始化一个完成量DECLARE_COMPLETION(com);

• 等待完成量wait_for_completion(&com);//线程将一直等待，且不会被中断打断。

• 释放完成量complete(&com);//只唤醒一个等待的进程complete_all(&com);//唤醒所有等待的进程

• 完成量的使用

struct completioncom;intxxx_init(void){    ...    init_completion(&com);    ...}intxxx_A(void){    ...    /* 代码1 */    wait_for_completion(&com);    /* 代码3 */    return0;}intxxx_B(void){    ...    /* 代码2 */    complete(&com);    ...    return0;}

初始化后com中的done值为0，此时若xxx_A先执行，则当执行完成代码1 后便会进入睡眠，等待进程xxx_B执行完代码2，并释放完成量（使done加1），此时系统会唤醒处于完成量com中的等待队列里正在睡觉的进程A，然后进程A继续执行完代码3.