iOS多线程：GCD源码分析<四>-dispatch_semaphore

发表于 | 分类于

dispatch_semaphore

概述

dispatch_semaphore是持有计数的信号，该信号是多线程编程中的计数类型信号。有3个api：create, wait signal。

使用

  1. 对常用资源进行加锁操作，防止多线程访问修改数据导致出现结果不一致甚至崩溃的问题：
1
2
3
4
5
6
//在init等函数初始化
_lock = dispatch_semaphore_create(1); 
dispatch_semaphore_wait(_lock, DISPATCH_TIME_FOREVER); 
//修改Array或字典等数据的信息

dispatch_semaphore_signal(_lock);
  1. 串行执行一系列异步任务
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
    - (void)chainRequestCurrentConfig {
    dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{
        NSArray *list = @[@"1",@"2",@"3"];
        dispatch_semaphore_t semaphore = dispatch_semaphore_create(0);
        [list enumerateObjectsUsingBlock:^(id  _Nonnull obj, NSUInteger idx, BOOL * _Nonnull stop) {
            [self fetchConfigurationWithCompletion:^(NSDictionary *dict) {
                dispatch_semaphore_signal(semaphore);
            }];
            //注意如果signal没有被调用线程会被挂起，造成死锁
            dispatch_semaphore_wait(semaphore, DISPATCH_TIME_FOREVER);
        }];
    });
}
- (void)fetchConfigurationWithCompletion:(void(^)(NSDictionary *dict))completion {
    //AFNetworking或其他网络请求库
    dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{
        //模拟网络请求
        sleep(2);
        !completion ? nil : completion(nil);
    });
}

源码分析

dispatch_semaphore_t
先看一下dispatch_semaphore_s的定义

1
2
3
4
5
6
7
8
9
10
11
12
struct dispatch_semaphore_s {
    DISPATCH_STRUCT_HEADER(semaphore);
    semaphore_t dsema_port;    //等同于mach_port_t信号
    long dsema_orig;           //初始化的信号量值
    long volatile dsema_value; //当前信号量值
    union {
        long volatile dsema_sent_ksignals;
        long volatile dsema_group_waiters;
    };
    struct dispatch_continuation_s *volatile dsema_notify_head; //notify的链表头部
    struct dispatch_continuation_s *volatile dsema_notify_tail; //notify的链表尾部
};

dispatch_semaphore_create

dispatch_semaphore_create 用来创建信号量，创建时要指定value，内部会将value的值存储到 dsema_value(当前值) 和 dsema_orig(初始值) 中，value的值必须大于等于0。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
dispatch_semaphore_t dispatch_semaphore_create(long value) {
    dispatch_semaphore_t dsema;
    if (value < 0) {
       //value值需大于或等于0
        return NULL;
    }
  //申请dispatch_semaphore_t的内存
    dsema = (dispatch_semaphore_t)_dispatch_alloc(DISPATCH_VTABLE(semaphore),
            sizeof(struct dispatch_semaphore_s) -
            sizeof(dsema->dsema_notify_head) -
            sizeof(dsema->dsema_notify_tail));
    //调用初始化函数
    _dispatch_semaphore_init(value, dsema);
    return dsema;
}
//初始化结构体信息
static void _dispatch_semaphore_init(long value, dispatch_object_t dou) {
    dispatch_semaphore_t dsema = dou._dsema;
    dsema->do_next = (dispatch_semaphore_t)DISPATCH_OBJECT_LISTLESS;
    dsema->do_targetq = dispatch_get_global_queue(
            DISPATCH_QUEUE_PRIORITY_DEFAULT, 0);
    dsema->dsema_value = value; //设置信号量的当前value值
    dsema->dsema_orig = value;  //设置信号量的初始value值
}

接着来看Dispatch Semaphore很容易忽略也是最容易造成App崩溃的地方，即信号量的释放。

创建Semaphore时，会将do_vtable 指向_dispatch_semaphore_vtable， _dispatch_semaphore_vtable的结构体定义了信号量销毁时会执行_dispatch_semaphore_dispose方法。相关代码如下：

1
2
3
4
5
6
7
//semaphore的vtable定义
DISPATCH_VTABLE_INSTANCE(semaphore,
    .do_type = DISPATCH_SEMAPHORE_TYPE,
    .do_kind = "semaphore",
    .do_dispose = _dispatch_semaphore_dispose,  //销毁时执行的回调函数
    .do_debug = _dispatch_semaphore_debug,      //debug函数
);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
//释放信号量的函数
void _dispatch_semaphore_dispose(dispatch_object_t dou) {
    dispatch_semaphore_t dsema = dou._dsema;

    if (dsema->dsema_value < dsema->dsema_orig) {
       //Warning:信号量还在使用的时候销毁会造成崩溃
        DISPATCH_CLIENT_CRASH(
                "Semaphore/group object deallocated while in use");
    }
    kern_return_t kr;
    if (dsema->dsema_port) {
        kr = semaphore_destroy(mach_task_self(), dsema->dsema_port);
        DISPATCH_SEMAPHORE_VERIFY_KR(kr);
    }
}

