OC底层探索23-GCD(下)OC底层探索23-GCD(下)

用户8893176

发布于 2021-08-09 11:08:55

3380

发布于 2021-08-09 11:08:55

文章被收录于专栏：小黑娃Henry

在上篇OC底层探索22-GCD(上)中分析了GCD的串/并队列的创建，同步、异步函数执行，而且留下了：死锁、栅栏函数的坑会在本文中补上；

1、单例

+ (instancetype)sharedInstance {
    static HRTest *instance;
    static dispatch_once_t onceToken;
    dispatch_once(&onceToken, ^{
        instance = [[HRTest alloc] init];
    });
    return instance;
}

对于这段代码，相信你已经非常熟悉了，就是安全的创建一个单例。底层做了哪些操作呢？

1.1 dispatch_once_t 谓词

/*!
 * @typedef dispatch_once_t
 *
 * @abstract
 * A predicate for use with dispatch_once(). It must be initialized to zero.
 * Note: static and global variables default to zero.
 */
typedef intptr_t dispatch_once_t;

是一个整型类的指针；
初始化值必须为0，后续使用要进行赋值；

1.2 dispatch_once 函数

void dispatch_once(dispatch_once_t *val, dispatch_block_t block)
{
    dispatch_once_f(val, block, _dispatch_Block_invoke(block));
}
// 核心函数
void
dispatch_once_f(dispatch_once_t *val, void *ctxt, dispatch_function_t func)
{
    // 对谓词进行类型转换
    dispatch_once_gate_t l = (dispatch_once_gate_t)val;

#if !DISPATCH_ONCE_INLINE_FASTPATH || DISPATCH_ONCE_USE_QUIESCENT_COUNTER
    // 判断谓词中的标识符是否等于DLOCK_ONCE_DONE
    uintptr_t v = os_atomic_load(&l->dgo_once, acquire);
    if (likely(v == DLOCK_ONCE_DONE)) {
        return;
    }
    ...
#endif
    // 
    if (_dispatch_once_gate_tryenter(l)) {
        return _dispatch_once_callout(l, ctxt, func);
    }
    return _dispatch_once_wait(l);
}

_dispatch_Block_invoke对block做了一层包装；
如果谓词的标识符等于DLOCK_ONCE_DONE，则表示：该任务已经执行过。

1.2.1 标识符首次匹配

static inline bool
_dispatch_once_gate_tryenter(dispatch_once_gate_t l)
{
    // 谓词的标示符和DLOCK_ONCE_UNLOCKED进行匹配
    return os_atomic_cmpxchg(&l->dgo_once, DLOCK_ONCE_UNLOCKED,
            (uintptr_t)_dispatch_lock_value_for_self(), relaxed);
}

在系统内核中对谓词的标示符和DLOCK_ONCE_UNLOCKED做了原子性匹配；

1.2.2 dispatch_once任务执行

static void
_dispatch_once_callout(dispatch_once_gate_t l, void *ctxt,
        dispatch_function_t func)
{
    _dispatch_client_callout(ctxt, func);
    _dispatch_once_gate_broadcast(l);
}
// 1. block执行
static inline void
_dispatch_client_callout(void *ctxt, dispatch_function_t f)
{
    // 对block进行执行
    return f(ctxt);
}

// 2. 标识符更新
static inline void
_dispatch_once_gate_broadcast(dispatch_once_gate_t l)
{
    // 线程锁，保证线程安全
    dispatch_lock value_self = _dispatch_lock_value_for_self();
    uintptr_t v;
    v = _dispatch_once_mark_done(l);
    ...
}
static inline uintptr_t
_dispatch_once_mark_done(dispatch_once_gate_t dgo)
{
    //将谓词标识符更新成DLOCK_ONCE_DONE-原子性操作
    return os_atomic_xchg(&dgo->dgo_once, DLOCK_ONCE_DONE, release);
}

