锁的原理（二）：自旋锁、互斥锁以及读写锁

作者: HotPotCat | 来源:发表于2021-08-23 18:13 被阅读0次

一、锁的分类

在分析其它锁之前，需要先区分清楚锁的区别，基本的锁包括了二类：互斥锁 和 自旋锁。

1.1 自旋锁

自旋锁：线程反复检查锁变量是否可用。由于线程在这一过程中保持执行，因此是一种忙等。一旦获取了自旋锁，线程会一直保持该锁，直至显式释放自旋锁。自旋锁避免了进程上下文的调度开销，因此对于线程只会阻塞很短时间的场合是有效的。

自旋锁 = 互斥锁 + 忙等。OSSpinLock就是自旋锁。

1.2 互斥锁

互斥锁：是一种用于多线程编程中，防止两条线程同时对同一公共资源(比如全局变量)进行读写的机制。该目的通过将代码切片成一个一个的临界区而达成。
在Posix Thread中定义有一套专⻔用于线程同步的mutex函数，mutex用于保证在任何时刻，都只能有一个线程访问该对象。当获取锁操作失败时，线程会进入睡眠，等待锁释放时被唤醒（闲等）。

创建和销毁：

POSIX定义了一个宏PTHREAD_MUTEX_INITIALIZER来静态初始化互斥锁。

int pthread_mutex_init(pthread_mutex_t *mutex, const pthread_mutexattr_t *mutexattr)

pthread_mutex_destroy ()用于注销一个互斥锁。

锁操作相关API：

 int pthread_mutex_lock(pthread_mutex_t *mutex)
 int pthread_mutex_unlock(pthread_mutex_t *mutex)
 int pthread_mutex_trylock(pthread_mutex_t *mutex)

pthread_mutex_trylock()语义与pthread_mutex_lock()类似，不同的是在锁已经被占据时返回EBUSY而不是挂起等待。

互斥锁 分为 递归锁 和 非递归锁

递归锁：
@synchronized：多线程可递归。
NSRecursiveLock：不支持多线程可递归。
pthread_mutex_t(recursive)：多线程可递归。
非递归锁：NSLock、pthread_mutex、dispatch_semaphore、os_unfair_lock。
条件锁：NSCondition、NSConditionLock。
信号量（semaphore）：是一种更高级的同步机制，互斥锁可以说是
semaphore在仅取值0/1时的特例。信号量可以有更多的取值空间，用来实
现更加复杂的同步，而不单单是线程间互斥。dispatch_semaphore

1.2.1 读写锁

读写锁实际是一种特殊的互斥锁，它把对共享资源的访问者划分成读者和写者，读者只对共享资源进行读访问，写者则需要对共享资源进行写操作。这种锁 相对于自旋锁而言，能提高并发性。因为在多处理器系统中，它允许同时有多个读者来访问共享资源，最大可能的读者数为实际的逻辑CPU 数。写者是排他性的，一个读写锁同时只能有一个写者或多个读者(与CPU数相关)，但不能同时既有读者又有写者，在读写锁保持期间也是抢占失效的。

如果读写锁当前没有读者，也没有写者，那么写者可以立刻获得读写锁，否则它必须自旋在那里，直到没有任何写者或读者。如果读写锁没有写者，那么读者可以立即获得该读写锁，否则读者必须自旋在那里，直到写者释放该读写锁。

一次只有一个线程可以占有写模式的读写锁，可以有多个线程同时占有读模式的读写锁。正是因为这个特性，当读写锁是写加锁状态时，在这个锁被解锁之前, 所有试图对这个锁加锁的线程都会被阻塞。当读写锁在读加锁状态时，所有试图以读模式对它进行加锁的线程都可以得到访问权，但是如果线程希望以写模式对此锁进行加锁, 它必须直到所有的线程释放锁。
通常当读写锁处于读模式锁住状态时，如果有另外线程试图以写模式加锁，读写锁通常会阻塞随后的读模式锁请求，这样可以避免读模式锁⻓期占用而导致等待的写模式锁请求⻓期阻塞。
读写锁适合于对数据结构的读次数比写次数多得多的情况。因为读模式锁定可以共享，写模式锁住时意味着独占，所以读写锁又叫共享-独占锁。

创建和销毁API：

#include <pthread.h>
int pthread_rwlock_init(pthread_rwlock_t *restrict rwlock, const pthread_rwlockattr_t *restrict attr);
int pthread_rwlock_destroy(pthread_rwlock_t *rwlock);

成功则返回0，出错则返回错误编号。

同互斥锁一样，在释放读写锁占用的内存之前, 需要先通过pthread_rwlock_destroy对读写锁进行清理工作，释放由init分配的资源。

锁操作相关API：

#include <pthread.h>
int pthread_rwlock_rdlock(pthread_rwlock_t *rwlock);
int pthread_rwlock_wrlock(pthread_rwlock_t *rwlock);
int pthread_rwlock_unlock(pthread_rwlock_t *rwlock);

成功则返回0，出错则返回错误编号。这3个函数分别实现获取读锁，获取写锁和释放锁的操作。获取锁的两个函数是阻塞操作，同样非阻塞的函数为:

#include <pthread.h>
int pthread_rwlock_tryrdlock(pthread_rwlock_t *rwlock);
int pthread_rwlock_trywrlock(pthread_rwlock_t *rwlock);

非阻塞的获取锁操作，如果可以获取则返回0，否则返回错误的EBUSY。

更多概念和解释：
互斥锁
 线程同步

二、NSLock & NSRecursiveLock 的应用以及原理

2.1 案例一

__block NSMutableArray *array;
for (int i = 0; i < 10000; i++) {
    dispatch_async(dispatch_get_global_queue(0, 0), ^{
        array = [NSMutableArray array];
    });
}

