当数组在Swift中有并发读取时，如何实现removeAtIndex的线程安全？

Question

我正在查看 this answer，它提供了用于并发读取的线程安全数组的代码。正如@tombardey 在评论中指出的那样，代码（下面的相关片段）并不完全安全：

public func removeAtIndex(index: Int) {

    self.accessQueue.async(flags:.barrier) {
        self.array.remove(at: index)
    }
}

public var count: Int {
    var count = 0

    self.accessQueue.sync {
        count = self.array.count
    }

    return count
}

...Say the sychronized array has one element, wouldn't this fail? if synchronizedArray.count == 1 { synchronizedArray.remove(at: 0) } It's a race condition, say two threads execute the statement. Both read a count of 1 concurrently, both enqueue a write block concurrently. The write blocks execute sequentially, the second one will fail... (cont.)

@Rob 回复：

@tombardey - You are absolutely right that this level of synchronization (at the property/method level) is frequently insufficient to achieve true thread-safety in broader applications. Your example is easily solved (by adding an method that dispatches block to the queue), but there are others that aren't (e.g. "synchronized" array simultaneously used by a UITableViewDataSource and mutated by some background operation). In those cases, you have to implement your own higher-level synchronization. But the above technique is nonetheless very useful in certain, highly constrained situations.

我正在努力弄清楚@Rob 所说的“您的示例很容易解决（通过添加一个将块分派到队列的方法）”的意思。我有兴趣查看此方法（或任何其他）技术的示例实现来解决问题。

Answer 1

我发现发布的代码和示例有几个问题：

函数 removeAtIndex 不检查它是否真的可以在提供的索引处删除。所以应该改成

public func removeAtIndex(index: Int) {

    // Check if it even makes sense to schedule an update
    // This is optional, but IMO just a better practice
    guard count > index else { return }

    self.accessQueue.async(flags: .barrier) {

        // Check again before removing to make sure array didn't change
        // Here we can actually check the size of the array, since other threads are blocked
        guard self.array.count > index else { return }
        self.array.remove(at: index)
    }
}

