Dispatch Group（调度组）
在追加到 Dispatch Queue 中的多个处理全部结束后想执行结束处理，这种情况会经常出现。只使用一个 Serial Dispatch Queue 时，只要将想执行的处理全部追加到该 Serial Dispatch Queue 中并在最后追加结束处理，即可实现。但是在使用 Concurrent Dispatch Queue 时或同时使用多个 Dispatch Queue 时，源代码就会变得颇为复杂。
在此种情况下使用 Dispatch Group，Dispatch Group 涉及下面 6 个函数互相搭配使用控制任务执⾏顺序。
-
dispatch_group_create创建组 -
dispatch_group_async进组任务 -
dispatch_group_notify进组任务执⾏完毕通知 -
dispatch_group_wait进组任务执⾏等待时间 -
dispatch_group_enter进组 -
dispatch_group_leave出组
例如下面源代码为：追加 3 个 Block 到 Global Dispatch Queue，这些 Block 如果全部执行完毕，就会通过 dispatch_group_notify 执行 Main Dispatch Queue中 结束处理用的 Block。
dispatch_queue_t queue = dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0);
dispatch_group_t group = dispatch_group_create();
dispatch_group_async(group, queue,^{ NSLog(@"blkO");});
dispatch_group_async(group, queue,^{NSLog(@"blk1");});
dispatch_group_async(group, queue,^{NSLog(@"blk2");});
dispatch_group_notify(group, dispatch_get_main_queue(), ^{NSLog("done");});}
或者使用 dispatch_group_enter 和 dispatch_group_leave
dispatch_queue_t queue = dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0);
dispatch_group_t group = dispatch_group_create();
dispatch_group_enter(group);
dispatch_async(queue,^{
NSLog(@"blkO");
dispatch_leave(group);
});
dispatch_group_notify(group, dispatch_get_main_queue(), ^{NSLog("done");});}
另外，在 Dispatch Group 中也可以使用 dispatch_group_wait 函数仅等待全部处理执行结束。
dispatch_queue_t queue = dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0);
dispatch_group_t group = dispatch_group_create();
dispatch_group_async(group,queue,^{NSLog("blkO");});
dispatch_group_async(group,queue,^{NSLog("blk1");});
dispatch_group_async(group,queue,^{NSLog("blk2");});
long result = dispatch_group_wait(group, DISPATCH_TIME_FOREVER);
dispatch_group_wait 函数的第二个参数指定为等待的时间(超时)。它属 dispatch_time_t 类型的值。该源代码使用DISPATCH_TIME_FOREVER,意味着永久等待。
如果 dispatch_group_wait 函数的返回值不为 0 ，就意味着虽然经过了指定的时间，但属于 Dispatch Group 的某-个处理还在执行中。如果返冋值为 0 ,那么全部处理执行结束。当等待时间为 DISPATCH_TIME_FOREVER 。由 dispatch_group_wait 函数返回时，由于属于 Dispatch Group 的处理必定全部执行结束，因此返回值恒为 0。
dispatch_group_create 实现
在 libdispatch 中查找 dispatch_group_create 函数，可以发现其实际调用 _dispatch_group_create_with_count函数，并传递参数 0。 创建一个 dispatch_group_t 类型值并返回。
dispatch_group_t
dispatch_group_create(void)
{
return _dispatch_group_create_with_count(0);
}
DISPATCH_ALWAYS_INLINE
static inline dispatch_group_t
_dispatch_group_create_with_count(uint32_t n)
{
//OS_dispatch_group
dispatch_group_t dg = _dispatch_object_alloc(DISPATCH_VTABLE(group),
sizeof(struct dispatch_group_s));
//#define DISPATCH_OBJECT_LISTLESS ((void *)0xffffffff89abcdef)
dg->do_next = DISPATCH_OBJECT_LISTLESS;
dg->do_targetq = _dispatch_get_default_queue(false);
if (n) {
os_atomic_store2o(dg, dg_bits,
(uint32_t)-n * DISPATCH_GROUP_VALUE_INTERVAL, relaxed);
os_atomic_store2o(dg, do_ref_cnt, 1, relaxed); // <rdar://22318411>
}
return dg;
}
dispatch_group_enter 实现
#define DISPATCH_GROUP_VALUE_INTERVAL 0x0000000000000004ULL
#define DISPATCH_GROUP_VALUE_MAX DISPATCH_GROUP_VALUE_INTERVAL
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.
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>
}
//0x0000000000000004ULL
if (unlikely(old_value == DISPATCH_GROUP_VALUE_MAX)) {
DISPATCH_CLIENT_CRASH(old_bits,
"Too many nested calls to dispatch_group_enter()");
}
}
dispatch_group_enter 其实就是通过 os_atomic_sub_orig2o 对 dg 的 dg_bits 进行 -1 操作，并且对 old_value 进行判断临界值。
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.
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;
do {
new_state = old_state;
if ((old_state & DISPATCH_GROUP_VALUE_MASK) == 0) {
new_state &= ~DISPATCH_GROUP_HAS_WAITERS;
new_state &= ~DISPATCH_GROUP_HAS_NOTIFS;
} else {
// If the group was entered again since the atomic_add above,
// we can't clear the waiters bit anymore as we don't know for
// which generation the waiters are for
new_state &= ~DISPATCH_GROUP_HAS_NOTIFS;
}
if (old_state == new_state) break;
} while (unlikely(!os_atomic_cmpxchgv2o(dg, dg_state,
old_state, new_state, &old_state, relaxed)));
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()");
}
}
dispatch_group_leave 流程如下：
- 通过
os_atomic_add_orig2o对dg的dg_state进行 +1 操作； - 根据
dg_state，do-while循环， 调用_dispatch_group_wake； - 如果
old_value == 0，调用过多的dispatch_group_leave，Crash
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(dg);
dc->dc_data = dg;
_dispatch_continuation_async(dq, dc, qos, dc->dc_flags);
}
static inline void
_dispatch_continuation_with_group_invoke(dispatch_continuation_t dc)
{
struct dispatch_object_s *dou = dc->dc_data;
unsigned long type = dx_type(dou);
if (type == DISPATCH_GROUP_TYPE) {
_dispatch_client_callout(dc->dc_ctxt, dc->dc_func);
_dispatch_trace_item_complete(dc);
dispatch_group_leave((dispatch_group_t)dou);
} else {
DISPATCH_INTERNAL_CRASH(dx_type(dou), "Unexpected object type");
}
}
dispatch_group_async 实现流程如下：
- 调用
dispatch_group_enter进组 - 保存
dg -
_dispatch_continuation_async异步执行任务，与dispatch_async相同。
static inline void
_dispatch_continuation_invoke_inline(dispatch_object_t dou,
dispatch_invoke_flags_t flags, dispatch_queue_class_t dqu)
{
//...
if (unlikely(dc_flags & DC_FLAG_GROUP_ASYNC)) {
_dispatch_continuation_with_group_invoke(dc);
}
//...
}
s
static inline void
_dispatch_continuation_with_group_invoke(dispatch_continuation_t dc)
{
struct dispatch_object_s *dou = dc->dc_data;
unsigned long type = dx_type(dou);
if (type == DISPATCH_GROUP_TYPE) {
_dispatch_client_callout(dc->dc_ctxt, dc->dc_func);
_dispatch_trace_item_complete(dc);
dispatch_group_leave((dispatch_group_t)dou);
} else {
DISPATCH_INTERNAL_CRASH(dx_type(dou), "Unexpected object type");
}
}
由于， dispatch_group_enter 与 dispatch_group_leave 是成对的。按照 dispatch_async 执行任务的流程，在 _dispatch_continuation_invoke_inline 中找到 DC_FLAG_GROUP_ASYNC 的判断 ，调用 _dispatch_continuation_with_group_invoke 函数，其中调用了 dispatch_group_leave出组。
因此，
dispatch_group_async等于dispatch_group_enter、dispatch_group_leave以及dispatch_async的搭配使用。
_dispatch_group_notify 实现
void
dispatch_group_notify(dispatch_group_t dg, dispatch_queue_t dq,
dispatch_block_t db)
{
dispatch_continuation_t dsn = _dispatch_continuation_alloc();
_dispatch_continuation_init(dsn, dq, db, 0, DC_FLAG_CONSUME);
_dispatch_group_notify(dg, dq, dsn);
}
DISPATCH_ALWAYS_INLINE
static inline void
_dispatch_group_notify(dispatch_group_t dg, dispatch_queue_t dq,
dispatch_continuation_t dsn)
{
uint64_t old_state, new_state;
dispatch_continuation_t prev;
dsn->dc_data = dq;
_dispatch_retain(dq);
prev = os_mpsc_push_update_tail(os_mpsc(dg, dg_notify), dsn, do_next);
if (os_mpsc_push_was_empty(prev)) _dispatch_retain(dg);
os_mpsc_push_update_prev(os_mpsc(dg, dg_notify), prev, dsn, do_next);
if (os_mpsc_push_was_empty(prev)) {
os_atomic_rmw_loop2o(dg, dg_state, old_state, new_state, release, {
new_state = old_state | DISPATCH_GROUP_HAS_NOTIFS;
if ((uint32_t)old_state == 0) {
os_atomic_rmw_loop_give_up({
return _dispatch_group_wake(dg, new_state, false);
});
}
});
}
}
dispatch_group_notify 通过 os_mpsc_push_update_tail 获取dg的状态码，判断 old_state == 0 时，调用 _dispatch_group_wake
_dispatch_group_wake 实现
DISPATCH_NOINLINE
static void
_dispatch_group_wake(dispatch_group_t dg, uint64_t dg_state, bool needs_release)
{
uint16_t refs = needs_release ? 1 : 0; // <rdar://problem/22318411>
if (dg_state & DISPATCH_GROUP_HAS_NOTIFS) {
dispatch_continuation_t dc, next_dc, tail;
// Snapshot before anything is notified/woken <rdar://problem/8554546>
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);
_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++;
}
//...
}
_dispatch_group_wake 函数判断当前 dg_state 有 DISPATCH_GROUP_HAS_NOTIFS，do-while 循环进行异步命中，调用 _dispatch_continuation_async 执行任务
