1、同步执行 dispatch_sync

  1. void
  2. dispatch_sync(dispatch_queue_t dq, dispatch_block_t work)
  3. {
  4.     uintptr_t dc_flags = DC_FLAG_BLOCK;
  5.     if (unlikely(_dispatch_block_has_private_data(work))) {
  6.         return _dispatch_sync_block_with_privdata(dq, work, dc_flags);
  7.     }
  8.     _dispatch_sync_f(dq, work, _dispatch_Block_invoke(work), dc_flags);
  9. }
  11. // 进一步查看 _dispatch_sync_f
  13. static void
  14. _dispatch_sync_f(dispatch_queue_t dq, void *ctxt, dispatch_function_t func,
  15.         uintptr_t dc_flags)
  16. {
  17.     _dispatch_sync_f_inline(dq, ctxt, func, dc_flags);
  18. }
  20. // 查看 _dispatch_sync_f_inline
  22. static inline void
  23. _dispatch_sync_f_inline(dispatch_queue_t dq, void *ctxt,
  24.         dispatch_function_t func, uintptr_t dc_flags)
  25. {
  26.     if (likely(dq->dq_width == 1)) {
  27.         return _dispatch_barrier_sync_f(dq, ctxt, func, dc_flags);
  28.     }
  30.     if (unlikely(dx_metatype(dq) != _DISPATCH_LANE_TYPE)) {
  31.         DISPATCH_CLIENT_CRASH(0, "Queue type doesn't support dispatch_sync");
  32.     }
  34.     dispatch_lane_t dl = upcast(dq)._dl;
  35.     // Global concurrent queues and queues bound to non-dispatch threads
  36.     // always fall into the slow case, see DISPATCH_ROOT_QUEUE_STATE_INIT_VALUE
  37.     if (unlikely(!_dispatch_queue_try_reserve_sync_width(dl))) {
  38.         return _dispatch_sync_f_slow(dl, ctxt, func, 0, dl, dc_flags);
  39.     }
  41.     if (unlikely(dq->do_targetq->do_targetq)) {
  42.         return _dispatch_sync_recurse(dl, ctxt, func, dc_flags);
  43.     }
  44.     _dispatch_introspection_sync_begin(dl);
  45.     _dispatch_sync_invoke_and_complete(dl, ctxt, func DISPATCH_TRACE_ARG(
  46.             _dispatch_trace_item_sync_push_pop(dq, ctxt, func, dc_flags)));
  47. }

可以看到首先通过 width 判定是串行队列还是并发队列，如果是并发队列则调用 _dispatch_sync_invoke_and_complete，如果是串行队列则调用 _dispatch_barrier_sync_f。我们先展开看一下串行队列的同步执行源代码：

  1. static void
  2. _dispatch_barrier_sync_f(dispatch_queue_t dq, void *ctxt,
  3.         dispatch_function_t func, uintptr_t dc_flags)
  4. {
  5.     _dispatch_barrier_sync_f_inline(dq, ctxt, func, dc_flags);
  6. }
  8. // 查看 _dispatch_barrier_sync_f_inline
  10. static inline void
  11. _dispatch_barrier_sync_f_inline(dispatch_queue_t dq, void *ctxt,
  12.         dispatch_function_t func, uintptr_t dc_flags)
  13. {
  14.     dispatch_tid tid = _dispatch_tid_self();
  16.     if (unlikely(dx_metatype(dq) != _DISPATCH_LANE_TYPE)) {
  17.         DISPATCH_CLIENT_CRASH(0, "Queue type doesn't support dispatch_sync");
  18.     }
  20.     dispatch_lane_t dl = upcast(dq)._dl;
  21.     // The more correct thing to do would be to merge the qos of the thread
  22.     // that just acquired the barrier lock into the queue state.
  23.     //
  24.     // However this is too expensive for the fast path, so skip doing it.
  25.     // The chosen tradeoff is that if an enqueue on a lower priority thread
  26.     // contends with this fast path, this thread may receive a useless override.
  27.     //
  28.     // Global concurrent queues and queues bound to non-dispatch threads
  29.     // always fall into the slow case, see DISPATCH_ROOT_QUEUE_STATE_INIT_VALUE
  30.     if (unlikely(!_dispatch_queue_try_acquire_barrier_sync(dl, tid))) {
  31.         return _dispatch_sync_f_slow(dl, ctxt, func, DC_FLAG_BARRIER, dl,
  32.                 DC_FLAG_BARRIER | dc_flags);
  33.     }
  35.     if (unlikely(dl->do_targetq->do_targetq)) {
  36.         return _dispatch_sync_recurse(dl, ctxt, func,
  37.                 DC_FLAG_BARRIER | dc_flags);
  38.     }
  39.     _dispatch_introspection_sync_begin(dl);
  40.     _dispatch_lane_barrier_sync_invoke_and_complete(dl, ctxt, func
  41.             DISPATCH_TRACE_ARG(_dispatch_trace_item_sync_push_pop(
  42.                     dq, ctxt, func, dc_flags | DC_FLAG_BARRIER)));
  43. }

可以看到上面的代码逻辑时首先获取线程 tid，然后处理死锁的情况，因此我们先看一下死锁的情况：

  1. static void
  2. _dispatch_sync_f_slow(dispatch_queue_class_t top_dqu, void *ctxt,
  3.         dispatch_function_t func, uintptr_t top_dc_flags,
  4.         dispatch_queue_class_t dqu, uintptr_t dc_flags)
  5. {
  6.     dispatch_queue_t top_dq = top_dqu._dq;
  7.     dispatch_queue_t dq = dqu._dq;
  8.     if (unlikely(!dq->do_targetq)) {
  9.         return _dispatch_sync_function_invoke(dq, ctxt, func);
  10.     }
  12.     pthread_priority_t pp = _dispatch_get_priority();
  13.     struct dispatch_sync_context_s dsc = {
  14.         .dc_flags    = DC_FLAG_SYNC_WAITER | dc_flags,
  15.         .dc_func     = _dispatch_async_and_wait_invoke,
  16.         .dc_ctxt     = &dsc,
  17.         .dc_other    = top_dq,
  18.         .dc_priority = pp | _PTHREAD_PRIORITY_ENFORCE_FLAG,
  19.         .dc_voucher  = _voucher_get(),
  20.         .dsc_func    = func,
  21.         .dsc_ctxt    = ctxt,
  22.         .dsc_waiter  = _dispatch_tid_self(),
  23.     };
  25.     _dispatch_trace_item_push(top_dq, &dsc);
  26.     __DISPATCH_WAIT_FOR_QUEUE__(&dsc, dq);
  28.     if (dsc.dsc_func == NULL) {
  29.         // dsc_func being cleared means that the block ran on another thread ie.
  30.         // case (2) as listed in _dispatch_async_and_wait_f_slow.
  31.         dispatch_queue_t stop_dq = dsc.dc_other;
  32.         return _dispatch_sync_complete_recurse(top_dq, stop_dq, top_dc_flags);
  33.     }
  35.     _dispatch_introspection_sync_begin(top_dq);
  36.     _dispatch_trace_item_pop(top_dq, &dsc);
  37.     _dispatch_sync_invoke_and_complete_recurse(top_dq, ctxt, func,top_dc_flags
  38.             DISPATCH_TRACE_ARG(&dsc));
  39. }
  41. // 查看 __DISPATCH_WAIT_FOR_QUEUE__
  43. static void
  44. __DISPATCH_WAIT_FOR_QUEUE__(dispatch_sync_context_t dsc, dispatch_queue_t dq)
  45. {
  46.     uint64_t dq_state = _dispatch_wait_prepare(dq);
  47.     if (unlikely(_dq_state_drain_locked_by(dq_state, dsc->dsc_waiter))) {
  48.         DISPATCH_CLIENT_CRASH((uintptr_t)dq_state,
  49.                 "dispatch_sync called on queue "
  50.                 "already owned by current thread");
  51.     }
  53.     // Blocks submitted to the main thread MUST run on the main thread, and
  54.     // dispatch_async_and_wait also executes on the remote context rather than
  55.     // the current thread.
  56.     //
  57.     // For both these cases we need to save the frame linkage for the sake of
  58.     // _dispatch_async_and_wait_invoke
  59.     _dispatch_thread_frame_save_state(&dsc->dsc_dtf);
  61.     if (_dq_state_is_suspended(dq_state) ||
  62.             _dq_state_is_base_anon(dq_state)) {
  63.         dsc->dc_data = DISPATCH_WLH_ANON;
  64.     } else if (_dq_state_is_base_wlh(dq_state)) {
  65.         dsc->dc_data = (dispatch_wlh_t)dq;
  66.     } else {
  67.         _dispatch_wait_compute_wlh(upcast(dq)._dl, dsc);
  68.     }
  70.     if (dsc->dc_data == DISPATCH_WLH_ANON) {
  71.         dsc->dsc_override_qos_floor = dsc->dsc_override_qos =
  72.                 (uint8_t)_dispatch_get_basepri_override_qos_floor();
  73.         _dispatch_thread_event_init(&dsc->dsc_event);
  74.     }
  75.     dx_push(dq, dsc, _dispatch_qos_from_pp(dsc->dc_priority));
  76.     _dispatch_trace_runtime_event(sync_wait, dq, 0);
  77.     if (dsc->dc_data == DISPATCH_WLH_ANON) {
  78.         _dispatch_thread_event_wait(&dsc->dsc_event); // acquire
  79.     } else {
  80.         _dispatch_event_loop_wait_for_ownership(dsc);
  81.     }
  82.     if (dsc->dc_data == DISPATCH_WLH_ANON) {
  83.         _dispatch_thread_event_destroy(&dsc->dsc_event);
  84.         // If _dispatch_sync_waiter_wake() gave this thread an override,
  85.         // ensure that the root queue sees it.
  86.         if (dsc->dsc_override_qos > dsc->dsc_override_qos_floor) {
  87.             _dispatch_set_basepri_override_qos(dsc->dsc_override_qos);
  88.         }
  89.     }
  90. }
  92. // 展开 _dq_state_drain_locked_by
  94. static inline bool
  95. _dq_state_drain_locked_by(uint64_t dq_state, dispatch_tid tid)
  96. {
  97.     return _dispatch_lock_is_locked_by((dispatch_lock)dq_state, tid);
  98. }
  100. // 然后看一下 _dispatch_lock_is_locked_by
  102. static inline bool
  103. _dispatch_lock_is_locked_by(dispatch_lock lock_value, dispatch_tid tid)
  104. {
  105.     // equivalent to _dispatch_lock_owner(lock_value) == tid
  106.     return ((lock_value ^ tid) & DLOCK_OWNER_MASK) == 0;
  107. }

