第二版linux操作系統(tǒng)原理與應(yīng)用chp7

第七章內(nèi)核中的同步臨界區(qū)和競(jìng)爭(zhēng)狀態(tài)內(nèi)核同步措施生產(chǎn)者-消費(fèi)者并發(fā)實(shí)例內(nèi)核多任務(wù)并發(fā)實(shí)例什么是臨界區(qū)（criticalregions）?就是訪問和操作共享數(shù)據(jù)的代碼段,這段代碼必須被原子地執(zhí)行什么是競(jìng)爭(zhēng)狀態(tài)?多個(gè)內(nèi)核任務(wù)同時(shí)訪問同一臨界區(qū)什么是同步?避免并發(fā)和防止競(jìng)爭(zhēng)狀態(tài)稱為同步（synchronization）

<>臨界區(qū)和競(jìng)爭(zhēng)狀態(tài)考慮一個(gè)非常簡(jiǎn)單的共享資源的例子：一個(gè)全局整型變量和一個(gè)簡(jiǎn)單的臨界區(qū)，其中的操作僅僅是將整型變量的值增加1：

i++

該操作可以轉(zhuǎn)化成下面三條機(jī)器指令序列：(1)得到當(dāng)前變量i的值并拷貝到一個(gè)寄存器中(2)將寄存器中的值加1(3)把i的新值寫回到內(nèi)存中

<>臨界區(qū)舉例內(nèi)核任務(wù)1內(nèi)核任務(wù)2獲得i(1)---增加i(1->2)---寫回i(2)---

獲得i(2)

增加i(2->3)

寫回i(3)<>臨界區(qū)舉例內(nèi)核任務(wù)1內(nèi)核任務(wù)2獲得i(1)------獲得i(1)

增加i(1->2)------增加i(1->2)

寫回i(2)

------寫回i(2)

可能的實(shí)際執(zhí)行結(jié)果：期望的結(jié)果

當(dāng)共享資源是一個(gè)復(fù)雜的數(shù)據(jù)結(jié)構(gòu)時(shí)，競(jìng)爭(zhēng)狀態(tài)往往會(huì)使該數(shù)據(jù)結(jié)構(gòu)遭到破壞。

對(duì)于這種情況，鎖機(jī)制可以避免競(jìng)爭(zhēng)狀態(tài)正如門鎖和門一樣，門后的房間可想象成一個(gè)臨界區(qū)。

在一個(gè)指定時(shí)間內(nèi)，房間里只能有個(gè)一個(gè)內(nèi)核任務(wù)存在，當(dāng)一個(gè)任務(wù)進(jìn)入房間后，它會(huì)鎖住身后的房門；當(dāng)它結(jié)束對(duì)共享數(shù)據(jù)的操作后，就會(huì)走出房間，打開門鎖。如果另一個(gè)任務(wù)在房門上鎖時(shí)來了,那么它就必須等待房間內(nèi)的任務(wù)出來并打開門鎖后，才能進(jìn)入房間。

<>共享隊(duì)列和加鎖任何要訪問隊(duì)列的代碼首先都需要占住相應(yīng)的鎖，這樣該鎖就能阻止來自其它內(nèi)核任務(wù)的并發(fā)訪問：

<>任務(wù)1

試圖鎖定隊(duì)列成功：獲得鎖訪問隊(duì)列…為隊(duì)列解除鎖…任務(wù)2試圖鎖定隊(duì)列失?。旱却却却晒Γ韩@得鎖

訪問隊(duì)列…

為隊(duì)列解除鎖共享隊(duì)列和加鎖

找出哪些數(shù)據(jù)需要保護(hù)是關(guān)鍵所在

內(nèi)核任務(wù)的局部數(shù)據(jù)僅僅被它本身訪問，顯然不需要保護(hù)

如果數(shù)據(jù)只會(huì)被特定的進(jìn)程訪問，也不需加鎖

大多數(shù)內(nèi)核數(shù)據(jù)結(jié)構(gòu)都需要加鎖：若有其它內(nèi)核任務(wù)可以訪問這些數(shù)據(jù)，那么就給這些數(shù)據(jù)加上某種形式的鎖；若任何其它東西能看到它，那么就要鎖住它

