concurrency - N>1 goroutines 的不同结果(在 N>1 Cpu :s). 为什么？

Question

我有一个测试程序，在多个 Cpu (Goroutines = Cpus) 上执行多个 goroutine 时会给出不同的结果。 “测试”是关于使用 channel 同步 goroutines，程序本身计算字符串中字符的出现次数。它在一个 Cpu/一个 goroutine 上产生一致的结果。

请参阅 playground 上的代码示例(注意:在本地计算机上运行以在多核上执行，并观察结果数字的变化):http://play.golang.org/p/PT5jeCKgBv .

代码摘要:该程序计算 (DNA) 字符串中 4 个不同字符(A、T、G、C)的出现次数。

问题:在多个 Cpu(goroutine)上执行时，结果(出现 n 个字符)会发生变化。为什么？

描述:

1. goroutine 将工作 (SpawnWork) 作为字符串发送给 Workers。设置人工字符串输入数据(硬编码字符串被复制 n 次)。
2. Goroutine Workers (Worker) 的创建数量等于 Cpu 的数量。
3. Workers 检查字符串中的每个字符并计算 A、T 并发送求和到一个 channel ，G，C 计数到另一个 channel 。
4. SpawnWork 关闭 workstring channel 以控制 Worker(它使用范围消耗字符串，当输入 channel 被 SpawnWork 关闭时退出)。
5. 当 Workers 消耗完其范围(字符)后，它会在退出 channel 上发送退出信号 (quit <- true)。这些“脉冲”将出现 Cpu 次数(Cpu 计数 = goroutines 计数)。
6. Main (select) 循环将在收到退出的 Cpu-count 数时退出信号。
7. Main 函数打印出现的字符(A、T、G、C)的摘要。

简化代码:

1.“Worker”(goroutines)计算行中的字符数:

func Worker(inCh chan *[]byte, resA chan<- *int, resB chan<- *int, quit chan bool) {
    //for p_ch := range inCh {
    for {
        p_ch, ok := <-inCh // similar to range
        if ok {
            ch := *p_ch
            for i := 0; i < len(ch); i++ {
                if ch[i] == 'A' || ch[i] == 'T' {        // Count A:s and T:s
                    at++
                } else if ch[i] == 'G' || ch[i] == 'C' { // Count G:s and C:s
                    gc++
                }
            }
            resA <- &at  // Send line results on separate channels
            resB <- &gc  // Send line results on separate channels
        } else {
            quit <- true // Indicate that we're all done
            break
        }
    }
}

2. 向 worker 生成工作(字符串):

func SpawnWork(inStr chan<- *[]byte, quit chan bool) {
    // Artificial input data
    StringData :=
        "NNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNN\n" +
        "NTGAGAAATATGCTTTCTACTTTTTTGTTTAATTTGAACTTGAAAACAAAACACACACAA\n" +
        "... etc\n" +
    // ...
    for scanner.Scan() {
        s := scanner.Bytes()
        if len(s) == 0 || s[0] == '>' {
            continue
        } else {
            i++
            inStr <- &s
        }
    }
    close(inStr) // Indicate (to Workers) that there's no more strings coming.
}

3.主程序:

func main() {
    // Count Cpus, and count down in final select clause
    CpuCnt := runtime.NumCPU() 
    runtime.GOMAXPROCS(CpuCnt)
    // Make channels
    resChA := make(chan *int)
    resChB := make(chan *int)
    quit := make(chan bool)
    inStr := make(chan *[]byte)

    // Set up Workers ( n = Cpu )
    for i := 0; i < CpuCnt; i++ {
        go Worker(inStr, resChA, resChB, quit)
    }
    // Send lines to Workers
    go SpawnWork(inStr, quit)

    // Count the number of "A","T" & "G","C" per line 
    // (comes in here as ints per row, on separate channels (at and gt))
    for {
        select {
        case tmp_at := <-resChA:
            tmp_gc := <-resChB // Ch A and B go in pairs anyway
            A += *tmp_at       // sum of A's and T's
            B += *tmp_gc       // sum of G's and C's
        case <-quit:
            // Each goroutine sends "quit" signals when it's done. Since 
            // the number of goroutines equals the Cpu counter, we count 
            // down each time a goroutine tells us it's done (quit at 0):
            CpuCnt--
            if CpuCnt == 0 { // When all goroutines are done then we're done.
                goto out     
            }
        }
    }
out:
    // Print report to screen
}

为什么只有在单个 cpu/goroutine 上执行时，这段代码才会始终如一地计数？也就是说， channel 似乎没有同步，或者主循环在所有 goroutine 完成之前强行退出？挠头。

