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

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

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

i++

該操作可以轉(zhuǎn)化成下面三條機器指令序列：(1)得到當(dāng)前變量i的值并拷貝到一個寄存器中(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)

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

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

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

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

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

<>任務(wù)1

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

訪問隊列…

為隊列解除鎖共享隊列和加鎖

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

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

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

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

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

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

典型的死鎖：

四路交通堵塞

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

<>死鎖

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

防止發(fā)生饑餓

不要重復(fù)請求同一個鎖。

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

<>死鎖的避免

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

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

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

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

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

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

原子操作自旋鎖信號量

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

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

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

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

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

typedefstruct{intcounter;}atomic_t;

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

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

<>自旋鎖

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

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

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

typedefstructraw_spinlock{unsignedintslock;}raw_spinlock_t;typedef

struct

{ raw_spinlock_traw_lock;

...

}

spinlock

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

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

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

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

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

down()操作

voiddown(structsemaphore*sem){unsignedlongflags;

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

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

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

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

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

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

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

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

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

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

<>信號量與自旋鎖的比較

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

問題描述

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

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

實現(xiàn)機制①數(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ā)實例

實現(xiàn)機制②生產(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ā)實例

實現(xiàn)機制③消費者線程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ā)實例

實現(xiàn)機制④模塊的插入和刪除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è)存在這樣一個的內(nèi)核共享資源－鏈表。另外我們構(gòu)造一個內(nèi)核多任務(wù)訪問鏈表的場景：內(nèi)核線程向鏈表加入新節(jié)點；內(nèi)核定時器定時刪除結(jié)點；系統(tǒng)調(diào)用銷毀鏈表。

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

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

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

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

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

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

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

<>實現(xiàn)機制

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

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

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

<>實現(xiàn)機制

qt_task：

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

share_init：

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

share_exit：

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

<>實現(xiàn)機制

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

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

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

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

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

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

模塊注冊函數(shù)share_init<>實現(xiàn)代碼staticintshare_init(void){ inti; printk(KERN_INFO"sharelistenter\n"); INIT_WORK(&queue,kthread_launcher);//初始化工作隊列 setup_timer(&mytimer,qt_task,0);//設(shè)置定時器 add_timer(&mytimer);//添加定時器 for(i=0;i<NTHREADS;i++)/／再啟動200個內(nèi)核線程來添加節(jié)點 start_kthread(); return0;}該函數(shù)是模塊注冊函數(shù)，也是通過它啟動定時器任務(wù)和內(nèi)核線程。它首先初始化定時器任務(wù)隊列，注冊定時器任務(wù)qt_task；然后依次啟動200個內(nèi)核線程start_kthread()。至此開始對鏈表進(jìn)行操作。<>實現(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);/*向隊列中添加新節(jié)點*/

list_len++;

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

通過kernel_thread方法啟動內(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)度工作隊列*/}<>實現(xiàn)代碼

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