並列実行と順次実行のクロージャの不一致

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

をその結果、あなたはいつもの代わりに各1の最後のfを開始しました。

Answer 2

私はあなたの問題があなたのRunInParallel機能にあると信じています。

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は、遊び場でのライブ例です。