Linux驱动--并发和竞争

概述： 并发和竞争，对于驱动来说，就是对临界区资源的保护。在面对多核CPU，很多进程同时运行，同时访问公共设备和数据，要保证这么多进程有序进行访问。

以下，将主要从用法上归纳总结linux驱动对并发和竞争处理提供的接口。

接口类别：

旗标（semaphore 也可以说信号量）Completions机制自旋锁（spinlock）顺序锁（seqlock）原子变量（atomic）原子位操作（bitops）RCU（read-copy-update:读取-拷贝-更新）

各种接口类别应用的场景：

旗标（semaphore） 保护临界区资源，同一时刻只能有一个进程获取旗标，可以访问保护的临界区，一个进程未释放旗标，另外的进程试图访问临界区，将不能获取旗标，会睡眠等待释放旗标，多个进程访问未释放旗标的临界区，这些进程将排队等待并睡眠，直到旗标释放，一个一个唤起。主要应用在进程上下文环境中，可以睡眠。不能用在不可睡眠的地方，如中断服务程序。

Completions 一个动作，需要等待另外一个动作完成，才能继续下去。这种场景用completions，虽然用旗标也可以做到，但是completions在这种场景要比旗标好，性能上和各种极端情况都要好。completions等待另外动作完成的进程，是不杀死的。completions典型应用是模块退出和线程退出时一起退出。

自旋锁（spinlock） 自旋锁和旗标功能一样，保护临界区。自旋锁保护的临界区code是原子性的，在获取锁时会关闭中断。因此，自旋锁可以用在不可睡眠的地方，如中断服务程序，不可用在可以睡眠的方。自旋锁的原子性，要求临界区的code不能睡眠，否则会死锁。

顺序锁（seqlock） 顺序锁(seqlock)是对读写锁的一种优化,提高了读锁和写锁的独立性。写锁不会被读锁阻塞，读锁也不会被写锁阻塞。写锁会被写锁阻塞。 如果读执行单元在读操作期间,写执行单元已经发生了写操作,那么,读执行单元必须重新去读数据,以便确保读到的数据是完整的;这种锁在读写操作同时进行的概率比较小,性能是非常好的,而且它允许读写操作同时进行,因而更大地提高了并发性。 顺序锁有一个限制:它必须要求被保护的共享资源中不能含有指针;因为写执行单元可能会使指针失效,当读执行单元如果正要访问该指针时,系统就会崩溃。 更加详细的可以参考这篇博文： Linux并发控制——顺序锁（seqlock）

原子变量（atomic） 原子变量，主要用于对保护的数据是一个简单的整数时用，用其他的又浪费（旗标，锁等对系统性能影响较大）。

原子位操作（bitops） 与原子变量一样，对于位操作用bitops比较快，而且对系统影响小，还是原子性的。

RCU RCU是用来应对读写场景，而且主要用来保护由指针指向的对象（而不保护指针本身），读不阻塞，读操作的优先权大于写操作（与seqlock相反），写的时候不阻塞读，写的时候发现在读，要等待读完成，才销毁指针指向旧地址空间数据，然后更新指针指向新的地址。 应用在读者与写者场景，目前性能最好（旗标，自旋锁都有实现读者写者的专用接口，当然顺序锁也是应用在这个场景，但性能都不及RCU）。 RCU可以参考如下博客: Linux并发控制——RCU linux内核 RCU机制详解 Linux 2.6内核中新的锁机制–RCU

加锁原则： （个人的理解，不当之处望大家指出） 优先考虑从上往下 1、能不加锁尽量不加锁，能用算法代替尽量用算法代替，锁的开销对系统效率影响大，来自网上的图片：

2、能用原子变量和原子位操作尽量用

3、加锁遵循下面规则：

模糊的规则和顺序的规则，有资料这么说，看描述大体总结下： （1）同一进程，不能同一时刻，两次获取同一把锁，否则将导致死锁。 （2）获取一把锁，必须释放后，才能获取其他锁，即尽量不要嵌套，否则，稍微不注意就死锁。 （3）用锁时，一定要清晰明确。 上面的规则要尽量遵循，当然锁是可以嵌套，但是嵌套逻辑复杂，除非很清楚。如遇到的，自旋锁嵌套，外层的锁，会被嵌套自旋锁，释放时，打开中断，导致外层锁不在具有原子性。

4、细-粗-粒度加锁 在考虑加锁时，尽量粗粒度。如一个驱动模块能用一把锁尽量用一把，绝对不用两把。

