并行 For 循环

Question

我希望 for 循环使用 go 例程并行。我尝试使用频道，但没有奏效。我的主要问题是，我想在继续之前等待所有迭代完成。这就是为什么在它不起作用之前简单地写go。我尝试使用通道（我认为是错误的方式），但这让我的代码更慢

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
}

我的结构如下所示：

type Individual struct {
    gene []bool
    fitness int
}

Answer 1

所以基本上 goroutine 不应该返回一个值，而是将它推下一个通道。如果你想等待所有的 goroutines 完成，你可以计算 goroutines 的数量，或者使用 WaitGroup。在这个例子中，这是一个矫枉过正的做法，因为大小是已知的，但无论如何这是一个很好的做法。这是一个修改过的例子：

package main

import (
    "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

对于您的具体问题，您根本不需要使用渠道。

但是，除非您的 createIndividual 花费一些时间进行计算，否则当并行运行时，goroutine 之间的上下文切换总是会慢得多。

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

输出：

┌─ 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 3

向这样的循环添加受控并行性的一种常见方法是生成许多将从通道读取任务的工作 goroutines。 runtime.NumCPU 函数可能有助于确定产生多少工人是有意义的（确保您适当地设置 GOMAXPROCS 以利用这些 CPU）。然后，您只需将作业写入通道，它们将由工作人员处理。

在这种情况下，工作是初始化人口切片的元素，因此使用*Individual 指针的通道可能是有意义的。像这样的：

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)

worker goroutine 看起来像这样：

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

由于任务没有提前分配，createIndividual 花费可变的时间并不重要：每个工人只会在最后一个完成时承担新任务，并在有时退出没有剩余任务（因为此时通道已关闭）。

但是我们如何知道工作何时完成？ sync.WaitGroup 类型可以在这里提供帮助。生成工作 goroutine 的代码可以这样修改：

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

initIndividuals函数也被修改为带附加参数，并添加defer wg.Done()作为第一条语句。现在对 wg.Wait() 的调用将阻塞，直到所有工作 goroutine 都完成。然后，您可以返回完全构造的 population 切片。

Answer 4

如果您想避免将并发逻辑与业务逻辑混为一谈，我编写了这个库https://github.com/shomali11/parallelizer 来帮助您。它封装了并发逻辑，所以你不用担心。

所以在你的例子中：

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
}

Answer 5

由于您事先知道您将拥有多少个人，我会避免使用频道，而只是在 goroutine createIndividual 中分配 population 的个人成员。 createIndividual 的签名将如下所示：

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

调用代码如下所示：

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

所以，每个 goroutine 只负责一个人，它分配给population 中的相应槽：

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

    // assign other attributes to `individual`
}

您可以查看完整的代码示例on play here。