Linux设备驱动中的并发

Posted 2023-04-12

参考技术A

并发就是多个执行单元或多个进程并行执行，而这多个执行单元对资源进行共享，比如访问同一个变量或同一个硬件资源，这个时候就很容易出现竞态(说简单点就是竞争同一个”女朋友”)。

为了处理并发带来的问题，Linux有几种处理方法:
1. 中断屏蔽
2. 原子操作
3. 自旋锁
4. 信号量
5. 互斥体
6. 完成量

以上几种处理并发的方式各有利弊，需要根据实际情况来选择使用哪一种。

它的原理就是让CPU不响应中断，一般使用这种方法的要求代码段要比较少，不能占用大量的时间。一般在驱动中不推荐使用这种方式。

原子量保证对一个整形数据的操作是排他性的。就是该操作绝不会在执行完毕前被任何其他任务或事件打断。

自旋锁是一种典型的对临界资源进行互斥访问的手段。当一个自旋锁已经被其他线程持有，而另一个线程试图去获取自旋锁，这时候就会一直在那里等待(原地自旋等待)。如果递归调用自旋锁，就会导致系统死锁。

与自旋锁不同的是，当获取不到信号量时，进程不会原地打转而是进入休眠等待状态。新的Linux内核倾向于直接使用mutex作为互斥手段，信号量用作互斥不再被推荐使用。

当进程占用资源时间较长时，用互斥体会比较好。

使用方法：

完成量的机制是实现一个线程发送一个信号通知另一个线程完成某个任务。一个线程调用wait_for_completion后就进入休眠，另一个线程执行完某个任务后就发送通知给进入休眠的线程，然后它就执行wait_for_completion后面的代码。

设备驱动中的并发控制

在为操作系统编写驱动设备时，因为涉及到中断、多任务和多处理器SMP的处理，所以内核提供了诸如中断屏蔽、原子操作、信号量、完成量等几种并发控制机制，对公用资源进行保护。下文将分别予以阐述。

0、中断

中断屏蔽使得中断与进程之间的并发不再发生，而且由于 Linux 内核的进程调度等操作都依赖
中断来实现，所以内核抢占进程之间的并发也就得以避免了。但由于 Linux 系统的异步 I/O、进程调度等很多重要操作都依赖于中断，中断对于内核的运行非常重要，在屏蔽中断期间所有的中断都无法得到处理，因此长时间屏蔽中断是很危险的，有可能造成数据丢失甚至系统崩溃。这就要求在屏蔽了中断之后，当前的内核执行路径应当尽快地执行完临界区的代码。
常用的开关中断的函数有：

• ​​local_irq_disable()​​​和​​local_irq_enable()​​​ 但它们只能禁止和使能本 CPU 内的中断，因此，
  并不能解决 SMP 多 CPU 引发的竞态。因此，单独使用中断屏蔽通常不是一种值得推荐的避免竞态的方法，它适宜与自旋锁联合使用。
• ​​local_irq_save(flags)​​​和​​local_irq_restore(flags)​​​ 与 local_irq_disable()不同的是，local_irq_save（flags）除了进行禁止中断的操作以外，还保存目前 CPU 的中断位信息，local_irq_restore（flags）进行的是与 local_irq_save
  （flags）相反的操作。
• ​​local_bh_disable()​​​和​​local_bh_enable()​​ 若只是想禁止中断的底半部，应使用 local_bh_disable()，使能被禁止的底半部应该调用 local_bh_enable()。

1、原子变量

原子变量就是，在对其进行操作时不会被其它任务或中断打断。而原子操作需要硬件的支持，因此时架构相关的，其API和原子类型的定义在内核​​include/asm/atomic.h​​文件中，都是用汇编语言实现的。它的优点是使用简单，但缺点是功能单一，只能做计数操作，保护的东西太少。在Linux中，原子变量的定义如下：

typedef struct
  volatile int counter;
atomic_t;

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

1.1 原子整型操作

• 定义并初始化atomic_t变量
  ​​​atomic_t counter = ATOMIC_INIT(0); //定义并初始化原子变量counter为0.​​
