Golank goroutine nie działa z kanałem wewnątrz

Question

Próbuję zaimplementować program liczący słowa, ale z pierwszym krokiem mam pewien problem.

Oto mój kod:

package main 

import (
    "fmt" 
    "os" 
    "bufio" 
    "sync" 
) 

// Load data into channel 
func laodData(arr []string,channel chan string,wg sync.WaitGroup) { 
    for _,path := range arr { 
     file,err := os.Open(path) 
     fmt.Println("begin to laodData ", path) 
     if err != nil { 
      fmt.Println(err) 
      os.Exit(-1) 
     } 
     defer file.Close() 
     reader := bufio.NewReaderSize(file, 32*10*1024) 
     i := 0 
     for { 
      line,err := reader.ReadString('\n') 
      channel <- line 
      if err != nil { 
       break 
      } 
      i++ 
      if i%200 == 0 { 
       fmt.Println(i," lines parsed") 
      } 
     } 
     fmt.Println("finish laodData ", path) 
    } 
    wg.Done() 
} 

// dispatch data lines into different mappers 
func dispatcher(channel chan string,wg sync.WaitGroup){ 
    fmt.Println("pull data 11") 
    line,ok := <- channel 
    fmt.Println(ok) 
    for ok { 
     fmt.Println(line) 
     line,ok = <- channel 
    } 
    fmt.Println("pull data 22") 
    wg.Done() 
} 

func main() { 
    path := os.Args 
    if len(path) < 2 { 
     fmt.Println("Need Input Files") 
     os.Exit(0) 
    } 
    var wg sync.WaitGroup 
    wg.Add(2) 

    channel := make(chan string) 
    defer close(channel) 

    fmt.Println("before dispatcher") 
    go laodData(path[1:],channel,wg) 
    go dispatcher(channel,wg) 
    wg.Wait() 

    fmt.Println("after dispatcher") 
} 

A oto moje wyjście:

... 

finish laodData result.txt 

throw: all goroutines are asleep - deadlock! 

goroutine 1 [semacquire]: 
sync.runtime_Semacquire(0x42154100, 0x42154100) 
    /usr/local/go/src/pkg/runtime/zsema_amd64.c:146 +0x25 
sync.(*WaitGroup).Wait(0x4213b440, 0x0) 
    /usr/local/go/src/pkg/sync/waitgroup.go:79 +0xf2 
main.main() 
    /Users/kuankuan/go/src/mreasy/main.go:66 +0x238 

goroutine 2 [syscall]: 
created by runtime.main 
    /usr/local/go/src/pkg/runtime/proc.c:221 

goroutine 4 [chan receive]: 
main.dispatcher(0x42115a50, 0x0, 0x2, 0x0) 
    /Users/kuankuan/go/src/mreasy/main.go:45 +0x223 
created by main.main 
    /Users/kuankuan/go/src/mreasy/main.go:65 +0x228 
exit status 2 

Dzięki!

Answer 1

Program kończy się, gdy główna goroutina kończy działanie, dzięki czemu dispatcher() nie ma czasu na nic. Musisz zablokować się w main() aż do ukończenia dispatcher(). Kanał można wykorzystać do tego:

package main 

import (
    "fmt" 
    "os" 
    "bufio" 
) 

var done = make(chan bool)    // create channel 

// Load files and send them into a channel for mappers reading. 
func dispatcher(arr []string,channel chan string) { 
    for _,path := range arr { 
     file,err := os.Open(path) 
     fmt.Println("begin to dispatch ", path) 
     if err != nil { 
      fmt.Println(err) 
      os.Exit(-1) 
     } 
     defer file.Close() 
     reader := bufio.NewReaderSize(file, 32*10*1024) 
     i := 0 
     for { 
      line,_ := reader.ReadString('\n') 
      channel <- line 
      i++ 
      if i%200 == 0 { 
       fmt.Println(i," lines parsed") 
      } 
     } 
     fmt.Println("finish dispatch ", path) 
    } 
    done <- true     // notify main() of completion 
} 

func main() { 
    path := os.Args 
    if len(path) < 2 { 
     fmt.Println("Need Input Files") 
     os.Exit(0) 
    } 
    channel := make(chan string) 
    fmt.Println("before dispatcher") 
    go dispatcher(path[1:],channel) 
    <-done     // wait for dispatcher() 
    fmt.Println("after dispatcher") 
} 

Answer 2

I modified your example do pracy na placu zabaw tam, gdzie nie ma pliku I/O; zamiast tego wysyła losowe numery na kanale.

@Wytłumaczenie Davida Deryagina i jego sugestia użycia "zrobionego" kanału jest poprawna. Powodem jest to, że twój goroutine wysyła na kanał, ale nikt nie czyta z niego, więc program utknął w tym momencie. W powyższym linku dodałem konsumenta goroutine. Następnie program działa równolegle zgodnie z przeznaczeniem.

Należy pamiętać, że aby poczekać na kilka goroutines, jest to bardziej przejrzyste i łatwiejsze w użyciu sync.WaitGroup.

Answer 3

W pierwotnym pytaniu należy poprawić dwie kwestie.

1. Musisz zamknąć kanał po zakończeniu wysyłania wszystkich danych. W func laodData, użyj close (channel) post wysyłając wszystkie dane.
2. Przekaż sync.Waitgroup jako odniesienie. Wysyłasz wg jako wartość w argumencie do następujących funkcji ... laodData i funkcji dyspozytorskich.

Naprawienie tych dwóch problemów rozwiąże problem z impasem. Przyczyny blokady w kodzie są następujące:

• Pozostawienie zamkniętego kanału wysyłającego spowoduje, że kanał niższego rzędu będzie czekał przez dłuższy czas.
• wysyłając argument o wartości sync.Waitgroup jako wartość. Powinien zostać wysłany jako odnośnik, w przeciwnym razie utworzy nową kopię obiektu, który wysyłasz.