GCD源码分析5 —— dispatch_semaphore篇

发表于 | 分类于
字数统计: 1,307 | 阅读时长 ≈ 6

前言

这篇文章主要分析dispatch_semaphore_t的底层实现

GCD信号量

dispatch_semaphore_t

dispatch_semaphore_s是性能稍次于自旋锁的的信号量对象，用来保证资源使用的安全性。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
DISPATCH_DECL(dispatch_semaphore);

#define DISPATCH_DECL(name) typedef struct name##_s *name##_t;

struct dispatch_semaphore_s {
    DISPATCH_STRUCT_HEADER(dispatch_semaphore_s, dispatch_semaphore_vtable_s);
    long dsema_value;   // 当前信号值，当这个值小于0时无法访问加锁资源
    long dsema_orig;    // 初始化信号值，限制了同时访问资源的线程数量
    
    size_t dsema_sent_ksignals; //由于mach信号可能会被意外唤醒，通过原子操作来避免虚假信号
    
    semaphore_t dsema_port;
    semaphore_t dsema_waiter_port;
    size_t dsema_group_waiters;
    struct dispatch_sema_notify_s *dsema_notify_head;
    struct dispatch_sema_notify_s *dsema_notify_tail;
};
  • 1、semaphore_t是mach_port_t信号
1
typedef mach_port_t semaphore_t;
  • 2、dispatch_sema_notify_s的定义如下：
1
2
3
4
5
6
struct dispatch_sema_notify_s {
    struct dispatch_sema_notify_s *dsn_next;
    dispatch_queue_t dsn_queue;
    void *dsn_ctxt;
    void (*dsn_func)(void *);
};

现在看起来还是一团散沙，硬着头皮往下看吧

dispatch_semaphore_create

创建一个具有初始值的计数信号量。如果有两个线程共同完成某一项工作时，可以使用一个初始值为0的semaphore。如果你需要管理一个有限的资源池时，应该使用一个初始值为资源池的大小的semaphore。如果不再使用这个semaphore时，应该使用dispatch_release销毁semaphore对象并且释放他的内存。(在ARC模式下不需要，系统会自动释放)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
dispatch_semaphore_t dispatch_semaphore_create(long value){
    dispatch_semaphore_t dsema;

    if (value < 0) {
        return NULL;
    }
    // 申请内存
    dsema = calloc(1, sizeof(struct dispatch_semaphore_s));
    
    if (fastpath(dsema)) {
        // do_vtable里面主要包含了这个 dispatch_semaphore_s 的操作函数
        dsema->do_vtable = &_dispatch_semaphore_vtable;

        //可以理解为链表的结尾标记
        dsema->do_next = DISPATCH_OBJECT_LISTLESS;

        // 引用计数
        dsema->do_ref_cnt = 1;
        dsema->do_xref_cnt = 1;

        // 目标队列
        dsema->do_targetq = dispatch_get_global_queue(0, 0);

        // 信号值
        dsema->dsema_value = value;
        dsema->dsema_orig = value;
    }
    
    return dsema;
}
  1. 首先value必须大于等于0，流程才能继续进行；
  2. 申请一块dispatch_semaphore_s内存，并初始化为0；
  3. 将dsema的do_vtable指向_dispatch_semaphore_vtable，_dispatch_semaphore_vtable的定义如下：
1
2
3
4
5
6
const struct dispatch_semaphore_vtable_s _dispatch_semaphore_vtable = {
    .do_type = DISPATCH_SEMAPHORE_TYPE,
    .do_kind = "semaphore",
    .do_dispose = _dispatch_semaphore_dispose,
    .do_debug = _dispatch_semaphore_debug,
};

_dispatch_semaphore_dispose，顾名思义，销毁一个信号量，定义如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
void _dispatch_semaphore_dispose(dispatch_semaphore_t dsema){
    kern_return_t kr;
    
    if (dsema->dsema_value < dsema->dsema_orig) {
        DISPATCH_CLIENT_CRASH("Semaphore/group object deallocated while in use");
    }
    
    if (dsema->dsema_port) {
        kr = semaphore_destroy(mach_task_self(), dsema->dsema_port);
        DISPATCH_SEMAPHORE_VERIFY_KR(kr);
    }
    if (dsema->dsema_waiter_port) {
        kr = semaphore_destroy(mach_task_self(), dsema->dsema_waiter_port);
        DISPATCH_SEMAPHORE_VERIFY_KR(kr);
    }
    
    _dispatch_dispose(dsema);
}

dispatch_semaphore_wait

等待一个semaphore，或者是减少semaphore的计数。每次会执行会将计数器-1.如果减完之后计数器小于0的话，会阻塞在当前线程直到接收到信号。

1
2
3
4
5
6
7
8
long dispatch_semaphore_wait(dispatch_semaphore_t dsema, dispatch_time_t timeout){
    
        // 如果信号量的值-1之后大于等于0,表示有资源可用
    if (dispatch_atomic_dec(&dsema->dsema_value) >= 0) {
        return 0;
    }
    return _dispatch_semaphore_wait_slow(dsema, timeout);
}