• 设置atomic_t变量的值
  ​​​void atomic_set(atomic_t *v, int i); //设置原子变量的值为i.​​
• 读atomic_t变量的值
  ​​​var = atomic_read(atomic_t *v); // 读原子变量counter的值到var中。​​
• 原子变量的加减法
  ​​​void atomic_add(int i, atomic_t *v); //将原子变量加i.​​​​void atomic_sub(i, atomic_t *v); //将原子变量减i.​​
• 原子变量的自增/自减
  ​​​void atomic_inc( atomic_t *v); //将原子变量自增1​​​​void atomic_dec( atomic_t *v); //将原子变量自减1​​
• 原子变量的自增、自减和减值后测试
  ​​​int atomic_inc_and_test( atomic_t *v); //将原子变量counter自增1,若结果为0返回真，否则返回假。​​​​int atomic_dec_and_test( atomic_t *v); //将原子变量counter自减1，若结果为0返回真，否则返回假。​​​​int atomic_sub_and_test( int i, atomic_t *v); //将原子变量counter自减1，若结果为0返回真，否则返回假。​​
• 原子变量的操作后返回
  ​​​int atomic_inc_return( atomic_t *v); //将原子变量自增1后返回新值。​​​​int atomic_dec_return( atomic_t *v); //将原子变量自减1后返回新值。​​​​int atomic_add_return( int i, atomic_t *v); //将原子变量加i后返回新值。​​​​int atomic_sub_return( int i, atomic_t *v); //将原子变量减i后返回新值。​​

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_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中还定义了一组与原子位操作函数功能相同的函数，其函数名是在原子位操作函数名前加两个下划线。区别在于他们不会保证是一个原子操作。
• 下面的程序使用原子变量使设备只能被一个进程打开：

static atomic_t xxx_available = ATOMIC_INIT(1); /*定义原子变量*/ 
 
static int xxx_open(struct inode *inode, struct file *filp) 
 
  ... 
  if (!atomic_dec_and_test(&xxx_available))
   
    atomic_inc(&xxx_available);
    return - EBUSY; /*已经打开*/ 
   
  ... 
  return 0; /* 成功 */ 
 
 
static int xxx_release(struct inode *inode, struct file *filp) 
 
  atomic_inc(&xxx_available); /* 释放设备 */ 
  return 0; 

2、自旋锁

自旋锁的类型也是一个结构体，即struct spinlock_t。下面对它的操作函数进行介绍：

2.1 定义和初始化自旋锁

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

spinlock_t lock;
spin_lock_init(&lock);

• 锁定自旋锁
  ​​spin_lock(lock); //这个宏一直等待，直到获得自旋锁​​​​spin_trylock(lock); //该宏尝试获得自旋锁 lock，如果能立即获得锁，它获得锁并返回真，否则立即返回假，实际上不再“在原地打转”​​
• 释放自旋锁
  ​​spin_unlock(lock); //这个宏立刻释放自旋锁​​
• 自旋锁的使用举例
  在驱动程序中，有些设备只允许打开一次，那么就需要一个自旋锁保护表示设备打开次数的count变量。如果不对count变量进行保护，当该设备被频繁打开的话，容易出现错误的count计数。

int count = 0;  //定义文件打开次数
spinlock_t lock;
int xxx_init(void)

  ...
  spin_lock_init(&lock);
  ...

/* 设备打开函数 */
static int xxx_open(struct inode *inode, struct file* filp)

  ...
  spin_lock(&lock);
  /* 临界代码 */
  if(count)  /*已经被其它程序打开过了*/
  
    spin_unlock(&lock);
    return -EBUSY;
  
  count++;  //增加使用计数
  spin_unlock(&lock);
  ...
  return 0;


/* 设备释放函数 */
static int xxx_release(struct inode *inode, struct file* filp)

  ...
  spin_lock(&lock);
  count--;  //减少使用计数
  spin_unlock(&lock);
  ...
  return 0;

2.2自旋锁的衍生函数

