先看進(jìn)程間的互斥。在linux內核中主要通過(guò)semaphore機制和spin_lock機制實(shí)現。主要的區別是在semaphore機制中，進(jìn)不了臨界區時(shí)會(huì )進(jìn)行進(jìn)程的切換，而spin_lock剛執行忙等（在SMP中）。先看內核中的semaphore機制。前提是對引用計數count增減的原子性操作。內核用atomic_t的數據結構和在它上面的一系列操作如atomic_add()、atomic_sub()等等實(shí)現。（定義在atomic.h中）semaphone機制主要通過(guò)up()和down()兩個(gè)操作實(shí)現。semaphone的結構為: struct semaphore { atomic_t count; int sleepers; wait_queue_head_t wait; };
相應的down()函數為: 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); } }
相應的up()函數為: void up(struct semaphore*sem) { sem->count ; //為原子操作 if(sem->count<=0) { //喚醒等待隊列中的一個(gè)符合條件的進(jìn)程（因為每個(gè)進(jìn)程都加了TASK_EXCLUSIVE標志） 。 };
假設開(kāi)始時(shí)，count=1;sleepers=0。當進(jìn)程A執行down()時(shí)，引用計數count--，如果這時(shí)它的值大于等于0，則從down()中直接返回。如果count少于0，則A的state改為T(mén)ASK_INTERRUPTIBLE后進(jìn)入這個(gè)信號量的等待隊列中，同時(shí)使sleepers ;然后重新計算count=sleepers - 1 count,若這時(shí)引用計數仍小于0（一般情況下應為-1，因為count = - sleepers,不過(guò)在SMP結構中，期間別的進(jìn)程可能執行了up()和down()從而使得引用計數的值可能變化），則執行進(jìn)程切換。
當進(jìn)程A又獲得機會(huì )運行時(shí)，它先執行wake_up(&sem->wait)操作，喚醒等待隊列里的一個(gè)進(jìn)程，接著(zhù)它進(jìn)入臨界區，從臨界區出來(lái)時(shí)執行up()操作，使sem->count ,（如果進(jìn)程A是從down()中直接返回，因為這時(shí)等待隊列一定為空，所以它不用執行wake_up()操作，直接進(jìn)入臨界區，在從臨界區出來(lái)時(shí)一樣執行up()操作，使 sem->count ）。這時(shí)如果count的值小于等于0，這表明在它在臨界區期間又有一個(gè)進(jìn)程（可能就是它進(jìn)入臨界區時(shí)喚醒的那個(gè)進(jìn)程）進(jìn)入睡眠了，則執行wake_up()操作，反之，如果count的值已經(jīng)大于0，這表明在它在臨界區期間沒(méi)有別的進(jìn)程（包括在它進(jìn)入臨界區時(shí)被它喚醒過(guò)的那個(gè)進(jìn)程）進(jìn)入睡眠，那么它就可以直接返回了。從被喚醒的那個(gè)進(jìn)程看看，如果在喚醒它的進(jìn)程沒(méi)執行up()之前它就得到了運行機會(huì )，這時(shí)它又重新計算count=sleepers - 1 count=-1；從而sleepers被賦值1；這時(shí)它又必須進(jìn)行調度讓出運行的機會(huì )給別的進(jìn)程，自己去睡眠。這正是發(fā)生在喚醒它的進(jìn)程在臨界區時(shí)運行的時(shí)候。如果是在喚醒它的進(jìn)程執行了up()操作后它才得到了運行機會(huì )，而且在喚醒它的進(jìn)程在臨界區期間時(shí)沒(méi)別的進(jìn)程執行down()，則count的值在進(jìn)程執行up()之前依然為0，這時(shí)在up()里面就不必要再執行wake_up()函數了?？梢酝ㄟ^(guò)一個(gè)例子來(lái)說(shuō)明具體的實(shí)現。設開(kāi)始sem->count=sem->sleepers=0。也就是有鎖但無(wú)等待隊列 （一個(gè)進(jìn)程已經(jīng)在運行中）。先后分別進(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),調度
down2: count(-1->-2),sleepers(1->2),sleepers-1 count(-1),count(-1),sleepers(1),調度
down3: count(-1->-2),sleepers(1->2),sleepers-1 count(-1),count(-1),sleepers(1),調度
up1:
count(-1->0),喚醒一個(gè)睡眠進(jìn)程（設為1）,（進(jìn)程1得到機會(huì )運行）
sleepers-1 count(0),count(0),sleepers(0),break,
喚醒另一個(gè)睡眠進(jìn)程（設為2），（進(jìn)程2得到機會(huì )運行） sleepers-1 count(-1),count(-1),sleepers(1),
調度（沒(méi)達到條件，又得睡覺(jué)）也可能是這樣的： up1`:
count(-1->0),喚醒一個(gè)睡眠進(jìn)程（設為1）,（進(jìn)程1得到機會(huì )運行）
sleepers-1 count(0),count(0),sleepers(0),break,
喚醒另一個(gè)睡眠進(jìn)程（設為2），進(jìn)程2在以后才得到機會(huì )運行）
up2:
count(-1->0),（因為count<=0）喚醒一個(gè)睡眠進(jìn)程（設為2）,進(jìn)程2得到機會(huì )運行）
sleepers- count(0) , count(0) , sleepers(0) ,break,
喚醒另一個(gè)睡眠進(jìn)程（設為3），進(jìn)程3得到機會(huì )運行）
sleepers-1 count(-1),count(-1),sleepers(1),
調度（沒(méi)達到條件，又得睡覺(jué)）對應上面的1`：
up2`:
count(0->1),（因為count>0,所以直接返回）進(jìn)程2得到機會(huì )運行）
sleepers-1 count(0),count(0),sleepers(0),break,
喚醒另一個(gè)睡眠進(jìn)程，（設為3） up3:
count(-1->0),（因為count<=0）喚醒一個(gè)睡眠進(jìn)程（設為3）,進(jìn)程3得到機會(huì )運行）
sleepers-1 count(0),count(0),sleepers(0),break,
喚醒另一個(gè)睡眠進(jìn)程（這時(shí)隊列里沒(méi)進(jìn)程了）
進(jìn)程3運行結束，執行up(), 使count =1 ,這時(shí)變成沒(méi)鎖狀態(tài) ）
對應上邊的2`：
up3`:
count(0->1),（因為count>0,所以直接返回）進(jìn)程3得到機會(huì )運行）
sleepers-1 count(0),count(0),sleepers(0),break,
喚醒另一個(gè)睡眠進(jìn)程（這時(shí)隊列里沒(méi)進(jìn)程了）
進(jìn)程3運行結束，執行up(), 使count =1 ,這時(shí)變成沒(méi)鎖狀態(tài) ）
當然，還有另一種情況，就是up()操作和down()操作是交叉出現的，一般的規律就是，如果進(jìn)程在臨界區期間又有進(jìn)程（無(wú)論是哪個(gè)進(jìn)程，新來(lái)的還是剛被喚醒的那個(gè)）進(jìn)入睡眠，就會(huì )令count的值從0變?yōu)?1，從而進(jìn)程在從臨界區出來(lái)執行up()里就必須執行一次wake_up()，以確保所有的進(jìn)程都能被喚醒，因為多喚醒幾個(gè)是沒(méi)關(guān)系的。如果進(jìn)程在臨界區期間沒(méi)有別的進(jìn)程進(jìn)入睡眠，則從臨界區出來(lái)執行up()時(shí)就用不著(zhù)去執行wake_up()了（當然，執行了也沒(méi)什么影響，不過(guò)多余罷了）。而為什么要把wake_up()和count 分開(kāi)呢，可以從上面的up1看出來(lái)，例如，進(jìn)程2第一次得到機會(huì )運行時(shí)，本來(lái)這時(shí)喚醒它的進(jìn)程還沒(méi)執行up()的，但有可能其它進(jìn)程執行了up()了，所以真有可能會(huì )發(fā)現count==1的情況，這時(shí)它就真的不用睡覺(jué)了，令count=sleepers=0，就可以接著(zhù)往下執行了。還可看出一點(diǎn)，一般的，( count ,sleepers)的值的取值范圍為(n ,0)[n>0] 和(0 ,0)和 (1 ,-1)。下邊看看spin_lock機制。Spin_lock采用的方式是讓一個(gè)進(jìn)程運行，另外的進(jìn)程忙等待，由于在只有一個(gè)cpu的機器(UP)上微觀(guān)上只有一個(gè)進(jì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機制解決的都是兩個(gè)進(jìn)程的互斥問(wèn)題，都是讓一個(gè)進(jìn)程退出臨界區后另一個(gè)進(jìn)程才進(jìn)入的方法，不過(guò)sempahore機制實(shí)行的是讓進(jìn)程暫時(shí)讓出CPU，進(jìn)入等待隊列等待的策略，而spin_lock實(shí)行的卻是卻進(jìn)程在原地空轉，等著(zhù)另一個(gè)進(jìn)程結束的策略。
下邊考慮中斷對臨界區的影響。要互斥的還有進(jìn)程和中斷服務(wù)程序之間。當一個(gè)進(jìn)程在執行一個(gè)臨界區的代碼時(shí)，可能發(fā)生中斷，而中斷函數可能就會(huì )調用這個(gè)臨界區的代碼，不讓它進(jìn)入的話(huà)就會(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())
對于UP來(lái)說(shuō)，上面已經(jīng)是足夠了，不過(guò)對于SMP來(lái)說(shuō)，還要防止來(lái)自其它c(diǎn)pu的影響，這時(shí)解決的方法就可以把上面的spin_lock機制也綜合進(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ò)，對于UP來(lái)說(shuō)，spin_lock（）是空的，所以以上的定義就一起適用于UP 和SMP的情形了。 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è)小小的變型而己了。