<>確定保護(hù)對(duì)象

死鎖產(chǎn)生的條件：有一個(gè)或多個(gè)并發(fā)執(zhí)行的內(nèi)核任務(wù)和一個(gè)或多個(gè)資源，每個(gè)任務(wù)都在等待其中的一個(gè)資源，但所有的資源都已經(jīng)被占用。所有任務(wù)都在相互等待，但它們永遠(yuǎn)不會(huì)釋放已經(jīng)占有的資源，于是任何任務(wù)都無法繼續(xù)

典型的死鎖：

四路交通堵塞

自死鎖：一個(gè)執(zhí)行任務(wù)試圖去獲得一個(gè)自己已經(jīng)持有的鎖

<>死鎖

加鎖的順序是關(guān)鍵。使用嵌套的鎖時(shí)必須保證以相同的順序獲取鎖，這樣可以阻止致命擁抱類型的死鎖

防止發(fā)生饑餓

不要重復(fù)請(qǐng)求同一個(gè)鎖。

越復(fù)雜的加鎖方案越有可能造成死鎖，因此設(shè)計(jì)應(yīng)力求簡(jiǎn)單

<>死鎖的避免

中斷——中斷幾乎可以在任何時(shí)刻異步發(fā)生，也可能隨時(shí)打斷正在執(zhí)行的代碼。

內(nèi)核搶占——若內(nèi)核具有搶占性，內(nèi)核中的任務(wù)就可能會(huì)被另一任務(wù)搶占

睡眠及與用戶空間的同步——在內(nèi)核執(zhí)行的進(jìn)程可能會(huì)睡眠，這將喚醒調(diào)度程序，導(dǎo)致調(diào)度一個(gè)新的用戶進(jìn)程執(zhí)行

對(duì)稱多處理——兩個(gè)或多個(gè)處理器可以同時(shí)執(zhí)行代碼

<>并發(fā)執(zhí)行的原因

為了避免并發(fā)，防止競(jìng)爭(zhēng)。內(nèi)核提供了一組同步方法來提供對(duì)共享數(shù)據(jù)的保護(hù)

原子操作自旋鎖信號(hào)量

<>內(nèi)核同步措施

原子操作可以保證指令以原子的方式被執(zhí)行

兩個(gè)原子操作絕對(duì)不可能并發(fā)地訪問同一個(gè)變量

Linux內(nèi)核提供了一個(gè)專門的atomic_t類型（一個(gè)24位原子訪問計(jì)數(shù)器）和一些專門的函數(shù)，這些函數(shù)作用于atomic_t類型的變量

。關(guān)于atomic_t類型的定義如下：<>原子操作

typedefstruct{intcounter;}atomic_t;

自旋鎖是專為防止多處理器并發(fā)而引入的一種鎖，它在內(nèi)核中大量應(yīng)用于中斷處理等部分，而對(duì)于單處理器來說，可簡(jiǎn)單采用關(guān)閉中斷的方式防止中斷處理程序的并發(fā)執(zhí)行

自旋鎖最多只能被一個(gè)內(nèi)核任務(wù)持有，若一個(gè)內(nèi)核任務(wù)試圖請(qǐng)求一個(gè)已被持有的自旋鎖，那么這個(gè)任務(wù)就會(huì)一直進(jìn)行忙循環(huán)，也就是旋轉(zhuǎn)，等待鎖重新可用

<>自旋鎖

設(shè)計(jì)自旋鎖的初衷是在短期間內(nèi)進(jìn)行輕量級(jí)的鎖定。一個(gè)被持有的自旋鎖使得請(qǐng)求它的任務(wù)在等待鎖重新可用期間進(jìn)行自旋，所以自旋鎖不應(yīng)該被持有時(shí)間過長(zhǎng)

自旋鎖在內(nèi)核中主要用來防止多處理器中并發(fā)訪問臨界區(qū)，防止內(nèi)核搶占造成的競(jìng)爭(zhēng)

自旋鎖不允許任務(wù)睡眠，持有自旋鎖的任務(wù)睡眠會(huì)造成自死鎖，因此自旋鎖能夠在中斷上下文中使用<>自旋鎖自旋鎖的定義如下：<>自旋鎖

