GCD - dispatch_semaphore详解

作者: Sevenuncle | 来源:发表于2017-09-10 00:50 被阅读0次

前言

文章内容包含很多作者个人的理解，如果不对，非常希望能及时指出来，防止误导大众，期待一起进步。

在使用semaphore的时候突然有一些疑问：

信号量函数wait是怎么实现等待的？通过轮询？通知？基于端口？
信号量的这个变量是全局的吗？要不然不同线程之间怎么互相等待、通知？

带着这些疑问，让我们通过源码（https://opensource.apple.com/source/libdispatch/libdispatch-84.5/src/）去一探究竟。

基本使用

dispatch_semaphore信号量通常用于解决生产者消费者问题。

dispatch_semaphore就三个函数，创建信号量、等待信号、发送信号：

创建信号量

dispatch_semaphore_t dispatch_semaphore_create(long)

a.创建一个信号量，参数为一个long型，使用的场景如果是多个线程竞争一个资源或者线程需要等待一个事件完成，则传入参数为0；比如一个下载图片线程，另一个等待图片下载完成的线程，此时可以创建一个为0的信号量，当下载图片完成，则将信号量加1，则等待的线程得到这个信号通知。

b.如果使用场景是管理有限数量的资源，则传入大于0的参数；类似的使用场景如可用于有缓冲界限的生产者消费者问题，当资源池里的数量达到上限，则不再生产资源（需要加另外一些处理逻辑），直到等到消费者取走了资源；

c.传入小于0的参数返回NULL。

等待信号

long dispatch_semaphore_wait(dispatch_semaphore_t, dispatch_time_t)

当收到信号量通知的时候，则会将信号量-1并返回0；否则返回非0值。

dispatch_semaphore_t 等待的信号量对象

dispatch_time_t 等待的时间，主要分为用户自定义时间长度、DISPATCH_TIME_NOW、DISPATCH_TIME_FOREVER三种，在制定时间内等待信号量，如果等到则返回0表示非超时的响应，非0表示等待超时返回；

发送信号

long dispatch_semaphore_signal(dispatch_semaphore_t)

将信号量对应的值加1，用于唤醒等待该信号量的一个线程，如果有线程被唤醒，则返回非0；否则返回0。

数据结构

信号量其实比较简单，就是围绕dispatch_semaphore_t的加减。iOS信号量dispatch_semaphore其实是基于mach微内核中semaphore_t封装实现的。

首先看一下创建信号量生成的dispatch_semaphore_t变量，其实它是指向结构体dispatch_semaphore_s的指针，而dispatch_semaphore_s的结构如下：

libDispatch/semaphore_internal.h

struct dispatch_semaphore_s {
    DISPATCH_STRUCT_HEADER(dispatch_semaphore_s, dispatch_semaphore_vtable_s); //定义了任意一个dispatch_xx_t结构体的基本数据（存储表、引用计数、目标队列、上下文等）
    long dsema_value;//信号量值（重点）
    long dsema_orig;//信号量初始值
    size_t dsema_sent_ksignals; //用于表示真是信号发送（重点）
    semaphore_t dsema_port; //信号
    semaphore_t dsema_waiter_port; //等待信号
    size_t dsema_group_waiters; //类似于上面，和group有关？
    struct dispatch_sema_notify_s *dsema_notify_head; //信号通知队列头
    struct dispatch_sema_notify_s *dsema_notify_tail;
};

DISPATCH_STRUCT_HEADER //是一个宏，定义了一些结构体引用计数、上下文、队列、以及一些内部处理函数（暂时没详细了解）

long desma_value //信号量的当前值

long desma_orig // 信号量的初始值，作用不详

size_t dsema_sent_ksignals; //配合desma_value工作，表示实际发送的信号数量，实际的用于唤醒等待dispatch_semaphore_wait()函数的等待（对，不是desma_value，而是它）

semaphore_t dsema_port //信号

semaphore_t dsema_waiter_port //等待信号

semaphore_t其实是mach提供的数据结构，实现如下：

