iOS開發探索多執行緒GCD常用函數

2022-07-26 14:03:07

正文

前篇文章我們瞭解了GCD的任務的原理，接下來我們在探索一下GCD中我們開發常用的函數

單例

下面我們從原始碼中看一下我們建立單例的時候使用的dispatch_once，都做了什麼，是通過什麼操作保證全域性唯一的

void dispatch_once(dispatch_once_t *val, dispatch_block_t block) {
    dispatch_once_f(val, block, _dispatch_Block_invoke(block));
}
void dispatch_once_f(dispatch_once_t *val, void *ctxt, dispatch_function_t func) {
    dispatch_once_gate_t l = (dispatch_once_gate_t)val;
    #if !DISPATCH_ONCE_INLINE_FASTPATH || DISPATCH_ONCE_USE_QUIESCENT_COUNTER
        uintptr_t v = os_atomic_load(&l->dgo_once, acquire);
        if (likely(v == DLOCK_ONCE_DONE)) {
            return;
        }
        #if DISPATCH_ONCE_USE_QUIESCENT_COUNTER
            if (likely(DISPATCH_ONCE_IS_GEN(v))) {
                return _dispatch_once_mark_done_if_quiesced(l, v);
            }
        #endif
    #endif
    if (_dispatch_once_gate_tryenter(l)) {
        return _dispatch_once_callout(l, ctxt, func);
    }
    return _dispatch_once_wait(l);
}
// 原子屬性，比較 + 交換 &l->dgo_once 是否等於 DLOCK_ONCE_UNLOCKED，相等還沒有執行過，將_dispatch_lock_value_for_self() 賦值給 &l->dgo_once，返回true；不等返回false
#define os_atomic_cmpxchg(p, e, v, m) 
    ({ _os_atomic_basetypeof(p) _r = (e); 
    atomic_compare_exchange_strong_explicit(_os_atomic_c11_atomic(p), 
    &_r, v, memory_order_##m, memory_order_relaxed); })
static inline bool _dispatch_once_gate_tryenter(dispatch_once_gate_t l) {
    return os_atomic_cmpxchg(&l->dgo_once, DLOCK_ONCE_UNLOCKED, (uintptr_t)_dispatch_lock_value_for_self(), relaxed);
}

和任務一樣，對block進行了封裝，同時通過判斷dispatch_once_t *val的狀態進行判斷，DLOCK_ONCE_DONE直接返回，也就是已經執行了直接返回，呼叫_dispatch_once_gate_tryenter判斷是否執行過block，未執行_dispatch_once_callout進行執行程式碼，執行過呼叫_dispatch_once_wait等待執行結束。

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); // 標記執行完成
}
static inline void _dispatch_once_gate_broadcast(dispatch_once_gate_t l) {
    dispatch_lock value_self = _dispatch_lock_value_for_self();
    uintptr_t v;
    #if DISPATCH_ONCE_USE_QUIESCENT_COUNTER
        v = _dispatch_once_mark_quiescing(l);
    #else
        v = _dispatch_once_mark_done(l); // 將 &l->dgo_once 設定成 DLOCK_ONCE_DONE,標記函數已經執行完成了,廣播的作用
    #endif
    if (likely((dispatch_lock)v == value_self)) return;
    _dispatch_gate_broadcast_slow(&l->dgo_gate, (dispatch_lock)v);
}
void _dispatch_once_wait(dispatch_once_gate_t dgo) {
    ...省略部分...
    for (;;) {
        // 從底層獲取 &dgo->dgo_onc 的狀態
        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_USE_QUIESCENT_COUNTER
            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);
                });
            }
            #endif
            new_v = old_v | (uintptr_t)DLOCK_WAITERS_BIT;
            if (new_v == old_v) os_atomic_rmw_loop_give_up(break);
        });
        ...省略部分...
    }
}

