深入剖析信号量

2019-08-06

遇到问题

国庆前我们公司的自研的广告SDK项目完成，新版b612相机已经包含该广告SDK，b612相机是一款工具类的app，广告是我们的主要收入来源，但同时用户体验的要求又非常的高。对于我们研发广告SDK带来不小的挑战，比如启动必须在1.5s完成，也就是使开屏广告必须在1.5秒就完成获取和展示（广告业界的一半建议标准是3s），别看这1.5s的优化，为这1.5秒，我们团队绞尽脑汁，深挖每行代码用的技术点，然后逐一进行优化。比如http的请求优化、代码执行耗时的优化、流量控制缓存选择的优化等等。后面有机会再去更多介绍我们采用的一些技术手段，今天只说说iOS中的信号量。

在开发广告SDK的时候，发现b612的还有品牌广告在韩国团队维护，暂时不能移交给我们维护，因此广告SDK需要等待品牌广告请求失败了才能去展示广告SDK的广告，为了将开屏广告请求时间控制在1.5s这个底线，因此广告SDK必须将请求和展示拆分：

1）在启动的时候请求广告
2）在品牌广告失败后去决定是否展示广告
3）广告SDK是有流量分配规则的，有直投的广告，也有接入优量汇、inmobi等第三方的广告联盟，之间也许要做流量控制。

SDK为了达到这个目的，使用到了信号量dispatch_semaphore。信号量本来不难，但是我们研发工程师在研发中遇到过信号量控制不生效的情况，执行到下一个流量分配去了，导致展示两次开屏广告。我在协助排查的时候觉得很诡异，陷入自我怀疑：会不会是iOS的bug，在多线程中使用信号量会不会不生效。当然最后排查的结果是dispatch_semaphore_wait使用不当导致，它有两个入参数，一个参数是信号量对象，另一个是等待时间。

1	dispatch_semaphore_wait(dispatch_semaphore_t _Nonnull dsema, dispatch_time_t timeout)

timeout的值可以有三个，DISPATCH_TIME_NOW（不等待）、DISPATCH_TIME_FOREVER（一直等待，直到信号出现并大于等于0）、自定义的dispatch_time_t。我们SDK为了无论如何在1.5s能释放线程资源（dispatch_semaphore_wait 是会创建线程池的），使用了自定义的信号，最大等待时间等于超时时间（1.5s），而我们自定义的dispatch_time_t使用了ms作为单位，而传入的时间是1.5,导致dispatch_semaphore_wait几乎不等待。

解决这个问题费了点时间，因为没想到会在这个地方犯错，而排查耗时较长的原因，是因为自我怀疑，以为是系统的bug。还是自己学的不深，所以会自我怀疑，当时解决完问题后就想找个时间再次深入去学习GCD（虽然我自以为深入学习过好几次😂），今天得空看了看信号量的源码GCD源码。

深入剖析信号量

一、信号量的定义和作用

信号量：就是一种可用来控制访问资源的数量的标识，设定了一个信号量，在线程访问之前，加上信号量的处理，则可告知系统按照我们指定的信号量数量来执行多个线程。其实，这有点类似锁机制了，只不过信号量都是系统帮助我们处理了，我们只需要在执行线程之前，设定一个信号量值，并且在使用时，加上信号量处理方法就行了。

也就是说系统提供信号量的方法，来帮我们开发者简化控制任务执行顺序。

二、苹果关于信号量的API不多，主要就3个。

dispatch_semaphore_create创建信号量对象dispatch_semaphore_t
dispatch_semaphore_wait等待一个semaphore，或者是减少semaphore的计数。每次会执行会将计数器-1.如果减完之后计数器小于0的话，会阻塞在当前线程直到接收到信号。
dispatch_semaphore_signal发送信号，或者增加一个信号量的计数。如果增加之前的信号值dsema_value大于等于0，说明没有线程因为该信号而阻塞，所以直接返回。

`三、dispatch_semaphore_create`

信号量在初始化（调用dispatch_semaphore_create初始化）时要指定 value，随后内部将这个 value 存储起来。实际操作时会存两个 value，一个是当前的 value，一个是记录初始 value。我们来看看源码：