//semaphore.h
struct __semaphore
  {
    __pthread_spinlock_t __lock;//自旋锁，即忙等待锁
    struct __pthread *__queue;
    int __pshared; //不为0时可以在进程间共享，为0时可以在进程内的所有线程共享
    int __value; //信号量的值
    void *__data;
  };

size_t dsema_group_waiters; //类似于上面dsema_sent_ksignals，和group有关？

struct dispatch_sema_notify_s *dsema_notify_head; //信号通知队列头

struct dispatch_sema_notify_s *dsema_notify_tail;

通知队列的结构如下：

struct dispatch_sema_notify_s {
    struct dispatch_sema_notify_s *dsn_next;
    dispatch_queue_t dsn_queue;
    void *dsn_ctxt;
    void (*dsn_func)(void *);
};

这个是一个用于链表结构的信号量通知队列，该结构体主要是封装了等待该信号量的回调块的一些基本信息：

dispatch_queue_t dsn_queue //要执行目标队列

void *dsn_ctxt //上下文信息

void (*dsn_func)(void *) //回调函数

struct dispatch_sema_notify_s *dsn_next //指向下一个等待信号量通知的节点

过程分析

好了，对于dispatch_semaphore_t变量的内部结构有了大致的了解，接下来看三个主要函数内部是怎么操作该变量实现信号量的功能的。

从创建信号量函数开始:

// libDispatch/semaphore.c
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)) {
        //DISPATCH_STRUCT_HEADER里面的内容初始化
        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;
}

如果传入参数小于0，则直接返回NULL；否则进行申请内存、基本结构体头初始化、信号量值初始化，我们需要的关注点不多，值得注意的一点是初始化的时候并没有初始化semaphore_t dsema_port, 这是因为该变量是赖加载的（后续会讲到），在等待或者发送信号里创建。

创建完信号量，然后开始等待这个信号量发消息

long
dispatch_semaphore_wait(dispatch_semaphore_t dsema, dispatch_time_t timeout)
{
    if (dispatch_atomic_dec(&dsema->dsema_value) >= 0) {
        return 0;
    }
    return _dispatch_semaphore_wait_slow(dsema, timeout);
}

dispatch_atomic_dec（）的作用即将参数以原子性的减一并作为返回值，如果信号量已经大于1则直接返回0，表示等到信号量的信息了；否则进入_dispatch_semaphore_wait_slow（）；可以看出，如果有线程正在等待信号量，dsema->dsema_value的值为负值；

static long
_dispatch_semaphore_wait_slow(dispatch_semaphore_t dsema, dispatch_time_t timeout)
{
    mach_timespec_t _timeout;
    kern_return_t kr;
    uint64_t nsec;
    long orig;
    
again:
    while ((orig = dsema->dsema_sent_ksignals)) {
        if (dispatch_atomic_cmpxchg(&dsema->dsema_sent_ksignals, orig, orig - 1)) {
            return 0;
        }
    }

    _dispatch_semaphore_create_port(&dsema->dsema_port);

    switch (timeout) {
    default:
        do {
            nsec = _dispatch_timeout(timeout);
            //秒
            _timeout.tv_sec = (typeof(_timeout.tv_sec))(nsec / NSEC_PER_SEC);
            //纳秒
            _timeout.tv_nsec = (typeof(_timeout.tv_nsec))(nsec % NSEC_PER_SEC);
            //利用mach的信号量函数等待
            kr = slowpath(semaphore_timedwait(dsema->dsema_port, _timeout));
        } while (kr == KERN_ABORTED); //返回值为系统信号器处理中断

        if (kr != KERN_OPERATION_TIMED_OUT) {
            DISPATCH_SEMAPHORE_VERIFY_KR(kr);
            break;
        }
        // Fall through and try to undo what the fast path did to dsema->dsema_value
    case DISPATCH_TIME_NOW:
        while ((orig = dsema->dsema_value) < 0) { //
            if (dispatch_atomic_cmpxchg(&dsema->dsema_value, orig, orig + 1)) {
                return KERN_OPERATION_TIMED_OUT;
            }
        }
        // Another thread called semaphore_signal().
        // Fall through and drain the wakeup.
    case DISPATCH_TIME_FOREVER:
        do {
            kr = semaphore_wait(dsema->dsema_port);
        } while (kr == KERN_ABORTED);
        DISPATCH_SEMAPHORE_VERIFY_KR(kr);
        break;
    }

    goto again;
}