_dispatch_once_wait中使用死迴圈，直到DLOCK_ONCE_DONE時呼叫os_atomic_rmw_loop_give_up(return);退出迴圈,通過這種方式保證只建立一份.

柵欄函數

柵欄函數：前面的任務沒有執行完成時，不執行柵欄函數中的任務，柵欄函數後面的任務需等待柵欄函數中的任務執行完成才能執行。

柵欄函數只能攔截同一佇列中的任務
柵欄函數無法攔截全域性佇列，因為系統操作也會使用全域性佇列
攔截同步佇列和普通的任務原理相同柵欄函數分為同步函數和非同步函數，我們先看一下同步函數的原始碼dispatch_barrier_sync

void dispatch_barrier_sync(dispatch_queue_t dq, dispatch_block_t work) {
    ...省略...
    _dispatch_barrier_sync_f(dq, work, _dispatch_Block_invoke(work), dc_flags); // 和之前一樣的封裝block
}
_dispatch_barrier_sync_f -> _dispatch_barrier_sync_f_inline
static inline void _dispatch_barrier_sync_f_inline(dispatch_queue_t dq, void *ctxt, dispatch_function_t func, uintptr_t dc_flags) {
    dispatch_tid tid = _dispatch_tid_self();
    dispatch_lane_t dl = upcast(dq)._dl;
    if (unlikely(!_dispatch_queue_try_acquire_barrier_sync(dl, tid))) {
        return _dispatch_sync_f_slow(dl, ctxt, func, DC_FLAG_BARRIER, dl, DC_FLAG_BARRIER | dc_flags);
    }
    if (unlikely(dl->do_targetq->do_targetq)) {
        return _dispatch_sync_recurse(dl, ctxt, func, DC_FLAG_BARRIER | dc_flags);
    }
    _dispatch_introspection_sync_begin(dl);
    _dispatch_lane_barrier_sync_invoke_and_complete(dl, ctxt, func DISPATCH_TRACE_ARG(_dispatch_trace_item_sync_push_pop(dq, ctxt, func, dc_flags | DC_FLAG_BARRIER)));
}

在_dispatch_barrier_sync_f_inline函數中有3個執行func的函數, 我們將這三個函數使用符號斷點來檢視其呼叫的是哪個函數,發現呼叫的是_dispatch_sync_f_slow