对于上面的代码运行会发生崩溃，常规处理是对它加一个锁，如下：

__block NSMutableArray *array;
self.lock = [[NSLock alloc] init];
for (int i = 0; i < 10000; i++) {
    dispatch_async(dispatch_get_global_queue(0, 0), ^{
        [self.lock lock];
        array = [NSMutableArray array];
        [self.lock unlock];
    });
}

这样就能解决array的创建问题了。

2.2 案例二

for (int i = 0; i < 10; i++) {
    dispatch_async(dispatch_get_global_queue(0, 0), ^{
        static void (^testMethod)(int);
        testMethod = ^(int value){
            if (value > 0) {
                NSLog(@"current value = %d",value);
                testMethod(value - 1);
            }
        };
        testMethod(10);
    });
}

上面的例子中最终输出会错乱：

image.png

可以在block调用前后加解锁解决：

for (int i = 0; i < 10; i++) {
    dispatch_async(dispatch_get_global_queue(0, 0), ^{
        static void (^testMethod)(int);
        testMethod = ^(int value){
            if (value > 0) {
                NSLog(@"current value = %d",value);
                testMethod(value - 1);
            }
        };
        [self.lock lock];
        testMethod(10);
        [self.lock unlock];
    });
}

但是在实际开发中锁往往是与业务代码绑定在一起的，如下：

image.png

这个时候block在执行前会同一时间进入多次，相当于多次加锁了（递归），这样就产生了死锁。NSLog只会执行一次。

将NSLock改为NSRecursiveLock可以解决NSLock存在的死锁问题：

for (int i = 0; i < 10; i++) {
    dispatch_async(dispatch_get_global_queue(0, 0), ^{
        static void (^testMethod)(int);
        testMethod = ^(int value){
            [self.recursiveLock lock];
            if (value > 0) {
                NSLog(@"current value = %d",value);
                testMethod(value - 1);
            }
            [self.recursiveLock unlock];
        };
        testMethod(10);
    });
}

但是在执行testMethod一次（也有可能是多次）递归调用后没有继续输出：

image.png

由于NSRecursiveLock不支持多线程可递归。所以改为@synchronized：

    for (int i = 0; i < 10; i++) {
        dispatch_async(dispatch_get_global_queue(0, 0), ^{
            static void (^testMethod)(int);
            testMethod = ^(int value){
                @synchronized (self) {
                    if (value > 0) {
                        NSLog(@"current value = %d",value);
                        testMethod(value - 1);
                    }
                }
            };
            testMethod(10);
        });
    }

就能完美解决问题了。

NSRecursiveLock 解决了 NSLock 递归问题，@synchronized 解决了 NSRecursiveLock 多线程可递归问题问题。

2.3 原理分析

NSLock与NSRecursiveLock是定义在Foundation框架中的，Foundation框架并没有开源。有三种方式来探索：

分析Foundation动态库的汇编代码。
断点跟踪加锁解锁流程。
Swift Foundation源码分析。虽然Foundation框架本身没有，但是苹果开源了Swift Foundation的代码。原理是想通的。swift-corelibs-foundation

当然有兴趣可以尝试编译可运行版本进行调试 swift-foundation 源码编译

在Foundation中lock与unlock是NSLocking协议提供的方法：

@protocol NSLocking
- (void)lock;
- (void)unlock;
@end

在Swift Foundation源码中同样有NSLocking协议：

public protocol NSLocking {
    func lock()
    func unlock()
}

2.3.1 NSLock 源码分析

image.png
底层是对pthread_mutex_init的封装。lock和unlock同样是对pthread_mutex_lock与pthread_mutex_unlock的封装：

image.png

通过宏定义可以看到Swift的跨平台支持。

2.3.2 NSRecursiveLock 源码分析

image.png
内部是对PTHREAD_MUTEX_RECURSIVE的封装。lock与unlock同样是对pthread_mutex_lock与pthread_mutex_unlock的封装。

三、NSCondition 原理

NSCondition实际上作为一个锁和一个 线程检查器。锁主要为了当检测条件时保护数据源，执行条件引发的任务；线程检查器主要是根据条件决定是否继续运行线程，即线程是否被阻塞。

[condition lock]：一般用于多线程同时访问、修改同一个数据源，保证在同一时间内数据源只被访问、修改一次，其他线程的命令需要在lock 外等待，只到 unlock后才可访问。
[condition unlock]：与lock同时使用。
[condition wait]：让当前线程处于等待状态。
[condition signal]：CPU发信号告诉线程不用在等待，可以继续执行。

3.1 生产者-消费者案例

- (void)testNSCondition {
    //创建生产-消费者
    for (int i = 0; i < 10; i++) {
        dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_HIGH, 0), ^{
            [self test_producer];
        });
        dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_HIGH, 0), ^{
            [self test_consumer];
        });
        
        dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_HIGH, 0), ^{
            [self test_consumer];
        });
        dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_HIGH, 0), ^{
            [self test_producer];
        });
    }
}

- (void)test_producer{
    [self.condition lock];
    self.ticketCount = self.ticketCount + 1;
    NSLog(@"生产 + 1 剩余: %zd",self.ticketCount);
    [self.condition signal]; // 信号
    [self.condition unlock];
}

- (void)test_consumer{
    [self.condition lock];
    if (self.ticketCount == 0) {
        NSLog(@"等待 剩余: %zd",self.ticketCount);
        [self.condition wait];
    }
    //消费行为，要在等待条件判断之后
    self.ticketCount -= 1;
    NSLog(@"消费 - 1 剩余: %zd ",self.ticketCount);
    [self.condition unlock];
}

输出：

