假设在任意位置,registerChan上最多有两个元素(工作人员地址)。然后由于某种原因,以下代码不会在最后两个goroutine中调用wg.Done()。
func schedule(jobName string, mapFiles []string, nReduce int, phase jobPhase, registerChan chan string) {
var ntasks int
var nOther int // number of inputs (for reduce) or outputs (for map)
switch phase {
case mapPhase:
ntasks = len(mapFiles)
nOther = nReduce
case reducePhase:
ntasks = nReduce
nOther = len(mapFiles)
}
fmt.Printf("Schedule: %v %v tasks (%d I/Os)\n", ntasks, phase, nOther)
const rpcname = "Worker.DoTask"
var wg sync.WaitGroup
for taskNumber := 0; taskNumber < ntasks; taskNumber++ {
file := mapFiles[taskNumber%len(mapFiles)]
taskArgs := DoTaskArgs{jobName, file, phase, taskNumber, nOther}
wg.Add(1)
go func(taskArgs DoTaskArgs) {
workerAddr := <-registerChan
print("hello\n")
// _ = call(workerAddr, rpcname, taskArgs, nil)
registerChan <- workerAddr
wg.Done()
}(taskArgs)
}
wg.Wait()
fmt.Printf("Schedule: %v done\n", phase)
}
如果我将wg.Done()
放在registerChan <- workerAddr
之前,那很好,我也不知道为什么。我也尝试过推迟wg.Done(),但是即使我希望这样做也似乎不起作用。我认为我对常规程序和渠道的工作方式有一些误解,这引起了我的困惑。
因为它已停止在这里:
workerAddr := <-registerChan
对于缓冲通道:为了使此workerAddr := <-registerChan
工作:通道registerChan
必须有一个值;否则,代码将在此处等待通道。
我设法以这种方式运行您的代码(尝试this):
package main
import (
"fmt"
"sync"
)
func main() {
registerChan := make(chan int, 1)
for i := 1; i <= 10; i++ {
wg.Add(1)
go fn(i, registerChan)
}
registerChan <- 0 // seed
wg.Wait()
fmt.Println(<-registerChan)
}
func fn(taskArgs int, registerChan chan int) {
workerAddr := <-registerChan
workerAddr += taskArgs
registerChan <- workerAddr
wg.Done()
}
var wg sync.WaitGroup
输出:
55
说明:此代码使用一个通道将1到10加到10个goroutines加一个main goroutine。
我希望这会有所帮助。
运行此语句registerChan <- workerAddr
时,如果通道容量已满,则无法添加,它将阻塞。如果您有一个名为10的workerAddr的池,则可以在调用schedule
之前将它们全部添加到容量为10的缓冲通道中。调用后不要添加,以确保如果您从通道中获取一个值,则有空间可以在其后再次添加它。在goroutine的开头使用defer
是很好的。