如何在Swift中实现线程安全HashTable (PhoneBook)数据结构？

Question

我正在尝试实现一个线程安全的PhoneBook对象。电话簿应该能够添加一个人，并根据他们的名字和phoneNumber查找一个人。从实现的角度来看，这只是涉及两个哈希表，一个是关联名称-> Person，另一个是关联phone# -> Person。

请注意，我希望这个对象是threadSafe。这意味着我希望能够在PhoneBook中支持并发查找，同时确保一次只能向PhoneBook添加一个人。这是基本的读者群问题，我正试图使用GrandCentralDispatch和调度障碍来解决这个问题。不过，当我遇到问题时，我正在努力解决这个问题。以下是我的Swift操场代码：

//: Playground - noun: a place where people can play

import UIKit
import PlaygroundSupport

PlaygroundPage.current.needsIndefiniteExecution = true

public class Person: CustomStringConvertible {
    public var description: String {
        get {
            return "Person: \(name), \(phoneNumber)"
        }
    }

    public var name: String
    public var phoneNumber: String
    private var readLock = ReaderWriterLock()

    public init(name: String, phoneNumber: String) {
        self.name = name
        self.phoneNumber = phoneNumber
    }


    public func uniquePerson() -> Person {
        let randomID = UUID().uuidString
        return Person(name: randomID, phoneNumber: randomID)
    }
}

public enum Qos {
    case threadSafe, none
}

public class PhoneBook {

    private var qualityOfService: Qos = .none
    public var nameToPersonMap = [String: Person]()
    public var phoneNumberToPersonMap = [String: Person]()
    private var readWriteLock = ReaderWriterLock()


    public init(_ qos: Qos) {
        self.qualityOfService = qos
    }

    public func personByName(_ name: String) -> Person? {
        var person: Person? = nil
        if qualityOfService == .threadSafe {
            readWriteLock.concurrentlyRead { [weak self] in
                guard let strongSelf = self else { return }
                person = strongSelf.nameToPersonMap[name]
            }
        } else {
            person = nameToPersonMap[name]
        }

        return person
    }

    public func personByPhoneNumber( _ phoneNumber: String) -> Person? {
        var person: Person? = nil
        if qualityOfService == .threadSafe {
            readWriteLock.concurrentlyRead { [weak self] in
                guard let strongSelf = self else { return }
                person = strongSelf.phoneNumberToPersonMap[phoneNumber]
            }
        } else {
            person = phoneNumberToPersonMap[phoneNumber]
        }

        return person
    }

    public func addPerson(_ person: Person) {
        if qualityOfService == .threadSafe {
            readWriteLock.exclusivelyWrite { [weak self] in
                guard let strongSelf = self else { return }
                strongSelf.nameToPersonMap[person.name] = person
                strongSelf.phoneNumberToPersonMap[person.phoneNumber] = person
            }
        } else {
            nameToPersonMap[person.name] = person
            phoneNumberToPersonMap[person.phoneNumber] = person
        }
    }

}


// A ReaderWriterLock implemented using GCD and OS Barriers.
public class ReaderWriterLock {

    private let concurrentQueue = DispatchQueue(label: "com.ReaderWriterLock.Queue", attributes: DispatchQueue.Attributes.concurrent)
    private var writeClosure: (() -> Void)!

    public func concurrentlyRead(_ readClosure: (() -> Void)) {
        concurrentQueue.sync {
            readClosure()
        }
    }

    public func exclusivelyWrite(_ writeClosure: @escaping (() -> Void)) {
        self.writeClosure = writeClosure
        concurrentQueue.async(flags: .barrier) { [weak self] in
            guard let strongSelf = self else { return }
            strongSelf.writeClosure()
        }
    }

}

// MARK: Testing the synchronization and thread-safety

