程序猿阵线联盟-汇总各类技术干货多线程iOS开发

深入理解GCD之dispatch_group

2018-08-21  本文已影响359人  NeroXie

之前已经介绍了dispatch_semaphore的底层实现，dispatch_group的实现是基于前者的。在看源码之前，我们先看一下我们是如何应用的。假设有这么场景：有一个A耗时操作，B和C两个网络请求和一个耗时操作C当ABC都执行完成后，刷新页面。我们可以用dispatch_group实现。关键如下：

- (void)viewDidLoad {
    [super viewDidLoad];
    
        __block NSInteger number = 0;
    
    dispatch_group_t group = dispatch_group_create();
    
    //A耗时操作
    dispatch_group_async(group, dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{
        sleep(3);
        number += 2222;
    });
    
    //B网络请求
    dispatch_group_enter(group);
    [self sendRequestWithCompletion:^(id response) {
        number += [response integerValue];
        dispatch_group_leave(group);
    }];
    
    //C网络请求
    dispatch_group_enter(group);
    [self sendRequestWithCompletion:^(id response) {
        number += [response integerValue];
        dispatch_group_leave(group);
    }];
    
    dispatch_group_notify(group, dispatch_get_main_queue(), ^{
        NSLog(@"%zd", number);
    });
}

- (void)sendRequestWithCompletion:(void (^)(id response))completion {
    //模拟一个网络请求
    dispatch_queue_t queue = dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0);
    dispatch_async(queue, ^{
        sleep(2);
        dispatch_async(dispatch_get_main_queue(), ^{
            if (completion) completion(@1111);
        });
    });
}

接下来我们根据上面的流程来看一下dispatch_group的相关API

dispatch_group_create

dispatch_group_t
dispatch_group_create(void)
{
    return (dispatch_group_t)dispatch_semaphore_create(LONG_MAX);
}

dispatch_group_create其实就是创建了一个value为LONG_MAX的dispatch_semaphore信号量

dispatch_group_async

void
dispatch_group_async(dispatch_group_t dg, dispatch_queue_t dq,
        dispatch_block_t db)
{
    dispatch_group_async_f(dg, dq, _dispatch_Block_copy(db),
            _dispatch_call_block_and_release);
}

dispatch_group_async只是dispatch_group_async_f的封装

dispatch_group_async_f

void
dispatch_group_async_f(dispatch_group_t dg, dispatch_queue_t dq, void *ctxt,
        dispatch_function_t func)
{
    dispatch_continuation_t dc;

    _dispatch_retain(dg);
    dispatch_group_enter(dg);

    dc = fastpath(_dispatch_continuation_alloc_cacheonly());
    if (!dc) {
        dc = _dispatch_continuation_alloc_from_heap();
    }

    dc->do_vtable = (void *)(DISPATCH_OBJ_ASYNC_BIT | DISPATCH_OBJ_GROUP_BIT);
    dc->dc_func = func;
    dc->dc_ctxt = ctxt;
    dc->dc_group = dg;

    // No fastpath/slowpath hint because we simply don't know
    if (dq->dq_width != 1 && dq->do_targetq) {
        return _dispatch_async_f2(dq, dc);
    }

    _dispatch_queue_push(dq, dc);
}

从上面的代码我们可以看出dispatch_group_async_f和dispatch_async_f相似。dispatch_group_async_f多了dispatch_group_enter(dg);，另外在do_vtable的赋值中dispatch_group_async_f多了一个DISPATCH_OBJ_GROUP_BIT的标记符。既然添加了dispatch_group_enter必定会存在dispatch_group_leave。在之前《深入理解GCD之dispatch_queue》介绍_dispatch_continuation_pop函数的源码中有一段代码如下：

    _dispatch_client_callout(dc->dc_ctxt, dc->dc_func);
    if (dg) {
        //group需要进行调用dispatch_group_leave并释放信号
        dispatch_group_leave(dg);
        _dispatch_release(dg);
    }

所以dispatch_group_async_f函数中的dispatch_group_leave是在_dispatch_continuation_pop函数中调用的。

这里概括一下dispatch_group_async_f的工作流程：

  1. 调用dispatch_group_enter
  2. 将block和queue等信息记录到dispatch_continuation_t结构体中，并将它加入到group的链表中。
  3. _dispatch_continuation_pop执行时会判断任务是否为group，是的话执行完任务再调用dispatch_group_leave以达到信号量的平衡。

dispatch_group_enter

void
dispatch_group_enter(dispatch_group_t dg)
{
    dispatch_semaphore_t dsema = (dispatch_semaphore_t)dg;

    (void)dispatch_semaphore_wait(dsema, DISPATCH_TIME_FOREVER);
}

dispatch_group_enter将dispatch_group_t转换成dispatch_semaphore_t，并调用dispatch_semaphore_wait，原子性减1后，进入等待状态直到有信号唤醒。所以说dispatch_group_enter就是对dispatch_semaphore_wait的封装。

