并行和顺序执行之间的闭包不一致

Question

我试图编写一个可以并行或顺序执行函数的通用函数。在测试它时，我发现了一些关于闭包的非常意外的行为。在下面的代码中，我定义了一个不接受参数并返回错误的函数列表。这些函数还在闭包中使用 for 循环变量，但我使用在循环中定义新变量的技巧以避免捕获。

我希望我可以按顺序或同时调用这些函数，但效果相同，但我看到了不同的结果。就好像闭包变量被捕获但仅在并发运行时。

据我所知，这不是捕获循环变量的常见情况。正如我所提到的，我在循环中定义了一个新变量。另外，我没有在循环中运行闭包函数。我在循环中生成一个函数列表，但我在循环之后执行函数。

我正在使用 go 版本 go1.8.3 linux/amd64。

package closure_test

import (
    "sync"
    "testing"
)

// MergeErrors merges multiple channels of errors.
// Based on https://blog.golang.org/pipelines.
func MergeErrors(cs ...<-chan error) <-chan error {
    var wg sync.WaitGroup
    out := make(chan error)

    // Start an output goroutine for each input channel in cs.  output
    // copies values from c to out until c is closed, then calls wg.Done.
    output := func(c <-chan error) {
        for n := range c {
            out <- n
        }
        wg.Done()
    }
    wg.Add(len(cs))
    for _, c := range cs {
        go output(c)
    }

    // Start a goroutine to close out once all the output goroutines are
    // done.  This must start after the wg.Add call.
    go func() {
        wg.Wait()
        close(out)
    }()
    return out
}

// WaitForPipeline waits for results from all error channels.
// It returns early on the first error.
func WaitForPipeline(errs ...<-chan error) error {
    errc := MergeErrors(errs...)
    for err := range errc {
        if err != nil {
            return err
        }
    }
    return nil
}

func RunInParallel(funcs ...func() error) error {
    var errcList [](<-chan error)
    for _, f := range funcs {
        errc := make(chan error, 1)
        errcList = append(errcList, errc)
        go func() {
            err := f()
            if err != nil {
                errc <- err
            }
            close(errc)
        }()
    }
    return WaitForPipeline(errcList...)
}

func RunSequentially(funcs ...func() error) error {
    for _, f := range funcs {
        err := f()
        if err != nil {
            return err
        }
    }
    return nil
}

func validateOutputChannel(t *testing.T, out chan int, n int) {
    m := map[int]bool{}
    for i := 0; i < n; i++ {
        m[<-out] = true
    }
    if len(m) != n {
        t.Errorf("Output channel has %v unique items; wanted %v", len(m), n)
    }
}

// This fails because j is being captured.
func TestClosure1sp(t *testing.T) {
    n := 4
    out := make(chan int, n*2)
    var funcs [](func() error)
    for i := 0; i < n; i++ {
        j := i // define a new variable that has scope only inside the current loop iteration
        t.Logf("outer i=%v, j=%v", i, j)
        f := func() error {
            t.Logf("inner i=%v, j=%v", i, j)
            out <- j
            return nil
        }
        funcs = append(funcs, f)
    }
    t.Logf("Running funcs sequentially")
    if err := RunSequentially(funcs...); err != nil {
        t.Fatal(err)
    }
    validateOutputChannel(t, out, n)
    t.Logf("Running funcs in parallel")
    if err := RunInParallel(funcs...); err != nil {
        t.Fatal(err)
    }
    close(out)
    validateOutputChannel(t, out, n)
}

下面是上面测试函数的输出。

closure_test.go:91: outer i=0, j=0
closure_test.go:91: outer i=1, j=1
closure_test.go:91: outer i=2, j=2
closure_test.go:91: outer i=3, j=3
closure_test.go:99: Running funcs sequentially
closure_test.go:93: inner i=4, j=0
closure_test.go:93: inner i=4, j=1
closure_test.go:93: inner i=4, j=2
closure_test.go:93: inner i=4, j=3
closure_test.go:104: Running funcs in parallel
closure_test.go:93: inner i=4, j=3
closure_test.go:93: inner i=4, j=3
closure_test.go:93: inner i=4, j=3
closure_test.go:93: inner i=4, j=3
closure_test.go:80: Output channel has 1 unique items; wanted 4

有什么想法吗？这是 Go 中的错误吗？

Answer 1

始终使用 -race 运行测试。在您的情况下，您忘记在 RunInParallel 的每次迭代中重新创建 f：

func RunInParallel(funcs ...func() error) error {
    var errcList [](<-chan error)
    for _, f := range funcs {

        f := f // << HERE

        errc := make(chan error, 1)
        errcList = append(errcList, errc)
        go func() {
            err := f()
            if err != nil {
                errc <- err
            }
            close(errc)
        }()
    }
    return WaitForPipeline(errcList...)
}

因此，您总是启动最后一个 f 而不是每个。

Answer 2

我相信你的问题在于你的RunInParallel func。

func RunInParallel(funcs ...func() error) error {
    var errcList [](<-chan error)
    for _, f := range funcs {
        errc := make(chan error, 1)
        errcList = append(errcList, errc)
        go func() {
            // This line probably isn't being reached until your range
            // loop has completed, meaning f is the last func by the time
            // each goroutine starts. If you capture f
            // in another variable inside the range, you won't have this issue.
            err := f()
            if err != nil {
                errc <- err
            }
            close(errc)
        }()
    }
    return WaitForPipeline(errcList...)
}

您也可以将 f 作为参数传递给匿名函数以避免此问题。

for _, f := range funcs {
    errc := make(chan error, 1)
    errcList = append(errcList, errc)
    go func(g func() error) {
        err := g()
        if err != nil {
            errc <- err
        }
        close(errc)
    }(f)
}

Here 是操场上的一个活生生的例子。