读者写者选择: （也可以说是生产者消费者） 1、都有基于旗标和自旋锁实现的读者写者，都分别继承各种的优势，要根据场景来选择。多个进程读不阻塞，多个进程写会阻塞，在写的时候，有读，此时读会被阻塞 2、顺序锁用于保护非指针数据。基于上面，顺序锁的写不会阻塞读，发现被修改，会重读，其他都和上面相同。 3、RCU性能最好，尽量考虑用。读不阻塞，读操作的优先权大于写操作（与seqlock相反），写的时候不阻塞读，写的时候发现在读，要等待读完成，才销毁指针指向旧地址空间数据，然后更新指针指向新的地址。

总之，总的原则来说，可以不用锁就不用，非用不可，需根据场景来选择不同的锁，保证设计思路清晰。对于读者写者，读多写越少，各接口性能就越好，写越多性能越不好。

函数接口：

旗标（semaphore） 定义头文件： #include <linux/semaphore.h>

首先，定义和初始化： 静态的用宏：

#define DEFINE_SEMAPHORE(name) \ struct semaphore name = __SEMAPHORE_INITIALIZER(name, 1)

动态的先定义semaphore指针变量，然后调用inline void sema_init(struct semaphore *sem, int val)，第2个参数一般为1。

函数：

extern void down(struct semaphore *sem); extern int __must_check down_interruptible(struct semaphore *sem); extern int __must_check down_killable(struct semaphore *sem); extern int __must_check down_trylock(struct semaphore *sem); extern int __must_check down_timeout(struct semaphore *sem, long jiffies); extern void up(struct semaphore *sem);

down()—–加锁，得不到锁的进程，将在此处睡眠等待，直到等到锁，中途睡眠不可打断。得到锁，down()返回继续往下执行。 用法：

DEFINE_SEMAPHORE(sem) void exampSem() { down(&sem); ... ... ... up(&sem) }

down_interruptible()—-加锁，得不到锁的进程，将在此睡眠，进程中途可被其他信号打断，打断后down_interruptible()将返回非0，进程继续运行。得到锁，down_interruptible()返回0. 用法：

DEFINE_SEMAPHORE(sem) void exampSem() { if(down_interruptible(&sem)) return -ERESTARTSYS; ... ... ... up(&sem) }

down_killable()和down_interruptible()用法差不多，返回值也差不多，在打断信号类型上有要求。

down_trylock()用法和down_interruptible()差不多，返回值也差不多，区别在于，不等待不睡眠，得不到锁立即返回。

down_timeout()用法上和down_interruptible()差不多，返回值也差不多，却别在于，睡眠不可打断，睡眠时间受第2个参数jiffies限制，时间到，还没获得锁，返回非0，进程不访问临界区，返回进程继续运行。

up()解锁。

旗标实现的读者写者：

定义的头文件：

#include <linux/rwsem.h>

首先一样是定义rw_semaphore类型变量： 静态：

#define DECLARE_RWSEM(name) \ struct rw_semaphore name = __RWSEM_INITIALIZER(name)

动态用：

#define init_rwsem(sem) \ do { \ static struct lock_class_key __key; \ \ __init_rwsem((sem), #sem, &__key); \ } while (0)

然后调用函数，和旗标各个版本函数接口用法差不多：

void down_read(struct rw_semaphore *sem); int down_read_trylock(struct rw_semaphore *sem); void up_read(struct rw_semaphore *sem); 获得和释放对读者/写者旗标的读存取的函数.

void down_write(struct rw_semaphore *sem); int down_write_trylock(struct rw_semaphore *sem); void up_write(struct rw_semaphore *sem); void downgrade_write(struct rw_semaphore *sem); 管理对读者/写者旗标写存取的函数.

2. Completion （1）头文件：

#include <linux/completion.h>

（2）初始化： 静态定义并初始化：

/** * DECLARE_COMPLETION - declare and initialize a completion structure * @work: identifier for the completion structure * * This macro declares and initializes a completion structure. Generally used * for static declarations. You should use the _ONSTACK variant for automatic * variables. */ #define DECLARE_COMPLETION(work) \ struct completion work = COMPLETION_INITIALIZER(work)

动态定义并初始化： 用init_completion(),如下：

struct completion comp; init_completion(&comp);

（3）等待函数接口：

extern void wait_for_completion(struct completion *); extern void wait_for_completion_io(struct completion *); extern int wait_for_completion_interruptible(struct completion *x); extern int wait_for_completion_killable(struct completion *x); extern unsigned long wait_for_completion_timeout(struct completion *x, unsigned long timeout); extern unsigned long wait_for_completion_io_timeout(struct completion *x, unsigned long timeout); extern long wait_for_completion_interruptible_timeout( struct completion *x, unsigned long timeout); extern long wait_for_completion_killable_timeout( struct completion *x, unsigned long timeout); extern bool try_wait_for_completion(struct completion *x); extern bool completion_done(struct completion *x);

