- dispatch_queue_create
dispatch_queue_create调用了_dispatch_lane_create_with_target，继续调用_dispatch_object_alloc、_dispatch_queue_init、_dispatch_trace_queue_create,最终返回了dispatch_queue_t。重要的代码：
dispatch_lane_t dq = _dispatch_object_alloc(vtable, sizeof(struct dispatch_lane_s));
_dispatch_queue_init(dq, dqf, dqai.dqai_concurrent ? DISPATCH_QUEUE_WIDTH_MAX : 1,DISPATCH_QUEUE_ROLE_INNER | (dqai.dqai_inactive ? DISPATCH_QUEUE_INACTIVE : 0));
这其中的dqai.dqai_concurrent看着跟队列的串行并发有关系。追根溯源，一方面我们搞清楚DISPATCH_QUEUE_CONCURRENT是什么：
//串行队列属性DISPATCH_QUEUE_SERIAL==NULL
#define DISPATCH_QUEUE_SERIAL NULL
//并发队列属性DISPATCH_QUEUE_CONCURRENT使用了DISPATCH_GLOBAL_OBJECT宏定义
#define DISPATCH_QUEUE_CONCURRENT DISPATCH_GLOBAL_OBJECT(dispatch_queue_attr_t, _dispatch_queue_attr_concurrent)
//DISPATCH_GLOBAL_OBJECT：将object转换为type类型
#define DISPATCH_GLOBAL_OBJECT(type, object) ((OS_OBJECT_BRIDGE type)&(object))
也就是我们使用的DISPATCH_QUEUE_CONCURRENT宏定义，实际上是struct dispatch_queue_attr_s _dispatch_queue_attr_concurrent;结构体。
再者，我们探究dqai创建的函数dispatch_queue_attr_info_t dqai = _dispatch_queue_attr_to_info(dqa);:
dispatch_queue_attr_info_t _dispatch_queue_attr_to_info(dispatch_queue_attr_t dqa){
dispatch_queue_attr_info_t dqai = { };
if (!dqa) return dqai;
if (dqa == &_dispatch_queue_attr_concurrent) {
dqai.dqai_concurrent = true;
return dqai;
}
...
return dqai;
}
如果是串行队列到参数dqa为NULL，则直接返回dqai;而并发队列的dqa== &_dispatch_queue_attr_concurrent，则标记dqai.dqai_concurrent = true。返回到上面的位置，调用_dispatch_queue_init的时候，如果是并发队列，第三个参数传入的是DISPATCH_QUEUE_WIDTH_MAX，而
#define DISPATCH_QUEUE_WIDTH_MAX (DISPATCH_QUEUE_WIDTH_FULL - 2)
#define DISPATCH_QUEUE_WIDTH_FULL 0x1000ull
即DISPATCH_QUEUE_WIDTH_MAX值为6，串行队列传入的是1。继续往下看：
static inline dispatch_queue_class_t _dispatch_queue_init(dispatch_queue_class_t dqu, dispatch_queue_flags_t dqf, uint16_t width, uint64_t initial_state_bits) {
uint64_t dq_state = DISPATCH_QUEUE_STATE_INIT_VALUE(width);
dispatch_queue_t dq = dqu._dq;
dispatch_assert((initial_state_bits & ~(DISPATCH_QUEUE_ROLE_MASK |
DISPATCH_QUEUE_INACTIVE)) == 0);
if (initial_state_bits & DISPATCH_QUEUE_INACTIVE) {
dq->do_ref_cnt += 2; // rdar://8181908 see _dispatch_lane_resume
if (dx_metatype(dq) == _DISPATCH_SOURCE_TYPE) {
dq->do_ref_cnt++; // released when DSF_DELETED is set
}
}
dq_state |= initial_state_bits;
dq->do_next = DISPATCH_OBJECT_LISTLESS;
dqf |= DQF_WIDTH(width);
os_atomic_store2o(dq, dq_atomic_flags, dqf, relaxed);
dq->dq_state = dq_state;
dq->dq_serialnum = os_atomic_inc_orig(&_dispatch_queue_serial_numbers, relaxed);
return dqu;
}
最终串行队列DQF_WIDTH(width)的width=1，而并发队列width=6,这还需要经过一些运算保存到queue的width属性上。
带着好奇心，我们打印下串行队列和并发队列、以及常用的全局并发队列和主队列的信息看看：
dispatch_queue_t concurrent = dispatch_queue_create("", DISPATCH_QUEUE_CONCURRENT);
dispatch_queue_t global = dispatch_get_global_queue(0, 0);
dispatch_queue_t serial = dispatch_queue_create("", DISPATCH_QUEUE_SERIAL);
dispatch_queue_t main = dispatch_get_main_queue();
打印结果如下：
队列的width.jpg
打印的结果显示：自己创建的串行队列和默认的主队列width==1，而自己创建的并发队列和全局队列width则为一个很大的数。
这里的dq->dq_serialnum又是什么呢？
// skip zero
// 1 - main_q
// 2 - mgr_q
// 3 - mgr_root_q
// 4,5,6,7,8,9,10,11,12,13,14,15 - global queues
// 17 - workloop_fallback_q
// we use 'xadd' on Intel, so the initial value == next assigned
#define DISPATCH_QUEUE_SERIAL_NUMBER_INIT 17
这里有对队列的dq_serialnum的注解，1为主队列，4-15为全局队列，这个可以到_dispatch_root_queues[]中查询到。
2.dispatch_sync
dispatch_sync的基本流程，dispatch_sync内部对任务进行封装_dispatch_Block_invoke(work),然后
调用_dispatch_sync_f：
void dispatch_sync(dispatch_queue_t dq, dispatch_block_t work){
uintptr_t dc_flags = DC_FLAG_BLOCK;
if (unlikely(_dispatch_block_has_private_data(work))) {
return _dispatch_sync_block_with_privdata(dq, work, dc_flags);
}
//_dispatch_Block_invoke(work)是对任务的一个封装
_dispatch_sync_f(dq, work, _dispatch_Block_invoke(work), dc_flags);
}
static void _dispatch_sync_f(dispatch_queue_t dq, void *ctxt,dispatch_function_t func,
uintptr_t dc_flags){
_dispatch_sync_f_inline(dq, ctxt, func, dc_flags);
}
static inline void _dispatch_sync_f_inline(dispatch_queue_t dq, void *ctxt, dispatch_function_t func, uintptr_t dc_flags){
if (likely(dq->dq_width == 1)) {
return _dispatch_barrier_sync_f(dq, ctxt, func, dc_flags);
}
if (unlikely(dx_metatype(dq) != _DISPATCH_LANE_TYPE)) {
DISPATCH_CLIENT_CRASH(0, "Queue type doesn't support dispatch_sync");
}
dispatch_lane_t dl = upcast(dq)._dl;
// Global concurrent queues and queues bound to non-dispatch threads
// always fall into the slow case, see DISPATCH_ROOT_QUEUE_STATE_INIT_VALUE
if (unlikely(!_dispatch_queue_try_reserve_sync_width(dl))) {
return _dispatch_sync_f_slow(dl, ctxt, func, 0, dl, dc_flags);
}
if (unlikely(dq->do_targetq->do_targetq)) {
return _dispatch_sync_recurse(dl, ctxt, func, dc_flags);
}
_dispatch_introspection_sync_begin(dl);
_dispatch_sync_invoke_and_complete(dl, ctxt, func DISPATCH_TRACE_ARG(
_dispatch_trace_item_sync_push_pop(dq, ctxt, func, dc_flags)));
}
/未完待续/
3.dispatch_async
/未完待续/
4.dispatch_once
dispatch_once常用于单例对象的初始化，而且是线程安全的，下面来看具体的函数调用：
void dispatch_once(dispatch_once_t *val, dispatch_block_t block){
//入参数，第一个val是dispatch_once_t*类型的静态变量，第二个block是dispatch_block_t类型
//调用dispatch_once_f函数，前2个参数同上，第三个参数是对block的封装
dispatch_once_f(val, block, _dispatch_Block_invoke(block));
}
//核心函数
void dispatch_once_f(dispatch_once_t *val, void *ctxt, dispatch_function_t func){
//将val(也就是外部传入的静态变量)转换为dispatch_once_gate_t类型的变量
dispatch_once_gate_t l = (dispatch_once_gate_t)val;
#if !DISPATCH_ONCE_INLINE_FASTPATH || DISPATCH_ONCE_USE_QUIESCENT_COUNTER
//通过os_atomic_load获取此时任务的标识符v
uintptr_t v = os_atomic_load(&l->dgo_once, acquire);
//如果当前任务的状态为DLOCK_ONCE_DONE，表示任务已经执行过了，则直接return，不会触发block的调用。
if (likely(v == DLOCK_ONCE_DONE)) {
return;
}
#if DISPATCH_ONCE_USE_QUIESCENT_COUNTER
//如果任务执行加锁失败，则走到_dispatch_once_mark_done_if_quiesced，并将标识符标记为DLOCK_ONCE_DONE
if (likely(DISPATCH_ONCE_IS_GEN(v))) {
//内部调用os_atomic_store(&dgo->dgo_once, DLOCK_ONCE_DONE, ...);
return _dispatch_once_mark_done_if_quiesced(l, v);
}
#endif
#endif
//通过_dispatch_once_gate_tryenter尝试进入任务，即解锁，然后通过_dispatch_once_callout执行block回调任务
//os_atomic_cmpxchg(&l->dgo_once, DLOCK_ONCE_UNLOCKED, (uintptr_t)_dispatch_lock_value_for_self(), relaxed)
if (_dispatch_once_gate_tryenter(l)) {
//触发任务的执行，并标记任务的状态
return _dispatch_once_callout(l, ctxt, func);
}
//如果此时有任务正在执行，这时再进来一个新任务，则通过_dispatch_once_wait让新任务进入无限次等待。
return _dispatch_once_wait(l);
}
这里读取任务状态是通过os_atomic_load(&dgo->dgo_once, ...)进行的，而标记任务状态则是通过os_atomic_store(&dgo->dgo_once, DLOCK_ONCE_DONE, ...);进行的两者相呼应。这里标记为DLOCK_ONCE_DONE则修改dgo_once==1。
//执行block任务，并将任务的状态标记为DLOCK_ONCE_DONE。
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);
}
//执行任务
_dispatch_client_callout(void *ctxt, dispatch_function_t f){
...
f(ctxt);
...
}
//标记任务的状态,广播通知
static inline void _dispatch_once_gate_broadcast(dispatch_once_gate_t l) {
dispatch_lock value_self = _dispatch_lock_value_for_self();
uintptr_t v = _dispatch_once_mark_done(l);//标记为DLOCK_ONCE_DONE
if (likely((dispatch_lock)v == value_self)) return;
_dispatch_gate_broadcast_slow(&l->dgo_gate, (dispatch_lock)v);
}
执行任务在_dispatch_client_callout中通过f(ctxt);触发。 _dispatch_once_mark_done中将任务的状态标记为DLOCK_ONCE_DONE了。
那么如果任务正在执行，这个无限次等待是怎么实现的呢？
void _dispatch_once_wait(dispatch_once_gate_t dgo){
...
for (;;) {
os_atomic_rmw_loop(&dgo->dgo_once, old_v, new_v, relaxed, {
if (likely(old_v == DLOCK_ONCE_DONE)) {
os_atomic_rmw_loop_give_up(return);
}
if (DISPATCH_ONCE_IS_GEN(old_v)) {
os_atomic_rmw_loop_give_up({
os_atomic_thread_fence(acquire);
return _dispatch_once_mark_done_if_quiesced(dgo, old_v);
});
}
new_v = old_v | (uintptr_t)DLOCK_WAITERS_BIT;
if (new_v == old_v) os_atomic_rmw_loop_give_up(break);
});
...
}
}
等待任务标记为完成后发出广播，这个就立即返回了。