for _ in 0..<5 {
    let iterations = 1000
    let phoneBook = PhoneBook(.none)

    let concurrentTestQueue = DispatchQueue(label: "com.PhoneBookTest.Queue", attributes: DispatchQueue.Attributes.concurrent)
    for _ in 0..<iterations {
        let person = Person(name: "", phoneNumber: "").uniquePerson()
        concurrentTestQueue.async {
            phoneBook.addPerson(person)
        }
    }

    sleep(10)
    print(phoneBook.nameToPersonMap.count)
}

为了测试我的代码，我运行了1000个并发线程，这些线程只需向PhoneBook添加一个新的人。每个人都是唯一的，所以在1000个线程完成后，我期望PhoneBook包含1000个计数。每次执行写操作时，我都会执行一个dispatch_barrier调用，更新哈希表并返回。据我所知，这是我们所需要做的；然而，在重复运行了1000个线程之后，我得到了PhoneBook中的条目数量不一致，而且到处都是：

Phone Book Entries: 856
Phone Book Entries: 901
Phone Book Entries: 876
Phone Book Entries: 902
Phone Book Entries: 912

有人能帮我弄清楚这是怎么回事吗？我的锁定代码有什么问题吗?或者更糟糕的是，我的测试是如何构造的？我是非常新的这个多线程的问题空间，谢谢！

Answer 1

问题是你的ReaderWriterLock。您正在将writeClosure保存为属性，然后异步分配一个调用保存属性的闭包。但是，如果在中间一段时间内出现了另一个exclusiveWrite，您的writeClosure属性将被新的闭包所取代。

在本例中，这意味着您可以多次添加相同的Person。而且因为你使用的是字典，这些副本有相同的键，因此不会导致你看到所有的1000个条目。

您实际上可以简化ReaderWriterLock，完全消除该属性。我还会使concurrentRead成为一个泛型，返回值(就像sync那样)，并重新抛出任何错误(如果有的话)。

public class ReaderWriterLock {
    private let queue = DispatchQueue(label: "com.domain.app.rwLock", attributes: .concurrent)
    
    public func concurrentlyRead<T>(_ block: (() throws -> T)) rethrows -> T {
        return try queue.sync {
            try block()
        }
    }
    
    public func exclusivelyWrite(_ block: @escaping (() -> Void)) {
        queue.async(flags: .barrier) {
            block()
        }
    }
}

另外几点，不相关的意见：