生产 + 1 剩余: 1
消费 - 1 剩余: 0 
等待 剩余: 0
生产 + 1 剩余: 1
消费 - 1 剩余: 0 
等待 剩余: 0
等待 剩余: 0
等待 剩余: 0
生产 + 1 剩余: 1
消费 - 1 剩余: 0 
等待 剩余: 0
生产 + 1 剩余: 1
消费 - 1 剩余: 0 
等待 剩余: 0
等待 剩余: 0
生产 + 1 剩余: 1
消费 - 1 剩余: 0 
生产 + 1 剩余: 1
消费 - 1 剩余: 0 
生产 + 1 剩余: 1
消费 - 1 剩余: 0 
等待 剩余: 0
等待 剩余: 0
生产 + 1 剩余: 1
消费 - 1 剩余: 0 
生产 + 1 剩余: 1
消费 - 1 剩余: 0 
等待 剩余: 0
生产 + 1 剩余: 1
消费 - 1 剩余: 0 
生产 + 1 剩余: 1
消费 - 1 剩余: 0 
生产 + 1 剩余: 1
生产 + 1 剩余: 2
消费 - 1 剩余: 1 
消费 - 1 剩余: 0 
生产 + 1 剩余: 1
生产 + 1 剩余: 2
生产 + 1 剩余: 3
消费 - 1 剩余: 2 
生产 + 1 剩余: 3
消费 - 1 剩余: 2 
生产 + 1 剩余: 3
消费 - 1 剩余: 2 
消费 - 1 剩余: 1 
生产 + 1 剩余: 2
生产 + 1 剩余: 3
消费 - 1 剩余: 2 
消费 - 1 剩余: 1 
消费 - 1 剩余: 0

因为有condition的存在保证了消费行为是在对应的生产行为之后。在这个过程中会有消费等待行为，signal信号通知消费。

生产和消费的加锁保证了各个事务的额安全。
wait和signal保证了事务之间的安全。

3.2 源码分析

image.png
内部也是对pthread_mutex_init的包装，多了一个pthread_cond_init。

open func lock() {
    pthread_mutex_lock(mutex)
}

open func unlock() {
    pthread_mutex_unlock(mutex)
}

open func wait() {
    pthread_cond_wait(cond, mutex)
}

open func signal() {
    pthread_cond_signal(cond)
}

open func broadcast() {
    pthread_cond_broadcast(cond)
}

代码中去掉了windows相关宏逻辑：

NSCondition：锁（pthread_mutex_t） + 线程检查器（pthread_cond_t）
锁（pthread_mutex_t）：lock(pthread_mutex_lock) + unlock(pthread_mutex_unlock)
线程检查器（pthread_cond_t）：wait(pthread_cond_wait) + signal(pthread_cond_signal)

四、NSConditionLock 使用和原理

NSConditionLock也是锁，一旦一个线程获得锁，其他线程一定等待。它同样遵循NSLocking协议，相关API:

- (void)lockWhenCondition:(NSInteger)condition;
- (void)unlockWithCondition:(NSInteger)condition;
- (BOOL)lockWhenCondition:(NSInteger)condition beforeDate:(NSDate *)limit;

[conditionLock lock]：表示 conditionLock 期待获得锁，如果没有其他线程获得锁(不需要判断内部的condition) 那它能执行此行以下代码，如果已经有其他线程获得锁(可能是条件锁，或者无条件锁)，则等待，直至其他线程解锁。
[conditionLock lockWhenCondition:A条件]：表示如果没有其他线程获得该锁，但是该锁内部的 condition不等于A条件，它依然不能获得锁，仍然等待。如果内部的condition等于A条件，并且没有其他线程获得该锁，则进入代码区，同时设置它获得该锁，其他任何线程都将等待它代码的完成，直至它解锁。
[conditionLock unlockWithCondition:A条件]: 表示释放锁，同时把内部的condition设置为A条件。
return = [conditionLock lockWhenCondition:A条件 beforeDate:A时间]：表示如果被锁定(没获得锁)，并超过该时间则不再阻塞线程。注意：返回的值是NO，它没有改变锁的状态，这个函数的目的在于可以实现两种状态下的处理。
所谓的condition就是整数，内部通过整数比较条件。

4.1案例

NSConditionLock *conditionLock = [[NSConditionLock alloc] initWithCondition:2];
dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_HIGH, 0), ^{
    [conditionLock lockWhenCondition:1];
    NSLog(@"1");
    [conditionLock unlockWithCondition:0];
});

dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_LOW, 0), ^{
    [conditionLock lockWhenCondition:2];
    sleep(1);
    NSLog(@"2");
    [conditionLock unlockWithCondition:1];
});

dispatch_async(dispatch_get_global_queue(0, 0), ^{
   [conditionLock lock];
   NSLog(@"3");
   [conditionLock unlock];
});

上面的案例2一定比1先执行，2与3之间无序。
输出：3 2 1，如果任务2的优先级改为High则输出顺序变为2 1 3。

那么有以下疑问：

1.NSConditionLock 与 NSCondition有关系么？
2.NSConditionLock初始化的时候condition是什么？
3.lockWhenCondition是如何控制的？
4.unlockWithCondition是如何控制的？

4.2 断点调试分析逻辑

在拿不到源码以及拿不到动态库的情况下，断点分析调用流程是一个比较好的方案。
分别在测试代码中打下以下断点：

image.png

运行工程到达断点后下符号断点-[NSConditionLock initWithCondition:]过掉断点:

image.png
这个时候就进入了initWithCondition的汇编实现。在汇编中对所有的b（跳转指令）下断点配合寄存器的值跟踪流程。

-[NSConditionLock initWithCondition:]:
image.png
可以通过lldb读取寄存器的值，也可以查看全部寄存器中对应的值。

过掉断点继续：

image.png
执行了：- [xxx init]。继续下一步：

image.png
调用了-[NSConditionLock init]，继续下一步：

image.png
调用了-[NSConditionLock zone]，继续下异步：

