不安全的雙檢鎖

從其他語言轉入Go語言的同學經常會陷入一個思考：如何建立一個單例？

有些同學可能會把其它語言中的雙檢鎖模式移植過來，雙檢鎖模式也稱為懶漢模式，首次用到的時候才建立範例。大部分人首次用Golang寫出來的範例大概是這樣的：

type Conn struct {
	Addr  string
	State int
}

var c *Conn
var mu sync.Mutex

func GetInstance() *Conn {
	if c == nil {
		mu.Lock()
		defer mu.Unlock()
		if c == nil {
			c = &Conn{"127.0.0.1:8080", 1}
		}
	}
	return c
}

這裡先解釋下這段程式碼的執行邏輯（已經清楚的同學可以直接跳過）：

GetInstance用於獲取結構體Conn的一個範例，其中：先判斷c是否為空，如果為空則加鎖，加鎖之後再判斷一次c是否為空，如果還為空，則建立Conn的一個範例，並賦值給c。這裡有兩次判空，所以稱為雙檢，需要第二次判空的原因是：加鎖之前可能有多個執行緒/協程都判斷為空，這些執行緒/協程都會在這裡等著加鎖，它們最終也都會執行加鎖操作，不過加鎖之後的程式碼在多個執行緒/協程之間是序列執行的，一個執行緒/協程判空之後建立了範例，其它執行緒/協程在判斷c是否為空時必然得出false的結果，這樣就能保證c僅建立一次。而且後續呼叫GetInstance時都會僅執行第一次判空，得出false的結果，然後直接返回c。這樣每個執行緒/協程最多隻執行一次加鎖操作，後續都只是簡單的判斷下就能返回結果，其效能必然不錯。

瞭解Java的同學可能知道Java中的雙檢鎖是非執行緒安全的，這是因為賦值操作中的兩個步驟可能會出現亂序執行問題。這兩個步驟是：物件記憶體空間的初始化和將記憶體地址設定給變數。因為編譯器或者CPU優化，它們的執行順序可能不確定，先執行第2步的話，鎖外邊的執行緒很有可能存取到沒有初始化完畢的變數，從而引發某些異常。針對這個問題，Java以及其它一些語言中可以使用volatile來修飾變數，實際執行時會通過插入記憶體柵欄阻止指令重排，強制按照編碼的指令順序執行。

那麼Go語言中的雙檢鎖是安全的嗎？

答案是也不安全。

先來看看指令重排問題：

在Go語言規範中，賦值操作分為兩個階段：第一階段對賦值操作左右兩側的表示式進行求值，第二階段賦值按照從左至右的順序執行。（參考：https://golang.google.cn/ref/spec#Assignments）

說的有點抽象，但沒有提到賦值存在指令重排的問題，隱約感覺不會有這個問題。為了驗證，讓我們看一下上邊那段程式碼中賦值操作的偽組合程式碼：

紅框圈出來的部分對應的程式碼是： c = &Conn{"127.0.0.1:8080", 1}

其中有一行：CMPL $0x0, runtime.writeBarrier(SB) ，這個指令就是插入一個記憶體柵欄。前邊是要賦值資料的初始化，後邊是賦值操作。如此看，賦值操作不存在指令重排的問題。

既然賦值操作沒有指令重排的問題，那這個雙檢鎖怎麼還是不安全的呢？

在Golang中，對於大於單個機器字的值，讀寫它的時候是以一種不確定的順序多次執行單機器字的操作來完成的。機器字大小就是我們通常說的32位元、64位元，即CPU完成一次無定點整數運算可以處理的二進位制位數，也可以認為是CPU資料通道的大小。比如在32位元的機器上讀寫一個int64型別的值就需要兩次操作。（參考：https://golang.google.cn/ref/mem#tmp_2）

因為Golang中對變數的讀和寫都沒有原子性的保證，所以很可能出現這種情況：鎖裡邊變數賦值只處理了一半，鎖外邊的另一個goroutine就讀到了未完全賦值的變數。所以這個雙檢鎖的實現是不安全的。

Golang中將這種問題稱為data race，說的是對某個資料產生了並行讀寫，讀到的資料不可預測，可能產生問題，甚至導致程式崩潰。可以在構建或者執行時檢查是否會發生這種情況：

$ go test -race mypkg    // to test the package
$ go run -race mysrc.go  // to run the source file
$ go build -race mycmd   // to build the command
$ go install -race mypkg // to install the package