进入_dispatch_once_callout会分为2部分来进行执行：block的执行，谓词标识符的更新;
谓词标识符的更新是线程安全的；
谓词标识符->dgo_once更新为DLOCK_ONCE_DONE，和前面判断处是一致的，形成流程闭环；

2、栅栏函数

栅栏函数除了用于任务有依赖关系时，同时还可以用于数据安全

2.1 简单应用-异步栅栏函数

- (void)demo{
    // 创建一个并发队列
    dispatch_queue_t concurrentQueue = dispatch_queue_create("Henry", DISPATCH_QUEUE_CONCURRENT);
    dispatch_async(concurrentQueue, ^{
        sleep(1);
        NSLog(@"1111---%@",[NSThread currentThread]);
    });
    /* 2. 异步栅栏函数 */
    dispatch_barrier_async(concurrentQueue, ^{
        NSLog(@"栅栏函数----%@",[NSThread currentThread]);
    });
    dispatch_async(concurrentQueue, ^{
        NSLog(@"2222----%@",[NSThread currentThread]);
    });
    NSLog(@"主线程----%@",[NSThread currentThread]);
}

输出：

异步栅栏函数会阻拦当前并发队列的任务执行；
在栅栏函数中使用串行队列没有意义，还会造成额外的消耗；

2.2 简单应用-同步栅栏函数

- (void)demo2{
    // 创建一个并发队列
    dispatch_queue_t concurrentQueue = dispatch_queue_create("Henry", DISPATCH_QUEUE_CONCURRENT);
    dispatch_async(concurrentQueue, ^{
        sleep(1);
        NSLog(@"1111---%@",[NSThread currentThread]);
    });
    /* 2. 同步栅栏函数 */
    dispatch_barrier_sync(concurrentQueue, ^{
        NSLog(@"栅栏函数----%@",[NSThread currentThread]);
    });
    dispatch_async(concurrentQueue, ^{
        NSLog(@"2222----%@",[NSThread currentThread]);
    });
    NSLog(@"主线程----%@",[NSThread currentThread]);
}

输出：

同步栅栏函数会阻拦当前线程;这个地方可能会出现面试题；
同步栅栏函数依旧会阻拦当前并发队列的执行;

小结：

栅栏函数无论同步栅栏函还是异步栅栏函数都会对当前并发队列进行阻拦;
同步栅栏函数还会阻拦当前线程的任务执行;
栅栏函数必须使用自定义并发队列，主线程和全局并发线程还会有系统任务需要执行，不允许进行栅栏阻拦；

2.3 源码分析-异步栅栏函数

void
dispatch_barrier_async(dispatch_queue_t dq, dispatch_block_t work)
{
    dispatch_continuation_t dc = _dispatch_continuation_alloc();
    uintptr_t dc_flags = DC_FLAG_CONSUME | DC_FLAG_BARRIER;
    dispatch_qos_t qos;

    qos = _dispatch_continuation_init(dc, dq, work, 0, dc_flags);
    _dispatch_continuation_async(dq, dc, qos, dc_flags);
}

和OC底层探索22-GCD(上) 第五部分dispatch_async异步函数几乎一致，只有uintptr_t dc_flags = DC_FLAG_CONSUME | DC_FLAG_BARRIER;标示多了一个DC_FLAG_BARRIER；
和dispatch_async异步函数的执行流程类似，但是需要等待队列中其他任务完成之后才会执行;

2.4 源码分析-同步栅栏函数

void
dispatch_barrier_sync(dispatch_queue_t dq, dispatch_block_t work)
{
    uintptr_t dc_flags = DC_FLAG_BARRIER | DC_FLAG_BLOCK;
    if (unlikely(_dispatch_block_has_private_data(work))) {
        return _dispatch_sync_block_with_privdata(dq, work, dc_flags);
    }
    _dispatch_barrier_sync_f(dq, work, _dispatch_Block_invoke(work), dc_flags);
}

