? 并發(fā)(concurrency)指的是多個執(zhí)行單元同時、并行被執(zhí)行。而并發(fā)的執(zhí)行單元對共享資源(硬件資源和軟件上的全局、靜態(tài)變量)的訪問則容易導致競態(tài)(race conditions)。
? SMP是一種緊耦合、共享存儲的系統(tǒng)模型，它的特點是多個CPU使用共同的系統(tǒng)總線，因此可訪問共同的外設和存儲器。
? 進程與搶占它的進程訪問共享資源的情況類似于SMP的多個CPU.
? 中斷可打斷正在執(zhí)行的進程，若中斷處理程序訪問進程正在訪問的資源，則競態(tài)也會發(fā)生。中斷也可能被新的更高優(yōu)先級的中斷打斷，因此，多個中斷之間也可能引起并發(fā)而導致競態(tài)。上述并發(fā)的發(fā)生情況除了SMP是真正的并行以外，其他的都是“宏觀并行、微觀串行”的，但其引發(fā)的實質(zhì)問題和SMP相似。解決競態(tài)問題的途徑是保證對共享資源的互斥訪問，即一個執(zhí)行單元在訪問共享資源的時候，其他的執(zhí)行單元被禁止訪問。
訪問共享資源的代碼區(qū)域成為臨界區(qū)(critical sections)，臨界區(qū)需要以某種互斥機制加以保護。中斷屏蔽、原子操作、自旋鎖和信號量等是Linux設備驅(qū)動中可采用的互斥途徑。

?

中斷屏蔽的使用方法為：

local_irq_disable()??//?屏蔽中斷

...

critical?section??// 臨界區(qū)

...

local_irq_enable()??// 開中斷

? 在屏蔽了中斷后，當前的內(nèi)核執(zhí)行路徑應當盡快執(zhí)行完臨界區(qū)代碼。上述兩個函數(shù)都只能禁止和使能本CPU內(nèi)的中斷，不能解決SMP多CPU引發(fā)的競態(tài)。

? local_irq_save(flags) 除禁止中斷的操作外，還保存目前CPU的中斷位信息；

? local_irq_restore(flags) 進行的是local_irq_save(flags)相反的操作；

? 若只想禁止中斷的底半部，應使用local_bh_disable(), 使能被local_bh_disable()禁止的底半部應調(diào)用local_bh_enable()。

?

原子操作指的是在執(zhí)行過程中不會被別的代碼路徑所中斷的操作。

整型原子操作：

// 設置原子變量的值

void?atomic_set(atomic_t?*v,?int?i);??// 設置原子變量的值為i

atomic_t?v?=?ATOMIC_INIT(0);??// 定義原子變量v，并初始化為0

?

// 獲取原子變量的值

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，為0則返回true，否則返回false

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

?

// 操作并返回： 對原子變量進行加/減和自增/自減操作，并返回新的值

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

? 位原子操作：

// 設置位

void?set_bit(nr,?void?*addr);??// 設置addr地址的第nr位，即將位寫1

?

// 清除位

void?clear_bit(nr,?void?*addr);??// 清除addr地址的第nr位，即將位寫0

?

// 改變位

void?change_bit(nr,?void?*addr);??// 對addr地址的第nr位取反

?

// 測試位

test_bit(nr,?void?*addr);?// 返回addr地址的第nr位

?

// 測試并操作：等同于執(zhí)行test_bit(nr, void *addr)后再執(zhí)行xxx_bit(nr, void *addr)

int?test_and_set_bit(nr,?void?*addr);

int?test_and_clear_bit(nr,?void?*addr);

int?test_and_change_bit(nr,?void?*addr);

? 原子變量使用實例，使設備只能被一個進程打開：

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

??? ??? return?-?EBUSY;??// 已經(jīng)打開

??? }

??? ...

??? return?0;??// 成功

}

?

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

{

??? atomic_inc(&xxx_available);??// 釋放設備

??? return?0;

}

?

自旋鎖(spin lock)——“在原地打轉(zhuǎn)”。若一個進程要訪問臨界資源，測試鎖空閑，則進程獲得這個鎖并繼續(xù)執(zhí)行；若測試結果表明鎖扔被占用，進程將在一個小的循環(huán)內(nèi)重復“測試并設置”操作，進行所謂的“自旋”，等待自旋鎖持有者釋放這個鎖。

? 自旋鎖的相關操作：

?

//?定義自旋鎖?

spinlock_t?spin;?

?

// 初始化自旋鎖

spin_lock_init(lock);

?

