在不使用 wg.Wait() 的情况下检查是否所有 goroutine 都已完成

Question

假设我有一个函数 IsAPrimaryColour()，它通过调用其他三个函数 IsRed()、 来工作IsGreen() 和 IsBlue()。由于这三个功能彼此完全独立，因此它们可以运行并发。 return 条件是：

如果三个函数中的任何一个return为真，IsAPrimaryColour() 也应该 return 为真。无需等待对方功能来完成。即：IsPrimaryColour() 是 true 如果 IsRed() 是 true 或 IsGreen() 是 true 或 IsBlue() 是真
如果所有函数 return 为假，IsAPrimaryColour() 也应该 return 错误的。即：IsPrimaryColour() 是 false 如果 IsRed() 是 false AND IsGreen() 是 false AND IsBlue() 是假
如果三个函数中的任何一个return出错，IsAPrimaryColour() 也应该return这个错误。无需等待对方函数以完成或收集任何其他错误。

我正在努力解决的问题是，如果任何其他三个函数 return 为真，如何退出该函数，如果它们都 return 为假，还要等待所有三个函数完成。如果我使用 sync.WaitGroup 对象，我将需要等待所有 3 个 go 例程完成才能从调用函数 return。

因此，我使用循环计数器来跟踪我在频道上收到消息的次数，并在收到所有 3 条消息后退出程序。

https://play.golang.org/p/kNfqWVq4Wix

package main

import (
    "errors"
    "fmt"
    "time"
)

func main() {
    x := "something"
    result, err := IsAPrimaryColour(x)

    if err != nil {
        fmt.Printf("Error: %v\n", err)
    } else {
        fmt.Printf("Result: %v\n", result)
    }
}

func IsAPrimaryColour(value interface{}) (bool, error) {
    found := make(chan bool, 3)
    errors := make(chan error, 3)
    defer close(found)
    defer close(errors)
    var nsec int64 = time.Now().UnixNano()

    //call the first function, return the result on the 'found' channel and any errors on the 'errors' channel
    go func() {
        result, err := IsRed(value)
        if err != nil {
            errors <- err
        } else {
            found <- result
        }
        fmt.Printf("IsRed done in %f nanoseconds \n", float64(time.Now().UnixNano()-nsec))
    }()

    //call the second function, return the result on the 'found' channel and any errors on the 'errors' channel
    go func() {
        result, err := IsGreen(value)
        if err != nil {
            errors <- err
        } else {
            found <- result
        }
        fmt.Printf("IsGreen done in %f nanoseconds \n", float64(time.Now().UnixNano()-nsec))
    }()

    //call the third function, return the result on the 'found' channel and any errors on the 'errors' channel
    go func() {
        result, err := IsBlue(value)
        if err != nil {
            errors <- err
        } else {
            found <- result
        }
        fmt.Printf("IsBlue done in %f nanoseconds \n", float64(time.Now().UnixNano()-nsec))
    }()

    //loop counter which will be incremented every time we read a value from the 'found' channel
    var counter int

    for {
        select {
        case result := <-found:
            counter++
            fmt.Printf("received a value on the results channel after %f nanoseconds. Value of counter is %d\n", float64(time.Now().UnixNano()-nsec), counter)
            if result {
                fmt.Printf("some goroutine returned true\n")
                return true, nil
            }
        case err := <-errors:
            if err != nil {
                fmt.Printf("some goroutine returned an error\n")
                return false, err
            }
        default:
        }

        //check if we have received all 3 messages on the 'found' channel. If so, all 3 functions must have returned false and we can thus return false also
        if counter == 3 {
            fmt.Printf("all goroutines have finished and none of them returned true\n")
            return false, nil
        }
    }
}

func IsRed(value interface{}) (bool, error) {
    return false, nil
}

func IsGreen(value interface{}) (bool, error) {
    time.Sleep(time.Millisecond * 100) //change this to a value greater than 200 to make this function take longer than IsBlue()
    return true, nil
}

func IsBlue(value interface{}) (bool, error) {
    time.Sleep(time.Millisecond * 200)
    return false, errors.New("something went wrong")
}

虽然这工作得很好，但我想知道我是否没有忽略一些语言功能来以更好的方式做到这一点？

