Thread-safety is a tricky territory, especially in languages like Swift where there is no built-in concurrency support. Instead, we have to go through Grand Central Dispatch. And where we get concurrency is where we can find thread-safety, so let’s repurpose queues for achieving thread-safety.

UPDATE: Created a more generic version here.

Serial Queue Method

By leveraging serial queues, we can enforce mutual exclusion on a resource. With a serial queue, only one process can run at a time, so if many processes are stuffed in a queue to modify the array, the serial queue will only let one process execute at a time; the array is safe from concurrent processes by design.

let queue = DispatchQueue(label: "MyArrayQueue") queue.async() { // Manipulate the array here } queue.sync() { // Read array here }

Dispatch queues are serial by default. We use this queue’s async method to write to the array and not worry about the result; we asynchronously set it and forget it. When reading from the array, we can use the sync method and get the results instantly.

We can still do better though. The reads are not optimized because multiple read requests have to wait for each other in a queue. However, reads should be able to happen concurrently, as long as there isn’t a write happening at the same time.

Concurrent Queue Method

This technique is more elegant and uses a shared exclusion lock on the array. We will still use Grand Central Dispatch, but this time with a concurrent queue instead of a serial one. That might work for concurrent reads, but we must disallow all concurrency when writing. This can be achieved with the barrier flag for the dispatch queue:

let queue = DispatchQueue(label: "MyArrayQueue", attributes: .concurrent) queue.async(flags: .barrier) { // Mutate array here } queue.sync() { // Read array here }

Notice the async method has the barrier flag set for writes. This means no other blocks may be scheduled from the queue while the async/barrier process runs. We continue to use the sync method for reads, but all readers will run in parallel this time because of the concurrent queue attribute.

Readers will still be blocked when a barrier process is running though. Even if there are several reader blocks already running in parallel, the barrier process will wait for all readers to finish before beginning the write. Once the barrier process is complete, then the readers behind it can run in parallel again. Sweet! 🙂

Where’s the Proof?

Concurrency is a difficult thing to test since it is non-deterministic, so how do we test if this even works?

Here’s a little script that shows thread-safety issues when modifying arrays concurrently. What we can do is modify the array in flight and create a race condition:

var array = [Int]() DispatchQueue.concurrentPerform(iterations: 1000) { index in let last = array.last ?? 0 array.append(last + 1) }

This is launching 1000 parallel processes to modify the same array. It is taking the last element, incrementing it by one, and appending to the array.

The race condition occurs during the append statement. Swift uses a copy-to-write technique for value types as soon as it starts mutating it. During the copy-to-write process, the array could have been changed by another parallel task in that microsecond. So work would be done on an older snapshot of the array which results in data loss.

If we were to run and print several results of this, it would look random like this:

Unsafe loop count: 989. Unsafe loop count: 992. Unsafe loop count: 986. Unsafe loop count: 998.

This is where the problem is: it should always end up with a 1000 elements!

Where’s the Goods?

Let’s encapsulate everything we learned into a nifty, thread-safe array called `SynchronizedArray`. That way, we can use it in our code where we need a shared array accessible by multiple threads. Here it goes:

/// A thread-safe array. public class SynchronizedArray<Element> { fileprivate let queue = DispatchQueue(label: "io.zamzam.ZamzamKit.SynchronizedArray", attributes: .concurrent) fileprivate var array = [Element]() } // MARK: - Properties public extension SynchronizedArray { /// The first element of the collection. var first: Element? { var result: Element? queue.sync { result = self.array.first } return result } /// The last element of the collection. var last: Element? { var result: Element? queue.sync { result = self.array.last } return result } /// The number of elements in the array. var count: Int { var result = 0 queue.sync { result = self.array.count } return result } /// A Boolean value indicating whether the collection is empty. var isEmpty: Bool { var result = false queue.sync { result = self.array.isEmpty } return result } /// A textual representation of the array and its elements. var description: String { var result = "" queue.sync { result = self.array.description } return result } } // MARK: - Immutable public extension SynchronizedArray { /// Returns the first element of the sequence that satisfies the given predicate or nil if no such element is found. /// /// - Parameter predicate: A closure that takes an element of the sequence as its argument and returns a Boolean value indicating whether the element is a match. /// - Returns: The first match or nil if there was no match. func first(where predicate: (Element) -> Bool) -> Element? { var result: Element? queue.sync { result = self.array.first(where: predicate) } return result } /// Returns an array containing, in order, the elements of the sequence that satisfy the given predicate. /// /// - Parameter isIncluded: A closure that takes an element of the sequence as its argument and returns a Boolean value indicating whether the element should be included in the returned array. /// - Returns: An array of the elements that includeElement allowed. func filter(_ isIncluded: (Element) -> Bool) -> [Element] { var result = [Element]() queue.sync { result = self.array.filter(isIncluded) } return result } /// Returns the first index in which an element of the collection satisfies the given predicate. /// /// - Parameter predicate: A closure that takes an element as its argument and returns a Boolean value that indicates whether the passed element represents a match. /// - Returns: The index of the first element for which predicate returns true. If no elements in the collection satisfy the given predicate, returns nil. func index(where predicate: (Element) -> Bool) -> Int? { var result: Int? queue.sync { result = self.array.index(where: predicate) } return result } /// Returns the elements of the collection, sorted using the given predicate as the comparison between elements. /// /// - Parameter areInIncreasingOrder: A predicate that returns true if its first argument should be ordered before its second argument; otherwise, false. /// - Returns: A sorted array of the collection’s elements. func sorted(by areInIncreasingOrder: (Element, Element) -> Bool) -> [Element] { var result = [Element]() queue.sync { result = self.array.sorted(by: areInIncreasingOrder) } return result } /// Returns an array containing the non-nil results of calling the given transformation with each element of this sequence. /// /// - Parameter transform: A closure that accepts an element of this sequence as its argument and returns an optional value. /// - Returns: An array of the non-nil results of calling transform with each element of the sequence. func flatMap<ElementOfResult>(_ transform: (Element) -> ElementOfResult?) -> [ElementOfResult] { var result = [ElementOfResult]() queue.sync { result = self.array.flatMap(transform) } return result } /// Calls the given closure on each element in the sequence in the same order as a for-in loop. /// /// - Parameter body: A closure that takes an element of the sequence as a parameter. func forEach(_ body: (Element) -> Void) { queue.sync { self.array.forEach(body) } } /// Returns a Boolean value indicating whether the sequence contains an element that satisfies the given predicate. /// /// - Parameter predicate: A closure that takes an element of the sequence as its argument and returns a Boolean value that indicates whether the passed element represents a match. /// - Returns: true if the sequence contains an element that satisfies predicate; otherwise, false. func contains(where predicate: (Element) -> Bool) -> Bool { var result = false queue.sync { result = self.array.contains(where: predicate) } return result } } // MARK: - Mutable public extension SynchronizedArray { /// Adds a new element at the end of the array. /// /// - Parameter element: The element to append to the array. func append( _ element: Element) { queue.async(flags: .barrier) { self.array.append(element) } } /// Adds a new element at the end of the array. /// /// - Parameter element: The element to append to the array. func append( _ elements: [Element]) { queue.async(flags: .barrier) { self.array += elements } } /// Inserts a new element at the specified position. /// /// - Parameters: /// - element: The new element to insert into the array. /// - index: The position at which to insert the new element. func insert( _ element: Element, at index: Int) { queue.async(flags: .barrier) { self.array.insert(element, at: index) } } /// Removes and returns the element at the specified position. /// /// - Parameters: /// - index: The position of the element to remove. /// - completion: The handler with the removed element. func remove(at index: Int, completion: ((Element) -> Void)? = nil) { queue.async(flags: .barrier) { let element = self.array.remove(at: index) DispatchQueue.main.async { completion?(element) } } } /// Removes and returns the element at the specified position. /// /// - Parameters: /// - predicate: A closure that takes an element of the sequence as its argument and returns a Boolean value indicating whether the element is a match. /// - completion: The handler with the removed element. func remove(where predicate: @escaping (Element) -> Bool, completion: ((Element) -> Void)? = nil) { queue.async(flags: .barrier) { guard let index = self.array.index(where: predicate) else { return } let element = self.array.remove(at: index) DispatchQueue.main.async { completion?(element) } } } /// Removes all elements from the array. /// /// - Parameter completion: The handler with the removed elements. func removeAll(completion: (([Element]) -> Void)? = nil) { queue.async(flags: .barrier) { let elements = self.array self.array.removeAll() DispatchQueue.main.async { completion?(elements) } } } } public extension SynchronizedArray { /// Accesses the element at the specified position if it exists. /// /// - Parameter index: The position of the element to access. /// - Returns: optional element if it exists. subscript(index: Int) -> Element? { get { var result: Element? queue.sync { guard self.array.startIndex..<self.array.endIndex ~= index else { return } result = self.array[index] } return result } set { guard let newValue = newValue else { return } queue.async(flags: .barrier) { self.array[index] = newValue } } } } // MARK: - Equatable public extension SynchronizedArray where Element: Equatable { /// Returns a Boolean value indicating whether the sequence contains the given element. /// /// - Parameter element: The element to find in the sequence. /// - Returns: true if the element was found in the sequence; otherwise, false. func contains(_ element: Element) -> Bool { var result = false queue.sync { result = self.array.contains(element) } return result } } // MARK: - Infix operators public extension SynchronizedArray { static func +=(left: inout SynchronizedArray, right: Element) { left.append(right) } static func +=(left: inout SynchronizedArray, right: [Element]) { left.append(right) } }

I’ve declared `SynchronizedArray` to mimic a regular array. In it contains a private queue and array. Several of the array’s properties and methods have been exposed. Also notice the queue is declared as concurrent.

The exposed array calls that mutate it have been wrapped in the queue’s async method with the barrier flag, and the reads are wrapped in the queue’s sync method. This allows concurrent reads to occur, but writes to block all requests until complete.

Note: we can probably create `SynchronizedDictionary` the same way too. I also wonder if wrapping lower level API’s would work, such as making `SynchronizedArray` conform to a Sequence or Collection protocol.

The tests can now be rewritten to show the comparison between an unsafe and safe array:

import Foundation import PlaygroundSupport // Thread-unsafe array do { var array = [Int]() var iterations = 1000 let start = Date().timeIntervalSince1970 DispatchQueue.concurrentPerform(iterations: iterations) { index in let last = array.last ?? 0 array.append(last + 1) DispatchQueue.global().sync { iterations -= 1 // Final loop guard iterations <= 0 else { return } let message = String(format: "Unsafe loop took %.3f seconds, count: %d.", Date().timeIntervalSince1970 - start, array.count) print(message) } } } // Thread-safe array do { var array = SynchronizedArray<Int>() var iterations = 1000 let start = Date().timeIntervalSince1970 DispatchQueue.concurrentPerform(iterations: iterations) { index in let last = array.last ?? 0 array.append(last + 1) DispatchQueue.global().sync { iterations -= 1 // Final loop guard iterations <= 0 else { return } let message = String(format: "Safe loop took %.3f seconds, count: %d.", Date().timeIntervalSince1970 - start, array.count) print(message) } } } PlaygroundPage.current.needsIndefiniteExecution = true

This should print something like this:

Unsafe loop took 1.031 seconds, count: 989. Safe loop took 1.363 seconds, count: 1000.

It is unfortunately 30% slower and incurs more memory due to the GCD overhead, but the tradeoff is that it’s accurate 😉

The full Playground gist is available for you to try.

Happy Coding!!