可以看到队列 push 以后就是用 _dispatch_lock_is_locked_by 判断将要调度的和当前等待的队列是不是同一个，如果相同则返回 YES，产生死锁DISPATCH_CLIENT_CRASH；如果没有产生死锁，则执行 _dispatch_trace_item_pop()出队列执行。如何执行调度呢，我们可以看一下_dispatch_sync_invoke_and_complete_recurse

  1. static void
  2. _dispatch_sync_invoke_and_complete_recurse(dispatch_queue_class_t dq,
  3.         void *ctxt, dispatch_function_t func, uintptr_t dc_flags
  4.         DISPATCH_TRACE_ARG(void *dc))
  5. {
  6.     _dispatch_sync_function_invoke_inline(dq, ctxt, func);
  7.     _dispatch_trace_item_complete(dc);
  8.     _dispatch_sync_complete_recurse(dq._dq, NULL, dc_flags);
  9. }
  11. // 看一下 _dispatch_sync_function_invoke_inline
  13. static inline void
  14. _dispatch_sync_function_invoke_inline(dispatch_queue_class_t dq, void *ctxt,
  15.         dispatch_function_t func)
  16. {
  17.     dispatch_thread_frame_s dtf;
  18.     _dispatch_thread_frame_push(&dtf, dq);
  19.     _dispatch_client_callout(ctxt, func);
  20.     _dispatch_perfmon_workitem_inc();
  21.     _dispatch_thread_frame_pop(&dtf);
  22. }
  24. // 看一下 _dispatch_client_callout
  26. void
  27. _dispatch_client_callout(void *ctxt, dispatch_function_t f)
  28. {
  29.     @try {
  30.         return f(ctxt);
  31.     }
  32.     @catch (...) {
  33.         objc_terminate();
  34.     }
  35. }
  37. 可以比较清楚的看到最终执行f函数，这个就是外界传过来的回调block。

2、异步调用 dispatch_async

今天我们重新来分析一遍 dispatch_async 的调用过程

  1. void
  2. dispatch_async(dispatch_queue_t dq, dispatch_block_t work)
  3. {
  4.     dispatch_continuation_t dc = _dispatch_continuation_alloc();
  5.     uintptr_t dc_flags = DC_FLAG_CONSUME;
  6.     dispatch_qos_t qos;
  8.     qos = _dispatch_continuation_init(dc, dq, work, 0, dc_flags);
  9.     _dispatch_continuation_async(dq, dc, qos, dc->dc_flags);
  10. }
  12. // 查看 _dispatch_continuation_init 代码，主要进行block初始化
  14. static inline dispatch_qos_t
  15. _dispatch_continuation_init(dispatch_continuation_t dc,
  16.         dispatch_queue_class_t dqu, dispatch_block_t work,
  17.         dispatch_block_flags_t flags, uintptr_t dc_flags)
  18. {
  19.     void *ctxt = _dispatch_Block_copy(work);
  21.     dc_flags |= DC_FLAG_BLOCK | DC_FLAG_ALLOCATED;
  22.     if (unlikely(_dispatch_block_has_private_data(work))) {
  23.         dc->dc_flags = dc_flags;
  24.         dc->dc_ctxt = ctxt;
  25.         // will initialize all fields but requires dc_flags & dc_ctxt to be set
  26.         return _dispatch_continuation_init_slow(dc, dqu, flags);
  27.     }
  29.     dispatch_function_t func = _dispatch_Block_invoke(work);
  30.     if (dc_flags & DC_FLAG_CONSUME) {
  31.         func = _dispatch_call_block_and_release;
  32.     }
  33.     return _dispatch_continuation_init_f(dc, dqu, ctxt, func, flags, dc_flags);
  34. }
  36. // 另外查看 _dispatch_continuation_async
  38. static inline void
  39. _dispatch_continuation_async(dispatch_queue_class_t dqu,
  40.         dispatch_continuation_t dc, dispatch_qos_t qos, uintptr_t dc_flags)
  41. {
  42. #if DISPATCH_INTROSPECTION
  43.     if (!(dc_flags & DC_FLAG_NO_INTROSPECTION)) {
  44.         _dispatch_trace_item_push(dqu, dc);
  45.     }
  46. #else
  47.     (void)dc_flags;
  48. #endif
  49.     return dx_push(dqu._dq, dc, qos);
  50. }
  52. // 进一步查看 dx_push
  54. #define dx_push(x, y, z) dx_vtable(x)->dq_push(x, y, z)
  56. // 本质是调用 dx_vtable 的 dq_push (其实就是调用对象的 do_push )，进一步查看 dq_push，我们假设是 global_queue 进行异步调用可以看到：
  58. DISPATCH_VTABLE_SUBCLASS_INSTANCE(queue_global, lane,
  59.     .do_type        = DISPATCH_QUEUE_GLOBAL_ROOT_TYPE,
  60.     .do_dispose     = _dispatch_object_no_dispose,
  61.     .do_debug       = _dispatch_queue_debug,
  62.     .do_invoke      = _dispatch_object_no_invoke,
  64.     .dq_activate    = _dispatch_queue_no_activate,
  65.     .dq_wakeup      = _dispatch_root_queue_wakeup,
  66.     .dq_push        = _dispatch_root_queue_push,
  67. );

可以看到 dx_push已经到了 _dispatch_root_queue_push ，这是可以接着查看 _dispatch_root_queue_push：

  1. void
  2. _dispatch_root_queue_push(dispatch_queue_global_t rq, dispatch_object_t dou,
  3.         dispatch_qos_t qos)
  4. {
  5. #if DISPATCH_USE_KEVENT_WORKQUEUE
  6.     dispatch_deferred_items_t ddi = _dispatch_deferred_items_get();
  7.     if (unlikely(ddi && ddi->ddi_can_stash)) {
  8.         dispatch_object_t old_dou = ddi->ddi_stashed_dou;
  9.         dispatch_priority_t rq_overcommit;
  10.         rq_overcommit = rq->dq_priority & DISPATCH_PRIORITY_FLAG_OVERCOMMIT;
  12.         if (likely(!old_dou._do || rq_overcommit)) {
  13.             dispatch_queue_global_t old_rq = ddi->ddi_stashed_rq;
  14.             dispatch_qos_t old_qos = ddi->ddi_stashed_qos;
  15.             ddi->ddi_stashed_rq = rq;
  16.             ddi->ddi_stashed_dou = dou;
  17.             ddi->ddi_stashed_qos = qos;
  18.             _dispatch_debug("deferring item %p, rq %p, qos %d",
  19.                     dou._do, rq, qos);
  20.             if (rq_overcommit) {
  21.                 ddi->ddi_can_stash = false;
  22.             }
  23.             if (likely(!old_dou._do)) {
  24.                 return;
  25.             }
  26.             // push the previously stashed item
  27.             qos = old_qos;
  28.             rq = old_rq;
  29.             dou = old_dou;
  30.         }
  31.     }
  32. #endif
  33. #if HAVE_PTHREAD_WORKQUEUE_QOS
  34.     if (_dispatch_root_queue_push_needs_override(rq, qos)) {
  35.         return _dispatch_root_queue_push_override(rq, dou, qos);
  36.     }
  37. #else
  38.     (void)qos;
  39. #endif
  40.     _dispatch_root_queue_push_inline(rq, dou, dou, 1);
  41. }
  43. // 多数情况下符合 HAVE_PTHREAD_WORKQUEUE_QOS，会执行 _dispatch_root_queue_push_override（对比的是 qos 与 root 队列的 qos 是否一致，基本上都不一致的。）
  45. static void
  46. _dispatch_root_queue_push_override(dispatch_queue_global_t orig_rq,
  47.         dispatch_object_t dou, dispatch_qos_t qos)
  48. {
  49.     bool overcommit = orig_rq->dq_priority & DISPATCH_PRIORITY_FLAG_OVERCOMMIT;
  50.     dispatch_queue_global_t rq = _dispatch_get_root_queue(qos, overcommit);
  51.     dispatch_continuation_t dc = dou._dc;
  53.     if (_dispatch_object_is_redirection(dc)) {
  54.         // no double-wrap is needed, _dispatch_async_redirect_invoke will do
  55.         // the right thing
  56.         dc->dc_func = (void *)orig_rq;
  57.     } else {
  58.         dc = _dispatch_continuation_alloc();
  59.         dc->do_vtable = DC_VTABLE(OVERRIDE_OWNING);
  60.         dc->dc_ctxt = dc;
  61.         dc->dc_other = orig_rq;
  62.         dc->dc_data = dou._do;
  63.         dc->dc_priority = DISPATCH_NO_PRIORITY;
  64.         dc->dc_voucher = DISPATCH_NO_VOUCHER;
  65.     }
  66.     _dispatch_root_queue_push_inline(rq, dc, dc, 1);
  67. }
  68. /*
  69. 上面 _dispatch_object_is_redirection 函数其实就是 return _dispatch_object_has_type(dou,DISPATCH_CONTINUATION_TYPE(ASYNC_REDIRECT)); 
  70. 所以自定义队列会走这个 if 语句，如果是 dispatch_get_global_queue 不会走 if 语句。
  71. 展开 _dispatch_root_queue_push_inline。注意_dispatch_root_queue_push_inline中 的 if 把任务装进队列，大多数不走进if语句。
  72. 但是第一个任务进来之前还是满足这个条件式的，会进入这个条件语句去激活队列来执行里面的任务，后面再加入的任务因为队列被激活了，所以也就不太需要再进入这个队列了，所以相对来说激活队列只要一次。
  73. */
  75. static inline void
  76. _dispatch_root_queue_push_inline(dispatch_queue_global_t dq,
  77.         dispatch_object_t _head, dispatch_object_t _tail, int n)
  78. {
  79.     struct dispatch_object_s *hd = _head._do, *tl = _tail._do;
  80.     if (unlikely(os_mpsc_push_list(os_mpsc(dq, dq_items), hd, tl, do_next))) {
  81.         return _dispatch_root_queue_poke(dq, n, 0);
  82.     }
  83. }
  85. // 我们可以看到，我们装入到自定义的任务都被扔到其挂靠的root队列里去了，所以我们我们自己创建的队列只是一个代理人身份，继续查看 _dispatch_root_queue_poke 源码
  87. void
  88. _dispatch_root_queue_poke(dispatch_queue_global_t dq, int n, int floor)
  89. {
  90.     if (!_dispatch_queue_class_probe(dq)) {
  91.         return;
  92.     }
  93. #if !DISPATCH_USE_INTERNAL_WORKQUEUE
  94. #if DISPATCH_USE_PTHREAD_POOL
  95.     if (likely(dx_type(dq) == DISPATCH_QUEUE_GLOBAL_ROOT_TYPE))
  96. #endif
  97.     {
  98.         if (unlikely(!os_atomic_cmpxchg2o(dq, dgq_pending, 0, n, relaxed))) {
  99.             _dispatch_root_queue_debug("worker thread request still pending "
  100.                     "for global queue: %p", dq);
  101.             return;
  102.         }
  103.     }
  104. #endif // !DISPATCH_USE_INTERNAL_WORKQUEUE
  105.     return _dispatch_root_queue_poke_slow(dq, n, floor);
  106. }
  108. // 继续查看 _dispatch_root_queue_poke_slow
  110. static void
  111. _dispatch_root_queue_poke_slow(dispatch_queue_global_t dq, int n, int floor)
  112. {
  113.     int remaining = n;
  114.     int r = ENOSYS;
  116.     _dispatch_root_queues_init();
  117.     _dispatch_debug_root_queue(dq, __func__);
  118.     _dispatch_trace_runtime_event(worker_request, dq, (uint64_t)n);
  120. #if !DISPATCH_USE_INTERNAL_WORKQUEUE
  121. #if DISPATCH_USE_PTHREAD_ROOT_QUEUES
  122.     if (dx_type(dq) == DISPATCH_QUEUE_GLOBAL_ROOT_TYPE)
  123. #endif
  124.     {
  125.         _dispatch_root_queue_debug("requesting new worker thread for global "
  126.                 "queue: %p", dq);
  127.         r = _pthread_workqueue_addthreads(remaining,
  128.                 _dispatch_priority_to_pp_prefer_fallback(dq->dq_priority));
  129.         (void)dispatch_assume_zero(r);
  130.         return;
  131.     }
  132. #endif // !DISPATCH_USE_INTERNAL_WORKQUEUE
  133. #if DISPATCH_USE_PTHREAD_POOL
  134.     dispatch_pthread_root_queue_context_t pqc = dq->do_ctxt;
  135.     if (likely(pqc->dpq_thread_mediator.do_vtable)) {
  136.         while (dispatch_semaphore_signal(&pqc->dpq_thread_mediator)) {
  137.             _dispatch_root_queue_debug("signaled sleeping worker for "
  138.                     "global queue: %p", dq);
  139.             if (!--remaining) {
  140.                 return;
  141.             }
  142.         }
  143.     }
  145.     bool overcommit = dq->dq_priority & DISPATCH_PRIORITY_FLAG_OVERCOMMIT;
  146.     if (overcommit) {
  147.         os_atomic_add2o(dq, dgq_pending, remaining, relaxed);
  148.     } else {
  149.         if (!os_atomic_cmpxchg2o(dq, dgq_pending, 0, remaining, relaxed)) {
  150.             _dispatch_root_queue_debug("worker thread request still pending for "
  151.                     "global queue: %p", dq);
  152.             return;
  153.         }
  154.     }
  156.     int can_request, t_count;
  157.     // seq_cst with atomic store to tail <rdar://problem/16932833>
  158.     t_count = os_atomic_load2o(dq, dgq_thread_pool_size, ordered);
  159.     do {
  160.         can_request = t_count < floor ? 0 : t_count - floor;
  161.         if (remaining > can_request) {
  162.             _dispatch_root_queue_debug("pthread pool reducing request from %d to %d",
  163.                     remaining, can_request);
  164.             os_atomic_sub2o(dq, dgq_pending, remaining - can_request, relaxed);
  165.             remaining = can_request;
  166.         }
  167.         if (remaining == 0) {
  168.             _dispatch_root_queue_debug("pthread pool is full for root queue: "
  169.                     "%p", dq);
  170.             return;
  171.         }
  172.     } while (!os_atomic_cmpxchgvw2o(dq, dgq_thread_pool_size, t_count,
  173.             t_count - remaining, &t_count, acquire));
  175. #if !defined(_WIN32)
  176.     pthread_attr_t *attr = &pqc->dpq_thread_attr;
  177.     pthread_t tid, *pthr = &tid;
  178. #if DISPATCH_USE_MGR_THREAD && DISPATCH_USE_PTHREAD_ROOT_QUEUES
  179.     if (unlikely(dq == &_dispatch_mgr_root_queue)) {
  180.         pthr = _dispatch_mgr_root_queue_init();
  181.     }
  182. #endif
  183.     do {
  184.         _dispatch_retain(dq); // released in _dispatch_worker_thread
  185.         while ((r = pthread_create(pthr, attr, _dispatch_worker_thread, dq))) {
  186.             if (r != EAGAIN) {
  187.                 (void)dispatch_assume_zero(r);
  188.             }
  189.             _dispatch_temporary_resource_shortage();
  190.         }
  191.     } while (--remaining);
  192. #else // defined(_WIN32)
  193. #if DISPATCH_USE_MGR_THREAD && DISPATCH_USE_PTHREAD_ROOT_QUEUES
  194.     if (unlikely(dq == &_dispatch_mgr_root_queue)) {
  195.         _dispatch_mgr_root_queue_init();
  196.     }
  197. #endif
  198.     do {
  199.         _dispatch_retain(dq); // released in _dispatch_worker_thread
  200. #if DISPATCH_DEBUG
  201.         unsigned dwStackSize = 0;
  202. #else
  203.         unsigned dwStackSize = 64 * 1024;
  204. #endif
  205.         uintptr_t hThread = 0;
  206.         while (!(hThread = _beginthreadex(NULL, dwStackSize, _dispatch_worker_thread_thunk, dq, STACK_SIZE_PARAM_IS_A_RESERVATION, NULL))) {
  207.             if (errno != EAGAIN) {
  208.                 (void)dispatch_assume(hThread);
  209.             }
  210.             _dispatch_temporary_resource_shortage();
  211.         }
  212.         if (_dispatch_mgr_sched.prio > _dispatch_mgr_sched.default_prio) {
  213.             (void)dispatch_assume_zero(SetThreadPriority((HANDLE)hThread, _dispatch_mgr_sched.prio) == TRUE);
  214.         }
  215.         CloseHandle((HANDLE)hThread);
  216.     } while (--remaining);
  217. #endif // defined(_WIN32)
  218. #else
  219.     (void)floor;
  220. #endif // DISPATCH_USE_PTHREAD_POOL
  221. }

