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