concurrency - 为什么添加并发会减慢这个 golang 代码？

Question

我有一些 Go 代码，我一直在修补这些代码来回答我对我姐夫玩的视频游戏的一点好奇心。

从本质上讲，下面的代码模拟了游戏中与怪物的互动，以及他在击败怪物后多久会掉落物品。我遇到的问题是，我希望这样的一段代码非常适合并行化，但是当我添加并发时，执行所有模拟所需的时间往往会比原来慢 4-6 倍没有并发。

为了让您更好地理解代码的工作原理，我提供了三个主要功能: 交互功能，它是玩家和怪物之间的简单交互。如果怪物掉落物品，则返回 1，否则返回 0。模拟函数运行多个交互并返回一段交互结果(即，1 和 0 代表成功/不成功的交互)。最后是测试函数，它运行一组模拟并返回模拟结果的一部分，这些结果是导致丢弃项目的交互总数。这是我尝试并行运行的最后一个函数。

现在，我可以理解为什么如果我为每个要运行的测试创建一个 goroutine 代码会变慢。假设我正在运行 100 个测试，我的 MacBook Air 的 4 个 CPU 上的每个 goroutine 之间的上下文切换会降低性能，但我只创建与处理器数量一样多的 goroutine，并将测试数量划分为协程。我希望这实际上会加快代码的性能，因为我正在并行运行我的每个测试，但是，当然，我的速度会大大减慢。

我很想弄清楚为什么会这样，所以任何帮助将不胜感激。

下面是没有 go 例程的常规代码:

package main

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

const (
    NUMBER_OF_SIMULATIONS = 1000
    NUMBER_OF_INTERACTIONS = 1000000
    DROP_RATE = 0.0003
)

/**
 * Simulates a single interaction with a monster
 *
 * Returns 1 if the monster dropped an item and 0 otherwise
 */
func interaction() int {
    if rand.Float64() <= DROP_RATE {
        return 1
    }
    return 0
}

/**
 * Runs several interactions and retuns a slice representing the results
 */
func simulation(n int) []int {
    interactions := make([]int, n)
    for i := range interactions {
        interactions[i] = interaction()
    }
    return interactions
}

/**
 * Runs several simulations and returns the results
 */
func test(n int) []int {
    simulations := make([]int, n)
    for i := range simulations {
        successes := 0
        for _, v := range simulation(NUMBER_OF_INTERACTIONS) {
            successes += v
        }
        simulations[i] = successes
    }
    return simulations
}

func main() {
    rand.Seed(time.Now().UnixNano())
    fmt.Println("Successful interactions: ", test(NUMBER_OF_SIMULATIONS))
}

而且，这是与 goroutines 的并发代码:

package main

import (
    "fmt"
    "math/rand"
    "time"
    "runtime"
)

const (
    NUMBER_OF_SIMULATIONS = 1000
    NUMBER_OF_INTERACTIONS = 1000000
    DROP_RATE = 0.0003
)

/**
 * Simulates a single interaction with a monster
 *
 * Returns 1 if the monster dropped an item and 0 otherwise
 */
func interaction() int {
    if rand.Float64() <= DROP_RATE {
        return 1
    }
    return 0
}

/**
 * Runs several interactions and retuns a slice representing the results
 */
func simulation(n int) []int {
    interactions := make([]int, n)
    for i := range interactions {
        interactions[i] = interaction()
    }
    return interactions
}

/**
 * Runs several simulations and returns the results
 */
func test(n int, c chan []int) {
    simulations := make([]int, n)
    for i := range simulations {
        for _, v := range simulation(NUMBER_OF_INTERACTIONS) {
            simulations[i] += v
        }
    }
    c <- simulations
}

func main() {
    rand.Seed(time.Now().UnixNano())

    nCPU := runtime.NumCPU()
    runtime.GOMAXPROCS(nCPU)
    fmt.Println("Number of CPUs: ", nCPU)

    tests := make([]chan []int, nCPU)
    for i := range tests {
        c := make(chan []int)
        go test(NUMBER_OF_SIMULATIONS/nCPU, c)
        tests[i] = c
    }

    // Concatentate the test results
    results := make([]int, NUMBER_OF_SIMULATIONS)
    for i, c := range tests {
        start := (NUMBER_OF_SIMULATIONS/nCPU) * i
        stop := (NUMBER_OF_SIMULATIONS/nCPU) * (i+1)
        copy(results[start:stop], <-c)
    }

    fmt.Println("Successful interactions: ", results)
}

更新 (01/12/13 18:05)

我在下面添加了一个新版本的并发代码，它根据下面“系统”的建议为每个 goroutine 创建一个新的 Rand 实例。与代码的串行版本相比，我现在看到了非常小的速度提升(总时间减少了大约 15-20%)。我很想知道为什么我没有看到时间减少接近 75%，因为我将工作量分散到我的 MBA 的 4 个内核上。有没有人有任何进一步的建议可以提供帮助？

package main

import (
    "fmt"
    "math/rand"
    "time"
    "runtime"
)

const (
    NUMBER_OF_SIMULATIONS = 1000
    NUMBER_OF_INTERACTIONS = 1000000
    DROP_RATE = 0.0003
)

/**
 * Simulates a single interaction with a monster
 *
 * Returns 1 if the monster dropped an item and 0 otherwise
 */
