Process Array in parallel using GCD

Question

I have a large array that I would like to process by handing slices of it to a few asynchronous tasks. As a proof of concept, I have the written the following code:

class TestParallelArrayProcessing {
    let array: [Int]
    var summary: [Int]

    init() {
        array = Array<Int>(count: 500000, repeatedValue: 0)
        for i in 0 ..< 500000 {
            array[i] = Int(arc4random_uniform(10))
        }
        summary = Array<Int>(count: 10, repeatedValue: 0)
    }

    func calcSummary() {
        let group = dispatch_group_create()
        let queue = dispatch_get_global_queue(QOS_CLASS_USER_INITIATED, 0)

        for i in 0 ..< 10 {
            dispatch_group_async(group, queue, {
                let base = i * 50000
                for x in base ..< base + 50000 {
                    self.summary[i] += self.array[x]
                }
            })
        }
        dispatch_group_notify(group, queue, {
            println(self.summary)
        })
    }
}

After init(), array will be initialized with random integers between 0 and 9.

The calcSummary function dispatches 10 tasks that take disjoint chunks of 50000 items from array and add them up, using their respective slot in summary as an accummulator.

This program crashes at the self.summary[i] += self.array[x] line. The error is:

 EXC_BAD_INSTRUCTION (code = EXC_I386_INVOP).

I can see, in the debugger, that it has managed to iterate a few times before crashing, and that the variables, at the time of the crash, have values within correct bounds.

I have read that EXC_I386_INVOP can happen when trying to access an object that has already been released. I wonder if this has anything to do with Swift making a copy of the array if it is modified, and, if so, how to avoid it.

Answer 1

This is a slightly different take on the approach in @Eduardo's answer, using the Array type's withUnsafeMutableBufferPointer<R>(body: (inout UnsafeMutableBufferPointer<T>) -> R) -> R method. That method's documentation states:

Call body(p), where p is a pointer to the Array's mutable contiguous storage. If no such storage exists, it is first created.

Often, the optimizer can eliminate bounds- and uniqueness-checks within an array algorithm, but when that fails, invoking the same algorithm on body's argument lets you trade safety for speed.

That second paragraph seems to be exactly what's happening here, so using this method might be more "idiomatic" in Swift, whatever that means:

func calcSummary() {
    let group = dispatch_group_create()
    let queue = dispatch_get_global_queue(QOS_CLASS_USER_INITIATED, 0)
    
    self.summary.withUnsafeMutableBufferPointer {
        summaryMem -> Void in
        for i in 0 ..< 10 {
            dispatch_group_async(group, queue, {
                let base = i * 50000
                for x in base ..< base + 50000 {
                    summaryMem[i] += self.array[x]
                }
            })
        }
    }

    dispatch_group_notify(group, queue, {
        println(self.summary)
    })
}

Answer 2

When you use the += operator, the LHS is an inout parameter -- I think you're getting race conditions when, as you mention in your update, Swift moves around the array for optimization. I was able to get it to work by summing the chunk in a local variable, then simply assigning to the right index in summary:

func calcSummary() {
    let group =  dispatch_group_create()
    let queue = dispatch_get_global_queue(QOS_CLASS_USER_INITIATED, 0)

    for i in 0 ..< 10 {
        dispatch_group_async(group, queue, {
            let base = i * 50000
            var sum = 0
            for x in base ..< base + 50000 {
                sum += self.array[x]
            }
            self.summary[i] = sum
        })
    }

    dispatch_group_notify(group, queue, {
        println(self.summary)
    })
}

Answer 3

You can also use concurrentPerform(iterations: Int, execute work: (Int) -> Swift.Void) (since Swift 3).

It has a much simpler syntax and will wait for all threads to finalise before returning.:

DispatchQueue.concurrentPerform(iterations: iterations) { i in
    performOperation(i)
}

Answer 4

I think Nate is right: there are race conditions with the summary variable. To fix it, I used summary's memory directly:

func calcSummary() {
    let group = dispatch_group_create()
    let queue = dispatch_get_global_queue(QOS_CLASS_USER_INITIATED, 0)

    let summaryMem = UnsafeMutableBufferPointer<Int>(start: &summary, count: 10)

    for i in 0 ..< 10 {
        dispatch_group_async(group, queue, {
           let base = i * 50000
           for x in base ..< base + 50000 {
              summaryMem[i] += self.array[x]
           }
        })
    }

    dispatch_group_notify(group, queue, {
        println(self.summary)
    })
}

This works (so far).

EDIT Mike S has a very good point, in his comment below. I have also found this blog post, which sheds some light on the problem.

Answer 5

Any solution that assigns the i'th element of the array concurrently risks race condition (Swift's array is not thread-safe). On the other hand, dispatching to the same queue (in this case main) before updating solves the problem but results in a slower performance overall. The only reason I see for taking either of these two approaches is if the array (summary) cannot wait for all concurrent operations to finish.

Otherwise, perform the concurrent operations on a local copy and assign it to summary upon completion. No race condition, no performance hit:

Swift 4

func calcSummary(of array: [Int]) -> [Int] {
    var summary = Array<Int>.init(repeating: 0, count: array.count)

    let iterations = 10 // number of parallel operations  

    DispatchQueue.concurrentPerform(iterations: iterations) { index in
        let start = index * array.count / iterations
        let end = (index + 1) * array.count / iterations

        for i in start..<end {
            // Do stuff to get the i'th element
            summary[i] = Int.random(in: 0..<array.count)
        }
    }

    return summary
}

I've answered a similar question here for simply initializing an array after computing on another array