到了这里可以清楚的看到对于全局队列使用 _pthread_workqueue_addthreads 开辟线程，对于其他队列使用 pthread_create 开辟新的线程。那么任务执行的代码为什么没看到？其实 _dispatch_root_queues_init 中会首先执行第一个任务：

  1. static inline void
  2. _dispatch_root_queues_init(void)
  3. {
  4.     dispatch_once_f(&_dispatch_root_queues_pred, NULL,
  5.             _dispatch_root_queues_init_once);
  6. }
  8. // 看一下 dispatch_once_f 就不展开了，可以看一下下面 dispatch_once 的分析，这里看一下 _dispatch_root_queues_init_once
  10. static void
  11. _dispatch_root_queues_init_once(void *context DISPATCH_UNUSED)
  12. {
  13.     _dispatch_fork_becomes_unsafe();
  14. #if DISPATCH_USE_INTERNAL_WORKQUEUE
  15.     size_t i;
  16.     for (i = 0; i < DISPATCH_ROOT_QUEUE_COUNT; i++) {
  17.         _dispatch_root_queue_init_pthread_pool(&_dispatch_root_queues[i], 0,
  18.                 _dispatch_root_queues[i].dq_priority);
  19.     }
  20. #else
  21.     int wq_supported = _pthread_workqueue_supported();
  22.     int r = ENOTSUP;
  24.     if (!(wq_supported & WORKQ_FEATURE_MAINTENANCE)) {
  25.         DISPATCH_INTERNAL_CRASH(wq_supported,
  26.                 "QoS Maintenance support required");
  27.     }
  29. #if DISPATCH_USE_KEVENT_SETUP
  30.     struct pthread_workqueue_config cfg = {
  31.         .version = PTHREAD_WORKQUEUE_CONFIG_VERSION,
  32.         .flags = 0,
  33.         .workq_cb = 0,
  34.         .kevent_cb = 0,
  35.         .workloop_cb = 0,
  36.         .queue_serialno_offs = dispatch_queue_offsets.dqo_serialnum,
  37. #if PTHREAD_WORKQUEUE_CONFIG_VERSION >= 2
  38.         .queue_label_offs = dispatch_queue_offsets.dqo_label,
  39. #endif
  40.     };
  41. #endif
  43. #pragma clang diagnostic push
  44. #pragma clang diagnostic ignored "-Wunreachable-code"
  45.     if (unlikely(!_dispatch_kevent_workqueue_enabled)) {
  46. #if DISPATCH_USE_KEVENT_SETUP
  47.         cfg.workq_cb = _dispatch_worker_thread2;
  48.         r = pthread_workqueue_setup(&cfg, sizeof(cfg));
  49. #else
  50.         r = _pthread_workqueue_init(_dispatch_worker_thread2,
  51.                 offsetof(struct dispatch_queue_s, dq_serialnum), 0);
  52. #endif // DISPATCH_USE_KEVENT_SETUP
  53. #if DISPATCH_USE_KEVENT_WORKLOOP
  54.     } else if (wq_supported & WORKQ_FEATURE_WORKLOOP) {
  55. #if DISPATCH_USE_KEVENT_SETUP
  56.         cfg.workq_cb = _dispatch_worker_thread2;
  57.         cfg.kevent_cb = (pthread_workqueue_function_kevent_t) _dispatch_kevent_worker_thread;
  58.         cfg.workloop_cb = (pthread_workqueue_function_workloop_t) _dispatch_workloop_worker_thread;
  59.         r = pthread_workqueue_setup(&cfg, sizeof(cfg));
  60. #else
  61.         r = _pthread_workqueue_init_with_workloop(_dispatch_worker_thread2,
  62.                 (pthread_workqueue_function_kevent_t)
  63.                 _dispatch_kevent_worker_thread,
  64.                 (pthread_workqueue_function_workloop_t)
  65.                 _dispatch_workloop_worker_thread,
  66.                 offsetof(struct dispatch_queue_s, dq_serialnum), 0);
  67. #endif // DISPATCH_USE_KEVENT_SETUP
  68. #endif // DISPATCH_USE_KEVENT_WORKLOOP
  69. #if DISPATCH_USE_KEVENT_WORKQUEUE
  70.     } else if (wq_supported & WORKQ_FEATURE_KEVENT) {
  71. #if DISPATCH_USE_KEVENT_SETUP
  72.         cfg.workq_cb = _dispatch_worker_thread2;
  73.         cfg.kevent_cb = (pthread_workqueue_function_kevent_t) _dispatch_kevent_worker_thread;
  74.         r = pthread_workqueue_setup(&cfg, sizeof(cfg));
  75. #else
  76.         r = _pthread_workqueue_init_with_kevent(_dispatch_worker_thread2,
  77.                 (pthread_workqueue_function_kevent_t)
  78.                 _dispatch_kevent_worker_thread,
  79.                 offsetof(struct dispatch_queue_s, dq_serialnum), 0);
  80. #endif // DISPATCH_USE_KEVENT_SETUP
  81. #endif
  82.     } else {
  83.         DISPATCH_INTERNAL_CRASH(wq_supported, "Missing Kevent WORKQ support");
  84.     }
  85. #pragma clang diagnostic pop
  87.     if (r != 0) {
  88.         DISPATCH_INTERNAL_CRASH((r << 16) | wq_supported,
  89.                 "Root queue initialization failed");
  90.     }
  91. #endif // DISPATCH_USE_INTERNAL_WORKQUEUE
  92. }
  94. // 继续查看 _dispatch_workloop_worker_thread
  96. static void
  97. _dispatch_workloop_worker_thread(uint64_t *workloop_id,
  98.         dispatch_kevent_t *events, int *nevents)
  99. {
  100.     if (!dispatch_assume(workloop_id && events && nevents)) {
  101.         return;
  102.     }
  103.     if (!dispatch_assume(*workloop_id != 0)) {
  104.         return _dispatch_kevent_worker_thread(events, nevents);
  105.     }
  106.     if (*nevents == 0 || *events == NULL) {
  107.         // events for worker thread request have already been delivered earlier
  108.         // or got cancelled before point of no return concurrently
  109.         return;
  110.     }
  111.     dispatch_wlh_t wlh = (dispatch_wlh_t)*workloop_id;
  112.     _dispatch_adopt_wlh(wlh);
  113.     _dispatch_wlh_worker_thread(wlh, *events, nevents);
  114.     _dispatch_preserve_wlh_storage_reference(wlh);
  115. }
  117. // 查看 _dispatch_worker_thread2
  119. static void
  120. _dispatch_worker_thread2(pthread_priority_t pp)
  121. {
  122.     bool overcommit = pp & _PTHREAD_PRIORITY_OVERCOMMIT_FLAG;
  123.     dispatch_queue_global_t dq;
  125.     pp &= _PTHREAD_PRIORITY_OVERCOMMIT_FLAG | ~_PTHREAD_PRIORITY_FLAGS_MASK;
  126.     _dispatch_thread_setspecific(dispatch_priority_key, (void *)(uintptr_t)pp);
  127.     dq = _dispatch_get_root_queue(_dispatch_qos_from_pp(pp), overcommit);
  129.     _dispatch_introspection_thread_add();
  130.     _dispatch_trace_runtime_event(worker_unpark, dq, 0);
  132.     int pending = os_atomic_dec2o(dq, dgq_pending, relaxed);
  133.     dispatch_assert(pending >= 0);
  134.     _dispatch_root_queue_drain(dq, dq->dq_priority,
  135.             DISPATCH_INVOKE_WORKER_DRAIN | DISPATCH_INVOKE_REDIRECTING_DRAIN);
  136.     _dispatch_voucher_debug("root queue clear", NULL);
  137.     _dispatch_reset_voucher(NULL, DISPATCH_THREAD_PARK);
  138.     _dispatch_trace_runtime_event(worker_park, NULL, 0);
  139. }
  141. // 查看  _dispatch_root_queue_drain
  143. static void
  144. _dispatch_root_queue_drain(dispatch_queue_global_t dq,
  145.         dispatch_priority_t pri, dispatch_invoke_flags_t flags)
  146. {
  147. #if DISPATCH_DEBUG
  148.     dispatch_queue_t cq;
  149.     if (unlikely(cq = _dispatch_queue_get_current())) {
  150.         DISPATCH_INTERNAL_CRASH(cq, "Premature thread recycling");
  151.     }
  152. #endif
  153.     _dispatch_queue_set_current(dq);
  154.     _dispatch_init_basepri(pri);
  155.     _dispatch_adopt_wlh_anon();
  157.     struct dispatch_object_s *item;
  158.     bool reset = false;
  159.     dispatch_invoke_context_s dic = { };
  160. #if DISPATCH_COCOA_COMPAT
  161.     _dispatch_last_resort_autorelease_pool_push(&dic);
  162. #endif // DISPATCH_COCOA_COMPAT
  163.     _dispatch_queue_drain_init_narrowing_check_deadline(&dic, pri);
  164.     _dispatch_perfmon_start();
  165.     while (likely(item = _dispatch_root_queue_drain_one(dq))) {
  166.         if (reset) _dispatch_wqthread_override_reset();
  167.         _dispatch_continuation_pop_inline(item, &dic, flags, dq);
  168.         reset = _dispatch_reset_basepri_override();
  169.         if (unlikely(_dispatch_queue_drain_should_narrow(&dic))) {
  170.             break;
  171.         }
  172.     }
  174.     // overcommit or not. worker thread
  175.     if (pri & DISPATCH_PRIORITY_FLAG_OVERCOMMIT) {
  176.         _dispatch_perfmon_end(perfmon_thread_worker_oc);
  177.     } else {
  178.         _dispatch_perfmon_end(perfmon_thread_worker_non_oc);
  179.     }
  181. #if DISPATCH_COCOA_COMPAT
  182.     _dispatch_last_resort_autorelease_pool_pop(&dic);
  183. #endif // DISPATCH_COCOA_COMPAT
  184.     _dispatch_reset_wlh();
  185.     _dispatch_clear_basepri();
  186.     _dispatch_queue_set_current(NULL);
  187. }
  189. // 查看 _dispatch_continuation_pop_inline 这个是出队列操作，这里注意一下首先看了有没有 vtable（_dispatch_object_has_vtable），这里解释了为什么 dispatch_barrier_async 尽管主要流程和 dispatch_async 一模一样但是无法应用到全局队列的原因，因为全局队列没有 v_table 结构会直接像dispatch_async 一样执行
  191. static inline void
  192. _dispatch_continuation_pop_inline(dispatch_object_t dou,
  193.         dispatch_invoke_context_t dic, dispatch_invoke_flags_t flags,
  194.         dispatch_queue_class_t dqu)
  195. {
  196.     dispatch_pthread_root_queue_observer_hooks_t observer_hooks =
  197.             _dispatch_get_pthread_root_queue_observer_hooks();
  198.     if (observer_hooks) observer_hooks->queue_will_execute(dqu._dq);
  199.     flags &= _DISPATCH_INVOKE_PROPAGATE_MASK;
  200.     if (_dispatch_object_has_vtable(dou)) {
  201.         dx_invoke(dou._dq, dic, flags);
  202.     } else {
  203.         _dispatch_continuation_invoke_inline(dou, flags, dqu);
  204.     }
  205.     if (observer_hooks) observer_hooks->queue_did_execute(dqu._dq);
  206. }
  208. // 查看 _dispatch_continuation_invoke_inline，这里_dispatch_client_callout 就是真正的执行 block 操作 ，当然还有一种情况这里还不会走就是 _dispatch_continuation_with_group_invoke，这个后面的 dispatch_group 会用到
  210. static inline void
  211. _dispatch_continuation_invoke_inline(dispatch_object_t dou,
  212.         dispatch_invoke_flags_t flags, dispatch_queue_class_t dqu)
  213. {
  214.     dispatch_continuation_t dc = dou._dc, dc1;
  215.     dispatch_invoke_with_autoreleasepool(flags, {
  216.         uintptr_t dc_flags = dc->dc_flags;
  217.         // Add the item back to the cache before calling the function. This
  218.         // allows the 'hot' continuation to be used for a quick callback.
  219.         //
  220.         // The ccache version is per-thread.
  221.         // Therefore, the object has not been reused yet.
  222.         // This generates better assembly.
  223.         _dispatch_continuation_voucher_adopt(dc, dc_flags);
  224.         if (!(dc_flags & DC_FLAG_NO_INTROSPECTION)) {
  225.             _dispatch_trace_item_pop(dqu, dou);
  226.         }
  227.         if (dc_flags & DC_FLAG_CONSUME) {
  228.             dc1 = _dispatch_continuation_free_cacheonly(dc);
  229.         } else {
  230.             dc1 = NULL;
  231.         }
  232.         if (unlikely(dc_flags & DC_FLAG_GROUP_ASYNC)) {
  233.             _dispatch_continuation_with_group_invoke(dc);
  234.         } else {
  235.             _dispatch_client_callout(dc->dc_ctxt, dc->dc_func);
  236.             _dispatch_trace_item_complete(dc);
  237.         }
  238.         if (unlikely(dc1)) {
  239.             _dispatch_continuation_free_to_cache_limit(dc1);
  240.         }
  241.     });
  242.     _dispatch_perfmon_workitem_inc();
  243. }

