竞赛暂停一组 goroutine答案

【问题标题】：Race pausing a group of goroutines竞赛暂停一组 goroutine
【发布时间】：2018-10-19 01:25:27
【问题描述】：

我有一堆 goroutine 在循环中做某事。我希望能够暂停所有这些，运行一些任意代码，然后恢复它们。我尝试这样做的方式可能不是惯用的（我希望有更好的解决方案），但我不明白为什么它不起作用。

精简到本质（驱动代码在底部）：

type looper struct {
    pause  chan struct{}
    paused sync.WaitGroup
    resume chan struct{}
}

func (l *looper) loop() {
    for {
        select {
        case <-l.pause:
            l.paused.Done()
            <-l.resume
        default:
            dostuff()
        }
    }
}

func (l *looper) whilePaused(fn func()) {
    l.paused.Add(32)
    l.resume = make(chan struct{})
    close(l.pause)
    l.paused.Wait()
    fn()
    l.pause = make(chan struct{})
    close(l.resume)
}

我启动了 32 个 goroutine，全部运行 loop()，然后连续调用 whilePaused 100 次，似乎一切正常……但如果我使用 -race 运行它，它会告诉我在 @ 上存在比赛987654327@ 在写到whilePaused (l.resume = make(chan struct{})) 和读到loop (<-l.resume) 之间。

我不明白为什么会这样。根据The Go Memory Model，close(l.pause) 应该发生在每个loop goroutine 中的<-l.pause 之前。这应该意味着 make(chan struct{}) 值在所有这些 loop goroutine 中作为 l.resume 的值可见，就像字符串 "hello world" 在 f goroutine 中作为 a 的值可见一样在文档示例中。

一些可能相关的附加信息：

如果我将 l.resume 替换为 unsafe.Pointer 并在 loop 中使用 atomic.LoadPointer 访问 chan struct{} 值，在 whilePaused 中访问 atomic.StorePointer 值，比赛就结束了。这似乎提供了与频道应该提供的完全相同的获取-释放顺序？
如果我在l.paused.Done()和<-l.resume之间添加time.Sleep(10 * time.Microsecond)，程序通常会在调用fn一两次后死锁。

.

这是我的其余代码，以防你想运行整个程序：

package main

import (
    "fmt"
    "sync"
    "sync/atomic"
)

// looper code from above

var n int64    
func dostuff() {
    atomic.AddInt64(&n, 1)
}

func main() {
    l := &looper{
        pause: make(chan struct{}),
    }
    var init sync.WaitGroup
    init.Add(32)
    for i := 0; i < 32; i++ {
        go func() {
            init.Done()
            l.loop()
        }()
    }
    init.Wait()
    for i := 0; i < 100; i++ {
        l.whilePaused(func() { fmt.Printf("%d ", i) })
    }
    fmt.Printf("\n%d\n", atomic.LoadInt64(&n))
}

【问题讨论】：

频道对于这种模式来说有点尴尬。这看起来是使用 sync.Condition 的完美候选人。
@JimB 除非 Go 条件与 POSIX/etc 有很大不同。在这种情况下，它们对于尝试等待的情况也很尴尬，在这种情况下你不能只阻止for !ready { c.Wait() }。另外，在我的真实代码中，case <-l.pause: 将在 select 内，我已经因为其他原因需要它，所以要使用 Condition，我是否必须有一个等待条件并发送的 goroutine还是一个频道？
是的，如果您需要针对该逻辑条件进行选择，sync.Cond 将无法正常工作。我看到重复的广播模式并立即跳转到条件变量，但实际上我也很少使用它们；）
@JimB 我的第一个想法也是一个条件，但那是因为我更习惯于带有任务结构队列的 C 风格线程池（或 Python/C#/etc. 执行器，我只是在将来的回调中恢复暂停任务）。我确信对于我正在尝试做的事情有一个更好的基于频道的习语，我只是不确定它是什么。（事实上，我的 WaitGroup 必须知道我有多少个 goroutines 似乎很笨拙……）
当然，如果没有完整的上下文，很难做出推荐，但您可以尝试重新考虑您在这里使用的整体模式。一个常见的陷阱是将 goroutine 视为线程。调度 goroutine 非常快，当您将它们视为一次性的、按需创建和销毁它们时，通常会有更简单的模式可用。

标签： go synchronization channel

【解决方案1】：

这是因为在线程执行 l.paused.Done() 之后，另一个线程能够绕过循环并再次分配 l.resume

这是操作顺序

Looper thread    |    Pauser thread
------------------------------------
l.paused.Done()  |   
                 |   l.paused.Wait()
                 |   l.pause = make(chan struct{})
                 |   round the loop
                 |   l.paused.Add(numThreads)
<- l.resume      |   l.resume = make(chan struct{})   !!!RACE!!

【讨论】：