본문으로 건너뛰기

동시성 패턴

이 챕터에서 다루는 것​

7-1부터 7-7까지가 부품이었다. 이 챕터는 그것들을 조립한 완성된 형태를 모은다.

  • 워커 풀 — 고루틴 개수를 묶는다
  • 파이프라인 — 단계를 이어 붙이고 한 번에 취소한다
  • fan-out / fan-in — 흩뿌리고 모은다
  • errgroup — 에러와 취소를 한 묶음으로

워커 풀​

7-3의 세마포어는 작업마다 고루틴을 만들고 실행만 제한했다. 워커 풀은 반대다. 고루틴을 고정된 개수만 만들고, 작업을 채널로 흘린다.

세마포어워커 풀
고루틴 수작업 수만큼고정
코드 길이짧다길다
작업 수가 매우 많을 때고루틴 폭증안전
작업 수가 미리 안 정해질 때곤란자연스럽다
examples/07-concurrency/08-worker-pool/main.go
// worker는 jobs가 닫힐 때까지 일한다.
// 워커는 자기가 만들지 않은 채널을 닫지 않는다 (7-3의 소유권 규칙).
func worker(id int, jobs <-chan Job, results chan<- Result) {
	for j := range jobs {
		var r Result
		r.JobID = j.ID
		if j.Input < 0 {
			r.Err = fmt.Errorf("job %d: 음수 입력 %d", j.ID, j.Input)
		} else {
			r.Output = j.Input * j.Input
		}
		results <- r
	}
}

에러를 Result에 담아 보내는 것이 핵심이다. 워커는 error를 반환할 수 없으므로 (고루틴에는 반환값이 없다) 결과 타입 안에 넣는다.

examples/07-concurrency/08-worker-pool/main.go
	jobs := make(chan Job)
	// 결과 채널에는 작업 수만큼 버퍼를 준다.
	// 소비자가 늦어도 워커가 막히지 않는다 (7-3).
	results := make(chan Result, len(inputs))

	var wg sync.WaitGroup
	for i := range numWorkers {
		wg.Go(func() { worker(i, jobs, results) })
	}

	// jobs의 소유자는 main이다. 다 넣고 나서 main이 닫는다.
	for i, in := range inputs {
		jobs <- Job{ID: i, Input: in}
	}
	close(jobs)

	// 워커가 전부 끝난 뒤에 results를 닫는다. 감시 고루틴이 필요 없다 —
	// 여기서는 main이 직접 기다려도 results에 버퍼가 있어 데드락이 나지 않는다.
	wg.Wait()
	close(results)

배선의 순서가 전부다.

  1. jobs를 만들고 워커 N개를 띄운다. 워커는 for range jobs로 대기한다.
  2. 작업을 전부 넣고 close(jobs)한다. 그래야 워커들의 range가 끝난다.
  3. wg.Wait()로 워커 종료를 기다린 뒤 close(results)한다.

3번에서 7-3의 merge와 달라진 점을 보라. 거기서는 wg.Wait()를 별도 고루틴에 넣어야 했다. 여기서는 results에 len(inputs)만큼 버퍼가 있어서 워커가 절대 막히지 않으므로, main이 직접 기다려도 된다. 버퍼가 없다면 감시 고루틴이 필요하다.

cd examples/07-concurrency
go run ./08-worker-pool
성공: [4 9 16 25 49 64]
실패: job 1: 음수 입력 -1
실패: job 4: 음수 입력 -9

완료 순서는 실행마다 다르다. slices.Sort로 정렬해서 출력을 고정했다.

:::tip 워커 수는 몇 개로?

  • CPU 바운드: runtime.GOMAXPROCS(0) 근처. 그보다 많으면 전환 비용만 든다.
  • I/O 바운드: 훨씬 많아도 된다. 병목은 CPU가 아니라 상대편이다. 다만 상대편이 감당할 수 있는 수가 진짜 상한이다 — DB 커넥션 풀 크기, API 레이트 리밋.
  • 모르면 측정한다. 파트 8의 벤치마크가 그 도구다. :::

파이프라인​

각 단계가 채널로 이어진 구조다. 7-3의 생성자 패턴을 여러 단계로 늘린 것이고, 여기에 취소를 얹는다.

examples/07-concurrency/08-pipeline/main.go
// generate는 무한 수열을 내보낸다.
// ctx가 취소되면 즉시 멈춘다 — 무한 생성자에 반드시 필요한 장치다.
func generate(ctx context.Context) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for i := 1; ; i++ {
			select {
			case out <- i:
			case <-ctx.Done():
				return
			}
		}
	}()
	return out
}