//?獲得自旋鎖：若能立即獲得鎖，它獲得鎖并返回，否則，自旋，直到該鎖持有者釋放

spin_lock(lock);?

?

//?嘗試獲得自旋鎖：若能立即獲得鎖，它獲得并返回真，否則立即返回假，不再自旋

spin_trylock(lock);?

?

// 釋放自旋鎖: 與spin_lock(lock)和spin_trylock(lock)配對使用

spin_unlock(lock);?

?

? 自旋鎖的使用：

// 定義一個自旋鎖

spinlock_t?lock;

spin_lock_init(&lock);

?

spin_lock(&lock);??// 獲取自旋鎖，保護臨界區(qū)

...??// 臨界區(qū)

spin_unlock();??// 解鎖

? 自旋鎖持有期間內(nèi)核的搶占將被禁止。

? 自旋鎖可以保證臨界區(qū)不受別的CPU和本CPU內(nèi)的搶占進程打擾，但是得到鎖的代碼路徑在執(zhí)行臨界區(qū)的時候還可能受到中斷和底半部(BH)的影響。

? 為防止這種影響，需要用到自旋鎖的衍生：

spin_lock_irq() = spin_lock() + local_irq_disable()

spin_unlock_irq() = spin_unlock() + local_irq_enable()

spin_lock_irqsave() = spin_lock() + 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()

? 注意：自旋鎖實際上是忙等待，只有在占用鎖的時間極短的情況下，使用自旋鎖才是合理的

????????? 自旋鎖可能導致死鎖：遞歸使用一個自旋鎖或進程獲得自旋鎖后阻塞。

? 自旋鎖使用實例，使設備只能被最多一個進程打開：

int?xxx_count?=?0;?// 定義文件打開次數(shù)計數(shù)

?

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

{

??? ...

??? spinlock(&xxx_lock);

??? if(xxx_count);??// 已經(jīng)打開

??? {

??? ??? spin_unlock(&xxx_lock);

??? ??? return?-?EBUSY;

??? }

??? xxx_count++;??// 增加使用計數(shù)

??? spin_unlock(&xxx_lock);

??? ...

??? return?0;??// 成功

}

?

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

{

??? ...

??? spinlock(&xxx_lock);

??? xxx_count--;?// 減少使用計數(shù)

??? spin_unlock(&xxx_lock);

???

??? return?0;

}

讀寫自旋鎖(rwlock)允許讀的并發(fā)。在寫操作方面，只能最多有一個寫進程，在讀操作方面，同時可以有多個讀執(zhí)行單元。當然，讀和寫也不能同時進行。

// 定義和初始化讀寫自旋鎖

rwlock_t?my_rwlock?=?RW_LOCK_UNLOCKED;??// 靜態(tài)初始化

rwlock_t?my_rwlock;

rwlock)init(&my_rwlock);??// 動態(tài)初始化

?

// 讀鎖定：在對共享資源進行讀取之前，應先調(diào)用讀鎖定函數(shù)，完成之后調(diào)用讀解鎖函數(shù)

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

?

// 寫鎖定：在對共享資源進行寫之前，應先調(diào)用寫鎖定函數(shù)，完成之后調(diào)用寫解鎖函數(shù)

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_irqsave(rwlock_t?*lock,?unsigned?long?flags);

void?write_unlock_irq(rwlock_t?*lock);

void?write_unlock_bh(rwlock_t?*lock);

? 讀寫自旋鎖一般用法：

rwlock_t?lock;??// 定義rwlock

rwlock_init(&lock);??// 初始化rwlock

?

// 讀時獲取鎖

read_lock(&lock);

...??// 臨界資源

read_unlock(&lock);

?

// 寫時獲取鎖

write_lock_irqsave(&lock,?flags);

...??// 臨界資源

write_unlock_irqrestore(&lock,?flags);

?

? 順序鎖(seqlock)是對讀寫鎖的優(yōu)化。

? 使用順序鎖，讀執(zhí)行單元不會被寫執(zhí)行單元阻塞，即讀執(zhí)行單元可以在寫執(zhí)行單元對被順序鎖保護的共享資源進行寫操作時仍然可以繼續(xù)讀，而不必等待寫執(zhí)行單元完成寫操作，寫執(zhí)行單元也不需要等待所有讀執(zhí)行單元完成讀操作才去進行寫操作。

? 寫執(zhí)行單元之間仍是互斥的。

? 若讀操作期間，發(fā)生了寫操作，必須重新讀取數(shù)據(jù)。

? 順序鎖必須要求被保護的共享資源不含有指針。