image.png
调用了+[NSCondition allocWithZone:]，继续下一步：

image.png
调用了-[NSCondition init]，继续：

image.png
最终返回了创建的NSConditionLock对象，它持有NSCondition对象以及初始化传的condition参数2。
-[NSConditionLock initWithCondition:]流程：

-[NSConditionLock initWithCondition:]
    -[xxx init]
    -[NSConditionLock init]
    -[NSConditionLock zone]
    +[NSCondition allocWithZone:]
    -[NSCondition init]

-[NSConditionLock lockWhenCondition:]：
同样添加-[NSConditionLock lockWhenCondition:]符号断点：
image.png
调用了[NSDate distantFuture]，继续：
image.png
调用了-[NSConditionLock lockWhenCondition:beforeDate:]，对它增加一个符号断点：
image.png

调用了-[NSCondition lock]，继续跟踪：

image.png
调用了-[NSCondition waitUntilDate:]

image.png
在调试的过程中会跳转另外一个线程的-[NSConditionLock lockWhenCondition:]，流程与上面的一致，继续调用会调用了-[NSCondition unlock]:

image.png
这个时候直接返回1(true)，接着对-[NSConditionLock unlockWithCondition:]下符号断点：

image.png
调用了-[NSCondition lock]：

image.png
调用了-[NSCondition broadcast]:

image.png
调用了-[NSCondition unlock]，这个时候继续过断点就又会回到线程4，调用逻辑和线程3相同。

完整调用逻辑如下：

线程4
-[NSConditionLock lockWhenCondition:]
    +[NSDate distantFuture]
    -[NSConditionLock lockWhenCondition:beforeDate:]
        -[NSCondition lock]
        -[NSCondition waitUntilDate:]
            线程3
            -[NSConditionLock lockWhenCondition:]
                +[NSDate distantFuture]
                -[NSConditionLock lockWhenCondition:beforeDate:]
                    -[NSCondition lock]
                    -[NSCondition unlock]
                    返回1（true）
            -[NSConditionLock unlockWithCondition:]
                -[NSCondition lock]
                -[NSCondition broadcast]
                -[NSCondition unlock]
        //回到线程4
        -[NSCondition unlock]
        返回1（true）
-[NSConditionLock unlockWithCondition:]
    -[NSCondition lock]
    -[NSCondition broadcast]
    -[NSCondition unlock]

流程总结：

线程4调用[NSConditionLock lockWhenCondition:]，此时因为不满足当前条件，所
以会进入 waiting 状态，当前进入到 waiting 时，会释放当前的互斥锁。
此时当前的线程2 调用[NSConditionLock lock:]，本质上是调用[NSConditionLock lockBeforeDate:]这里不需要比对条件值，所以任务3会执行。
接下来线程3 执行[NSConditionLock lockWhenCondition:]，因为满足条件值，所以线任务2会执行，执行完成后会调用[NSConditionLock unlockWithCondition:],这个时候将
condition 设置为 1，并发送 boradcast, 此时线程 4接收到当前的信号，唤醒执行并打印。
这个时候任务执行顺序为任务3 -> 任务2 -> 任务1。
[NSConditionLock lockWhenCondition:]会根据传入的 condition进
行对比，如果不相等，这里就会阻塞，进入线程池，否则的话就继续代码执行。
[NSConditionLock unlockWithCondition:] 会先更改当前的condition值，然后进行广播，唤醒当前的线程。

4.3 源码分析

initWithCondition：

image.png
保存了condition参数以及NSCondition的创建。
lockWhenCondition：

open func lock(whenCondition condition: Int) {
    let _ = lock(whenCondition: condition, before: Date.distantFuture)
}

内部调用了lockWhenCondition: before:，默认值传的Date.distantFuture：

open func lock(whenCondition condition: Int, before limit: Date) -> Bool {
    _cond.lock()
    while _thread != nil || _value != condition {
        if !_cond.wait(until: limit) {
            _cond.unlock()
            return false
        }
    }
    _thread = pthread_self()
    _cond.unlock()
    return true
}

NSCondition加锁判断condition条件是否满足，不满足调用NSCondition的wait waitUntilDate方法进入等待，超时后解锁返回false。满足的情况下赋值_thread解锁返回true。

unlockWithCondition：

open func unlock(withCondition condition: Int) {
    _cond.lock()
    _thread = nil
    _value = condition
    _cond.broadcast()
    _cond.unlock()
}

加锁后释放_thread，更新condition，调用broadcast后解锁。

lock

open func lock() {
    let _ = lock(before: Date.distantFuture)
}

open func lock(before limit: Date) -> Bool {
    _cond.lock()
    while _thread != nil {
        if !_cond.wait(until: limit) {
            _cond.unlock()
            return false
        }
    }
    _thread = pthread_self()
    _cond.unlock()
    return true
}

判断是否有其它任务阻塞，没有阻塞直接创建_thread返回true。

unlock

open func unlock() {
    _cond.lock()
    _thread = nil
    _cond.broadcast()
    _cond.unlock()
}

广播并且释放_thread。

4.4 反汇编分析

initWithCondition：

image.png
lockWhenCondition：

-(int)lockWhenCondition:(int)arg2 {
    r0 = [arg0 lockWhenCondition:arg2 beforeDate:[NSDate distantFuture]];
    return r0;
}

调用自己的lockWhenCondition: beforeDate :：

image.png

unlockWithCondition:

-(int)unlockWithCondition:(int)arg2 {
    r0 = object_getIndexedIvars(arg0);
    [*r0 lock];
    *(int128_t *)(r0 + 0x8) = 0x0;
    *(int128_t *)(r0 + 0x10) = arg2;
    [*r0 broadcast];
    r0 = *r0;
    r0 = [r0 unlock];
    return r0;
}

lock：