和OC底层探索22-GCD(上) 第；六部分dispatch_sync同步函数完全一致;
所以说同步函数串行队列和同步栅栏函数实现原理是一致的；

3、死锁

这个奔溃堆栈就是著名的死锁。

static void
_dispatch_sync_f_slow(dispatch_queue_class_t top_dqu, void *ctxt,
        dispatch_function_t func, uintptr_t top_dc_flags,
        dispatch_queue_class_t dqu, uintptr_t dc_flags)
{
    ...
    __DISPATCH_WAIT_FOR_QUEUE__(&dsc, dq);
    ...
}

static void
__DISPATCH_WAIT_FOR_QUEUE__(dispatch_sync_context_t dsc, dispatch_queue_t dq)
{
    uint64_t dq_state = _dispatch_wait_prepare(dq);
    if (unlikely(_dq_state_drain_locked_by(dq_state, dsc->dsc_waiter))) {
        DISPATCH_CLIENT_CRASH((uintptr_t)dq_state,
                "dispatch_sync called on queue "
                "already owned by current thread");
    }
    ...
}

static inline bool
_dispatch_lock_is_locked_by(dispatch_lock lock_value, dispatch_tid tid)
{
    // equivalent to _dispatch_lock_owner(lock_value) == tid
    //即判断 当前要等待完成执行的任务 和 要执行的任务是否一样
    return ((lock_value ^ tid) & DLOCK_OWNER_MASK) == 0;
}

同一任务既要等待又要执行，就会出现矛盾；

4、信号量

根据信号量来控制线程的并发数；

// 初始化信号量，规定并发数
dispatch_semaphore_t sem = dispatch_semaphore_create(1);
// 开始执行
dispatch_semaphore_wait(sem, DISPATCH_TIME_FOREVER);
// 执行结束
dispatch_semaphore_signal(sem);

使用很简单，dispatch_semaphore_wait，dispatch_semaphore_signal成对出现即可；

4.1 dispatch_semaphore_create

dispatch_semaphore_t
dispatch_semaphore_create(intptr_t value)
{
    dispatch_semaphore_t dsema;
    // 对象创建
    dsema = _dispatch_object_alloc(DISPATCH_VTABLE(semaphore),
            sizeof(struct dispatch_semaphore_s));
    dsema->do_next = DISPATCH_OBJECT_LISTLESS;
    dsema->do_targetq = _dispatch_get_default_queue(false);
    dsema->dsema_value = value;
    _dispatch_sema4_init(&dsema->dsema_sema, _DSEMA4_POLICY_FIFO);
    dsema->dsema_orig = value;
    return dsema;
}

简单的对象创建，需要注意的是dsema_value代表并发数;

4.2 dispatch_semaphore_wait

intptr_t
dispatch_semaphore_wait(dispatch_semaphore_t dsema, dispatch_time_t timeout)
{
    // 原子性完成--操作后赋值给dsema_value
    long value = os_atomic_dec2o(dsema, dsema_value, acquire);
    if (likely(value >= 0)) {
        return 0;
    }
    // 进入长等待
    return _dispatch_semaphore_wait_slow(dsema, timeout);
}

os_atomic_dec2o原子性完成--操作后赋值给dsema_value；dsema_value代表并发数;
自减后value >= 0 : 当前任务队列处于就绪状态；
自减后value < 0 : 当前（线程）任务队列处于堵塞状态，需要等待；

4.3 dispatch_semaphore_signal

intptr_t
dispatch_semaphore_signal(dispatch_semaphore_t dsema)
{
    // 原子性自增
    long value = os_atomic_inc2o(dsema, dsema_value, release);
    if (likely(value > 0)) {
        return 0;
    }
    // LONG_MIN最小值
    if (unlikely(value == LONG_MIN)) {
        DISPATCH_CLIENT_CRASH(value,
                "Unbalanced call to dispatch_semaphore_signal()");
    }
    return _dispatch_semaphore_signal_slow(dsema);
}

