Wie kann man warten, bis alle Goroutinen fertig sind, ohne Zeit zu verlieren?

108

Dieser Code wählt alle XML-Dateien im selben Ordner aus, da die aufgerufene ausführbare Datei die Verarbeitung asynchron auf jedes Ergebnis der Rückrufmethode anwendet (im folgenden Beispiel wird nur der Name der Datei ausgedruckt).

Wie vermeide ich die Verwendung der Schlafmethode, um zu verhindern, dass die Hauptmethode beendet wird? Ich habe Probleme, meinen Kopf um Kanäle zu wickeln (ich nehme an, das ist nötig, um die Ergebnisse zu synchronisieren), daher ist jede Hilfe willkommen!

package main

import (
    "fmt"
    "io/ioutil"
    "path"
    "path/filepath"
    "os"
    "runtime"
    "time"
)

func eachFile(extension string, callback func(file string)) {
    exeDir := filepath.Dir(os.Args[0])
    files, _ := ioutil.ReadDir(exeDir)
    for _, f := range files {
            fileName := f.Name()
            if extension == path.Ext(fileName) {
                go callback(fileName)
            }
    }
}


func main() {
    maxProcs := runtime.NumCPU()
    runtime.GOMAXPROCS(maxProcs)

    eachFile(".xml", func(fileName string) {
                // Custom logic goes in here
                fmt.Println(fileName)
            })

    // This is what i want to get rid of
    time.Sleep(100 * time.Millisecond)
}
Dante
quelle

Antworten:

173

Sie können sync.WaitGroup verwenden . Zitieren des verknüpften Beispiels:

package main

import (
        "net/http"
        "sync"
)

func main() {
        var wg sync.WaitGroup
        var urls = []string{
                "http://www.golang.org/",
                "http://www.google.com/",
                "http://www.somestupidname.com/",
        }
        for _, url := range urls {
                // Increment the WaitGroup counter.
                wg.Add(1)
                // Launch a goroutine to fetch the URL.
                go func(url string) {
                        // Decrement the counter when the goroutine completes.
                        defer wg.Done()
                        // Fetch the URL.
                        http.Get(url)
                }(url)
        }
        // Wait for all HTTP fetches to complete.
        wg.Wait()
}
zzzz
quelle
11
Gibt es einen Grund, warum Sie wg.Add (1) außerhalb der Go-Routine ausführen müssen? Können wir es kurz vor dem Aufschieben von wg.Done () tun?
saß am
18
saß, ja, es gibt einen Grund, es ist in sync.WaitGroup.Add docs beschrieben: Note that calls with positive delta must happen before the call to Wait, or else Wait may wait for too small a group. Typically this means the calls to Add should execute before the statement creating the goroutine or other event to be waited for. See the WaitGroup example.
Wobmene
15
Das Anpassen dieses Codes verursachte eine lange Debugging-Sitzung, da meine Goroutine eine benannte Funktion war und die Übergabe der WaitGroup als Wert sie kopiert und wg.Done () unwirksam macht. Während dies durch Übergeben eines Zeigers & wg behoben werden könnte, besteht eine bessere Möglichkeit, solche Fehler zu vermeiden, darin, die Variable WaitGroup zunächst als Zeiger zu deklarieren: wg := new(sync.WaitGroup)anstelle von var wg sync.WaitGroup.
Robert Jack Will
Ich denke, es ist gültig, wg.Add(len(urls))direkt über der Zeile zu schreiben. for _, url := range urlsIch glaube, es ist besser, wenn Sie das Hinzufügen nur einmal verwenden.
Victor
@ RobertJackWill: Gute Nachricht! Übrigens wird dies in den Dokumenten behandelt : "Eine WaitGroup darf nach der ersten Verwendung nicht kopiert werden. Schade, dass Go dies nicht erzwingen kann . Tatsächlich go veterkennt es diesen Fall jedoch und warnt mit" func übergibt die Sperre nach Wert : sync.WaitGroup enthält sync.noCopy ".
Brent Bradburn
56

WaitGroups sind definitiv der kanonische Weg, dies zu tun. Der Vollständigkeit halber ist hier die Lösung, die üblicherweise vor der Einführung von WaitGroups verwendet wurde. Die Grundidee besteht darin, einen Kanal zu verwenden, um "Ich bin fertig" zu sagen, und die Hauptgoroutine warten zu lassen, bis jede gespawnte Routine ihren Abschluss gemeldet hat.

func main() {
    c := make(chan struct{}) // We don't need any data to be passed, so use an empty struct
    for i := 0; i < 100; i++ {
        go func() {
            doSomething()
            c <- struct{}{} // signal that the routine has completed
        }()
    }

    // Since we spawned 100 routines, receive 100 messages.
    for i := 0; i < 100; i++ {
        <- c
    }
}
joshlf
quelle
9
Schön, eine Lösung mit einfachen Kanälen zu sehen. Ein zusätzlicher Bonus: Wenn doSomething()ein Ergebnis zurückgegeben wird, können Sie es auf den Kanal setzen und die Ergebnisse in der zweiten for-Schleife sammeln und verarbeiten (sobald sie fertig sind)
andras
4
Es funktioniert nur, wenn Sie bereits wissen, wie viele Gorutinen Sie starten möchten. Was ist, wenn Sie eine Art HTML-Crawler schreiben und Gorutinen für jeden Link auf der Seite rekursiv starten?
Shinydev
Sie müssen dies irgendwie irgendwie verfolgen. Mit WaitGroups ist es ein bisschen einfacher, weil Sie jedes Mal, wenn Sie eine neue Goroutine erzeugen, dies zuerst tun können wg.Add(1)und somit den Überblick behalten. Mit Kanälen wäre es etwas schwieriger.
Joshlf
c wird blockieren, da alle Go-Routinen versuchen, darauf zuzugreifen, und es ist ungepuffert
Edwin Ikechukwu Okonkwo
Wenn mit "blockieren" gemeint ist, dass das Programm blockiert, ist das nicht wahr. Sie können versuchen, es selbst auszuführen. Der Grund ist, dass die einzigen Goroutinen, in die geschrieben wird, csich von der Hauptgoroutine unterscheiden, aus der gelesen wird c. Somit ist die Hauptgoroutine immer verfügbar, um einen Wert aus dem Kanal zu lesen. Dies geschieht, wenn eine der Goroutinen verfügbar ist, um einen Wert in den Kanal zu schreiben. Sie haben Recht, wenn dieser Code keine Goroutinen hervorbringt, sondern alles in einer einzigen Goroutine ausführt, würde er zum Stillstand kommen.
Joshlf
8

