我写的由 GCD 代码支持的读写器锁导致并行测试中的死锁

Question

我在 GCD 中实现了这个读写器锁，但在并行测试中失败了。我可以解释为什么它失败了吗？

这是用于 iOS 开发的。该代码基于Objective C。我在GCD中编写了一个带有读写器锁的RWCache，用于数据保护。

@interface RWCache : NSObject

- (void)setObject:(id)object forKey:(id <NSCopying>)key;

- (id)objectForKey:(id <NSCopying>)key;

@end

@interface RWCache()
@property (nonatomic, strong) NSMutableDictionary *memoryStorage;
@property (nonatomic, strong) dispatch_queue_t storageQueue;
@end

@implementation RWCache

- (instancetype)init {
    self = [super init];
    if (self) {
        _memoryStorage = [NSMutableDictionary new];
        _storageQueue = dispatch_queue_create("Storage Queue", DISPATCH_QUEUE_CONCURRENT);
    }
    return self;
}

- (void)setObject:(id)object forKey:(id <NSCopying>)key {
    dispatch_barrier_async(self.storageQueue, ^{
        self.memoryStorage[key] = object;
    });
}

- (id)objectForKey:(id <NSCopying>)key {
    __block id object = nil;
    dispatch_sync(self.storageQueue, ^{
        object = self.memoryStorage[key];
    });
    return object;
}

@end


int main(int argc, const char * argv[]) {
    @autoreleasepool {
        RWCache *cache = [RWCache new];
        dispatch_queue_t testQueue = dispatch_queue_create("Test Queue", DISPATCH_QUEUE_CONCURRENT);
        dispatch_group_t group = dispatch_group_create();
        for (int i = 0; i < 100; i++) {
            dispatch_group_async(group, testQueue, ^{
                [cache setObject:@(i) forKey:@(i)];
            });
            dispatch_group_async(group, testQueue, ^{
                [cache objectForKey:@(i)];
            });
        }
        dispatch_group_wait(group, DISPATCH_TIME_FOREVER);
    }
    return 0;
}

如果没有死锁，程序退出0，否则程序挂起不退出。

Answer 1

问题不在于读取器/写入器模式，本身，，而是因为此代码中的一般线程爆炸。请参阅 WWDC 2015 视频 Building Responsive and Efficient Apps with GCD 中的“线程爆炸导致死锁”讨论。 WWDC 2016 Concurrent Programming With GCD in Swift 3 也是一个不错的视频。在这两个链接中，我会将您带到视频的相关部分，但都值得完整观看。

归根结底，您正在耗尽 GCD 线程池中数量非常有限的工作线程。只有 64 个。但是你有 100 个带屏障的写入，这意味着在该队列上的所有其他内容完成之前，调度的块无法运行。它们散布着 100 次读取，因为它们是同步的，所以会阻塞您调度它的工作线程，直到它返回。

让我们将其简化为能体现问题的更简单的东西：

dispatch_queue_t queue1 = dispatch_queue_create("queue1", DISPATCH_QUEUE_CONCURRENT);
dispatch_queue_t queue2 = dispatch_queue_create("queue2", DISPATCH_QUEUE_CONCURRENT);
for (int i = 0; i < 100; i++) {
    dispatch_async(queue2, ^{
        dispatch_barrier_async(queue1, ^{
            NSLog(@"barrier async %d", i);
        });
        dispatch_sync(queue1, ^{
            NSLog(@"sync %d", i);
        });
    });
}
NSLog(@"done dispatching all blocks to queue1");

这会产生类似的东西：

开始
完成将所有块分派到 queue1
屏障异步 0
同步 0

它会死锁。

但如果我们将其限制为不超过 30 个项目可以在 queue2 上同时运行，那么问题就消失了：

dispatch_queue_t queue1 = dispatch_queue_create("queue1", DISPATCH_QUEUE_CONCURRENT);
dispatch_queue_t queue2 = dispatch_queue_create("queue2", DISPATCH_QUEUE_CONCURRENT);
dispatch_semaphore_t semaphore = dispatch_semaphore_create(30);

for (int i = 0; i < 100; i++) {
    dispatch_semaphore_wait(semaphore, DISPATCH_TIME_FOREVER);
    dispatch_async(queue2, ^{
        dispatch_barrier_async(queue1, ^{
            NSLog(@"barrier async %d", i);
        });
        dispatch_sync(queue1, ^{
            NSLog(@"sync %d", i);
        });
        dispatch_semaphore_signal(semaphore);
    });
}
NSLog(@"done dispatching all blocks to queue1");

或者，另一种方法是使用dispatch_apply，它实际上是一个并行化的for 循环，但在任何给定时刻将并发任务的数量限制为机器上的核心数量（使我们远低于阈值耗尽工作线程）：

dispatch_queue_t queue1 = dispatch_queue_create("queue1", DISPATCH_QUEUE_CONCURRENT);
dispatch_queue_t queue2 = dispatch_queue_create("queue2", DISPATCH_QUEUE_CONCURRENT);

dispatch_apply(100, queue2, ^(size_t i) {
    dispatch_barrier_async(queue1, ^{
        NSLog(@"barrier async %ld", i);
    });
    dispatch_sync(queue1, ^{
        NSLog(@"sync %ld", i);
    });
});
NSLog(@"done dispatching all blocks to queue1");