int -[NSConditionLock lock](int arg0) {
    r0 = [arg0 lockBeforeDate:[NSDate distantFuture]];
    return r0;
}

lockBeforeDate:

image.png

unlock:

int -[NSConditionLock unlock]() {
    r0 = object_getIndexedIvars(r0);
    [*r0 lock];
    *(r0 + 0x8) = 0x0;
    [*r0 broadcast];
    r0 = *r0;
    r0 = [r0 unlock];
    return r0;
}

汇编、源码以及断点调试逻辑相同。
NSConditionLock 内部封装了NSCondition。

五、OSSpinLock & os_unfair_lock

image.png
OSSpinLock的API注释以及它自己的命名说明了这是一把自旋锁，自iOS10之后被os_unfair_lock替代。

image.png

os_unfair_lock必须以OS_UNFAIR_LOCK_INIT初始化。
它是用来代替OSSpinLock的。
它不是自旋锁（忙等），是被内核唤醒的（闲等）。

image.png

image.png
可以看到这两个锁都是定义在libsystem_platform.dylib中的。可以在openSource中找到他们libplatform的源码libplatform，实现是在/src/os目录下的lock.c文件中。

5.1 OSSpinLock 源码分析

OSSpinLock的使用一般会用到以下API：

OSSpinLock hp_spinlock = OS_SPINLOCK_INIT;
OSSpinLockLock(&hp_spinlock);
OSSpinLockUnlock(&hp_spinlock);

OSSpinLock：

typedef int32_t OSSpinLock OSSPINLOCK_DEPRECATED_REPLACE_WITH(os_unfair_lock);

#define OS_SPINLOCK_INIT    0

OSSpinLock本身是一个int32_t类型的值，初始化默认值为0。

5.1.1 OSSpinLockLock

void
OSSpinLockLock(volatile OSSpinLock *l)
{
    OS_ATOMIC_ALIAS(spin_lock, OSSpinLockLock);
    OS_ATOMIC_ALIAS(_spin_lock, OSSpinLockLock);
    bool r = os_atomic_cmpxchg(l, 0, _OSSpinLockLocked, acquire);
    if (likely(r)) return;
    return _OSSpinLockLockSlow(l);
}

#if TARGET_OS_IPHONE && !TARGET_OS_SIMULATOR
static const OSSpinLock _OSSpinLockLocked = 1;
#else
static const OSSpinLock _OSSpinLockLocked = -1;
#endif

OS_ATOMIC_ALIAS定义如下：

#undef OS_ATOMIC_ALIAS
#define OS_ATOMIC_ALIAS(n, o)
static void _OSSpinLockLock(volatile OSSpinLock *l);

这里相当于分了两条路径，通过_OSSpinLockLocked标记是否被锁定。在源码中并没有找到_OSSpinLockLock函数的实现。

5.1.1.1 _OSSpinLockLockSlow

#if OS_ATOMIC_UP
void
_OSSpinLockLockSlow(volatile OSSpinLock *l)
{
    return _OSSpinLockLockYield(l); // Don't spin on UP
}
#elif defined(__arm64__)
// Exclusive monitor must be held during WFE <rdar://problem/22300054>
#if defined(__ARM_ARCH_8_2__)
void
_OSSpinLockLockSlow(volatile OSSpinLock *l)
{
    uint32_t tries = OS_LOCK_SPIN_SPIN_TRIES;
    OSSpinLock lock;
_spin:
    while (unlikely(lock = os_atomic_load_exclusive(l, relaxed))) {
        if (unlikely(lock != _OSSpinLockLocked)) {
            os_atomic_clear_exclusive();
            return _os_lock_corruption_abort((void *)l, (uintptr_t)lock);
        }
        if (unlikely(!tries--)) {
            os_atomic_clear_exclusive();
            return _OSSpinLockLockYield(l);
        }
        OS_LOCK_SPIN_PAUSE();
    }
    os_atomic_clear_exclusive();
    bool r = os_atomic_cmpxchg(l, 0, _OSSpinLockLocked, acquire);
    if (likely(r)) return;
    goto _spin;
}
#else // !__ARM_ARCH_8_2__
void
_OSSpinLockLockSlow(volatile OSSpinLock *l)
{
    uint32_t tries = OS_LOCK_SPIN_SPIN_TRIES;
    OSSpinLock lock;
    os_atomic_rmw_loop(l, lock, _OSSpinLockLocked, acquire, if (unlikely(lock)){
        if (unlikely(lock != _OSSpinLockLocked)) {
            os_atomic_rmw_loop_give_up(return
                    _os_lock_corruption_abort((void *)l, (uintptr_t)lock));
        }
        if (unlikely(!tries--)) {
            os_atomic_rmw_loop_give_up(return _OSSpinLockLockYield(l));
        }
        OS_LOCK_SPIN_PAUSE();
        continue;
    });
}
#endif // !__ARM_ARCH_8_2__
#else // !OS_ATOMIC_UP
void
_OSSpinLockLockSlow(volatile OSSpinLock *l)
{
    uint32_t tries = OS_LOCK_SPIN_SPIN_TRIES;
    OSSpinLock lock;
    while (unlikely(lock = *l)) {
_spin:
        if (unlikely(lock != _OSSpinLockLocked)) {
            return _os_lock_corruption_abort((void *)l, (uintptr_t)lock);
        }
        if (unlikely(!tries--)) return _OSSpinLockLockYield(l);
        OS_LOCK_SPIN_PAUSE();
    }
    bool r = os_atomic_cmpxchgv(l, 0, _OSSpinLockLocked, &lock, acquire);
    if (likely(r)) return;
    goto _spin;
}
#endif // !OS_ATOMIC_UP

可以看到内部有自转逻辑，这里直接分析_OSSpinLockLockYield

5.1.1.2 _OSSpinLockLockYield

