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

以上是Linux裝置驅動中如何解決並發控制問題？的詳細內容。更多資訊請關注PHP中文網其他相關文章！