Swift 2: struct thread-safety

Question

In my swift practice, I wrote simple struct named OrderedSet.

I tried OrderedSet to be a thread-safe with GCD serial queue.

But it’s not working. The test result is unstable. I expected something like:

20:[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19]

but received something like like

2:[3, 19]

here is playground code:

import Foundation
import XCPlayground

struct OrderedSet<T: Equatable> {
    mutating func append(e: T) {
        dispatch_sync(q) {
            if !self.__elements.contains(e) {
                self.__elements.append(e)
            }
        }
    }
    var elements: [T] {
        var elements: [T] = []
        dispatch_sync(q) {
            elements = self.__elements
        }
        return elements
    }
    var count: Int {
        var ret = 0
        dispatch_sync(q) {
            ret = self.__elements.count
        }
        return ret
    }
    private var __elements: [T] = []
    private let q = dispatch_queue_create("OrderedSet.private.serial.queue", DISPATCH_QUEUE_SERIAL)
}
extension OrderedSet: CustomStringConvertible {
    var description: String {
        var text = ""
        dispatch_sync(q) {
            text = "\(self.__elements.count):\(self.__elements)"
        }
        return text
    }
}

// Test code
let globalQueue = dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0)
let group = dispatch_group_create()

var testSet = OrderedSet<Int>()
for i in 0..<20 {
    dispatch_group_async(group, globalQueue) {
        testSet.append(i)
    }
}
dispatch_group_notify(group, globalQueue) {
    print("\(testSet)") // unstable result
}

XCPSetExecutionShouldContinueIndefinitely()

I’ve checked below:

It’s OK if defined OrderdSet as a class (not struct).

It’s OK if using semaphore instead of using serial queue.

I would like to know the reason why the pair of struct and serial queue is unstable.

---- updated

I got the expected result with these.

1. class instead of struct

   import Foundation
   import XCPlayground
   
   class OrderedSet<T: Equatable> {
       func append(e: T) {
           dispatch_sync(q) {
               if !self.__elements.contains(e) {
                   self.__elements.append(e)
               }
           }
       }
       var elements: [T] {
           var elements: [T] = []
           dispatch_sync(q) {
               elements = self.__elements
           }
           return elements
       }
       var count: Int {
           var ret = 0
           dispatch_sync(q) {
               ret = self.__elements.count
           }
           return ret
       }
       private var __elements: [T] = []
       private let q = dispatch_queue_create("OrderedSet.private.serial.queue", DISPATCH_QUEUE_SERIAL)
   }
   extension OrderedSet: CustomStringConvertible {
       var description: String {
           var text = ""
           dispatch_sync(q) {
               text = "\(self.__elements.count):\(self.__elements)"
           }
           return text
       }
   }
   
   // Test code
   let globalQueue = dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0)
   let group = dispatch_group_create()
   
   var testSet = OrderedSet<Int>()
   for i in 0..<20 {
       dispatch_group_async(group, globalQueue) {
           testSet.append(i)
       }
   }
   dispatch_group_notify(group, globalQueue) {
       print("\(testSet)") // It's OK
   }
   
   XCPSetExecutionShouldContinueIndefinitely()
   

2. semaphore instead of serial queue

   import Foundation
   import XCPlayground
   
   struct OrderedSet<T: Equatable> {
       mutating func append(e: T) {
           dispatch_semaphore_wait(s, DISPATCH_TIME_FOREVER)
           if !self.__elements.contains(e) {
               self.__elements.append(e)
           }
           dispatch_semaphore_signal(s)
       }
       var elements: [T] {
           var elements: [T] = []
           dispatch_semaphore_wait(s, DISPATCH_TIME_FOREVER)
           elements = self.__elements
           dispatch_semaphore_signal(s)
           return elements
       }
       var count: Int {
           var ret = 0
           dispatch_semaphore_wait(s, DISPATCH_TIME_FOREVER)
           ret = self.__elements.count
           dispatch_semaphore_signal(s)
           return ret
       }
       private var __elements: [T] = []
       private let s = dispatch_semaphore_create(1)
   }
   extension OrderedSet: CustomStringConvertible {
       var description: String {
           var text = ""
           dispatch_semaphore_wait(s, DISPATCH_TIME_FOREVER)
           text = "\(self.__elements.count):\(self.__elements)"
           dispatch_semaphore_signal(s)
           return text
       }
   }
   
   // Test code
   let globalQueue = dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0)
   let group = dispatch_group_create()
   
   var testSet = OrderedSet<Int>()
   for i in 0..<20 {
       dispatch_group_async(group, globalQueue) {
           testSet.append(i)
       }
   }
   dispatch_group_notify(group, globalQueue) {
       print("\(testSet)") // It's OK
   }
   
   XCPSetExecutionShouldContinueIndefinitely()
   

3. serial queue with OrderdSet itself.

   import Foundation
   import XCPlayground
   
   struct OrderedSet<T: Equatable> {
       mutating func append(e: T) {
           if !self.__elements.contains(e) {
               self.__elements.append(e)
           }
       }
       var elements: [T] {
           return self.__elements
       }
       var count: Int {
           return self.__elements.count
       }
       private var __elements: [T] = []
   }
   extension OrderedSet: CustomStringConvertible {
       var description: String {
           return "\(self.__elements.count):\(self.__elements)"
       }
   }
   
   // Test code
   let globalQueue = dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0)
   let serialQueue = dispatch_queue_create("serial", DISPATCH_QUEUE_SERIAL)
   
   let group = dispatch_group_create()
   
   var testSet = OrderedSet<Int>()
   for i in 0..<20 {
       dispatch_group_async(group, globalQueue) {
           dispatch_sync(serialQueue) {
               testSet.append(i)
           }
       }
   }
   dispatch_group_notify(group, serialQueue) {
       print("\(testSet)") // It's OK
   }
   
   XCPSetExecutionShouldContinueIndefinitely()
   

Answer 1

This code will capture current value of testSet:

dispatch_group_async(group, globalQueue) {
    testSet.append(i) // `testSet` inside the closure will be a copy of the `testSet` variable outside 
}

After the execution of the closure, the value of the inside testSet will be copied to the outside testSet variable.

Imagine a concurrent world:

• 10 closures are running simultaneously, capturing the initial value of the outside testSet, which is "0:[]".

• Once finished, 10 copy of testSets inside closures try to copy back to the only outside testSet. However, there is only one winner, say, current value of the outside testSet is "1:[3]".

• Yet another round start, capturing current value of the outside testSet which is "1:[3]", appending i, and copying back, yielding the weird result, say, "2:[3, 19]"

In your updated case 1, changing OrderedSet to class, things are pretty straight forward, testSet is captured by reference, and all the threads are sharing the same object.

In your updated case 3, by using serial queue, I guess every appending and copying back operation is serial, so you yield a perfect ordered set.

Case 2 is more complicated. Actually I haven't figure out what's going on under the hood. And I think it's more about a implementation detail of the swift compiler and may change over different swift versions. It seems like semaphore is a reference type, thus all the copy of the 'testSet's are sharing the same semaphore. I guess complier decide to do some optimization in this case and make all the copy of the testSets' __element point to the same array. So the result contains all the 0..<20 elements but the order is unpredictable.