static void
_OSSpinLockLockYield(volatile OSSpinLock *l)
{
    int option = SWITCH_OPTION_DEPRESS;
    mach_msg_timeout_t timeout = 1;
    uint64_t deadline = _os_lock_yield_deadline(timeout);
    OSSpinLock lock;
    while (unlikely(lock = *l)) {
_yield:
        if (unlikely(lock != _OSSpinLockLocked)) {
            _os_lock_corruption_abort((void *)l, (uintptr_t)lock);
        }
        thread_switch(MACH_PORT_NULL, option, timeout);
        if (option == SWITCH_OPTION_WAIT) {
            timeout++;
        } else if (!_os_lock_yield_until(deadline)) {
            option = SWITCH_OPTION_WAIT;
        }
    }
    bool r = os_atomic_cmpxchgv(l, 0, _OSSpinLockLocked, &lock, acquire);
    if (likely(r)) return;
    goto _yield;
}

内部有超时时间以及线程切换逻辑。

5.1.2 OSSpinLockUnlock

void
OSSpinLockUnlock(volatile OSSpinLock *l)
{
    OS_ATOMIC_ALIAS(spin_unlock, OSSpinLockUnlock);
    OS_ATOMIC_ALIAS(_spin_unlock, OSSpinLockUnlock);
    return _os_nospin_lock_unlock((_os_nospin_lock_t)l);
}

内部调用了_os_nospin_lock_unlock。

5.1.2.1 _os_nospin_lock_unlock

void
_os_nospin_lock_unlock(_os_nospin_lock_t l)
{
    os_lock_owner_t self = _os_lock_owner_get_self();
    os_ulock_value_t current;
    current = os_atomic_xchg(&l->oul_value, OS_LOCK_NO_OWNER, release);
    if (likely(current == self)) return;
    return _os_nospin_lock_unlock_slow(l, current);
}

_os_nospin_lock_unlock_slow：

static void
_os_nospin_lock_unlock_slow(_os_nospin_lock_t l, os_ulock_value_t current)
{
    os_lock_owner_t self = _os_lock_owner_get_self();
    if (unlikely(OS_ULOCK_OWNER(current) != self)) {
        return; // no unowned_abort for drop-in compatibility with OSSpinLock
    }
    if (current & OS_ULOCK_NOWAITERS_BIT) {
        __LIBPLATFORM_INTERNAL_CRASH__(current, "unlock_slow with no waiters");
    }
    for (;;) {
        int ret = __ulock_wake(UL_COMPARE_AND_WAIT | ULF_NO_ERRNO, l, 0);
        if (unlikely(ret < 0)) {
            switch (-ret) {
            case EINTR:
                continue;
            case ENOENT:
                break;
            default:
                __LIBPLATFORM_INTERNAL_CRASH__(-ret, "ulock_wake failure");
            }
        }
        break;
    }
}

5.2 os_unfair_lock 源码分析

typedef struct os_unfair_lock_s {
    uint32_t _os_unfair_lock_opaque;
} os_unfair_lock, *os_unfair_lock_t;

初始化OS_UNFAIR_LOCK_INIT直接设置了默认值0：

#define OS_UNFAIR_LOCK_INIT ((os_unfair_lock){0})

5.2.1 os_unfair_lock_lock

void
os_unfair_lock_lock(os_unfair_lock_t lock)
{
    _os_unfair_lock_t l = (_os_unfair_lock_t)lock;
    os_lock_owner_t self = _os_lock_owner_get_self();
    bool r = os_atomic_cmpxchg(&l->oul_value, OS_LOCK_NO_OWNER, self, acquire);
    if (likely(r)) return;
    return _os_unfair_lock_lock_slow(l, OS_UNFAIR_LOCK_NONE, self);
}

_os_lock_owner_get_self：

OS_ALWAYS_INLINE OS_CONST
static inline os_lock_owner_t
_os_lock_owner_get_self(void)
{
    os_lock_owner_t self;
    self = (os_lock_owner_t)_os_tsd_get_direct(__TSD_MACH_THREAD_SELF);
    return self;
}

_os_unfair_lock_lock_slow：

static void
_os_unfair_lock_lock_slow(_os_unfair_lock_t l,
        os_unfair_lock_options_t options, os_lock_owner_t self)
{
    os_unfair_lock_options_t allow_anonymous_owner =
            options & OS_UNFAIR_LOCK_ALLOW_ANONYMOUS_OWNER;
    options &= ~OS_UNFAIR_LOCK_ALLOW_ANONYMOUS_OWNER;
    if (unlikely(options & ~OS_UNFAIR_LOCK_OPTIONS_MASK)) {
        __LIBPLATFORM_CLIENT_CRASH__(options, "Invalid options");
    }
    os_ulock_value_t current, new, waiters_mask = 0;
    while (unlikely((current = os_atomic_load(&l->oul_value, relaxed)) !=
            OS_LOCK_NO_OWNER)) {
_retry:
        if (unlikely(OS_ULOCK_IS_OWNER(current, self, allow_anonymous_owner))) {
            return _os_unfair_lock_recursive_abort(self);
        }
        new = current & ~OS_ULOCK_NOWAITERS_BIT;
        if (current != new) {
            // Clear nowaiters bit in lock value before waiting
            if (!os_atomic_cmpxchgv(&l->oul_value, current, new, &current,
                    relaxed)){
                continue;
            }
            current = new;
        }
        int ret = __ulock_wait(UL_UNFAIR_LOCK | ULF_NO_ERRNO | options,
                l, current, 0);
        if (unlikely(ret < 0)) {
            switch (-ret) {
            case EINTR:
            case EFAULT:
                continue;
            case EOWNERDEAD:
                _os_unfair_lock_corruption_abort(current);
                break;
            default:
                __LIBPLATFORM_INTERNAL_CRASH__(-ret, "ulock_wait failure");
            }
        }
        if (ret > 0) {
            // If there are more waiters, unset nowaiters bit when acquiring lock
            waiters_mask = OS_ULOCK_NOWAITERS_BIT;
        }
    }
    new = self & ~waiters_mask;
    bool r = os_atomic_cmpxchgv(&l->oul_value, OS_LOCK_NO_OWNER, new,
            &current, acquire);
    if (unlikely(!r)) goto _retry;
}

