Linux设备驱动中如何解决并发控制问题？

在Linux设备驱动中，当多个执行单元同时访问相同的资源时，可能会引发“竞态”，导致数据不一致或系统崩溃。因此，我们必须对共享资源进行并发控制，保证其互斥访问。本文将介绍Linux内核中解决并发控制的常用方法，包括中断屏蔽、原子操作、自旋锁、信号量、互斥体等，并给出相应的示例代码。

Linux 设备驱动中必须解决的一个问题是多个进程对共享资源的并发访问，并发的访问会导致竞态。
中断屏蔽、原子操作、自旋锁和信号量都是解决并发问题的机制。中断屏蔽很少单独被使用，原子操作只能针对整数进行，因此自旋锁和信号量应用最为广泛。
自旋锁会导致死循环，锁定期间不允许阻塞，因此要求锁定的临界区小。信号量允许临界区阻塞，可以适用于临界区大的情况。
读写自旋锁和读写信号量分别是放宽了条件的自旋锁和信号量，它们允许多个执行单元对共享资源的并发读。

中断屏蔽

访问共享资源的代码区域称为临界区（ critical sections），在单 CPU 范围内避免竞态的一种简单而省事的方法是在进入临界区之前屏蔽系统的中断。中断屏蔽将使得中断与进程之间的并发不再发生，而且，由于 Linux 内核的进程调度等操作都依赖中断来实现，内核抢占进程之间的并发也得以避免了。

    local_irq_disable(); /* 屏蔽中断 */
    ...
    critical section /* 临界区*/
    ...
    local_irq_enable(); /* 开中断 */

但是由于 Linux 的异步 I/O、进程调度等很多重要操作都依赖于中断，长时间屏蔽中断是很危险的；而且中断屏蔽只对本 CPU 内的中断有效，因此也并不能解决 SMP 多 CPU 引发的竞态。在实际应用中并不推荐直接使用，适宜与下文的自旋锁结合使用。

原子操作

Linux 内核提供了一系列函数来实现内核中的原子操作，这些函数又分为两类，分别针对位和整型变量进行原子操作。它们的共同点是在任何情况下操作都是原子的，内核代码可以安全地调用它们而不被打断。

整型原子操作

• 设置原子变量的值

  #include 
  void atomic_set(atomic_t *v, int i); /* 设置原子变量的值为 i */
  atomic_t v = ATOMIC_INIT(0);         /* 定义原子变量 v 并初始化为 0 */
  

• 获取原子变量的值

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

• 原子变量加/减

  void atomic_add(int i, atomic_t *v); /* 原子变量增加 i */
  void atomic_sub(int i, atomic_t *v); /* 原子变量减少 i */
  
  void atomic_inc(atomic_t *v); /* 原子变量自增 1 */
  void atomic_dec(atomic_t *v); /* 原子变量自减 1 */
  
  /* 操作完结果==0, return true */
  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);
  
  /* 操作完结果 return true */
  int atomic_add_negative(int i, atomic_t *v);
  
  /* 操作并返回结果 */
  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);
  

位原子操作

位原子操作相当快，一般只需一个机器指令，不需关中断。

• set/clear/toggle

  #include 
  
  /* 更改指针addr所指数据的第nr位 */
  void set_bit(nr, void *addr);
  void clear_bit(nr, void *addr);
  void change_bit(nr, void *addr);
  

• test

  int test_bit(nr, void *addr); /* 返回第nr位 */
  

• 测试并操作

  /* 操作第nr位，并返回操作前的值 */
  int test_and_set_bit(nr, void *addr);
  int test_and_clear_bit(nr, void *addr);
  int test_and_change_bit(nr, void *addr);
  

自旋锁(spinlock)

自旋锁（spinlock）是一种典型的对临界资源进行互斥访问的手段，其名称来源于它的工作方式。为了获得一个自旋锁， 在某 CPU 上运行的代码需先执行一个原子操作，该操作测试并设置（ test-and-set） 某个内存变量，由于它是原子操作，所以在该操作完成之前其他执行单元不可能访问这个内存变量。如果测试结果表明锁已经空闲，则程序获得这个自旋锁并继续执行； 如果测试结果表明锁仍被占用，程序将在一个小的循环内重复这个“ 测试并设置” 操作，即进行所谓的“ 自旋”，通俗地说就是“在原地打转”。 当自旋锁的持有者通过重置该变量释放这个自旋锁后，某个等待的“测试并设置” 操作向其调用者报告锁已释放。

Basic

• 定义/初始化

  #include 
  
  /* 静态初始化 */
  spinlock_t my_lock = SPIN_LOCK_UNLOCKED;
  /* 动态初始化 */
  void spin_lock_init(spinlock_t *lock);
  

• 获取/释放

  /* 基本操作 */
  void spin_lock(spinlock_t *lock);
  void spin_unlock(spinlock_t *lock);
  
  /* 保存中断状态并关闭 == spin_lock() + local_irq_save() */
  void spin_lock_irqsave(spinlock_t *lock, unsigned long flags);
  void spin_unlock_irqsave(spinlock_t *lock, unsigned long flags);
  
  /* 忽略操作前中断状态 */
  void spin_lock_irq(spinlock_t *lock);
  void spin_unlock_irq(spinlock_t *lock);
  
  /* 关闭中断底部（即关闭软件中断，打开硬件中断，详见后续中断的讲解） */
  void spin_lock_bh(spinlock_t *lock);
  void spin_unlock_bh(spinlock_t *lock);
  
  /* 非阻塞获取，成功返回非0 */
  int spin_trylock(spinlock_t *lock);
  int spin_trylock_bh(spinlock_t *lock);
  

