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)
 })
}

• Eduardo: Thanks. I am wondering if this really works, or if it is by chance because there are only ten assignments (my original code is able to iterate a few times before crashing). I will try with a larger summary array and see what happens.
• Eduardo: Interesting ... based on your code, I have increased the tasks to 100 and it still works. My problem is that the code I posted is a simplification of what I need, and, in real life, I really need to accumulate on summary. I will try to solve it writing to memory directly.
• Rob: This strikes me that this dramatically lowers the chance of the problem, but doesn't eliminate it. The problematic update is still there, but just done 10 times instead of 500,000 times. I'd be inclined to marry this approach (reducing updates of array) with some other synchronization technique (e.g. reader-writer pattern).
• Eduardo: Thank, Nate. And thank you for your blog; I am a fan :-)
• Rob: @MikeS Agreed, though I generally use the reader-writer pattern, i.e. create concurrent queue, and do reads with dispatch_sync and do writes with dispatch_barrier_async. It achieves the same thing as the serial queue you suggest, though it allows multithreaded reads, and only enforces serial behavior on writes. In this case, I don't think it's a difference that matters (because of the simplistic algorithm, we aren't reading from summary), but in more practical scenarios, the reader-writer pattern becomes valuable.

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.

• Rob: BTW, have you benchmarked this vs Array<T> approach? The reason I ask is that I noticed doing extensive manipulation of pixels in a 30mb pixel buffer was 100x faster when using the memory pointer (like here) rather than using the Array<T> technique. I'm just wondering if you've seen similar issues...
• Eduardo: @Rob: I have not benchmarked it yet, but it runs pretty fast. Not sure if much faster than Array with full optimization and no bound-checks.
• Mike S: I don't think there's any guarantee that Array's memory is contiguous, so you could easily write to an invalid memory location using this method. It should work if summary is a ContiguousArray though.
• Eduardo: @MikeS: good catch; I have updated the answer, and linked to an interesting blog post that explains the topic.

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