如果销毁时信号量还在使用，那么dsema_value值会小于dsema_orig，引起崩溃。如下代码：

1
2
3
4
dispatch_semaphore_t semephore = dispatch_semaphore_create(1);
dispatch_semaphore_wait(semephore, DISPATCH_TIME_FOREVER);
//重新赋值或者将semephore = nil都会造成崩溃,因为此时信号量还在使用中
semephore = dispatch_semaphore_create(0);

1
2
3
4
5
6
7
8
9
10
11
- (void)func {
    dispatch_semaphore_t semaphore = dispatch_semaphore_create(1);
        dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{

        sleep(2);
        NSLog(@"111");
//        dispatch_semaphore_signal(semaphore);
    });
    dispatch_wait(semaphore, DISPATCH_TIME_FOREVER);
}
//作用域结束时信号量会销毁，触发`dispose`方法，造成崩溃

dispatch_semaphore_wait

1
2
3
4
5
6
7
8
9
10
long dispatch_semaphore_wait(dispatch_semaphore_t dsema, dispatch_time_t timeout){
//value原子减1并返回value的值
    long value = dispatch_atomic_dec2o(dsema, dsema_value, acquire);
    //如果大于等于0 立即返回
    if (fastpath(value >= 0)) {
        return 0;
    }
    //否则调用这个方法，等待信号量唤醒或超时。
    return _dispatch_semaphore_wait_slow(dsema, 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
43
44
45
static long _dispatch_semaphore_wait_slow(dispatch_semaphore_t dsema,
        dispatch_time_t timeout) {
    long orig;
    mach_timespec_t _timeout;
    kern_return_t kr;
again:
    orig = dsema->dsema_sent_ksignals;
    while (orig) {
        if (dispatch_atomic_cmpxchgvw2o(dsema, dsema_sent_ksignals, orig,
                orig - 1, &orig, relaxed)) {
            return 0;
        }
    }

    _dispatch_semaphore_create_port(&dsema->dsema_port);
    switch (timeout) {
    default:
        do {
            uint64_t nsec = _dispatch_timeout(timeout);
            _timeout.tv_sec = (typeof(_timeout.tv_sec))(nsec / NSEC_PER_SEC);
            _timeout.tv_nsec = (typeof(_timeout.tv_nsec))(nsec % NSEC_PER_SEC);
            kr = slowpath(semaphore_timedwait(dsema->dsema_port, _timeout));
        } while (kr == KERN_ABORTED);

        if (kr != KERN_OPERATION_TIMED_OUT) {
            DISPATCH_SEMAPHORE_VERIFY_KR(kr);
            break;
        }
    case DISPATCH_TIME_NOW:
        orig = dsema->dsema_value;
        while (orig < 0) {
            if (dispatch_atomic_cmpxchgvw2o(dsema, dsema_value, orig, orig + 1,
                    &orig, relaxed)) {
                return KERN_OPERATION_TIMED_OUT;
            }
        }
    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: 调用系统semaphore_wait方法，直到收到signal调用。
  3. default：调用内核方法semaphore_timedwait计时等待，直到有信号到来或者超时了。

dispatch_semaphore_signal

1
2
3
4
5
6
7
8
9
10
11
12
long dispatch_semaphore_signal(dispatch_semaphore_t dsema) {
//将dsema_value 原子加一并赋值于value
    long value = dispatch_atomic_inc2o(dsema, dsema_value, release);
    if (fastpath(value > 0)) {
        return 0;
    }
    if (slowpath(value == LONG_MIN)) {
       //Warning：value值有误会造成崩溃，详见dispatch_group的分析
        DISPATCH_CLIENT_CRASH("Unbalanced call to dispatch_semaphore_signal()");
    }
    return _dispatch_semaphore_signal_slow(dsema);
}

首先将semaphore_value原子加一，如果大于0直接返回0，否则进入_dispatch_semaphore_signal_slow方法，该函数会调用内核的semaphore_signal函数唤醒等待中的线程。

1
2
3
4
5
6
7
8
9
10
long _dispatch_semaphore_signal_slow(dispatch_semaphore_t dsema) {
    _dispatch_retain(dsema);
    (void)dispatch_atomic_inc2o(dsema, dsema_sent_ksignals, relaxed);
    _dispatch_semaphore_create_port(&dsema->dsema_port);
    kern_return_t kr = semaphore_signal(dsema->dsema_port);
    DISPATCH_SEMAPHORE_VERIFY_KR(kr);

    _dispatch_release(dsema);
    return 1;
}

总结

Dispatch Semaphore 信号量wait方法会将信号量值减1，如果大于等于0就立即返回，否则等待信号量唤醒或超时；signal方法会将信号量值加1，如果value大于0立即返回，否则会唤醒某个等待中的线程。

需要注意信号量销毁或重新创建的时候如果还在使用会引起崩溃。