另外对于 _dispatch_continuation_init 的代码中的并没有对其进行展开，其实 _dispatch_continuation_init 中的 func 就是 _dispatch_call_block_and_release(源码如下)，它在 dx_push 调用时包装进了 qos。

  1. void
  2. _dispatch_call_block_and_release(void *block)
  3. {
  4.     void (^b)(void) = block;
  5.     b();
  6.     Block_release(b);
  7. }

dispatch_async 代码实现看起来比较复杂，因为其中的数据结构较多，分支流程控制比较复杂。不过思路其实很简单，用链表保存所有提交的 block（先进先出，在队列本身维护了一个链表新加入 block 放到链表尾部），然后在底层线程池中，依次取出 block 并执行。

类似的可以看到 dispatch_barrier_async 源码和 dispatch_async 几乎一致，仅仅多了一个标记位 DC_FLAG_BARRIER，这个标记位用于在取出任务时进行判断，正常的异步调用会依次取出，而如果遇到了 DC_FLAG_BARRIER 则会返回，所以可以等待所有任务执行结束执行 dx_push（不过提醒一下dispatch_barrier_async 必须在自定义队列才有用，原因是 global 队列没有 v_table 结构，同时不要试图在主队列调用，否则会 crash)：

  1. void
  2. dispatch_barrier_async(dispatch_queue_t dq, dispatch_block_t work)
  3. {
  4.     dispatch_continuation_t dc = _dispatch_continuation_alloc();
  5.     uintptr_t dc_flags = DC_FLAG_CONSUME | DC_FLAG_BARRIER;
  6.     dispatch_qos_t qos;
  8.     qos = _dispatch_continuation_init(dc, dq, work, 0, dc_flags);
  9.     _dispatch_continuation_async(dq, dc, qos, dc_flags);
  10. }

3、单次执行 dispatch_once

下面的代码在objc开发中应该很常见，这种方式可以保证instance只会创建一次：

  1. + (instancetype)sharedInstance {
  2.     static MyClass *instance;
  3.     static dispatch_once_t onceToken;
  4.     dispatch_once(&onceToken, ^{
  5.         instance = [[MyClass alloc] init];
  6.     });
  7.     return instance;
  8. }

我们不妨分析一下 dispatch_once 的源码：

  1. void
  2. dispatch_once(dispatch_once_t *val, dispatch_block_t block)
  3. {
  4.     dispatch_once_f(val, block, _dispatch_Block_invoke(block));
  5. }
  7. // 展开 dispatch_once_f
  9. void
  10. dispatch_once_f(dispatch_once_t *val, void *ctxt, dispatch_function_t func)
  11. {
  12.     dispatch_once_gate_t l = (dispatch_once_gate_t)val;
  14. #if !DISPATCH_ONCE_INLINE_FASTPATH || DISPATCH_ONCE_USE_QUIESCENT_COUNTER
  15.     uintptr_t v = os_atomic_load(&l->dgo_once, acquire);
  16.     if (likely(v == DLOCK_ONCE_DONE)) {
  17.         return;
  18.     }
  19. #if DISPATCH_ONCE_USE_QUIESCENT_COUNTER
  20.     if (likely(DISPATCH_ONCE_IS_GEN(v))) {
  21.         return _dispatch_once_mark_done_if_quiesced(l, v);
  22.     }
  23. #endif
  24. #endif
  25.     if (_dispatch_once_gate_tryenter(l)) {
  26.         return _dispatch_once_callout(l, ctxt, func);
  27.     }
  28.     return _dispatch_once_wait(l);
  29. }
  31. // 如果 os_atomic_load 为 DLOCK_ONCE_DONE 则直接返回，否则进入_dispatch_once_gate_tryenter，在这里首先判断对象是否存储过，如果存储过则则标记为 unlock
  33. static inline bool
  34. _dispatch_once_gate_tryenter(dispatch_once_gate_t l)
  35. {
  36.     return os_atomic_cmpxchg(&l->dgo_once, DLOCK_ONCE_UNLOCKED,
  37.             (uintptr_t)_dispatch_lock_value_for_self(), relaxed);
  38. }
  40. // 如果没有存储过则执行 _dispatch_once_callout，主要是执行 block
  42. static void
  43. _dispatch_once_callout(dispatch_once_gate_t l, void *ctxt,
  44.         dispatch_function_t func)
  45. {
  46.     _dispatch_client_callout(ctxt, func);
  47.     _dispatch_once_gate_broadcast(l);
  48. }
  50. // 执行过 block 则调用 _dispatch_once_gate_broadcast
  52. static inline void
  53. _dispatch_once_gate_broadcast(dispatch_once_gate_t l)
  54. {
  55.     dispatch_lock value_self = _dispatch_lock_value_for_self();
  56.     uintptr_t v;
  57. #if DISPATCH_ONCE_USE_QUIESCENT_COUNTER
  58.     v = _dispatch_once_mark_quiescing(l);
  59. #else
  60.     v = _dispatch_once_mark_done(l);
  61. #endif
  62.     if (likely((dispatch_lock)v == value_self)) return;
  63.     _dispatch_gate_broadcast_slow(&l->dgo_gate, (dispatch_lock)v);
  64. }
  66. // 在 _dispatch_once_gate_broadcast 中由于执行完毕，使用_dispatch_once_mark_done 标记为 done
  68. static inline uintptr_t
  69. _dispatch_once_mark_done(dispatch_once_gate_t dgo)
  70. {
  71.     return os_atomic_xchg(&dgo->dgo_once, DLOCK_ONCE_DONE, release);
  72. }

