sync.Cond 测试广播 - 为什么要检查循环？

Question

我试图使用sync.Cond - 等待和广播。我无法理解其中的某些部分：

等待电话的评论说：

   41    // Because c.L is not locked when Wait first resumes, the caller

   42    // typically cannot assume that the condition is true when

   43    // Wait returns.  Instead, the caller should Wait in a loop:

   44    //

   45    //    c.L.Lock()

   46    //    for !condition() {

   47    //        c.Wait()

   48    //    }

   49    //    ... make use of condition ...

   50    //    c.L.Unlock()

需要这样做的原因是什么？

所以这意味着以下程序可能不正确（尽管它可以工作）：

package main

import (

    "bufio"

    "fmt"

    "os"

    "sync"

)

type processor struct {

    n       int

    c       *sync.Cond

    started chan int

}

func newProcessor(n int) *processor {

    p := new(processor)

    p.n = n

    p.c = sync.NewCond(&sync.Mutex{})

    p.started = make(chan int, n)

    return p

}

func (p *processor) start() {

    for i := 0; i < p.n; i++ {

        go p.process(i)

    }

    for i := 0; i < p.n; i++ {

        <-p.started

    }

    p.c.L.Lock()

    p.c.Broadcast()

    p.c.L.Unlock()

}

func (p *processor) process(f int) {

    fmt.Printf("fork : %d\n", f)

    p.c.L.Lock()

    p.started <- f

    p.c.Wait()

    p.c.L.Unlock()

    fmt.Printf("process: %d - out of wait\n", f)

}

func main() {

    p := newProcessor(5)

    p.start()

    reader := bufio.NewReader(os.Stdin)

    _,_ =reader.ReadString('\n')

}

Answer 1

条件变量不会保持信号状态，它们只会唤醒在 .Wait() 中阻塞的其他 go 例程。因此，除非您有一个谓词来检查是否需要等待，或者您想要等待的事情是否已经发生，否则这会导致竞争条件。在您的特定情况下，您通过使用您的频道在 go 例程调用.Wait()和调用之间添加了同步，据我所知，这种方式不应呈现我在本文中进一步描述的竞争条件。虽然我不会打赌，但我个人只会按照文档描述的惯用方式来做。.BroadCast()p.started考虑您的start()函数正在这些行中执行广播：p.c.L.Lock()p.c.Broadcast()在那个特定的时间点考虑你的其他 go 例程之一已经在你的process()函数中到达了这个点fmt.Printf("fork : %d\n", f)go 例程要做的下一件事是锁定互斥锁（至少在 go 例程start()释放该互斥锁之前它不会拥有它）并等待条件变量。p.c.L.Lock()p.started <- fp.c.Wait()但是 Wait 永远不会回来，因为此时没有人会发出信号/广播它 - 信号已经发生。所以你需要另一个可以测试自己的条件，这样当你已经知道条件已经发生时，你不需要调用 Wait()，例如type processor struct {&nbsp; &nbsp; n&nbsp; &nbsp; &nbsp; &nbsp;int&nbsp; &nbsp; c&nbsp; &nbsp; &nbsp; &nbsp;*sync.Cond&nbsp; &nbsp; started chan int&nbsp; &nbsp; done&nbsp; &nbsp; bool //added}...func (p *processor) start() {&nbsp; &nbsp; for i := 0; i < p.n; i++ {&nbsp; &nbsp; &nbsp; &nbsp; go p.process(i)&nbsp; &nbsp; }&nbsp; &nbsp; for i := 0; i < p.n; i++ {&nbsp; &nbsp; &nbsp; &nbsp; <-p.started&nbsp; &nbsp; }&nbsp; &nbsp; p.c.L.Lock()&nbsp; &nbsp; p.done = true //added&nbsp; &nbsp; p.c.Broadcast()&nbsp; &nbsp; p.c.L.Unlock()}func (p *processor) process(f int) {&nbsp; &nbsp; fmt.Printf("fork : %d\n", f)&nbsp; &nbsp; p.c.L.Lock()&nbsp; &nbsp; p.started <- f&nbsp; &nbsp; for !p.done { //added&nbsp; &nbsp; &nbsp; &nbsp; p.c.Wait()&nbsp; &nbsp; }&nbsp; &nbsp; p.c.L.Unlock()&nbsp; &nbsp; fmt.Printf("process: %d - out of wait\n", f)}