自旋锁主要针对 SMP 或单 CPU 但内核可抢占的情况，对于单 CPU 和内核不支持抢占的系统，自旋锁退化为空操作。在单 CPU 和内核可抢占的系统中，自旋锁持有期间内核的抢占将被禁止。由于内核可抢占的单 CPU 系统的行为实际很类似于 SMP系统，因此，在这样的单 CPU 系统中使用自旋锁仍十分必要。
尽管用了自旋锁可以保证临界区不受别的 CPU 和本 CPU 内的抢占进程打扰，但是得到锁的代码路径在执行临界区的时候还可能受到中断和底半部（BH）的影响。为了防止这种影响，就需要用到自旋锁的衍生。
spin_lock()/spin_unlock()是自旋锁机制的基础，它们和关中断 local_irq_ disable()/开中断 local_irq_enable()、关底半部
local_bh_disable()/开底半部 local_bh_enable()、关中断并保存状态字 local_irq_save()/
开中断并恢复状态 local_irq_restore()结合就形成了整套自旋锁机制，关系如下所示：

spin_lock_irq() = spin_lock() + local_irq_disable() 
spin_unlock_irq() = spin_unlock() + local_irq_enable() 
spin_lock_irqsave() = spin_unlock() + 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. 自旋锁不能递归使用，否则会引起死锁。
3. 如果进程获得自旋锁之后再阻塞，也有可能导致死锁的发生。copy_from_user()、copy_to_user()和 kmalloc()等函数都有可能引
   起阻塞，因此在自旋锁的占用期间不能调用这些函数。

2.3 读写自旋锁

读写自旋锁是一种比自旋锁粒度更小的锁机制，它保留了“自旋”的概念，但是在写操作方面，只能最多有一个写进程，在读操作方面，同时可以有多个读执行单元。当然，读和写也不能同时进行。

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

在对共享资源进行读取之前，应该先调用读锁定函数，完成之后应调用读解锁函数。
read_lock_irqsave()、read_lock_irq()和 read_lock_bh()分别是 read_lock()分别与
local_irq_save() 、 local_irq_disable() 和 local_bh_disable() 的 组 合 ， 读 解 锁 函 数
read_unlock_irqrestore()、read_unlock_ irq()、read_unlock_bh()的情况与此类似。

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); 
int write_trylock(rwlock_t *lock);

5. 写解锁

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

write_lock_irqsave() 、 write_lock_irq() 、 write_lock_bh() 分别是 write_lock() 与
local_irq_save() 、 local_irq_disable() 和 local_bh_disable() 的 组 合 ， 写 解 锁 函 数
write_unlock_irqrestore()、write_unlock_irq()、write_unlock_bh()的情况与此类似。
在对共享资源写之前，应该先调用写锁定函数，完成之后应调用写解锁函数。和 spin_trylock()一样，write_trylock()也只是尝试获取读写自旋锁，不管成功失败，都会立即返回。

6. 读写自旋锁的使用

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

2.4 顺序自旋锁

顺序锁（seqlock）是对读写锁的一种优化。读执行单元可以在写执行单元对被顺序锁保护的共享资
源进行写操作时仍然可以继续读，而不必等待写执行单元完成写操作，写执行单元也不需要等待所有读执行单元完成读操作才去进行写操作。
但是，写执行单元与写执行单元之间仍然是互斥的，即如果有写执行单元在进行写操作，其他写执行单元必须自旋在那里，直到写执行单元释放了顺序锁。如果读执行单元在读操作期间，写执行单元已经发生了写操作，那么，读执行单元必须重新读取数据，以便确保得到的数据是完整的。这种锁在读写同时进行的概率比较小时，性能是非常好的，而且它允许读写同时进行，因而更大地提高了并发性。
顺序锁有一个限制，它必须要求被保护的共享资源不含有指针，因为写执行单元可能使得指针失效，但读执行单元如果正要访问该指针，将导致 Oops。

1. 定义和初始化顺序锁

seqlock_t my_seqlock = SEQ_LOCK_UNLOCKED; /* 静态初始化 */ 
seqlock_t my_seqlock; 
seqlock_init(&my_seqlock); /* 动态初始化 */

2. 获得顺序锁

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)

3. 释放顺序锁

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

