第七章內核中的同步臨界區和競爭狀態內核同步措施生產者-消費者并發實例內核多任務并發實例什么是臨界區（criticalregions）?就是訪問和操作共享數據的代碼段,這段代碼必須被原子地執行什么是競爭狀態?多個內核任務同時訪問同一臨界區什么是同步?避免并發和防止競爭狀態稱為同步（synchronization）

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

i++

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

<>臨界區舉例內核任務1內核任務2獲得i(1)---增加i(1->2)---寫回i(2)---

獲得i(2)

增加i(2->3)

寫回i(3)<>臨界區舉例內核任務1內核任務2獲得i(1)------獲得i(1)

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

寫回i(2)

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

可能的實際執行結果：期望的結果

當共享資源是一個復雜的數據結構時，競爭狀態往往會使該數據結構遭到破壞。

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

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

<>共享隊列和加鎖任何要訪問隊列的代碼首先都需要占住相應的鎖，這樣該鎖就能阻止來自其它內核任務的并發訪問：

<>任務1

試圖鎖定隊列成功：獲得鎖訪問隊列…為隊列解除鎖…任務2試圖鎖定隊列失敗：等待…等待…等待…成功：獲得鎖

訪問隊列…

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

找出哪些數據需要保護是關鍵所在

內核任務的局部數據僅僅被它本身訪問，顯然不需要保護

如果數據只會被特定的進程訪問，也不需加鎖

大多數內核數據結構都需要加鎖：若有其它內核任務可以訪問這些數據，那么就給這些數據加上某種形式的鎖；若任何其它東西能看到它，那么就要鎖住它

<>確定保護對象

死鎖產生的條件：有一個或多個并發執行的內核任務和一個或多個資源，每個任務都在等待其中的一個資源，但所有的資源都已經被占用。所有任務都在相互等待，但它們永遠不會釋放已經占有的資源，于是任何任務都無法繼續

典型的死鎖：

四路交通堵塞

自死鎖：一個執行任務試圖去獲得一個自己已經持有的鎖

<>死鎖

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

防止發生饑餓

不要重復請求同一個鎖。

越復雜的加鎖方案越有可能造成死鎖，因此設計應力求簡單

<>死鎖的避免

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

內核搶占——若內核具有搶占性，內核中的任務就可能會被另一任務搶占

睡眠及與用戶空間的同步——在內核執行的進程可能會睡眠，這將喚醒調度程序，導致調度一個新的用戶進程執行

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

<>并發執行的原因

為了避免并發，防止競爭。內核提供了一組同步方法來提供對共享數據的保護

原子操作自旋鎖信號量

<>內核同步措施

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

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

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

。關于atomic_t類型的定義如下：<>原子操作

typedefstruct{intcounter;}atomic_t;

自旋鎖是專為防止多處理器并發而引入的一種鎖，它在內核中大量應用于中斷處理等部分，而對于單處理器來說，可簡單采用關閉中斷的方式防止中斷處理程序的并發執行

自旋鎖最多只能被一個內核任務持有，若一個內核任務試圖請求一個已被持有的自旋鎖，那么這個任務就會一直進行忙循環，也就是旋轉，等待鎖重新可用

<>自旋鎖

設計自旋鎖的初衷是在短期間內進行輕量級的鎖定。一個被持有的自旋鎖使得請求它的任務在等待鎖重新可用期間進行自旋，所以自旋鎖不應該被持有時間過長

自旋鎖在內核中主要用來防止多處理器中并發訪問臨界區，防止內核搶占造成的競爭

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

typedefstructraw_spinlock{unsignedintslock;}raw_spinlock_t;typedef

struct

{ raw_spinlock_traw_lock;

...

}

spinlock

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

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

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

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

信號量<>信號量的相關操作

down()操作

voiddown(structsemaphore*sem){unsignedlongflags;

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

操作在關閉中斷的狀態下進行，防止多處

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

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

操作解鎖*/}<>信號量的相關操作