另外上邊說單條賦值操作沒有重排序的問題，但是重排序問題在Golang中還是存在的，稍不注意就可能寫出BUG來。比如下邊這段程式碼：

a=1
b=1
c=a+b

在執行這段程式的goroutine中並不會出現問題，但是另一個goroutine讀取到b1時並不代表此時a1，因為a=1和b=1的執行順序可能會被改變。針對重排序問題，Golang並沒有暴露類似volatile的關鍵字，因為理解和正確使用這類能力進行並行程式設計的門檻比較高，所以Golang只是在一些自己認為比較適合的地方插入了記憶體柵欄，儘量保持語言的簡單。對於goroutine之間的資料同步，Go提供了更好的方式，那就是Channel，不過這不是本文的重點，這裡就不介紹了。

sync.Once的啟示

還是回到最開始的問題，如何在Golang中建立一個單例？

很多人應該會被推薦使用 sync.Once ，這裡看下如何使用：

type Conn struct {
	Addr  string
	State int
}

var c *Conn
var once sync.Once

func setInstance() {
	fmt.Println("setup")
	c = &Conn{"127.0.0.1:8080", 1}
}

func doPrint() {
	once.Do(setInstance)
	fmt.Println(c)
}

func loopPrint() {
	for i := 0; i < 10; i++ {
		go doprint()
	}
}

這裡重用上文的結構體Conn，設定Conn單例的方法是setInstance，這個方法在doPrint中被once.Do呼叫，這裡的once就是sync.Once的一個範例，然後我們在loopPrint方法中建立10個goroutine來呼叫doPrint方法。

按照sync.Once的語意，setInstance應該近執行一次。可以實際執行下看看，我這裡直接貼出結果：

setup
&{127.0.0.1:8080 1}
&{127.0.0.1:8080 1}
&{127.0.0.1:8080 1}
&{127.0.0.1:8080 1}
&{127.0.0.1:8080 1}
&{127.0.0.1:8080 1}
&{127.0.0.1:8080 1}
&{127.0.0.1:8080 1}
&{127.0.0.1:8080 1}
&{127.0.0.1:8080 1}

無論執行多少遍，都是這個結果。那麼sync.Once是怎麼做到的呢？原始碼很短很清楚：

type Once struct {
	done uint32
	m    Mutex
}

func (o *Once) Do(f func()) {
	if atomic.LoadUint32(&o.done) == 0 {
		o.doSlow(f)
	}
}

func (o *Once) doSlow(f func()) {
	o.m.Lock()
	defer o.m.Unlock()
	if o.done == 0 {
		defer atomic.StoreUint32(&o.done, 1)
		f()
	}
}

Once是一個結構體，其中第一個欄位標識是否執行過，第二個欄位是一個互斥量。Once僅公開了一個Do方法，用於執行目標函數f。

這裡重點看下目標函數f是怎麼被執行的？

1. Do方法中第一行是判斷欄位done是否為0，為0則代表沒執行過，為1則代表執行過。這裡用了原子讀，寫的時候也要原子寫，這樣可以保證讀寫不會同時發生，能夠讀到當前最新的值。
2. 如果done為0，則呼叫doSLow方法，從名字我們就可以體會到這個方法比較慢。
3. doSlow中首先會加鎖，使用的是Once結構體的第二個欄位。
4. 然後再判斷done是否為0，注意這裡沒有使用原子讀，為什麼呢？因為鎖中的方法是序列執行的，不會發生並行讀寫。
5. 如果done為0，則呼叫目標函數f，執行相關的業務邏輯。
6. 在執行目標函數f前，這裡還宣告了一個defer：defer atomic.StoreUint32(&o.done, 1) ，使用原子寫改變done的值為1，代表目標函數已經執行過。它會在目標函數f執行完畢，doSlow方法返回之前執行。這個設計很精妙，精確控制了改寫done值的時機。

可以看出，這裡用的也是雙檢鎖的模式，只不過做了兩個增強：一是使用原子讀寫，避免了並行讀寫的記憶體資料不一致問題；二是在defer中更改完成標識，保證了程式碼執行順序，不會出現完成標識更改邏輯被編譯器或者CPU優化提前執行。

需要注意，如果目標函數f中發生了panic，目標函數也僅執行一次，不會執行多次直到成功。

安全的雙檢鎖

有了對sync.Once的理解，我們可以改造之前寫的雙檢鎖邏輯，讓它也能安全起來。

