固定工人数量模式的竞争条件

Question

我正在玩一些用于学习目的的代码，并且在使用标志时，我得到了一个关于其执行的竞争条件，我想了解原因。该代码启动一组固定的 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

Answer 1

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))

}