Linux設(shè)備驅(qū)動的并發(fā)控制

Linux 設(shè)備驅(qū)動中必須解決的一個問題是多個進(jìn)程對共享資源的并發(fā)訪問，并發(fā)的訪問會導(dǎo)致競態(tài)。

中斷屏蔽、原子操作、自旋鎖和信號量都是解決并發(fā)問題的機制。中斷屏蔽很少單獨被使用，原子操作只能針對整數(shù)進(jìn)行，因此自旋鎖和信號量應(yīng)用最為廣泛。
自旋鎖會導(dǎo)致死循環(huán)，鎖定期間不允許阻塞，因此要求鎖定的臨界區(qū)小。信號量允許臨界區(qū)阻塞，可以適用于臨界區(qū)大的情況。
讀寫自旋鎖和讀寫信號量分別是放寬了條件的自旋鎖和信號量，它們允許多個執(zhí)行單元對共享資源的并發(fā)讀。

中斷屏蔽

訪問共享資源的代碼區(qū)域稱為臨界區(qū)（ critical sections），在單 CPU 范圍內(nèi)避免競態(tài)的一種簡單而省事的方法是在進(jìn)入臨界區(qū)之前屏蔽系統(tǒng)的中斷。中斷屏蔽將使得中斷與進(jìn)程之間的并發(fā)不再發(fā)生，而且，由于 Linux 內(nèi)核的進(jìn)程調(diào)度等操作都依賴中斷來實現(xiàn)，內(nèi)核搶占進(jìn)程之間的并發(fā)也得以避免了。

local_irq_disable(); /* 屏蔽中斷 */ ... critical section /* 臨界區(qū)*/ ... local_irq_enable(); /* 開中斷 */

但是由于 Linux 的異步 I/O、進(jìn)程調(diào)度等很多重要操作都依賴于中斷，長時間屏蔽中斷是很危險的；而且中斷屏蔽只對本 CPU 內(nèi)的中斷有效，因此也并不能解決 SMP 多 CPU 引發(fā)的競態(tài)。在實際應(yīng)用中并不推薦直接使用，適宜與下文的自旋鎖結(jié)合使用。

原子操作

Linux 內(nèi)核提供了一系列函數(shù)來實現(xiàn)內(nèi)核中的原子操作，這些函數(shù)又分為兩類，分別針對位和整型變量進(jìn)行原子操作。它們的共同點是在任何情況下操作都是原子的，內(nèi)核代碼可以安全地調(diào)用它們而不被打斷。

整型原子操作

設(shè)置原子變量的值

#include void atomic_set(atomic_t *v, int i); /* 設(shè)置原子變量的值為 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 *//* 操作完結(jié)果==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);/* 操作完結(jié)果 <0, return true */int atomic_add_negative(int i, atomic_t *v);/* 操作并返回結(jié)果 */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);

位原子操作

位原子操作相當(dāng)快，一般只需一個機器指令，不需關(guān)中斷。

set/clear/toggle

#include /* 更改指針addr所指數(shù)據(jù)的第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）是一種典型的對臨界資源進(jìn)行互斥訪問的手段，其名稱來源于它的工作方式。為了獲得一個自旋鎖， 在某 CPU 上運行的代碼需先執(zhí)行一個原子操作，該操作測試并設(shè)置（ test-and-set） 某個內(nèi)存變量，由于它是原子操作，所以在該操作完成之前其他執(zhí)行單元不可能訪問這個內(nèi)存變量。如果測試結(jié)果表明鎖已經(jīng)空閑，則程序獲得這個自旋鎖并繼續(xù)執(zhí)行； 如果測試結(jié)果表明鎖仍被占用，程序?qū)⒃谝粋€小的循環(huán)內(nèi)重復(fù)這個“ 測試并設(shè)置” 操作，即進(jìn)行所謂的“ 自旋”，通俗地說就是“在原地打轉(zhuǎn)”。 當(dāng)自旋鎖的持有者通過重置該變量釋放這個自旋鎖后，某個等待的“測試并設(shè)置” 操作向其調(diào)用者報告鎖已釋放。

Basic

定義/初始化

#include /* 靜態(tài)初始化 */spinlock_t my_lock = SPIN_LOCK_UNLOCKED;/* 動態(tài)初始化 */void spin_lock_init(spinlock_t *lock);

獲取/釋放

/* 基本操作 */void spin_lock(spinlock_t *lock);void spin_unlock(spinlock_t *lock);/* 保存中斷狀態(tài)并關(guān)閉 == 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);/* 忽略操作前中斷狀態(tài) */void spin_lock_irq(spinlock_t *lock);void spin_unlock_irq(spinlock_t *lock);/* 關(guān)閉中斷底部（即關(guān)閉軟件中斷，打開硬件中斷，詳見后續(xù)中斷的講解） */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; /* 靜態(tài)初始化 */rwlock_t my_rwlock;rwlock_init(&my_rwlock); /* 動態(tài)初始化 */