同理os_atomic_inc2o原子性完成++操作后赋值给dsema_value；
自增后value > 0当前线程处于就绪状态，；
自增后value <= 0当前（线程）任务队列处于堵塞状态，需要等待；

4.3.1 宏定义

os_atomic_inc2o(p, f, m) 
?
os_atomic_add2o(p, f, 1, m)
?
os_atomic_add(&(p)->f, (v), m)
?
_os_atomic_c11_op((p), (v), m, add, +)
?
({ _os_atomic_basetypeof(p) _v = (v), _r = \
        atomic_fetch_##o##_explicit(_os_atomic_c11_atomic(p), _v, \
        memory_order_##m); (__typeof__(_r))(_r op _v); })

atomic_fetch_add_explicit通过原子性完成 ++操作；

小结

使用信号量可以控制当前线程的并发数,本质上也是简单的加减操作;

5、调度组

根据调度组来实现，成组任务的监听；它的实现和信号量非常类似；

dispatch_group_create 创建组 
//进组和出组一般是成对使用的
dispatch_group_enter 进组 
dispatch_group_leave 出组
//简便写法（一般使用）
dispatch_group_async 异步线程组
// 成组任务完成后的监听监听
dispatch_group_notify 进组任务执行完毕通知 dispatch_group_wait 进组任务执行等待时间

5.1 dispatch_group_create 创建

dispatch_group_t
dispatch_group_create(void)
{
    return _dispatch_group_create_with_count(0);
}

static inline dispatch_group_t
_dispatch_group_create_with_count(uint32_t n)
{
    // GCD对象创建
    
    dispatch_group_t dg = _dispatch_object_alloc(DISPATCH_VTABLE(group),
            sizeof(struct dispatch_group_s));
    dg->do_next = DISPATCH_OBJECT_LISTLESS;
    dg->do_targetq = _dispatch_get_default_queue(false);
    return dg;
}

dispatch_group_t对象的创建；
DISPATCH_VTABLE(group) == _dispatch_group_vtable 字符串的拼接，类名的由来；

5.2 dispatch_group_enter 进入

void
dispatch_group_enter(dispatch_group_t dg)
{
    // The value is decremented on a 32bits wide atomic so that the carry
    // for the 0 -> -1 transition is not propagated to the upper 32bits.
    // 从0 -> -1，自减
    uint32_t old_bits = os_atomic_sub_orig2o(dg, dg_bits,
            DISPATCH_GROUP_VALUE_INTERVAL, acquire);
    uint32_t old_value = old_bits & DISPATCH_GROUP_VALUE_MASK;
    if (unlikely(old_value == 0)) {
        _dispatch_retain(dg); // <rdar://problem/22318411>
    }
    // DISPATCH_GROUP_VALUE_MAX 最大线程组数
    if (unlikely(old_value == DISPATCH_GROUP_VALUE_MAX)) {
        DISPATCH_CLIENT_CRASH(old_bits,
                "Too many nested calls to dispatch_group_enter()");
    }
}

os_atomic_sub_orig2o原子性自减;可以看做是信号量中的dispatch_semaphore_wait看待；
达到最大线程组数后，会出现奔溃；

5.3 dispatch_group_leave 推出

void
dispatch_group_leave(dispatch_group_t dg)
{
    // The value is incremented on a 64bits wide atomic so that the carry for
    // the -1 -> 0 transition increments the generation atomically.
    // 从 -1 -> 0 自增
    uint64_t new_state, old_state = os_atomic_add_orig2o(dg, dg_state,
            DISPATCH_GROUP_VALUE_INTERVAL, release);
    uint32_t old_value = (uint32_t)(old_state & DISPATCH_GROUP_VALUE_MASK);

    if (unlikely(old_value == DISPATCH_GROUP_VALUE_1)) {
        old_state += DISPATCH_GROUP_VALUE_INTERVAL;
        ...
        return _dispatch_group_wake(dg, old_state, true);
    }
    if (unlikely(old_value == 0)) {
        DISPATCH_CLIENT_CRASH((uintptr_t)old_value,
                "Unbalanced call to dispatch_group_leave()");
    }
}

os_atomic_add_orig2o自增+1；
old_value == 0,如果0 + 1 = 1，enter-leave不平衡会出现崩溃，所以该函数必须成对出现，而且必须是先enter再leave；
_dispatch_group_wake达到条件之后就会weak;否则就循环等待；

5.2.1 _dispatch_group_wake 唤醒

static void
_dispatch_group_wake(dispatch_group_t dg, uint64_t dg_state, bool needs_release)
{
    if (dg_state & DISPATCH_GROUP_HAS_NOTIFS) {
        dispatch_continuation_t dc, next_dc, tail;

        //在当前线程创建一个notify任务；
        dc = os_mpsc_capture_snapshot(os_mpsc(dg, dg_notify), &tail);
        do {
            dispatch_queue_t dsn_queue = (dispatch_queue_t)dc->dc_data;
            next_dc = os_mpsc_pop_snapshot_head(dc, tail, do_next);
            //异步函数进行notify任务的执行
            _dispatch_continuation_async(dsn_queue, dc,
                    _dispatch_qos_from_pp(dc->dc_priority), dc->dc_flags);
            _dispatch_release(dsn_queue);
        } while ((dc = next_dc));

        refs++;
    }

    if (dg_state & DISPATCH_GROUP_HAS_WAITERS) {
        _dispatch_wake_by_address(&dg->dg_gen);
    }

    if (refs) _dispatch_release_n(dg, refs);
}

通过一个异步函数async来执行通知notify；
在每一次dispatch_group_leave后都会进行weak的判断；

5.4 dispatch_group_async 异步线程组

void
dispatch_group_async(dispatch_group_t dg, dispatch_queue_t dq,
        dispatch_block_t db)
{
    // 对象创建
    dispatch_continuation_t dc = _dispatch_continuation_alloc();
    uintptr_t dc_flags = DC_FLAG_CONSUME | DC_FLAG_GROUP_ASYNC;
    dispatch_qos_t qos;
    //任务包装器
    qos = _dispatch_continuation_init(dc, dq, db, 0, dc_flags);
    _dispatch_continuation_group_async(dg, dq, dc, qos);
}

static inline void
_dispatch_continuation_group_async(dispatch_group_t dg, dispatch_queue_t dq,
        dispatch_continuation_t dc, dispatch_qos_t qos)
{
    // 系统调用dispatch_group_enter
    dispatch_group_enter(dg);
    dc->dc_data = dg;
    // 异步任务的执行
    _dispatch_continuation_async(dq, dc, qos, dc->dc_flags);
}

简便使用系统底层帮我们做了dispatch_group_enter和dispatch_group_leaave的调用；

5.4.1 证实调用-dispatch_group_leaave

根据之前分析得知_dispatch_continuation_async最终会走到dx_push后面流程，系统做了隐藏，所以通过打印堆栈信息。

异步函数的任务执行离不开_dispatch_client_callout这个函数，那我们就全局搜索这个函数;

最后还是给我找到了具体的调用，只是流程没法补全只能脑补中间流程了；

dispatch_group小结

借助信号量的实现原理，就很好理解dispatch_group了；

总结

到此GCD中常见的方法都分析到了，或深或浅！当然libdispath中这只有九牛一毛，还有很多很多的东西，以后有机会一定继续深入。现在我头有点疼我先缓缓。

欢迎在留言和我沟通！

本文参与腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。

原始发表：2021/7/1 下午，如有侵权请联系 cloudcommunity@tencent.com 删除

海外加速

本文分享自作者个人站点/博客前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体同步曝光计划，欢迎热爱写作的你一起参与！

海外加速

登录后参与评论

0 条评论

热度