内部是wait等待逻辑。

5.2.2 os_unfair_lock_unlock

void
os_unfair_lock_unlock(os_unfair_lock_t lock)
{
    _os_unfair_lock_t l = (_os_unfair_lock_t)lock;
    os_lock_owner_t self = _os_lock_owner_get_self();
    os_ulock_value_t current;
    current = os_atomic_xchg(&l->oul_value, OS_LOCK_NO_OWNER, release);
    if (likely(current == self)) return;
    return _os_unfair_lock_unlock_slow(l, self, current, 0);
}

内部调用了_os_unfair_lock_unlock_slow：

OS_NOINLINE
static void
_os_unfair_lock_unlock_slow(_os_unfair_lock_t l, os_lock_owner_t self,
        os_ulock_value_t current, os_unfair_lock_options_t options)
{
    os_unfair_lock_options_t allow_anonymous_owner =
            options & OS_UNFAIR_LOCK_ALLOW_ANONYMOUS_OWNER;
    options &= ~OS_UNFAIR_LOCK_ALLOW_ANONYMOUS_OWNER;
    if (unlikely(OS_ULOCK_IS_NOT_OWNER(current, self, allow_anonymous_owner))) {
        return _os_unfair_lock_unowned_abort(OS_ULOCK_OWNER(current));
    }
    if (current & OS_ULOCK_NOWAITERS_BIT) {
        __LIBPLATFORM_INTERNAL_CRASH__(current, "unlock_slow with no waiters");
    }
    for (;;) {
        int ret = __ulock_wake(UL_UNFAIR_LOCK | ULF_NO_ERRNO, l, 0);
        if (unlikely(ret < 0)) {
            switch (-ret) {
            case EINTR:
                continue;
            case ENOENT:
                break;
            default:
                __LIBPLATFORM_INTERNAL_CRASH__(-ret, "ulock_wake failure");
            }
        }
        break;
    }
}

可以看到内部是有唤醒逻辑的。

六、读写锁

读操作可以共享，写操作是排他的，读可以有多个在读，写只有唯一个在写，同时写的时候不允许读。要实现读写锁核心逻辑是：

多读单写
写写互斥
读写互斥
写不能阻塞任务执行

有两套方案：

1.使用栅栏函数相关API。
2.使用pthread_rwlock_t相关API。

6.1 dispatch_barrier_async 实现多读单写

写：通过栅栏函数可以实现写写互斥以及读写互斥，写使用async可以保证写逻辑不阻塞当前任务执行。
读：使用dispatch_sync同步效果实现多读（放入并发队列中）。

首先定义一个并发队列以及字典存储数据：

@property (nonatomic, strong) dispatch_queue_t concurrent_queue;
@property (nonatomic, strong) NSMutableDictionary *dataDic;

//初始化
self.concurrent_queue = dispatch_queue_create("rw_queue", DISPATCH_QUEUE_CONCURRENT);
self.dataDic = [NSMutableDictionary dictionary];

写入操作：

- (void)safeSetter:(NSString *)name time:(int)time {
    dispatch_barrier_async(self.concurrent_queue, ^{
        sleep(time);
        [self.dataDic setValue:name forKey:@"HotpotCat"];
        NSLog(@"write name:%@,currentThread:%@",name,[NSThread currentThread]);
    });
}

为了方便测试key值写死，并且传入一个time。barrier保证了写之间互斥以及读写互斥。

读取操作：

- (NSString *)safeGetterWithTime:(int)time {
    __block NSString *result;
    //多条线程同时读，阻塞的是当前线程，多条线程访问就是多读了。同步使用concurrent_queue是为了配合栅栏函数读写互斥。
    dispatch_sync(self.concurrent_queue, ^{
        sleep(time);
        result = self.dataDic[@"HotpotCat"];
    });
    NSLog(@"result:%@,currentThread:%@,time:%@",result,[NSThread currentThread],@(time));
    return result;
}

使用同步函数配合栅栏函数（栅栏函数只能针对同一队列）实现读写互斥，当多条线程同时访问safeGetterWithTime时就实现了多读操作。

写入验证：

//调用
[self safeSetter:@"1" time:4];
[self safeSetter:@"2" time:1];
[self safeSetter:@"3" time:2];
[self safeSetter:@"4" time:1];

输出：

write name:1,currentThread:<NSThread: 0x281fea4c0>{number = 5, name = (null)}
write name:2,currentThread:<NSThread: 0x281fea4c0>{number = 5, name = (null)}
write name:3,currentThread:<NSThread: 0x281fea4c0>{number = 5, name = (null)}
write name:4,currentThread:<NSThread: 0x281fea4c0>{number = 5, name = (null)}

很明显写之间是互斥的，任务1没有执行完之前其它任务都在等待。

读取验证：

for (int i = 0; i < 5; i++) {
    dispatch_async(dispatch_get_global_queue(0, 0), ^{
        NSString *result = [self safeGetterWithTime:5 - i];
        NSLog(@"result:%@",result);
    });
}

输出：

result:4,currentThread:<NSThread: 0x281f80600>{number = 7, name = (null)},time:1
result:4,currentThread:<NSThread: 0x281fce540>{number = 8, name = (null)},time:2
result:4,currentThread:<NSThread: 0x281f80980>{number = 9, name = (null)},time:3
result:4,currentThread:<NSThread: 0x281feb540>{number = 10, name = (null)},time:4
result:4,currentThread:<NSThread: 0x281f80a80>{number = 11, name = (null)},time:5

