Go 언어 errgroup 라이브러리: 강력한 동시성 제어 도구

errgroup은 여러 goroutine을 동시에 실행하고 오류를 처리하는 데 사용되는 공식 Go 라이브러리 x의 유틸리티입니다. sync.WaitGroup을 기반으로 errgroup.Group을 구현하여 동시성 프로그래밍을 위한 더 강력한 기능을 제공합니다.

errgroup의 장점

sync.WaitGroup과 비교하여 errgroup.Group은 다음과 같은 장점이 있습니다.

1. 오류 처리: sync.WaitGroup은 goroutine의 완료를 기다리는 역할만 하며 반환 값이나 오류를 처리하지 않습니다. 반면 errgroup.Group은 반환 값을 직접 처리할 수는 없지만 goroutine에서 오류가 발생하면 즉시 다른 실행 중인 goroutine을 취소하고 Wait 메서드에서 첫 번째 non-nil 오류를 반환할 수 있습니다.
2. 컨텍스트 취소: errgroup은 context.Context와 함께 사용할 수 있습니다. goroutine에서 오류가 발생하면 다른 goroutine을 자동으로 취소하여 리소스를 효과적으로 제어하고 불필요한 작업을 방지할 수 있습니다.
3. 동시성 프로그래밍 간소화: errgroup을 사용하면 오류 처리를 위한 상용구 코드를 줄일 수 있습니다. 개발자는 오류 상태 및 동기화 로직을 수동으로 관리할 필요가 없어 동시성 프로그래밍이 더 간단하고 유지 관리하기 쉬워집니다.
4. 동시성 수 제한: errgroup은 과부하를 방지하기 위해 동시 실행되는 goroutine 수를 제한하는 인터페이스를 제공하며, 이는 sync.WaitGroup에는 없는 기능입니다.

sync.WaitGroup 사용 예시

errgroup.Group을 소개하기 전에 먼저 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)
    }
}

실행 결과:

$ go run waitgroup/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

sync.WaitGroup의 일반적인 관용구:

var wg sync.WaitGroup

for ... {
    wg.Add(1)

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

wg.Wait()

errgroup.Group 사용 예시

기본 사용법

errgroup.Group의 사용 패턴은 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)
    }
}

실행 결과:

$ 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

컨텍스트 취소

errgroup은 취소 기능을 추가하기 위해 errgroup.WithContext를 제공합니다.

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

실행 결과:

$ 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

http://www.somestupidname.com/에 대한 요청에서 오류가 보고되었으므로 프로그램에서 http://www.golang.org/에 대한 요청을 취소했습니다.

동시성 수 제한

errgroup은 동시에 실행되는 goroutine 수를 제한하기 위해 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++ {
        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.")
}

실행 결과:

$  go run examples/setlimit/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.

시작 시도

errgroup은 작업 시작을 시도하기 위해 errgroup.TryGo를 제공하며, 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.")
}

실행 결과:

$ go run examples/trygo/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 2 is done
Goroutine 3 is starting
Goroutine 3 is done
Goroutine 1 is done
All goroutines complete.

소스 코드 해석

errgroup의 소스 코드는 주로 3개의 파일로 구성됩니다.

• 주요 로직 코드
• Go 1.24 이상 버전에 대한 withCancelCause 구현
• Go 1.24 이전 버전에 대한 withCancelCause 구현

핵심 구조

type token struct{}

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

• token: 동시성 수를 제어하기 위해 신호를 전달하는 데 사용되는 빈 구조체입니다.
• Group:
  • cancel: 컨텍스트가 취소될 때 호출되는 함수입니다.
  • wg: 내부적으로 사용되는 sync.WaitGroup입니다.
  • sem: 동시 코루틴 수를 제어하는 신호 채널입니다.
  • errOnce: 오류가 한 번만 처리되도록 합니다.
  • err: 첫 번째 오류를 기록합니다.

주요 메서드

• SetLimit: 동시성 수를 제한합니다.

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: 작업을 실행하기 위해 새 코루틴을 시작합니다.

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: 모든 작업이 완료될 때까지 기다리고 첫 번째 오류를 반환합니다.

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

• TryGo: 작업 시작을 시도합니다.

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
}

결론

errgroup은 sync.WaitGroup을 기반으로 오류 처리 기능을 추가하는 공식 확장 라이브러리로, 동기화, 오류 전파 및 컨텍스트 취소와 같은 기능을 제공합니다. WithContext 메서드를 사용하면 취소 함수를 추가할 수 있고, SetLimit는 동시성 수를 제한할 수 있으며, TryGo는 작업 시작을 시도할 수 있습니다. 소스 코드는 기발하게 설계되었으며 참조할 가치가 있습니다.

Leapcell: 웹 호스팅, 비동기 작업 및 Redis를 위한 차세대 서버리스 플랫폼

마지막으로 golang 배포에 가장 적합한 플랫폼인 **Leapcell**을 추천합니다.

1. 다국어 지원

• JavaScript, Python, Go 또는 Rust로 개발하세요.

2. 무제한 프로젝트를 무료로 배포하세요

• 사용량에 대해서만 지불하세요. 요청도 없고, 요금도 없습니다.

3. 최고의 비용 효율성

• 유휴 요금 없이 사용량에 따라 지불하세요.
• 예: $25로 평균 응답 시간 60ms에서 694만 건의 요청을 지원합니다.

4. 간소화된 개발자 경험

• 간편한 설정을 위한 직관적인 UI.
• 완전 자동화된 CI/CD 파이프라인 및 GitOps 통합.
• 실행 가능한 통찰력을 위한 실시간 메트릭 및 로깅.

5. 손쉬운 확장성 및 고성능

• 고도의 동시성을 쉽게 처리할 수 있는 자동 확장.
• 운영 오버헤드가 제로이므로 구축에만 집중하세요.

자세한 내용은 설명서를 참조하세요!

Leapcell Twitter: https://x.com/LeapcellHQ