down()操作中__down()函數調用__down_common(),這是各種down()操作的統一函數。釋放信號量的up()操作：

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

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

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

操作進行解鎖*/}<>信號量的操作函數列表

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

<>生產者-消費者并發實例

問題描述

一個生產廠家、一個經銷商、一個倉庫。廠家生產一批產品并放在倉庫里，再通知經銷商來批發。經銷商賣完產品再向廠家下訂單，生產廠家才生產下一批產品。

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

實現機制①數據定義#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;<>生產者-消費者并發實例

實現機制②生產者線程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;}<>生產者-消費者并發實例

實現機制③消費者線程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;}<>生產者-消費者并發實例

實現機制④模塊的插入和刪除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");

假設存在這樣一個的內核共享資源－鏈表。另外我們構造一個內核多任務訪問鏈表的場景：內核線程向鏈表加入新節點；內核定時器定時刪除結點；系統調用銷毀鏈表。

上面三種內核任務并發執行時，有可能會破壞鏈表數據的完整性，所以我們必須對鏈表進行同步訪問保護，以確保數據一致性。

<>內核多任務并發控制實例系統調用：是用戶程序通過門機制來進入內核執行的內核例程，它運行在內核態，處于進程上下文中，可以認為是代表用戶進程的內核任務

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

。定時器任務隊列：任務隊列屬于下半部，在每次產生時鐘節拍時得到處理。<>內核任務及其之間的并發關系

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

<>內核任務及其之間的并發關系

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

<>實現機制

內核線程sharelist：該函數是作為內核線程由keventd調度執行的，作用是向鏈表中加入新節點

start_kthread：該函數用來構建內核線程Sharelist的封裝函數kthread_launcher，并啟動它

kthread_launcher：該函數作用僅僅是通過kernel_thread方法啟動內核線程sharelist

<>實現機制

qt_task：

該函數刪除鏈表節點，作為定時器任務運行

share_init：

該函數是我們的模塊注冊函數，也是通過它啟動定時器任務和內核線程

share_exit：

這是模塊注銷函數，負責銷毀鏈表

<>實現機制

share_exitsharelist鏈表qt_taskkeventstart_kthread進程上下文中斷上下文進程上下文downup上鎖添加節點刪除節點實現機制

并發控制實例示意圖

鏈表是內核開發中的常見數據組織形式，為了方便開發和統一結構，內核提供了一套接口來操作鏈表，我們用到的接口其主要功能為：LIST_HEAD：聲明鏈表結構list_add()：添加節點到鏈表list_del()：刪除節點list_entry()：遍歷鏈表

任務隊列結構為structtq_struct

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

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

模塊注冊函數share_init<>實現代碼staticintshare_init(void){ inti; printk(KERN_INFO"sharelistenter\n"); INIT_WORK(&queue,kthread_launcher);//初始化工作隊列 setup_timer(&mytimer,qt_task,0);//設置定時器 add_timer(&mytimer);//添加定時器 for(i=0;i<NTHREADS;i++)/／再啟動200個內核線程來添加節點 start_kthread(); return0;}該函數是模塊注冊函數，也是通過它啟動定時器任務和內核線程。它首先初始化定時器任務隊列，注冊定時器任務qt_task；然后依次啟動200個內核線程start_kthread()。至此開始對鏈表進行操作。<>實現代碼#definestaticintsharelist(void*data){ structmy_struct*p; if(count++%4==0) printk("\n"); spin_lock(&my_lock);/*添加鎖，保護共享資源*/ 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);/*向隊列中添加新節點*/

list_len++;

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

通過kernel_thread方法啟動內核線程sharelistvoidkthread_launcher(structwork_struct*q){

/*創建內核線程*/ kernel_thread(sharelist,NULL,CLONE_KERNEL|SIGCHLD); up(&sem);}

調度內核線程的start_kthreadstaticvoidstart_kthread(void){ down(&sem); schedule_work(&queue);/*調度工作隊列*/}<>實現代碼

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