(同样:在 Playground 上查看/运行完整代码:http://play.golang.org/p/PT5jeCKgBv)

//罗尔夫·兰帕

Answer 1

这是一个工作版本，无论使用多少 cpu，它始终产生相同的结果。

这是我做的

• 删除 *int 的传递 - 在 channel 中传递非常活泼!
• 删除 *[]byte 的传递 - 毫无意义，因为 slice 无论如何都是引用类型
• 在将 slice 放入 channel 之前复制 slice - slice 指向同一内存导致竞争
• 修复 Worker 中 at 和 gc 的初始化 - 它们在错误的位置 - 这是导致结果差异的主要原因
• 使用sync.WaitGroup用于同步和 channel 关闭()

我使用了 -race parameter of go build查找并修复数据竞争。

package main

import (
    "bufio"
    "fmt"
    "runtime"
    "strings"
    "sync"
)

func Worker(inCh chan []byte, resA chan<- int, resB chan<- int, wg *sync.WaitGroup) {
    defer wg.Done()
    fmt.Println("Worker started...")
    for ch := range inCh {
        at := 0
        gc := 0
        for i := 0; i < len(ch); i++ {
            if ch[i] == 'A' || ch[i] == 'T' {
                at++
            } else if ch[i] == 'G' || ch[i] == 'C' {
                gc++
            }
        }
        resA <- at
        resB <- gc
    }

}

func SpawnWork(inStr chan<- []byte) {
    fmt.Println("Spawning work:")
    // An artificial input source.
    StringData :=
        "NNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNN\n" +
            "NTGAGAAATATGCTTTCTACTTTTTTGTTTAATTTGAACTTGAAAACAAAACACACACAA\n" +
            "CTTCCCAATTGGATTAGACTATTAACATTTCAGAAAGGATGTAAGAAAGGACTAGAGAGA\n" +
            "TATACTTAATGTTTTTAGTTTTTTAAACTTTACAAACTTAATACTGTCATTCTGTTGTTC\n" +
            "AGTTAACATCCCTGAATCCTAAATTTCTTCAGATTCTAAAACAAAAAGTTCCAGATGATT\n" +
            "TTATATTACACTATTTACTTAATGGTACTTAAATCCTCATTNNNNNNNNCAGTACGGTTG\n" +
            "TTAAATANNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNN\n" +
            "NNNNNNNCTTCAGAAATAAGTATACTGCAATCTGATTCCGGGAAATATTTAGGTTCATAA\n"
    // Expand data n times
    tmp := StringData
    for n := 0; n < 1000; n++ {
        StringData = StringData + tmp
    }
    scanner := bufio.NewScanner(strings.NewReader(StringData))
    scanner.Split(bufio.ScanLines)

    var i int
    for scanner.Scan() {
        s := scanner.Bytes()
        if len(s) == 0 || s[0] == '>' {
            continue
        } else {
            i++
            s_copy := append([]byte(nil), s...)
            inStr <- s_copy
        }
    }
    close(inStr)
}

func main() {
    CpuCnt := runtime.NumCPU() // Count down in select clause
    CpuOut := CpuCnt           // Save for print report
    runtime.GOMAXPROCS(CpuCnt)
    fmt.Printf("Processors: %d\n", CpuCnt)

    resChA := make(chan int)
    resChB := make(chan int)
    inStr := make(chan []byte)

    fmt.Println("Spawning workers:")
    var wg sync.WaitGroup
    for i := 0; i < CpuCnt; i++ {
        wg.Add(1)
        go Worker(inStr, resChA, resChB, &wg)
    }
    fmt.Println("Spawning work:")
    go func() {
        SpawnWork(inStr)
        wg.Wait()
        close(resChA)
        close(resChB)
    }()

    A := 0
    B := 0
    LineCnt := 0
    for tmp_at := range resChA {
        tmp_gc := <-resChB // Theese go together anyway
        A += tmp_at
        B += tmp_gc
        LineCnt++
    }

    if !(A+B > 0) {
        fmt.Println("No A/B was found!")
    } else {
        ABFraction := float32(B) / float32(A+B)
        fmt.Println("\n----------------------------")
        fmt.Printf("Cpu's  : %d\n", CpuOut)
        fmt.Printf("Lines  : %d\n", LineCnt)
        fmt.Printf("A+B    : %d\n", A+B)
        fmt.Printf("A      : %d\n", A)
        fmt.Printf("B      : %d\n", A)
        fmt.Printf("AB frac: %v\n", ABFraction*100)
        fmt.Println("----------------------------")
    }
}