我正在玩一些用于學習目的的代碼，并且在使用標志時，我得到了一個關于其執行的競爭條件，我想了解原因。該代碼啟動一組固定的 goroutine，這些 goroutine 充當從通道消耗任務的工作線程，沒有固定數量的任務，只要通道接收到工作線程必須繼續工作的任務。-race我在調用函數時遇到爭用條件。根據我的理解（看看數據競爭報告），當第一個調用由一個生成的goroutine和主要例程調用同時執行時，就會發生競爭條件。這是對的嗎？如果是這樣，這意味著我必須始終在主例程上執行對Add的調用，以避免資源上的這種競爭？但是，這也意味著我需要知道工人需要提前處理多少任務，如果我需要代碼處理一旦工人運行可能遇到的任何數量的任務，那么哪種任務很糟糕......WaitGroupwg.Addwg.Wait代碼：func Test(t *testing.T) { t.Run("", func(t *testing.T) { var wg sync.WaitGroup queuedTaskC := make(chan func()) for i := 0; i < 5; i++ { wID := i + 1 go func(workerID int) { for task := range queuedTaskC { wg.Add(1) task() } }(wID) } taskFn := func() { fmt.Println("executing task...") wg.Done() } queuedTaskC <- taskFn queuedTaskC <- taskFn queuedTaskC <- taskFn queuedTaskC <- taskFn queuedTaskC <- taskFn queuedTaskC <- taskFn queuedTaskC <- taskFn queuedTaskC <- taskFn queuedTaskC <- taskFn wg.Wait() close(queuedTaskC) fmt.Println(len(queuedTaskC)) })}報告：==================WARNING: DATA RACERead at 0x00c0001280d8 by goroutine 11: internal/race.Read() /src/internal/race/race.go:37 +0x206 sync.(*WaitGroup).Add() /src/sync/waitgroup.go:71 +0x219 workerpool.Test.func1.1() /workerpool/workerpool_test.go:36 +0x64Previous write at 0x00c0001280d8 by goroutine 8: internal/race.Write() /src/internal/race/race.go:41 +0x125 sync.(*WaitGroup).Wait() /src/sync/waitgroup.go:128 +0x126 workerpool.Test.func1() /workerpool/workerpool_test.go:57 +0x292 testing.tRunner() /src/testing/testing.go:1123 +0x202Goroutine 11 (running) created at: workerpool.Test.func1() /workerpool/workerpool_test.go:34 +0xe4 testing.tRunner() /src/testing/testing.go:1123 +0x202
1 回答
繁華開滿天機
TA貢獻1816條經驗 獲得超4個贊
WaitGroup實現基于內部計數器，該計數器由 和 方法更改。在計數器歸零之前，該方法不會返回。也可以重復使用，但在文檔中描述的某些條件下：AddDoneWaitWaitGroup
// If a WaitGroup is reused to wait for several independent sets of events,
// new Add calls must happen after all previous Wait calls have returned.
盡管您的代碼沒有重用，但它能夠多次將計數器清零。當在給定時間沒有處理任何任務時，就會發生這種情況，這在并發代碼中是完全可能的。由于您的代碼在調用之前不會等待返回，因此您會收到爭用條件錯誤。wgWaitGroupWaitAdd
正如每個人在評論中建議的那樣，您應該放棄跟蹤任務的想法，轉而控制正在運行的戈魯丁。附加代碼建議。WaitGroup
func Test(t *testing.T) {
var wg sync.WaitGroup
queuedTaskC := make(chan func(), 10)
for i := 0; i < 5; i++ {
wID := i + 1
wg.Add(1)
go func(workerID int) {
defer wg.Done()
for task := range queuedTaskC {
task()
}
}(wID)
}
for i := 0; i < 10; i++ {
queuedTaskC <- func() {
fmt.Println("executing task...")
}
}
close(queuedTaskC)
wg.Wait()
fmt.Println(len(queuedTaskC))
}
- 1 回答
- 0 關注
- 106 瀏覽
添加回答
舉報
0/150
提交
取消