4、信号量 dispatch_semaphore

信号量是线程同步操作中很常用的一个操作，常用的几个类型：

  • dispatch_semaphore_t：信号量类型
  • dispatch_semaphore_create：创建一个信号量
  • dispatch_semaphore_wait：发送一个等待信号，信号量-1，当信号量为0阻塞线程，大于0则开始执行后面的逻辑（也就是说执行dispatch_semaphore_wait 前如果信号量 <=0 则阻塞，否则正常执行后面的逻辑）
  • dispatch_semaphore_signal：发送唤醒信号，信号量会+1

比如我们有个操作 run() 在异步线程已经开始执行，同时可能用户会手动再次触发动作 watch()，但是 watch 依赖 run 完成则可以使用信号量:

  1. - (void)run {
  2.     dispatch_semaphore_t semaphore = dispatch_semaphore_create(0);
  3.     dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{
  5.         // 这里执行其他任务。。。
  6.         // TODO:
  8.         // 执行完发送信号
  9.         dispatch_semaphore_signal(semaphore);
  10.     });
  12.     self->semaphore = semaphore;
  13. }
  15. - (void)watch {
  16.     // 等待上面的操作完成，如果60s还没有完成则超时继续执行下面的逻辑
  17.     dispatch_semaphore_wait(self.semaphore, dispatch_time(DISPATCH_TIME_NOW, 60*NSEC_PER_SEC));
  19.     // 这里执行其他任务。。。但是依赖上面的操作完成
  20.     // TODO:
  21. }

那么信号量是如何实现的呢，不妨看一下它的源码：

  1. // 首先看一下 dispatch_semaphore_t，没错和上面一样本质就是 dispatch_semaphore_s，dsema_value 代表当前信号量，dsema_orig 表示初始信号量
  3. DISPATCH_CLASS_DECL(semaphore, OBJECT);
  4. struct dispatch_semaphore_s {
  5.     DISPATCH_OBJECT_HEADER(semaphore);
  6.     long volatile dsema_value;
  7.     long dsema_orig;
  8.     _dispatch_sema4_t dsema_sema;
  9. };
  11. // 查看 dispatch_semaphore_create 源码，其实并不复杂创建分配 DISPATCH_VTABLE 结构的空间，设置初始信号量，但是可以清楚的看到同样指定了目标队列，这是一个优先级为 DISPATCH_QUEUE_PRIORITY_DEFAULT 的非过载队列
  13. dispatch_semaphore_t
  14. dispatch_semaphore_create(long value)
  15. {
  16.     dispatch_semaphore_t dsema;
  18.     // If the internal value is negative, then the absolute of the value is
  19.     // equal to the number of waiting threads. Therefore it is bogus to
  20.     // initialize the semaphore with a negative value.
  21.     if (value < 0) {
  22.         return DISPATCH_BAD_INPUT;
  23.     }
  25.     dsema = _dispatch_object_alloc(DISPATCH_VTABLE(semaphore),
  26.             sizeof(struct dispatch_semaphore_s));
  27.     dsema->do_next = DISPATCH_OBJECT_LISTLESS;
  28.     dsema->do_targetq = _dispatch_get_default_queue(false);
  29.     dsema->dsema_value = value;
  30.     _dispatch_sema4_init(&dsema->dsema_sema, _DSEMA4_POLICY_FIFO);
  31.     dsema->dsema_orig = value;
  32.     return dsema;
  33. }
  35. // 下面看一下 dispatch_semaphore_wait，首先 os_atomic_dec2o 信号量减一，当然递减之后信号量大于等于0它其实什么也不做继续执行就好了，但是如果不满足执行 _dispatch_semaphore_wait_slow 等待信号量唤醒或者 timeout 超时
  37. long
  38. dispatch_semaphore_wait(dispatch_semaphore_t dsema, dispatch_time_t timeout)
  39. {
  40.     long value = os_atomic_dec2o(dsema, dsema_value, acquire);
  41.     if (likely(value >= 0)) {
  42.         return 0;
  43.     }
  44.     return _dispatch_semaphore_wait_slow(dsema, timeout);
  45. }
  47. // 看一下 _dispatch_semaphore_wait_slow 源码，这里首先对于两种极端情况：如果是 DISPATCH_TIME_NOW 则执行信号量 +1 并返回超时信号，DISPATCH_TIME_FOREVER 则一直等待，默认则调用 _dispatch_sema4_timedwait
  49. static long
  50. _dispatch_semaphore_wait_slow(dispatch_semaphore_t dsema,
  51.         dispatch_time_t timeout)
  52. {
  53.     long orig;
  55.     _dispatch_sema4_create(&dsema->dsema_sema, _DSEMA4_POLICY_FIFO);
  56.     switch (timeout) {
  57.     default:
  58.         if (!_dispatch_sema4_timedwait(&dsema->dsema_sema, timeout)) {
  59.             break;
  60.         }
  61.         // Fall through and try to undo what the fast path did to
  62.         // dsema->dsema_value
  63.     case DISPATCH_TIME_NOW:
  64.         orig = dsema->dsema_value;
  65.         while (orig < 0) {
  66.             if (os_atomic_cmpxchgvw2o(dsema, dsema_value, orig, orig + 1,
  67.                     &orig, relaxed)) {
  68.                 return _DSEMA4_TIMEOUT();
  69.             }
  70.         }
  71.         // Another thread called semaphore_signal().
  72.         // Fall through and drain the wakeup.
  73.     case DISPATCH_TIME_FOREVER:
  74.         _dispatch_sema4_wait(&dsema->dsema_sema);
  75.         break;
  76.     }
  77.     return 0;
  78. }
  80. // 查看 _dispatch_sema4_timedwait 调用 mach 的内核函数semaphore_timedwait 等待收到信号直至超时
  82. bool
  83. _dispatch_sema4_timedwait(_dispatch_sema4_t *sema, dispatch_time_t timeout)
  84. {
  85.     mach_timespec_t _timeout;
  86.     kern_return_t kr;
  88.     do {
  89.         uint64_t nsec = _dispatch_timeout(timeout);
  90.         _timeout.tv_sec = (__typeof__(_timeout.tv_sec))(nsec / NSEC_PER_SEC);
  91.         _timeout.tv_nsec = (__typeof__(_timeout.tv_nsec))(nsec % NSEC_PER_SEC);
  92.         kr = semaphore_timedwait(*sema, _timeout);
  93.     } while (unlikely(kr == KERN_ABORTED));
  95.     if (kr == KERN_OPERATION_TIMED_OUT) {
  96.         return true;
  97.     }
  98.     DISPATCH_SEMAPHORE_VERIFY_KR(kr);
  99.     return false;
  100. }
  102. // 最后看一下 dispatch_semaphore_signal，首先信号量 +1，如果信号量大于 0 就什么也不做（通常到了这里 dispatch_semaphore_wait 还没调用），否则执行 _dispatch_semaphore_signal_slow
  104. long
  105. dispatch_semaphore_signal(dispatch_semaphore_t dsema)
  106. {
  107.     long value = os_atomic_inc2o(dsema, dsema_value, release);
  108.     if (likely(value > 0)) {
  109.         return 0;
  110.     }
  111.     if (unlikely(value == LONG_MIN)) {
  112.         DISPATCH_CLIENT_CRASH(value,
  113.                 "Unbalanced call to dispatch_semaphore_signal()");
  114.     }
  115.     return _dispatch_semaphore_signal_slow(dsema);
  116. }
  118. // 查看 _dispatch_semaphore_signal_slow，调用内核 semaphore_signal 唤醒线程，如 Apple Api 描述 “如果唤醒线程则返回非0，否则返回0”。
  120. long
  121. _dispatch_semaphore_signal_slow(dispatch_semaphore_t dsema)
  122. {
  123.     _dispatch_sema4_create(&dsema->dsema_sema, _DSEMA4_POLICY_FIFO);
  124.     _dispatch_sema4_signal(&dsema->dsema_sema, 1);
  125.     return 1;
  126. }
  128. // 查看 _dispatch_sema4_signal 源码 
  130. void
  131. _dispatch_sema4_signal(_dispatch_sema4_t *sema, long count)
  132. {
  133.     do {
  134.         int ret = sem_post(sema);
  135.         DISPATCH_SEMAPHORE_VERIFY_RET(ret);
  136.     } while (--count);
  137. }

信号量是一个比较重要的内容，合理使用可以让你的程序更加的优雅，比如说一个常见的情况：大家知道 PHImageManager.requestImage 是一个释放消耗内存的方法，有时我们需要批量获取到图片执行一些操作的话可能就没办法直接for循环，不然内存会很快爆掉，因为每个 requestImage 操作都需要占用大量内存，即使外部嵌套 autoreleasepool 也不一定可以及时释放（想想 for 执行的速度，释放肯定来不及），那么 requestImage 又是一个异步操作，如此只能让一个操作执行完再执行另一个循环操作才能解决。也就是说这个问题就变成 for 循环内部的异步操作串行执行的问题。要解决这个问题有几种思路： 1.使用 requestImage 的同步请求照片 2.使用递归操作一个操作执行完再执行另外一个操作移除 for 操作 3.使用信号量解决。 当然第一个方法并非普适，有些异步操作并不能轻易改成同步操作，第二个方法相对普适，但是递归调用本身因为要改变原来的代码结构看起来不是那么优雅，自然当前讨论的信号量是更好的方式。我们假设 requestImage 是一个 watch(callback:((_ image)-> Void)) 操作,整个请求是一个 run(callback:((_ images)->Void)) 那么它的实现方式如下:

  1. - (void)run:(CallbackWithImages)callback {
  2.    dispatch_queue_t globalQueue = dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0);
  3.    dispatch_semaphore_t semaphore = dispatch_semaphore_create(0);
  4.    dispatch_async(globalQueue, ^{
  5.        NSMutableArray *array = [[NSMutableArray alloc] init];
  6.        for (int i=0; i<100; ++i) {
  7.            [self watch:^(UIImage *image){
  8.                [array addObject:image];
  9.                dispatch_semaphore_signal(semaphore);
  10.            }];
  11.            dispatch_semaphore_wait(semaphore, DISPATCH_TIME_FOREVER);
  12.        }
  13.        dispatch_async(dispatch_get_main_queue(), ^{
  14.            callback([array copy]);
  15.        });
  16.    });
  17. }
  19. - (void)watch:(CallbackWithImage)callback {
  20.    dispatch_queue_t globalQueue = dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0);
  21.    dispatch_async(globalQueue, ^{
  22.        callback([UIImage new]);
  23.    });
  24. }

5、调度组 dispatch_group

dispatch_group 常常用来同步多个任务（注意和 dispatch_barrier_sync 不同的是它可以是多个队列的同步），所以其实上面先分析 dispatch_semaphore 也是这个原因，它本身是依靠信号量来完成的同步管理。典型的用法如下：

  1. dispatch_queue_t globalQueue = dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0);
  2. dispatch_group_t group = dispatch_group_create();
  4. dispatch_group_async(group, globalQueue, ^{
  5.     sleep(10);
  6.     NSLog(@"任务1完成");
  7. });
  9. dispatch_group_async(group, globalQueue, ^{
  10.     NSLog(@"任务2完成");
  11. });
  13. dispatch_group_notify(group, globalQueue, ^{
  14.     NSLog(@"两个任务全部完成");
  15. });
  17. dispatch_async(globalQueue, ^{
  18.     // 等待5s超时后继续执行，此时dispatch_group中的任务未必全部完成，注意：dispatch_group_wait是同步操作必须放到异步队列否则阻塞当前线程
  19.     dispatch_group_wait(group, dispatch_time(DISPATCH_TIME_NOW, 5*NSEC_PER_SEC));
  20.     NSLog(@"等待到了上限，开始执行。。。");
  21. });

