Язык GO: фатальная ошибка: все горуты спит - тупик

Код ниже отлично работает с жестко закодированными данными JSON, однако не работает, когда я читаю данные JSON из файла. Я получаю ошибку fatal error: all goroutines are asleep - deadlock при использовании sync.WaitGroup.

РАБОЧИЙ ПРИМЕР С ТЯЖЕЛЫМИ КОДИРОВАННЫМИ ДАННЫМИ JSON:

package main

import (
    "bytes"
    "fmt"
    "os/exec"
    "time"
)

func connect(host string) {
    cmd := exec.Command("ssh", host, "uptime")
    var out bytes.Buffer
    cmd.Stdout = &out
    err := cmd.Run()
    if err != nil {
        fmt.Println(err)
    }
    fmt.Printf("%s: %q\n", host, out.String())
    time.Sleep(time.Second * 2)
    fmt.Printf("%s: DONE\n", host)
}

func listener(c chan string) {
    for {
        host := <-c
        go connect(host)
    }
}

func main() {
    hosts := [2]string{"[email protected]", "[email protected]"}
    var c chan string = make(chan string)
    go listener(c)

    for i := 0; i < len(hosts); i++ {
        c <- hosts[i]
    }
    var input string
    fmt.Scanln(&input)
}

ВЫВОД:

[email protected]:~/go$ go run channel.go
[email protected]: " 09:46:40 up 86 days, 18:16,  0 users,  load average: 5"
[email protected]: " 09:46:40 up 86 days, 17:27,  1 user,  load average: 9"
[email protected]: DONE
[email protected]: DONE

НЕ РАБОТАЕТ - ПРИМЕР С ЧТЕНИЕМ ФАЙЛА ДАННЫХ JSON:

package main

import (
    "bytes"
    "fmt"
    "os/exec"
    "time"
    "encoding/json"
    "os"
    "sync"
)

func connect(host string) {
    cmd := exec.Command("ssh", host, "uptime")
    var out bytes.Buffer
    cmd.Stdout = &out
    err := cmd.Run()
    if err != nil {
        fmt.Println(err)
    }
    fmt.Printf("%s: %q\n", host, out.String())
    time.Sleep(time.Second * 2)
    fmt.Printf("%s: DONE\n", host)
}

func listener(c chan string) {
    for {
        host := <-c
        go connect(host)
    }
}

type Content struct {
    Username string `json:"username"`
    Ip       string `json:"ip"`
}

func main() {
    var wg sync.WaitGroup

    var source []Content
    var hosts []string
    data := json.NewDecoder(os.Stdin)
    data.Decode(&source)

    for _, value := range source {
        hosts = append(hosts, value.Username + "@" + value.Ip)
    }

    var c chan string = make(chan string)
    go listener(c)

    for i := 0; i < len(hosts); i++ {
        wg.Add(1)
        c <- hosts[i]
        defer wg.Done()
    }

    var input string
    fmt.Scanln(&input)

    wg.Wait()
}

OUTPUT

[email protected]:~/go$ go run deploy.go < hosts.txt 
[email protected]: " 09:46:40 up 86 days, 18:16,  0 users,  load average: 5"
[email protected]: " 09:46:40 up 86 days, 17:27,  1 user,  load average: 9"
[email protected] : DONE
[email protected]: DONE
fatal error: all goroutines are asleep - deadlock!

goroutine 1 [semacquire]:
sync.runtime_Semacquire(0xc210000068)
    /usr/lib/go/src/pkg/runtime/sema.goc:199 +0x30
sync.(*WaitGroup).Wait(0xc210047020)
    /usr/lib/go/src/pkg/sync/waitgroup.go:127 +0x14b
main.main()
    /home/user/go/deploy.go:64 +0x45a

goroutine 3 [chan receive]:
main.listener(0xc210038060)
    /home/user/go/deploy.go:28 +0x30
created by main.main
    /home/user/go/deploy.go:53 +0x30b
exit status 2
[email protected]:~/go$

HOSTS.TXT

[
   {
      "username":"user1",
      "ip":"111.79.154.111"
   },
   {
      "username":"user2",
      "ip":"111.79.190.222"
   }
]

Ответ 1

Программа

Go заканчивается, когда заканчивается основная функция.

Из спецификация языка

Выполнение программы начинается с инициализации основного пакета и последующего вызова функции main. Когда возвращается эта функция, программа завершает работу. Он не дожидается завершения других (не главных) goroutines.

Поэтому вам нужно подождать, пока ваши goroutines будут завершены. Общим решением для этого является использование объекта sync.WaitGroup.

Самый простой возможный код для синхронизации goroutine:

package main

import "fmt"
import "sync"

var wg sync.WaitGroup // 1

func routine() {
    defer wg.Done() // 3
    fmt.Println("routine finished")
}

func main() {
    wg.Add(1) // 2
    go routine() // *
    wg.Wait() // 4
    fmt.Println("main finished")
}

И для синхронизации нескольких goroutines

package main

import "fmt"
import "sync"

var wg sync.WaitGroup // 1

func routine(i int) {
    defer wg.Done() // 3
    fmt.Printf("routine %v finished\n", i)
}

func main() {
    for i := 0; i < 10; i++ {
        wg.Add(1) // 2
        go routine(i) // *
    }
    wg.Wait() // 4
    fmt.Println("main finished")
}

Использование WaitGroup в порядке выполнения.

Объявление глобальной переменной. Сделать его глобальным - это самый простой способ сделать его видимым для всех функций и методов.
Увеличение счетчика. Это должно быть сделано в главном goroutine, потому что нет гарантии, что недавно начатый goroutine будет выполняться до 4 из-за модели памяти гарантирует.
Уменьшение счетчика. Это должно быть сделано на выходе из горутины. Используя отложенный вызов, мы убеждаемся, что будет вызываться всякий раз, когда заканчивается функция, независимо от того, как это закончится.
Ожидание того, что счетчик достигнет 0. Это необходимо сделать в главном goroutine, чтобы предотвратить выход программы.

* Фактические параметры оцениваются перед запуском новой gouroutine. Таким образом, необходимо явно оценить их до wg.Add(1), чтобы, возможно, панический код не оставил увеличенный счетчик.

Использование

param := f(x)
wg.Add(1)
go g(param)

вместо

wg.Add(1)
go g(f(x))

Ответ 2

Спасибо за очень приятное и подробное объяснение Grzegorz Żur. Одна вещь, которую я хочу указать на это, как правило, func, который должен быть потоковым, не будет в main(), поэтому у нас будет что-то вроде этого:

package main

import (
    "bufio"
    "fmt"
    "io"
    "io/ioutil"
    "math/rand"
    "os"
    "reflect"
    "regexp"
    "strings"
    "sync"
    "time"
)

var wg sync.WaitGroup    // VERY IMP to declare this globally, other wise one   //would hit "fatal error: all goroutines are asleep - deadlock!"

func doSomething(arg1 arg1Type) {
     // cured cancer
}

func main() {
    r := rand.New(rand.NewSource(time.Now().UnixNano()))
    randTime := r.Intn(10)
    wg.Add(1)    
    go doSomething(randTime)
    wg.Wait()
    fmt.Println("Waiting for all threads to finish")
}

То, что я хочу указать на это, заключается в том, что глобальная дефляция wg очень важна для завершения всех потоков до main()