首先是一个While循环，注意看while的条件是一个赋值语句:

while ((orig = dsema->dsema_sent_ksignals))

所以只要dsema_sent_ksignals非0，总能进入循环；而下面的if判断：

if (dispatch_atomic_cmpxchg(&dsema->dsema_sent_ksignals, orig, orig - 1))

dispatch_atomic_cmpxchg()展开后为：

#define dispatch_atomic_cmpxchg2o(p, f, e, n) \
        dispatch_atomic_cmpxchg(&(p)->f, (e), (n))
#define dispatch_atomic_cmpxchg(p, e, n) \
        __sync_bool_compare_and_swap((p), (e), (n))

看最后一个函数__sync_bool_compare_and_swap，基本意思就是：如果(p)和（e）相等，则将(n)赋值给(p)，并且函数返回true；否则返回flase;

非常明显，在dsema_sent_ksignals非0的情况下，while循环的赋值语句进入之后，dsema->dsema_sent_ksignals与orig的值每次都是相等的，所以if的结果每次都是true，因为函数直接返回0，表示等到信号；而信号量初始化的时候未给dsema->dsema_sent_ksignals赋值，默认为0，所以，如果没有信号量的实际通知（_dispatch_semaphore_signal_slow（）里有对该变量的+1操作，下文会讲到）或者遭受了系统异常通知，并不会解除等待；

看到这里，大家或许已经看得一脸懵逼了，作者表示夜看得一脸黑人问号？dsema_sent_ksignals 这东西貌似有抢帮夺权，干了dsema_value干的活，的确，源码中也用英文注释说明了这一点，dipatch_semaphore底层的实现机制和开放的文档有一点出入，但这实现逻辑实际上是可行有效的。

// From xnu/osfmk/kern/sync_sema.c:
// wait_semaphore->count = -1;  /* we don't keep an actual count */
//
// The code above does not match the documentation, and that fact is
// not surprising. The documented semantics are clumsy to use in any
// practical way. The above hack effectively tricks the rest of the
// Mach semaphore logic to behave like the libdispatch algorithm.

好吧，先继续讲完剩下部分，即使现在理不清楚夜没关系，看完了signal部分的代码，再回头看这一部分，就明白了。

_dispatch_semaphore_create_port(&dsema->dsema_port)，这函数就是懒加载初始化dsema_port，如果已经初始化则直接返回，否则会进行默认的对semaphore_t dsema_port变量初始化；

Switch(timeout) {
    default : //这里计算剩余时间，利用mach内核的信号量等待函数semaphore_timedwait()进行等待，参考了POSIX的sem_timedwait()函数，该函数是阻塞式等待，如果在指定时间内没有得到通知，则会一直阻塞住，监听dsema_port等待其通知；当超时时返回超时结果？怎么结束GOTO语句？（仔细看下源码，default分支最后居然没有break，也就是说semaphore_timedwait()超时了之后还是执行下面的NOW分支，哎，这这这，害我懵逼了半天超时等待过期时怎么退出该函数）
    case DISPATCH_TIME_NOW: //如果超时了，因为超时了（两种超时情况），所以撤销wait函数里一开始对dsema->dsema_value的减1操作，然后直接return 说明等待超时了，结束函数；如果没有超时，则什么都不执行，执行GOTO语句，返回；
    case DISPATCH_TIME_FOREVER：semaphore_wait（）函数是一直阻塞等待，如果中途信息处理器中断了，则重新开启等待；如果有一天终于发信号了，则等待结束，执行GOTO语句，此时dsema_sent_ksignals已经大于0了，返回0等待信号成功；
    
}

