идти рутина для диапазона по каналам

Question

Я давно работаю в Голанге.Но все же я сталкиваюсь с этой проблемой, хотя знаю решение своей проблемы.Но так и не выяснил, почему это происходит.

Например, если у меня есть конвейерная ситуация для входящих и исходящих каналов, как показано ниже:

package main

import (
    "fmt"
)

func main() {
    for n := range sq(sq(gen(3, 4))) {
        fmt.Println(n)
    }
    fmt.Println("Process completed")
}

func gen(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        for _, n := range nums {
            out <- n
        }
        close(out)
    }()
    return out
}

func sq(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        for n := range in {
            out <- n * n
        }
        close(out)
    }()
    return out
}

Это не дает мне тупиковую ситуацию.Но если я удалю подпрограмму go из исходящего кода, как показано ниже:

func sq(in <-chan int) <-chan int {
    out := make(chan int)
    for n := range in {
        out <- n * n
    }
    close(out)
    return out
}

, я получу ошибку взаимоблокировки.Почему так, что циклическое переключение каналов с использованием диапазона без go приводит к тупику.

Answer 1

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

func sq(in <-chan int) <-chan int {
    // Nobody else has this channel yet...
    out := make(chan int)
    for n := range in {
        // ...but this line will block until somebody reads the value...
        out <- n * n
    }
    close(out)
    // ...and nobody else can possibly read it until after this return.
    return out
}

Если вы оберните цикл в процедуру, тогда и цикл, и sq функция может продолжаться;даже если цикл блокируется, оператор return out по-прежнему может выполняться, и в конечном итоге вы сможете подключить считыватель к каналу.

(Нет ничего плохого в том, чтобы зацикливаться на каналах за пределами процедур;main функция делает это безобидно и правильно.)

Answer 2

Причина тупика заключается в том, что основной ожидает возврата и завершения sq, но sq ожидает, пока кто-то прочитает чан, и он может продолжить.

Я упростил ваш код, удалив слой вызова sq, и разделил одно предложение на 2:

func main() {
    result := sq(gen(3, 4)) // <-- block here, because sq doesn't return
    for n := range result { 
        fmt.Println(n)
    }
    fmt.Println("Process completed")
}

func gen(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        for _, n := range nums {
            out <- n
        }
        close(out)
    }()
    return out
}

func sq(in <-chan int) <-chan int {
    out := make(chan int)
    for n := range in {
        out <- n * n   // <-- block here, because no one is reading from the chan
    }
    close(out)
    return out
}

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

func main() {
    result := sq(gen(3, 4)) // will not blcok here, because the sq just start a goroutine and return
    for n := range result {
        fmt.Println(n)
    }
    fmt.Println("Process completed")
}

func gen(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        for _, n := range nums {
            out <- n
        }
        close(out)
    }()
    return out
}

func sq(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        for n := range in {
            out <- n * n // will not block here, because main will continue and read the out chan
        }
        close(out)
    }()
    return out
}

Answer 3

Эта ситуация, вызванная тем, что выходной канал функции sq не буферизован.Таким образом, sq ожидает, пока следующая функция не прочитает данные с выхода, но если sq не является асинхронным, это не произойдет ( Playground link ):

package main

import (
    "fmt"
    "sync"
)

var wg sync.WaitGroup

func main() {
    numsCh := gen(3, 4)
    sqCh := sq(numsCh) // if there is no sq in body - we are locked here until input channel will be closed
    result := sq(sqCh) // but if output channel is not buffered, so `sq` is locked, until next function will read from output channel

    for n := range result {
        fmt.Println(n)
    }
    fmt.Println("Process completed")
}

func gen(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        for _, n := range nums {
            out <- n
        }
        close(out)
    }()
    return out
}

func sq(in <-chan int) <-chan int {
    out := make(chan int, 100)
    for n := range in {
        out <- n * n
    }
    close(out)
    return out
}