? 寫執(zhí)行單元操作：

// 獲得順序鎖

void?write_seqlock(seqlock_t?*sl);

int?write_tryseqlock(seqlock_t?*sl);

write_seqlock_irqsave(lock,?flags)

write_seqlock_irq(lock)

write_seqlock_bh()

?

// 釋放順序鎖

void?write_sequnlock(seqlock_t?*sl);

write_sequnlock_irqrestore(lock,?flags)

write_sequnlock_irq(lock)

write_sequnlock_bh()

?

// 寫執(zhí)行單元使用順序鎖的模式如下：

write_seqlock(&seqlock_a);

...??// 寫操作代碼塊

write_sequnlock(&seqlock_a);

? 讀執(zhí)行單元操作：

// 讀開始：返回順序鎖sl當前順序號

unsigned?read_seqbegin(const?seqlock_t?*sl);

read_seqbegin_irqsave(lock,?flags)

?

// 重讀：讀執(zhí)行單元在訪問完被順序鎖sl保護的共享資源后需要調(diào)用該函數(shù)來檢查，在讀訪問期間是否有寫操作。若有寫操作，重讀

int?read_seqretry(const?seqlock_t?*sl,?unsigned?iv);

read_seqretry_irqrestore(lock,?iv,?flags)

?

// 讀執(zhí)行單元使用順序鎖的模式如下：

do{

??? seqnum?=?read_seqbegin(&seqlock_a);

??? // 讀操作代碼塊?

??? ...

}while(read_seqretry(&seqlock_a,?seqnum));

?

? RCU(Read-Copy Update 讀-拷貝-更新)

? RCU可看作讀寫鎖的高性能版本，既允許多個讀執(zhí)行單元同時訪問被保護的數(shù)據(jù)，又允許多個讀執(zhí)行單元和多個寫執(zhí)行單元同時訪問被保護的數(shù)據(jù)。

? 但是RCU不能替代讀寫鎖。因為如果寫操作比較多時，對讀執(zhí)行單元的性能提高不能彌補寫執(zhí)行單元導致的損失。因為使用RCU時，寫執(zhí)行單元之間的同步開銷會比較大，它需要延遲數(shù)據(jù)結構的釋放，復制被修改的數(shù)據(jù)結構，它也必須使用某種鎖機制同步并行的其他寫執(zhí)行單元的修改操作。

? RCU操作：

// 讀鎖定

rcu_read_lock()

rcu_read_lock_bh()

?

// 讀解鎖

rcu_read_unlock()

rcu_read_unlock_bh()

?

// 使用RCU進行讀的模式如下：

rcu_read_lock()

...??// 讀臨界區(qū)

rcu_read_unlock()

? rcu_read_lock() 和rcu_read_unlock()實質(zhì)是禁止和使能內(nèi)核的搶占調(diào)度：

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

? 同步RCU

synchronize_rcu()

? 由RCU寫執(zhí)行單元調(diào)用，保證所有CPU都處理完正在運行的讀執(zhí)行單元臨界區(qū)。

?

?

?

? 信號量的使用

??? 信號量(semaphore)與自旋鎖相同，只有得到信號量才能執(zhí)行臨界區(qū)代碼，但，當獲取不到信號量時，進程不會原地打轉(zhuǎn)而是進入休眠等待狀態(tài)。

??? 信號量的操作：

//?定義信號量?

struct?semaphore?sem;?

?

//?初始化信號量:

//?初始化信號量，并設置sem的值為val

void?sema_init(struct?semaphore?*sem,?int?val);?

?

//?初始化一個用于互斥的信號量，sem的值設置為1。等同于sema_init(struct?semaphore?*sem,?1)?

void?init_MUTEX(struct?semaphore?*sem);?

?

// 等同于sema_init(struct?semaphore?*sem, 0)?

void?init_MUTEX_LOCKED(struct?semaphore?*sem);

?

// 下面兩個宏是定義并初始化信號量的“快捷方式”：

DECLEAR_MUTEX(name)?

DECLEAR_MUTEX_LOCKED(name)?

?

//? 獲得信號量：

//? 用于獲得信號量，它會導致睡眠，不能在中斷上下文使用

void?down(struct?semaphore?*sem);?

?

//?類似down(),因為down()而進入休眠的進程不能被信號打斷，而因為down_interruptible()而進入休眠的進程能被信號打斷， // 信號也會導致該函數(shù)返回，此時返回值非0

void?down_interruptible(struct?semaphore?*sem);?

?