dispatch_semaphore_t dispatch_semaphore_create(long value){
    dispatch_semaphore_t dsema;

    if (value < 0) {
        return NULL;
    }
    // 申请内存
    dsema = calloc(1, sizeof(struct dispatch_semaphore_s));
    
    if (fastpath(dsema)) {
        // do_vtable里面主要包含了这个 dispatch_semaphore_s 的操作函数
        dsema->do_vtable = &_dispatch_semaphore_vtable;

        //可以理解为链表的结尾标记
        dsema->do_next = DISPATCH_OBJECT_LISTLESS;

        // 引用计数
        dsema->do_ref_cnt = 1;
        dsema->do_xref_cnt = 1;

        // 目标队列
        dsema->do_targetq = dispatch_get_global_queue(0, 0);

        // 信号值
        dsema->dsema_value = value;
        dsema->dsema_orig = value;
    }
    
    return dsema;
    
 }

首先value必须大于等于0，流程才能继续进行；
申请一块dispatch_semaphore_s内存，并初始化为0；
将dsema的do_vtable指向_dispatch_semaphore_vtable，_dispatch_semaphore_vtable的定义如下：

const struct dispatch_semaphore_vtable_s _dispatch_semaphore_vtable = {
    .do_type = DISPATCH_SEMAPHORE_TYPE,
    .do_kind = "semaphore",
    .do_dispose = _dispatch_semaphore_dispose,
    .do_debug = _dispatch_semaphore_debug,
};

_dispatch_semaphore_dispose，顾名思义，销毁一个信号量，定义如下：

void _dispatch_semaphore_dispose(dispatch_semaphore_t dsema){
    kern_return_t kr;
    
    if (dsema->dsema_value < dsema->dsema_orig) {
        DISPATCH_CLIENT_CRASH("Semaphore/group object deallocated while in use");
    }
    
    if (dsema->dsema_port) {
        kr = semaphore_destroy(mach_task_self(), dsema->dsema_port);
        DISPATCH_SEMAPHORE_VERIFY_KR(kr);
    }
    if (dsema->dsema_waiter_port) {
        kr = semaphore_destroy(mach_task_self(), dsema->dsema_waiter_port);
        DISPATCH_SEMAPHORE_VERIFY_KR(kr);
    }
    
    _dispatch_dispose(dsema);
}

注意DISPATCH_CLIENT_CRASH这段，如果初始值dsema_orig和销毁时最终的值dsema_value不一致，会crash。这就是苹果官有这么一段警告的原因：

Important

Calls to dispatch_semaphore_signal must be balanced with calls to wait(). Attempting to dispose of a semaphore with a count lower than value causes an EXC_BAD_INSTRUCTION exception.

https://developer.apple.com/documentation/dispatch/1452955-dispatch_semaphore_create?language=objc

所以 dispatch_semaphore_signal和dispatch_semaphore_wait 请配对使用，确保在心好凉释放的时候不会crash。

`四、dispatch_semaphore_wait`

先看源码：


DISPATCH_NOINLINE long dispatch_semaphore_signal(dispatch_semaphore_t dsema){
    if (dispatch_atomic_inc(&dsema->dsema_value) > 0) {
        return 0;
    }
    return _dispatch_semaphore_signal_slow(dsema);
}

第一行的 dispatch_atomic_dec2o 是一个宏，会调用 GCC 内置的函数 __sync_sub_and_fetch，实现减法的原子性操作。因此这一行的意思是将 dsema 的值减一，并把新的值赋给 value。

如果减一后的 value 大于等于 0 就立刻返回，没有任何操作，否则进入等待状态。

_dispatch_semaphore_wait_slow 函数针对不同的 timeout 参数，分了三种情况考虑:

case DISPATCH_TIME_NOW:  
    while ((orig = dsema->dsema_value) < 0) {
        if (dispatch_atomic_cmpxchg2o(dsema, dsema_value, orig, orig + 1)) {
            return KERN_OPERATION_TIMED_OUT;
        }
    }

这种情况下会立刻判断 dsema->dsema_value 与 orig 是否相等。如果 while 判断成立，内部的 if 判断一定也成立，此时会将 value 加一(也就是变为 0) 并返回。加一的原因是为了抵消 wait 函数一开始的减一操作。此时函数调用方会得到返回值 KERN_OPERATION_TIMED_OUT，表示由于等待时间超时而返回。

实际上 while 判断一定会成立，因为如果 value 大于等于 0，在上一个函数 dispatch_semaphore_wait 中就已经返回了。

第二种情况是 DISPATCH_TIME_FOREVER 这个 case:


case DISPATCH_TIME_FOREVER:  
    do {
        kr = semaphore_wait(dsema->dsema_port);
    } while (kr == KERN_ABORTED);
    break;

进入 do-while 循环后会调用系统的 semaphore_wait 方法，KERN_ABORTED 表示调用者被一个与信号量系统无关的原因唤醒。因此一旦发生这种情况，还是要继续等待，直到收到 signal 调用。

在其他情况下(default 分支)，我们指定一个超时时间，这和 DISPATCH_TIME_FOREVER 的处理比较类似，不同的是我们调用了内核提供的 semaphore_timedwait 方法可以指定超时时间。

整个函数的框架如下:


static long _dispatch_semaphore_wait_slow(dispatch_semaphore_t dsema, dispatch_time_t timeout) {  
again:  
    while ((orig = dsema->dsema_sent_ksignals)) {
        if (dispatch_atomic_cmpxchg2o(dsema, dsema_sent_ksignals, orig,
                orig - 1)) {
            return 0;
        }
    }
    switch (timeout) {
        default:  /* semaphore_timedwait */
        case DISPATCH_TIME_NOW: /* KERN_OPERATION_TIMED_OUT */
        case DISPATCH_TIME_FOREVER: /* semaphore_wait */
    }
    goto again;
}

可见信号量被唤醒后，会回到最开始的地方，进入 while 循环。这个判断条件一般都会成立，极端情况下由于内核存在 bug，导致 orig 和 dsema_sent_ksignals 不相等，也就是收到虚假 signal 信号时会忽略。

进入 while 循环后，if 判断一定成立，因此返回 0，正如文档所说，返回 0 表示成功，否则表示超时。

`五、dispatch_semaphore_signal`

发送信号量，或者增加一个信号量的计数。如果增加之前的信号值dsema_value大于等于0，说明没有线程因为该信号而阻塞，所以直接返回。简化版源码如下:


long dispatch_semaphore_signal(dispatch_semaphore_t dsema) {  
    long value = dispatch_atomic_inc2o(dsema, dsema_value);
    if (fastpath(value > 0)) {
        return 0;
    }
    return _dispatch_semaphore_signal_slow(dsema);
}

首先会调用原子方法让 value 加1，如果大于零就立刻返回 0，否则返回 _dispatch_semaphore_signal_slow，将会唤起一个在spatch_semaphore_wait中等待的线程。（如果有多个线程在等待，iOS会根据线程的优先级来判断具体唤醒哪个线程）:


DISPATCH_NOINLINE static long _dispatch_semaphore_signal_slow(dispatch_semaphore_t dsema){
    kern_return_t kr;
    
    _dispatch_semaphore_create_port(&dsema->dsema_port);

    // Before dsema_sent_ksignals is incremented we can rely on the reference
    // held by the waiter. However, once this value is incremented the waiter
    // may return between the atomic increment and the semaphore_signal(),
    // therefore an explicit reference must be held in order to safely access
    // dsema after the atomic increment.
    _dispatch_retain(dsema);
    
    dispatch_atomic_inc(&dsema->dsema_sent_ksignals);
    
    // 利用系统的信号量库实现发送信号量的功能
    kr = semaphore_signal(dsema->dsema_port);
    DISPATCH_SEMAPHORE_VERIFY_KR(kr);

    _dispatch_release(dsema);
    
    return 1;
}

使用场景

iOS中的信号量其实蛮简单的，信号量设计的目的是阻塞然后等待执行。因此在我们开发中可以运用在很多场景中，随便列举几个：

限制线程的最大并发数
阻塞发请求的线程
多个请求结束后统一操作
多个请求顺序执行

举例：

dispatch_semaphore_t sema = dispatch_semaphore_create(M);
for (NSInteger i = 0;i<N;i++） {
     dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{
        dispatch_semaphore_wait(sema, DISPATCH_TIME_FOREVER); 
        // doing
        dispatch_semaphore_signal(sema);
    });
}

如上述代码可知，总共异步执行N个任务，但是由于我们设置了值为M的信号量，每一次执行任务的时候都会导致信号量的减1，而在任务结束后使信号量加1，当信号量减到0的时候，说明正在执行的任务有M个，这个时候其它任务就会阻塞，直到有任务被完成时，这些任务才会执行。

总结

信号量虽然很简单，但用好了能帮我们简化控制任务控制逻辑。但苹果封装它的背后逻辑，如果我们清楚了，自然就更加了如指掌，也更有信心和能力运用好信号量。