dispatch_group_leave

void
dispatch_group_leave(dispatch_group_t dg)
{
    dispatch_semaphore_t dsema = (dispatch_semaphore_t)dg;
    dispatch_atomic_release_barrier();
    long value = dispatch_atomic_inc2o(dsema, dsema_value);//dsema_value原子性加1
    if (slowpath(value == LONG_MIN)) {//内存溢出，由于dispatch_group_leave在dispatch_group_enter之前调用
        DISPATCH_CLIENT_CRASH("Unbalanced call to dispatch_group_leave()");
    }
    if (slowpath(value == dsema->dsema_orig)) {//表示所有任务已经完成，唤醒group
        (void)_dispatch_group_wake(dsema);
    }
}

从上面的源代码中我们看到dispatch_group_leave将dispatch_group_t转换成dispatch_semaphore_t后将dsema_value的值原子性加1。如果value为LONG_MIN程序crash；如果value等于dsema_orig表示所有任务已完成，调用_dispatch_group_wake唤醒group（_dispatch_group_wake的用于和notify有关，我们会在后面介绍）。因为在enter的时候进行了原子性减1操作。所以在leave的时候需要原子性加1。

这里先说明一下enter和leave之间的关系：

  1. dispatch_group_leave与dispatch_group_enter配对使用。当调用了dispatch_group_enter而没有调用dispatch_group_leave时，由于value不等于dsema_orig不会走到唤醒逻辑，dispatch_group_notify中的任务无法执行或者dispatch_group_wait收不到信号而卡住线程。

  2. dispatch_group_enter必须在dispatch_group_leave之前出现。当dispatch_group_leave比dispatch_group_enter多调用了一次或者说在dispatch_group_enter之前被调用的时候，dispatch_group_leave进行原子性加1操作，相当于value为LONGMAX+1，发生数据长度溢出，变成LONG_MIN，由于value == LONG_MIN成立，程序发生crash。

dispatch_group_notify

void
dispatch_group_notify(dispatch_group_t dg, dispatch_queue_t dq,
        dispatch_block_t db)
{
    dispatch_group_notify_f(dg, dq, _dispatch_Block_copy(db),
            _dispatch_call_block_and_release);
}

dispatch_group_notify是dispatch_group_notify_f的封装，具体实现在后者。

dispatch_group_notify_f

void
dispatch_group_notify_f(dispatch_group_t dg, dispatch_queue_t dq, void *ctxt,
        void (*func)(void *))
{
    dispatch_semaphore_t dsema = (dispatch_semaphore_t)dg;
    struct dispatch_sema_notify_s *dsn, *prev;

    //封装dispatch_continuation_t结构体
    // FIXME -- this should be updated to use the continuation cache
    while (!(dsn = calloc(1, sizeof(*dsn)))) {
        sleep(1);
    }

    dsn->dsn_queue = dq;
    dsn->dsn_ctxt = ctxt;
    dsn->dsn_func = func;
    _dispatch_retain(dq);
    dispatch_atomic_store_barrier();
    //将结构体放到链表尾部，如果链表为空同时设置链表头部节点并唤醒group
    prev = dispatch_atomic_xchg2o(dsema, dsema_notify_tail, dsn);
    if (fastpath(prev)) {
        prev->dsn_next = dsn;
    } else {
        _dispatch_retain(dg);
        (void)dispatch_atomic_xchg2o(dsema, dsema_notify_head, dsn);
        if (dsema->dsema_value == dsema->dsema_orig) {//任务已经完成，唤醒group
            _dispatch_group_wake(dsema);
        }
    }
}

所以dispatch_group_notify函数只是用链表把所有回调通知保存起来，等待调用。

_dispatch_group_wake

static long
_dispatch_group_wake(dispatch_semaphore_t dsema)
{
    struct dispatch_sema_notify_s *next, *head, *tail = NULL;
    long rval;
    //将dsema的dsema_notify_head赋值为NULL，同时将之前的内容赋给head
    head = dispatch_atomic_xchg2o(dsema, dsema_notify_head, NULL);
    if (head) {
        // snapshot before anything is notified/woken <rdar://problem/8554546>
        //将dsema的dsema_notify_tail赋值为NULL，同时将之前的内容赋给tail
        tail = dispatch_atomic_xchg2o(dsema, dsema_notify_tail, NULL);
    }
    //将dsema的dsema_group_waiters设置为0，并返回原来的值
    rval = dispatch_atomic_xchg2o(dsema, dsema_group_waiters, 0);
    if (rval) {
        //循环调用semaphore_signal唤醒当初等待group的信号量，使得dispatch_group_wait函数返回。
        // wake group waiters
#if USE_MACH_SEM
        _dispatch_semaphore_create_port(&dsema->dsema_waiter_port);
        do {
            kern_return_t kr = semaphore_signal(dsema->dsema_waiter_port);
            DISPATCH_SEMAPHORE_VERIFY_KR(kr);
        } while (--rval);
#elif USE_POSIX_SEM
        do {
            int ret = sem_post(&dsema->dsema_sem);
            DISPATCH_SEMAPHORE_VERIFY_RET(ret);
        } while (--rval);
#endif
    }
    if (head) {
        //获取链表，依次调用dispatch_async_f异步执行在notify函数中的任务即Block。
        // async group notify blocks
        do {
            dispatch_async_f(head->dsn_queue, head->dsn_ctxt, head->dsn_func);
            _dispatch_release(head->dsn_queue);
            next = fastpath(head->dsn_next);
            if (!next && head != tail) {
                while (!(next = fastpath(head->dsn_next))) {
                    _dispatch_hardware_pause();
                }
            }
            free(head);
        } while ((head = next));
        _dispatch_release(dsema);
    }
    return 0;
}