// 嘗試獲得信號量sem，若立即獲得，它就獲得該信號量并返回0，否則，返回非0.它不會導致調(diào)用者睡眠，可在中斷上下文使用

int?down_trylock(struct?semaphore?*sem);?

?

//? 使用down_interruptible()獲取信號量時，對返回值一般會進行檢查，若非0，通常立即返回-ERESTARTSYS，如：

if(down_interruptible(&sem))?

{?

????return?-?ERESTARTSYS;?

}?

?

// 釋放信號量

// 釋放信號量sem, 喚醒等待者

void?up(struct?semaphore?*sem);?

?

// 信號量一般這樣被使用：

DECLARE_MUTEX(mount_sem);?

down(&mount_sem);??// 獲取信號量，保護臨界區(qū)

...?

critical?section??// 臨界區(qū)

...?

up(&mount_sem);??// 釋放信號量

?

?

??? Linux自旋鎖和信號量鎖采用的“獲取鎖-訪問臨界區(qū)-釋放鎖”的方式存在于幾乎所有的多任務操作系統(tǒng)之中。

??? 用信號量實現(xiàn)設備只能被一個進程打開的例子：

static?DECLEAR_MUTEX(xxx_lock)??// 定義互斥鎖

?

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

{

??? ...

??? if(down_trylock(&xxx_lock))??// 獲得打開鎖

??? ??? return?-?EBUSY; ?// 設備忙

??? ...

??? return?0;??// 成功

}

?

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

{

??? up(&xxx_lock);??// 釋放打開鎖?

??? return?0;

}

? 信號量用于同步

??? 若信號量被初始化為0，則它可以用于同步，同步意味著一個執(zhí)行單元的繼續(xù)執(zhí)行需等待另一執(zhí)行單元完成某事，保證執(zhí)行的先后順序。

? 完成量用于同步

??? 完成量(completion)提供了一種比信號量更好的同步機制，它用于一個執(zhí)行單元等待另一個執(zhí)行單元執(zhí)行完某事。

??? completion相關操作：

// 定義完成量

struct?completion?my_completion;

?

// 初始化completion

init_completion(&my_completion);

?

// 定義和初始化快捷方式：

DECLEAR_COMPLETION(my_completion);

?

// 等待一個completion被喚醒

void?wait_for_completion(struct?completion?*c);

?

// 喚醒完成量

void?cmplete(struct?completion?*c);

void?cmplete_all(struct?completion?*c);

? 自旋鎖和信號量的選擇

??? 當鎖不能被獲取時，使用信號量的開銷是進程上下文切換時間Tsw，使用自旋鎖的開銷是等待獲取自旋鎖(由臨界區(qū)執(zhí)行時間決定)Tcs，若Tcs較小，應使用自旋鎖，若Tcs較大，應使用信號量。

??? 信號量保護的臨界區(qū)可包含可能引起阻塞的代碼，而自旋鎖則絕對要避免用來保護包含這樣代碼的臨界區(qū)。因為阻塞意味著要進行進程切換，若進程被切換出去后，另一個進程企圖獲取本自旋鎖，死鎖就會發(fā)生。

??? 信號量存在于進程上下文，因此，若被保護的共享資源需要在中斷或軟中斷情況下使用，則在信號量和自旋鎖之間只能選擇自旋鎖。若一定要使用信號量，則只能通過down_trylock()方式進行，不能獲取就立即返回避免阻塞。

? 讀寫信號量

??? 讀寫信號量與信號量的關系與讀寫自旋鎖和自旋鎖的關系類似，讀寫信號量可能引起進程阻塞，但它可允許N個讀執(zhí)行單元同事訪問共享資源，而最多只能有一個寫執(zhí)行單元。

??? 讀寫自旋鎖的操作：

// 定義和初始化讀寫信號量

struct?rw_semaphore?my_res;??// 定義

void?init_rwsem(struct?rw_semaphore?*sem);??// 初始化

?

// 讀信號量獲取

void?down_read(struct?rw_semaphore?*sem);

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

?

// 讀寫信號量的使用：

rw_semaphore?rw_sem;??// 定義

init_rwsem(&rw_sem);??// 初始化

?

// 讀時獲取信號量

down_read(&rw_sem);

...??// 臨街資源

up_read(&rw_sem);

?

// 寫時獲取信號量

down_write(&rw_sem);

...??// 臨界資源

up_writer(&rw_sem);

轉(zhuǎn)載于:https://www.cnblogs.com/yangzd/archive/2010/10/16/1852975.html

總結

以上是生活随笔為你收集整理的Linux设备驱动中的并发控制总结的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。