下面看一下 dispatch_group 相关的源码：

  1. // 和其他对象一样，dispatch_group_t 的本质就是 dispatch_group_s 指针，这里重点关注一下 dg_state 和 dg_bits 是一个计数器
  3. struct dispatch_group_s {
  4.     DISPATCH_OBJECT_HEADER(group);
  5.     DISPATCH_UNION_LE(uint64_t volatile dg_state,
  6.             uint32_t dg_bits,
  7.             uint32_t dg_gen
  8.     ) DISPATCH_ATOMIC64_ALIGN;
  9.     struct dispatch_continuation_s *volatile dg_notify_head;
  10.     struct dispatch_continuation_s *volatile dg_notify_tail;
  11. };
  13. // 查看 dispatch_group_create
  15. dispatch_group_t
  16. dispatch_group_create(void)
  17. {
  18.     return _dispatch_group_create_with_count(0);
  19. }
  21. // 展开 _dispatch_group_create_with_count，其实就是一个dispatch_group_s 对象，指定了 do_targetq 是默认队列并且不支持过载
  23. static inline dispatch_group_t
  24. _dispatch_group_create_with_count(uint32_t n)
  25. {
  26.     dispatch_group_t dg = _dispatch_object_alloc(DISPATCH_VTABLE(group),
  27.             sizeof(struct dispatch_group_s));
  28.     dg->do_next = DISPATCH_OBJECT_LISTLESS;
  29.     dg->do_targetq = _dispatch_get_default_queue(false);
  30.     if (n) {
  31.         os_atomic_store2o(dg, dg_bits,
  32.                 (uint32_t)-n * DISPATCH_GROUP_VALUE_INTERVAL, relaxed);
  33.         os_atomic_store2o(dg, do_ref_cnt, 1, relaxed); // <rdar://22318411>
  34.     }
  35.     return dg;
  36. }
  38. // 首先看一下 dispatch_group_enter，它的核心就是 os_atomic_sub_orig2o 对 dg_bits 进行 -1 操作
  40. void
  41. dispatch_group_enter(dispatch_group_t dg)
  42. {
  43.     // The value is decremented on a 32bits wide atomic so that the carry
  44.     // for the 0 -> -1 transition is not propagated to the upper 32bits.
  45.     uint32_t old_bits = os_atomic_sub_orig2o(dg, dg_bits,
  46.             DISPATCH_GROUP_VALUE_INTERVAL, acquire);
  47.     uint32_t old_value = old_bits & DISPATCH_GROUP_VALUE_MASK;
  48.     if (unlikely(old_value == 0)) {
  49.         _dispatch_retain(dg); // <rdar://problem/22318411>
  50.     }
  51.     if (unlikely(old_value == DISPATCH_GROUP_VALUE_MAX)) {
  52.         DISPATCH_CLIENT_CRASH(old_bits,
  53.                 "Too many nested calls to dispatch_group_enter()");
  54.     }
  55. }
  57. // 然后看一下 dispatch_group_leave，核心就是 os_atomic_add_orig2o 执行 dg_state+1 操作，如果 +1 之后还等于 0 那么说明之前没有调用dispatch_group_enter，就里会 crash，当然这里核心在 _dispatch_group_wake
  59. void
  60. dispatch_group_leave(dispatch_group_t dg)
  61. {
  62.     // The value is incremented on a 64bits wide atomic so that the carry for
  63.     // the -1 -> 0 transition increments the generation atomically.
  64.     uint64_t new_state, old_state = os_atomic_add_orig2o(dg, dg_state,
  65.             DISPATCH_GROUP_VALUE_INTERVAL, release);
  66.     uint32_t old_value = (uint32_t)(old_state & DISPATCH_GROUP_VALUE_MASK);
  68.     if (unlikely(old_value == DISPATCH_GROUP_VALUE_1)) {
  69.         old_state += DISPATCH_GROUP_VALUE_INTERVAL;
  70.         do {
  71.             new_state = old_state;
  72.             if ((old_state & DISPATCH_GROUP_VALUE_MASK) == 0) {
  73.                 new_state &= ~DISPATCH_GROUP_HAS_WAITERS;
  74.                 new_state &= ~DISPATCH_GROUP_HAS_NOTIFS;
  75.             } else {
  76.                 // If the group was entered again since the atomic_add above,
  77.                 // we can't clear the waiters bit anymore as we don't know for
  78.                 // which generation the waiters are for
  79.                 new_state &= ~DISPATCH_GROUP_HAS_NOTIFS;
  80.             }
  81.             if (old_state == new_state) break;
  82.         } while (unlikely(!os_atomic_cmpxchgv2o(dg, dg_state,
  83.                 old_state, new_state, &old_state, relaxed)));
  84.         return _dispatch_group_wake(dg, old_state, true);
  85.     }
  87.     if (unlikely(old_value == 0)) {
  88.         DISPATCH_CLIENT_CRASH((uintptr_t)old_value,
  89.                 "Unbalanced call to dispatch_group_leave()");
  90.     }
  91. }
  93. // 查看 _dispatch_group_wake 源码，到了这里通常就是出了调度组，如果有 notify 等待则执行 notify 遍历并且在对应队列中执行，如果有 wait 任务则唤醒其执行任务（注意这里比较牛叉的 _dispatch_wake_by_address 可以根据地址进行函数调用，本身是调用的 WakeByAddressAll 这个系统调用）
  95. static void
  96. _dispatch_group_wake(dispatch_group_t dg, uint64_t dg_state, bool needs_release)
  97. {
  98.     uint16_t refs = needs_release ? 1 : 0; // <rdar://problem/22318411>
  100.     if (dg_state & DISPATCH_GROUP_HAS_NOTIFS) {
  101.         dispatch_continuation_t dc, next_dc, tail;
  103.         // Snapshot before anything is notified/woken <rdar://problem/8554546>
  104.         dc = os_mpsc_capture_snapshot(os_mpsc(dg, dg_notify), &tail);
  105.         do {
  106.             dispatch_queue_t dsn_queue = (dispatch_queue_t)dc->dc_data;
  107.             next_dc = os_mpsc_pop_snapshot_head(dc, tail, do_next);
  108.             _dispatch_continuation_async(dsn_queue, dc,
  109.                     _dispatch_qos_from_pp(dc->dc_priority), dc->dc_flags);
  110.             _dispatch_release(dsn_queue);
  111.         } while ((dc = next_dc));
  113.         refs++;
  114.     }
  116.     if (dg_state & DISPATCH_GROUP_HAS_WAITERS) {
  117.         _dispatch_wake_by_address(&dg->dg_gen);
  118.     }
  120.     if (refs) _dispatch_release_n(dg, refs);
  121. }
  123. // 查看 dispatch_group_async 源码，dispatch_continuation_t 仅仅是封装任务，核心是 _dispatch_continuation_group_async
  125. void
  126. dispatch_group_async(dispatch_group_t dg, dispatch_queue_t dq,
  127.         dispatch_block_t db)
  128. {
  129.     dispatch_continuation_t dc = _dispatch_continuation_alloc();
  130.     uintptr_t dc_flags = DC_FLAG_CONSUME | DC_FLAG_GROUP_ASYNC;
  131.     dispatch_qos_t qos;
  133.     qos = _dispatch_continuation_init(dc, dq, db, 0, dc_flags);
  134.     _dispatch_continuation_group_async(dg, dq, dc, qos);
  135. }
  137. // 展开 _dispatch_continuation_group_async，这里重点记住  dispatch_group_enter() 方法，至于 _dispatch_continuation_async 前面已经介绍过
  139. static inline void
  140. _dispatch_continuation_group_async(dispatch_group_t dg, dispatch_queue_t dq,
  141.         dispatch_continuation_t dc, dispatch_qos_t qos)
  142. {
  143.     dispatch_group_enter(dg);
  144.     dc->dc_data = dg;
  145.     _dispatch_continuation_async(dq, dc, qos, dc->dc_flags);
  146. }
  148. // 继续查看 dispatch_group_notify，注意这里将 dispatch_block_t 存储到了共享数据 dispatch_continuation_t 中
  150. void
  151. dispatch_group_notify(dispatch_group_t dg, dispatch_queue_t dq,
  152.         dispatch_block_t db)
  153. {
  154.     dispatch_continuation_t dsn = _dispatch_continuation_alloc();
  155.     _dispatch_continuation_init(dsn, dq, db, 0, DC_FLAG_CONSUME);
  156.     _dispatch_group_notify(dg, dq, dsn);
  157. }
  159. // 展开 _dispatch_group_notify，其实最主要的方法就是 _dispatch_group_wake 
  161. static inline void
  162. _dispatch_group_notify(dispatch_group_t dg, dispatch_queue_t dq,
  163.         dispatch_continuation_t dsn)
  164. {
  165.     uint64_t old_state, new_state;
  166.     dispatch_continuation_t prev;
  168.     dsn->dc_data = dq;
  169.     _dispatch_retain(dq);
  171.     prev = os_mpsc_push_update_tail(os_mpsc(dg, dg_notify), dsn, do_next);
  172.     if (os_mpsc_push_was_empty(prev)) _dispatch_retain(dg);
  173.     os_mpsc_push_update_prev(os_mpsc(dg, dg_notify), prev, dsn, do_next);
  174.     if (os_mpsc_push_was_empty(prev)) {
  175.         os_atomic_rmw_loop2o(dg, dg_state, old_state, new_state, release, {
  176.             new_state = old_state | DISPATCH_GROUP_HAS_NOTIFS;
  177.             if ((uint32_t)old_state == 0) {
  178.                 os_atomic_rmw_loop_give_up({
  179.                     return _dispatch_group_wake(dg, new_state, false);
  180.                 });
  181.             }
  182.         });
  183.     }
  184. }

简单的说就是 dispatch_group_async 和 dispatch_group_notify 本身就是和 dispatch_group_enter、dispatch_group_leave 没有本质区别，后者相对更加灵活。当然这里还有一个重要的操作就是dispatch_group_wait，还没有看：

  1. // os_atomic_rmw_loop2o 不断遍历，如果 (old_state & DISPATCH_GROUP_VALUE_MASK) == 0 表示执行完，直接返回 0，如果当前如果超时立即返回，其他情况调用 _dispatch_group_wait_slow
  3. long
  4. dispatch_group_wait(dispatch_group_t dg, dispatch_time_t timeout)
  5. {
  6.     uint64_t old_state, new_state;
  8.     os_atomic_rmw_loop2o(dg, dg_state, old_state, new_state, relaxed, {
  9.         if ((old_state & DISPATCH_GROUP_VALUE_MASK) == 0) {
  10.             os_atomic_rmw_loop_give_up_with_fence(acquire, return 0);
  11.         }
  12.         if (unlikely(timeout == 0)) {
  13.             os_atomic_rmw_loop_give_up(return _DSEMA4_TIMEOUT());
  14.         }
  15.         new_state = old_state | DISPATCH_GROUP_HAS_WAITERS;
  16.         if (unlikely(old_state & DISPATCH_GROUP_HAS_WAITERS)) {
  17.             os_atomic_rmw_loop_give_up(break);
  18.         }
  19.     });
  21.     return _dispatch_group_wait_slow(dg, _dg_state_gen(new_state), timeout);
  22. }
  24. // 查看 _dispatch_group_wait_slow，最终调用_dispatch_wait_on_address 直至 __ulock_wait
  26. static long
  27. _dispatch_group_wait_slow(dispatch_group_t dg, uint32_t gen,
  28.         dispatch_time_t timeout)
  29. {
  30.     for (;;) {
  31.         int rc = _dispatch_wait_on_address(&dg->dg_gen, gen, timeout, 0);
  32.         if (likely(gen != os_atomic_load2o(dg, dg_gen, acquire))) {
  33.             return 0;
  34.         }
  35.         if (rc == ETIMEDOUT) {
  36.             return _DSEMA4_TIMEOUT();
  37.         }
  38.     }
  39. }