static void _dispatch_sync_f_slow(dispatch_queue_class_t top_dqu, void *ctxt, dispatch_function_t func, uintptr_t top_dc_flags, dispatch_queue_class_t dqu, uintptr_t dc_flags) {
    dispatch_queue_t top_dq = top_dqu._dq;
    dispatch_queue_t dq = dqu._dq;
    if (unlikely(!dq->do_targetq)) {
        return _dispatch_sync_function_invoke(dq, ctxt, func);
    }
    ...省略部分...
    // 等待前面的任務執行完成
    _dispatch_trace_item_push(top_dq, &dsc);
    __DISPATCH_WAIT_FOR_QUEUE__(&dsc, dq);
    if (dsc.dsc_func == NULL) {
        // dsc_func being cleared means that the block ran on another thread ie.
        // case (2) as listed in _dispatch_async_and_wait_f_slow.
        dispatch_queue_t stop_dq = dsc.dc_other;
        return _dispatch_sync_complete_recurse(top_dq, stop_dq, top_dc_flags);
    }
    _dispatch_introspection_sync_begin(top_dq);
    _dispatch_trace_item_pop(top_dq, &dsc);
    _dispatch_sync_invoke_and_complete_recurse(top_dq, ctxt, func,top_dc_flags
}

繼續檢視_dispatch_sync_f_slow的原始碼, 內部執行func的函數有兩個,也加入符號斷點檢視,最後得到呼叫的是_dispatch_sync_invoke_and_complete_recurse, 該方法就是同步任務呼叫的執行block任務的函數, 該函數內部有個_dispatch_sync_complete_recurse函數

static void _dispatch_sync_complete_recurse(dispatch_queue_t dq, dispatch_queue_t stop_dq, uintptr_t dc_flags){
bool barrier = (dc_flags & DC_FLAG_BARRIER);
    do {
        if (dq == stop_dq) return;
        if (barrier) {
            dx_wakeup(dq, 0, DISPATCH_WAKEUP_BARRIER_COMPLETE);
        } else {
            _dispatch_lane_non_barrier_complete(upcast(dq)._dl, 0);
        }
        dq = dq->do_targetq;
        barrier = (dq->dq_width == 1);
    } while (unlikely(dq->do_targetq));
}

該函數是使用do while,內部使用barrier來判斷是否使用柵欄函數,沒有柵欄函數就呼叫佇列中接下來的任務,有柵欄函數就呼叫dx_wakeup

#define dx_wakeup(x, y, z) dx_vtable(x)->dq_wakeup(x, y, z)

dq_wakeup和之前非同步任務的dq_push一樣,是針對不同佇列呼叫不同的方法

DISPATCH_VTABLE_SUBCLASS_INSTANCE(queue_serial, lane,
    .do_type        = DISPATCH_QUEUE_SERIAL_TYPE,
    ......
    .dq_wakeup        = _dispatch_lane_wakeup,
);
DISPATCH_VTABLE_SUBCLASS_INSTANCE(queue_concurrent, lane,
    .do_type        = DISPATCH_QUEUE_CONCURRENT_TYPE,
    ......
    .dq_wakeup        = _dispatch_lane_wakeup,
);
DISPATCH_VTABLE_SUBCLASS_INSTANCE(queue_global, lane,
    .do_type        = DISPATCH_QUEUE_GLOBAL_ROOT_TYPE,
    ......
    .dq_wakeup        = _dispatch_root_queue_wakeup,
);
DISPATCH_VTABLE_SUBCLASS_INSTANCE(queue_main, lane,
    .do_type        = DISPATCH_QUEUE_MAIN_TYPE,
    ......
    .dq_wakeup        = _dispatch_main_queue_wakeup,
);

// 序列佇列及自己建立的並行佇列
void _dispatch_lane_wakeup(dispatch_lane_class_t dqu, dispatch_qos_t qos, dispatch_wakeup_flags_t flags){
dispatch_queue_wakeup_target_t target = DISPATCH_QUEUE_WAKEUP_NONE;
    if (unlikely(flags & DISPATCH_WAKEUP_BARRIER_COMPLETE)) {
        // 柵欄函數任務完成後呼叫該函數
        return _dispatch_lane_barrier_complete(dqu, qos, flags);
    }
    if (_dispatch_queue_class_probe(dqu)) {
        target = DISPATCH_QUEUE_WAKEUP_TARGET;
    }
    // 喚醒後面佇列中的任務,執行柵欄函數後面佇列裡的任務
    return _dispatch_queue_wakeup(dqu, qos, flags, target);
}
// 全域性並行佇列 沒有任何關於柵欄函數的操作,所以柵欄函數對全域性並行佇列無效
void _dispatch_root_queue_wakeup(dispatch_queue_global_t dq, DISPATCH_UNUSED dispatch_qos_t qos, dispatch_wakeup_flags_t flags){
    if (!(flags & DISPATCH_WAKEUP_BLOCK_WAIT)) {
        DISPATCH_INTERNAL_CRASH(dq->dq_priority, "Don't try to wake up or override a root queue");
    }
    if (flags & DISPATCH_WAKEUP_CONSUME_2) {
        return _dispatch_release_2_tailcall(dq);
    }
}

我們可以看到全域性並行佇列的dq_wakeup關聯的函數中沒有任何關於柵欄函數的操作,所以柵欄函數對全域性並行佇列無效.

可是柵欄函數到底是怎麼攔截的呢?

我們還是用全域性符號斷點來進行檢視,將柵欄函數前面的任務進行延時操作,我們執行發現,呼叫_dispatch_sync_f_slow函數後,並沒有立即呼叫_dispatch_sync_invoke_and_complete_recurse,而是等到我們前面的延時操作結束後,才進行的_dispatch_sync_invoke_and_complete_recurse呼叫, 也就是說在_dispatch_sync_f_slow函數呼叫後,會等待前面的函數全部執行完成後,才會執行柵欄函數本身的任務及喚醒柵欄函數後面的任務.

那為什麼柵欄函數還區分同步和非同步函數呢?

其實就是柵欄函數本身的任務是否需要開闢執行緒去進行執行來區分使用同步還是非同步函數

排程組 dispatch_group_t

柵欄函數有自身的侷限性,他只能攔截同一佇列中的任務,當有多個佇列的任務時,無法生效,這是我們應該使用排程組dispatch_group_t來進行任務攔截,保障任務的執行順序. dispatch_group_t有兩種書寫方式:

dispatch_group_async(g, dispatch_get_global_queue(0, 0), ^{});

dispatch_group_enter(g); + dispatch_group_leave(g);

這兩種書寫方式,效果是一模一樣的,dispatch_group_async就是dispatch_group_enter和dispatch_group_leave的整合

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_async(dispatch_queue_class_t dqu, dispatch_continuation_t dc, dispatch_qos_t qos, uintptr_t dc_flags) {
    #if DISPATCH_INTROSPECTION
        if (!(dc_flags & DC_FLAG_NO_INTROSPECTION)) {
            _dispatch_trace_item_push(dqu, dc);
        }
    #else
        (void)dc_flags;
    #endif
    return dx_push(dqu._dq, dc, qos);
}

dispatch_group_async其內部先進行dispatch_group_enter,然後和非同步函數一樣進行了dx_push呼叫來執行任務,和我們之前的非同步任務是一樣的流程,在_dispatch_root_queues_init_once-...->_dispatch_continuation_invoke_inline檔案裡我們可以看到dispatch_group_leave的呼叫

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);
    }
    ...省略部分...
}
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的原始碼

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>
    }
    if (unlikely(old_value == DISPATCH_GROUP_VALUE_MAX)) {
        DISPATCH_CLIENT_CRASH(old_bits,
        "Too many nested calls to dispatch_group_enter()");
    }
}
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.
    // 修改 dg_state
    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_enter進行減1操作,增加組內當前未完成任務的參照計數; dispatch_group_leave進行加1操作,減少組內未完成任務的參照計數;

