我有一个测试程序，在多个 Cpu (Goroutines = Cpus) 上执行多个 goroutine 时会给出不同的结果。 “测试”是关于使用 channel 同步 goroutines，程序本身计算字符串中字符的出现次数。它在一个 Cpu/一个 goroutine 上产生一致的结果。
请参阅 playground 上的代码示例(注意:在本地计算机上运行以在多核上执行，并观察结果数字的变化):http://play.golang.org/p/PT5jeCKgBv .
代码摘要:该程序计算 (DNA) 字符串中 4 个不同字符(A、T、G、C)的出现次数。
问题:在多个 Cpu(goroutine)上执行时，结果(出现 n 个字符)会发生变化。为什么？
描述:
- goroutine 将工作 (SpawnWork) 作为字符串发送给 Workers。设置 人工字符串输入数据(硬编码字符串被复制 n 次)。
- Goroutine Workers (Worker) 的创建数量等于 Cpu 的数量。
- Workers 检查字符串中的每个字符并计算 A、T 并发送 求和到一个 channel ，G，C 计数到另一个 channel 。
- SpawnWork 关闭 workstring channel 以控制 Worker(它使用范围消耗字符串，当输入 channel 被 SpawnWork 关闭时退出)。
- 当 Workers 消耗完其范围(字符)后，它会在退出 channel 上发送退出信号 (quit <- true)。这些“脉冲”将出现 Cpu 次数(Cpu 计数 = goroutines 计数)。
- Main (select) 循环将在收到退出的 Cpu-count 数时退出 信号。
- 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)
//罗尔夫·兰帕
最佳答案
这是一个工作版本，无论使用多少 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("----------------------------")
}
}
关于concurrency - N>1 goroutines 的不同结果(在 N>1 Cpu :s). 为什么？，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/17098722/