Posted on 2008-04-10 23:07
silentneil 閱讀(185)
評(píng)論(0) 編輯 收藏 引用
本文將Linux內(nèi)核中用于同步的幾種機(jī)制集中起來(lái)分析，強(qiáng)調(diào)了它們之間在實(shí)現(xiàn)和使用上的不同。
同步通常是為了達(dá)到多線程協(xié)同的目的而設(shè)計(jì)的一種機(jī)制，通常包含異步信號(hào)機(jī)制和互斥機(jī)制作為其實(shí)現(xiàn)的底層。在Linux 2.4內(nèi)核中也有相應(yīng)的技術(shù)實(shí)現(xiàn)，包括信號(hào)量、自旋鎖、原子操作和等待隊(duì)列，其中原子操作和等待隊(duì)列又是實(shí)現(xiàn)信號(hào)量的底層。
等待隊(duì)列和異步信號(hào)
wait queue很早就作為一個(gè)基本的功能單位出現(xiàn)在Linux內(nèi)核里了，它以隊(duì)列為基礎(chǔ)數(shù)據(jù)結(jié)構(gòu)，與進(jìn)程調(diào)度機(jī)制緊密結(jié)合，能夠用于實(shí)現(xiàn)核心的異步事件通知機(jī)制。我們從它的使用范例著手，看看等待隊(duì)列是如何實(shí)現(xiàn)異步信號(hào)功能的。
在核心運(yùn)行過(guò)程中，經(jīng)常會(huì)因?yàn)槟承l件不滿(mǎn)足而需要掛起當(dāng)前線程，直至條件滿(mǎn)足了才繼續(xù)執(zhí)行。在2.4內(nèi)核中提供了一組新接口來(lái)實(shí)現(xiàn)這樣的功能，下面的代碼節(jié)選自kernel/printk.c：
unsigned long log_size;
1: DECLARE_WAIT_QUEUE_HEAD(log_wait);...
4: spinlock_t console_lock = SPIN_LOCK_UNLOCKED;...
int do_syslog(int type,char *buf,int len){
...
2: error=wait_event_interruptible(log_wait,log_size);
if(error)
goto out;
...
5: spin_lock_irq(&console_lock);
...
log_size--;
...
6: spin_unlock_irq(&console_lock);
...
}
asmlinkage int printk(const char *fmt,...){
...
7: spin_lock_irqsave(&console_lock,flags);
...
log_size++;...
8: spin_unlock_irqrestore(&console_lock,flags);
3: wake_up_interruptible(&log_wait);
...
}
這段代碼實(shí)現(xiàn)了printk調(diào)用和syslog之間的同步，syslog需要等待printk送數(shù)據(jù)到緩沖區(qū)，因此，在2:處等待log_size非0；而 printk一邊傳送數(shù)據(jù)，一邊增加log_size的值，完成后喚醒在log_wait上等待的所有線程（這個(gè)線程不是用戶(hù)空間的線程概念，而是核內(nèi)的一個(gè)執(zhí)行序列）。執(zhí)行了3:的wake_up_interruptible()后，2:處的wait_event_interruptible()返回 0，從而進(jìn)入syslog的實(shí)際動(dòng)作。
1:是定義log_wait全局變量的宏調(diào)用。
在實(shí)際操作log_size全局變量的時(shí)候，還使用了spin_lock自旋鎖來(lái)實(shí)現(xiàn)互斥，關(guān)于自旋鎖，這里暫不作解釋?zhuān)珡倪@段代碼中已經(jīng)可以清楚的知道它的使用方法了。
所有wait queue使用上的技巧體現(xiàn)在wait_event_interruptible()的實(shí)現(xiàn)上，代碼位于include/linux/sched.h中，前置數(shù)字表示行號(hào)：
779 #define __wait_event_interruptible(wq, condition, ret) \
780 do { \
781 wait_queue_t __wait; \
782 init_waitqueue_entry(&__wait, current); \
783 \
784 add_wait_queue(&wq, &__wait); \
785 for (;;) { \
786 set_current_state(TASK_INTERRUPTIBLE); \
787 if (condition) \
788 break; \
789 if (!signal_pending(current)) { \
790 schedule(); \
791 continue; \
792 } \
793 ret = -ERESTARTSYS; \
794 break; \
795 } \
796 current->state = TASK_RUNNING; \
797 remove_wait_queue(&wq, &__wait); \
798 } while (0)
799
800 #define wait_event_interruptible(wq, condition) \
801 ({ \
802 int __ret = 0; \
803 if (!(condition)) \
804 __wait_event_interruptible(wq, condition, __ret); \
805 __ret; \
806 })
在wait_event_interruptible ()中首先判斷condition是不是已經(jīng)滿(mǎn)足，如果是則直接返回0，否則調(diào)用__wait_event_interruptible()，并用 __ret來(lái)存放返回值。__wait_event_interruptible()首先定義并初始化一個(gè)wait_queue_t變量__wait，其中數(shù)據(jù)為當(dāng)前進(jìn)程結(jié)構(gòu)current（struct task_struct），并把__wait入隊(duì)。在無(wú)限循環(huán)中，__wait_event_interruptible()將本進(jìn)程置為可中斷的掛起狀態(tài)，反復(fù)檢查condition是否成立，如果成立則退出，如果不成立則繼續(xù)休眠；條件滿(mǎn)足后，即把本進(jìn)程運(yùn)行狀態(tài)置為運(yùn)行態(tài)，并將__wait從等待隊(duì)列中清除掉，從而進(jìn)程能夠調(diào)度運(yùn)行。如果進(jìn)程當(dāng)前有異步信號(hào)（POSIX的），則返回-ERESTARTSYS。
掛起的進(jìn)程并不會(huì)自動(dòng)轉(zhuǎn)入運(yùn)行的，因此，還需要一個(gè)喚醒動(dòng)作，這個(gè)動(dòng)作由wake_up_interruptible()完成，它將遍歷作為參數(shù)傳入的 log_wait等待隊(duì)列，將其中所有的元素（通常都是task_struct）置為運(yùn)行態(tài)，從而可被調(diào)度到，執(zhí)行 __wait_event_interruptible()中的代碼。
DECLARE_WAIT_QUEUE_HEAD(log_wait)經(jīng)過(guò)宏展開(kāi)后就是定義了一個(gè)log_wait等待隊(duì)列頭變量：
struct __wait_queue_head log_wait = {
lock: SPIN_LOCK_UNLOCKED,
task_list: { &log_wait.task_list, &log_wait.task_list }
}
其中task_list是struct list_head變量，包括兩個(gè)list_head指針，一個(gè)next、一個(gè)prev，這里把它們初始化為自身，屬于隊(duì)列實(shí)現(xiàn)上的技巧，其細(xì)節(jié)可以參閱關(guān)于內(nèi)核list數(shù)據(jù)結(jié)構(gòu)的討論，add_wait_queue()和remove_wait_queue()就等同于list_add()和 list_del()。
wait_queue_t結(jié)構(gòu)在include/linux/wait.h中定義，關(guān)鍵元素即為一個(gè)struct task_struct變量表征當(dāng)前進(jìn)程。
除了wait_event_interruptible()/wake_up_interruptible()以外，與此相對(duì)應(yīng)的還有wait_event ()和wake_up()接口，interruptible是更安全、更常用的選擇，因?yàn)榭芍袛嗟牡却梢越邮招盘?hào)，從而掛起的進(jìn)程允許被外界kill。
wait_event* ()接口是2.4內(nèi)核引入并推薦使用的，在此之前，最常用的等待操作是interruptible_sleep_on (wait_queue_head_t *wq)，當(dāng)然，與此配套的還有不可中斷版本sleep_on()，另外，還有帶有超時(shí)控制的*sleep_on_timeout()。sleep_on 系列函數(shù)的語(yǔ)義比wait_event簡(jiǎn)單，沒(méi)有條件判斷功能，其余動(dòng)作與wait_event完全相同，也就是說(shuō)，我們可以用 interruptible_sleep_on()來(lái)實(shí)現(xiàn)wait_event_interruptible()（僅作示意〉：
do{
interruptible_sleep_on(&log_wait);
if(condition)
break;
}while(1);
相對(duì)而言，這種操作序列有反復(fù)的入隊(duì)、出隊(duì)動(dòng)作，更加耗時(shí)，而很大一部分等待操作的確是需要判斷一個(gè)條件是否滿(mǎn)足的，因此2.4才推薦使用wait_event接口。
在wake_up系列接口中，還有一類(lèi)wake_up_sync()和wake_up_interruptible_sync()接口，保證調(diào)度在wake_up返回之后進(jìn)行。
回頁(yè)首
原子操作和信號(hào)量
POSIX 有信號(hào)量，SysV IPC有信號(hào)量，核內(nèi)也有信號(hào)量，接口很簡(jiǎn)單，一個(gè)down()，一個(gè)up()，分別對(duì)應(yīng)P操作和V操作，down()調(diào)用可能引起線程掛起，因此和 sleep_on類(lèi)似，也有interruptible系列接口。down意味著信號(hào)量減1，up意味著信號(hào)量加1，這兩個(gè)操作顯然需要互斥。在 Linux 2.4中，并沒(méi)有如想象中的用鎖實(shí)現(xiàn)，而是使用了原子操作。
在include/asm/atomic.h中定義了一系列原子操作，包括原子讀、原子寫(xiě)、原子加等等，大多是直接用匯編語(yǔ)句來(lái)實(shí)現(xiàn)的，這里就不詳細(xì)解釋。
我們從信號(hào)量數(shù)據(jù)結(jié)構(gòu)開(kāi)始，它定義在include/asm/semaphore.h中：
struct semaphore {
atomic_t count;
int sleepers;
wait_queue_head_t wait;
}
down ()操作可以理解為申請(qǐng)資源，up()操作可以理解為釋放資源，因此，信號(hào)量實(shí)際表示的是資源的數(shù)量以及是否有進(jìn)程正在等待。在semaphore結(jié)構(gòu)中，count相當(dāng)于資源計(jì)數(shù)，為正數(shù)或0時(shí)表示可用資源數(shù)，-1則表示沒(méi)有空閑資源且有等待進(jìn)程。而等待進(jìn)程的數(shù)量并不關(guān)心。這種設(shè)計(jì)主要是考慮與信號(hào)量的原語(yǔ)相一致，當(dāng)某個(gè)進(jìn)程執(zhí)行up()函數(shù)釋放資源，點(diǎn)亮信號(hào)燈時(shí)，如果count恢復(fù)到0，則表示尚有進(jìn)程在等待該資源，因此執(zhí)行喚醒操作。一個(gè)典型的down()-up()流程是這樣的：
down()-->count做原子減1操作，如果結(jié)果不小于0則表示成功申請(qǐng)，從down()中返回；
-->如果結(jié)果為負(fù)（實(shí)際上只可能是-1），則表示需要等待，則調(diào)用__down_fail()；
__down_fail()調(diào)用__down()，__down()用C代碼實(shí)現(xiàn)，要求已不如down()和__down_fail()嚴(yán)格，在此作實(shí)際的等待（arch/i386/kernel/semaphore.c）：
void __down(struct semaphore * sem)
{
struct task_struct *tsk = current;
DECLARE_WAITQUEUE(wait, tsk);
tsk->state = TASK_UNINTERRUPTIBLE;
add_wait_queue_exclusive(&sem->wait, &wait);
spin_lock_irq(&semaphore_lock);
sem->sleepers++;
for (;;) {
int sleepers = sem->sleepers;
/*
* Add "everybody else" into it. They aren't
* playing, because we own the spinlock.
*/
if (!atomic_add_negative(sleepers - 1, &sem->count)) {
sem->sleepers = 0;
break;
}
sem->sleepers = 1; /* us - see -1 above */
spin_unlock_irq(&semaphore_lock);
schedule();
tsk->state = TASK_UNINTERRUPTIBLE;
spin_lock_irq(&semaphore_lock);
}
spin_unlock_irq(&semaphore_lock);
remove_wait_queue(&sem->wait, &wait);
tsk->state = TASK_RUNNING;
wake_up(&sem->wait);
}
__down()-->當(dāng)前進(jìn)程進(jìn)入wait等待隊(duì)列，狀態(tài)為不可中斷的掛起，sleepers++，如果這是第一次申請(qǐng)失敗，則sleepers值為1，否則為2--這個(gè)設(shè)置純粹是為了下面這句原子加而安排的。
在真正進(jìn)入休眠以前，__down()還是需要判斷一下是不是確實(shí)沒(méi)有資源可用，因?yàn)樵趕pin_lock之前什么都可能發(fā)生。 atomic_add_negative()將sleepers-1（只可能是0或者1，分別表示僅有一個(gè)等待進(jìn)程或是多個(gè)）加到count（如果有多個(gè)進(jìn)程申請(qǐng)資源失敗進(jìn)入__down()，count可能為-2、-3等）之上，這個(gè)加法完成后，結(jié)果為0只可能是在sleepers等于1的時(shí)候發(fā)生（因?yàn)槿绻鹲leepers等于2，表示有多個(gè)進(jìn)程執(zhí)行了down()，則count必然小于-1，因此sleepers-1+count必然小于0），表示 count在此之前已經(jīng)變?yōu)?了，也就是說(shuō)已經(jīng)有進(jìn)程釋放了資源，因此本進(jìn)程不用休眠而是獲得資源退出__down()，從而也從down()中返回；如果沒(méi)有進(jìn)程釋放資源，那么在所有等待進(jìn)程的這一加法完成后，count將等于-1。因此，從down()調(diào)用外面看（無(wú)論是在down()中休眠還是獲得資源離開(kāi)down()），count為負(fù)時(shí)只可能為-1（此時(shí)sleepers等于1），這么設(shè)計(jì)使得up()操作只需要對(duì)count加1，判斷是否為0 就可以知道是否有必要執(zhí)行喚醒操作__up_wakeup()了。
獲得了資源的進(jìn)程將把sleepers設(shè)為 0，并喚醒所有其他等待進(jìn)程，這個(gè)操作實(shí)際上只是起到恢復(fù)count為-1，并使它們?cè)俅芜M(jìn)入休眠的作用，因?yàn)榈谝粋€(gè)被喚醒的等待進(jìn)程執(zhí)行 atomic_add_negative()操作后會(huì)將count恢復(fù)為-1，然后將sleepers置為1；以后的等待進(jìn)程則會(huì)像往常一樣重新休眠。
將down()操作設(shè)計(jì)得如此復(fù)雜的原因和結(jié)果就是up操作相當(dāng)簡(jiǎn)單。up()利用匯編原子地將count加1，如果小于等于0表示有等待進(jìn)程，則調(diào)用__up_wakeup()-->__up()喚醒wait；否則直接返回。
在down()中競(jìng)爭(zhēng)獲得資源的進(jìn)程并不是按照優(yōu)先級(jí)排序的，只是在up()操作完成后第一個(gè)被喚醒或者正在__down()中運(yùn)行而暫未進(jìn)入休眠的進(jìn)程成功的可能性稍高一些。
盡管可以將信號(hào)量的count初始化為1從而實(shí)現(xiàn)一種互斥鎖（mutex），但Linux并不保證這個(gè)count不會(huì)超過(guò)1，因?yàn)閡p操作并不考慮 count的初值，所以只能依靠程序員自己來(lái)控制不要無(wú)謂的執(zhí)行up()從而破壞mutex的語(yǔ)義。相關(guān)的初始化接口定義在 include/asm/semaphore.h中，但一般程序員可以通過(guò)sema_init()接口來(lái)初始化信號(hào)量：
#define DECLARE_MUTEX(name) __DECLARE_SEMAPHORE_GENERIC(name,1)
#define DECLARE_MUTEX_LOCKED(name) __DECLARE_SEMAPHORE_GENERIC(name,0)
static inline void sema_init (struct semaphore *sem, int val)
static inline void init_MUTEX (struct semaphore *sem)
static inline void init_MUTEX_LOCKED (struct semaphore *sem)
除了down()以外，Linux還提供了一個(gè)down_interruptible()，操作序列與down()基本相同，僅在休眠狀態(tài)為可中斷和信號(hào)處理上有所不同。在標(biāo)準(zhǔn)的信號(hào)量以外，還有一套讀寫(xiě)信號(hào)量，用于將資源的讀寫(xiě)區(qū)分開(kāi)來(lái)進(jìn)行同步以提高效率，采用讀寫(xiě)鎖來(lái)實(shí)現(xiàn)，有興趣的可以參閱文后列出的參考資料。
回頁(yè)首
自旋鎖
鎖是一個(gè)概念，正如上面提到的mutex互斥鎖僅僅是其中的一種。互斥鎖被鎖定時(shí)進(jìn)入休眠，而系統(tǒng)還能正常運(yùn)轉(zhuǎn)，但有很多時(shí)候，鎖應(yīng)該不僅僅互斥訪問(wèn)，甚至應(yīng)該讓系統(tǒng)掛起，直至鎖成功，也就是說(shuō)在鎖操作中"自旋"，這就是Linux中的spinlock機(jī)制。
從實(shí)現(xiàn)上來(lái)說(shuō)，自旋鎖比較簡(jiǎn)單，主要分為兩部分，一部分是中斷處理，一部分是自旋處理，最基礎(chǔ)的部分在spin_lock_string和spin_unlock_string這兩段匯編代碼中：
#define spin_lock_string \
"\n1:\t" \
"lock ; decb %0\n\t" \
"js 2f\n" \
".section .text.lock,\"ax\"\n" \
"2:\t" \
"cmpb $0,%0\n\t" \
"rep;nop\n\t" \
"jle 2b\n\t" \
"jmp 1b\n" \
".previous"
#define spin_unlock_string \
"movb $1,%0"
不詳細(xì)解釋這段匯編代碼的語(yǔ)義，spin_lock_string對(duì)鎖原子減1，循環(huán)檢查鎖值，直到鎖值大于0；而spin_unlock_string則是對(duì)鎖賦值1。spin_lock_string用于構(gòu)成spin_lock()函數(shù)，spin_unlock_string用于構(gòu)成 spin_unlock()函數(shù)。
spin_lock()/spin_unlock()構(gòu)成了自旋鎖機(jī)制的基礎(chǔ)，它們和關(guān)中斷l(xiāng)ocal_irq_disable()/開(kāi)中斷l(xiāng)ocal_irq_enable()、關(guān)bh local_bh_disable()/開(kāi)bh local_bh_enable()、關(guān)中斷并保存狀態(tài)字local_irq_save()/開(kāi)中斷并恢復(fù)狀態(tài)字local_irq_restore() 結(jié)合就形成了整套自旋鎖機(jī)制，接口定義在include/linux/spinlock.h中，這里就不列舉了。
實(shí)際上，以上提到的spin_lock()都是在CONFIG_SMP的前提下才會(huì)生成的，也就是說(shuō)，在單處理機(jī)系統(tǒng)中，spin_lock()是一條空語(yǔ)句，因?yàn)樵谔幚頇C(jī)執(zhí)行它的語(yǔ)句時(shí)，不可能受到打擾，語(yǔ)句肯定是串行的。在這種簡(jiǎn)單情況下，spin_lock_irq()就只需要鎖中斷就可以完成任務(wù)了。在include/linux/spinlock.h中就用#ifdef CONFIG_SMP來(lái)區(qū)分兩種不同的情況。
自旋鎖有很多種，信號(hào)量也可以用來(lái)構(gòu)成互斥鎖，原子操作也有鎖功能，而且還有與標(biāo)準(zhǔn)鎖機(jī)制類(lèi)似的讀寫(xiě)鎖變種，在不同的應(yīng)用場(chǎng)合應(yīng)該選擇不同的鎖，下一節(jié)就是介紹如何選擇。
回頁(yè)首
鎖的使用
1.用戶(hù)上下文之間
如果所訪問(wèn)的共享資源僅在用戶(hù)上下文中使用，最高效的辦法就是使用信號(hào)量。在net/core/netfilter.c中有一處使用信號(hào)量的例子：
static DECLARE_MUTEX(nf_sockopt_mutex);
int nf_register_sockopt(struct nf_sockopt_ops *reg)
{
...
if (down_interruptible(&nf_sockopt_mutex) != 0)
return -EINTR;
...
out:
up(&nf_sockopt_mutex);
return ret;
}
2.用戶(hù)上下文與bottom half之間
此時(shí)有兩種情況需要使用鎖，一是用戶(hù)上下文被bottom half中斷，二是多個(gè)處理機(jī)同時(shí)進(jìn)入一個(gè)臨界段。一般情況下，使用spin_lock_bh()/spin_unlock_bh()可以滿(mǎn)足要求，它將關(guān)閉當(dāng)前CPU的bottom half，然后再獲取鎖，直至離開(kāi)臨界段再釋放并對(duì)bottom half重新使能。
3.用戶(hù)上下文與軟中斷（Tasklet）之間
tasklet與bottom half的實(shí)現(xiàn)機(jī)制是一樣的，實(shí)際上spin_lock_bh()也同時(shí)關(guān)閉了tasklet的執(zhí)行，因此，在這種情況下，用戶(hù)上下文與tasklet之間的同步也使用spin_lock_bh()/spin_unlock_bh()。
4.bottom half之間
bottom half本身的機(jī)制就保證了不會(huì)有多于1個(gè)的bottom half同時(shí)處于運(yùn)行態(tài)，即使對(duì)于SMP系統(tǒng)也是如此，因此，在設(shè)計(jì)共享數(shù)據(jù)的bottom half時(shí)無(wú)需考慮互斥。
5.tasklet之間
tasklet 和bottom half類(lèi)似，也是受到local_bh_disable()保護(hù)的，因此，同一個(gè)tasklet不會(huì)同時(shí)在兩個(gè)CPU上運(yùn)行；但不同的tasklet卻有可能，因此，如果需要同步不同的tasklet訪問(wèn)共享數(shù)據(jù)的話，就應(yīng)該使用spin_lock()/spin_unlock()。正如上面提到的，這種保護(hù)僅對(duì)SMP系統(tǒng)有意義，UP系統(tǒng)中tasklet的運(yùn)行不會(huì)受到另一個(gè)tasklet（不論它是否與之相同）的打擾，因此也就沒(méi)有必要上鎖。
6.softirq之間
softirq 是實(shí)現(xiàn)tasklet和bottom half的基礎(chǔ)，限制較后二者都少，允許兩個(gè)softirq同時(shí)運(yùn)行于不同的CPU之上，而不論它們是否來(lái)自同一個(gè)softirq代碼，因此，在這種情況下，都需要用spin_lock()/spin_unlock()來(lái)同步。
7.硬中斷和軟中斷之間
硬中斷是指硬件中斷的處理程序上下文，軟中斷包括softirq和在它基礎(chǔ)上實(shí)現(xiàn)的tasklet和bottom half等，此時(shí)，為了防止硬件中斷軟中斷的運(yùn)行，同步措施必須包括關(guān)閉硬件中斷，spin_lock_irq()/spin_unlock_irq() 就包括這個(gè)動(dòng)作。還有一對(duì)API，spin_lock_irqsave()/spin_unlock_irqrestore()，不僅關(guān)閉中斷，還保存機(jī)器狀態(tài)字，并在打開(kāi)中斷時(shí)恢復(fù)。
8.其他注意事項(xiàng)
首先需要提醒的是"死鎖"，這在操作系統(tǒng)原理的課本上都做過(guò)介紹，無(wú)論是使用信號(hào)量還是使用自旋鎖，都有可能產(chǎn)生死鎖，特別是自旋鎖，如果死鎖在spin_lock上，整個(gè)系統(tǒng)就會(huì)掛起。如何避免死鎖是理論課的問(wèn)題，這里就不多說(shuō)了。
另外一點(diǎn)就是盡可能短時(shí)間的鎖定，因此，"對(duì)數(shù)據(jù)上鎖，而不是對(duì)代碼上鎖"就成了一個(gè)簡(jiǎn)單的原則；在有可能的情況下，使用讀寫(xiě)鎖，而不要總是使用互斥鎖；對(duì)讀寫(xiě)排序，使用原子操作，從而完全避免使用鎖，也是一個(gè)不錯(cuò)的設(shè)計(jì)思想。
不要在鎖定狀態(tài)下調(diào)用可能引起休眠的操作，以下這些操作就是目前可能因此休眠的函數(shù)：
1. 對(duì)用戶(hù)內(nèi)存的訪問(wèn)：copy_from_user()、copy_to_user()、get_user()、put_user()
2. kmalloc(GFP_KERNEL)
3. down_interruptible()和down()，如果需要在spinlock中使用信號(hào)量，可以選擇down_trylock()，它不會(huì)引起掛起 printk()的靈巧設(shè)計(jì)使得它不會(huì)掛起，因此可以在任何上下文中使用。
摘自: http://baoguanghua.blog.sohu.com/79808640.html