Parallel For-Loop

Question

Desidero che il ciclo sia parallelo utilizzando le routine di go. Ho provato a usare i canali ma non ha funzionato. Il mio problema principale è che voglio aspettare che tutte le iterazioni siano terminate prima di continuare. Ecco perché scrivere semplicemente go prima che non funzioni. Ho cercato di utilizzare i canali (penso che il modo sbagliato), ma che ha reso il mio codice ancora più lento

func createPopulation(populationSize int, individualSize int) []Individual { 
    population := make([]Individual, populationSize) 

    //i want this loop to be work parallel 
    for i := 0; i < len(population); i++ { 
     population[i] = createIndividual(individualSize) 
    } 

    return population 
} 

func createIndividual(size int) Individual { 
    var individual = Individual{make([]bool, size), 0} 

    for i := 0; i < len(individual.gene); i++ { 
     if rand.Intn(2)%2 == 1 { 
      individual.gene[i] = true 
     } else { 
      individual.gene[i] = false 
     } 
    } 

    return individual 
} 

mio struct è simile al seguente:

type Individual struct { 
    gene []bool 
    fitness int 
} 

Answer 1

Quindi, in pratica il goroutine non dovrebbe restituire un valore, ma spinta giù per un canale. Se vuoi aspettare che tutte le goroutine finiscano, puoi contare sul numero di goroutine o usare un WaitGroup. In questo esempio è eccessivo perché la dimensione è nota, ma è comunque una buona pratica. Ecco un esempio modificato:

package main 

import (
    "fmt" 
    "math/rand" 
    "sync" 
) 

type Individual struct { 
    gene []bool 
    fitness int 
} 


func createPopulation(populationSize int, individualSize int) []Individual { 

    // we create a slice with a capacity of populationSize but 0 size 
    // so we'll avoid extra unneeded allocations 
    population := make([]Individual, 0, populationSize) 

    // we create a buffered channel so writing to it won't block while we wait for the waitgroup to finish 
    ch := make(chan Individual, populationSize) 

    // we create a waitgroup - basically block until N tasks say they are done 
    wg := sync.WaitGroup{} 

    for i := 0; i < populationSize; i++ { 

     //we add 1 to the wait group - each worker will decrease it back 
     wg.Add(1) 

     //now we spawn a goroutine 
     go createIndividual(individualSize, ch, &wg) 
    } 

    // now we wait for everyone to finish - again, not a must. 
    // you can just receive from the channel N times, and use a timeout or something for safety 
    wg.Wait() 

    // we need to close the channel or the following loop will get stuck 
    close(ch) 

    // we iterate over the closed channel and receive all data from it 
    for individual := range ch { 

     population = append(population, individual) 
    } 
    return population 

} 

func createIndividual(size int, ch chan Individual, wg *sync.WaitGroup) { 

    var individual = Individual{make([]bool, size), 0} 

    for i := 0; i < len(individual.gene); i++ { 
     if rand.Intn(2)%2 == 1 { 
      individual.gene[i] = true 
     } else { 
      individual.gene[i] = false 
     } 
    } 

    // push the population object down the channel 
    ch <- individual 
    // let the wait group know we finished 
    wg.Done() 

} 

Answer 2

Dal momento che si sapere in anticipo quante persone si avrà, vorrei evitare di utilizzare i canali e basta assegnare i singoli membri del population nel goroutine createIndividual. La firma di createIndividual sarebbe quindi simile a questa:

func createIndividual(wg *sync.WaitGroup, individual *Individual, size int) 

e il codice chiamante sarebbe simile a questa:

population := make([]Individual, populationSize) 
wg := &sync.WaitGroup{} 
wg.Add(len(population)) 

for i := 0; i < len(population); i++ { 
    go createIndividual(wg, &population[i], individualSize) 
} 

wg.Wait() 

Così, ogni andare routine è responsabile esattamente un individuo, che assegna alla relativa scanalatura in population:

func createIndividual(wg *sync.WaitGroup, individual *Individual, size int) { 
    defer wg.Done() 
    *individual = Individual{make([]bool, size), 0} 

    // assign other attributes to `individual` 
} 

È possibile vedere un esempio di codice completo on play here.

Answer 3

Un modo comune per aggiungere un parallelismo controllato a un ciclo come questo è generare un numero di goroutine di lavoro che leggeranno le attività da un canale. La funzione runtime.NumCPU può aiutare a decidere quanti lavoratori ha il senso di spawnare (assicurati di impostare GOMAXPROCS in modo appropriato per trarre vantaggio da tali CPU). Quindi scrivi semplicemente i lavori sul canale e questi verranno gestiti dagli operatori.