_dispatch_group_wake主要的作用有两个：

  1. 调用semaphore_signal唤醒当初等待group的信号量，使得dispatch_group_wait函数返回。

  2. 获取链表，依次调用dispatch_async_f异步执行在notify函数中的任务即Block。

到这里我们已经差不多知道了dispatch_group工作过程，我们用一张图表示：

dispatch_group.png

dispatch_group_wait

dispatch_group_wait用于等待group中的任务完成。

long
dispatch_group_wait(dispatch_group_t dg, dispatch_time_t timeout)
{
    dispatch_semaphore_t dsema = (dispatch_semaphore_t)dg;

    if (dsema->dsema_value == dsema->dsema_orig) {//没有需要执行的任务
        return 0;
    }
    if (timeout == 0) {//返回超时
#if USE_MACH_SEM
        return KERN_OPERATION_TIMED_OUT;
#elif USE_POSIX_SEM
        errno = ETIMEDOUT;
        return (-1);
#endif
    }
    return _dispatch_group_wait_slow(dsema, timeout);
}

_dispatch_group_wait_slow

static long
_dispatch_group_wait_slow(dispatch_semaphore_t dsema, dispatch_time_t timeout)
{
    long orig;

again:
    // check before we cause another signal to be sent by incrementing
    // dsema->dsema_group_waiters
    if (dsema->dsema_value == dsema->dsema_orig) {
        return _dispatch_group_wake(dsema);
    }
    // Mach semaphores appear to sometimes spuriously wake up. Therefore,
    // we keep a parallel count of the number of times a Mach semaphore is
    // signaled (6880961).
    (void)dispatch_atomic_inc2o(dsema, dsema_group_waiters);
    // check the values again in case we need to wake any threads
    if (dsema->dsema_value == dsema->dsema_orig) {
        return _dispatch_group_wake(dsema);
    }

#if USE_MACH_SEM
    mach_timespec_t _timeout;
    kern_return_t kr;

    _dispatch_semaphore_create_port(&dsema->dsema_waiter_port);

    // 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.

    switch (timeout) {
    default:
        do {
            uint64_t 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);
            kr = slowpath(semaphore_timedwait(dsema->dsema_waiter_port,
                    _timeout));
        } while (kr == KERN_ABORTED);

        if (kr != KERN_OPERATION_TIMED_OUT) {
            DISPATCH_SEMAPHORE_VERIFY_KR(kr);
            break;
        }
        // Fall through and try to undo the earlier change to
        // dsema->dsema_group_waiters
    case DISPATCH_TIME_NOW:
        while ((orig = dsema->dsema_group_waiters)) {
            if (dispatch_atomic_cmpxchg2o(dsema, dsema_group_waiters, 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_waiter_port);
        } while (kr == KERN_ABORTED);
        DISPATCH_SEMAPHORE_VERIFY_KR(kr);
        break;
    }
#elif USE_POSIX_SEM
//这部分代码省略
#endif

    goto again;
}

从上面的代码我们发现_dispatch_group_wait_slow和_dispatch_semaphore_wait_slow的逻辑很接近。都利用mach内核的semaphore进行信号的发送。区别在于_dispatch_semaphore_wait_slow在等待结束后是return，而_dispatch_group_wait_slow在等待结束是调用_dispatch_group_wake去唤醒这个group。

总结

  1. dispatch_group是一个初始值为LONG_MAX的信号量，group中的任务完成是判断其value是否恢复成初始值。

  2. dispatch_group_enter和dispatch_group_leave必须成对使用并且支持嵌套。

  3. 如果dispatch_group_enter比dispatch_group_leave多，由于value不等于dsema_orig不会走到唤醒逻辑，dispatch_group_notify中的任务无法执行或者dispatch_group_wait收不到信号而卡住线程。如果是dispatch_group_leave多，则会引起崩溃。

上一篇下一篇

猜你喜欢

热点阅读