송신도 select 안에 넣는다. 이것이 7-3의 generate와 결정적으로 다른 점이다. 그냥 out <- i라고 쓰면 아무도 안 받을 때 영원히 블록되고, ctx.Done()을 볼 기회조차 없다.

examples/07-concurrency/08-pipeline/main.go
// square는 한 단계다. 입력이 닫히거나 ctx가 취소되면 끝난다.
func square(ctx context.Context, in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for v := range in {
			select {
			case out <- v * v:
			case <-ctx.Done():
				return
			}
		}
	}()
	return out
}

단계는 전부 같은 모양이다. <-chan T를 받아 <-chan U를 반환하고, 자기가 만든 것만 닫고, ctx를 존중한다.

examples/07-concurrency/08-pipeline/main.go
func main() {
	// 조기 종료: 필요한 만큼만 받고 취소한다.
	// defer cancel()이 파이프라인 전체를 정리한다.
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	stream := filter(ctx, square(ctx, generate(ctx)), func(v int) bool {
		return v%2 == 1 // 홀수 제곱만
	})

	var got []int
	for v := range stream {
		got = append(got, v)
		if len(got) == 5 {
			break // 여기서 나가면 defer cancel()이 위쪽 단계를 전부 멈춘다
		}
	}

	// 파이프라인은 순서를 보존한다. fan-in과 다른 점이다.
	fmt.Println("홀수 제곱 5개:", got)
}
go run ./08-pipeline
홀수 제곱 5개: [1 9 25 49 81]

break로 나갔는데 무한 생성자가 멈춘다. defer cancel() 한 줄이 세 단계 전부를 정리한다. 이것이 파이프라인에 context를 꿰는 이유다 — 소비자가 언제 그만둘지 생산자가 알 필요가 없다.

ctx 없이 만들었다면 break 이후 generate, square, filter 고루틴 셋이 전부 누수된다.

:::note 이터레이터와 무엇이 다른가 5-4의 iter.Seq도 값을 순서대로 내보내고 break로 조기 종료할 수 있다. 차이는 누가 실행 흐름을 쥐고 있느냐다.

  • 이터레이터: 고루틴이 하나다. 생산과 소비가 번갈아 일어난다. break가 yield의 false 반환으로 바로 전달된다.
  • 파이프라인: 단계마다 고루틴이 있다. 단계들이 동시에 돈다. 그래서 I/O 대기가 겹쳐지고 여러 코어를 쓸 수 있다. 대신 취소를 직접 설계해야 한다.

동시성이 필요 없으면 이터레이터가 훨씬 간단하다. 단계 중에 느린 I/O가 있어서 겹쳐 실행하는 것이 이득일 때만 파이프라인으로 간다. :::

fan-out / fan-in​

한 단계가 병목이면 그 단계만 여러 개로 늘린다(fan-out). 그리고 결과를 다시 하나로 모은다(fan-in).

generate ──┬─→ square #1 ──┐
           ├─→ square #2 ──┼─→ merge ─→ 소비
           └─→ square #3 ──┘

merge는 7-3에서 이미 만들었다. 그대로 쓰면 된다.

fan-in의 대가는 순서다. 파이프라인은 순서를 보존하지만 fan-in은 못 한다. 순서가 필요하면 두 가지 방법이 있다.

1. 인덱스를 미리 잡아 둔다. 7-1부터 계속 쓴 기법이다. 결과 개수를 미리 알 때 가장 간단하다.

2. 값에 시퀀스 번호를 붙여 보내고 소비자가 재정렬한다. 스트리밍이라 개수를 모를 때 쓴다. 소비자가 다음 번호가 올 때까지 버퍼링해야 하므로, 가장 느린 워커가 전체를 막는다 — 그럴 바에는 fan-out을 안 하는 게 나을 수도 있다.

errgroup​

여기까지의 패턴에는 공통된 불편이 남아 있다.

  • wg.Go는 func()을 받으므로 에러를 반환할 수 없다.
  • 하나가 실패했을 때 나머지를 멈추려면 별도로 context를 배선해야 한다.
  • 동시 실행 수를 제한하려면 세마포어를 손으로 만들어야 한다.

golang.org/x/sync/errgroup이 셋을 한 번에 해결한다.

go get golang.org/x/sync@v0.22.0

이 예제는 golang.org/x/sync v0.22.0 기준이다. x/ 저장소는 표준 라이브러리는 아니지만 Go 팀이 관리한다. 동시성 코드에서 사실상 표준 확장으로 쓰인다.