该函数先把信号量的信号值原子减1，如果大于等于0，则表示有资源可用，直接返回；否则调用_dispatch_semaphore_wait_slow，等到FIFO队列中信号量的到来，直到timeout为止。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
static long _dispatch_semaphore_wait_slow(dispatch_semaphore_t dsema, dispatch_time_t timeout){
    mach_timespec_t _timeout;
    kern_return_t kr;
    uint64_t nsec;
    long orig;
    
again:
    _dispatch_semaphore_create_port(&dsema->dsema_port);

    switch (timeout) {
    default:
        do {
            // timeout() already calculates relative time left
            nsec = _dispatch_timeout(timeout);
            _timeout.tv_sec = (typeof(_timeout.tv_sec))(nsec / NSEC_PER_SEC);
            _timeout.tv_nsec = (typeof(_timeout.tv_nsec))(nsec % NSEC_PER_SEC);
            kr = slowpath(semaphore_timedwait(dsema->dsema_port, _timeout));
        } while (kr == KERN_ABORTED);

        if (kr != KERN_OPERATION_TIMED_OUT) {
            DISPATCH_SEMAPHORE_VERIFY_KR(kr);
            break;
        }
    // Fall through and try to undo what the fast path did to dsema->dsema_value
    case DISPATCH_TIME_NOW:
        while ((orig = dsema->dsema_value) < 0) {
            if (dispatch_atomic_cmpxchg(&dsema->dsema_value, orig, orig + 1)) {
                return KERN_OPERATION_TIMED_OUT;
            }
        }
    // Another thread called semaphore_signal().
        // Fall through and drain the wakeup.
    case DISPATCH_TIME_FOREVER:
        do {
            kr = semaphore_wait(dsema->dsema_port);
        } while (kr == KERN_ABORTED);
        DISPATCH_SEMAPHORE_VERIFY_KR(kr);
        break;
    }

    goto again;
}

进入_dispatch_semaphore_wait_slow之后，根据timeout的类型分三种情况：

  • 1、DISPATCH_TIME_NOW：检测后立即返回，不等待，若dsema_value小于0，那么将其增加1, 然后返回超时信号KERN_OPERATION_TIMED_OUT；

  • 2、DISPATCH_TIME_FOREVER：无线等待，直到dsema_value大于0，若有了信号，会跳转到again，重新执行本函数的流程；

    1
    2
    /* 没有超时的代码 */
    kr = semaphore_wait(dsema->dsema_port);

  • 3、计时等待，直到有信号到来或者超时。

    1
    2
    /* 有超时的代码 */
    kr = slowpath(semaphore_timedwait(dsema->dsema_port, _timeout));

dispatch_semaphore_signal

发送信号，或者增加一个信号量的计数。如果增加之前的信号值dsema_value大于等于0，说明没有线程因为该信号而阻塞，所以直接返回。

1
2
3
4
5
6
DISPATCH_NOINLINE long dispatch_semaphore_signal(dispatch_semaphore_t dsema){
    if (dispatch_atomic_inc(&dsema->dsema_value) > 0) {
        return 0;
    }
    return _dispatch_semaphore_signal_slow(dsema);
}

如果增加之前的信号值小于0，则调用_dispatch_semaphore_signal_slow，将会唤起一个在spatch_semaphore_wait中等待的线程。（如果有多个线程在等待，iOS会根据线程的优先级来判断具体唤醒哪个线程）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
DISPATCH_NOINLINE static long _dispatch_semaphore_signal_slow(dispatch_semaphore_t dsema){
    kern_return_t kr;
    
    _dispatch_semaphore_create_port(&dsema->dsema_port);

    // Before dsema_sent_ksignals is incremented we can rely on the reference
    // held by the waiter. However, once this value is incremented the waiter
    // may return between the atomic increment and the semaphore_signal(),
    // therefore an explicit reference must be held in order to safely access
    // dsema after the atomic increment.
    _dispatch_retain(dsema);
    
    dispatch_atomic_inc(&dsema->dsema_sent_ksignals);
    
    // 利用系统的信号量库实现发送信号量的功能
    kr = semaphore_signal(dsema->dsema_port);
    DISPATCH_SEMAPHORE_VERIFY_KR(kr);

    _dispatch_release(dsema);
    
    return 1;
}

总结

GCD信号量比较简单：记住这3个关键API：create、wait和signal即可

-------------本文结束 感谢您的阅读-------------

本文标题:GCD源码分析5 —— dispatch_semaphore篇

文章作者:lingyun

发布时间:2018年02月08日 - 13:02

最后更新:2018年02月12日 - 18:02

原始链接:https://tsuijunxi.github.io/2018/02/08/GCD源码分析5 —— dispatch-semaphore篇/

许可协议: 署名-非商业性使用-禁止演绎 4.0 国际 转载请保留原文链接及作者。

坚持原创技术分享，您的支持将鼓励我继续创作！

本文标题:GCD源码分析5 —— dispatch_semaphore篇

文章作者:lingyun

发布时间:2018年02月08日 - 13:02

最后更新:2018年02月12日 - 18:02

原始链接:https://tsuijunxi.github.io/2018/02/08/GCD源码分析5 —— dispatch-semaphore篇/

许可协议: 署名-非商业性使用-禁止演绎 4.0 国际 转载请保留原文链接及作者。