【问题标题】:Multithreaded Functional Programming in SwiftSwift 中的多线程函数式编程
【发布时间】:2015-12-09 01:55:32
【问题描述】:

我最近一直在 Swift 2.1 中操作字节数组,我经常发现自己在编写这样的代码:

// code to add functions to a [UInt8] object
extension CollectionType where Generator.Element == UInt8 {

    func xor(with byte: UInt8) -> [UInt8] {
        return map { $0 ^ byte }
    }
}

// example usage: [67, 108].xor(with: 0) == [67, 108]

有没有一种简单的方法可以并行化这个map 调用,以便多个线程可以同时对数组的非重叠区域进行操作?

我可以编写代码来手动将数组划分为子数组,并在不同线程中对每个子数组调用map。 但我想知道 Swift 中是否存在一些框架来自动进行除法,因为map 是一个函数调用,可以在线程安全的环境中工作而没有副作用。

澄清说明:

  1. 代码只需要在[UInt8] 对象上工作,不一定是每个CollectionType

【问题讨论】:

    标签: arrays multithreading swift


    【解决方案1】:

    并行执行计算循环的最简单方法是concurrentPerform(以前称为dispatch_apply;请参阅并发编程指南中的Performing Loop Iterations Concurrently)。但是,不,没有map 可以为您执行此操作。你必须自己做。

    例如,您可以编写一个扩展来执行并发任务:

    extension Array {
        public func concurrentMap<T>(_ transform: (Element) -> T) -> [T] {
            var results = [Int: T](minimumCapacity: count)
    
            let lock = NSLock()
    
            DispatchQueue.concurrentPerform(iterations: count) { index in
                let result = transform(self[index])
                lock.synchronized {
                    results[index] = result
                }
            }
    
            return (0 ..< results.count).compactMap { results[$0] }
        }
    }
    

    在哪里

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

    你可以使用任何你想要的同步机制(锁、串行队列、读写器),但想法是同时执行transform,然后同步集合的更新。

    注意:

    • 这将阻塞您从中调用它的线程(就像非并发 map 一样),因此请确保将其分派到后台队列。

    • 需要确保每个线程上有足够的工作来证明管理所有这些线程的固有开销是合理的。 (例如,每个循环一个简单的异或调用是不够的,你会发现它实际上比非并发版本慢。)在这些情况下,确保你大步前进(参见Improving Loop Code,它平衡了每个并发的工作量堵塞)。例如,与其对一个极其简单的操作进行 5000 次迭代,不如在每个循环中进行 500 次操作的 10 次迭代。您可能需要尝试合适的跨步值。


    虽然我怀疑您不需要这个讨论,但对于不熟悉concurrentPerform(以前称为dispatch_apply)的读者,我将在下面说明它的用法。有关该主题的更完整讨论,请参阅上面的链接。

    例如,让我们考虑一些比简单的xor 复杂得多的东西(因为这样简单的东西,开销超过了所获得的任何性能),例如一个简单的斐波那契实现:

    func fibonacci(_ n: Int) -> Int {
        if n == 0 || n == 1 {
            return n
        }
        return fibonacci(n - 1) + fibonacci(n - 2)
    }
    

    如果您想要计算 arrayInt 值,而不是:

    let results = array.map { fibonacci($0) }
    

    你可以:

    var results = [Int](count: array.count, repeatedValue: 0)
    DispatchQueue.concurrentPerform(iterations: array.count) { index in
        let result = self.fibonacci(array[index])
        synchronize.update { results[index] = result }      // use whatever synchronization mechanism you want
    }
    

    或者,如果你想要一个功能再现,你可以使用我上面定义的extension

    let results = array.concurrentMap { fibonacci($0) }
    

    对于 Swift 2 版本,请参阅 previous revision of this answer

    【讨论】:

    • 就像我在现已删除的答案中所说的那样,我一直在做一些基准测试。如果将数组拆分为 x = NSProcessInfo().activeProcessorCount * 4 的 x 片,速度会稍快一些。每个切片都有自己的dispatch_apply
    • 当我将它插入my test framework 时,这个实现会捕获解包 nil 可选,我认为即使修复了它也会非常低效,因为它会为每个元素排队一个异步操作。跨度>
    • 是的,如果您的计算对于concurrentPerform 的每次迭代都不够重要,您想大步前进。请参阅stackoverflow.com/a/39949292/1271826 以获取跨步示例。
    【解决方案2】:

    与我见过的所有其他实施相比,我的实施似乎是正确的并且表现良好。测试和基准是here

    extension RandomAccessCollection {
        /// Returns `self.map(transform)`, computed in parallel.
        ///
        /// - Requires: `transform` is safe to call from multiple threads.
        func concurrentMap<B>(_ transform: (Element) -> B) -> [B] {
            let batchSize = 4096 // Tune this
            let n = self.count
            let batchCount = (n + batchSize - 1) / batchSize
            if batchCount < 2 { return self.map(transform) }
    
            return Array(unsafeUninitializedCapacity: n) {
                uninitializedMemory, resultCount in
                resultCount = n
                let baseAddress = uninitializedMemory.baseAddress!
    
                DispatchQueue.concurrentPerform(iterations: batchCount) { b in
                    let startOffset = b * n / batchCount
                    let endOffset = (b + 1) * n / batchCount
                    var sourceIndex = index(self.startIndex, offsetBy: startOffset)
                    for p in baseAddress+startOffset..<baseAddress+endOffset {
                        p.initialize(to: transform(self[sourceIndex]))
                        formIndex(after: &sourceIndex)
                    }
                }
            }
        }
    }
    

    希望对你有帮助,

    -戴夫

    【讨论】:

    • 你链接到的要点看起来很有趣,谢谢!
    【解决方案3】:

    您可以使用parMap(),它是平行地图。您可以使用活动监视器来检查它是否是平行地图。

    func map<T: Collection, U>( _ transform: (T.Iterator.Element) -> U, _ xs: T) -> [U] {
        return xs.reduce([U](), {$0 + [transform($1)]})
    }
    
    public func parMap<T,U>(_ transform: @escaping (T)->U, _ xs: [T]) -> [U] {
        let len     = xs.count
        var results = [U?](repeating: nil, count: len)
        let process = { (i: Int) -> Void in results[i] = transform(xs[i]) }
        DispatchQueue.concurrentPerform(iterations: len, execute: process)
        return map({$0!}, results)
    }
    
    func test() {
        parMap({_ in Array(1...10000000).reduce(0,+)}, Array(1...10))
    }
    

    【讨论】:

    • 这个结果有竞争条件
    猜你喜欢
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 2010-09-06
    • 2018-09-20
    • 2012-05-10
    • 2021-08-06
    • 1970-01-01
    • 1970-01-01
    相关资源
    最近更新 更多