讀

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）是對讀寫鎖的一種優(yōu)化，采用了重讀機制，讀寫不相互阻塞。

定義/初始化

#include seqlock_t lock1 = SEQLOCK_UNLOCKED; /* 靜態(tài) */seqlock_t lock2;seqlock_init(&lock2); /* 動態(tài) */

讀

/* 讀之前先獲取個順序號，讀完與當(dāng)前順序號對比，如不一致則重讀 */unsigned int seq;do { seq = read_seqbegin(&the_lock); /* Do what you need to do */} while read_seqretry(&the_lock, seq);/* 如果這個鎖可能會出現(xiàn)在中斷程序中獲取，則在這里應(yīng)使用關(guān)中斷版本 */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 保護(hù)的共享數(shù)據(jù)結(jié)構(gòu)，讀執(zhí)行單元不需要獲得任何鎖就可以訪問它，因此讀執(zhí)行單元沒有任何同步開銷。使用 RCU 的寫執(zhí)行單元在訪問它前需首先拷貝一個副本，然后對副本進(jìn)行修改，最后使用一個回調(diào)機制在適當(dāng)?shù)臅r機把指向原來數(shù)據(jù)的指針重新指向新的被修改的數(shù)據(jù)，這個時機就是所有引用該數(shù)據(jù)的 CPU 都退出對共享數(shù)據(jù)的操作的時候。寫執(zhí)行單元的同步開銷則取決于使用的寫執(zhí)行單元間同步機制。RCU在驅(qū)動中很少使用，這里暫不詳述。

注意事項

自旋鎖實際上是忙等鎖，當(dāng)鎖不可用時， CPU 一直循環(huán)執(zhí)行“測試并設(shè)置”該鎖直到可用而取得該鎖， CPU 在等待自旋鎖時不做任何有用的工作，僅僅是等待。 因此，只有在占用鎖的時間極短的情況下，使用自旋鎖才是合理的。 當(dāng)臨界區(qū)很大，或有共享設(shè)備的時候，需要較長時間占用鎖，使用自旋鎖會降低系統(tǒng)的性能。

自旋鎖可能導(dǎo)致系統(tǒng)死鎖。引發(fā)這個問題最常見的情況是遞歸使用一個自旋鎖，即如果一個已經(jīng)擁有某個自旋鎖的 CPU 想第二次獲得這個自旋鎖，則該 CPU 將死鎖。

自旋鎖鎖定期間不能調(diào)用可能引起進(jìn)程調(diào)度而導(dǎo)致休眠的函數(shù)。如果進(jìn)程獲得自旋鎖之后再阻塞， 如調(diào)用 copy_from_user()、 copy_to_user()、 kmalloc()和 msleep()等函數(shù)，則可能導(dǎo)致內(nèi)核的崩潰。

信號量 semaphore

使用方式和自旋鎖類似，不同的是，當(dāng)獲取不到信號量時，進(jìn)程不會原地打轉(zhuǎn)而是進(jìn)入休眠等待狀態(tài)。

定義/初始化

#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, 會導(dǎo)致睡眠，因此不能在中斷上下文使用 */int down_interruptible(struct semaphore * sem); /* 與down不同的是，進(jìn)入睡眠后的進(jìn)程可被打斷返回非0 */ int down_trylock(struct semaphore * sem); /* 非阻塞版本，獲得返回0，不會導(dǎo)致睡眠，可在中斷上下文使用 */

釋放信號量

void up(struct semaphore * sem);

Reader/Writer Semaphores

讀寫信號量與信號量的關(guān)系與讀寫自旋鎖和自旋鎖的關(guān)系類似，讀寫信號量可能引起進(jìn)程阻塞，但它可允許 N 個讀執(zhí)行單元同時訪問共享資源， 而最多只能有 1 個寫執(zhí)行單元。因此，讀寫信號量是一種相對放寬條件的粒度稍大于信號量的互斥機制。

定義/初始化

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

寫

/* 寫比讀優(yōu)先級高，寫時所有讀只能等待 */void down_write(struct rw_semaphore *sem);int down_write_trylock(struct rw_semaphore *sem);void up_write(struct rw_semaphore *sem);

完成量 completion

輕量級，用于一個執(zhí)行單元等待另一個執(zhí)行單元執(zhí)行完某事。

定義/初始化

#include /* 靜態(tài) */DECLARE_COMPLETION(name);/* 動態(tài) */struct completion my_completion;init_completion(struct completion *c);INIT_COMPLETION(struct completion c); /* 重新初始化已經(jīng)定義并使用過的 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) */

?