In questo caso in cui il lavoro deve inizializzare gli elementi della porzione di popolazione, è consigliabile utilizzare un canale di punti di riferimento *Individual. Qualcosa di simile a questo:

ch := make(chan *Individual) 
for i := 0; i < nworkers; i++ { 
    go initIndividuals(individualSize, ch) 
} 

population := make([]Individual, populationSize) 
for i := 0; i < len(population); i++ { 
    ch <- &population[i] 
} 
close(ch) 

Il goroutine operaio sarebbe simile a questa:

func initIndividuals(size int, ch <-chan *Individual) { 
    for individual := range ch { 
     // Or alternatively inline the createIndividual() code here if it is the only call 
     *individual = createIndividual(size) 
    } 
} 

Dal momento che le attività non sono porzionati fuori prima del tempo, non importa se createIndividual prende una quantità di tempo variabile : ogni lavoratore assumerà una nuova attività solo quando l'ultimo è completo e uscirà quando non ci sono più attività (dal momento che il canale è chiuso in quel punto).

Ma come facciamo a sapere quando il lavoro è terminato? Il tipo sync.WaitGroup può essere d'aiuto qui. Il codice per deporre le uova le goroutines lavoratore può essere modificato in questo modo:

ch := make(chan *Individual) 
var wg sync.WaitGroup 
wg.Add(nworkers) 
for i := 0; i < nworkers; i++ { 
    go initIndividuals(individualSize, ch, &wg) 
} 

La funzione initIndividuals è inoltre modificato per prendere il parametro aggiuntivo, e aggiungere defer wg.Done() come la prima istruzione. Ora una chiamata a wg.Wait() bloccherà fino a quando tutte le goroutine non saranno state completate.È quindi possibile restituire la slice population completamente costruita.

Answer 4

Per il tuo problema specifico non è necessario utilizzare i canali.

Tuttavia, a meno che il tuo createIndividual impieghi un po 'di tempo a fare calcoli, il contesto cambia tra le goroutine sempre sarà molto più lento quando viene eseguito in parallelo.

type Individual struct { 
    gene []bool 
    fitness int 
} 

func createPopulation(populationSize int, individualSize int) (population []*Individual) { 
    var wg sync.WaitGroup 
    population = make([]*Individual, populationSize) 

    wg.Add(populationSize) 
    for i := 0; i < populationSize; i++ { 
     go func(i int) { 
      population[i] = createIndividual(individualSize) 
      wg.Done() 
     }(i) 
    } 
    wg.Wait() 
    return 
} 

func createIndividual(size int) *Individual { 
    individual := &Individual{make([]bool, size), 0} 

    for i := 0; i < size; i++ { 
     individual.gene[i] = rand.Intn(2)%2 == 1 
    } 

    return individual 
} 

func main() { 
    numcpu := flag.Int("cpu", runtime.NumCPU(), "") 
    flag.Parse() 
    runtime.GOMAXPROCS(*numcpu) 
    pop := createPopulation(1e2, 21e3) 
    fmt.Println(len(pop)) 
} 

uscita:

┌─ oneofone@Oa [/tmp]                            
└──➜ go build blah.go; xtime ./blah -cpu 1 
100 
0.13u 0.00s 0.13r 4556kB ./blah -cpu 1 
┌─ oneofone@Oa [/tmp]                            
└──➜ go build blah.go; xtime ./blah -cpu 4 
100 
2.10u 0.12s 0.60r 4724kB ./blah -cpu 4 

Answer 5

Se volete evitare di mescolare la logica di concorrenza con la logica di business, ho scritto questa libreria https://github.com/shomali11/parallelizer per aiutarvi in ​​questo. Incapsula la logica della concorrenza in modo da non doversene preoccupare.

Quindi nel tuo esempio:

package main 

import (
    "github.com/shomali11/parallelizer" 
    "fmt" 
) 

func main() { 
    populationSize := 100 
    results = make([]*Individual, populationSize) 

    options := &Options{ Timeout: time.Second } 
    group := parallelizer.NewGroup(options) 
    for i := 0; i < populationSize; i++ { 
     group.Add(func(index int, results *[]*Individual) { 
      return func() { 
       ... 

       results[index] = &Individual{...} 
      } 
     }(i, &results)) 
    } 

    err := group.Run() 

    fmt.Println("Done") 
    fmt.Println(fmt.Sprintf("Results: %v", results)) 
    fmt.Printf("Error: %v", err) // nil if it completed, err if timed out 
}