1. 顺便说一句，这个简化的ReaderWriterLock恰好解决了另一个问题。我们现在已经删除的writeClosure属性可以很容易地引入一个强大的引用周期。 是的，您在使用[weak self]时非常谨慎，所以没有任何强大的引用周期，但这是可能的。我建议，无论您在何处使用闭包属性，在您完成闭包操作时，都应该将该闭包属性设置为nil，因此任何可能意外地导致闭包的强引用都将被解析。这样，一个持久的强参考周期是不可能的。(另外，闭包本身和它拥有的任何局部变量或其他外部引用都将被解析。)
2. 你睡了10秒。这应该足够了，但我建议不要只添加随机的sleep调用(因为您永远无法100%确定)。幸运的是，您有一个并发队列，因此可以使用： ConcurrentTestQueue.async(标志：.barrier) { print(phoneBook.count) } 因为有了这个屏障，它就会等到你放在队列上的所有东西都完成了。
3. 注意，我不只是打印nameToPersonMap.count。这个数组在PhoneBook中已经被仔细地同步了，所以您不能让随机的外部类直接访问它而不需要同步。 每当您有一些内部同步的属性时，它应该是private，然后创建一个线程安全函数/变量来检索所需的任何内容： 公共类PhoneBook {私有变量nameToPersonMap = 串:人私有变量phoneNumberToPersonMap = 串:人 .变量计数: Int {返回readWriteLock.concurrentlyRead { nameToPersonMap.count }}
4. 您说您正在测试线程安全性，但随后使用.none选项创建了.none(实现没有线程安全性)。在这种情况下，我认为会有问题。您必须使用PhoneBook选项创建您的.threadSafe。
5. 您有许多strongSelf模式。那可不太好喝。在Swift中通常不需要它，因为您可以使用[weak self]，然后执行可选的链接。

把这一切结合起来，这是我最后的操场：

PlaygroundPage.current.needsIndefiniteExecution = true

public class Person {
    public let name: String
    public let phoneNumber: String
    
    public init(name: String, phoneNumber: String) {
        self.name = name
        self.phoneNumber = phoneNumber
    }
    
    public static func uniquePerson() -> Person {
        let randomID = UUID().uuidString
        return Person(name: randomID, phoneNumber: randomID)
    }
}

extension Person: CustomStringConvertible {
    public var description: String {
        return "Person: \(name), \(phoneNumber)"
    }
}

public enum ThreadSafety { // Changed the name from Qos, because this has nothing to do with quality of service, but is just a question of thread safety
    case threadSafe, none
}

public class PhoneBook {
    
    private var threadSafety: ThreadSafety
    private var nameToPersonMap = [String: Person]()        // if you're synchronizing these, you really shouldn't expose them to the public
    private var phoneNumberToPersonMap = [String: Person]() // if you're synchronizing these, you really shouldn't expose them to the public
    private var readWriteLock = ReaderWriterLock()
    
    public init(_ threadSafety: ThreadSafety) {
        self.threadSafety = threadSafety
    }
    
    public func personByName(_ name: String) -> Person? {
        if threadSafety == .threadSafe {
            return readWriteLock.concurrentlyRead { [weak self] in
                self?.nameToPersonMap[name]
            }
        } else {
            return nameToPersonMap[name]
        }
    }
    
    public func personByPhoneNumber(_ phoneNumber: String) -> Person? {
        if threadSafety == .threadSafe {
            return readWriteLock.concurrentlyRead { [weak self] in
                self?.phoneNumberToPersonMap[phoneNumber]
            }
        } else {
            return phoneNumberToPersonMap[phoneNumber]
        }
    }
    
    public func addPerson(_ person: Person) {
        if threadSafety == .threadSafe {
            readWriteLock.exclusivelyWrite { [weak self] in
                self?.nameToPersonMap[person.name] = person
                self?.phoneNumberToPersonMap[person.phoneNumber] = person
            }
        } else {
            nameToPersonMap[person.name] = person
            phoneNumberToPersonMap[person.phoneNumber] = person
        }
    }
    
    var count: Int {
        return readWriteLock.concurrentlyRead {
            nameToPersonMap.count
        }
    }
}

// A ReaderWriterLock implemented using GCD concurrent queue and barriers.

public class ReaderWriterLock {
    private let queue = DispatchQueue(label: "com.domain.app.rwLock", attributes: .concurrent)
    
    public func concurrentlyRead<T>(_ block: (() throws -> T)) rethrows -> T {
        return try queue.sync {
            try block()
        }
    }
    
    public func exclusivelyWrite(_ block: @escaping (() -> Void)) {
        queue.async(flags: .barrier) {
            block()
        }
    }
}


for _ in 0 ..< 5 {
    let iterations = 1000
    let phoneBook = PhoneBook(.threadSafe)
    
    let concurrentTestQueue = DispatchQueue(label: "com.PhoneBookTest.Queue", attributes: .concurrent)
    for _ in 0..<iterations {
        let person = Person.uniquePerson()
        concurrentTestQueue.async {
            phoneBook.addPerson(person)
        }
    }
    
    concurrentTestQueue.async(flags: .barrier) {
        print(phoneBook.count)
    }
}

就我个人而言，我倾向于更进一步

• 将同步移到泛型类中；以及
• 将模型更改为Person对象的数组，以便：
  • 该模型支持多个具有相同或电话号码的人；以及
  • 如果需要，可以使用值类型。

例如：

public struct Person {
    public let name: String
    public let phoneNumber: String
    
    public static func uniquePerson() -> Person {
        return Person(name: UUID().uuidString, phoneNumber: UUID().uuidString)
    }
}

public struct PhoneBook {
    
    private var synchronizedPeople = Synchronized([Person]())
    
    public func people(name: String? = nil, phone: String? = nil) -> [Person]? {
        return synchronizedPeople.value.filter {
            (name == nil || $0.name == name) && (phone == nil || $0.phoneNumber == phone)
        }
    }
    
    public func append(_ person: Person) {
        synchronizedPeople.writer { people in
            people.append(person)
        }
    }
    
    public var count: Int {
        return synchronizedPeople.reader { $0.count }
    }
}

/// A structure to provide thread-safe access to some underlying object using reader-writer pattern.

public class Synchronized<T> {
    /// Private value. Use `public` `value` computed property (or `reader` and `writer` methods)
    /// for safe, thread-safe access to this underlying value.
    
    private var _value: T
    
    /// Private reader-write synchronization queue
    
    private let queue = DispatchQueue(label: Bundle.main.bundleIdentifier! + ".synchronized", qos: .default, attributes: .concurrent)
    
    /// Create `Synchronized` object
    ///
    /// - Parameter value: The initial value to be synchronized.
    
    public init(_ value: T) {
        _value = value
    }
    
    /// A threadsafe variable to set and get the underlying object, as a convenience when higher level synchronization is not needed        
    
    public var value: T {
        get { reader { $0 } }
        set { writer { $0 = newValue } }
    }
    
    /// A "reader" method to allow thread-safe, read-only concurrent access to the underlying object.
    ///
    /// - Warning: If the underlying object is a reference type, you are responsible for making sure you
    ///            do not mutating anything. If you stick with value types (`struct` or primitive types),
    ///            this will be enforced for you.
    
    public func reader<U>(_ block: (T) throws -> U) rethrows -> U {
        return try queue.sync { try block(_value) }
    }
    
    /// A "writer" method to allow thread-safe write with barrier to the underlying object
    
    func writer(_ block: @escaping (inout T) -> Void) {
        queue.async(flags: .barrier) {
            block(&self._value)
        }
    }
}

Answer 2

在某些情况下，您使用的是可能的NSCache类。文档声称它是线程安全的：

您可以从不同的线程中添加、删除和查询缓存中的项，而不必自己锁定缓存。

下面是一个文章，它描述了与NSCache相关的非常有用的技巧

Answer 3

我不认为你用错了。)

