iOS中為什麼會有這麼多鎖呢?

語言: CN / TW / HK

其實iOS領域很多文章都談到了關於鎖的文章,但是我為什麼要在這裡重新寫一篇文章呢?一是很多文章使用的觀點依然是很老的觀點,和我的測試結果不符合,二則是自己對這方面也比較生疏,所以就在最近重新梳理一下自己對著方面的調查,梳理一下這一塊的知識點。

首先是一波對比,我使用了10^7次遍歷,使用的開發語言是Swift,在iOS15.5系統版本的iPhone13真機上跑出的資料:

iOS中不同鎖的效能對比.png

整體來說NSConditionLock的效能會略慢,但是其他的效能都類似,在這個量級的資料處理下,它們的表現都非常的接近。從圖中可以看出效能最好的三個鎖是os_unfair_lockpthread_mutex以及DispatchSemaphore,前兩者是互斥鎖,後者是訊號量。

首先我想提出一個問題,那就是鎖的目的是什麼?

在聊鎖的目的之前,那首先我們來看一個概念,那就是執行緒安全。什麼是執行緒安全?我的定義是當多執行緒都需要操作某個共享資料時,並不會引起意料之外的情況,能保證該共享資料的正確性。可是如何去實現一個執行緒安全類呢?通用的方式就是在一些資料的操作上加鎖。而鎖的目的就是確保多執行緒操作共享資料時,能保證資料的準確性和可預測性。

os_unfair_lock

我相信有很多人都閱讀過ibireme關於鎖的效能對比的知名文章《不再安全的 OSSpinLock》,其中提到了OSSpinLock不再安全的理由,但是由此卻引發一個問題,那就是OSSpinLock主要的使用場景是哪裡呢?

我們都知道在Objective-C中定義一個屬性的時候,有時屬性會被宣告為atomic,這就是說這個屬性的set操作和get操作是原子性的,那麼如何確保這些 操作的原子性呢?我想這個時候你已經猜到答案了,Apple使用的方案是OSSpinLock,這是一個自旋鎖,但是這個鎖有一個很嚴重的問題,那就是優先順序反轉問題會導致自旋鎖發生死鎖。

iOS 系統中維護了 5 個不同的執行緒優先順序/QoS: background,utility,default,user-initiated,user-interactive。高優先順序執行緒始終會在低優先順序執行緒前執行,一個執行緒不會受到比它更低優先順序執行緒的干擾。這種執行緒排程演算法會產生潛在的優先順序反轉問題,從而破壞了 spin lock。

具體來說,如果一個低優先順序的執行緒獲得鎖並訪問共享資源,這時一個高優先順序的執行緒也嘗試獲得這個鎖,它會處於 spin lock 的忙等狀態從而佔用大量 CPU。此時低優先順序執行緒無法與高優先順序執行緒爭奪 CPU 時間,從而導致任務遲遲完不成、無法釋放 lock。這並不只是理論上的問題,libobjc 已經遇到了很多次這個問題了,於是蘋果的工程師停用了 OSSpinLock。

蘋果工程師 Greg Parker 提到,對於這個問題,一種解決方案是用 truly unbounded backoff 演算法,這能避免 livelock 問題,但如果系統負載高時,它仍有可能將高優先順序的執行緒阻塞數十秒之久;另一種方案是使用 handoff lock 演算法,這也是 libobjc 目前正在使用的。鎖的持有者會把執行緒 ID 儲存到鎖內部,鎖的等待者會臨時貢獻出它的優先順序來避免優先順序反轉的問題。理論上這種模式會在比較複雜的多鎖條件下產生問題,但實踐上目前還一切都好。

而在iOS 10之後,Apple使用了os_unfair_lock來替代了OSSpinLock, 這是一個高效能的互斥鎖,而不是自旋鎖,如果是阻止兩個執行緒可以同時訪問臨界區,那麼這個鎖無疑可以很好的完成工作,包括上述的pthread_mutex_lock 以及訊號量都可以,但是如果我們需要鎖具備某些特性,那麼這個時候就需要其他多種類的鎖了。

```swift // os_unfair_lock的使用 var unfairLock = os_unfair_lock() os_unfair_lock_lock(&unfairLock) os_unfair_lock_unlock(&unfairLock)

// pthreadMutex的使用 var pthreadMutex = pthread_mutex_t() pthread_mutex_lock(&pthreadMutex) pthread_mutex_unlock(&pthreadMutex) ```