6、延迟执行 dispatch_after

dispatch_after 也是一个常用的延迟执行的方法，比如常见的使用方法是：

  1. dispatch_after(dispatch_time(DISPATCH_TIME_NOW, (int64_t)(1.0 * NSEC_PER_SEC)), dispatch_get_main_queue(), ^{
  2.     NSLog(@"...");
  3. });

在查看 dispatch_after 源码之前先看一下另一个内容事件源 dispatch_source_t，其实 dispatch_source_t 是一个很少让开发者和 GCD 联想到一起的一个类型，它本身也有对应的创建方法 dispatch_source_create（事实上它的使用甚至可以追踪到 Runloop）。多数开发者认识 dispatch_source_t 都是通过定时器，很多文章会教你如何创建一个比较准确的定时器，比如下面的代码：

  1. dispatch_source_t timerSource = dispatch_source_create(DISPATCH_SOURCE_TYPE_TIMER, 0, 0, dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0));
  2. dispatch_source_set_timer(timerSource, dispatch_time(DISPATCH_TIME_NOW, 0), 3*NSEC_PER_SEC, 0);
  3. dispatch_source_set_event_handler(timerSource, ^{
  4.     NSLog(@"dispatch_source_t...");
  5. });
  6. dispatch_resume(timerSource);
  7. self.source = timerSource;

如果你知道上面一个定时器如何执行的那么下面看一下 dispatch_after 应该就比较容易明白了：

  1. void
  2. dispatch_after(dispatch_time_t when, dispatch_queue_t queue,
  3.         dispatch_block_t work)
  4. {
  5.     _dispatch_after(when, queue, NULL, work, true);
  6. }
  8. // 查看 _dispatch_after
  10. static inline void
  11. _dispatch_after(dispatch_time_t when, dispatch_queue_t dq,
  12.         void *ctxt, void *handler, bool block)
  13. {
  14.     dispatch_timer_source_refs_t dt;
  15.     dispatch_source_t ds;
  16.     uint64_t leeway, delta;
  18.     if (when == DISPATCH_TIME_FOREVER) {
  19. #if DISPATCH_DEBUG
  20.         DISPATCH_CLIENT_CRASH(0, "dispatch_after called with 'when' == infinity");
  21. #endif
  22.         return;
  23.     }
  25.     delta = _dispatch_timeout(when);
  26.     if (delta == 0) {
  27.         if (block) {
  28.             return dispatch_async(dq, handler);
  29.         }
  30.         return dispatch_async_f(dq, ctxt, handler);
  31.     }
  32.     leeway = delta / 10; // <rdar://problem/13447496>
  34.     if (leeway < NSEC_PER_MSEC) leeway = NSEC_PER_MSEC;
  35.     if (leeway > 60 * NSEC_PER_SEC) leeway = 60 * NSEC_PER_SEC;
  37.     // this function can and should be optimized to not use a dispatch source
  38.     ds = dispatch_source_create(&_dispatch_source_type_after, 0, 0, dq);
  39.     dt = ds->ds_timer_refs;
  41.     dispatch_continuation_t dc = _dispatch_continuation_alloc();
  42.     if (block) {
  43.         _dispatch_continuation_init(dc, dq, handler, 0, 0);
  44.     } else {
  45.         _dispatch_continuation_init_f(dc, dq, ctxt, handler, 0, 0);
  46.     }
  47.     // reference `ds` so that it doesn't show up as a leak
  48.     dc->dc_data = ds;
  49.     _dispatch_trace_item_push(dq, dc);
  50.     os_atomic_store2o(dt, ds_handler[DS_EVENT_HANDLER], dc, relaxed);
  52.     dispatch_clock_t clock;
  53.     uint64_t target;
  54.     _dispatch_time_to_clock_and_value(when, &clock, &target);
  55.     if (clock != DISPATCH_CLOCK_WALL) {
  56.         leeway = _dispatch_time_nano2mach(leeway);
  57.     }
  58.     dt->du_timer_flags |= _dispatch_timer_flags_from_clock(clock);
  59.     dt->dt_timer.target = target;
  60.     dt->dt_timer.interval = UINT64_MAX;
  61.     dt->dt_timer.deadline = target + leeway;
  62.     dispatch_activate(ds);
  63. }

代码并不是太复杂，无时间差则直接调用 dispatch_async，否则先创建一个dispatch_source_t，不同的是这里的类型并不是 DISPATCH_SOURCE_TYPE_TIMER 而是 _dispatch_source_type_after，查看源码不难发现它只是 dispatch_source_type_s 类型的一个常量和 _dispatch_source_type_timer 并没有明显区别：

  1. const dispatch_source_type_s _dispatch_source_type_after = {
  2.     .dst_kind           = "timer (after)",
  3.     .dst_filter         = DISPATCH_EVFILT_TIMER_WITH_CLOCK,
  4.     .dst_flags          = EV_DISPATCH,
  5.     .dst_mask           = 0,
  6.     .dst_timer_flags    = DISPATCH_TIMER_AFTER,
  7.     .dst_action         = DISPATCH_UNOTE_ACTION_SOURCE_TIMER,
  8.     .dst_size           = sizeof(struct dispatch_timer_source_refs_s),
  10.     .dst_create         = _dispatch_source_timer_create,
  11.     .dst_merge_evt      = _dispatch_source_merge_evt,
  12. };

而 dispatch_activate() 其实和 dispatch_resume() 是一样的开启定时器。那么为什么看不到 dispatch_source_set_event_handler 来给 timer 设置 handler 呢？不妨看一下 dispatch_source_set_event_handler 的源代码：

  1. void
  2. dispatch_source_set_event_handler(dispatch_source_t ds,
  3.         dispatch_block_t handler)
  4. {
  5.     _dispatch_source_set_handler(ds, handler, DS_EVENT_HANDLER, true);
  6. }
  8. // 查看 _dispatch_source_set_handler 
  10. static void
  11. _dispatch_source_set_handler(dispatch_source_t ds, void *func,
  12.         uintptr_t kind, bool is_block)
  13. {
  14.     dispatch_continuation_t dc;
  16.     dc = _dispatch_source_handler_alloc(ds, func, kind, is_block);
  18.     if (_dispatch_lane_try_inactive_suspend(ds)) {
  19.         _dispatch_source_handler_replace(ds, kind, dc);
  20.         return _dispatch_lane_resume(ds, DISPATCH_RESUME);
  21.     }
  23.     dispatch_queue_flags_t dqf = _dispatch_queue_atomic_flags(ds);
  24.     if (unlikely(dqf & DSF_STRICT)) {
  25.         DISPATCH_CLIENT_CRASH(kind, "Cannot change a handler of this source "
  26.                 "after it has been activated");
  27.     }
  28.     // Ignore handlers mutations past cancelation, it's harmless
  29.     if ((dqf & DSF_CANCELED) == 0) {
  30.         _dispatch_ktrace1(DISPATCH_PERF_post_activate_mutation, ds);
  31.         if (kind == DS_REGISTN_HANDLER) {
  32.             _dispatch_bug_deprecated("Setting registration handler after "
  33.                     "the source has been activated");
  34.         } else if (func == NULL) {
  35.             _dispatch_bug_deprecated("Clearing handler after "
  36.                     "the source has been activated");
  37.         }
  38.     }
  39.     dc->dc_data = (void *)kind;
  40.     _dispatch_barrier_trysync_or_async_f(ds, dc,
  41.             _dispatch_source_set_handler_slow, 0);
  42. }

可以看到最终还是封装成一个 dispatch_continuation_t 进行同步或者异步调用，而上面 _dispatch_after 直接构建了 dispatch_continuation_t 进行执行。

取消延迟执行的任务

使用 dispatch_after 还有一个问题就是取消问题，当然通常遇到了这种问题大部分答案就是使用下面的方式：

  1. [self performSelector:@selector(myDelayedMethod) withObject: self afterDelay: desiredDelay];
  3. [NSObject cancelPreviousPerformRequestsWithTarget: self selector:@selector(myDelayedMethod) object: self];

不过如果你使用的是 iOS 8 及其以上的版本，那么其实是可以取消的（如下），当然如果你还在支持 iOS 8以下 的版本不妨试试这个自定义的dispatch_cancelable_block_t类：

  1. dispatch_block_t block = dispatch_block_create(DISPATCH_BLOCK_INHERIT_QOS_CLASS, ^{
  2.     NSLog(@"dispatch_after...");
  4. });
  5. dispatch_after(dispatch_time(DISPATCH_TIME_NOW, 3*NSEC_PER_SEC), dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), block);
  7. // 取消
  8. dispatch_block_cancel(block);

7、dispatch_apply