Answer 1

你不必阻塞 Wait 上的主 goroutine，你可以阻塞其他东西，例如：

doneCh := make(chan struct{}{})

go func() {
    wg.Wait()
    close(doneCh)
}()

然后您可以在 select 中等待 doneCh，看看是否所有例程都已完成。

Answer 2

一些评论，

您不需要关闭通道，您事先知道要读取的信号的预期数量。这足以满足退出条件。
您不需要重复手动函数调用，使用切片。
因为你使用了一个切片，你甚至不需要一个计数器，或者一个静态值 3，只要看看你的 func 切片的长度就可以了。
切换到默认情况下是没用的。只需阻止您正在等待的输入。

所以一旦你摆脱了所有的脂肪，代码看起来像


func IsAPrimaryColour(value interface{}) (bool, error) {
    fns := []func(interface{}) (bool, error){IsRed, IsGreen, IsBlue}
    found := make(chan bool, len(fns))
    errors := make(chan error, len(fns))

    for i := 0; i < len(fns); i++ {
        fn := fns[i]
        go func() {
            result, err := fn(value)
            if err != nil {
                errors <- err
                return
            }
            found <- result
        }()
    }

    for i := 0; i < len(fns); i++ {
        select {
        case result := <-found:
            if result {
                return true, nil
            }
        case err := <-errors:
            if err != nil {
                return false, err
            }
        }
    }
    return false, nil
}

您不需要观察每次异步调用的时间，只需观察整个调用者花费的时间 return。

func main() {
    now := time.Now()
    x := "something"
    result, err := IsAPrimaryColour(x)

    if err != nil {
        fmt.Printf("Error: %v\n", err)
    } else {
        fmt.Printf("Result: %v\n", result)
    }
    fmt.Println("it took", time.Since(now))
}

https://play.golang.org/p/bARHS6c6m1c

Answer 3

处理多个并发函数调用并取消任何未完成的条件的惯用方法是使用上下文值。像这样：

func operation1(ctx context.Context) bool { ... }
func operation2(ctx context.Context) bool { ... }
func operation3(ctx context.Context) bool { ... }

func atLeastOneSuccess() bool {
    ctx, cancel := context.WithCancel(context.Background()
    defer cancel() // Ensure any functions still running get the signal to stop
    results := make(chan bool, 3) // A channel to send results
    go func() {
        results <- operation1(ctx)
    }()
    go func() {
        results <- operation2(ctx)
    }()
    go func() {
        results <- operation3(ctx)
    }()
    for i := 0; i < 3; i++ {
        result := <-results
        if result {
            // One of the operations returned success, so we'll return that
            // and let the deferred call to cancel() tell any outstanding
            // functions to abort.
            return true
        }
    }
    // We've looped through all return values, and they were all false
    return false
}

当然，这假设每个 operationN 函数实际上都支持已取消的上下文。讨论如何做到这一点。

Answer 4

errgroup.WithContext 可以帮助简化这里的并发。

您想在发生错误或找到结果时停止所有 goroutine。如果您可以将“找到结果”表示为可区分的错误（沿 io.EOF 的行），则可以使用 errgroup 的内置“在第一个错误时取消”行为来关闭全组：

func IsAPrimaryColour(ctx context.Context, value interface{}) (bool, error) {
    var nsec int64 = time.Now().UnixNano()

    errFound := errors.New("result found")
    g, ctx := errgroup.WithContext(ctx)

    g.Go(func() error {
        result, err := IsRed(ctx, value)
        if result {
            err = errFound
        }
        fmt.Printf("IsRed done in %f nanoseconds \n", float64(time.Now().UnixNano()-nsec))
        return err
    })

    …

    err := g.Wait()
    if err == errFound {
        fmt.Printf("some goroutine returned errFound\n")
        return true, nil
    }
    if err != nil {
        fmt.Printf("some goroutine returned an error\n")
        return false, err
    }
    fmt.Printf("all goroutines have finished and none of them returned true\n")
    return false, nil
}

(https://play.golang.org/p/MVeeBpDv4Mn)

在不使用 wg.Wait() 的情况下检查是否所有 goroutine 都已完成

Check if all goroutines have finished without using wg.Wait()

go

concurrency

waitgroup