Как дождаться завершения всех горутин, не используя time.Sleep?

111

Этот код выбирает все файлы xml в той же папке в качестве вызываемого исполняемого файла и асинхронно применяет обработку к каждому результату в методе обратного вызова (в приведенном ниже примере выводится только имя файла).

Как мне избежать использования метода сна, чтобы не допустить выхода из основного метода? У меня проблемы с мыслями о каналах (я предполагаю, что это то, что нужно, чтобы синхронизировать результаты), поэтому любая помощь приветствуется!

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)
}
Данте
источник

Ответы:

175

Вы можете использовать sync.WaitGroup . Цитируя связанный пример:

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
источник
11
Есть ли причина, по которой вам нужно выполнять wg.Add (1) вне программы go? Можем ли мы сделать это внутри непосредственно перед defer wg.Done ()?
сб
19
сел, да, есть причина, это описано в sync.WaitGroup.Add docs: 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
18
Адаптация этого кода вызвала у меня длительный сеанс отладки, потому что моя горутина была именованной функцией и передача WaitGroup в качестве значения скопирует ее и сделает wg.Done () неэффективным. Хотя это можно исправить, передав указатель & wg, лучший способ предотвратить такие ошибки - это в первую очередь объявить переменную WaitGroup как указатель: wg := new(sync.WaitGroup)вместо var wg sync.WaitGroup.
Роберт Джек Уилл
Я полагаю, что допустимо писать wg.Add(len(urls))чуть выше линии for _, url := range urls, я считаю, что лучше, если вы используете Добавить только один раз.
Виктор
@RobertJackWill: Хорошее примечание! Кстати, это описано в документации : «WaitGroup не должна копироваться после первого использования. Жаль, что Go не имеет способа принудительно это сделать . На самом деле, однако, go vetобнаруживает этот случай и предупреждает с помощью« func передает блокировку по значению. : sync.WaitGroup содержит sync.noCopy ".
Брент
57

WaitGroups - определенно канонический способ сделать это. Однако для полноты картины вот решение, которое обычно использовалось до появления WaitGroups. Основная идея состоит в том, чтобы использовать канал, чтобы сказать «Я готов», и заставить основную горутину ждать, пока каждая порожденная процедура не сообщит о своем завершении.

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
    }
}
Джошлф
источник
9
Приятно видеть решение с простыми каналами. Дополнительный бонус: если doSomething()возвращается какой-то результат, вы можете поместить его на канал, и вы можете собирать и обрабатывать результаты во втором цикле for (как только они будут готовы)
andras
5
Это работает только в том случае, если вы уже знаете, какое количество горутинов хотите начать. Что, если вы пишете какой-то html-сканер и запускаете горутины рекурсивно для каждой ссылки на странице?
shinydev
Вам нужно как-то отслеживать это в любом случае. С WaitGroups это немного проще, потому что каждый раз, когда вы создаете новую горутину, вы можете сначала сделать это, wg.Add(1)и, таким образом, он будет отслеживать их. С каналами было бы несколько сложнее.
joshlf
c будет заблокирован, так как все процедуры go будут пытаться получить к нему доступ, и он не буферизован
Эдвин Икечукву Оконкво
Если под словом «блокировать» вы имеете в виду, что программа зайдет в тупик, это неправда. Вы можете попробовать запустить его самостоятельно. Причина в том, что единственные горутины, которые пишут c, отличаются от основной горутины, которая читает из c. Таким образом, основная горутина всегда доступна для чтения значения из канала, что произойдет, когда одна из горутин будет доступна для записи значения в канал. Вы правы, если бы этот код не порождал горутины, а вместо этого запускал бы все в одной горутине, он бы заблокировался.
joshlf
8

sync.WaitGroup может вам здесь помочь.

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()
}
тусклый
источник
1

Хотя sync.waitGroup(wg) - это канонический путь вперед, он требует, чтобы вы выполнили хотя бы некоторые из ваших wg.Addвызовов перед вами, wg.Waitчтобы все завершились. Это может оказаться невозможным для простых вещей, таких как поисковый робот, где вы заранее не знаете количество рекурсивных вызовов и требуется время, чтобы получить данные, которые управляют wg.Addвызовами. В конце концов, вам нужно загрузить и проанализировать первую страницу, прежде чем вы узнаете размер первого пакета дочерних страниц.

Я написал решение, используя каналы, избегая waitGroupв своем решении упражнения Tour of Go - поискового робота . Каждый раз, когда запускается одна или несколько программ, вы отправляете номер в childrenканал. Каждый раз, когда процедура го приближается к завершению, вы отправляете 1на doneканал. Когда сумма детей сравняется с суммой готовых, все готово.

Единственное, что меня беспокоит, - это жестко заданный размер resultsканала, но это (текущее) ограничение Go.


// 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
            }
        }
    }
}

Полный исходный код решения

диркджот
источник
1

Вот решение, в котором используется WaitGroup.

Сначала определите 2 служебных метода:

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()
}

Затем замените вызов callback:

go callback(fileName)

С вызовом вашей служебной функции:

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

Последний шаг: добавьте эту строку в конец вашего main, а не вашего sleep. Это гарантирует, что основной поток ожидает завершения всех подпрограмм, прежде чем программа сможет остановиться.

func main() {
  // ...
  util.WaitForAllNodes()
}
гамлиела
источник