如何在 Swift 中实现线程安全哈希表 (PhoneBook) 数据结构?

How to implement a Thread Safe HashTable (PhoneBook) Data Structure in Swift?

我正在尝试实现一个线程安全的电话簿对象。 phone本子应该可以加人,根据名字和phone号码查人。从实现的角度来看,这只涉及两个哈希表,一个关联 name -> Person,另一个关联 phone# -> Person。

需要注意的是我希望这个对象是线程安全的。这意味着我希望能够支持电话簿中的并发查找,同时确保一次只有一个线程可以将一个人添加到电话簿。这是基本的 reader-writers 问题,我正在尝试使用 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 个并发线程,它们只是将一个新的 Person 添加到 PhoneBook。每个人都是独一无二的,所以在 1000 个线程完成后,我希望 PhoneBook 包含 1000 个计数。每次执行写入时,我都会执行 dispatch_barrier 调用、更新哈希表和 return。据我所知,这就是我们需要做的;然而,在 1000 个线程中重复 运行s 之后,我发现电话簿中的条目数量不一致并且到处都是:

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

谁能帮我弄清楚这是怎么回事?我的锁定代码有问题,或者更糟的是我的测试构建方式有问题吗?我对这个多线程问题很陌生 space,谢谢!

我不认为你用错了:)。

原始(在 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

如果您从 ReaderWriter 队列中删除“.concurrent”,“问题就会消失”。© 如果恢复 .concurrent,但将编写器端的异步调用更改为同步:

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

如果不是 swift 会有点令人吃惊吗? 我深入研究,通过插入哈希函数将基于“字符串”的数组替换为 Int 数组,将 sleep(10) 替换为屏障分派以清除任何滞后的块,这使得它更容易重现地崩溃,更有帮助:

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

但是当源代码搜索显示没有 malloc 或 free 时,堆栈转储可能更有用。

无论如何,解决问题的最佳方法:改用 go;它实际上是有道理的。

问题是你的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(flags: .barrier) { 
        print(phoneBook.count) 
    }
    

    由于这个障碍,它会一直等到你放入该队列的所有其他内容都完成。

  3. 注意,我不只是打印 nameToPersonMap.count。此数组已在 PhoneBook 内仔细同步,因此您不能让随机的外部 classes 在不同步的情况下直接访问它。

    每当您有一些要在内部同步的 属性 时,它应该是 private,然后创建一个 thread-safe function/variable 来检索您需要的任何内容:

    public class PhoneBook {
    
        private var nameToPersonMap = [String: Person]()
        private var phoneNumberToPersonMap = [String: Person]()
    
        ...
    
        var count: Int {
            return readWriteLock.concurrentlyRead {
                nameToPersonMap.count
            }
        }
    }
    
  4. 你说你正在测试线程安全,但随后创建了带有 .none 选项的 PhoneBook(没有达到 thread-safety)。在那种情况下,我预计会出现问题。您必须使用 .threadSafe 选项创建 PhoneBook

  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)
    }
}

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

  • 将同步移至通用class;和
  • 将模型更改为 Person 对象的数组,以便:
    • 模型支持多人同号或phone号;和
    • 如果需要,您可以使用值类型。

例如:

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 || [=14=].name == name) && (phone == nil || [=14=].phoneNumber == phone)
        }
    }
    
    public func append(_ person: Person) {
        synchronizedPeople.writer { people in
            people.append(person)
        }
    }
    
    public var count: Int {
        return synchronizedPeople.reader { [=14=].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 { [=14=] } }
        set { writer { [=14=] = 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)
        }
    }
}