sync.WaitGroup kann Ihnen hier helfen.

package main

import (
    "fmt"
    "sync"
    "time"
)


func wait(seconds int, wg * sync.WaitGroup) {
    defer wg.Done()

    time.Sleep(time.Duration(seconds) * time.Second)
    fmt.Println("Slept ", seconds, " seconds ..")
}


func main() {
    var wg sync.WaitGroup

    for i := 0; i <= 5; i++ {
        wg.Add(1)   
        go wait(i, &wg)
    }
    wg.Wait()
}
dimmg
quelle
1

Obwohl sync.waitGroup(wg) der kanonische Weg nach vorne ist, müssen Sie zumindest einige Ihrer wg.AddAnrufe vor Ihnen tätigen, wg.Waitdamit alle fertig sind. Dies ist möglicherweise für einfache Dinge wie einen Webcrawler nicht möglich, bei denen Sie die Anzahl der rekursiven Aufrufe im Voraus nicht kennen und es eine Weile dauert, bis die Daten abgerufen sind, die die wg.AddAufrufe steuern. Schließlich müssen Sie die erste Seite laden und analysieren, bevor Sie die Größe des ersten Stapels untergeordneter Seiten kennen.

Ich habe eine Lösung über Kanäle geschrieben und waitGroupin meiner Lösung die Tour of Go - Web-Crawler- Übung vermieden . Jedes Mal, wenn eine oder mehrere Go-Routinen gestartet werden, senden Sie die Nummer an den childrenKanal. Jedes Mal, wenn eine Go-Routine abgeschlossen ist, senden Sie eine 1an den doneKanal. Wenn die Summe der Kinder gleich der Summe der erledigten Kinder ist, sind wir fertig.

Meine einzige verbleibende Sorge ist die fest codierte Größe des resultsKanals, aber das ist eine (aktuelle) Go-Einschränkung.


// recursionController is a data structure with three channels to control our Crawl recursion.
// Tried to use sync.waitGroup in a previous version, but I was unhappy with the mandatory sleep.
// The idea is to have three channels, counting the outstanding calls (children), completed calls 
// (done) and results (results).  Once outstanding calls == completed calls we are done (if you are
// sufficiently careful to signal any new children before closing your current one, as you may be the last one).
//
type recursionController struct {
    results  chan string
    children chan int
    done     chan int
}

// instead of instantiating one instance, as we did above, use a more idiomatic Go solution
func NewRecursionController() recursionController {
    // we buffer results to 1000, so we cannot crawl more pages than that.  
    return recursionController{make(chan string, 1000), make(chan int), make(chan int)}
}

// recursionController.Add: convenience function to add children to controller (similar to waitGroup)
func (rc recursionController) Add(children int) {
    rc.children <- children
}

// recursionController.Done: convenience function to remove a child from controller (similar to waitGroup)
func (rc recursionController) Done() {
    rc.done <- 1
}

// recursionController.Wait will wait until all children are done
func (rc recursionController) Wait() {
    fmt.Println("Controller waiting...")
    var children, done int
    for {
        select {
        case childrenDelta := <-rc.children:
            children += childrenDelta
            // fmt.Printf("children found %v total %v\n", childrenDelta, children)
        case <-rc.done:
            done += 1
            // fmt.Println("done found", done)
        default:
            if done > 0 && children == done {
                fmt.Printf("Controller exiting, done = %v, children =  %v\n", done, children)
                close(rc.results)
                return
            }
        }
    }
}

Vollständiger Quellcode für die Lösung

Dirkjot
quelle
1

Hier ist eine Lösung, die WaitGroup verwendet.

Definieren Sie zunächst zwei Dienstprogrammmethoden:

package util

import (
    "sync"
)

var allNodesWaitGroup sync.WaitGroup

func GoNode(f func()) {
    allNodesWaitGroup.Add(1)
    go func() {
        defer allNodesWaitGroup.Done()
        f()
    }()
}

func WaitForAllNodes() {
    allNodesWaitGroup.Wait()
}

Ersetzen Sie dann den Aufruf von callback:

go callback(fileName)

Mit einem Aufruf Ihrer Utility-Funktion:

util.GoNode(func() { callback(fileName) })

Im letzten Schritt fügen Sie diese Zeile am Ende Ihres mainstatt Ihres hinzu sleep. Dadurch wird sichergestellt, dass der Hauptthread auf den Abschluss aller Routinen wartet, bevor das Programm gestoppt werden kann.

func main() {
  // ...
  util.WaitForAllNodes()
}
Gamliela
quelle