• 写执行单元使用顺序锁的模式如下：

write_seqlock(&seqlock_a); 
...              //写操作代码块
write_sequnlock(&seqlock_a);

• 读执行单元涉及如下顺序锁操作:

开始读：读执行单元在对被顺序锁 s1 保护的共享资源进行访问前需要调用一下函数之一，它们仅返回顺序锁 s1 的当前顺序号:

unsigned read_seqbegin(const seqlock_t *sl); 
read_seqbegin_irqsave(lock, flags)

重读：读执行单元在访问完被顺序锁 s1 保护的共享资源后需要调用以下函数之一来检查，在读访问期间是否有写操作。如果有写操作，读执行单元就需要重新进行读操作：

int read_seqretry(const seqlock_t *sl, unsigned iv); 
read_seqretry_irqrestore(lock, iv, flags)

读执行单元使用顺序锁的模式如下：

do  
  seqnum = read_seqbegin(&seqlock_a); 
  //读操作代码块
  ... 
 while (read_seqretry(&seqlock_a, seqnum));

2.5 RCU（读-拷贝-更新）

RCU 可以看做读写锁的高性能版本，相比读写锁，RCU 的优点在于既允许多个读执行单元同时访问被保护的数据，又允许多个读执行单元和多个写执行单元同时访问被保护的数据。
对于被 RCU 保护的共享数据结构，读执行单元不需要获得任何锁就可以访问它，不使用原子指令，而且在除 Alpha 的所有架构上也不需要内存栅（Memory Barrier），因此不会导致锁竞争、内存延迟以及流水线停滞。不需要锁也使得使用更容易，因为死锁问题就不需要考虑了。
使用 RCU 的写执行单元在访问它前需首先复制一个副本，然后对副本进行修改，最后使用一个回调机制在适当的时机把指向原来数据的指针重新指向新的被修改的数据，这个时机就是所有引用该数据的 CPU 都退出对共享数据的操作的时候。读执行单元没有任何同步开销，而写执行单元的同步开销则取决于使用的写执行单元间的同步机制。
但是，RCU 不能替代读写锁，因为如果写比较多时，对读执行单元的性能提高不能弥补写执行单元导致的损失。因为使用 RCU 时，写执行单元之间的同步开销会比较大，它需要延迟数据结构的释放，复制被修改的数据结构，它也必须使用某种锁机制同步并行的其他写执行单元的修改操作。

1. 读锁定

rcu_read_lock() 
rcu_read_lock_bh()

2. 读解锁

rcu_read_unlock() 
rcu_read_unlock_bh()

使用 RCU 进行读的模式如下：

rcu_read_lock() 
...//读临界区
rcu_read_unlock()

其中 rcu_read_lock()和 rcu_read_unlock()实质只是禁止和使能内核的抢占调度，如下所示：

#define rcu_read_lock()  preempt_disable()
#define rcu_read_unlock() preempt_enable()

其变种 rcu_read_lock_bh()、rcu_read_unlock_bh()则定义为：

#define rcu_read_lock_bh() local_bh_disable()
#define rcu_read_unlock_bh() local_bh_enable()

3. 同步 RCU

synchronize_rcu()

该函数由 RCU 写执行单元调用，它将阻塞写执行单元，直到所有的读执行单元已经完成读执行单元临界区，写执行单元才可以继续下一步操作。如果有多个 RCU写执行单元调用该函数，它们将在一个 grace period（即所有的读执行单元已经完成对
临界区的访问）之后全部被唤醒。synchronize_rcu()保证所有 CPU 都处理完正在运行的读执行单元临界区。

synchronize_kernel()

内核代码使用该函数来等待所有 CPU 处于可抢占状态，目前功能等同于synchronize_rcu()，但现在已经不建议使用，而是用​​synchronize_sched()​​，该函数用于等待所有 CPU 都处在可抢占状态，它能保证正在运行的中断处理函数处理完毕，但不能保证正在运行的软中断处理完毕。

4. 挂接回调

void fastcall call_rcu(struct rcu_head *head, void (*func)(struct rcu_head *rcu));