原始(在macos上)生成：

0  swift                    0x000000010c9c536a PrintStackTraceSignalHandler(void*) + 42
1  swift                    0x000000010c9c47a6 SignalHandler(int) + 662
2  libsystem_platform.dylib 0x00007fffbbdadb3a _sigtramp + 26
3  libsystem_platform.dylib 000000000000000000 _sigtramp + 1143284960
4  libswiftCore.dylib       0x0000000112696944 _T0SSwcp + 36
5  libswiftCore.dylib       0x000000011245fa92 _T0s24_VariantDictionaryBufferO018ensureUniqueNativeC0Sb11reallocated_Sb15capacityChangedtSiF + 1634
6  libswiftCore.dylib       0x0000000112461fd2 _T0s24_VariantDictionaryBufferO17nativeUpdateValueq_Sgq__x6forKeytF + 1074

如果从.concurrent队列中删除‘ReaderWriter’，则如果恢复.concurrent，则“问题消失”.©，但将写入端的异步调用更改为同步：

swift(10504,0x70000896f000) malloc：*对象0x7fcaa440cee8:释放对象后可能修改了不正确的校验和。

如果它不迅速的话，这会有点令人吃惊吗？我深入研究，通过插入一个散列函数，将基于“string”的数组替换为Int数组，将睡眠(10)替换为一个屏障分派，以清除任何滞后的块，这使得它更具有可重复性，更有帮助：

x(10534,0x700000f01000) malloc：*对象错误对象0x7f8c9ee008:释放对象后可能修改了不正确的校验和。

但是，当对源的搜索显示没有malloc或空闲时，堆栈转储可能更有用。

无论如何，解决你的问题的最好方法是:用go代替，这实际上是有意义的。