有点多，从函数名字顾名思义， wait_for_completion()等待通知完成，进程将在此睡眠等待。注意这个函数挂起的进程，是不可杀死的。

（4）通知完成函数：

extern void complete(struct completion *); extern void complete_all(struct completion *);

等待在一个completion上的可能又很多进程，complete()只能一次唤起一个， 如果用complete_all()将一次唤醒等待在同一个completion上的全部进程，用这个函数通知后，再次使用，需要重新调用init_completion()初始化complation。

void complete_and_exit(struct completion *c, long retval); 通过调用 complete 来发出一个 completion 事件, 并且为当前线程调用 exit。

3. 自旋锁（spinlock） （1）头文件：

#include <linux/spinlock.h>

（2）初始化：

struct spinlock_t splk; spin_lock_init(&splk);

（3）加锁接口：

void spin_lock(spinlock_t *lock); void spin_lock_irqsave(spinlock_t *lock, unsigned long flags); void spin_lock_irq(spinlock_t *lock); void spin_lock_bh(spinlock_t *lock);

spin_lock()—不禁止中断，只禁止内核抢占，小心使用，被打段后，再有其他函数来获取同一个锁，会导致死锁。 spin_lock_irq()—禁止所有中断，禁止内核抢占 spin_lock_irqsave()—在spin_lock_irq()基础上一防止改变中断状态，先保存在第2个参数中，退出时用spin_unlock_irqrestore()会恢复中断状态。 spin_lock_bh()—禁止软件中断，硬件其他中断不禁止，禁止内核抢占。

（4）对应释放锁的函数接口：

void spin_unlock(spinlock_t *lock); void spin_unlock_irqrestore(spinlock_t *lock, unsigned long flags); void spin_unlock_irq(spinlock_t *lock); void spin_unlock_bh(spinlock_t *lock);

上锁和解锁接口对应使用。

（5）自旋锁实现的读者写者： 头文件：

#include <linux/rwlock.h>

初始化：

rwlock_t rwl; rwlock_init(&rwl);

读者加锁解锁接口：

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); void read_unlock(rwlock_t *lock); void read_unlock_irqrestore(rwlock_t *lock, unsigned long flags); void read_unlock_irq(rwlock_t *lock); void read_unlock_bh(rwlock_t *lock);

写者加锁解锁：

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); int write_trylock(rwlock_t *lock); void write_unlock(rwlock_t *lock); void write_unlock_irqrestore(rwlock_t *lock, unsigned long flags); void write_unlock_irq(rwlock_t *lock); void write_unlock_bh(rwlock_t *lock);

4、顺序锁（seqlock）

（1）头文件：

#include <linux/seqlock.h>

（2）定义或者初始化： 静态：

DEFINE_SEQLOCK(seql) seqlock_init(&seql);

动态的：

seqlock_t seql； seqlock_init(&seql);

获得seqlock保护的读：

unsigned int read_seqbegin(seqlock_t *lock); int read_seqretry(seqlock_t *lock, unsigned int seq);

用法：

seqlock_t seql； seqlock_init(&seql); unsigned int seqSta; do{ seqSta = read_seqbegin(&seql); ... ... ... }while(read_seqretry(&seql,seqSta));

当读时，read_seqbegin(&seql)获取并保证在seqSta变量中，在完成读后，read_seqretry()比较seqSta，如发现不一样表示读期间有写，于是read_seqretry()返回非0，重新读，如果发现和传入的seqSta值一样，返回0，不再重新读取。

获取seqlock写保护：

void write_seqlock(seqlock_t *lock); void write_seqlock_irqsave(seqlock_t *lock, unsigned long flags); void write_seqlock_irq(seqlock_t *lock); void write_seqlock_bh(seqlock_t *lock);

seqlock写锁释放：

void write_sequnlock(seqlock_t *lock); void write_sequnlock_irqrestore(seqlock_t *lock, unsigned long flags); void write_sequnlock_irq(seqlock_t *lock); void write_sequnlock_bh(seqlock_t *lock);

seqlock写接口和spinloc差不多，各个对应版本意思也差不多。

5、原子变量 （1）头文件：

#include <asm/atomic.h>

（2）定义和初始化：

atomic_t v = ATOMIC_INIT(value);

