Почему получается deadlock? Что-то не так с получателем?

Question

[Everything_is_not_so_bad]

Разбираюсь с работой Go каналов. Написал тестовую программу, суть которой в последовательной обработке данных, получаемых в одном канале из другого.
Получаю ошибку

fatal error: all goroutines are asleep - deadlock!

код

package main

import (
	"fmt"
	"time"
)

var numChan chan int
var labelChan <-chan string

func main() {
	input := []int{3, 1, 8, 7, 5, 9, 2}
	numChan = make(chan int)

	go func() {
		labelsReceiver(labelChan)
	}()
	time.Sleep(time.Millisecond) // Так попытался убедиться, что получатель стартует первым. Не знаю как правильно сделать.

	go func() {
		labelChan = labelsEmitter(numChan)
	}()

	for _, num := range input {
		fmt.Println(num)
		numChan <- num
	}
}

func labelsEmitter(numbers chan int) <-chan string {
	fmt.Println("go labelsEmitter")
	out := make(chan string, 3)
	defer close(out)
	for num := range numbers {
		time.Sleep(time.Millisecond * 200)
		str := fmt.Sprintf("Number %d", num)
		out <- str
	}
	return out
}

func labelsReceiver(labelChanRcv <-chan string) {
	fmt.Println("go labelsReceiver")
	for label := range labelChanRcv {
		fmt.Println(label)
	}
}

Но если увеличить размер буферизованного канала в сткроке
out := make(chan string, 3)
, скажем, до 10, то программа хоть и не выводит label, но и не валится с ошибкой выше.

Answer 1

[Алексей Маслов]

Попробуйте так:

package main

import (
	"fmt"
	"time"
)

var numChan chan int
var outChan chan string

func main() {
	input := []int{3, 1, 8, 7, 5, 9, 2}
	numChan = make(chan int)
	outChan = make(chan string)
	go func() {
		labelsReceiver(outChan)
	}()
	time.Sleep(time.Millisecond) // Так попытался убедиться, что получатель стартует первым. Не знаю как правильно сделать.

	go func() {
		labelsEmitter(numChan, outChan)
	}()

	for _, num := range input {
		fmt.Println(num)
		numChan <- num
	}
	close(numChan)
	close(outChan)
}

func labelsEmitter(numbers chan int, outChan chan string) {
	fmt.Println("go labelsEmitter")
	//out := make(chan string, 3)
	//defer close(out)
	for num := range numbers {
		time.Sleep(time.Millisecond * 200)
		str := fmt.Sprintf("Number %d", num)
		outChan <- str
	}
	return
}

func labelsReceiver(labelChanRcv <-chan string) {
	fmt.Println("go labelsReceiver")
	for label := range labelChanRcv {
		fmt.Println("label", label)
	}
}

Comments

[Алексей Маслов]

И вместо time.Sleep(time.Millisecond) лучше использовать waitgroup.
Как то так:

var wg = sync.WaitGroup{}

	wg.Add(1)
	go labelsReceiver(outChan, &wg)
	//time.Sleep(time.Millisecond) // Так попытался убедиться, что получатель стартует первым. Не знаю как правильно сделать.
	wg.Wait() // ждем пока не будет wg.done()

func labelsReceiver(labelChanRcv <-chan string, wg *sync.WaitGroup) {
	wg.Done()
	fmt.Println("go labelsReceiver")
	for label := range labelChanRcv {
		fmt.Println("label", label)
	}
}

[Everything_is_not_so_bad]

Не все числа выводит в label. Так и не понял в чём проблема в моём коде.

[Алексей Маслов]

Роман Мирр, перед закрытием каналов задержку поставьте: time.Sleep(time.Second * 5)

Данные он записал, но т.к. labelsEmitter стоит задержка - не успевал все вычитать, а программа к тому времени завершалась.

Проблема в том, что вы возвращали канал labelsEmitter
А до туда программа не доходила, т.к. ждала, пока из канала кто то вычитает, на этом и случался deadlock

