惯用的例行程序终止和错误处理

Question

I have a simple concurrency use case in go, and it's driving me nuts I can't figure out an elegant solution. Any help would be appreciated.

I want to write a method fetchAll that queries an unspecified number of resources from remote servers in parallel. If any of the fetches fails, I want to return that first error immediately.

My initial, naive implementation, leaks goroutines:

package main

import (
  "fmt"
  "math/rand"
  "sync"
  "time"
)

func fetchAll() error {
  wg := sync.WaitGroup{}
  errs := make(chan error)
  leaks := make(map[int]struct{})
  defer fmt.Println("these goroutines leaked:", leaks)

  // run all the http requests in parallel
  for i := 0; i < 4; i++ {
    leaks[i] = struct{}{}
    wg.Add(1)
    go func(i int) {
      defer wg.Done()
      defer delete(leaks, i)

      // pretend this does an http request and returns an error
      time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond)
      errs <- fmt.Errorf("goroutine %d's error returned", i)
    }(i)
  }

  // wait until all the fetches are done and close the error
  // channel so the loop below terminates
  go func() {
    wg.Wait()
    close(errs)
  }()

  // return the first error
  for err := range errs {
    if err != nil {
      return err
    }
  }

  return nil
}

func main() {
  fmt.Println(fetchAll())
}

Playground: https://play.golang.org/p/Be93J514R5

I know from reading https://blog.golang.org/pipelines that I can create a signal channel to cleanup the other threads. Alternatively, I could probably use context to accomplish it. But it seems like such a simple use case should have a simpler solution that I'm missing.

Answer 1

All but one of your goroutines are leaked, because they're still waiting to send to the errs channel - you never finish the for-range that empties it. You're also leaking the goroutine who's job is to close the errs channel, because the waitgroup is never finished.

(Also, as Andy pointed out, deleting from map is not thread-safe, so that'd need protection from a mutex.)

However, I don't think maps, mutexes, waitgroups, contexts etc. are even necessary here. I'd rewrite the whole thing to just use basic channel operations, something like the following:

package main

import (
    "fmt"
    "math/rand"
    "time"
)

func fetchAll() error {
    var N = 4
    quit := make(chan bool)
    errc := make(chan error)
    done := make(chan error)
    for i := 0; i < N; i++ {
        go func(i int) {
            // dummy fetch
            time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond)
            err := error(nil)
            if rand.Intn(2) == 0 {
                err = fmt.Errorf("goroutine %d's error returned", i)
            }
            ch := done // we'll send to done if nil error and to errc otherwise
            if err != nil {
                ch = errc
            }
            select {
            case ch <- err:
                return
            case <-quit:
                return
            }
        }(i)
    }
    count := 0
    for {
        select {
        case err := <-errc:
            close(quit)
            return err
        case <-done:
            count++
            if count == N {
                return nil // got all N signals, so there was no error
            }
        }
    }
}

func main() {
    rand.Seed(time.Now().UnixNano())
    fmt.Println(fetchAll())
}

Playground link: https://play.golang.org/p/mxGhSYYkOb

EDIT: There indeed was a silly mistake, thanks for pointing it out. I fixed the code above (I think...). I also added some randomness for added Realism™.

Also, I'd like to stress that there really are multiple ways to approach this problem, and my solution is but one way. Ultimately it comes down to personal taste, but in general, you want to strive towards "idiomatic" code - and towards a style that feels natural and easy to understand for you.