Выход из цикла навсегда с использованием каналов - проблемы с Go Playground

Question

Я пытаюсь реализовать простую логику, при которой производитель отправляет данные в канал ch с навсегда for цикл, и Потребитель читает из канала ch.

Продюсер прекращает производство и выходит из цикла навсегда, когда получает сигнал на канале. quit.

Код такой (см. Также эту площадку)

func main() {
    ch := make(chan int)
    quit := make(chan bool)
    var wg sync.WaitGroup
    wg.Add(1)
    go produce(ch, quit, &wg)
    go consume(ch)
    time.Sleep(1 * time.Millisecond)
    fmt.Println("CLOSE")
    close(quit)
    wg.Wait()
}

func produce(ch chan int, quit chan bool, wg *sync.WaitGroup) {
    for i := 0; ; i++ {
        select {
        case <-quit:
            close(ch)
            fmt.Println("exit")
            wg.Done()
            return //we exit
        default:
            ch <- i
            fmt.Println("Producer sends", i)
        }
    }
}

func consume(ch chan int) {
    for {
        runtime.Gosched() // give the opportunity to the main goroutine to close the "quit" channel
        select {
        case i, more := <-ch:
            if !more {
                fmt.Println("exit consumer")
                return
            }
            fmt.Println("Consumer receives", i)
        }
    }
}

Если я запустил этот фрагмент кода на своей машине (Mac с 4 ядрами), все будет работать нормально. Если я попробую тот же код на Go Playgroud, он всегда истечет. Я предполагаю, что это потому, что Go Playground - одноядерный, и поэтому бесконечный цикл не дает возможности запускать другие горутины, но тогда я не понимаю, почему инструкция runtime.Gosched() не имеет никакого эффекта.

Просто для полноты картины я видел это, если я поставлю GOMAXPROCS=1на моем Mac программа по-прежнему работает нормально и завершается должным образом. Если я установлю GOMAXPROCS=1 на моем Mac и удалите runtime.Gosched() инструкции, поведение становится неустойчивым: иногда программа завершается, как ожидалось, иногда кажется, что никогда не выходит из бесконечного цикла.

Answer 1

Вы создали патологическую ситуацию, которой не должно быть в реальной программе, поэтому планировщик не оптимизирован для этого. В сочетании с реализацией фальшивого времени на игровой площадке, вы получаете слишком много циклов производителя и потребителя, прежде чем достигнете тайм-аута.

Горутин-производитель создает ценности как можно быстрее, а потребитель всегда готов их принять. С участием GOMAPXPROCS=1, планировщик тратит все свое время, прыгая между ними, прежде чем он будет вынужден прервать доступную работу для проверки основной горутины, что занимает больше времени, чем позволяет игровая площадка.

Если мы добавим что-то для пары производитель-потребитель, мы сможем ограничить количество времени, которое у них есть, чтобы монополизировать планировщик. Например, добавив time.Sleep(time.Microsecond)потребителю заставит игровую площадку печатать 1000 значений. Это также показывает, насколько "точным" является смоделированное время на игровой площадке, поскольку это было бы невозможно с обычным оборудованием, которое требует ненулевого количества времени для обработки каждого сообщения.

Хотя это и интересный случай, это мало влияет на реальные программы.

Несколько заметок, вы можете range по каналу, чтобы получать все значения, вы всегда должны defer wg.Done в начале горутины, когда это возможно, вы можете отправлять значения в select case который позволяет вам фактически отменить цикл for-select, когда отправка не готова, и если вы хотите, чтобы сообщение "выход потребителя", вам нужно отправить WaitGroup для потребителя.

https://play.golang.org/p/WyPmpY9pFl7

func main() {
    ch := make(chan int)
    quit := make(chan bool)
    var wg sync.WaitGroup
    wg.Add(2)
    go produce(ch, quit, &wg)
    go consume(ch, &wg)
    time.Sleep(50 * time.Microsecond)
    fmt.Println("CLOSE")
    close(quit)
    wg.Wait()
}

func produce(ch chan int, quit chan bool, wg *sync.WaitGroup) {
    defer wg.Done()

    for i := 0; ; i++ {
        select {
        case <-quit:
            close(ch)
            fmt.Println("exit")
            return
        case ch <- i:
            fmt.Println("Producer sends", i)
        }
    }
}

func consume(ch chan int, wg *sync.WaitGroup) {
    defer wg.Done()
    
    for i := range ch {
        fmt.Println("Consumer receives", i)
        time.Sleep(time.Microsecond)
    }
    
    fmt.Println("exit consumer")
    return
}