當參照計數變成0,就會呼叫dispatch_group_notify來執行後續程式碼.

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)) {
        // 監聽 dg_state
        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中會判斷old_state == 0才會執行後續程式碼,如果dispatch_group_enter和dispatch_group_leave不是一一對應的則永遠不會執行dispatch_group_notify,而且dispatch_group_leave中對old_value == 0,也進行了crash判斷,dispatch_group_leave比dispatch_group_enter多的話會直接crash.

號誌 dispatch_semaphore_t

GCD中另一種常用的控制任務執行順序的就是號誌,其主要是控制並行數量

通過dispatch_semaphore_create(0)來建立號誌並指定訊號的大小
dispatch_semaphore_signal(sem), 傳送號誌,將號誌的值 +1
dispatch_semaphore_wait(sem, DISPATCH_TIME_FOREVER) 等待號誌, 將號誌的值 -1, 當訊號值的值小於0時會阻塞執行緒一直進行等待,當訊號值的值大於等於0時執行後續程式碼

接下來我們還是檢視原始碼來看看號誌的實現原理

intptr_t dispatch_semaphore_signal(dispatch_semaphore_t dsema) {
    // 對號誌的值 + 1
    long value = os_atomic_inc2o(dsema, dsema_value, release);
    if (likely(value > 0)) {
        return 0;
    }
    if (unlikely(value == LONG_MIN)) {
        DISPATCH_CLIENT_CRASH(value, "Unbalanced call to dispatch_semaphore_signal()");
    }
    return _dispatch_semaphore_signal_slow(dsema);
}
intptr_t _dispatch_semaphore_signal_slow(dispatch_semaphore_t dsema) {
    _dispatch_sema4_create(&dsema->dsema_sema, _DSEMA4_POLICY_FIFO);
    _dispatch_sema4_signal(&dsema->dsema_sema, 1);
    return 1;
}

我們可以看到signal就只有+1操作,下面我們主要看一下wait是如何進行等待的

intptr_t dispatch_semaphore_wait(dispatch_semaphore_t dsema, dispatch_time_t timeout) {
    // 對號誌進行減1操作
    long value = os_atomic_dec2o(dsema, dsema_value, acquire);
    // 當號誌的值 >= 0直接返回,執行後續程式碼
    if (likely(value >= 0)) {
        return 0;
    }
    return _dispatch_semaphore_wait_slow(dsema, timeout);
}
static intptr_t _dispatch_semaphore_wait_slow(dispatch_semaphore_t dsema, dispatch_time_t timeout){
    long orig;
    _dispatch_sema4_create(&dsema->dsema_sema, _DSEMA4_POLICY_FIFO);
    switch (timeout) {
        default:
            if (!_dispatch_sema4_timedwait(&dsema->dsema_sema, timeout)) {
                break;
            }
        // Fall through and try to undo what the fast path did to
        // dsema->dsema_value
        case DISPATCH_TIME_NOW:
            orig = dsema->dsema_value;
            while (orig < 0) {
                if (os_atomic_cmpxchgv2o(dsema, dsema_value, orig, orig + 1, &orig, relaxed)) {
                    return _DSEMA4_TIMEOUT();
                }
            }
        // Another thread called semaphore_signal().
        // Fall through and drain the wakeup.
        case DISPATCH_TIME_FOREVER:
            _dispatch_sema4_wait(&dsema->dsema_sema);
            break;
    }
    return 0;
}
void _dispatch_sema4_wait(_dispatch_sema4_t *sema) {
    int ret = 0;
    do {
        ret = sem_wait(sema);
    } while (ret == -1 && errno == EINTR);
    DISPATCH_SEMAPHORE_VERIFY_RET(ret);
}

我們可以看到wait進行等待的函數是根據我們傳遞的timeout進行的判斷,然後進行do while迴圈阻塞當前執行緒獲取號誌的值,如果號誌>=0跳出迴圈,執行後續程式碼.

需要注意的是dispatch_semaphore_signal的數量不能比dispatch_semaphore_wait少,否則號誌記憶體無法被釋放,會導致程式崩潰,_dispatch_semaphore_dispose函數中會進行判斷當前的號誌的值 < 原始號誌的值會觸發崩潰.

void _dispatch_semaphore_dispose(dispatch_object_t dou, DISPATCH_UNUSED bool *allow_free) {
    dispatch_semaphore_t dsema = dou._dsema;
    if (dsema->dsema_value < dsema->dsema_orig) {
        DISPATCH_CLIENT_CRASH(dsema->dsema_orig - dsema->dsema_value, "Semaphore object deallocated while in use");
    }
    _dispatch_sema4_dispose(&dsema->dsema_sema, _DSEMA4_POLICY_FIFO);
}

dispatch_source

dispatch_source 和runloop的source相同, 用於監聽事件的,比如計時器,系統記憶體壓力,mach port 待處理訊息等.

dispatch_source_create(DISPATCH_SOURCE_TYPE_TIMER, 0, 0, dispatch_get_global_queue(0, 0)) 建立事件源

dispatch_source_set_event_handler(source, ^{}) 設定源事件回撥

dispatch_source_merge_data(source, 1) 設定源事件資料

dispatch_source_get_data(source) 獲取源事件資料

dispatch_resume(source); 啟動/繼續

dispatch_suspend(source); 掛起，暫停

dispatch_source_cancel(source); 非同步取消源事件

dispatch_cancel(source); 取消源事件