Swift 2: struct thread-safety

Question

W mojej szybkiej praktyce napisałem prostą strukturę o nazwie OrderedSet.

Próbowałem, aby OrderedSet był wątkowo bezpieczny z kolejką szeregową GCD.

Ale to nie działa. Wynik testu jest niestabilny. Spodziewałem się czegoś takiego:

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

ale otrzymał coś takiego jak

2:[3, 19] 

tutaj jest kod zabaw:

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

Sprawdziłem poniżej:

to OK, jeśli zdefiniowano OrderdSet jako klasa (nie struct).

Jest OK, jeśli używasz semafora zamiast używania kolejki szeregowej.

Chciałbym poznać powód, dla którego para kolejki struct i serial jest niestabilna.

---- aktualizowane

mam oczekiwany rezultat z nich.

1. klasy zamiast struktury

   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. semafora zamiast kolejki seryjny

   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. seryjny kolejki z samą OrderdSet.

   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

Kod ten będzie przechwytywać wartość prądu z testSet:

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

Po wykonaniu zamknięcia, wartość wewnętrzna testSet zostaną skopiowane do zewnętrznego testSet zmiennej.

Wyobraźmy sobie świat współbieżne:

• 10 zamknięcia są uruchomione jednocześnie, przechwytywanie wartość początkową zewnętrznej testSet, który jest "0: []".

• Po zakończeniu 10 kopii zamkniętych wewnątrz zamknięć testSet próbuje skopiować z powrotem na jedyną zewnętrzną stronę: testSet. Jednak jest tylko jeden zwycięzca, powiedzmy, obecna wartość zewnętrznego testSet to "1: [3]".

• Jeszcze inny okrągły początek, uchwycenie bieżącą wartość poza testSet która jest "1: [3]", dodanie i i kopiowania z powrotem, w wyniku czego otrzymuje się dziwne wyniku, na przykład, "2: [3, 19]"

W zaktualizowanym przypadku 1, zmieniając OrderedSet do klasy, rzeczy są dość proste, testSet zostaje schwytany przez odniesienie, a wszystkie wątki dzielą ten sam obiekt.

W zaktualizowanym przypadku 3, używając kolejki szeregowej, domyślam się, że każda operacja dodawania i kopiowania jest seryjna, więc otrzymujesz idealny zestaw uporządkowany.

Przypadek 2 jest bardziej skomplikowany. Właściwie to nie wiem, co się dzieje pod maską. I myślę, że chodzi bardziej o szczegóły implementacji szybkiego kompilatora i może się zmieniać w różnych szybkich wersjach. Wygląda na to, że semafor jest typem referencyjnym, a więc cała kopia "zestawu testowego dzieli ten sam semafor. Przypuszczam, że kompilator zdecyduje się przeprowadzić optymalizację w tym przypadku i uczynić całą kopię punktu testSet s ' tą samą tablicą. Więc wynik zawiera wszystkie 0 .. < 20 elementów, ale kolejność jest nieprzewidywalna.

Answer 2

Myślę, że to, co się dzieje, gdy dispatch_sync jest używane wewnątrz struktury, jest to, że self jest niejawnie przechwytywane przez zamknięcie jako parametr inout.

Oznacza to, że zmodyfikowano kopię, która następnie zastępuje zewnętrzną self po powrocie. Tak więc istnieje wiele równoległych kopii zmutowanych samemu sobie, a następnie zbijanie oryginału.

W przypadku semaforów nie ma zamknięcia, więc nie ma przechwytywania, więc samo jest self to jaźń. Mutowanie dzieje się na oryginalnym zewnętrznym ja, a semafory zapewniają, że każdy robi to w uporządkowanej linii.

Zdarzyło mi się uruchomić to samo, gdy używa się opakowania zamykającego wokół pthread mutexes dla modułów pobierających i ustawiających wewnątrz struktury. Mimo że parametr zamknięcia nie jest unikany, struktura (np. self) nadal wydaje się być traktowana jako inout, więc zdarzają się nieporządne rzeczy.