予焦啦！我們昨日實作完簡易排程，確保 Golang 執行緒（M）都會被排到 CPU 資源。但是卻有不定時炸彈會出現，那就是試圖解鎖非上鎖的鎖的錯誤；大部分時候都運行良好，但偶爾這個錯誤會自己發生，又或者像是昨日最後的附加實驗那樣，調整計時器中斷的頻率或是增加其它系統活動，都可能會看到這個問題：
fatal error: unlock of unlocked lock
在深入研究之前，我們得先檢視 Golang 執行期的鎖是怎麼回事。尤其是，我們現在使用的鎖是怎麼來的？
爬梳一下發現，我們當初在參賽第 3 天，產出可執行檔的過程中，lock 與 unlock 這兩個一度是非定義符號，但我們強行引用 src/runtime/lock_js.go 將之解消了。
然而，若是讓我們瀏覽一下 js/wasm 系統組合的這個檔案內容，不難發現一些註解，似乎給予我們遲來的警告：
// js/wasm has no support for threads yet. There is no preemption.
...
func lock2(l *mutex) {
if l.key == mutex_locked {
// js/wasm is single-threaded so we should never
// observe this.
...
顯然這個系統組合只支援單一執行緒，所以很可能它的鎖的實作，會有一些對於多執行緒系統行為來說過強的假設。說得更白話就是，當初 opensbi/riscv64 選擇用 lock_js.go 來逃避相關實作，現在可能要認賠殺出啦。
筆者承認，筆者閱讀完
js/wasm這裡的實作之後，還是想不通出現unlock unlocked lock的錯誤是怎麼回事。也許是這裡仗著單一執行緒的特性，所以在lock/unlock函數當中都只是加加減減，而沒有牽涉到任何原子性操作的緣故？這些都只是臆測，筆者不想再花力氣在這一組鎖的實作上了。
那麼，下一步該何去何從？老樣子，先參考其他 Golang 所支援的系統組合，都是如何實作它們所使用的鎖。
$ grep 'func lock' -R src/runtime/ | grep linux
$ grep 'func lock' -R src/runtime/ | grep freebsd
$
空空如也！執行期之大，難道 Linux 和 FreeBSD 卻沒有一個鎖的機制嗎？原來是多慮了。事實上，關於鎖的實作，除了比較特立獨行的 js/wasm 系統組合之外，只有兩個陣營：
$ grep 'func lock(' -R src/runtime/
src/runtime/lock_futex.go:func lock(l *mutex) {
src/runtime/lock_sema.go:func lock(l *mutex) {
src/runtime/lock_js.go:func lock(l *mutex) {
src/runtime/lock_opensbi.go:func lock(l *mutex) {
一個是快速使用者空間互斥鎖（futex），另一個則是旗號（semaphore）。前者支援的作業系統是兩款 BSD 與 Linux：
//go:build dragonfly || freebsd || linux
後者則是幾款其他的類 Unix 作業系統與 Plan9 與 Windows：
//go:build aix || darwin || netbsd || openbsd || plan9 || solaris || windows
src/runtime/lock_futex.go最一開始，就有一段註解在描述，作業系統應該實作的部分。實際上還蠻少的，只有兩個（中文是筆者的翻譯）：
// This implementation depends on OS-specific implementations of
//
// futexsleep(addr *uint32, val uint32, ns int64)
// 原子性地進行以下操作：
// if *addr == val { sleep }
// 不會睡超過 ns 的時間長度。若 ns 小於 0，代表永遠睡眠。
// 這個睡眠的狀態可以虛假地（spuriously）被叫醒。
//
// futexwakeup(addr *uint32, cnt uint32)
// 叫醒至多 cnt 個執行緒，若它們睡在 addr 上的話。
這個檔案中的其餘部分，都是這兩個呼叫的使用者。最主要的兩個組合為
lock/unlock：作為互斥鎖之用的 API。noteclear/notesleep/notewakeup：圍繞 note 結構體發展出來出來的三個基本方法。這些 API 的抽象功能在 src/runtime/runtime2.go 裡面有定義。簡單來說， note 是一種一次性的事件，使用前須有 noteclear 對之初始化。只能夠恰有一個執行緒呼叫 notesleep，表示這個執行緒將會進入睡眠狀態。喚醒它的方法，是必須恰有一個執行緒呼叫 notewakeup 且針對該事件。以下我們再分兩小節，稍微深入地觀察其中的機制吧。
lock_futex.go 的互斥鎖同步機制實作這裡顯示的是
lock2而非lock函數沒有錯，但後綴為 2 的互斥鎖函數才是這一組同步機制的核心。
首先，上鎖的部份可以約略區分為五個區塊。其中後三者是一起處在一個無窮迴圈裡面，除非取到這個鎖回傳，否則無法離開迴圈與這個上鎖的函數。
func lock2(l *mutex) {
gp := getg()
if gp.m.locks < 0 {
throw("runtime·lock: lock count")
}
gp.m.locks++
第一階段是很簡單的維護 gp.m，也就是當前共常式所屬的執行緒的一個資料成員，locks。從程式語意上不難看出，這個值用來代表一個執行緒當前取得的鎖的數量。這個成員變數的一個用例是，當因為某些原因需要中止這個執行緒，而在 Golang 執行期內呼叫了 stopm 函數，其中就有一個條件判斷：若是 g.m.locks 的值非為 0，這個執行緒就不該被停止；理由是，如果它握有其它執行緒執行下去必須取得的鎖，那整個流程都會出問題。
// Speculative grab for lock.
v := atomic.Xchg(key32(&l.key), mutex_locked)
if v == mutex_unlocked {
return
}
這裡動用了原子性的交換（Xchg）功能，試著去換換看互斥鎖結構（這裡是變數 l.key）的狀態。mutex_ 開頭的常數是可以顧名思義的狀態常數，除了這一段看到的上鎖狀態與非上鎖狀態之外，還有一個是睡眠狀態。
換過之後的回傳值（這裡儲存在 v）是原先的狀態。所以如果原本是非上鎖狀態，那就表示已經成功取得了這個鎖，那就可以回傳了。反之，則表示互斥鎖處在上鎖狀態或是睡眠狀態，所以當前的執行緒本身需要再等一下。
wait := v
...
for {
// Try for lock, spinning.
for i := 0; i < spin; i++ {
for l.key == mutex_unlocked {
if atomic.Cas(key32(&l.key), mutex_unlocked, wait) {
return
}
}
procyield(active_spin_cnt)
}
第三階段起始，都包在這個無條件的無窮迴圈之內。簡單來說，就是不斷的嘗試取得這個鎖。現在擷取的這一段，核心概念是透過比對後交換（Cas，Compare And Swap）來達成互斥鎖本身的狀態轉移與伺機取得。我們可以忽略 i 變數迴圈，那是細節。在其中會先讀取 l.key，看它是否是非上鎖狀態。注意這個是單純的讀取，而不是原子性讀取，所以無論當前的執行流程觀察的結果如何，其它執行緒都很有可能並行地造成這個鎖狀態的改變。因此這裡只是先預先偷偷地讀取一下，看看有沒有必要試著改變互斥鎖的狀態。
如果這個讀取本身就失敗了，表示互斥鎖並不是非上鎖狀態，那就可以直接呼叫到 procyield 函數去並準備離開這一段了。這個函數在 RISC-V 沒有實作，但是在 ARM64 的話，他們有實作 YIELD 指令，可以接收到這裡傳入的 active_spin_cnt 變數並有意義地運用；大致上，是讓當前核心可以停止那麼多個 cycle 的意思。procyied 函數是與架構相依的實作，所以 RISC-V 的部份出現在 src/runtime/asm_riscv64.s 裡面。
如果這個讀取成功了，則無論其它並行的執行緒正在作什麼動作、是否修改到了這個互斥鎖的狀態，至少蠻值得一試。所以這裡的行為是，我原子性地嘗試改變互斥鎖狀態，如果它現在是非上鎖狀態，那麼我將之改變為 wait 變數的狀態（上鎖或睡眠）。至於 Cas 函數的回傳值，是在成功替換時回傳 1，無法替換時回傳 0。所以如果成功將非上鎖的狀態改變為不可用的狀態之任一，當前執行緒就取得了鎖，可以離開上鎖函數了。反之，則回到先前的讀取去。
// Try for lock, rescheduling.
for i := 0; i < passive_spin; i++ {
for l.key == mutex_unlocked {
if atomic.Cas(key32(&l.key), mutex_unlocked, wait) {
return
}
}
osyield()
}
第四階段與第三階段幾乎一樣，但是 i 迴圈的數量其實比前一階段少。從註解也可以看出端倪，前一階段是一邊試著拿到鎖一邊空轉（spin），這個階段則是一邊試著拿到鎖，一邊重新排程，顯然時間尺度較大一些。最主要的差異在於這裡用了 osyield，這是作業系統相依的實作部份，所以以 Linux 為例的話，實作在 src/runtime/sys_linux_riscv64.s 裡面，內容是呼叫 Linux 的 sched_yield 系統呼叫，代表說，這個執行緒自願交出控制權，讓作業系統可以優先安排其它執行緒來使用。
// Sleep.
v = atomic.Xchg(key32(&l.key), mutex_sleeping)
if v == mutex_unlocked {
return
}
wait = mutex_sleeping
futexsleep(key32(&l.key), mutex_sleeping, -1)
}
最後一個階段如註解，前兩種試過之後都沒有消息，就有點放棄想直接進入睡眠模式了吧。首先是直接試著改變互斥鎖狀態為睡眠狀態，若是剛好很巧發現原先狀態是非上鎖，表示雖然前面幾個階段在試的時候都有其它執行緒佔用，但這裡剛好運氣好排到了，那也是表示成功取得了鎖，可以回傳。反之，表示它可能在非上鎖或是睡眠狀態。整個大無窮迴圈的最後這一行，正式使用到 futexsleep 功能，也就是原子性地判斷，若是互斥鎖狀態為睡眠狀態，那就不設時限的睡下去吧。
至於解鎖部份，比較單純一點，只分為兩個階段就可以解釋完基本功能：
func unlock2(l *mutex) {
v := atomic.Xchg(key32(&l.key), mutex_unlocked)
if v == mutex_unlocked {
throw("unlock of unlocked lock")
}
if v == mutex_sleeping {
futexwakeup(key32(&l.key), 1)
}
第一組條件判斷就是我們先前遭遇過的，不該解鎖已經是非上鎖狀態的鎖，無須贅言。第二組判斷的意思是，如果這不只是上鎖狀態，而是某個睡眠狀態的鎖，表示有其它執行緒正陷入睡眠，因此這裡叫醒一個。
gp := getg()
gp.m.locks--
if gp.m.locks < 0 {
throw("runtime·unlock: lock count")
}
...
第二段將執行緒本身取得的鎖 locks 成員變數維護好，就可以算是解鎖離開關鍵區域（critical section）了。
lock_futex.go 的一次性事件同步機制實作首先介紹 noteclear，也就是初始化一個 note 結構：
func noteclear(n *note) {
n.key = 0
}
沒有其它的繁文縟節，設為 0 之後就算可以使用的事件了。然後是 notesleep，
func notesleep(n *note) {
gp := getg()
if gp != gp.m.g0 {
throw("notesleep not on g0")
}
ns := int64(-1)
if *cgo_yield != nil {
// Sleep for an arbitrary-but-moderate interval to poll libc interceptors.
ns = 10e6
}
for atomic.Load(key32(&n.key)) == 0 {
gp.m.blocked = true
futexsleep(key32(&n.key), 0, ns)
if *cgo_yield != nil {
asmcgocall(*cgo_yield, nil)
}
gp.m.blocked = false
}
}
筆者承認，這裡實在不太清楚
cgo_yield的功能是什麼，但透過 GDB 觀察，指到_cgo_yield的部份都是 0，也就是說相關的判斷與呼叫，我們這裡應該是可以忽略的。雖然不難猜測，應該是連接到某些 C 語言界面時，可以實行更細緻的讓出資源（也就是 yield 的本意）動作。
當 note 結構是乾淨狀，則直接無時限地睡下去，因為 ns 變數為 -1。這裡也維護了 blocked 成員變數，目前看起來只有 windows 有在利用這個屬性，代表一個執行緒是否正在被某個一次性事件卡著。這個前提是，n.key 的值一直都還是 0，沒有改變過。但這個判斷的邏輯實際上也代表，如果 n.key 被修改過而變成不是 0 了，那麼根本就不會走進這些流程，也就不需要睡了。
參照 notewakeup，就可以理解這整組的效果，
func notewakeup(n *note) {
old := atomic.Xchg(key32(&n.key), 1)
if old != 0 {
print("notewakeup - double wakeup (", old, ")\n")
throw("notewakeup - double wakeup")
}
futexwakeup(key32(&n.key), 1)
}
這個喚醒呼叫迎面就來個 n.key 的交換，它預期要嘛這個 note 的狀態是剛初始化完但卻沒有對應的執行緒睡眠之，或是已經有執行緒對之睡眠。總之只有一個排除錯誤狀況的判斷，之後就是使用 futexwakeup 去喚醒至多一個睡眠於這個 note 的執行緒。
觀察 os_linux.go 裡面的 futex 系列，可以看到這兩者都使用 futex 系統呼叫實作，差異在於 futexsleep 使用了 _FUTEX_WAIT_PRIVATE 參數，而 futexwakeup 使用的是 _FUTEX_WAKE_PRIVATE 參數。至於 sys_linux_riscv64.s 中的 futex 系統呼叫，也只是很簡單地將暫存器轉手之後再呼叫核心而已，因此就先都省略了。
如果是旗號陣營的作業系統的話，Golang 要求實作的介面（列在 src/runtime/lock_sema.go 之中）有三個（中文為筆者的翻譯），
// This implementation depends on OS-specific implementations of
//
// func semacreate(mp *m)
// 如果不是已經創造過了的話，創造一個 mp 的旗號
//
// func semasleep(ns int64) int32
// 如果 ns 小於 0，取得 m 的旗號並且回傳 0；
// 如果 ns 大於等於 0，試著取得 m 的旗號，持續至多 ns 奈秒的睡眠狀態；若是取得了旗號則回傳 0，
// 若是被中斷或是時間到了，則回傳 -1。
//
// func semawakeup(mp *m)
// 叫醒 mp。它的狀態是正在睡眠，或可能馬上準備要睡眠。
//
這個抽象層也能夠提供上述的互斥鎖對以及一次性事件的兩組 API，讀者可以自行瀏覽，會發現實作上非常的相似。
筆者比較熟悉的 Linux 和勉強玩過一陣子的 FreeBSD 都是 futex 陣營的作業系統，理論上是該往那邊靠攏才對，但是 futex 系統呼叫提供的功能實在太強大了，若要實作整個系統呼叫，我們必然需要從位址對應到等待該位址的執行緒的一種查詢機制，而且這種機制本身也必須被保護在關鍵區域之內才行，這實在是有點麻煩。
所以不得已，我們只好投奔旗號陣營了。以 NetBSD 的實作先參考一下：
semacreateNetBSD 多使用一個成員變數，waitsemacount，在 mOS 之中。創造旗號的這個 API 則是留空，畢竟都已經直接對應在執行緒的 mOS 裡面了。
semawakeup相較於進入睡眠的邏輯，喚醒的邏輯相當簡單，如下
func semawakeup(mp *m) {
atomic.Xadd(&mp.waitsemacount, 1)
// From NetBSD's _lwp_unpark(2) manual:
// "If the target LWP is not currently waiting, it will return
// immediately upon the next call to _lwp_park()."
ret := lwp_unpark(int32(mp.procid), unsafe.Pointer(&mp.waitsemacount))
if ret != 0 && ret != _ESRCH {
// semawakeup can be called on signal stack.
systemstack(func() {
print("thrwakeup addr=", &mp.waitsemacount, " sem=", mp.waitsemacount, " ret=", ret, "\n")
})
}
}
先將 waitsemacount 的值增加 1，然後引用 lwp_unpark 系統呼叫，讓作業系統去幫忙排時間給指定的執行緒。
semasleepfunc semasleep(ns int64) int32 {
_g_ := getg()
var deadline int64
if ns >= 0 {
deadline = nanotime() + ns
}
for {
v := atomic.Load(&_g_.m.waitsemacount)
if v > 0 {
if atomic.Cas(&_g_.m.waitsemacount, v, v-1) {
return 0 // semaphore acquired
}
continue
}
上半段，先在給定時限的情況下計算死線（deadline）為何，然後進入到無條件的無窮迴圈之中。新增的 waitsemacount 於是派上用場，其值被讀取之後，只有大於 0 時可以算是取得了這個旗號。怎麼說呢？雖然我們前一個段落跳過了旗號部份的邏輯，但與快速使用者互斥鎖的邏輯並不相差太多。會需要呼叫到睡眠 API 時，都是因為這個互斥鎖當前是被其它執行緒上鎖的，歷經短期的空轉與中期的重新排程未果，只好選擇睡眠一途。所以，這個執行緒若要取得互斥鎖，得有其它執行緒先釋放這個鎖才行。若能夠成功從大於 0 的狀態減一，則可回報取得這個旗標，重新回到搶奪 lock 當中互斥鎖的狀態，或是從針對單一事件的 notesleep 當中醒轉。
// Sleep until unparked by semawakeup or timeout.
var tsp *timespec
var ts timespec
if ns >= 0 {
wait := deadline - nanotime()
if wait <= 0 {
return -1
}
ts.setNsec(wait)
tsp = &ts
}
ret := lwp_park(_CLOCK_MONOTONIC, _TIMER_RELTIME, tsp, 0, unsafe.Pointer(&_g_.m.waitsemacount), nil)
if ret == _ETIMEDOUT {
return -1
}
}
}
下半段的部份，即是必須要認真考慮睡覺這回事了。在 ns 有確實傳入的狀況，就重新計算一次看看是否死線已經超過了。若否，則依賴 lwp_park 系統呼叫幫忙停住這麼多時間。
OpenSBI 當然沒有通知其它執行緒或是幫助我們暫停特定時間的功能。前者是完全不相干的抽象。後者和我們打通的計時器中斷當然是能夠牽扯在一起，但我們就必須先把計時器中斷機制抽象掉，另外維護一層排序過的事件，在系統執行期間都僅有最近要發生的事件會被設定計時器中斷，這聽起來又太麻煩了。
不過巧的是，我們這裡其實可以耍詐。反正以計時器中斷為基礎的固定式執行緒切換已經在運作了，所以陷在 semasleep 裡面的執行緒雖然沒有一個系統呼叫來幫助它睡指定的時間，但是它總還是會被計時器中斷通知，並且轉移控制權。雖然有點浪費 CPU 資源，但是理論上是可行的。
所以筆者就將 NetBSD 的系統呼叫拔掉之後，就完成今天的主要部份。
有一個先前先透過無窮迴圈卡住的
gcenable函數，我們必須移除這個路障，讓它能夠繼續進行下去。
結果還是跑一跑之後就會出現瘋狂的錯誤！以下的訊息不斷洗頻：
...
goroutine 0 [idle]:
runtime.throw({0xffffffc000098ae6, 0x14})
/home/noner/FOSS/hoddarla/ithome/go/src/runtime/panic.go:965 +0x60
runtime.ethanol_trap1(0xffffffcf04009ee0)
/home/noner/FOSS/hoddarla/ithome/go/src/runtime/os_opensbi.go:320 +0x21c
runtime.ethanol_trap()
/home/noner/FOSS/hoddarla/ithome/go/src/runtime/sys_opensbi_riscv64.s:78 +0xe4
for thread 0xffffffcf04000000(0xffffffc0001195e0): [0xffffffc000119440, 0xffffffcf04000000, 0xffffffcf040001a0]
fatal error: unexpected exception
配合 GDB 設置斷點在 throw，觀察這個錯誤，發現是在 runtime.main 的最後面的階段：
ffffffc000036f2c: 00027f97 auipc t6,0x27
ffffffc000036f30: 0d4f80e7 jalr 212(t6) # ffffffc00005e000 <runtime.exit>
ffffffc000036f34: 00000193 li gp,0
ffffffc000036f38: 0001a023 sw zero,0(gp)
ffffffc000036f3c: ff9ff06f j ffffffc000036f34 <runtime.main+0x2fc>
由於呼叫 exit 沒有效果，有一個強制寫入 0 到位址 0 的區段：
277 exit(0)
278 for {
279 var x *int32
280 *x = 0
281 }
所以這裡我們為 exit 與概念相近的 exitThread 補上實作，如下：
TEXT runtime·exit(SB),NOSPLIT,$0
MOV $0, A0
MOV $0, A1
MOV $0, A6
MOV $0x53525354, A7
ECALL
RET
TEXT runtime·exitThread(SB),NOSPLIT,$0
CALL runtime·exit(SB)
exit裡面的序列有點魔術，其實只是呼叫 OpenSBI，讓它將平台關機的指令。
今天的內容可以在 github 取得。
...
[0xffffffcf04000d00|0xffffffcf0402c580] wakes [0xffffffc000119440|0xffffffc0001195e0]
[0xffffffcf04000d00|0xffffffcf0402c580] try sema in -1 nanosecs
for thread 0xffffffcf04000b60(0xffffffcf0402c580): [0xffffffcf04000d00, 0xffffffcf04000b60, 0x0]
for thread 0xffffffcf04000b60(0xffffffcf0402c580): [0xffffffcf04000d00, 0xffffffcf04000b60, 0x0]
for thread 0xffffffcf04000b60(0xffffffcf0402c580): [0xffffffcf04000d00, 0xffffffcf04000b60, 0x0]
for thread 0xffffffcf04000340(0xffffffcf0402c000): [0xffffffcf040004e0, 0xffffffcf04000340, 0x0]
for thread 0xffffffcf04000340(0xffffffcf0402c000): [0xffffffcf040004e0, 0xffffffcf04000340, 0x0]
for thread 0xffffffcf04000340(0xffffffcf0402c000): [0xffffffcf040004e0, 0xffffffcf04000340, 0x0]
...
繞過 gcenable 的效果是，為了垃圾回收，Golang 執行期會再多創一個執行緒（m）與兩個共常式（g）。所以，加上筆者補上的一些印出資訊，執行之後可以看到執行緒之間的換手歷程，以及我們補進來的旗號系列函數被呼叫的過程，然後停止執行，QEMU 正常離開。
今天是我們第一次成功讓 QEMU 從系統啟動到它自然結束，照理來說固然是值得慶幸的，但是這時候我們不禁會覺得納悶，Golang 執行期就這樣度過了，但 ethanol/ethanol.go 裡面的那個 Hello World 訊息又怎麼了呢？從 QEMU 的輸出訊息，我們什麼也沒有看到，顯然這裡還有我們沒有處理的部份，否則怎麼沒有印出訊息呢？
無論如何，這一章也可以算是結束了。這一章的基本發想是，Golang 有如此複雜的執行期功能，應該有些抽象足以挪用為作業系統的執行緒；我們因此將腦筋動到 m 結構上面，然後比對了 newosproc 所需要的抽象層。之後，我們加上了一個很簡單的 Round-Robin 式的固定排程；然後再基於旗號機制，補上了相關的函數。我們因此得以擺脫先前的 js/wasm 系統組合單一執行緒假設造成的錯誤，以及未完備的同步機制造成 gcenable 與 Golang 頻道（channel）不能使用的現象。最後也有一個小型修補，是執行緒離開（exit）的機制。
明天開始，我們就進入本篇的最後一章。各位讀者，我們明天再會！
iThome鐵人賽
看影片追技術