使用线程安全的 class 也意味着您使用 一个操作 来检查和操作一个项目是线程安全的。因此，如果您检查数组大小然后将其删除，就会破坏线程安全信封，这不是 class 的正确用法。特殊情况 synchronizedArray.count == 1 { synchronizedArray.remove(at: 0) } 已通过调整上述功能解决（您不再需要检查计数，因为功能已经这样做了）。但是，如果您仍然需要一个既可以验证计数又可以删除项目的函数，则必须在线程安全 class 中创建一个函数来执行这两个操作，而其他线程不可能修改数组之间。您甚至可能需要 2 个函数：synchronizedArray.getCountAndRemove（获取计数，然后删除）, and synchronizedArray.removeAndGetCount`（删除，然后获取计数）。
```
public func getCountAndRemoveAtIndex(index: Int) -> Int {

    var currentCount = count
    guard currentCount > index else { return currentCount }

    // Has to run synchronously to ensure the value is returned
    self.accessQueue.sync {

        currentCount = self.array.count
        guard currentCount > index else { break }
        self.array.remove(at: index)
    }

    return currentCount
}
```
一般来说，从多线程使用的数组的索引处删除项目是毫无意义的。您甚至不能确定要删除的是什么。也许在某些情况下它是有意义的，但通常更有意义的是通过某种逻辑（例如特定值）删除，或者具有 returns 它删除的项目的值的函数（例如 func getAndRemoveAtIndex(index: Int) -> T)

始终测试每个功能及其组合。例如，如果原始发布者像这样测试删除：

let array = SynchronizedArray<Int>()
array.append(newElement: 1)
array.append(newElement: 2)
array.append(newElement: 3)

DispatchQueue.concurrentPerform(iterations: 5) {_ in

    array.removeAtIndex(index: 0)
}

他会在 5 个线程中的 2 个中得到 Fatal error: Index out of range: file Swift/Array.swift, line 1298，因此很明显这个函数的原始实现是不正确的。使用我上面发布的函数尝试相同的测试，您会看到不同之处。

顺便说一句，我们只讨论 removeAtIndex，但 subscript 也有类似的问题。但有趣的是 first() 实现正确。

Answer 2

这是一个很好的例子，说明了为什么对单个属性的“原子”可变操作很少是足够的，而且如果不小心添加就会很危险。

此示例中的根本问题是，每当修改数组时，都会使现有索引无效。为了安全地使用索引，必须确保整个“获取索引，使用索引”操作是原子的。您不能只确保每个 piece 都是原子的。孤立地写 removeAtIndex 没有安全的方法，因为没有安全的方法来获取你传递的索引。在您获取索引和使用它之间，数组可能已经以任意方式更改。

要点是，没有像普通数组一样可以使用而不必担心并发问题的“线程安全（可变）数组”这样的东西。 “线程安全”可变数组不能 return 或接受索引，因为它的索引不稳定。究竟哪种数据结构合适取决于您要解决的问题。这里没有人回答。

在大多数情况下，答案是“较少并发”。与其尝试管理对单个数据结构的并发访问，不如考虑更大范围的“工作单元”，这些“工作单元”携带所有自己的数据并对其进行独占访问。将那些较大的工作单元放到队列中。（在很多情况下，即使这样也有些矫枉过正。如果您不仔细设计，添加货币经常会使事情变得更慢，您会感到震惊。）有关更多建议，请参阅 Modernizing Grand Central Dispatch Usage。

Answer 3

你说：

I am struggling to work out what @Rob means by “Your example is easily solved (by adding [a] method that dispatches block to the queue)”. I would be interested to see an example implementation of this method (or any other) technique to solve the problem.

让我们扩展我在回答您的其他问题时发布的示例（参见中的第 3 点），添加更多 Array 方法：

class SynchronizedArray<T> {
    private var array: [T]
    private let accessQueue = DispatchQueue(label: "com.domain.app.reader-writer", attributes: .concurrent)

    init(_ array: [T] = []) {
        self.array = array
    }

    subscript(index: Int) -> T {
        get { reader { [=10=][index] } }
        set { writer { [=10=][index] = newValue } }
    }

    var count: Int {
        reader { [=10=].count }
    }

    func append(newElement: T) {
        writer { [=10=].append(newElement) }
    }

    func remove(at index: Int) {
        writer { [=10=].remove(at: index) }
    }

    func reader<U>(_ block: ([T]) throws -> U) rethrows -> U {
        try accessQueue.sync { try block(array) }
    }

    func writer(_ block: @escaping (inout [T]) -> Void) {
        accessQueue.async(flags: .barrier) { block(&self.array) }
    }
}

所以，假设您想要删除一个项目，如果数组中只有一个项目。考虑：

let numbers = SynchronizedArray([42])

...

if numbers.count == 1 { 
    numbers.remove(at: 0) 
}

这看起来很无辜，但它不是线程安全的。如果其他线程正在插入或删除值，您可能会出现竞争条件。例如，如果其他线程在您测试 count 和您删除该值之间附加了一个值。

您可以通过将整个操作（if 测试和随后的删除）包装在同步的单个块中来解决这个问题。因此你可以：

numbers.writer { array in
    if array.count == 1 { 
        array.remove(at: 0) 
    } 
}

此 writer 方法（在此 reader 基于写入器的同步中）是我所说的“将块分派到队列的方法”的示例。

现在，显然，您还可以为 SynchronizedArray 提供自己的方法来为您完成此操作，例如：

func safelyRemove(at index: Int) {
    writer { array in
        if index < array.count {
            array.remove(at: index)
        }
    }
}

那么你可以这样做：

numbers.safelyRemove(at: index)

...这是线程安全的，但仍然享有 reader-writer 同步的性能优势。

但一般的想法是，在处理线程安全的集合时，您总是有一系列的任务需要在更高的抽象层次上一起同步。通过公开 reader 和 writer 的同步方法，您有一个简单的通用机制来执行此操作。

综上所述，正如其他人所说，编写线程安全代码的最佳方式是完全避免并发访问。但是，如果您必须使可变对象成为线程安全的，那么调用者有责任确定必须作为单个同步操作执行的一系列任务。

当数组在Swift中有并发读取时，如何实现removeAtIndex的线程安全？

How to achieve thread safety for removeAtIndex when array has concurrent reads in Swift?

swift

multithreading

grand-central-dispatch

race-condition