swift - 使用队列和信号量进行并发和属性包装？

Question

我正在尝试创建一个线程安全的属性包装器。我只能认为GCD队列和信号量是最快捷，最可靠的方法。信号量是不是性能更高(如果是真的)，还是出于并发性而使用一个信号量另一个原因？

以下是原子属性包装器的两种变体:

@propertyWrapper
struct Atomic<Value> {
    private var value: Value
    private let queue = DispatchQueue(label: "Atomic serial queue")

    var wrappedValue: Value {
        get { queue.sync { value } }
        set { queue.sync { value = newValue } }
    }

    init(wrappedValue value: Value) {
        self.value = value
    }
}

@propertyWrapper
struct Atomic2<Value> {
    private var value: Value
    private var semaphore = DispatchSemaphore(value: 1)

    var wrappedValue: Value {
        get {
            semaphore.wait()
            let temp = value
            semaphore.signal()
            return temp
        }

        set {
            semaphore.wait()
            value = newValue
            semaphore.signal()
        }
    }

    init(wrappedValue value: Value) {
        self.value = value
    }
}

struct MyStruct {
    @Atomic var counter = 0
    @Atomic2 var counter2 = 0
}

func test() {
    var myStruct = MyStruct()

    DispatchQueue.concurrentPerform(iterations: 1000) {
        myStruct.counter += $0
        myStruct.counter2 += $0
   }
}

如何对其进行适当的测试和衡量，以查看这两种实现之间的差异以及它们是否有效？

Answer 1

FWIW的另一种选择是具有并发队列的读取器-写入器模式，其中读取是同步完成的，但允许相对于其他读取并发运行，但是写入是异步完成的，但是有一个障碍(即，相对于其他任何事件都不是同时进行的)读取或写入):

@propertyWrapper
class Atomic<Value> {
    private var value: Value
    private let queue = DispatchQueue(label: "com.domain.app.atomic", attributes: .concurrent)

    var wrappedValue: Value {
        get { queue.sync { value } }
        set { queue.async(flags: .barrier) { self.value = newValue } }
    }

    init(wrappedValue value: Value) {
        self.value = value
    }
}

另一个是 NSLock:

@propertyWrapper
struct Atomic<Value> {
    private var value: Value
    private var lock = NSLock()

    var wrappedValue: Value {
        get { lock.synchronized { value } }
        set { lock.synchronized { value = newValue } }
    }

    init(wrappedValue value: Value) {
        self.value = value
    }
}

在哪里

extension NSLocking {
    func synchronized<T>(block: () throws -> T) rethrows -> T {
        lock()
        defer { unlock() }
        return try block()
    }
}

或者，您可以使用不公平的锁:

@propertyWrapper
struct SynchronizedUnfairLock<Value> {
    private var value: Value
    private var lock = UnfairLock()

    var wrappedValue: Value {
        get { lock.synchronized { value } }
        set { lock.synchronized { value = newValue } }
    }

    init(wrappedValue value: Value) {
        self.value = value
    }
}

在哪里

// One should not use `os_unfair_lock` directly in Swift (because Swift
// can move `struct` types), so we'll wrap it in a `UnsafeMutablePointer`.
// See https://github.com/apple/swift/blob/88b093e9d77d6201935a2c2fb13f27d961836777/stdlib/public/Darwin/Foundation/Publishers%2BLocking.swift#L18
// for stdlib example of this pattern.

final class UnfairLock: NSLocking {
    private let unfairLock: UnsafeMutablePointer<os_unfair_lock> = {
        let pointer = UnsafeMutablePointer<os_unfair_lock>.allocate(capacity: 1)
        pointer.initialize(to: os_unfair_lock())
        return pointer
    }()

    deinit {
        unfairLock.deinitialize(count: 1)
        unfairLock.deallocate()
    }

    func lock() {
        os_unfair_lock_lock(unfairLock)
    }

    func tryLock() -> Bool {
        os_unfair_lock_trylock(unfairLock)
    }

    func unlock() {
        os_unfair_lock_unlock(unfairLock)
    }
}

我们应该认识到，尽管这些以及您自己提供了原子性，但您必须小心，因为根据您的使用方式，它可能不是线程安全的。
考虑这个简单的实验，我们将整数增加一百万次:

func threadSafetyExperiment() {
    @Atomic var foo = 0

    DispatchQueue.global().async {
        DispatchQueue.concurrentPerform(iterations: 10_000_000) { _ in
            foo += 1
        }
        print(foo)
    }
}

您希望 foo等于10,000,000，但事实并非如此。这是因为“获取值并增加值并保存它”的整个交互过程都需要包装在一个同步机制中。
但是您可以添加一个原子增量方法:

extension Atomic where Value: Numeric {
    mutating func increment(by increment: Value) {
        lock.synchronized { value += increment }
    }
}

然后这很好用:

func threadSafetyExperiment() {
    @Atomic var foo = 0

    DispatchQueue.global().async {
        DispatchQueue.concurrentPerform(iterations: iterations) { _ in
            _foo.increment(by: 1)
        }
        print(foo)
    }
}

How can they be properly tested and measured to see the difference between the two implementations and if they even work?

一些想法:

我建议进行1000次以上的迭代。您希望进行足够的迭代，以秒为单位而不是毫秒来衡量结果。在我的示例中，我使用了一千万次迭代。

单元测试框架非常适合进行正确性测试以及使用measure方法测量性能(该方法对每个单元测试重复进行10次性能测试，结果将被单元测试报告捕获):

因此，请创建一个具有单元测试目标的项目(或根据需要将单元测试目标添加到现有项目中)，然后创建单元测试，并使用Command + U执行它们。

如果您为目标编辑方案，则可以选择随机化测试顺序，以确保测试的执行顺序不会影响性能:

我还将使测试目标使用发布版本，以确保您正在测试优化的版本。

不用说，虽然我正在通过运行10m次迭代来对锁进行压力测试，但每次迭代都增加一，所以效率非常低。在每个线程上根本没有足够的工作来证明线程处理的开销是合理的。通常，将跨过数据集并在每个线程中执行更多迭代，并减少同步次数。
这样做的实际含义是，在精心设计的并行算法中，您在做足够多的工作来证明多个线程合理，从而减少了正在发生的同步数。因此，无法观察到不同同步技术中的微小差异。如果同步机制具有可观察到的性能差异，则可能表明并行化算法中存在更深层的问题。专注于减少同步，而不是使同步更快。