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!!
I am making a mistake / don’t understand why the code above fails. I don’t use playground (I’ll save you my rant on Xcode). I copied/pasted the code into a swift source files and use the swift package manager environment on a MacOS (all updated no worries on versioning).
Would you consider commenting running the code on MacOS, I believe it runs concurrently (truly), fails outright and I lack the insight why playground succeeds and package manager fails.
var shouldKeepRunning = true
let runLoop = RunLoop.current
let distantFuture = Date()
var isRunning = true
// Thread-unsafe array
// * * * Test1 fails on MacOs ie, use super simple Swift Package Manager
func Test1()
{
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)
iterations -= 1
guard iterations == 0 else { return }
// 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)
}
}
while shouldKeepRunning == true &&
runLoop.run(mode:RunLoopMode.defaultRunLoopMode, before: distantFuture) {}
—————————————
I did experiment – the following doesn't crash but exhibits no concurrency… why? Your insight?
var array = [Int]()
func kernel(_ index: Int) {
let last = array.last ?? 0
let cnt = last + 1
array.append(cnt)
print("index:\(index)")
}
func Test2() {
let queue = DispatchQueue(label: "my", qos: .userInitiated, attributes: .concurrent)
let start = Date().timeIntervalSince1970
var iterations = 100
queue.async {
DispatchQueue.concurrentPerform(iterations: iterations) { index in
kernel(index)
iterations -= 1
guard iterations <= 0 else { return }
let message = String(format: "Unsafe loop took %.3f seconds, count: %d.",
Date().timeIntervalSince1970 – start,
array.count)
print(message)
}
}
}
while shouldKeepRunning == true &&
runLoop.run(mode:RunLoopMode.defaultRunLoopMode, before: distantFuture) {}
I failed to clearly mention this, but the thread-unsafe test *should* intermittently crash, because it will try to read and write to the same memory access at some points. Do you get a “bad access” crash?
If you get lucky enough (or unlucky enough), the thread-unsafe test will suffer from corrupt data and will print the results.
The thread-safe test should always have correct results and never crash. Is this the case if you isolate just the thread-safe test?
Great post! Just wondering how it might be possible to use SynchronizedArray with multiple concurrent queues.
I have several singleton arrays and would like a concurrent queue for each. I can declare each array as SynchronizedArray to share the great helper functions, but then it uses the one concurrent queue. Any suggestions on how to use separate queues?
I’m glad you found it helpful 🙂
If I understand correctly, I did something similar when creating a wrapper for Core Location manager. See how I implemented multiple queues there: http://basememara.com/swifty-locations-observables/
Let me know if I misunderstood or have any questions from here.
Hi,
I have a question that’s not directly related to the threading stuff but with this SynchronizedArray class, how do you declare an array of arrays? For example I have a normal array declared like this – var outputArray: [[Double]] = []. How do I convert that to use the SynchronizedArray?
Hi Isuru, you can instantiate it like this to make `outputArray` thread-safe: `SynchronizedArray<[Double]>()`. However, keep in mind that your inner array will not be thread-safe, which may be ok if it is protected so it cannot be accessed from other threads. If you need to expose the inner array to other threads, then you may want to split it up into two different arrays.
Thanks for the reply.
Can I make the inner array, an array of Synchronized array too? I mean like this, `SynchronizedArray<SynchronizedArray>()`?
Hey please correct me if I’m wrong but I don’t your testing code is thread safe either:
`
concurrentPerform(iterations: iterations) { index in
iterations -= 1
}
iterations might end up with any value between 0 and 999. I understand that you take this into account but I don’t think your test for `iterations <= 0` is feasible because it should never be a negative.
Anyways, feedback on that would be awesome.
Hey Maxim, good catch! I added `DispatchQueue.global().sync` around the iteration counter and updated the post. Thanks!
Exactly aha I was looking for! You just fixed a race condition I was having for a long time sir! you just made my day 🙂
hi @BASEM EMARA. I’ve gotten the code of you. This is my project it’s crash. https://github.com/HieuLsw/Demo . Please help me the solution. Tks .
Hi Hieu, in `DataManager` you’re getting an index out of range fatal error. You remove 0…4 sub-range then try to remove it again right after. I’d group all removals together or put a guard to check if the index exists before removing.
Hello!
When I run the code under // Thread-safe array I get 1000 elements but they are like [1, 1, 1, 1, 2, 3, 3, 4, 4, 5, 5, 5, 6 … 257, 257, 257, 258]
But I expect to see [1, 2, 3, 4, 5 … 998, 999, 1000]. Can you explain is it right behavior or is there something wrong?
This is the expected behavior since `DispatchQueue.concurrentPerform` fires all the processes at the same time so it is not guaranteed which one will complete first. This is the nature of the test because executing serially would not test thread safety.
Playground only has one thread though.
for conforming to the Sequence Protocol:
extension SynchronizedArray: Sequence {
public typealias Iterator = IndexingIterator
public func makeIterator() -> Iterator {
var iterator: Iterator?
queue.sync {
iterator = self.array.makeIterator()
}
return iterator!
}
}
Thanks. This fixes some issues for me.
I needed a copy initializer and added
init() {
// Do nothing
}
// Copy an array
init(_ newArray: [Element]) {
array = newArray
}
I also added conformance to Sequence and InteratorProtocol with the following…
var curPos = 0 // Iterator variable
public func next() -> Element? {
if curPos Bool) -> SynchronizedArray {
var result = SynchronizedArray()
queue.sync { result = SynchronizedArray(self.array.sorted(by: areInIncreasingOrder)) }
return result
}
Very cool, thx! I think we can even take it one step further with Property Wrappers 😎
Hi I have a question about the concurrent sync read implementation in this post. Say if we have an array using you implementation. Also, there is a background thread that is crazily writing to this array and the main thread is crazily reading from the array. Will that possibly freeze the UI? Because `.barrier` is actually blocking any other operations to this array, including the synchronized read operations from the main thread.
If so, is there any better way to implement the data structure in this scenario?
You always want to send tasks asynchronously so the main thread won’t freeze. Those tasks will just get queued up until handled and fired later back later.
Basem, Great article and provided the basis of a solution that resolved many threading issues for me. I would add that I eventually had to change all “queue.async..” lines to “queue.sync..” to prevent some sequencing problems. When async was used for an action, there was no guarantee that its closure would be complete before the need of the next line needing the result, and when the following code depended on the action in the async closure, it led to some strange outcomes. Making every closure synchronized resolved all threading issues with arrays. I took another hit on speed (not significant for my app) but was necessary to ensure actions were undertaken in the desired sequence.
Hi Gordon, thanks for the feedback and sharing all your findings with us here!
It’s strange that the completion handler doesn’t fire at the right time for your scenario. It should execute the closure after it does its action on the array. I also noticed that I only put the completion handler on the remove methods, but would be handy on the append and insert functions. Would this help your scenario?
It would be great if you can share a test that reproduces this so we can see what’s happening if possible.
Hello I am deploying my first app to the app store and your synchronized array has been a life saver, however I have some questions, How is it able to access elements using array style notations “[]”, my second question is when you overloaded the operators why must it be inout?
remove part not working as expected, it does not remove concurrently even though queue is .barrier and concurrent.
Can you elaborate or have tests that can show this? That would be helpful.
Thank you very much for writing this article!!!
All descriptions and this working code were very helpful!
Thanks for the great article Basem Emara!
I had one doubt though. As the queue being used in the solution is a concurrent queue. Probably this is why you said
“We continue to use the sync method for reads, but all readers will run in parallel this time because of the concurrent queue attribute.”
But sync method blocks the thread on which it is fired. So until one read operation is complete, another read operation would not get a chance to run as essentially all read operations are fed into the same queue. If I am thinking right, all read operators would run serially?
Can you correct me if I am mis-understanding something here?