Reader/Writer Spinlocks

粒度更小，可多Reader同时读，但Writer只能单独，且读与写不能同时，适用于写很少读很多的情况。

• 定义/初始化

  rwlock_t my_rwlock = RW_LOCK_UNLOCKED; /* 静态初始化 */
  rwlock_t my_rwlock;
  rwlock_init(&my_rwlock); /* 动态初始化 */
  

• 读

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

seqlock

顺序锁（seqlock）是对读写锁的一种优化，采用了重读机制，读写不相互阻塞。

• 定义/初始化

  #include 
  
  seqlock_t lock1 = SEQLOCK_UNLOCKED; /* 静态 */
  seqlock_t lock2;
  seqlock_init(&lock2); /* 动态 */
  

• 读

  /* 读之前先获取个顺序号，读完与当前顺序号对比，如不一致则重读 */
  unsigned int seq;
  do {
      seq = read_seqbegin(&the_lock);
      /* Do what you need to do */
  } while read_seqretry(&the_lock, seq);
  
  /* 如果这个锁可能会出现在中断程序中获取，则在这里应使用关中断版本 */
  unsigned int read_seqbegin_irqsave(seqlock_t *lock,unsigned long flags);
  int read_seqretry_irqrestore(seqlock_t *lock, unsigned int seq,unsigned long flags);
  

• 写

  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);
  int write_tryseqlock(seqlock_t *lock);
  
  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);
  

RCU(Read-Copy-Update)

对于被 RCU 保护的共享数据结构，读执行单元不需要获得任何锁就可以访问它，因此读执行单元没有任何同步开销。使用 RCU 的写执行单元在访问它前需首先拷贝一个副本，然后对副本进行修改，最后使用一个回调机制在适当的时机把指向原来数据的指针重新指向新的被修改的数据，这个时机就是所有引用该数据的 CPU 都退出对共享数据的操作的时候。写执行单元的同步开销则取决于使用的写执行单元间同步机制。RCU在驱动中很少使用，这里暂不详述。

注意事项

• 自旋锁实际上是忙等锁，当锁不可用时， CPU 一直循环执行“测试并设置”该锁直到可用而取得该锁， CPU 在等待自旋锁时不做任何有用的工作，仅仅是等待。 因此，只有在占用锁的时间极短的情况下，使用自旋锁才是合理的。 当临界区很大，或有共享设备的时候，需要较长时间占用锁，使用自旋锁会降低系统的性能。
• 自旋锁可能导致系统死锁。引发这个问题最常见的情况是递归使用一个自旋锁，即如果一个已经拥有某个自旋锁的 CPU 想第二次获得这个自旋锁，则该 CPU 将死锁。
• 自旋锁锁定期间不能调用可能引起进程调度而导致休眠的函数。如果进程获得自旋锁之后再阻塞， 如调用 copy_from_user()、 copy_to_user()、 kmalloc()和 msleep()等函数，则可能导致内核的崩溃。

信号量 semaphore

使用方式和自旋锁类似，不同的是，当获取不到信号量时，进程不会原地打转而是进入休眠等待状态。

• 定义/初始化

  #include 
  
  struct semaphore sem;
  void sema_init(struct semaphore *sem, int val);
  
  /* 通常我们将val的值置1,即使用互斥模式 */
  DECLARE_MUTEX(name);
  DECLARE_MUTEX_LOCKED(name);
  void init_MUTEX(struct semaphore *sem);
  void init_MUTEX_LOCKED(struct semaphore *sem);
  

• 获得信号量

  void down(struct semaphore * sem); /* 信号量减1, 会导致睡眠，因此不能在中断上下文使用 */
  int down_interruptible(struct semaphore * sem); /* 与down不同的是，进入睡眠后的进程可被打断返回非0 */ 
  int down_trylock(struct semaphore * sem); /* 非阻塞版本，获得返回0，不会导致睡眠，可在中断上下文使用 */
  

• 释放信号量

  void up(struct semaphore * sem);
  

Reader/Writer Semaphores

读写信号量与信号量的关系与读写自旋锁和自旋锁的关系类似，读写信号量可能引起进程阻塞，但它可允许 N 个读执行单元同时访问共享资源， 而最多只能有 1 个写执行单元。因此，读写信号量是一种相对放宽条件的粒度稍大于信号量的互斥机制。

• 定义/初始化

  #include 
  
  struct rw_semaphore;
  void init_rwsem(struct rw_semaphore *sem);
  

• 读

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

完成量 completion

轻量级，用于一个执行单元等待另一个执行单元执行完某事。

• 定义/初始化

  #include 
  
  /* 静态 */
  DECLARE_COMPLETION(name);
  /* 动态 */
  struct completion my_completion;
  init_completion(struct completion *c);
  
  INIT_COMPLETION(struct completion c); /* 重新初始化已经定义并使用过的 completion */
  

• 等待完成

  void wait_for_completion(struct completion *c);
  

• 完成信号

  void complete(struct completion *c); /* 唤醒1个 */
  void complete_all(struct completion *c); /* 唤醒所有waiter */
  void complete_and_exit(struct completion *c, long retval); /* call complete() and exit(retval) */
  

本文总结了Linux设备驱动中的并发控制问题及其解决方法。通过使用合适的互斥机制，我们可以避免竞态的发生，提高设备驱动的稳定性和性能。在实际开发中，我们需要根据不同的场景选择最优的方案，并注意避免死锁、优先级反转等潜在的问题。

以上就是良许教程网为各位朋友分享的Linu系统相关内容。想要了解更多Linux相关知识记得关注公众号“良许Linux”，或扫描下方二维码进行关注，更多干货等着你 ！