ErrGroup: Unlocking Go’s Concurrency Power

7 min read1 day ago

Leapcell: The Next-Gen Serverless Platform for Web Hosting, Async Tasks, and Redis

Go Language errgroup Library: A Powerful Concurrency Control Tool

errgroup is a utility in the official Go library x used for concurrently executing multiple goroutines and handling errors. It implements errgroup.Group based on sync.WaitGroup, providing more powerful functions for concurrent programming.

Advantages of errgroup

Compared with sync.WaitGroup, errgroup.Group has the following advantages:

Error Handling: sync.WaitGroup is only responsible for waiting for the goroutines to complete and does not handle return values or errors. While errgroup.Group cannot directly handle return values, it can immediately cancel other running goroutines when a goroutine encounters an error and return the first non-nil error in the Wait method.
Context Cancellation: errgroup can be used in conjunction with context.Context. When a goroutine encounters an error, it can automatically cancel other goroutines, effectively controlling resources and avoiding unnecessary work.
Simplifying Concurrent Programming: Using errgroup can reduce the boilerplate code for error handling. Developers do not need to manually manage error states and synchronization logic, making concurrent programming simpler and more maintainable.
Limiting the Number of Concurrency: errgroup provides an interface to limit the number of concurrent goroutines to avoid overloading, which is a feature that sync.WaitGroup does not have.

Example of Using sync.WaitGroup

Before introducing errgroup.Group, let's first review the usage of sync.WaitGroup.

package main

import (
    "fmt"
    "net/http"
    "sync"
)

func main() {
    var urls = []string{
        "http://www.golang.org/",
        "http://www.google.com/",
        "http://www.somestupidname.com/", 
    }
    var err error
    var wg sync.WaitGroup 
    for _, url := range urls {
        wg.Add(1) 
        go func() {
            defer wg.Done() 
            resp, e := http.Get(url)
            if e != nil { 
                err = e
                return
            }
            defer resp.Body.Close()
            fmt.Printf("fetch url %s status %s\n", url, resp.Status)
        }()
    }
    wg.Wait()
    if err != nil { 
        fmt.Printf("Error: %s\n", err)
    }
}

Execution result:

$ go run examples/main.go
fetch url http://www.google.com/ status 200 OK
fetch url http://www.golang.org/ status 200 OK
Error: Get "http://www.somestupidname.com/": dial tcp: lookup www.somestupidname.com: no such host

Typical idiom of sync.WaitGroup:

var wg sync.WaitGroup

for ... {
    wg.Add(1)
    go func() {
        defer wg.Done()
        // do something
    }()
}

wg.Wait()

Example of Using errgroup.Group

Basic Usage

The usage pattern of errgroup.Group is similar to that of sync.WaitGroup.

package main

import (
    "fmt"
    "net/http"
    "golang.org/x/sync/errgroup"
)

func main() {
    var urls = []string{
        "http://www.golang.org/",
        "http://www.google.com/",
        "http://www.somestupidname.com/", 
    }
    var g errgroup.Group 
    for _, url := range urls {
        g.Go(func() error {
            resp, err := http.Get(url)
            if err != nil {
                return err 
            }
            defer resp.Body.Close()
            fmt.Printf("fetch url %s status %s\n", url, resp.Status)
            return nil 
        })
    }
    if err := g.Wait(); err != nil {
        fmt.Printf("Error: %s\n", err)
    }
}

Execution result:

$ go run examples/main.go
fetch url http://www.google.com/ status 200 OK
fetch url http://www.golang.org/ status 200 OK
Error: Get "http://www.somestupidname.com/": dial tcp: lookup www.somestupidname.com: no such host

Context Cancellation

errgroup provides errgroup.WithContext to add a cancellation function.

package main

import (
    "context"
    "fmt"
    "net/http"
    "sync"
    "golang.org/x/sync/errgroup"
)

func main() {
    var urls = []string{
        "http://www.golang.org/",
        "http://www.google.com/",
        "http://www.somestupidname.com/", 
    }
    g, ctx := errgroup.WithContext(context.Background())
    var result sync.Map
    for _, url := range urls {
        g.Go(func() error {
            req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
            if err != nil {
                return err 
            }
            resp, err := http.DefaultClient.Do(req)
            if err != nil {
                return err 
            }
            defer resp.Body.Close()
            result.Store(url, resp.Status)
            return nil 
        })
    }
    if err := g.Wait(); err != nil {
        fmt.Println("Error: ", err)
    }
    result.Range(func(key, value any) bool {
        fmt.Printf("fetch url %s status %s\n", key, value)
        return true
    })
}

Execution result:

$ go run examples/withcontext/main.go
Error:  Get "http://www.somestupidname.com/": dial tcp: lookup www.somestupidname.com: no such host
fetch url http://www.google.com/ status 200 OK

Since the request to http://www.somestupidname.com/ reported an error, the program cancelled the request to http://www.golang.org/.

Limiting the Number of Concurrency

errgroup provides errgroup.SetLimit to limit the number of concurrently executing goroutines.

package main