typedefstructraw_spinlock{unsignedintslock;}raw_spinlock_t;typedef

struct

{ raw_spinlock_traw_lock;

...

}

spinlock

t;使用自旋鎖的基本形式如下：DEFINE_SPINLOCK(mr_lock);/*定義一個(gè)自旋鎖*/spin_lock(&mr_lock);/*臨界區(qū)*/spin_unlock(&mr_lock);

Linux中的信號(hào)量是一種睡眠鎖。若有一個(gè)任務(wù)試圖獲得一個(gè)已被持有的信號(hào)量時(shí)，信號(hào)量會(huì)將其推入等待隊(duì)列，然后讓其睡眠。這時(shí)處理器獲得自由而去執(zhí)行其它代碼。當(dāng)持有信號(hào)量的進(jìn)程將信號(hào)量釋放后，在等待隊(duì)列中的一個(gè)任務(wù)將被喚醒，從而便可以獲得這個(gè)信號(hào)量

信號(hào)量具有睡眠特性，適用于鎖會(huì)被長(zhǎng)時(shí)間持有的情況，只能在進(jìn)程上下文中使用<>信號(hào)量信號(hào)量的使用<>信號(hào)量的定義structsemaphore{spinlock_tlock;unsignedintcount;structlist_headwait_list;}staticDECLARE_MUTEX(mr_sem);／*聲明并初始化互斥信號(hào)量*/if(!down_interruptible(&mr_sem))/*信號(hào)被接收,信號(hào)量還未獲取*/

/*臨界區(qū)…*/up(&mr_sem);

信號(hào)量<>信號(hào)量的相關(guān)操作

down()操作

voiddown(structsemaphore*sem){unsignedlongflags;

spin_lock_irqsave(&sem->lock,flags);/*加鎖，使信號(hào)量的

操作在關(guān)閉中斷的狀態(tài)下進(jìn)行，防止多處

理器并發(fā)操作造成錯(cuò)誤*/if(sem->count>0))/*若信號(hào)量可用，則將引用計(jì)數(shù)減1*/sem->count--;else/*如果無信號(hào)量可用，則調(diào)用__down()函數(shù)進(jìn)入睡眠

等待狀態(tài)*/__down(sem);spin_unlock_irqrestore(&sem->lock,flags);/*對(duì)信號(hào)量的

操作解鎖*/}<>信號(hào)量的相關(guān)操作

down()操作中__down()函數(shù)調(diào)用__down_common(),這是各種down()操作的統(tǒng)一函數(shù)。釋放信號(hào)量的up()操作：

voidup(structsemaphore*sem){unsignedlongflags;spin_lock_irqsave(&sem->lock,flags);/*對(duì)信號(hào)量操作

進(jìn)行加鎖*/iflist_empty(&sem->wait_list)/*如果該信號(hào)量的等待

隊(duì)列為空，則釋放信號(hào)量*/sem->count++;else/*否則喚醒該信號(hào)量的等待隊(duì)列隊(duì)頭的進(jìn)程*/__up(sem);spin_unlock_irqrestore(&sem->lock,flags);/*對(duì)信號(hào)量

操作進(jìn)行解鎖*/}<>信號(hào)量的操作函數(shù)列表

<>信號(hào)量與自旋鎖的比較

<>生產(chǎn)者-消費(fèi)者并發(fā)實(shí)例

問題描述

一個(gè)生產(chǎn)廠家、一個(gè)經(jīng)銷商、一個(gè)倉(cāng)庫(kù)。廠家生產(chǎn)一批產(chǎn)品并放在倉(cāng)庫(kù)里，再通知經(jīng)銷商來批發(fā)。經(jīng)銷商賣完產(chǎn)品再向廠家下訂單，生產(chǎn)廠家才生產(chǎn)下一批產(chǎn)品。

