如果你是一个嵌入式Linux开发者，你可能会遇到这样的问题：如何在多个任务或线程之间安全地共享设备资源？如何避免数据的竞争和不一致？如何提高系统的性能和可靠性？这些问题都涉及到并发控制技术，也就是如何协调多个执行实体对共享资源的访问。在本文中，我们将介绍Linux驱动中常用的并发控制技术，包括原子操作，自旋锁，信号量，互斥锁，读写锁，顺序锁和RCU等，并举例说明它们的使用方法和注意事项。
为了实现对临界资源的有效管理，应用层的程序有原子变量，条件变量，信号量来控制并发，同样的问题也存在与驱动开发中，比如一个驱动同时被多个应用层程序调用，此时驱动中的全局变量会同时属于多个应用层进程的进程空间，这种情况下也要使用一些技术来实现对并发的控制。本文将讨论内核中下述并发控制技术的技术特点和应用场景。
顾名思义，就是屏蔽所有的中断。在嵌入式系统，中断屏蔽可以有三级，1. 硬件接口的屏蔽，2. 硬件GIC的屏蔽，3. CPU(内核)的屏蔽。如果在接口处屏蔽了，那么中断来了就丢了，根本找不到。如果在GIC处屏蔽了，那么在屏蔽期间如果来了irq_1,irq_2,irq_3个中断，因为只有一个pending标志位，所以最后irq_3来的时候会将pending置位，之后解除屏蔽了，CPU发现pending有置位，还是会处理，但是1,2就肯定丢了。在ARM处的屏蔽，即内核中的屏蔽，看怎么设置了，如果就是local_irq_disable，那么丢了就是丢了，和在接口处屏蔽一样，如果是local_irq_save就和第二种一样，追到最后一个中断，内核也有相应的机制进行中断计数，知道这期间来了多少个中断，但是实际操作中，大部分情况我们都不会追着执行错过的中断，除非这个中断非常重要。
我们这里讨论的，就是在内核中进行中断屏蔽。由于内核中很多重要的操作都要依赖于中断，所以屏蔽所有的中断是十分危险的，里面执行的代码要尽可能的快，而且，由于内核的进程调度也是由中断驱动的，所以中断屏蔽中不能有可能引发休眠的代码，否则无法被唤醒。注意，中断屏蔽只是屏蔽了本CPU的中断，所以并不能解决SMP引发的竞泰问题，通常，中断屏蔽要和自旋锁联合使用，用于防止访问自旋锁保护的临界区时被中断打断
local_irq_disable(); //屏蔽中断 //或 local_irq_save(flags); //屏蔽中断并保存目前CPU中的中断位信息 /* 临界区 */ local_irq_enable(); //解除屏蔽 //或 local_irq_restore(flags); //解除屏蔽并恢复中断位信息
local_bh_disable(); //屏蔽中断 /* 临界区 */ local_bh_enable();
原子操作即不能被打断的操作，和应用层的概念一样，内核中的原子操作模板如下：
//asm/atomic.h //创建并初始化原子变量 atomic_t tv = ATOMIC_INIT(初值); //读原子变量 int atomic_read(atomic_t *v); //写原子变量 void atomic_set(atomic_t *v, int i); /** *atomic_dec_and_test - 尝试将原子变量-1 *v:如果-1之后原子变量变为0，返回非0, 否则返回0 */ int atomic_dec_and_test(volatile atomic_t *v); int atomic_inc_and_test(volatile atomic_t *v); int atomic_sub_and_test(int i, volatile atomic_t *v); //操作并返回 int atomic_add_return(int i, atomic *v); int atomic_sub_return(int i, atomic *v); int atomic_inc_return(atomic *v); int atomic_dev_return(atomic *v);
static atomic_t tv;
static int demo_open(struct inode *inode, struct file *filp)
{
if(!atomic_dec_and_test(&tv)){
atomic_inc(&tv);
return -EBUSY;
}
/* 操作代码 */
return 0;
}
static int demo_release(struct inode *inode, struct file *filp)
{
atomic_inc(&tv);
return 0;
}
static int __init demo_init(void)
{
// init atomic_t
atomic_set(&tv, 1);
}
位原子操作即原子的位操作，内核中大量使用”位”来记录信息，比如位图，这些操作都必须是原子性的，内核API如下：
//设置位 void set_bit(nr,void *addr); //清除位 void clear_bit(nr,void *addr); //改变位 void change_bit(nr,void *addr); //测试位 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);
意即”在原地打转”，当加锁不成功时，自旋，自旋锁会不断的占用CPU进行变量的测试，由于属于原子操作，所以该CPU的占用会升为100%，所以，使用自旋锁时，临界区的代码需要很短，否则会影响系统性能，此外，作为锁机制的一种，使用自旋锁同样需要注意死锁的出现，自旋锁锁定期间不能调用可能引起进程调度的函数，如果进程获得自旋锁之后再阻塞，eg，copy_from_user(),copy_to_user(),kmalloc()，msleep()等，一旦阻塞发生就可能导致内核 崩溃。
自旋锁可以用来解决SMP竞态问题。不同类型的自旋锁有自己的处理机制，适用于不同的情况。包括传统自旋锁，读写自旋锁，RCU机制，顺序锁等，自旋锁是信号量，互斥体的的底层实现工具。
| 比较类型 | 传统自旋锁 | 读写自旋锁 | 顺序锁 | RCU机制 |
|---|---|---|---|---|
| 应用场合 | 需要上锁者独占的资源 | 需要写者独占的资源 | 很少同时读写的资源 | 读多写少的资源 |
| 读+读 并发 | × | √ | √ | √ |
| 读+写 并发 | × | × | √ | √ |
| 写+写 并发 | × | × | × | √ |
和其他锁机制的一样，使用自旋锁保护数据分为抢锁-操作-解锁,下面就是一个典型的使用锁的流程，通过自旋锁实现一个文件只被一个进程打开。
int cnt=0;
lock_t lock;
static int my_open()
{
lock(&lock);
if(cnt){
unlock(&lock)
}
cnt++;
unlock(&lock);
}
static int release()
{
lock(&lock);
cnt--;
unlock(&lock);
}
是一种比较粗暴的自旋锁，使用这种锁的时候，被锁定的临界区域不允许其他CPU访问，需要注意的是，尽管获得锁之后执行的临界区操作不会被其他CPU和本CPU内其他抢占进程的打扰，但是仍然会被中断和底半部的影响，所以通常我们会使用下述API中的衍生版本，比如上文中提到的将自旋锁+中断屏蔽来防止使用自旋锁访问临界资源的时候被中断打断，对应的宏函数就是spin_lock_irq和spin_lock_irqsave。
//定义并初始化自旋锁 spinlock_t spinlock void spin_lock_init(spinlock_t *); //加锁 //spin_lock - 加锁函数（忙等） void spin_lock(spinlock_t *lock); int spin_trylock(spinlock_t *lock); spin_lock_irq(); //=spin_lock() + local_irq_disable() spin_lock_irqsave(); //= spin_lock() + lock_irq_save(); spin_lock_bh(); //=spin_lock() + local_bh_disable(); //解锁 void spin_unlock(spinlock_t *lock); spin_unlock_irq(); //=spin_unlock() + local_irq_enable() spin_unlock_irqrestore(); //= spin_unlock() + lock_irq_restore(); spin_unlock_bh(); //=spin_unlock() + local_bh_enable();
传统的自旋锁粗暴的将临界资源划归给一个CPU，但是很多资源都不会因为读而被破坏，所以我们可以允许多个CPU同时读临界资源，但不允许同时写资源，类似于应用层的文件锁，内核的读写锁机制同样有下述互斥原则:
//include/linux/rwlock.h //定义并初始化自旋锁 rwlock_t rwlock; void rwlock_init(rwlock_t *lock); //加读锁 void read_lock(rwlock_t *lock); int read_trylock(rwlock_t *lock); void read_lock_irqsave(rwlock_t *lock,unsigned long flags); void read_lock_irq(rwlock_t *lock, unsigned long flags); void read_lock_bh(rwlock_t *lock); //解读锁 void read_unlock(rwlock_t *lock) void read_unlock_irqrestrore(rwlock_t *lock,unsigned long flags); void read_unlock_irq(rwlock_t *lock, unsigned long flags); void read_unlock_bh(rwlock_t *lock); //加写锁 void write_lock(rwlock_t *lock) int write_trylock(rwlock_t *lock) void write_lock_irqsave(rwlock_t *lock,unsigned long flags); void write_lock_irq(rwlock_t *lock, unsigned long flags); void write_lock_bh(rwlock_t *lock); //解写锁 void write_unlock(rwlock_t *lock) void write_unlock_irqrestrore(rwlock_t *lock,unsigned long flags); void write_unlock_irq(rwlock_t *lock, unsigned long flags); void write_unlock_bh(rwlock_t *lock);
顺序锁可以看作读写锁的升级版，读写锁不允许同时存在读者和写者，顺序锁对这一情况进行了改良，它允许写者和读者同时访问临界区，不必再向读写锁那样读者要读必须等待写者写完，写者要写必须等待读者读完。不过，使用顺序锁的时候，临界区不能有指针，因为写者可能会修改指针的指向，如果读者去读，就会Oops，此外，如果读者读过的数据被写者改变了，读者需要重新读，以维持数据是最新的，虽然有这两个约束，但是顺序锁提供了比读写锁更大的灵活性。对于写者+写者的情况，顺序锁的机制和读写锁一样，必须等！
//include/linux/seqlock.h //定义顺序锁 struct seqlock_t sl; //获取顺序锁 void write_seqlock(seqlock_t *sl); void write_tryseqlock(seqlock_t *sl); void write_seqlock_irqsave(lock,flags); //=local_irq_save() + write_seqlock() void write_seqlock_irq(seqlock_t *sl); //=local_irq_disable() + write_seqlock() void write_seqlock_bh(seqlock_t *sl); //local_bh_disable() + write_seqlock() //释放顺序锁 void write_sequnlock(seqlock_t *sl); void write_sequnlock_irqsave(lock,flags); //=local_irq_restore() + write_sequnlock() void write_sequnlock_irq(seqlock_t *sl); //=local_irq_enable() + write_sequnlock() void write_sequnlock_bh(seqlock_t *sl); //local_bh_enable() + write_sequnlock() //读开始 unsigned read_seqbegin(const seqlock_t *sl); read_seqbegin_irqsave(lock,flags); //=local_irq_save() + read_seqbegin(); //重读 int read_seqretry(const seqlock_t *sl,unsigned iv); read_seqretry_irqrestore(lock,iv,flags); //=local_irq_restore() + read_seqretry();
RCU即Read-Copy Update，即读者直接读，写者先拷贝再择时更新，是另外一种读写锁的升级版，这种机制在VFS层被大量使用。正如其名，读者访问临界资源不需要锁，从下面的rcu_read_lock的定义即可看出，写者在写之前先将临界资源进行备份，去修改这个副本，等所有的CPU都退出对这块临界区的引用后，再通过回调机制，将引用这块资源的原指针指向已经修改的备份。从中可以看出，在RCU机制下，读者的开销大大降低，也没有顺序锁的指针问题，但是写者的开销很大，所以RCU适用于读多写少的临界资源。如果写操作很多，就有可能将读操作节约的性能抵消掉，得不偿失。
内核会为每一个CPU维护两个数据结构-rcu_data和rcu_bh_data，他们用于保存回调函数，函数call_rcu()把回调函数注册到rcu_data，而call_rcu_bh()则把回调函数注册到rcu_bh_data，在每一个数据结构上，回调函数们会组成一个队列。
使用RCU时，读执行单元必须提供一个信号给写执行单元以便写执行单元能够确定数据可以被安全地释放或修改的时机。内核中有一个专门的垃圾收集器来探测读执行单元的信号，一旦所有的读执行单元都已经发送信号告诉收集器自己都没有使用RCU的数据结构，收集器就调用回调函数完成最后的数据释放或修改操作。
读即RCU中的R，从下面的宏定义可以看出，读操作其实就是禁止内核的抢占调度，并没有使用一个锁对象。
//读锁定 //include/linux/rcupdate.h rcu_read_lock(); //preempt_disbale() rcu_read_lock_bh(); //local_bh_disable() //读解锁 rcu_read_unlock() //preempt_enable() rcu_read_unlock_bh(); //local_bh_enable()
同步即是RCU写操作的最后一个步骤-Update，下面这个接口会则色写执行单元，直到所有的读执行单元已经完成读执行单元临界区，写执行单元才可以继续下一步操作。如果有多个RCU写执行单元调用该函数，他们将在一个grace period(即所有的读执行单元已经完成对临界区的访问)之后全部被唤醒。
synchrosize_rcu()
下面这个接口也由RCU写执行单元调用，它不会使写执行单元阻塞，因而可以在中断上下文或软中断中使用，该函数把func挂接到RCU回调函数链上，然后立即返回。函数sychrosize_rcu()其实也会调用call_rcu()。
void call_rcu(struct rcu_head *head,void (*func)(struct rcu_head *rcu));
下面这个接口会将软中断的完成也当作经历一个quiecent state(静默状态)，因此如果写执行单元调用了该函数，在进程上下文的读执行单元必须使用rcu_read_lock_bh();
void call_rcu_bh(struct rcu_head *head,void (*func)(struct rcu_head *rcu));
RCU机制被大量的运用在内核链表的读写中，下面这些就是内核中使用RCU机制保护的数据结构，函数的功能和非RCU版本一样，具体可以参考内核文件**”include/linux/list.h”**，只不过这里的操作都会使用RCU保护起来。
void list_add_rcu(struct list_head *new, struct list_head *head); void list_add_tail_rcu(struct list_head *new,struct list_head *head); void list_del_rcu(struct list_head *entry); void list_replace_rcu(struct list_head *old,struct list_head *new); list_for_each_rcu(pos,head); list_for_each_safe_rcu(pos,n,head); list_for_each_entry_rcu(pos,head,member); void hlist_del_rcu(struct hlist_node *n); void hlist_add_head_rcu(struct hlist_node *n, struct hlist_head *h); list_for_each_rcu(pos,head); hlist_for_each_entry_rcu(tpos,pos,head,member);
自旋锁一节提过，如果一个CPU不能获取临界资源，就会造成”原地自旋”，所以自旋锁保护的临界区的执行时间不能太长，但如果我们的确需要保护一段执行时间比较长的临界区呢？答案就是信号量
，信号量的底层依托于自旋锁来实现其原子性，并进一步将其提高到”进程”的维度，称为一种可以运行在进程上下文的”锁”，正是这种能运行在进程上下文的能力赋予了信号量和自旋锁的诸多不同。
使用信号量，如果试图获取信号量的进程获取失败，内核就会将其调度为睡眠状态，执行其他进程，避免了CPU的忙等。不过，进程上下文的切换也是有成本的，所以通常，信号量在内核中都是只用于保护大块临界区。
此外，一个进程一旦无法获取期待的信号量就会进入睡眠，所以信号量保护的临界区可以有睡眠的代码。在这方面，自旋锁保护的区域是不能睡眠也不能执行schedule()的，因为一旦一个抢到了锁的CPU进行了进程上下文的切换或睡眠，那么其他等待这个自旋锁的CPU就会一直在那忙等，就永远无法等到这个锁，，形成死锁，除非有其他进程将其唤醒(通常都不会有)。
也正是由于信号量操作可能引起阻塞，所以信号量不能用于中断上下文。总结一下刚才罗嗦这一段：
| 项目 | 信号量 | 自旋锁 |
|---|---|---|
| 临界区时间 | 进程切换时间更短 | 临界区执行时间更短 |
| 进程上下文 | 临界区可以睡眠或调度 | 临界区不可以睡眠或调度 |
| 中断上下文 | 只有down_trylock()可以 | 可以 |
内核的信号量和应用层的信号量的使用方式类似，但没有获取信号量这一步骤，因为内核中中的信号量可以映射到所有调用这个模块的用户进程的内核空间。这些用户进程也就直接共享了一个信号量，所以也就没有获取信号量一说，相关的内容我在”Linux IPC System V 信号量”一文中有所讨论。
和应用层的信号量一样，内核信号量也是用于对临界资源的互斥/顺序访问，同样，虽然在使用信号量的时候我们可以初始化为任意值，但实际操作上我们通常只初始化为1或0，下述是Linux内核提供的信号量API。
//include/linux/semaphore.h //定义并初始化semaphore对象 struct semphore sem; //初始化信号量 void sem_init(struct semaphore * sem,int val); init_MUTEX(sem); init_MUTEX_LOCKED(sem); DECLARE_MUTEX(sem); DECLARE_MUTEX_LOCKED(sem); //P操作 //down()会导致睡眠，不能用于中断上下文 void down(struct semaphore *sem); //down_interruptible同样会进入休眠，但能被打断 int down_interruptible(struct semaphore *sem); //down_trylock不能获得锁时会立即返回，不会睡眠，可以用在中断上下文 int down_trylock(struct semaphore *sem); //V操作 void up(struct semaphore *sem);
读写信号量与信号量的关系 和 读写自旋锁与自旋锁的关系类似，他们的互斥逻辑都是一样的，这里不再赘述
//定义并初始化读写信号量 struct rw_semaphore my_rwsem; void init_rwsem(struct rw_semaphore *sem); //P读信号量 void down_read(struct rw_semaphore *sem); int down_read_trylock(struct rw_semaphore *sem); //V读信号量 void up_read(struct rw_semaphore *sem); //P写信号量 void down_write(struct rw_semaphore *sem); int down_write_trylock(struct rw_semaphore *sem); //V写信号量 void up_write(struct rw_semaphore *sem);
struct rw_semaphore my_rwsem; void init_rwsem(&my_rwsem); //读前获取读信号量 down_read(&my_rwsem); //若要非阻塞：down_read_trylock(&my_rwsem); /* 读临界区 */ //读完释放读信号量 up_read(&my_rwsem); //写前获取写信号量 down_write(&my_rwsem); //若要非阻塞：down_write_trylock(&my_rwsem); /* 写临界区 */ //写完释放写信号量 up_write(&my_rwsem);
完成量用于一个执行单元等待另一个执行单元执行完某事，和传统信号量一样，主要是用来实现队临界区的顺序/互斥访问。但是完成量还提供一种唤醒一个或唤醒所有等待进程的接口，有点类似与应用层的条件变量。
//定义并初始化完成量 struct completion my_completion; init_completion(&my_completion); //或 DECLARE_COMPLETION(my_completion) //等待completion void wait_for_completion(struct completion *c); //唤醒completion void complete(struct completion *c); //只唤醒一个等待的执行单元 void complete_all(struct completion *c); //释放所有等待该完成量的执行单元
除了信号量，Linux内核还提供了一种专门用于实现互斥的机制-互斥体，相关的内核API如下：
//include/linux/mutex.h //定义并初始化mutex对象 struct mutex my_mutex; mutex_init(&my_mutex); //获取mutex void mutex_lock(struct mutex *lock); int mutex_trylock(struct mutex *lock); int mutex_lock_interruptible(struct mutex *lock); //释放mutex void mutex_unlock(struct mutex *lock);
通过本文，我们了解了Linux驱动中的并发控制技术，它们各有优缺点和适用场景。我们应该根据实际需求选择合适的技术，并遵循一些基本原则，如最小化临界区，避免死锁，减少上下文切换等。并发控制技术是Linux驱动开发中不可或缺的一部分，它可以保证设备资源的正确性和高效性，也可以提升系统的稳定性和可扩展性。希望本文能够对你有所帮助和启发。
以上是Linux驱动中的并发控制技术：原理与实践的详细内容。更多信息请关注PHP中文网其他相关文章！