任务并行执行，顺序是由于设置了sleep时间，如果去掉时间或者时间一致，每次执行结果都不同了。

6.2 pthread_rwlock_t 实现多读单写

定义锁以及字典数据：

{
    pthread_rwlock_t rw_lock;
    pthread_rwlockattr_t rw_lock_attr;
}
@property (nonatomic, strong) NSMutableDictionary *dataDic;

pthread_rwlockattr_t读写属性有两种：lockkind与pshared。
lockkind：读写策略，包括读取优先（默认属性）、写入优先。苹果系统里面没有提供 pthread_rwlockattr_getkind_np 与 pthread_rwlockattr_setkind_np 相关函数。
pshared：PTHREAD_PROCESS_PRIVATE（进程内竞争读写锁，默认属性）PTHREAD_PROCESS_SHARED(进程间竞争读写锁)

初始化:

self.dataDic = [NSMutableDictionary dictionary];
//初始化
pthread_rwlockattr_init(&rw_lock_attr);
pthread_rwlock_init(&rw_lock, &rw_lock_attr);
//进程内
pthread_rwlockattr_setpshared(&rw_lock_attr, PTHREAD_PROCESS_PRIVATE);

pthread_rwlockattr_t读写设置为进程内。

写入操作如下：

- (void)safeSetter:(NSString *)name {
    //写锁
    pthread_rwlock_wrlock(&rw_lock);
    [self.dataDic setValue:name forKey:@"HotpotCat"];
    NSLog(@"write name:%@,currentThread:%@",name,[NSThread currentThread]);
    //释放
    pthread_rwlock_unlock(&rw_lock);
}

读取操作如下：

- (NSString *)safeGetter {
    //读锁
    pthread_rwlock_rdlock(&rw_lock);
    NSString *result = self.dataDic[@"HotpotCat"];
    //释放
    pthread_rwlock_unlock(&rw_lock);
    NSLog(@"result:%@,currentThread:%@",result,[NSThread currentThread]);
    return result;
}

写入验证：

dispatch_async(dispatch_get_global_queue(0, 0), ^{
    [self safeSetter:@"1"];
});
dispatch_async(dispatch_get_global_queue(0, 0), ^{
    [self safeSetter:@"2"];
});
dispatch_async(dispatch_get_global_queue(0, 0), ^{
    [self safeSetter:@"3"];
});
dispatch_async(dispatch_get_global_queue(0, 0), ^{
    [self safeSetter:@"4"];
});

输出：

LockDemo[52251:5873172] write name:4,currentThread:<NSThread: 0x60000072e980>{number = 4, name = (null)}
LockDemo[52251:5873177] write name:1,currentThread:<NSThread: 0x60000075d100>{number = 6, name = (null)}
LockDemo[52251:5873170] write name:2,currentThread:<NSThread: 0x60000072f600>{number = 7, name = (null)}
LockDemo[52251:5873178] write name:3,currentThread:<NSThread: 0x60000073d480>{number = 5, name = (null)}

这里就与队列调度有关了，顺序不定，如果不加锁大量并发调用下则会crash。

读取验证：

for (int i = 0; i < 5; i++) {
    dispatch_async(dispatch_get_global_queue(0, 0), ^{
        NSString *result = [self safeGetter];
    });
}

输出:

result:4,currentThread:<NSThread: 0x600001cdc200>{number = 5, name = (null)}
result:4,currentThread:<NSThread: 0x600001cd1080>{number = 7, name = (null)}
result:4,currentThread:<NSThread: 0x600001c95f40>{number = 6, name = (null)}
result:4,currentThread:<NSThread: 0x600001c91ec0>{number = 3, name = (null)}
result:4,currentThread:<NSThread: 0x600001c94d80>{number = 4, name = (null)}

输出顺序也不一定。当然混合读写测试也可以，用数组更容易测试。

参考链接:pthread_rwlock

对于读数据比修改数据频繁的应用，用读写锁代替互斥锁可以提高效率。因为使用互斥锁时，即使是读出数据（相当于操作临界区资源）都要上互斥锁，而采用读写锁，则可以在任一时刻允许多个读出者存在，提高了更高的并发度，同时在某个写入者修改数据期间保护该数据，以免任何其它读出者或写入者的干扰。
获取一个读写锁用于读称为共享锁，获取一个读写锁用于写称为独占锁，因此对于某个给定资源的共享访问也称为共享-独占上锁。

锁对应关系

锁的原理（二）：自旋锁、互斥锁以及读写锁

一、锁的分类

1.1 自旋锁

1.2 互斥锁

1.2.1 读写锁

二、NSLock & NSRecursiveLock 的应用以及原理

2.1 案例一

2.2 案例二

2.3 原理分析

2.3.1 NSLock 源码分析

2.3.2 NSRecursiveLock 源码分析

三、NSCondition 原理

3.1 生产者-消费者 案例

3.2 源码分析

四、NSConditionLock 使用和原理

4.1案例

4.2 断点调试分析逻辑

4.3 源码分析

4.4 反汇编分析

五、OSSpinLock & os_unfair_lock

5.1 OSSpinLock 源码分析

5.1.1 OSSpinLockLock

5.1.1.1 _OSSpinLockLockSlow

5.1.1.2 _OSSpinLockLockYield

5.1.2 OSSpinLockUnlock

5.1.2.1 _os_nospin_lock_unlock

5.2 os_unfair_lock 源码分析

5.2.1 os_unfair_lock_lock

5.2.2 os_unfair_lock_unlock

六、读写锁

6.1 dispatch_barrier_async 实现多读单写

6.2 pthread_rwlock_t 实现多读单写

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

3.1 生产者-消费者案例