dispatch_apply 设计的主要目的是提高并行能力，所以一般我们用来并行执行多个结构类似的任务，比如：

  1. void
  2. dispatch_apply(size_t iterations, dispatch_queue_t dq, void (^work)(size_t))
  3. {
  4.     dispatch_apply_f(iterations, dq, work,
  5.             (dispatch_apply_function_t)_dispatch_Block_invoke(work));
  6. }
  8. // 查看 dispatch_apply_f，对于 width=1 或者单核心数cpu其实这个是一个同步调用，核心方法就是 _dispatch_apply_f
  10. void
  11. dispatch_apply_f(size_t iterations, dispatch_queue_t _dq, void *ctxt,
  12.         void (*func)(void *, size_t))
  13. {
  14.     if (unlikely(iterations == 0)) {
  15.         return;
  16.     }
  17.     dispatch_thread_context_t dtctxt =
  18.             _dispatch_thread_context_find(_dispatch_apply_key);
  19.     size_t nested = dtctxt ? dtctxt->dtc_apply_nesting : 0;
  20.     dispatch_queue_t old_dq = _dispatch_queue_get_current();
  21.     dispatch_queue_t dq;
  23.     if (likely(_dq == DISPATCH_APPLY_AUTO)) {
  24.         dq = _dispatch_apply_root_queue(old_dq)->_as_dq;
  25.     } else {
  26.         dq = _dq; // silence clang Nullability complaints
  27.     }
  28.     dispatch_qos_t qos = _dispatch_priority_qos(dq->dq_priority) ?:
  29.             _dispatch_priority_fallback_qos(dq->dq_priority);
  30.     if (unlikely(dq->do_targetq)) {
  31.         // if the queue passed-in is not a root queue, use the current QoS
  32.         // since the caller participates in the work anyway
  33.         qos = _dispatch_qos_from_pp(_dispatch_get_priority());
  34.     }
  35.     int32_t thr_cnt = (int32_t)_dispatch_qos_max_parallelism(qos,
  36.             DISPATCH_MAX_PARALLELISM_ACTIVE);
  38.     if (likely(!nested)) {
  39.         nested = iterations;
  40.     } else {
  41.         thr_cnt = nested < (size_t)thr_cnt ? thr_cnt / (int32_t)nested : 1;
  42.         nested = nested < DISPATCH_APPLY_MAX && iterations < DISPATCH_APPLY_MAX
  43.                 ? nested * iterations : DISPATCH_APPLY_MAX;
  44.     }
  45.     if (iterations < (size_t)thr_cnt) {
  46.         thr_cnt = (int32_t)iterations;
  47.     }
  48.     struct dispatch_continuation_s dc = {
  49.         .dc_func = (void*)func,
  50.         .dc_ctxt = ctxt,
  51.         .dc_data = dq,
  52.     };
  53.     dispatch_apply_t da = (__typeof__(da))_dispatch_continuation_alloc();
  54.     da->da_index = 0;
  55.     da->da_todo = iterations;
  56.     da->da_iterations = iterations;
  57.     da->da_nested = nested;
  58.     da->da_thr_cnt = thr_cnt;
  59. #if DISPATCH_INTROSPECTION
  60.     da->da_dc = _dispatch_continuation_alloc();
  61.     *da->da_dc = dc;
  62.     da->da_dc->dc_flags = DC_FLAG_ALLOCATED;
  63. #else
  64.     da->da_dc = &dc;
  65. #endif
  66.     da->da_flags = 0;
  68.     if (unlikely(dq->dq_width == 1 || thr_cnt <= 1)) {
  69.         return dispatch_sync_f(dq, da, _dispatch_apply_serial);
  70.     }
  71.     if (unlikely(dq->do_targetq)) {
  72.         if (unlikely(dq == old_dq)) {
  73.             return dispatch_sync_f(dq, da, _dispatch_apply_serial);
  74.         } else {
  75.             return dispatch_sync_f(dq, da, _dispatch_apply_redirect);
  76.         }
  77.     }
  79.     dispatch_thread_frame_s dtf;
  80.     _dispatch_thread_frame_push(&dtf, dq);
  81.     _dispatch_apply_f(upcast(dq)._dgq, da, _dispatch_apply_invoke);
  82.     _dispatch_thread_frame_pop(&dtf);
  83. }
  85. // 查看 _dispatch_apply_f 
  87. static inline void
  88. _dispatch_apply_f(dispatch_queue_global_t dq, dispatch_apply_t da,
  89.         dispatch_function_t func)
  90. {
  91.     int32_t i = 0;
  92.     dispatch_continuation_t head = NULL, tail = NULL;
  93.     pthread_priority_t pp = _dispatch_get_priority();
  95.     // The current thread does not need a continuation
  96.     int32_t continuation_cnt = da->da_thr_cnt - 1;
  98.     dispatch_assert(continuation_cnt);
  100.     for (i = 0; i < continuation_cnt; i++) {
  101.         dispatch_continuation_t next = _dispatch_continuation_alloc();
  102.         uintptr_t dc_flags = DC_FLAG_CONSUME;
  104.         _dispatch_continuation_init_f(next, dq, da, func,
  105.                 DISPATCH_BLOCK_HAS_PRIORITY, dc_flags);
  106.         next->dc_priority = pp | _PTHREAD_PRIORITY_ENFORCE_FLAG;
  107.         next->do_next = head;
  108.         head = next;
  110.         if (!tail) {
  111.             tail = next;
  112.         }
  113.     }
  115.     _dispatch_thread_event_init(&da->da_event);
  116.     // FIXME: dq may not be the right queue for the priority of `head`
  117.     _dispatch_trace_item_push_list(dq, head, tail);
  118.     _dispatch_root_queue_push_inline(dq, head, tail, continuation_cnt);
  119.     // Call the first element directly
  120.     _dispatch_apply_invoke_and_wait(da);
  121. }
  123. // 展开 _dispatch_apply_invoke_and_wait
  125. static void
  126. _dispatch_apply_invoke_and_wait(void *ctxt)
  127. {
  128.     _dispatch_apply_invoke2(ctxt, DISPATCH_APPLY_INVOKE_WAIT);
  129.     _dispatch_perfmon_workitem_inc();
  130. }
  132. // 查看 _dispatch_apply_invoke2，主要是循环使用 _dispatch_perfmon_workitem_inc 调用任务，同时在最后一个任务调用完恢复线程 _dispatch_thread_event_signal
  134. static inline void
  135. _dispatch_apply_invoke2(dispatch_apply_t da, long invoke_flags)
  136. {
  137.     size_t const iter = da->da_iterations;
  138.     size_t idx, done = 0;
  140.     idx = os_atomic_inc_orig2o(da, da_index, acquire);
  141.     if (unlikely(idx >= iter)) goto out;
  143.     // da_dc is only safe to access once the 'index lock' has been acquired
  144.     dispatch_apply_function_t const func = (void *)da->da_dc->dc_func;
  145.     void *const da_ctxt = da->da_dc->dc_ctxt;
  147.     _dispatch_perfmon_workitem_dec(); // this unit executes many items
  149.     // Handle nested dispatch_apply rdar://problem/9294578
  150.     dispatch_thread_context_s apply_ctxt = {
  151.         .dtc_key = _dispatch_apply_key,
  152.         .dtc_apply_nesting = da->da_nested,
  153.     };
  154.     _dispatch_thread_context_push(&apply_ctxt);
  156.     dispatch_thread_frame_s dtf;
  157.     dispatch_priority_t old_dbp = 0;
  158.     if (invoke_flags & DISPATCH_APPLY_INVOKE_REDIRECT) {
  159.         dispatch_queue_t dq = da->da_dc->dc_data;
  160.         _dispatch_thread_frame_push(&dtf, dq);
  161.         old_dbp = _dispatch_set_basepri(dq->dq_priority);
  162.     }
  163.     dispatch_invoke_flags_t flags = da->da_flags;
  165.     // Striding is the responsibility of the caller.
  166.     do {
  167.         dispatch_invoke_with_autoreleasepool(flags, {
  168.             _dispatch_client_callout2(da_ctxt, idx, func);
  169.             _dispatch_perfmon_workitem_inc();
  170.             done++;
  171.             idx = os_atomic_inc_orig2o(da, da_index, relaxed);
  172.         });
  173.     } while (likely(idx < iter));
  175.     if (invoke_flags & DISPATCH_APPLY_INVOKE_REDIRECT) {
  176.         _dispatch_reset_basepri(old_dbp);
  177.         _dispatch_thread_frame_pop(&dtf);
  178.     }
  180.     _dispatch_thread_context_pop(&apply_ctxt);
  182.     // The thread that finished the last workitem wakes up the possibly waiting
  183.     // thread that called dispatch_apply. They could be one and the same.
  184.     if (!os_atomic_sub2o(da, da_todo, done, release)) {
  185.         _dispatch_thread_event_signal(&da->da_event);
  186.     }
  187. out:
  188.     if (invoke_flags & DISPATCH_APPLY_INVOKE_WAIT) {
  189.         _dispatch_thread_event_wait(&da->da_event);
  190.         _dispatch_thread_event_destroy(&da->da_event);
  191.     }
  192.     if (os_atomic_dec2o(da, da_thr_cnt, release) == 0) {
  193. #if DISPATCH_INTROSPECTION
  194.         _dispatch_continuation_free(da->da_dc);
  195. #endif
  196.         _dispatch_continuation_free((dispatch_continuation_t)da);
  197.     }
  198. }

8、补充

8.1 线程和锁

在GCD中其实总共有两个线程池进行线程管理，一个是主线程池，另一个是除了主线程池之外的线程池。主线程池由序列为1的主队列管理，使用 objc.io 上的一幅图表示如下：

image.png

大家都知道使用 dispatch_sync 很有可能会发生死锁那么这是为什么呢？

不妨回顾一下 dispatch_sync 的过程：

  1. dispatch_sync
  2.     _dispatch_sync_f:区分并发还是串行队列，如果是串行队列
  3.         _dispatch_barrier_sync_f:是不是同一个队列，如果是
  4.             _dispatch_sync_f_slow

重点在 _dq_state_drain_locked_by(dq_state, dsc->dsc_waiter) 这个条件，成立则会发生死锁，那么它成立的条件就是 ((lock_value ^ tid) & DLOCK_OWNER_MASK) == 0 首先 lock_value 和 tid 进行异或操作，相同为 0 不同为 1，然后和 DLOCK_OWNER_MASK（0xfffffffc） 进行按位与操作，一个为0则是0，所以如果 lock_value 和 tid 相同则会发生死锁。

8.2 likely 和 unlikely

  1. #define likely(x) __builtin_expect(!!(x), 1)
  2. #define unlikely(x) __builtin_expect(!!(x), 0)

看了 likely 和 unlikely 可以了解，likely 表示更大可能成立，unlikely 表示更大可能不成立。likely 就是 if(likely(x == 0)) 就是 if (x==0)。

8.3 __builtin_expect

__builtin_expect 是一个针对编译器优化的内置函数，让编译更加优化。比如说我们会写这种代码：

  1. if a {
  2.     print(a)
  3. } else {
  4.     print(b)
  5. }

如果我们更加倾向于使用 a 那么可将其设为默认值，极特殊情况下才会使用 b 条件。CPU读取指定是多条一起加载的，可能先加载进来的是 a，那么如果遇到执行 b 的情况则再加载 b，那么对于条件 a 的情况就造成了性能浪费。long builtin_expect (long EXP, long C) 第一个参数是要预测变量，第二个参数是预测值，这样 builtin_expect(a,false) 说明多数情况 a 应该是false，极少数情况可能是 true，这样不至于造成性能浪费。其实对于编译器在汇编时会优化成 if !a 的形式：

  1. if !a {
  2.     print(b)
  3. } else {
  4.     print(a)
  5. }

8.4 os_atomic_cmpxchg

第二个参数与第一个参数值比较，如果相等，第三个参数的值替换第一个参数的值。如果不相等，把第一个参数的值赋值到第二个参数上。

  1. #define os_atomic_cmpxchg(p, e, v, m) \
  2.         ({ _os_atomic_basetypeof(p) _r = (e); \
  3.         atomic_compare_exchange_strong_explicit(_os_atomic_c11_atomic(p), \
  4.         &_r, v, memory_order_##m, memory_order_relaxed); })

8.5 os_atomic_store2o

将第二个参数保存到第一个参数中

  1. #define os_atomic_store2o(p, f, v, m) \
  2.         os_atomic_store(&(p)->f, (v), m)
  3. #define os_atomic_store(p, v, m) \
  4.         atomic_store_explicit(_os_atomic_c11_atomic(p), v, memory_order_##m)

8.6 os_atomic_inc_orig

第一个参数赋值为1

  1. #define os_atomic_inc_orig(p, m) \
  2.         os_atomic_add_orig((p), 1, m)
  3. #define os_atomic_add_orig(p, v, m) \
  4.         _os_atomic_c11_op_orig((p), (v), m, add, +)
  5. #define _os_atomic_c11_op_orig(p, v, m, o, op) \
  6.         atomic_fetch_##o##_explicit(_os_atomic_c11_atomic(p), v, \
  7.         memory_order_##m)

8.7 os_atomic_inc_orig2o

第二个参数加1并返回

  1. #define os_atomic_inc_orig2o(p, f, m) \
  2.         os_atomic_add_orig2o(p, f, 1, m)
  3. #define os_atomic_add_orig2o(p, f, v, m) \
  4.         os_atomic_add_orig(&(p)->f, (v), m)

8.8 os_atomic_dec2o

第二个参数-1并返回

  1. #define os_atomic_dec2o(p, f, m) \
  2.         os_atomic_sub2o(p, f, 1, m)
  3. #define os_atomic_sub2o(p, f, v, m) \
  4.         os_atomic_sub(&(p)->f, (v), m)