當一個以上的CPU同時存取相同的資源時就會需要用到鎖以確保資料的安全與完整,鎖的概念被廣泛的運用在各種場合,如火車通過時閉塞區間必須先取得鎖才能通行,在作業系統中也經常發生複數CPU或是interrupt handler競爭相同的資料結構,而使用鎖可以保證同一個時間只會有一個CPU可以存取該資料結構,因此可以保證資料的正確性與完整性。
鎖的類型有很多,xv6主要使用的是spinlock(自旋鎖),spinlock基本上就是維護一組boolean的不變量,藉由該不變量的true/false來判斷是否有其他cpu正在競爭相同的資料結構,如果答案是True則無窮迴圈等待,反之則繼續執行且同時將不變量設定為True。
在xv6中使用spinlock的地方相當多,幾乎所有跟I/O有關的操作都需要spinlock的保護,以下列出spinlock的呼叫:
File | init(line num) | acquire(line num) | release(line num) |
---|---|---|---|
Bio.c | 43 | 66 | 27,87 |
128 | 140 | ||
Console.c | 291 | 63 | 103 |
196 | 229 | ||
243 | 247,267 | ||
279 | 282 | ||
File.c | 22 | 31 | 35,39 |
47 | 51 | ||
61 | 65,71 |
以上僅列出部分呼叫。
在所有用到spinlock的地方都會先呼叫initlock(0838),對該spinlock命名並且初始化。
當執行acquire
會試圖取得鎖,首先會呼叫第三行的pushcli()
取得中斷狀態並關閉中斷cli()
,紀錄狀態在mycpu()->ineta
,為了CPU在取得複數的spinlock後又release
時誤開啟中斷,因此加入一個計數器,只有當CPU沒有取得任何spinlock時才恢復先前中斷觸發的狀態。
接著會CPU會試圖設定和取得locked
,只有當locked
為FALSE才允許進入,否則無窮迴圈等待。此處要注意xchg
實際上的作用是原子操作賦值和取值,目的為確保不變量locked
不會同時被讀取然後修改而導致多個CPU同時進入,特別需注意的是xchg
實際上是asm的xchgl
操作:
asm volatile("lock; xchgl %0, %1" :
"+m" (*addr), "=a" (result) :
"1" (newval) :
"cc");
利用指定的memory address和register交換可以達到同時取得該記憶體空間之值且同時設定新的value,只有當spinlock被release
時才可以取得FALSE並繼續進行,否則永遠都是等待。
__sync_synchronize()
是使用full memory barrier,意即不允許Compiler將barrier之前的命令re-ordering到barrier之後,反之亦然。其目的在於保證任何取得spinlock(通過xchg
)後執行的命令都不會再取得spinlock之前被執行,避免spinlock失效導致災難。
最後將目前的CPU id與call stack存入spinlock中做Debug用。
void
acquire(struct spinlock *lk)
{
pushcli(); // disable interrupts to avoid deadlock.
if(holding(lk))
panic("acquire");
// The xchg is atomic.
while(xchg(&lk->locked, 1) != 0)
;
__sync_synchronize();
// Record info about lock acquisition for debugging.
lk->cpu = mycpu();
getcallerpcs(&lk, lk->pcs);
}
pushcli(void)
{
int eflags;
eflags = readeflags();
cli();
if(mycpu()->ncli == 0)
mycpu()->intena = eflags & FL_IF;
mycpu()->ncli += 1;
}
如果要檢查是否擁有spinlock,則需要呼叫holding()
,檢查該CPU是否是spinlock擁有者,要注意的是xv6並沒有實作遞迴鎖,所以如果在holding()
中去acquire lock會導致deadlock,此處實際上用在檢查同個CPU是否在沒有鎖的情況下release
或是有鎖時又重複acquire
都會導致panic。
p.s. 呼叫holding時應該處於有lock的狀態嗎?如果是的話acquire
又怎麼解釋?
int
holding(struct spinlock *lock)
{
return lock->locked && lock->cpu == mycpu();
}
在release
部分與acquire類似,只是流程反轉,唯一要特別注意的是在此使用asm去設定locked而非lk->locked=0
:
asm volatile("movl $0, %0" : "+m" (lk->locked) : );
其原因是由於某些情況下C在Assign int並非atomic因此使用asm以保證。
最後呼叫popcli()
去恢復中斷狀態,如果計數器(ncli
)顯示0代表CPU目前並沒有取得任何spinlock,此時即可恢復先前的中斷狀態intena
。
void
release(struct spinlock *lk)
{
if(!holding(lk))
panic("release");
lk->pcs[0] = 0;
lk->cpu = 0;
__sync_synchronize();
asm volatile("movl $0, %0" : "+m" (lk->locked) : );
popcli();
}
void
popcli(void)
{
if(readeflags()&FL_IF)
panic("popcli - interruptible");
if(--mycpu()->ncli < 0)
panic("popcli");
if(mycpu()->ncli == 0 && mycpu()->intena)
sti();
}
struct spinlock {
uint locked; // 不變量,是否鎖上,必須以atomic操作。
// 偵錯用:
char *name; // 名稱.
struct cpu *cpu; // The cpu holding the lock.
uint pcs[10]; // Acquire時的Call stack (an array of program )
};