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 mainimport (    "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 <- f

p.c.Wait()

但是 Wait 永远不会回来，因为此时没有人会发出信号/广播它 - 信号已经发生。

所以你需要另一个可以测试自己的条件，这样当你已经知道条件已经发生时，你不需要调用 Wait()，例如

type processor struct {

    n       int

    c       *sync.Cond

    started chan int

    done    bool //added

}

...

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.done = true //added

    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

    for !p.done { //added

        p.c.Wait()

    }

    p.c.L.Unlock()

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

}