type Conn struct {
	Addr  string
	State int
}

var c *Conn
var mu sync.Mutex
var done uint32

func getInstance() *Conn {
	if atomic.LoadUint32(&done) == 0 {
		mu.Lock()
		defer mu.Unlock()
		if done == 0 {
			defer atomic.StoreUint32(&done, 1)
			c = &Conn{"127.0.0.1:8080", 1}
		}
	}
	return c
}

改變的地方就是sync.Once做的兩個增強；原子讀寫和defer中更改完成標識。

當然如果要做的工作僅限於此，還不如直接使用sync.Once。

有時候我們需要的單例不是一成不變的，比如在ylog中需要每小時建立一個紀錄檔檔案的範例，再比如需要為每一個使用者建立不同的單例；再比如建立範例的過程中發生了錯誤，可能我們還會期望再執行範例的建立過程，直到成功。這兩個需求是sync.Once無法做到的。

處理panic

這裡在建立Conn的時候模擬一個panic。

i:=0
func newConn() *Conn {
	fmt.Println("newConn")
	div := i
	i++
	k := 10 / div
	return &Conn{"127.0.0.1:8080", k}
}

第1次執行newConn時會發生一個除零錯誤，並引發 panic。再執行時則可以正常建立。

panic可以通過recover進行處理，因此可以在捕捉到panic時不更改完成標識，之前的getInstance方法可以修改為：

func getInstance() *Conn {
	if atomic.LoadUint32(&done) == 0 {
		mu.Lock()
		defer mu.Unlock()

		if done == 0 {
			defer func() {
				if r := recover(); r == nil {
					defer atomic.StoreUint32(&done, 1)
				}
			}()

			c = newConn()
		}
	}
	return c
}

可以看到這裡只是改了下defer函數，捕捉不到panic時才去更改完成標識。注意此時c並沒有建立成功，會返回零值，或許你還需要增加其它的錯誤處理。

處理error

如果業務程式碼不是丟擲panic，而是返回error，這時候怎麼處理？

可以將error轉為panic，比如newConn是這樣實現的：

func newConn() (*Conn, error) {
	fmt.Println("newConn")
	div := i
	i++
	if div == 0 {
		return nil, errors.New("the divisor is zero")
	}
	k := 1 / div
	return &Conn{"127.0.0.1:8080", k}, nil
}

我們可以再把它包裝一層：

func mustNewConn() *Conn {
	conn, err := newConn()
	if err != nil {
		panic(err)
	}
	return conn
}

如果不使用panic，還可以再引入一個變數，有error時對它賦值，在defer函數中增加對這個變數的判斷，如果有錯誤值，則不更新完成標識位。程式碼也比較容易實現，不過還要增加變數，感覺複雜了，這裡就不測試這種方法了。

有範圍的單例

前文提到過有時單例不是一成不變的，我這裡將這種單例稱為有範圍的單例。

這裡還是複用前文的Conn結構體，不過需求修改為要為每個使用者建立一個Conn範例。

看一下User的定義：

type User struct {
	done uint32
	Id   int64
	mu   sync.Mutex
	c    *Conn
}

其中包括一個使用者Id，其它三個欄位還是用於獲取當前使用者的Conn單例的。

再看看getInstance函數怎麼改：

func getInstance(user *User) *Conn {
	if atomic.LoadUint32(&user.done) == 0 {
		user.mu.Lock()
		defer user.mu.Unlock()

		if user.done == 0 {
			defer func() {
				if r := recover(); r == nil {
					defer atomic.StoreUint32(&user.done, 1)
				}
			}()

			user.c = newConn()
		}
	}
	return user.c
}

這裡增加了一個引數 user，方法內的邏輯基本沒變，只不過操作的東西都變成user的欄位。這樣就可以為每個使用者建立一個Conn單例。

這個方法有點泛型的意思了，當然不是泛型。

有範圍單例的另一個範例：在ylog中需要每小時建立一個紀錄檔檔案用於記錄當前小時的紀錄檔，在每個小時只需建立並開啟這個檔案一次。

先看看Logger的定義（這裡省略和建立單例無關的內容。）：

type FileLogger struct {
	lastHour int64
	file     *os.File
	mu       sync.Mutex
	...
}

lastHour是記錄的小時數，如果當前小時數不等於記錄的小時數，則說明應該建立新的檔案，這個變數類似於sync.Once中的done欄位。

file是開啟的檔案範例。

mu是建立檔案範例時需要加的鎖。

