errgroup 剖析

type Group struct {
  // cancel  就是 context 返 回 的 
  // cancel  函 数 
  cancel func()


  // wg 管 理 一 组 子 任 务 协 程 
  wg sync.WaitGroup


  // errOnce 单例模式，保证失败只发 
  // 一次，并且由出现错误的子协程最先 
  // 发送 
  errOnce sync.Once
  
  // err 存储子协程的错误 
  // 同步返回后，通过它取得 
  // 失败的原因 
  err     error
}

// 提供了 Go 接口，用于启动子任务协程
// f: 子任务, 需要满足接口 func() error
func (g *Group) Go(f func() error) {
  // WaitGroup 添加一个子任务
  g.wg.Add(1)


  // 启动子任务协程
  go func() {
    // 同步通知，减少 WaitGroup 等待的任务数目
    defer g.wg.Done()


    // 错误处理
    if err := f(); err != nil {
      // 子任务出现错误，单例启动处理
      // 错误，错误处理程序包含：
      // 1. 存储 错误 error
      // 2. cancel 发送信号，结束 context 上下文
      g.errOnce.Do(func() {
        g.err = err
        if g.cancel != nil {
          g.cancel()
        }
      })
    }
  }()
}

// Wait 调用后，内部调用 WaitGroup 的
// Wait 进入同步阻塞等待
func (g *Group) Wait() error {
  // WaitGroup 同步等待
  g.wg.Wait()
  
  // 给所有的 context 关联的子协程发送
  // cancel 信号
  if g.cancel != nil {
    g.cancel()
  }
  
  // 返回协程的错误
  return g.err
}

// WithContext 接收上下文，一次为上下文创建
// 可被取消的 context，同时返回 Group 组
func WithContext(ctx context.Context) (*Group, context.Context) {
  ctx, cancel := context.WithCancel(ctx)
  // Group 的 cancel 由 context 得到
  return &Group{cancel: cancel}, ctx
}

func main() {
  // 计算当前路径下文件的 md5 值
  m, err := MD5All(context.Background(), ".")
  if err != nil {
    log.Fatal(err)
  }


  for k, sum := range m {
    fmt.Printf("%s:\t%x\n", k, sum)
  }
}


// 结果集
type result struct {
  path string
  sum  [md5.Size]byte
}


// 计算所有文件的 md5 
func MD5All(ctx context.Context, root string) (map[string][md5.Size]byte, error) {


  // 创建含有上下文信息的 Group
  g, ctx := errgroup.WithContext(ctx)


  // 用来发送读到的文件路径
  paths := make(chan string)


  // G-pro 协程，遍历目录
  g.Go(func() error {
    // 发送方只有一个，发送负责关闭 chan
    defer close(paths)


    // 遍历目录
    return filepath.Walk(root, func(path string, info os.FileInfo, err error) error {
      if err != nil {
        return err
      }


      if !info.Mode().IsRegular() {
        return nil
      }


      // select 监听取消信号
      // 出错时，其它协程会发送
      // cancel 信号，此时监听
      // ctx.Done,便于结束
      select {
      case paths <- path:


      // 被取消
      case <-ctx.Done():
        return ctx.Err()
      }


      return nil
    })
  })


  // 结果通道
  c := make(chan result)


  const numDigesters = 20


  // G-w* 子任务组协程，计算一部分文件的 md5
  for i := 0; i < numDigesters; i++ {
    g.Go(func() error {


      // 监听读到的文件路径
      for path := range paths {


        // 读取文件内容
        data, err := ioutil.ReadFile(path)
        if err != nil {
          return err
        }


        // 计算 md5 并发送计算结果
        select {
        case c <- result{path, md5.Sum(data)}:


        // 同样等待出错结束任务的通知
        case <-ctx.Done():
          return ctx.Err()
        }
      }
      return nil
    })
  }


  // G-wait 协程，等待 errgroup 的返回
  go func() {
    g.Wait()
    close(c)
  }()


  // G-result 协程，汇总存储结果
  m := make(map[string][md5.Size]byte)
  for r := range c {
    m[r.path] = r.sum
  }
  // Check whether any of the goroutines failed. Since g is accumulating the
}