import (
    "fmt"
    "time"
    "golang.org/x/sync/errgroup"
)

func main() {
    var g errgroup.Group
    g.SetLimit(3)
    for i := 1; i <= 10; i++ {
        g.Go(func() error {
            fmt.Printf("Goroutine %d is starting\n", i)
            time.Sleep(2 * time.Second) 
            fmt.Printf("Goroutine %d is done\n", i)
            return nil
        })
    }
    if err := g.Wait(); err != nil {
        fmt.Printf("Encountered an error: %v\n", err)
    }
    fmt.Println("All goroutines complete.")
}

Execution result:

$  go run examples/main.go
Goroutine 3 is starting
Goroutine 1 is starting
Goroutine 2 is starting
Goroutine 2 is done
Goroutine 1 is done
Goroutine 5 is starting
Goroutine 3 is done
Goroutine 6 is starting
Goroutine 4 is starting
Goroutine 6 is done
Goroutine 5 is done
Goroutine 8 is starting
Goroutine 4 is done
Goroutine 7 is starting
Goroutine 9 is starting
Goroutine 9 is done
Goroutine 8 is done
Goroutine 10 is starting
Goroutine 7 is done
Goroutine 10 is done
All goroutines complete.

Try to Start

errgroup provides errgroup.TryGo to try to start a task, which needs to be used in conjunction with errgroup.SetLimit.

package main

import (
    "fmt"
    "time"
    "golang.org/x/sync/errgroup"
)

func main() {
    var g errgroup.Group
    g.SetLimit(3)
    for i := 1; i <= 10; i++ {
        if g.TryGo(func() error {
            fmt.Printf("Goroutine %d is starting\n", i)
            time.Sleep(2 * time.Second) 
            fmt.Printf("Goroutine %d is done\n", i)
            return nil
        }) {
            fmt.Printf("Goroutine %d started successfully\n", i)
        } else {
            fmt.Printf("Goroutine %d could not start (limit reached)\n", i)
        }
    }
    if err := g.Wait(); err != nil {
        fmt.Printf("Encountered an error: %v\n", err)
    }
    fmt.Println("All goroutines complete.")
}

Execution result:

$ go run examples/main.go
Goroutine 1 started successfully
Goroutine 1 is starting
Goroutine 2 is starting
Goroutine 2 started successfully
Goroutine 3 started successfully
Goroutine 4 could not start (limit reached)
Goroutine 5 could not start (limit reached)
Goroutine 6 could not start (limit reached)
Goroutine 7 could not start (limit reached)
Goroutine 8 could not start (limit reached)
Goroutine 9 could not start (limit reached)
Goroutine 10 could not start (limit reached)
Goroutine 3 is starting
Goroutine 2 is done
Goroutine 3 is done
Goroutine 1 is done
All goroutines complete.

Source Code Interpretation

The source code of errgroup mainly consists of 3 files:

Core Structure

type token struct{}

type Group struct {
    cancel func(error)
    wg sync.WaitGroup
    sem chan token
    errOnce sync.Once
    err     error
}

token: An empty structure used to pass signals to control the number of concurrency.
Group:
cancel: The function called when the context is cancelled.
wg: The internally used sync.WaitGroup.
sem: The signal channel that controls the number of concurrent coroutines.
errOnce: Ensures that the error is handled only once.
err: Records the first error.

Main Methods

SetLimit: Limits the number of concurrency.

func (g *Group) SetLimit(n int) {
    if n < 0 {
        g.sem = nil
        return
    }
    if len(g.sem) != 0 {
        panic(fmt.Errorf("errgroup: modify limit while %v goroutines in the group are still active", len(g.sem)))
    }
    g.sem = make(chan token, n)
}

Go: Starts a new coroutine to execute the task.

func (g *Group) Go(f func() error) {
    if g.sem != nil {
        g.sem <- token{}
    }

    g.wg.Add(1)
    go func() {
        defer g.done()
        if err := f(); err != nil {
            g.errOnce.Do(func() {
                g.err = err
                if g.cancel != nil {
                    g.cancel(g.err)
                }
            })
        }
    }()
}

Wait: Waits for all tasks to complete and returns the first error.

func (g *Group) Wait() error {
    g.wg.Wait()
    if g.cancel != nil {
        g.cancel(g.err)
    }
    return g.err
}

TryGo: Tries to start a task.

func (g *Group) TryGo(f func() error) bool {
    if g.sem != nil {
        select {
        case g.sem <- token{}:
        default:
            return false
        }
    }

    g.wg.Add(1)
    go func() {
        defer g.done()
        if err := f(); err != nil {
            g.errOnce.Do(func() {
                g.err = err
                if g.cancel != nil {
                    g.cancel(g.err)
                }
            })
        }
    }()
    return true
}

Conclusion

errgroup is an official extended library that adds error handling capabilities on the basis of sync.WaitGroup, providing functions such as synchronization, error propagation, and context cancellation. Its WithContext method can add a cancellation function, SetLimit can limit the number of concurrency, and TryGo can try to start a task. The source code is ingeniously designed and worthy of reference.