func WithContext(ctx context.Context) (*Group, context.Context)
func (g *Group) Go(f func() error)
func (g *Group) SetLimit(n int)
func (g *Group) TryGo(f func() error) bool
func (g *Group) Wait() error
examples/07-concurrency/08-errgroup/main.go
// runAll은 모든 입력을 동시에 처리한다.
// 하나라도 실패하면 ctx가 취소되고 나머지가 중단된다.
func runAll(inputs []int, limit int) ([]int, error) {
	g, ctx := errgroup.WithContext(context.Background())
	g.SetLimit(limit) // 동시 실행 고루틴 수 제한. 세마포어를 손으로 만들 필요가 없다.

	out := make([]int, len(inputs))
	for i, v := range inputs {
		g.Go(func() error {
			r, err := process(ctx, v)
			if err != nil {
				return err
			}
			out[i] = r // 인덱스로 쓰므로 순서가 보존되고 경합도 없다
			return nil
		})
	}

	if err := g.Wait(); err != nil {
		return nil, err
	}
	return out, nil
}

errgroup.WithContext가 돌려주는 ctx를 작업 함수에 넘기는 것이 핵심이다. 누군가 nil이 아닌 에러를 반환하는 순간 그 ctx가 취소되고, ctx.Done()을 확인하는 형제 작업들이 스스로 멈춘다.

examples/07-concurrency/08-errgroup/main.go
// process는 음수를 만나면 에러를 낸다.
// ctx.Done()을 확인하므로 형제 작업이 실패하면 스스로 멈춘다.
func process(ctx context.Context, v int) (int, error) {
	if v < 0 {
		return 0, fmt.Errorf("%d: %w", v, errBadInput)
	}
	select {
	case <-time.After(10 * time.Millisecond):
		return v * v, nil
	case <-ctx.Done():
		return 0, ctx.Err()
	}
}
go run ./08-errgroup
정상 결과: [1 4 9 16 25 36 49 64] err: <nil>
실패 결과: []
err: -42: 잘못된 입력
errors.Is(err, errBadInput): true

%w로 감쌌으므로 errors.Is가 통한다. 4-7의 래핑 체인이 동시성 코드에서도 그대로 작동한다.

:::warning errgroup은 첫 에러 하나만 준다 g.Wait()는 가장 먼저 발생한 에러를 돌려준다. 나머지는 버려진다. 전부 모으고 싶으면 errgroup을 쓰지 말고 결과 슬라이스에 에러를 담은 뒤 4-7의 errors.Join으로 합친다.

"하나라도 실패하면 전체 실패"가 맞는 상황에서만 errgroup이 맞는 도구다. :::

:::tip SetLimit는 반드시 Go 호출 전에 SetLimit(n)을 이미 고루틴이 도는 중에 부르면 패닉이다. 그리고 n을 넘겨서 Go를 부르면 자리가 날 때까지 Go 자체가 블록된다. 블록되는 것이 싫으면 TryGo가 false를 돌려준다. :::

wg.Go와 errgroup.Go​

이름이 같아서 헷갈리기 쉽다.

sync.WaitGroup.Goerrgroup.Group.Go
시그니처func()func() error
에러못 받는다첫 에러를 Wait가 반환
취소 전파없음WithContext와 함께 쓰면 있다
개수 제한없음SetLimit
의존성표준 라이브러리golang.org/x/sync

에러가 없거나 신경 쓰지 않는다면 sync.WaitGroup이면 충분하다. 의존성을 하나 줄이는 것도 가치다.

정상 종료​

서버를 끄는 순서는 정해져 있다.

  1. 새 작업 받기를 멈춘다 (리스너를 닫는다).
  2. 진행 중인 작업이 끝나기를 기다린다 — 단, 데드라인을 둔다.
  3. 자원을 정리한다 (DB, 파일, 커넥션).

패턴으로 옮기면 이렇다.

// 1. 종료 신호를 컨텍스트로 만든다.
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()

// 2. 워커들이 그 ctx를 존중한다.
g, gctx := errgroup.WithContext(ctx)
g.Go(func() error { return runWorker(gctx) })

// 3. 신호가 오면 유예 시간을 두고 기다린다.
<-ctx.Done()
shutdownCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
defer cancel()
_ = shutdownCtx

context.WithoutCancel(ctx)가 중요하다. 이미 취소된 컨텍스트를 부모로 삼으면 자식도 즉시 취소된다. 종료 유예 시간은 취소로부터 독립적이어야 한다.

os/signal.NotifyContext는 취소 원인에 어떤 시그널이었는지를 담아 준다. context.Cause(ctx)로 꺼낼 수 있다. 서버 종료의 전체 코드는 net/http가 필요하니 파트 12(프로덕션)의 몫이다.

안티패턴​

1. 결과를 공유 슬라이스에 append​