func interaction(generator *rand.Rand) int {
    if generator.Float64() <= DROP_RATE {
        return 1
    }
    return 0
}

/**
 * Runs several interactions and retuns a slice representing the results
 */
func simulation(n int, generator *rand.Rand) []int {
    interactions := make([]int, n)
    for i := range interactions {
        interactions[i] = interaction(generator)
    }
    return interactions
}

/**
 * Runs several simulations and returns the results
 */
func test(n int, c chan []int) {
    source := rand.NewSource(time.Now().UnixNano())
    generator := rand.New(source)
    simulations := make([]int, n)
    for i := range simulations {
        for _, v := range simulation(NUMBER_OF_INTERACTIONS, generator) {
            simulations[i] += v
        }
    }
    c <- simulations
}

func main() {
    rand.Seed(time.Now().UnixNano())

    nCPU := runtime.NumCPU()
    runtime.GOMAXPROCS(nCPU)
    fmt.Println("Number of CPUs: ", nCPU)

    tests := make([]chan []int, nCPU)
    for i := range tests {
        c := make(chan []int)
        go test(NUMBER_OF_SIMULATIONS/nCPU, c)
        tests[i] = c
    }

    // Concatentate the test results
    results := make([]int, NUMBER_OF_SIMULATIONS)
    for i, c := range tests {
        start := (NUMBER_OF_SIMULATIONS/nCPU) * i
        stop := (NUMBER_OF_SIMULATIONS/nCPU) * (i+1)
        copy(results[start:stop], <-c)
    }

    fmt.Println("Successful interactions: ", results)
}

更新 (01/13/13 17:58)

感谢大家帮助解决我的问题。我终于得到了我正在寻找的答案，所以我想我会在这里为任何有同样问题的人总结一下。

基本上我有两个主要问题:第一，即使我的代码是 embarrassingly parallel ，当我将它拆分到可用的处理器中时，它的运行速度变慢了，其次，该解决方案引发了另一个问题，即我的串行代码的运行速度是在单处理器上运行的并发代码的两倍，而这正是您期望的大致相同。在这两种情况下，问题都是随机数生成器函数 rand.Float64 .基本上，这是 rand提供的便利功能。包裹。在那个包中，一个全局实例 Rand struct 由每个便利函数创建和使用。这个全局 Rand实例有一个与之关联的互斥锁。由于我使用了这个方便的函数，我无法真正并行化我的代码，因为每个 goroutine 都必须排队才能访问全局 Rand实例。解决方案(如下面的“系统”建议)是创建 Rand 的单独实例。每个 goroutine 的结构。这解决了第一个问题，但造成了第二个问题。

第二个问题是我的非并行并发代码(即我的并发代码仅使用一个处理器运行)的运行速度是顺序代码的两倍。这样做的原因是，即使我只使用一个处理器和一个 goroutine 运行，该 goroutine 也有自己的 Rand 实例。我创建的结构，并且我在没有互斥锁的情况下创建了它。顺序码仍在使用 rand.Float64利用全局互斥保护的便利函数 Rand实例。获取该锁的成本导致顺序代码运行速度减慢两倍。

因此，这个故事的寓意是，无论何时性能很重要，请确保创建 Rand 的实例。 struct 并从中调用您需要的函数，而不是使用包提供的便利函数。

Answer 1

问题似乎来自您对 rand.Float64() 的使用。 ，它使用一个带有互斥锁的共享全局对象。

相反，如果为每个 CPU 创建一个单独的 rand.New() ，通过它到interactions() ，并用它来创建 Float64() ，有很大的改进。

更新以显示现在使用 rand.New() 的问题中新示例代码的更改
test()函数被修改为使用给定的 channel ，或返回结果。

func test(n int, c chan []int) []int {
    source := rand.NewSource(time.Now().UnixNano())
    generator := rand.New(source)
    simulations := make([]int, n)
    for i := range simulations {
        for _, v := range simulation(NUMBER_OF_INTERACTIONS, generator) {
            simulations[i] += v
        }   
    }   
    if c == nil {
        return simulations
    }   
    c <- simulations
    return nil 
}

main()函数已更新以运行两个测试，并输出计时结果。

func main() {
    rand.Seed(time.Now().UnixNano())

    nCPU := runtime.NumCPU()
    runtime.GOMAXPROCS(nCPU)
    fmt.Println("Number of CPUs: ", nCPU)

    start := time.Now()
    fmt.Println("Successful interactions: ", len(test(NUMBER_OF_SIMULATIONS, nil)))
    fmt.Println(time.Since(start))

    start = time.Now()
    tests := make([]chan []int, nCPU)
    for i := range tests {
        c := make(chan []int)
        go test(NUMBER_OF_SIMULATIONS/nCPU, c)
        tests[i] = c
    }

    // Concatentate the test results
    results := make([]int, NUMBER_OF_SIMULATIONS)
    for i, c := range tests {
        start := (NUMBER_OF_SIMULATIONS/nCPU) * i
        stop := (NUMBER_OF_SIMULATIONS/nCPU) * (i+1)
        copy(results[start:stop], <-c)
    }
    fmt.Println("Successful interactions: ", len(results))
    fmt.Println(time.Since(start))
}

输出是我收到的:

> CPU 数量:2
>
> 成功互动:1000
> 1m20.39959s
>
> 成功互动:1000
> 41.392299s