下邊看一下開啟檔案的方法：

func (l *FileLogger) ensureFile() (err error) {
	curTime := time.Now()
	curHour := getTimeHour(curTime)
	if atomic.LoadInt64(&l.lastHour) != curHour {
		return l.ensureFileSlow(curTime, curHour)
	}

	return
}

func (l *FileLogger) ensureFileSlow(curTime time.Time, curHour int64) (err error) {
	l.mu.Lock()
	defer l.mu.Unlock()
	if l.lastHour != curHour {
		defer func() {
			if r := recover(); r == nil {
				atomic.StoreInt64(&l.lastHour, curHour)
			}
		}()
		l.createFile(curTime, curHour)
	}
	return
}

這裡模仿sync.Once中的處理方法，有兩點主要的不同：數值比較不再是0和1，而是每個小時都會變化的數位；增加了對panic的處理。如果開啟檔案失敗，則還會再次嘗試開啟檔案。

要檢視完整的程式碼請存取Github：https://github.com/bosima/ylog/tree/1.0

雙檢鎖的效能

從原理上分析，雙檢鎖的效能要好過互斥鎖，因為互斥鎖每次都要加鎖；不使用原子操作的雙檢鎖要比使用原子操作的雙檢鎖好一些，畢竟原子操作也是有些成本的。那麼實際差距是多少呢？

這裡做一個Benchmark Test，還是處理上文的Conn結構體，為了方便測試，定義一個上下文：

type Context struct {
	done uint32
	c    *Conn
	mu   sync.Mutex
}

編寫三個用於測試的方法：

func ensure_unsafe_dcl(context *Context) {
	if context.done == 0 {
		context.mu.Lock()
		defer context.mu.Unlock()
		if context.done == 0 {
			defer func() { context.done = 1 }()
			context.c = newConn()
		}
	}
}

func ensure_dcl(context *Context) {
	if atomic.LoadUint32(&context.done) == 0 {
		context.mu.Lock()
		defer context.mu.Unlock()
		if context.done == 0 {
			defer atomic.StoreUint32(&context.done, 1)
			context.c = newConn()
		}
	}
}

func ensure_mutex(context *Context) {
	context.mu.Lock()
	defer context.mu.Unlock()
	if context.done == 0 {
    defer func() { context.done = 1 }()
		context.c = newConn()
	}
}

這三個方法分別對應不安全的雙檢鎖、使用原子操作的安全雙檢鎖和每次都加互斥鎖。它們的作用都是確保Conn結構體的範例存在，如果不存在則建立。

使用的測試方法都是下面這種寫法，按照計算機邏輯處理器的數量並行執行測試方法：

func BenchmarkInfo_DCL(b *testing.B) {
	context := &Context{}
	b.ResetTimer()
	b.RunParallel(func(pb *testing.PB) {
		for pb.Next() {
			ensure_dcl(context)
			processConn(context.c)
		}
	})
}

先看一下Benchmark Test的結果：

可以看到使用雙檢鎖相比每次加鎖的提升是兩個數量級，這是正常的。

而不安全的雙檢鎖和使用原子操作的安全雙檢鎖時間消耗相差無幾，為什麼呢？

主要原因是這裡寫只有1次，剩下的全是讀。即使使用了原子操作，絕大部分情況下CPU讀資料的時候也不用在多個核心之間同步（鎖匯流排、鎖快取等），只需要讀快取就可以了。這也從一個方面證明了雙檢鎖模式的意義。

另外上文提到過Go讀寫超過一個機器字的變數時是非原子的，那如果讀寫只有1個機器字呢？在64位元機器上讀寫int64本身就是原子操作，也就是說讀寫應該都只需1次操作，不管用不用atomic方法。這可以在編譯器檔案或者CPU手冊中驗證。（Reference：https://preshing.com/20130618/atomic-vs-non-atomic-operations/）

不過這兩個分析不是說我們使用原子操作沒有意義，不安全雙檢鎖的執行結果是沒有Go語言規範保證的，上邊的結果只是在特定編譯器、特定平臺下的基準測試結果，不同的編譯器、CPU，甚至不同版本的Go都不知道會出什麼么蛾子，執行的效果也就無法保證。我們不得不考慮程式的可移植性。

---

以上就是本文主要內容，如有問題歡迎反饋。完整程式碼已經上傳到Github，歡迎存取：https://github.com/bosima/go-demo/tree/main/double-check-locking