問題分析 生產(chǎn)廠家相當(dāng)于“生產(chǎn)者”，經(jīng)銷商相當(dāng)于“消費(fèi)者”，倉(cāng)庫(kù)則為“公共緩沖區(qū)”。該問題屬于單一生產(chǎn)者，單一消費(fèi)者，單一緩沖區(qū)。這是典型的進(jìn)程同步問題。生產(chǎn)者和消費(fèi)者是不同的線程，“公共緩沖區(qū)”為臨界區(qū)。同一時(shí)刻，只能有一個(gè)線程訪問臨界區(qū)。<>生產(chǎn)者-消費(fèi)者并發(fā)實(shí)例

實(shí)現(xiàn)機(jī)制①數(shù)據(jù)定義#include<linux/init.h>#include<linux/module.h>#include<linux/semaphore.h>#include<linux/sched.h>#include<asm/atomic.h>#include<linux/delay.h>#definePRODUCT_NUMS10staticstructsemaphoresem_producer;staticstructsemaphoresem_consumer;staticcharproduct[12];staticatomic_tnum;staticintproducer(void*product);staticintconsumer(void*product);staticintid=1;staticintconsume_num=1;<>生產(chǎn)者-消費(fèi)者并發(fā)實(shí)例

實(shí)現(xiàn)機(jī)制②生產(chǎn)者線程staticintproducer(void*p){ char*product=(char*)p; inti; atomic_inc(&num); printk("producer[%d]start...\n",current->pid); for(i=0;i<PRODUCT_NUMS;i++){

down(&sem_producer);

snprintf(product,12,"2010-01-%d",id++);

printk("producer[%d]produce%s\n",current->pid,product);

up(&sem_consumer); } printk("producer[%d]exit...\n",current->pid); return0;}<>生產(chǎn)者-消費(fèi)者并發(fā)實(shí)例

實(shí)現(xiàn)機(jī)制③消費(fèi)者線程staticintconsumer(void*p){ char*product=(char*)p; printk("consumer[%d]start...\n",current->pid); for(;;){

msleep(100);

down_interruptible(&sem_consumer);

if(consume_num>=PRODUCT_NUMS*atomic_read(&num)) break;

printk("consumer[%d]consume%s\n",current->pid,product);

consume_num++;

memset(product,'\0',12);

up(&sem_producer); } printk("consumer[%d]exit...\n",current->pid); return0;}<>生產(chǎn)者-消費(fèi)者并發(fā)實(shí)例

實(shí)現(xiàn)機(jī)制④模塊的插入和刪除staticintprocon_init(void){ printk(KERN_INFO"showproducerandconsumer\n"); init_MUTEX(&sem_producer); init_MUTEX_LOCKED(&sem_consumer); atomic_set(&num,0); kernel_thread(producer,product,CLONE_KERNEL); kernel_thread(consumer,product,CLONE_KERNEL); return0;}staticvoidprocon_exit(void){ printk(KERN_INFO"exitproducerandconsumer\n");}module_init(procon_init);module_exit(procon_exit);MODULE_LICENSE("GPL");MODULE_DESCRIPTION("producerandconsumerModule");MODULE_ALIAS("asimplestmodule");

假設(shè)存在這樣一個(gè)的內(nèi)核共享資源－鏈表。另外我們構(gòu)造一個(gè)內(nèi)核多任務(wù)訪問鏈表的場(chǎng)景：內(nèi)核線程向鏈表加入新節(jié)點(diǎn)；內(nèi)核定時(shí)器定時(shí)刪除結(jié)點(diǎn)；系統(tǒng)調(diào)用銷毀鏈表。

上面三種內(nèi)核任務(wù)并發(fā)執(zhí)行時(shí)，有可能會(huì)破壞鏈表數(shù)據(jù)的完整性，所以我們必須對(duì)鏈表進(jìn)行同步訪問保護(hù)，以確保數(shù)據(jù)一致性。

<>內(nèi)核多任務(wù)并發(fā)控制實(shí)例系統(tǒng)調(diào)用：是用戶程序通過門機(jī)制來進(jìn)入內(nèi)核執(zhí)行的內(nèi)核例程，它運(yùn)行在內(nèi)核態(tài)，處于進(jìn)程上下文中，可以認(rèn)為是代表用戶進(jìn)程的內(nèi)核任務(wù)