函数 call_rcu()也由 RCU 写执行单元调用，它不会使写执行单元阻塞，因而可以在中断上下文或软中断中使用。该函数将把函数 func 挂接到 RCU 回调函数链上，然后立即返回。

void fastcall call_rcu_bh(struct rcu_head *head, void (*func)(struct rcu_head *rcu));

call_ruc_bh()函数的功能几乎与 call_rcu()完全相同，唯一差别就是它把软中断的完成也当做经历一个 quiescent state（静默状态），因此如果写执行单元使用了该函数，在进程上下文的读执行单元必须使用 rcu_read_lock_bh()。
每个 CPU 维护两个数据结构 rcu_data 和 rcu_bh_data，它们用于保存回调函数，函数 call_rcu()把回调函数注册到 rcu_data，而 call_rcu_bh()则把回调函数注册到rcu_bh_data，在每一个数据结构上，回调函数被组成一个链表，先注册的排在前头，后注册的排在末尾。
使用 RCU 时，读执行单元必须提供一个信号给写执行单元以便写执行单元能够确定数据可以被安全地释放或修改的时机。有一个专门的垃圾收集器来探测读执行单元的信号，一旦所有的读执行单元都已经发送信号告知它们都不在使用被 RCU 保护的数据结构，垃圾收集器就调用回调函数完成最后的数据释放或修改操作。

5. RCU的链表操作函数

static inline void list_add_rcu(struct list_head *new, struct list_head 
*head);

该函数把链表元素 new 插入到 RCU 保护的链表 head 的开头，内存栅保证了在引用这个新插入的链表元素之前，新链表元素的链接指针的修改对所有读执行单元是可见的。

static inline void list_add_tail_rcu(struct list_head *new, 
 struct list_head *head);

该函数类似于 list_add_rcu()，它将把新的链表元素 new 添加到被 RCU 保护的链表的末尾。

static inline void list_del_rcu(struct list_head *entry);

该函数从 RCU 保护的链表中删除指定的链表元素 entry。

static inline void list_replace_rcu(struct list_head *old, struct 
list_head *new);

该函数是 RCU 新添加的函数，并不存在非 RCU 版本。它使用新的链表元素 new取代旧的链表元素 old，内存栅保证在引用新的链表元素之前，它对链接指针的修正对所有读执行单元是可见的。

list_for_each_rcu(pos, head)

该宏用于遍历由 RCU 保护的链表 head，只要在读执行单元临界区使用该函数，它就可以安全地和其他_rcu 链表操作函数并发运行如 list_add_rcu()。

list_for_each_safe_rcu(pos, n, head)

该宏类似于 list_for_each_rcu，不同之处在于它允许安全地删除当前链表元素 pos。

list_for_each_entry_rcu(pos, head, member)

该宏类似于 list_for_each_rcu，不同之处在于它用于遍历指定类型的数据结构链表，当前链表元素 pos 为一个包含 struct list_head 结构的特定的数据结构。

static inline void hlist_del_rcu(struct hlist_node *n)

它从由 RCU 保护的哈希链表中移走链表元素 n。

static inline void hlist_add_head_rcu(struct hlist_node *n, struct hlist_head *h);

该函数用于把链表元素 n 插入到被 RCU 保护的哈希链表的开头，但同时允许读执行单元对该哈希链表的遍历。内存栅确保在引用新链表元素之前，它对指针的修改对所有读执行单元可见。

hlist_for_each_rcu(pos, head)

该宏用于遍历由 RCU 保护的哈希链表 head，只要在读端临界区使用该函数，它就可以安全地和其他_rcu 哈希链表操作函数（如 hlist_add_rcu）并发运行。

hlist_for_each_entry_rcu(tpos, pos, head, member)

类似于 hlist_for_each_rcu()，不同之处在于它用于遍历指定类型的数据结构哈希链表，当前链表元素 pos 为一个包含 struct list_head 结构的特定的数据结构。
应用举例：
原先的 audit_filter_task 函数，读链
表前获得读写锁 ：