func labelsEmitter(numbers chan int) <-chan string {
  fmt.Println("go labelsEmitter")
  out := make(chan string, 3)
  defer close(out)
  for num := range numbers {
    time.Sleep(time.Millisecond * 200)
    str := fmt.Sprintf("Number %d", num)
    out <- str
  }
  return out
}

Из вашего кода ни когда не дойдет до return out, пока емкость канала out будет меньше передаваемого numbers.

В вашем случае он записывает три значения и ждет, пока кто то вычитает из этого канала значения. А читать не кому, т.к. канал отдается только в return.

[Everything_is_not_so_bad]

Алексей Маслов,
сделал так:

wg.Add(1)
	go func() {
		defer wg.Done()
		labelsReceiver(outChan)
	}()
	wg.Wait()

и получаю снова deadlock.
Я так понял, что строка time.Sleep(time.Millisecond) теперь не нужна.

[Алексей Маслов]

Сделайте, как я написал (waitgorup передавать параметром)
В вашем случае labelsReceiver не завершится до тех, пор, пока outChan не закроется, соответсвенно wg.Done не выполнится и будет вечно висеть wg.Wait()

[Everything_is_not_so_bad]

В общем, мой окончательный вариант, с минорными изменениями, таков:

код

package main

import (
	"fmt"
	"time"
)

var numChan chan int
var labelChan chan string

func main() {
	input := []int{3, 1, 8, 0, 7, 5, 9, 2}
	numChan = make(chan int)
	labelChan = make(chan string)

	go func() {
		labelsReceiver(labelChan)
	}()
	// убрал отсюда time.Sleep(time.Millisecond)

	go func() {
		labelsEmitter(numChan, labelChan)
	}()

	for _, num := range input {
		fmt.Println(num)
		numChan <- num
	}

	time.Sleep(time.Second * 5)
	close(numChan)
	close(labelChan)
}

func labelsEmitter(numbers chan int, outChan chan string) {
	fmt.Println("go labelsEmitter")
	for num := range numbers {
		time.Sleep(time.Millisecond * 250)
		str := fmt.Sprintf("Number %d", num)
		outChan <- str
	}
}

func labelsReceiver(labelChanRcv <-chan string) {
	fmt.Println("go labelsReceiver")
	for label := range labelChanRcv {
		fmt.Println(label)
	}
}

Answer 2

[Илья Зябиров]

Могу ошибаться, но по-моему, дело в том, что вы переопределяете labelChan. Сначала вы в labelsReceiver передаете переменную, которая была объявлена над main() (она передается по значению - то есть в labelsReceiver вы взаимодействуете с копией переменной labelChan), а затем вы из метода labelsEmitter возвращаете другой канал, в который вы пишете, но из которого не читаете. А при увеличении буфера до 10 ошибки нет потому, что чисел в input'е меньше 10-ти => они просто записываются в буфер, но их все еще никто не читает.

Comments

[Everything_is_not_so_bad]

go func() {
    labelsReceiver(labelChan)
  }()
  time.Sleep(time.Millisecond) // Так попытался убедиться, что получатель стартует первым. Не знаю как правильно сделать.

  go func() {
    labelChan = labelsEmitter(numChan)
  }()

1-я является получателем, 2-я отправителем.
А как правильно передавать канал в параметрах?
Вот так тоже самое:

go func(c <-chan string) {
		labelsReceiver(c)
	}(labelChan)

[Илья Зябиров]

Роман Мирр, ниже вам верно ответили - стоит создавать канал out снаружи и передавать его в labelsEmitter в качестве второго параметра. И тот же канал передавать в labelsReceiver

[Everything_is_not_so_bad]

Спасибо

[Everything_is_not_so_bad]

вы переопределяете labelChan

А где у меня в коде происходит переопределение?

[Илья Зябиров]

go func() {
    labelChan = labelsEmitter(numChan)
}()