這裡再補充說明一下,Apple使用在保證原子性時實際會呼叫到的方法如下:

c static inline void reallySetProperty() { ... if (!atomic) { oldValue = *slot; *slot = newValue; } else { //PropertyLocks是一個StripedMap<spinlock_t>型別的全域性變數 //而StripedMap是一個用陣列來實現的hashmap,key是指標,value是型別是spinlock_t物件 //而spinlock_t則是mutex_tt<LOCKDEBUG>的類,而mutex_tt類內部是由os_unfair_lock mLock來實現 //所以,PropertyLocks[slot]目的就是獲取os_unfair_lock物件 spinlock_t& slotlock = PropertyLocks[slot]; slotlock.lock(); oldValue = *slot; *slot = newValue; slotlock.unlock(); } ... }

它通過地址從PropertyLocks陣列中取出了spinlock_t鎖,可是如何使用地址作為陣列下標呢?它使用了一個很巧妙的hash演算法,來實現指標到陣列下標的轉化:

c static unsigned int indexForPointer(const void *p) { uintptr_t addr = reinterpret_cast<uintptr_t>(p); // 這是一個雜湊演算法,可以將物件的地址轉化為陣列的下標 // 使得陣列元素在0~StripeCount之間 return ((addr >> 4) ^ (addr >> 9)) % StripeCount; }

當然這種方法也會偶爾導致雜湊衝突,兩個不同的地址會導致獲取到同一個Lock,這樣會造成資源閒置,沒有充分利用CPU的資源,但是不妨礙這個雜湊演算法整體上是高效的。

NSLock

既然已經有了效能比較高的互斥鎖,那為什麼還需要有其它這些雜七雜八的鎖呢?比如說接下來我們要提到的NSLock,這個鎖也是一個互斥鎖,而它是基於pthread_mutex_lock的封裝,而在原有的基礎上增加了一個特性那就是超時!沒錯這就是有其他各種鎖的原因,給不同的鎖不同的特性,以滿足具體的開發場景,NSLock的API如下:

``swift open class NSLock : NSObject, NSLocking { open functry`() -> Bool

open func lock(before limit: Date) -> Bool

open var name: String?

} ```

在某些時候,超時這個特性是非常有效的,因為在一些可能發生死鎖的場景中,使用NSLock可以讓我們有一個保險機制,即使發生了死鎖,也可以在一定的時間之後走出加鎖狀態,恢復到正常的程式處理邏輯。但是和以上的互斥鎖一樣,它都無法應對遞迴的情況,那使用什麼來處理遞迴鎖呢?NSRecursiveLock!

NSRecursiveLock

使用NSRecursiveLock可以使得該鎖被同一執行緒多次獲取而不會導致執行緒死鎖。但是每一次lock都對應一次unlock,這樣unlock結束之後,鎖才會釋放。而顧名思義,這種型別的鎖被用於一個遞迴方法內部來防止執行緒被阻塞。

```swift let rlock = NSRecursiveLock()

class RThread : Thread {

override func main(){
    rlock.lock()
    print("Thread acquired lock")
    callMe()
    rlock.unlock()
    print("Exiting main")
}

func callMe(){
    rlock.lock()
    print("Thread acquired lock")
    rlock.unlock()
    print("Exiting callMe")
}

}

var tr = RThread() tr.start()

// 多次申請鎖,並不會導致崩潰,這就是遞迴鎖的作用 ```

NSConditionLock

條件鎖滿足NSLocking 協議,所以基本的NSLock型別鎖的基本lock,unlock這種全域性的鎖方法它也是具備的,初次之外,它還具備自己的特性,通常情況下,當執行緒需要以某種特定的順序執行任務時,比如一個執行緒生產資料,而另一個執行緒消耗資料時,可以使用NSConditionLock(比如常見的生產者消費者模型)。接下來我們來看一個例項:

```swift let NODATA = 1 let GOTDATA = 2 let clock = NSConditionLock(condition: NODATA) var shareInt = 0

class ProducerThread: Thread { override func main() { for _ in 0..<100 { clock.lock(whenCondition: NODATA) LockFile.ProducerThread.sleep(forTimeInterval: 0.5) sharedInt = sharedInt + 1 NSLog("生產者:(sharedInt)") clock.unlock(withCondition: GOTDATA) } } }

class ConsumerThread: Thread {
    override func main() {
        for _ in 0..<100 {
            clock.lock(whenCondition: GOTDATA)
            sharedInt = sharedInt - 1
            NSLog("消費者:\(sharedInt)")
            clock.unlock(withCondition: NODATA)
        }
    }

}

let pt = ProducerThread.init() let ct = ConsumerThread.init() pt.start() ct.start() ```

當建立一個條件鎖的時候,需要指定一個特定Int型別的值。而lock(whenCondition:) 方法當條件滿足時會獲取這個鎖,或者條件和另一個執行緒在使用unlock(withCondition:) 釋放鎖時設定的值滿足時,NSConditionLock物件就會獲取鎖執行後續的程式碼片段,但是當lock(whenCondition:) 方法沒有獲取鎖的時候(條件沒滿足時),這個方法會阻塞執行緒的執行,直到獲得鎖為止。

NSCondition

NSCondition和前者是很容易混淆的,但是這個鎖解決了什麼問題呢?

當一個已獲得鎖的執行緒發現執行其工作所需的附加條件(它需要一些資源、另一個處於特定狀態的物件等)暫時還沒有得到滿足時,它需要一種方法來暫停,並且一旦滿足條件就繼續工作的機制,可是如何實現呢?可以通過連續的檢查(忙等待)來實現,但是這樣做的話,執行緒持有的鎖會發生什麼?我們應該在等待時保留它們還是釋放它們?還是在滿足條件時再次獲得它們?

NSCondition提供了一種簡潔的方式來提供了這種問題的解決方案,一旦一個執行緒被放在該Condition的等待列表中,它可以通過另一個執行緒Signal來喚醒。以下是具體的案例:

```swift let cond = NSCondition.init() var available = false var sharedString = ""

class WriterThread: Thread { override func main() { for _ in 0..<100 { cond.lock() sharedString = "🤣" available = true cond.signal() cond.unlock() } } }

class PrinterThread: Thread { override func main() { for _ in 0..<100 { cond.lock() while (!available) { cond.wait() } sharedString = "" available = false cond.unlock() } } } ```

當執行緒waits一個條件時,這個Condition物件會unlock當前鎖並且阻塞執行緒。當Condition發出訊號時,系統會喚醒執行緒,然後這個Condition物件會在wait()或者wait(until:)返回之前,這個Condition物件會重新獲取到它的鎖,因此,從執行緒的角度來看,它似乎一直持有者鎖(雖然中途它會失去鎖)。

Dispatch Semaphore

最後我們聊一聊訊號量,簡而言之,訊號量是需要在不同的執行緒中進行鎖定和解鎖時使用的鎖。因為它的wait方法會阻塞當前執行緒,所以需要其他執行緒發來signal訊號來喚醒它。

swift let semaphore = DispatchSemaphore.init(value: 0) DispatchQueue.global(qos: .userInitiated).async { // to do some thing semaphore.signal() } semaphore.wait() // will block thread

如上述例子一樣,訊號量通常用於鎖定一個執行緒,直到另外一個執行緒中事件的完成後發出signal訊號。從上述的測試圖示,以及其他諸多文章,訊號量的速度是很快的。上述的生產者消費者模型也可以使用訊號量來實現:

```swift let semaphore = DispatchSemaphore.init(value: 0)

DispatchQueue.global(qos: .userInitiated).async { while true { sleep(1) sharedInt = sharedInt + 1 NSLog("生產了: (sharedInt)") _ = semaphore.signal() } }

DispatchQueue.global(qos: .userInitiated).async { while true { if sharedInt <= 0 { _ = semaphore.wait(timeout: .distantFuture) } else { sharedInt = sharedInt - 1 NSLog("消耗了: (sharedInt)") } } } ```

好了,簡單說了一下我對於鎖的梳理,希望大家也可以從中學到一點東西吧~ 如果有什麼問題,或者錯誤希望大家可以留言指點。

參考

1、《不再安全的OSSPinLock

2、Apple Thread Programming Guide

3、Concurrency in Swift

4、thread safety in swift