创建组
dispatch_group_t
dispatch_group_create(void)
{
dispatch_group_t dg = _dispatch_alloc(DISPATCH_VTABLE(group),
sizeof(struct dispatch_semaphore_s));
_dispatch_semaphore_init(LONG_MAX, dg);
return dg;
}
- 调用
_dispatch_alloc()创建 - 调用
_dispatch_semaphore_init()初始化
2.1 初始化dispatch_semaphore_t结构体的计数器、执行队列、目标值
进入组
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_semaphore_wait()加锁等待,返回值为等待数量- 信号量计数器先自减 value，然后返回.
- value > 0 返回 0 表示目前 waitor 数量为0，因为传入的 value 无穷大，所以到此返回。
因此，调用 group_enter() 之后，当前线程不会被阻塞。
离开组
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);
if (slowpath(value == LONG_MIN)) {
DISPATCH_CLIENT_CRASH("Unbalanced call to dispatch_group_leave()");
}
if (slowpath(value == dsema->dsema_orig)) {
(void)_dispatch_group_wake(dsema);
}
}
- 信号量计数器先自加 value，然后返回.
- 如果自加后的value 为 dsema_orig 调用
_dispatch_group_wake()唤醒notify
设置组监控
#ifdef __BLOCKS__
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);
}
#endif
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;
// 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();
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) {
_dispatch_group_wake(dsema);
}
}
}
- 调用
dispatch_group_notify_f()传入组，执行队列，block，block调用方法指针。- 初始化一个
dispatch_sema_notify_s结构体 dsn - 在 group 中，notify 为一个链表，此时调用
dispatch_atomic_xchg2o获取前驱节点。 - 如果找到前驱节点，那么链接 dsn;
- 如果没找到前驱节点，把 dsn 作为首节点
- 检查所有任务是否执行完毕
- 初始化一个
组唤醒
static long
_dispatch_group_wake(dispatch_semaphore_t dsema)
{
struct dispatch_sema_notify_s *next, *head, *tail = NULL;
long rval;
head = dispatch_atomic_xchg2o(dsema, dsema_notify_head, NULL);
if (head) {
// snapshot before anything is notified/woken <rdar://problem/8554546>
tail = dispatch_atomic_xchg2o(dsema, dsema_notify_tail, NULL);
}
rval = dispatch_atomic_xchg2o(dsema, dsema_group_waiters, 0);
if (rval) {
// 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);
#endif
}
if (head) {
// 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;
}
每个组leave的时候，都会检查是否所有任务都已经结束，然后等到所有任务都结束后，会调用wake 函数来执行 notify
- 获取notify链表的头/尾结点和等待执行的任务数
- 如果有任务在等待，向内核发送信号，唤起等待任务
- 异步执行链表中所有的 notify
组等待
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;
#endif
}
return _dispatch_group_wait_slow(dsema, timeout);
}
static long
_dispatch_group_wait_slow(dispatch_semaphore_t dsema, dispatch_time_t timeout)
{
long orig;
#if USE_MACH_SEM
mach_timespec_t _timeout;
kern_return_t kr;
#endif
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
_dispatch_semaphore_create_port(&dsema->dsema_port);
#endif
// 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:
#if USE_MACH_SEM
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;
}
#endif
// 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)) {
#endif
}
}
// Another thread called semaphore_signal().
// Fall through and drain the wakeup.
case DISPATCH_TIME_FOREVER:
#if USE_MACH_SEM
do {
kr = semaphore_wait(dsema->dsema_waiter_port);
} while (kr == KERN_ABORTED);
DISPATCH_SEMAPHORE_VERIFY_KR(kr);
#endif
break;
}
goto again;
}
- 先检查所有的任务是否都执行完了，则调用wake，【并且处理特殊情况】
- 然后调用 semaphore_wait() 向 dsema->dsema_waiter_port 内核发送等待信号。
总结
信号量结构体里面，有一个计数器，每次wait 和 signal 的时候，都会改这个计数器，wait 的时候如果达到临界值，计数器为0，就会向内核发送中断的信号，体现为不继续向下执行。如果sign 被调用，如果没达到临界值，那就继续计数器增加，如果达到临界值，为0，就会向内核发送唤醒信号，然后让中断的代码继续执行。
对于group，实际就是创建信号量计数为long_max的信号量。所以不会等待，所有添加到group的任务都能立即执行。notify函数作用就是把block加到信号量里面的一个链表中，等所有的group都执行完后，遍历这个链表，执行notify。【不涉及内核的中断和恢复】
但是 group_wait 就和上面信号量等待类似，涉及内核的中断和恢复了。
