Linux內(nèi)核中的同步與互斥
先看進(jìn)程間的互斥。在linux內(nèi)核中主要通過(guò)semaphore機(jī)制和spin_lock機(jī)制實(shí)現(xiàn)。主要的區(qū)別是在semaphore機(jī)制中，進(jìn)不了臨界區(qū)時(shí)會(huì)進(jìn)行進(jìn)程的切換，而spin_lock剛執(zhí)行忙等（在SMP中）。
先看內(nèi)核中的semaphore機(jī)制。前提是對(duì)引用計(jì)數(shù)count增減的原子性操作。內(nèi)核用ato
mic_t的數(shù)據(jù)結(jié)構(gòu)和在它上面的一系列操作如atomic_add()、atomic_sub()等等實(shí)現(xiàn)。（定義在atomic.h中）
semaphone機(jī)制主要通過(guò)up()和down()兩個(gè)操作實(shí)現(xiàn)。
semaphone的結(jié)構(gòu)為
- struct semaphore {
- atomic_t count;
- int sleepers;
- wait_queue_head_t wait;
- };
相應(yīng)的down()函數(shù)為
- static inline void down(struct semaphore*sem)
- {
/* 1 */sem->count--; //為原子操作
- if(sem->count<0)
- {
- 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);
/* 2 */ sem->sleepers++;
- for (;;) {
- int sleepers = sem->sleepers;
- /*
- * Add "everybody else" into it. They aren't
- * playing, because we own the spinlock.
- */
/* 3 */ if (!atomic_add_negative(sleepers - 1, &sem->count)) {
/* 4 */ sem->sleepers = 0; //這時(shí)sem->count=0
- break;
- }
/* 4 */ sem->sleepers = 1; /* us - see -1 above */ // 這時(shí)sem
- ->count
- =-1
- 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);
- }
- }
相應(yīng)的up()函數(shù)為
- void up(struct semaphore*sem)
- {
sem->count++; //為原子操作
- if(sem->count<=0)
- {
//喚醒等待隊(duì)列中的一個(gè)符合條件的進(jìn)程（因?yàn)槊總€(gè)進(jìn)程都加了TASK_EXCLUSIVE標(biāo)志）
。
};
假設(shè)開始時(shí)，count=1;sleepers=0
#p#
當(dāng)進(jìn)程A執(zhí)行down()時(shí)，引用計(jì)數(shù)count--，如果這時(shí)它的值大于等于0，則從down()中直接返回。如果count少于0，則A的state改為TASK_INTERRUPTIBLE后進(jìn)入這個(gè)信號(hào)量的等待隊(duì)列中，同時(shí)使sleepers++;然后重新計(jì)算count=sleepers - 1 + count,若這時(shí)引用計(jì)數(shù)仍小于0（一般情況下應(yīng)為-1，因?yàn)閏ount = - sleepers,不過(guò)在SMP結(jié)構(gòu)中，期間別的進(jìn)程可能執(zhí)行了up()和down()從而使得引用計(jì)數(shù)的值可能變化），則執(zhí)行進(jìn)程切換。
當(dāng)進(jìn)程A又獲得機(jī)會(huì)運(yùn)行時(shí)，它先執(zhí)行wake_up(&sem->wait)操作，喚醒等待隊(duì)列里的一個(gè)進(jìn)程，接著它進(jìn)入臨界區(qū)，從臨界區(qū)出來(lái)時(shí)執(zhí)行up()操作，使sem>count++,（如果進(jìn)程A是從down()中直接返回，因?yàn)檫@時(shí)等待隊(duì)列一定為空，所以它不用執(zhí)行wake_up()操作，直接進(jìn)入臨界區(qū)，在從臨界區(qū)出來(lái)時(shí)一樣執(zhí)行up()操作，使 sem->count++）。這時(shí)如果count的值小于等于0，這表明在它在臨界區(qū)期間又有一個(gè)進(jìn)程（可能就是它進(jìn)入臨界區(qū)時(shí)喚醒的那個(gè)進(jìn)程）進(jìn)入睡眠了，則執(zhí)行wake_up()操作，反之，如果count的值已經(jīng)大于0，這表明在它在臨界區(qū)期間沒(méi)有別的進(jìn)程（包括在它進(jìn)入臨界區(qū)時(shí)被它喚醒過(guò)的那個(gè)進(jìn)程）進(jìn)入睡眠，那么它就可以直接返回了。
從被喚醒的那個(gè)進(jìn)程看看，如果在喚醒它的進(jìn)程沒(méi)執(zhí)行up()之前它就得到了運(yùn)行機(jī)會(huì),這時(shí)它又重新計(jì)算count=sleepers - 1 + count=-1；從而sleepers被賦值1；這時(shí)它又必須進(jìn)行調(diào)度讓出運(yùn)行的機(jī)會(huì)給別的進(jìn)程，自己去睡眠。這正是發(fā)生在喚醒它的進(jìn)程在臨界區(qū)時(shí)運(yùn)行的時(shí)候。
如果是在喚醒它的進(jìn)程執(zhí)行了up()操作后它才得到了運(yùn)行機(jī)會(huì)，而且在喚醒它的進(jìn)程在臨界區(qū)期間時(shí)沒(méi)別的進(jìn)程執(zhí)行down()，則count的值在進(jìn)程執(zhí)行up()之前依然為0，這時(shí)在up()里面就不必要再執(zhí)行wake_up()函數(shù)了。
可以通過(guò)一個(gè)例子來(lái)說(shuō)明具體的實(shí)現(xiàn)。設(shè)開始sem->count=sem->sleepers=0。也就是有鎖但無(wú)等待隊(duì)列 （一個(gè)進(jìn)程已經(jīng)在運(yùn)行中）。先后分別進(jìn)行3個(gè)down()操作，和3個(gè)up()操作，如下：
為了闡述方便，只保留了一些會(huì)改變sleepers和count值的步驟，并且遵循從左到右依次進(jìn)行的原則。
down1:
count(0->-1),sleepers(0->1),sleepers-1+count(-1),count(-1),sleepers(1),調(diào)度
down2:
count(-1->-2),sleepers(1->2),sleepers-1+count(-1),count(-1),sleepers(1),調(diào)度
down3:
count(-1->-2),sleepers(1->2),sleepers-1+count(-1),count(-1),sleepers(1),調(diào)度
up1:
count(-1->0),喚醒一個(gè)睡眠進(jìn)程（設(shè)為1）,（進(jìn)程1得到機(jī)會(huì)運(yùn)行）sleepers-1+count
(0),count(0),sleepers(0),break,
喚醒另一個(gè)睡眠進(jìn)程（設(shè)為2），
?。ㄟM(jìn)程2得到機(jī)會(huì)運(yùn)行）sleepers-1+count(-1),count(-1),sleepers(1),調(diào)度（沒(méi)達(dá)到
條件，又得睡覺(jué)）
也可能是這樣的：
up1`:
count(-1->0),喚醒一個(gè)睡眠進(jìn)程（設(shè)為1）,（進(jìn)程1得到機(jī)會(huì)運(yùn)行）sleepers-1+count
(0),count(0),sleepers(0),break,
喚醒另一個(gè)睡眠進(jìn)程（設(shè)為2），
進(jìn)程2在以后才得到機(jī)會(huì)運(yùn)行）
up2:
count(-1->0),（因?yàn)閏ount<=0）喚醒一個(gè)睡眠進(jìn)程（設(shè)為2）,
進(jìn)程2得到機(jī)會(huì)運(yùn)行）sleepers-+count(0) , count(0) , sleepers(0) ,break,
喚醒另一個(gè)睡眠進(jìn)程（設(shè)為3），
進(jìn)程3得到機(jī)會(huì)運(yùn)行）sleepers-1+count(-1),count(-1),sleepers(1),調(diào)度（沒(méi)達(dá)到條
件，又得睡覺(jué)）
對(duì)應(yīng)上面的1`：
up2`:
count(0->1),（因?yàn)閏ount>0,所以直接返回）
進(jìn)程2得到機(jī)會(huì)運(yùn)行）sleepers-1+count(0),count(0),sleepers(0),break,
喚醒另一個(gè)睡眠進(jìn)程，（設(shè)為3）
up3:
count(-1->0),（因?yàn)閏ount<=0）喚醒一個(gè)睡眠進(jìn)程（設(shè)為3）,
進(jìn)程3得到機(jī)會(huì)運(yùn)行）sleepers-1+count(0),count(0),sleepers(0),break,
喚醒另一個(gè)睡眠進(jìn)程（這時(shí)隊(duì)列里沒(méi)進(jìn)程了）
進(jìn)程3運(yùn)行結(jié)束，執(zhí)行up(), 使count =1 ,這時(shí)變成沒(méi)鎖狀態(tài) ）
對(duì)應(yīng)上邊的2`：
up3`:
count(0->1),（因?yàn)閏ount>0,所以直接返回）
進(jìn)程3得到機(jī)會(huì)運(yùn)行）sleepers-1+count(0),count(0),sleepers(0),break,
喚醒另一個(gè)睡眠進(jìn)程（這時(shí)隊(duì)列里沒(méi)進(jìn)程了）
進(jìn)程3運(yùn)行結(jié)束，執(zhí)行up(), 使count =1 ,這時(shí)變成沒(méi)鎖狀態(tài) ）
當(dāng)然，還有另一種情況，就是up()操作和down()操作是交*出現(xiàn)的，
一般的規(guī)律就是，如果進(jìn)程在臨界區(qū)期間又有進(jìn)程（無(wú)論是哪個(gè)進(jìn)程，新來(lái)的還是剛被喚醒的那個(gè)）進(jìn)入睡眠，就會(huì)令count的值從0變?yōu)?1，從而進(jìn)程在從臨界區(qū)出來(lái)執(zhí)行up()里就必須執(zhí)行一次wake_up()，以確保所有的進(jìn)程都能被喚醒，因?yàn)槎鄦拘褞讉€(gè)是沒(méi)關(guān)系的。如果進(jìn)程在臨界區(qū)期間沒(méi)有別的進(jìn)程進(jìn)入睡眠，則從臨界區(qū)出來(lái)執(zhí)行up()時(shí)就用不著去執(zhí)行wake_up()了（當(dāng)然，執(zhí)行了也沒(méi)什么影響，不過(guò)多余罷了）。
而為什么要把wake_up()和count++分開呢，可以從上面的up1看出來(lái)，例如，進(jìn)程2第一次得到機(jī)會(huì)運(yùn)行時(shí)，本來(lái)這時(shí)喚醒它的進(jìn)程還沒(méi)執(zhí)行up()的，但有可能其它進(jìn)程執(zhí)行了 up()了，所以真有可能會(huì)發(fā)現(xiàn)count==1的情況，這時(shí)它就真的不用睡覺(jué)了，令count=sl eepers=0，就可以接著往下執(zhí)行了。
還可看出一點(diǎn)，一般的，( count ,sleepers)的值的取值范圍為(n ,0)[n>0] 和(0
,0
)和 (1 ,-1)。
#p#
下邊看看spin_lock機(jī)制。
Spin_lock采用的方式是讓一個(gè)進(jìn)程運(yùn)行，另外的進(jìn)程忙等待，由于在只有一個(gè)cpu
的機(jī)
器(UP)上微觀上只有一個(gè)進(jìn)程在運(yùn)行。所以在UP中，spin_lock和spin_unlock就都是空
的了。
在SMP中，spin_lock()和spin_unlock()定義如下。
- typedef struct {
- volatile unsigned int lock;
- } spinlock_t;
static inline void spin_lock(spinlock_t *lock)
- {
- __asm__ __volatile__(
- "\n1:\t"
- "lock ; decb %0\n\t"
- "js 2f\n" //lock->lock< 0 ,jmp 2 forward
- ".section .text.lock,\"ax\"\n"
- "2:\t"
- "cmpb $0,%0\n\t" //wait lock->lock==1
- "rep;nop\n\t"
- "jle 2b\n\t"
- "jmp 1b\n"
- ".previous"
- :"=m" (lock->lock) : : "memory");
- }
static inline void spin_unlock(spinlock_t *lock)
- {
- __asm__ __volatile__(
- "movb $1,%0"
- :"=m" (lock->lock) : : "memory"); //lock->lock=1
- }
一般是如此使用:
- #define SPIN_LOCK_UNLOCKED (spinlock_t) { 1 }
- spinlock_t xxx_lock = SPIN_LOCK_UNLOCKED;
- spin_lock_(&xxx_lock)
- ...
- critical section
- ...
- spin_unlock (&xxx_lock)
可以看出，它和semaphore機(jī)制解決的都是兩個(gè)進(jìn)程的互斥問(wèn)題，都是讓一個(gè)進(jìn)程退出臨界區(qū)后另一個(gè)進(jìn)程才進(jìn)入的方法，不過(guò)sempahore機(jī)制實(shí)行的是讓進(jìn)程暫時(shí)讓出CPU，進(jìn)入等待隊(duì)列等待的策略，而spin_lock實(shí)行的卻是卻進(jìn)程在原地空轉(zhuǎn)，等著另一個(gè)進(jìn)程結(jié)束的策略。
下邊考慮中斷對(duì)臨界區(qū)的影響。要互斥的還有進(jìn)程和中斷服務(wù)程序之間。當(dāng)一個(gè)進(jìn)程在執(zhí)行一個(gè)臨界區(qū)的代碼時(shí)，可能發(fā)生中斷，而中斷函數(shù)可能就會(huì)調(diào)用這個(gè)臨界區(qū)的代碼，不讓它進(jìn)入的話就會(huì)產(chǎn)生死鎖。這時(shí)一個(gè)有效的方法就是關(guān)中斷了。
- #define local_irq_save(x) __asm__ __volatile__("pushfl ; popl %0 ;
- cli":
- "=g" (x): /* no input */ :"memory")
- #define local_irq_restore(x) __asm__ __volatile__("pushl %0 ; popfl": /*
- no
- output */ :"g" (x):"memory")
- #define local_irq_disable() __asm__ __volatile__("cli": : :"memory")
- #define local_irq_enable() __asm__ __volatile__("sti": : :"memory")
- #define cpu_bh_disable(cpu) do { local_bh_count(cpu)++; barrier(); } while (
- 0)
- #define cpu_bh_enable(cpu) do { barrier(); local_bh_count(cpu)--; } while
- (0
- )
- #define local_bh_disable() cpu_bh_disable(smp_processor_id())
- #define local_bh_enable() cpu_bh_enable(smp_processor_id())
對(duì)于UP來(lái)說(shuō)，上面已經(jīng)是足夠了，不過(guò)對(duì)于SMP來(lái)說(shuō)，還要防止來(lái)自其它c(diǎn)pu的影響，這
時(shí)解決的方法就可以把上面的spin_lock機(jī)制也綜合進(jìn)來(lái)了。
- #define spin_lock_irqsave(lock, flags) do {
- local_irq_save(flags); sp
- in_lock(lock); } while (0)
- #define spin_lock_irq(lock) do { local_irq_disable();
- spin_lock(lo
- ck); } while (0)
- #define spin_lock_bh(lock) do { local_bh_disable();
- spin_lock(loc
- k); } while (0)
- #define spin_unlock_irqrestore(lock, flags) do { spin_unlock(lock); local_i
- rq_restore(flags); } while (0)
- #define spin_unlock_irq(lock) do { spin_unlock(lock);
- local_irq_enable();
- } while (0)
- #define spin_unlock_bh(lock) do { spin_unlock(lock);
- local_bh_enable();
- } while (0)
前面說(shuō)過(guò)，對(duì)于UP來(lái)說(shuō)，spin_lock（）是空的，所以以上的定義就一起適用于UP 和SM
P的情形了。
而read_lock_irqsave(lock, flags) , read_lock_irq(lock),
read_lock_bh(lock) 和
write_lock_irqsave(lock, flags) , write_lock_irq(lock),
write_lock_bh(lock
) 就是spin_lock的一個(gè)小小的變型而己了。
一口能看完的都是大蝦，希望本文能對(duì)你們有很大的幫助。
【編輯推薦】
- Linux就這樣被黑客入侵
- Linux Kernel 2.6.37發(fā)布 Ext4性能提升
- 如何學(xué)好Linux 十一大建議
- 從頭學(xué)習(xí)Linux基礎(chǔ)的8大建議
- 眼見為實(shí) 近看Sandy Bridge內(nèi)核真身
- 2010年度報(bào)告：是誰(shuí)在編寫Linux內(nèi)核？
- Linux內(nèi)核編譯之高手教程
