Swift 3 并行 for/map 循环

Question

然而，关于这个的话题有很多Using Grand Central Dispatch in Swift to parallelize and speed up “for" loops?使用 Swift <3.0 代码，我无法在 3 中获得等效代码(参见代码)。 Process Array in parallel using GCD使用指针并且它变得有点丑陋所以我要在这里断言我正在寻找nice Swift 3方法来做到这一点(当然尽可能高效)。我还听说组很慢(？)也许有人可以证实这一点。我也无法让小组工作。

这是我实现的跨步并行映射函数(在 Array 的扩展中)。它希望在全局队列上执行，以免阻塞 UI。可能是并发位不需要在范围内，只需要余数循环。

extension Array {

    func parallelMap<R>(striding n: Int, f: @escaping (Element) -> R, completion: @escaping ([R]) -> ()) {
        let N = self.count
        var res = [R?](repeating: nil, count: self.count)
        DispatchQueue.global(qos: .userInitiated).async {
            DispatchQueue.concurrentPerform(iterations: N/n) {k in
                for i in (k * n)..<((k + 1) * n) {
                    res[i] = f(self[i]) //Error 1 here
                }
            }
            //remainder loop
            for i in (N - (N % n))..<N {
                res[i] = f(self[i])
            }
            DispatchQueue.main.sync { //But it will pause on this line.
                print("\nPlease work!\n") //This doesn't execute either.
                completion(unwrap(res)!) //Error 2 here, or rather "not here"
            }
        }
    }

}   

public func unwrap<T>(_ arr: Array<T?>) -> Array<T>? {
    if (arr.contains{$0 == nil}) {
        return nil
    } else {
        return arr.map{(x: T?) -> T in x!}
    }
}

错误 1:我们的老 friend EXC_BAD_ACCESS 在内部数组赋值行上有大约一半的测试次数。我猜这表明存在同时访问问题。

错误 2:completion 永远不会执行!

错误 3:错误永远存在，我相信一旦上述问题得到解决，这种情况就会发生。

最后:最快并行的代码(确保它尽可能并行，我不喜欢代码中的“并发”) map/for 函数可能。这是修复我的代码的补充。

Answer 1

Martin 原来的做法仍然是正确的做法。将您的方法与他的方法合并并转换为 Swift 3 非常简单(尽管我去掉了您的可选值，只是手动处理了内存)。

extension Array {
    func parallelMap<R>(striding n: Int, f: @escaping (Element) -> R, completion: @escaping ([R]) -> ()) {
        let N = self.count

        let res = UnsafeMutablePointer<R>.allocate(capacity: N)

        DispatchQueue.concurrentPerform(iterations: N/n) { k in
            for i in (k * n)..<((k + 1) * n) {
                res[i] = f(self[i])
            }
        }

        for i in (N - (N % n))..<N {
            res[i] = f(self[i])
        }

        let finalResult = Array<R>(UnsafeBufferPointer(start: res, count: N))
        res.deallocate(capacity: N)

        DispatchQueue.main.async {
            completion(finalResult)
        }
    }
}

Martin 的版本避免了额外的复制，因为他有一个“零”值来初始化数组。如果你知道你的类型有一个微不足道的 init()，你可以避免额外的副本:

protocol TriviallyInitializable {
    init()
}

extension Array {
    func parallelMap<R>(striding n: Int, f: @escaping (Element) -> R, completion: @escaping ([R]) -> ()) where R: TriviallyInitializable {
        let N = self.count

        var finalResult = Array<R>(repeating: R(), count: N)

        finalResult.withUnsafeMutableBufferPointer { res in
            DispatchQueue.concurrentPerform(iterations: N/n) { k in
                for i in (k * n)..<((k + 1) * n) {
                    res[i] = f(self[i])
                }
            }
        }

        for i in (N - (N % n))..<N {
            finalResult[i] = f(self[i])
        }

        DispatchQueue.main.async {
            completion(finalResult)
        }
    }
}