Golang concurrency: jak dołączyć do tego samego plasterka z różnych goroutines

Question

Mam równoczesne goroutines, które chcą dołączyć (wskaźnik do a) struct do tego samego plasterka. Jak piszesz to w Go, aby zapewnić bezpieczeństwo współbieżności?

To byłaby moja współbieżności-niebezpieczny kod, wykorzystując grupę oczekiwania:

var wg sync.WaitGroup 
MySlice = make([]*MyStruct) 
for _, param := range params { 
    wg.Add(1) 
    go func(param string) { 
     defer wg.Done() 
     OneOfMyStructs := getMyStruct(param) 
     MySlice = append(MySlice, &OneOfMyStructs) 
    }(param) 
} 
wg.Wait() 

Chyba trzeba by użyć przejść kanałów współbieżności-bezpieczeństwo. Czy ktokolwiek może przyczynić się do przykładu?

Answer 1

Kanał jest najlepszym sposobem rozwiązania tego problemu. Oto przykład, który można uruchomić na go playground.

package main 

import "fmt" 
import "sync" 
import "runtime" 

type T int 

func main() { 
    var slice []T 
    var wg sync.WaitGroup 

    queue := make(chan T, 1) 

    // Create our data and send it into the queue. 
    wg.Add(100) 
    for i := 0; i < 100; i++ { 
     go func(i int) { 
      defer wg.Done() 

      // Do stuff. 
      runtime.Gosched() 

      queue <- T(i) 
     }(i) 
    } 

    // Poll the queue for data and append it to the slice. 
    // Since this happens synchronously and in the same 
    // goroutine/thread, this can be considered safe. 
    go func() { 
     defer wg.Done() 
     for t := range queue { 
      slice = append(slice, t) 
     } 
    }() 

    // Wait for everything to finish. 
    wg.Wait() 

    fmt.Println(slice) 
} 

Uwaga: The runtime.Gosched() połączenie jest tam, ponieważ te goroutines nie dają do harmonogramu. Co spowodowałoby zakleszczenie, gdybyśmy nie zrobili czegoś, aby uruchomić program planujący. Inną opcją mogłoby być wykonanie pewnych operacji wejścia/wyjścia (np .: print to stdout). Ale uważam, że runtime.Gosched() jest łatwiejszy i bardziej przejrzysty w swoim zamyśle.

Answer 2

Nie ma nic złego w ochronie MySlice = append(MySlice, &OneOfMyStructs) z sync.Mutex. Ale oczywiście możesz mieć kanał wyników o rozmiarze bufora len(params) wszystkie goroutiny wysyłają swoje odpowiedzi i po zakończeniu pracy zbieracie z tego kanału wyników.

Jeśli params ma stały rozmiar:

MySlice = make([]*MyStruct, len(params)) 
for i, param := range params { 
    wg.Add(1) 
    go func(i int, param string) { 
     defer wg.Done() 
     OneOfMyStructs := getMyStruct(param) 
     MySlice[i] = &OneOfMyStructs 
    }(i, param) 
} 

Ponieważ wszystkie goroutines napisać do innej pamięci nie jest pikantny.

Answer 3

Odpowiedź wysłana przez @jimt nie jest całkiem poprawna, ponieważ pomija ostatnią wartość wysłaną na kanale, a ostatnia defer wg.Done() nigdy nie jest wywoływana. Poniższy fragment zawiera poprawki.

https://play.golang.org/p/7N4sxD-Bai

package main 

import "fmt" 
import "sync" 

type T int 

func main() { 
    var slice []T 
    var wg sync.WaitGroup 

    queue := make(chan T, 1) 

    // Create our data and send it into the queue. 
    wg.Add(100) 
    for i := 0; i < 100; i++ { 
     go func(i int) { 
      // defer wg.Done() <- will result in the last int to be missed in the receiving channel 
      queue <- T(i) 
     }(i) 
    } 

    go func() { 
     // defer wg.Done() <- Never gets called since the 100 `Done()` calls are made above, resulting in the `Wait()` to continue on before this is executed 
     for t := range queue { 
      slice = append(slice, t) 
      wg.Done() // ** move the `Done()` call here 
     } 
    }() 

    wg.Wait() 

    // now prints off all 100 int values 
    fmt.Println(slice) 
} 

Answer 4

użyć albo rozwiązanie daniele i do WaitGroup, to znaczy, gdy ostateczna wielkość kawałek jest znany, albo use a channel do zbierania przedmiotów asynchronicznie, a następnie rozwijać kawałek dynamicznie.

Nie potrzebujesz wtedy dodatkowej grupy WaitGroup. Kanały zapewniają już wymagane funkcje synchronizacji. Po przeczytaniu danych wystarczy zamknąć kanał.

package main 

import "fmt" 

type T int 

func main() { 
    var slice []T 
    queue := make(chan T, 1) 

    // concurrently produce some data 
    for i := 0; i < 20; i++ { 
     go func(i int) { 
      queue <- T(i) 
     }(i) 
    } 

    remaining := 20 
    for t := range queue { 
     // This loop blocks until a new item is available in the channel. 
     // You can grow your slice here, but must also care to close the 
     // channel, when you decide that you obtained enough data. 
     if t != 13 { // because it is evil ;) 
      slice = append(slice, t) 
     } 
     if remaining--; remaining == 0 { 
      close(queue) // do not forget to close the channel 
     } 
    } 

    fmt.Printf("slice: %v, len: %v", slice, len(slice)) 
}