var results []int
for _, v := range items {
	wg.Go(func() { results = append(results, f(v)) })   // 경합
}

7-6에서 본 그대로다. 인덱스로 자리를 잡거나 채널로 보낸다.

2. 고루틴을 안 세고 띄운다​

for _, item := range millionItems {
	go process(item)   // 고루틴 100만 개
}

고루틴은 싸지만 그 안에서 여는 커넥션과 버퍼는 싸지 않다. 워커 풀이나 SetLimit으로 묶는다.

3. 라이브러리가 몰래 고루틴을 띄운다​

func NewClient() *Client {
	c := &Client{}
	go c.backgroundRefresh()   // 누가 멈추나?
}

호출자가 이 고루틴의 존재를 모르고, 멈출 방법도 없다. Close() 메서드를 주거나 ctx를 받는다. 그리고 문서에 적는다. 6-1에서 정한 "init에서 부수 효과를 만들지 않는다"의 연장이다.

4. 워커에 의존성을 전역 변수로 준다​

워커가 DB나 캐시를 써야 한다면 생성자로 주입한다. 6-5의 소비자 쪽 인터페이스를 그대로 쓴다.

// 워커가 필요로 하는 것만 인터페이스로 선언한다 — 쓰는 쪽에서.
type Store interface {
	Save(ctx context.Context, r Result) error
}

type Pool struct {
	store   Store
	workers int
}

func NewPool(store Store, workers int) *Pool { ... }

*sql.DB를 통째로 받지 않는다. 필요한 메서드만 받으면 테스트에서 가짜로 바꿔 끼울 수 있고(파트 8), 순환 import도 생기지 않는다.

5. 채널을 쓸 자리가 아닌데 채널을 쓴다​

7-5의 판단 기준으로 돌아간다. 소유권이 옮겨 다니면 채널, 제자리 상태를 지키면 뮤텍스. 카운터 하나에 고루틴과 채널을 붙이는 것은 과설계다.

6. 파이프라인 단계마다 ctx를 확인하지 않는다​

송신을 select 밖에 두면 취소가 그 지점에서 막힌다. 단계 전부에 넣거나, 아예 파이프라인을 쓰지 않는다.

정리​

  • 워커 풀은 고루틴 수를 고정한다. jobs를 만든 쪽이 다 넣고 닫고, wg.Wait() 뒤에 results를 닫는다. 에러는 결과 타입 안에 담아 보낸다.
  • 파이프라인은 단계마다 <-chan T를 받아 <-chan U를 반환하고, 자기가 만든 것만 닫는다. 송신도 select 안에 넣어야 취소가 통한다.
  • defer cancel() 하나가 파이프라인 전체를 정리한다. 소비자가 break해도 생산자가 멈춘다.
  • fan-in은 순서를 잃는다. 필요하면 인덱스를 미리 잡거나 시퀀스 번호로 재정렬한다.
  • errgroup은 에러 반환 + 첫 실패 시 취소 + 개수 제한을 한 묶음으로 준다. WithContext가 준 ctx를 작업에 넘기지 않으면 취소가 작동하지 않는다. 첫 에러 하나만 돌려준다.
  • 에러가 없으면 sync.WaitGroup으로 충분하다.
  • 정상 종료는 "받기 중단 → 유예 대기 → 정리" 순서이고, 유예 컨텍스트는 context.WithoutCancel로 취소에서 떼어낸다.
  • 동시성이 필요 없으면 이터레이터가 낫다. 파이프라인의 비용은 취소 설계다.

연습문제​

  1. 08-worker-pool에서 results의 버퍼를 없애고(make(chan Result)) 실행해 보자. 어떻게 멈추는가? 7-3의 merge처럼 감시 고루틴을 넣어 고쳐 보고, 버퍼를 주는 방법과 비교해 각각 언제 나은지 정리해 보자.

  2. 08-pipeline의 generate에서 select를 없애고 out <- i로 바꾼 뒤, main에서 5개를 받고 break한 다음 runtime.NumGoroutine()을 찍어 보자. 몇 개가 남는가? defer cancel()이 왜 소용없어지는지 설명해 보자.

  3. 08-errgroup의 process에서 case <-ctx.Done()을 빼 보자. 실패 케이스에서 전체 소요 시간이 어떻게 달라지는가? "errgroup이 취소를 전파한다"는 말이 작업 함수의 협조가 필요하다는 뜻임을 확인해 보자.

  4. 08-worker-pool을 errgroup으로 다시 써 보자. jobs 채널이 필요한가? 어느 쪽이 짧은가? 작업 수가 1,000만 개라면 어느 쪽을 택하겠는가?