內(nèi)核線程：內(nèi)核線程可以理解成在內(nèi)核中運(yùn)行的特殊進(jìn)程，它有自己的“進(jìn)程上下文”

。定時(shí)器任務(wù)隊(duì)列：任務(wù)隊(duì)列屬于下半部，在每次產(chǎn)生時(shí)鐘節(jié)拍時(shí)得到處理。<>內(nèi)核任務(wù)及其之間的并發(fā)關(guān)系

系統(tǒng)調(diào)用和內(nèi)核線程可能和各種內(nèi)核任務(wù)并發(fā)執(zhí)行，除了中斷（定時(shí)器任務(wù)隊(duì)列屬于軟中斷范疇）搶占它產(chǎn)生并發(fā)外，它們還有可能自發(fā)性地主動(dòng)睡眠（比如在一些阻塞性的操作中），于是放棄處理器，從而重新調(diào)度其它任務(wù)，所以系統(tǒng)調(diào)用和內(nèi)核線程除與定時(shí)器任務(wù)隊(duì)列發(fā)生競(jìng)爭(zhēng)，也會(huì)與其他（包括自己）系統(tǒng)調(diào)用與內(nèi)核線程發(fā)生競(jìng)爭(zhēng)。

<>內(nèi)核任務(wù)及其之間的并發(fā)關(guān)系

主要的共享資源是鏈表（mine）,操作它的內(nèi)核任務(wù)有三個(gè)：一是200個(gè)內(nèi)核線程（sharelist），它們負(fù)責(zé)從表頭將新節(jié)點(diǎn)（structmy_struct）插入鏈表。二是定時(shí)器任務(wù)(qt_task)，它負(fù)責(zé)每個(gè)時(shí)鐘節(jié)拍時(shí)從鏈表頭刪除一個(gè)節(jié)點(diǎn)。三是系統(tǒng)調(diào)用share_exit，它負(fù)責(zé)銷毀鏈表并卸載模塊。

<>實(shí)現(xiàn)機(jī)制

內(nèi)核線程sharelist：該函數(shù)是作為內(nèi)核線程由keventd調(diào)度執(zhí)行的，作用是向鏈表中加入新節(jié)點(diǎn)

start_kthread：該函數(shù)用來構(gòu)建內(nèi)核線程Sharelist的封裝函數(shù)kthread_launcher，并啟動(dòng)它

kthread_launcher：該函數(shù)作用僅僅是通過kernel_thread方法啟動(dòng)內(nèi)核線程sharelist

<>實(shí)現(xiàn)機(jī)制

qt_task：

該函數(shù)刪除鏈表節(jié)點(diǎn)，作為定時(shí)器任務(wù)運(yùn)行

share_init：

該函數(shù)是我們的模塊注冊(cè)函數(shù)，也是通過它啟動(dòng)定時(shí)器任務(wù)和內(nèi)核線程

share_exit：

這是模塊注銷函數(shù)，負(fù)責(zé)銷毀鏈表

<>實(shí)現(xiàn)機(jī)制

share_exitsharelist鏈表qt_taskkeventstart_kthread進(jìn)程上下文中斷上下文進(jìn)程上下文downup上鎖添加節(jié)點(diǎn)刪除節(jié)點(diǎn)實(shí)現(xiàn)機(jī)制

并發(fā)控制實(shí)例示意圖

鏈表是內(nèi)核開發(fā)中的常見數(shù)據(jù)組織形式，為了方便開發(fā)和統(tǒng)一結(jié)構(gòu)，內(nèi)核提供了一套接口來操作鏈表，我們用到的接口其主要功能為：LIST_HEAD：聲明鏈表結(jié)構(gòu)list_add()：添加節(jié)點(diǎn)到鏈表list_del()：刪除節(jié)點(diǎn)list_entry()：遍歷鏈表

任務(wù)隊(duì)列結(jié)構(gòu)為structtq_struct

kill_proc，該函數(shù)在模塊注銷時(shí)被調(diào)用，其主要作用有兩個(gè)：第一殺死我們生成的內(nèi)核線程；第二告訴keventd回收相關(guān)子線程，以免產(chǎn)生“殘疾”子線程序<>關(guān)鍵代碼解釋