default : //这里计算剩余时间，利用mach内核的信号量等待函数semaphore_timedwait()进行等待，参考了POSIX的sem_timedwait()函数，该函数是阻塞式等待，如果在指定时间内没有得到通知，则会一直阻塞住，监听dsema_port等待其通知；当超时时返回超时结果？怎么结束GOTO语句？（仔细看下源码，default分支最后居然没有break，也就是说semaphore_timedwait()超时了之后还是执行下面的NOW分支，哎，这这这，害我懵逼了半天超时等待过期时怎么退出该函数）

case DISPATCH_TIME_NOW: //如果超时了，因为超时了（两种超时情况），所以撤销wait函数里一开始对dsema->dsema_value的减1操作，然后直接return 说明等待超时了，结束函数；如果没有超时，则什么都不执行，执行GOTO语句，返回重新执行一遍；

case DISPATCH_TIME_FOREVER：semaphore_wait（）函数是一直阻塞等待，如果中途信息处理器中断了，则重新开启等待；如果有一天别的线程终于发信号了，则等待结束，执行GOTO语句，此时dsema_sent_ksignals已经大于0了，返回0，表示等待信号成功；

这段代码的核心就是千万别漏掉default分支后面是没有break的，否则打死也想不明白指定时间等待超时该怎么恢复dsema_value，并且结束函数返回超时的结果；另外有一点，其实dispatch_semaphore是同时利用了dsema_value和dsema_sent_ksignals来处理等待和解除等待的；

发送信号函数
在去除了一些编译指令优化的代码之后，如下：

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_inc()是原子性的加1操作，可以看出，如果该信号量没有等待者（dsema->dsema_value>=0），则直接增加信号量的值然后返回0（代表没有线程被唤醒）即可，否则进入_dispatch_semaphore_signal_slow()函数：

static long
_dispatch_semaphore_signal_slow(dispatch_semaphore_t dsema)
{
    kern_return_t kr;
    //懒加载方式初始化semaphore_t dsema_port
    _dispatch_semaphore_create_port(&dsema->dsema_port);
    
    dispatch_atomic_inc(&dsema->dsema_sent_ksignals);
    
    kr = semaphore_signal(dsema->dsema_port);
    DISPATCH_SEMAPHORE_VERIFY_KR(kr);//校验函数返回值，如果系统的信号量方法调用异常（此处是semaphore_signal()），则抛出异常
    
    return 1;
}

接下来看上面部分代码，根据上述的分析，进入这部分代码的条件是有线程在等待当前信号量，此时才去创建semaphore_t dsema_port,然后dsema_sent_ksignals+1，之后通过dsema->dsema_port系统信号量去唤醒一个等待该底层semaphore_t信号量的线程，实际上唤醒的并不是iOS用户等待的线程，而是_dispatch_semaphore_wait_slow里，switch分支里semaphore_time_wait和semaphore_wait()函数，进而间接的唤醒了_dispatch_semaphore_wait_slow函数；

小结

dispatch_semaphore底层利用了mach内核提供的信号量接口进行二次封装实现；
dispatch_semaphore利用了两个变量long desma_value 和 size_t dsema_sent_ksignals，当不需要唤醒任何线程的时候，只操作desma_value变量，当有线程对信号量进行等待时，发送的信号量数目以dsema_sent_ksignals的值为准。

回顾文章开头的疑问？都解决了吗？

因为dispatch_semaphore是基于mach提供的semaphore_t的相关api，而mach传递消息的方式都是通过端口，所以是的，应该是通过端口轮询的；
dispatch_semaphore变量不是全局的，底层实际的数据访问通过指针传递；

参考

http://www.jianshu.com/p/7d97901baca2
https://opensource.apple.com/source/libdispatch/libdispatch-84.5/src/semaphore.c.auto.html
http://www.jianshu.com/p/947153c6b409
https://bestswifter.com/deep-gcd/

GCD - dispatch_semaphore详解

前言

基本使用

数据结构

过程分析

小结

回顾文章开头的疑问？都解决了吗？

参考

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读