static enum audit_state audit_filter_task(struct task_struct *tsk) 
 
  struct audit_entry *e; 
  enum audit_state state; 
  read_lock(&auditsc_lock); 
  /* 遍历链表 */ 
  list_for_each_entry(e, &audit_tsklist, list) 
   
    if (audit_filter_rules(tsk, &e->rule, NULL, &state)) 
     
      read_unlock(&auditsc_lock);
      return state; 
     
   
  read_unlock(&auditsc_lock); 
  return AUDIT_BUILD_CONTEXT; 

修改后的 audit_filter_task 函数，采用 RCU:

static enum audit_state audit_filter_task(struct task_struct *tsk) 
 
  struct audit_entry *e; 
  enum audit_state state; 
  rcu_read_lock(); 
  /* 遍历链表 */ 
  list_for_each_entry_rcu(e, &audit_tsklist, list) 
   
    if (audit_filter_rules(tsk, &e->rule, NULL, &state)) 
     
      rcu_read_unlock(); 
      return state; 
     
   
  rcu_read_unlock(); 
  return AUDIT_BUILD_CONTEXT; 

原先的 audit_add_rule，添加链表元
素前获得读写锁:

static inline int audit_add_rule(struct audit_entry *entry, struct list_head*list) 
 
  write_lock(&auditsc_lock); 
  if (entry->rule.flags &AUDIT_PREPEND) 
   
    entry->rule.flags &= ~AUDIT_PREPEND; 
    list_add(&entry->list, list);
    
  else 
   
    list_add_tail(&entry->list, list); 
   
  write_unlock(&auditsc_lock); 
  return 0; 

修改后的 audit_add_rule，在添加链表元素
时使用 list_add_xxx 函数:

static inline int audit_add_rule(struct audit_entry *entry, struct list_head*list) 
  
  if (entry->rule.flags &AUDIT_PREPEND) 
  
    entry->rule.flags &= ~AUDIT_PREPEND; 
    list_add_rcu(&entry->list, list);
   
  else 
   
    list_add_tail_rcu(&entry->list, list); 
   
  return 0; 

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信号量的使用

• 定义一个信号量
  ​​​struct semaphore sema;​​
• 初始化一个信号量
  ​​​sema_init(sema, val); // 初始化信号量sema为val​​

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

此外，下面两个宏是定义并初始化信号量的“快捷方式”。

DECLARE_MUTEX(name) 
DECLARE_MUTEX_LOCKED(name)

前者定义一个名为 name 的信号量并初始化为 1，后者定义一个名为 name 的信号量
并初始化为 0。

• 获得信号量
  ​​​down(&sema);​​​ 如果请求不到会导致进程睡眠，且不会被其它信号唤醒，故不能用于中断上下文中；
  ​​​down_interruptible(&sema);​​如果请求不到会导致进程睡眠，但可以被其它信号唤醒。所以在调用该函数时，要检查返回值，以判断被唤醒的原因。例如：

if (down_interruptible(&sem)) 
 
  return - ERESTARTSYS; 

​​int down_trylock(struct semaphore * sem);​​​该函数尝试获得信号量 sem，如果能够立刻获得，它就获得该信号量并返回 0，
否则，返回非 0 值。它不会导致调用者睡眠，可以在中断上下文使用。

• 释放信号量
  ​​​up(&sema);​​
• 信号量的使用
  以下是使用信号量实现设备只能被一个进程打开的例子：

struct semaphore sema;
int xxx_init(void)

  ...
  init_MUTEX(&sema);
  ...


int xxx_open(struct inode *inode, struct file *filp)

  ...
  if (down_trylock(&sema))  //获得打开锁
    return -EBUSY;
  不允许其它进程访问该临界资源区
  ...
  return 0;  //成功


int xxx_release(struct inode *inode, struct file *filp)

  
以上是关于Linux设备驱动中的并发的主要内容，如果未能解决你的问题，请参考以下文章
 
 linux设备驱动中的并发控制
 设备驱动中的并发控制
 Linux设备驱动基础01之并发与竞态
 Linux设备驱动基础01之并发与竞态
 Linux设备驱动基础01之并发与竞态
 linux驱动程序中的并发控制-8（完成量（completion））-50