變量聲明<>實(shí)現(xiàn)代碼#defineNTHREADS200/*線程數(shù)*/structmy_struct{ structlist_headlist; intid; intpid;};staticstructwork_structqueue;/*定義工作隊(duì)列*/staticstructtimer_listmytimer;/*定時(shí)器隊(duì)列*/staticLIST_HEAD(mine);/*sharelist頭*/staticunsignedintlist_len=0;staticDECLARE_MUTEX(sem);/*內(nèi)核線程進(jìn)行同步的信號(hào)量*/staticspinlock_tmy_lock=SPIN_LOCK_UNLOCKED;/*保護(hù)對(duì)鏈表的操作*/staticatomic_tmy_count=ATOMIC_INIT(0);/*以原子方式進(jìn)行追加*/staticlongcount=0;/*行計(jì)數(shù)器，每行打印4個(gè)信息*/staticinttimer_over=0;/*定時(shí)器結(jié)束標(biāo)志*/staticintsharelist(void*data);/*從共享鏈表增刪節(jié)點(diǎn)的線程*/staticvoidkthread_launcher(structwork_struct*q);/*創(chuàng)建內(nèi)核線程*/staticvoidstart_kthread(void);/*調(diào)度內(nèi)核線程*/

模塊注冊(cè)函數(shù)share_init<>實(shí)現(xiàn)代碼staticintshare_init(void){ inti; printk(KERN_INFO"sharelistenter\n"); INIT_WORK(&queue,kthread_launcher);//初始化工作隊(duì)列 setup_timer(&mytimer,qt_task,0);//設(shè)置定時(shí)器 add_timer(&mytimer);//添加定時(shí)器 for(i=0;i<NTHREADS;i++)/／再啟動(dòng)200個(gè)內(nèi)核線程來添加節(jié)點(diǎn) start_kthread(); return0;}該函數(shù)是模塊注冊(cè)函數(shù)，也是通過它啟動(dòng)定時(shí)器任務(wù)和內(nèi)核線程。它首先初始化定時(shí)器任務(wù)隊(duì)列，注冊(cè)定時(shí)器任務(wù)qt_task；然后依次啟動(dòng)200個(gè)內(nèi)核線程start_kthread()。至此開始對(duì)鏈表進(jìn)行操作。<>實(shí)現(xiàn)代碼#definestaticintsharelist(void*data){ structmy_struct*p; if(count++%4==0) printk("\n"); spin_lock(&my_lock);/*添加鎖，保護(hù)共享資源*/ if(list_len<100){

if((p=kmalloc(sizeof(structmy_struct),GFP_KERNEL))==NULL) return-ENOMEM;

p->id=atomic_read(&my_count);/*原子變量操作*/

atomic_inc(&my_count);

p->pid=current->pid;

list_add(&p->list,&mine);/*向隊(duì)列中添加新節(jié)點(diǎn)*/

list_len++;

printk("THREADADD:%-5d\t",p->id); }else{/*隊(duì)列超過定長(zhǎng)則刪除節(jié)點(diǎn)*/ structmy_struct*my=NULL; my=list_entry(mine.prev,structmy_struct,list); list_del(mine.prev);/*從隊(duì)列尾部刪除節(jié)點(diǎn)*/ list_len--; printk("THREADDEL:%-5d\t",my->id); kfree(my);} spin_unlock(&my_lock); return0;}對(duì)共享鏈表操作的內(nèi)核線程share_list<>實(shí)現(xiàn)代碼

通過kernel_thread方法啟動(dòng)內(nèi)核線程sharelistvoidkthread_launcher(structwork_struct*q){

/*創(chuàng)建內(nèi)核線程*/ kernel_thread(sharelist,NULL,CLONE_KERNEL|SIGCHLD); up(&sem);}

調(diào)度內(nèi)核線程的start_kthreadstaticvoidstart_kthread(void){ down(&sem); schedule_work(&queue);/*調(diào)度工作隊(duì)列*/}<>實(shí)現(xiàn)代碼

刪除結(jié)點(diǎn)的定時(shí)器任務(wù)qt_taskvoidqt_task(unsignedlongdata){ if(!list_empty(&mine)){ structmy_struct