Array procesowy równolegle za pomocą GCD

Question

Mam dużą tablicę, którą chciałbym przetworzyć, przekazując plasterki do kilku asynchronicznych zadań. Jako dowód koncepcji, mam napisane następujący kod:

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

Po init(), array zostanie zainicjowany z losowych liczb całkowitych od 0 do 9.

Funkcja calcSummary wywołuje 10 zadań, które mają rozłączne fragmenty z 50000 pozycji z array i dodaj je, używając odpowiedniego gniazda w summary jako pulsatora.

Ten program zawiesza się na linii self.summary[i] += self.array[x]. Błąd jest:

EXC_BAD_INSTRUCTION (code = EXC_I386_INVOP). 

widzę w debugger, że udało się iteracji kilka razy przed upaść, i że zmienne, w momencie katastrofy, mają wartości w prawidłowych granicach.

Przeczytałem, że EXC_I386_INVOP może się zdarzyć podczas próby uzyskania dostępu do obiektu, który został już zwolniony. Zastanawiam się, czy ma to coś wspólnego z tym, że Swift tworzy kopię tablicy, jeśli jest zmodyfikowana, a jeśli tak, to jak tego uniknąć.

Answer 1

To jest nieco inne podejście do podejścia w odpowiedzi @ Eduardo, przy użyciu metody Array typu withUnsafeMutableBufferPointer<R>(body: (inout UnsafeMutableBufferPointer<T>) -> R) -> R. That method's documentation states:

połączeń body(p), gdzie p jest wskaźnikiem do zmienny przyległej Przechowywanie Array „s. Jeśli takie miejsce nie istnieje, jest ono najpierw tworzone.

Często optymalizator może wyeliminować kontrole granic i unikalności w ramach algorytmu tablicowego, ale gdy to się nie powiedzie, wywołanie tego samego algorytmu na argumencie body pozwala obrócić bezpieczeństwo na szybkość.

To akapit drugi wydaje się być dokładnie to, co się tu dzieje, więc za pomocą tej metody może być bardziej „idiomatycznych” w Swift, czyli cokolwiek:

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

Kiedy używasz operatora +=, LHS jest parametrem inout - Myślę, że uzyskujesz warunki wyścigu, kiedy, jak wspominasz w swojej aktualizacji, Swift porusza się po tablicy w celu optymalizacji. Udało mi się zmusić go do pracy poprzez zsumowanie klocek w zmiennej lokalnej, a następnie po prostu przypisanie do prawej indeksu w 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) 
    }) 
} 

Answer 3

myślę Nate ma rację: istnieją warunki wyścigu ze zmiennej summary. Aby to naprawić, kiedyś pamięć summary jest bezpośrednio:

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

To działa (do tej pory).

EDIT Mike S ma bardzo dobry punkt, w swoim komentarzu poniżej. Znalazłem także this blog post, co rzuca nieco światła na problem.

Answer 4

Można również użyć concurrentPerform(iterations: Int, execute work: (Int) -> Swift.Void) (od Swift 3).

To ma znacznie prostszą składnię:

DispatchQueue.concurrentPerform(iterations: iterations) {i in 
     performOperation(i) 
} 

i będzie czekać na wszystkie wątki sfinalizować przed powrotem.