（3）函数接口： void atomic_set(atomic_t *v, int i);//将v初始化成i的值 int atomic_read(atomic_t *v);//读取v的值，返回值就是v的值 void atomic_add(int i, atomic_t *v);//v加上i的值 void atomic_sub(int i, atomic_t *v);//v减去i的值 void atomic_inc(atomic_t *v);//v的值加1 void atomic_dec(atomic_t *v);//v的值减1 int atomic_inc_and_test(atomic_t *v);//v加上1，并测试v的值是否为0，为0返回1，否则返回0 int atomic_dec_and_test(atomic_t *v);//v减1，并测试v的值是否为0，为0返回1，否则返回0 int atomic_sub_and_test(int i, atomic_t *v);//v减去i，并测试v的值是否为0，为0返回1，否则返回0 int atomic_add_negative(int i, atomic_t *v);//v加i，并测试v的值是否为负数，若是返回1，否则返回0 int atomic_add_return(int i, atomic_t *v);//v加上i,并返回结果 int atomic_sub_return(int i, atomic_t *v);//v减去i，并返回结果 int atomic_inc_return(atomic_t *v);//v加1，并返回结果 int atomic_dec_return(atomic_t *v);//v减1，并返回结果

6、位原子操作 （1）头文件：

#include <asm/bitops.h>

（2）定义和使用：

int a=0; set_bit(3,&a)

将a的第3位置1，结果a等于8。

（3）接口： void set_bit(nr, void *addr);//第nr位置1 void clear_bit(nr, void *addr);//第nr位清0 void change_bit(nr, void *addr);//第nr位取反 test_bit(nr, void *addr);//测试第nr位，为1则返回1，否则返回0 int test_and_set_bit(nr, void *addr);//获取第nr为值并返回这位置(0或1)，然后将第nr位置1 int test_and_clear_bit(nr, void *addr);//获取第nr为值并返回这位置(0或1)，然后将第nr清0 int test_and_change_bit(nr, void *addr);//获取第nr为值并返回这位置(0或1)，然后将第nr取反

7、RCU（read-copy-update） （1）头文件：

#include <linux/rcupdate.h>

（2）读保护获取和释放：

rcu_read_lock() ... ... ... rcu_read_unlock

rcu_read_lock()会禁止抢占 rcu_read_unlock()会打开抢占 所以保护代码的临界区，不会被其它进程抢占(设想，中断服务程序调用，会导致？) rcu_read_lock()和rcu_read_unlock所包含的code，不会阻塞读进程，在rcu_read_unlock()未执行完时，写者，是不能修改更新指针地址的。

（3）写者保护：

extern void call_rcu(struct rcu_head *head, void (*func)(struct rcu_head *head));

call_rcu()调用到时，将func加到队列。 所有受rcu_read_lock()和rcu_read_unlock()保护的临界区释放，func才会被调用来释放更新指针指的数据。如临界区的rcu_read_lock()被调用，有rcu_read_unlock()还没有被调用，则func不会被调用，直到所有rcu_read_unlock()调用完成，func才能被调用。

用法：

void *p = NULL; char *tp = NULL; struct rcu_head eam; spinlock_t sp; void read(char *x) { int i=0; rcu_read_lock(); tp = rcu_dereference(p); for(i=0;i<200;i++) { *(x+i) = *(tp+i); } rcu_read_unlock(); } void free(struct rcu_head *head) { kfree(tp); } void write(char *x) { spin_lock(&sp) rcu_assign_pointer(p, x) spin_unlock(&sp); call_rcu(&ema,free); } void init() { spin_init_lock(&sp); }

rcu_assign_pointer(p, x)将p的值安全修改成x，这个函数是原子性的; rcu_dereference(p)将指针p的值给tp,也是原子性的，安全的。 当写完成，如果rcu_read_unlock()没被调用，free()不会被调用，只有rcu_read_unlock()被调用，free才会被执行。 这样就能并发的读写。

call_rcu可以用synchronize_rcu代替，synchronize_rcu阻塞写进程，如下：

void *p = NULL; char *tp = NULL; struct rcu_head eam; spinlock_t sp; void read(char *x) { int i=0; rcu_read_lock(); tp = rcu_dereference(p); for(i=0;i<200;i++) { *(x+i) = *(tp+i); } rcu_read_unlock(); } void write(char *x) { spin_lock(&sp) rcu_assign_pointer(p, x) spin_unlock(&sp); synchronize_rcu(); kfree(tp); } void init() { spin_init_lock(&sp); }

进程会在synchronize_rcu()处等待，直到rcu_read_unlock()完成后，才继续执行下去。