종합 실습 — 병렬 수집기

만들 것

여러 URL을 동시에 가져와 결과를 모으는 수집기다. 요구사항은 이렇다.

1. 동시 실행 수를 제한한다 — 상대편을 때려잡지 않는다.
2. 전체 예산(타임아웃)을 둔다 — 느린 하나가 전체를 붙들지 않는다.
3. URL 하나의 실패가 전체 실패가 아니다 — 실패도 결과다.
4. 결과 순서는 입력 순서를 따른다 — 도착 순서가 아니다.
5. 진행률을 보고한다.
6. 고루틴이 하나도 남지 않는다.

파트 7에서 배운 것이 거의 전부 들어간다.

:::note 네트워크는 파트 9의 주제다 제목은 "URL 수집기"지만 net/http는 쓰지 않는다. 진짜 네트워크에 의존하면 예제가 오프라인에서 안 돌고, 실행할 때마다 결과가 달라져 이 파트에서 훈련한 "결정적 출력"이 무너진다.

대신 지연과 실패를 내가 통제하는 가짜 fetcher를 만든다. 워커 풀·타임아웃· 누수 검사라는 진짜 주제에 집중하기 위해서다. 파트 9에서 net/http를 배운 뒤 이 Fetcher의 구현만 갈아 끼우면 그대로 진짜 수집기가 된다. 그렇게 갈아 끼울 수 있도록 설계하는 것이 이 실습의 절반이다. :::

구조

파트 6에서 배운 대로 패키지를 나눈다. 경계가 생기는 곳에서만 나눈다.

examples/07-concurrency/09-collector/
├── main.go                            배선과 출력만
└── internal/
    ├── fetch/fetch.go                 가짜 fetcher (나중에 진짜로 교체할 자리)
    └── collect/collect.go             워커 풀과 집계 — fetch를 import하지 않는다

internal/이 09-collector/ 아래에 있으므로 이 프로그램 밖에서는 import할 수 없다(6-5).

가장 중요한 설계 결정은 화살표의 방향이다.

main ──→ fetch
  │
  └────→ collect        collect ↛ fetch

collect는 fetch를 모른다.

1단계 — 소비자 쪽 인터페이스

collect가 필요로 하는 것은 "URL을 주면 크기를 돌려주는 무언가"뿐이다. 그 인터페이스를 쓰는 쪽인 collect에 선언한다.

examples/07-concurrency/09-collector/internal/collect/collect.go

// Fetcher는 URL 하나를 가져와 본문 크기를 돌려준다.
// 메서드 하나짜리 최소 인터페이스다.
type Fetcher interface {
	Fetch(ctx context.Context, url string) (int, error)
}

6-5의 소비자 쪽 정의 그대로다. 결과가 셋이다.

• collect가 fetch를 import하지 않으므로 순환이 생길 여지가 없다.
• fetch.Sim은 Fetcher라는 이름을 한 번도 쓰지 않는다 — 암묵적 구현.
• 나중에 net/http 구현으로 바꿔도 collect는 한 줄도 안 바뀐다.

2단계 — 통제 가능한 가짜 fetcher

examples/07-concurrency/09-collector/internal/fetch/fetch.go

// ErrNotFound는 존재하지 않는 URL을 요청했을 때의 센티널 에러다.
var ErrNotFound = errors.New("찾을 수 없음")

// Sim은 URL 문자열만 보고 지연과 크기를 결정하는 가짜 수집기다.
// 같은 URL에 대해 항상 같은 결과를 낸다.
type Sim struct {
	// Unit은 지연의 기본 단위다. 테스트나 데모에서 전체 속도를 조절한다.
	Unit time.Duration
}

// Delay는 이 URL을 가져오는 데 걸리는 시간을 돌려준다.
func (s Sim) Delay(url string) time.Duration {
	if strings.HasSuffix(url, "/slow") {
		return 100 * s.Unit
	}
	return time.Duration(len(url)%5+1) * s.Unit
}

// Fetch는 본문 바이트 수를 돌려준다.
// ctx가 먼저 끝나면 즉시 ctx.Err()를 반환한다 — "컨텍스트를 존중한다"는 것이
// 바로 이 select다.
func (s Sim) Fetch(ctx context.Context, url string) (int, error) {
	if strings.HasSuffix(url, "/missing") {
		return 0, ErrNotFound
	}

	t := time.NewTimer(s.Delay(url))
	defer t.Stop()

	select {
	case <-t.C:
		return len(url) * 10, nil
	case <-ctx.Done():
		return 0, ctx.Err()
	}
}

두 가지가 의도적이다.

Unit을 필드로 뺀 것. 지연을 곱셈 계수 하나로 조절할 수 있으면 데모와 (파트 8의) 테스트에서 속도를 마음대로 바꿀 수 있다. 시간을 하드코딩하지 않는다.

ctx.Done()을 확인하는 것. 7-7에서 말한 "컨텍스트를 받는 것과 존중하는 것은 다르다"가 여기 있다. 이 select가 없으면 타임아웃이 걸려도 100단위 지연을 다 채운다.

3단계 — 워커 풀

examples/07-concurrency/09-collector/internal/collect/collect.go

// Result는 URL 하나의 수집 결과다. 에러도 값으로 담는다 —
// 고루틴은 에러를 반환할 수 없으므로 결과 타입에 넣는다.
type Result struct {
	URL   string
	Bytes int
	Err   error
}

// Collector는 워커 수와 진행률 보고 방식을 들고 있다.
// 의존성(Fetcher)은 생성자로 주입받는다. 전역 변수를 쓰지 않는다.
type Collector struct {
	fetcher Fetcher
	workers int

	// Progress는 완료 개수가 25% 구간을 넘을 때마다 호출된다.
	// 집계 고루틴 하나에서만 호출되므로 이 함수 안에서는 동기화가 필요 없다.
	Progress func(done, total int)
}

func New(f Fetcher, workers int) *Collector {
	if workers < 1 {
		workers = 1
	}
	return &Collector{fetcher: f, workers: workers}
}

Run이 본체다.

examples/07-concurrency/09-collector/internal/collect/collect.go

func (c *Collector) Run(ctx context.Context, urls []string) ([]Result, error) {
	results := make([]Result, len(urls))
	if len(urls) == 0 {
		return results, nil
	}

	// 진행률 집계. 상태를 고루틴 하나가 독점하므로 뮤텍스가 없다.
	ticks := make(chan struct{})
	reported := make(chan struct{})
	go func() {
		defer close(reported)
		n := 0
		for range ticks {
			n++
			if c.Progress != nil && milestone(n, len(urls)) {
				c.Progress(n, len(urls))
			}
		}
	}()

	g, gctx := errgroup.WithContext(ctx)
	g.SetLimit(c.workers)

	for i, u := range urls {
		g.Go(func() error {
			n, err := c.fetcher.Fetch(gctx, u)
			results[i] = Result{URL: u, Bytes: n, Err: err}
			ticks <- struct{}{}
			return nil
		})
	}

	// 여기서 나오는 에러는 없다(항상 nil을 반환하므로). 형태를 위해 확인한다.
	if err := g.Wait(); err != nil {
		close(ticks)
		<-reported
		return results, err
	}

	// 워커가 전부 끝난 뒤에 닫는다. 집계 고루틴이 정리될 때까지 기다린다 —
	// 이 대기를 빼면 프로그램 종료 시점에 고루틴이 남는다.
	close(ticks)
	<-reported

	return results, ctx.Err()
}

여기서 내린 결정 다섯 개를 따로 짚는다.

1. results를 미리 만들고 인덱스로 쓴다. 7-1부터 계속 쓴 기법이다. 잠금이 없고, 순서가 보존되고, -race가 조용하다. fan-in으로 모았다면 순서를 잃었을 것이다.

2. g.Go에 넘기는 함수가 항상 nil을 반환한다. 7-8에서 본 대로 errgroup은 첫 에러가 나면 형제를 전부 취소한다. 수집기에서는 그것이 틀린 동작이다 — URL 하나가 404라고 나머지 수집을 중단할 이유가 없다. 그래서 errgroup은 SetLimit과 Wait만 쓰고 에러 전파는 쓰지 않는다.

:::tip 도구의 일부만 쓰는 것은 정당하다 errgroup을 쓴다고 에러 전파까지 받아들일 필요는 없다. 여기서 얻는 것은 동시 실행 제한 + 전원 대기 두 가지다. 반대로 "하나 실패하면 전부 중단"이 맞는 작업이라면 err를 그대로 반환하면 된다. 정책은 도메인이 정한다. :::

3. Run이 반환하는 에러는 ctx.Err()뿐이다. 개별 실패는 Result.Err에, 전체 예산 초과는 반환값에. 호출자가 둘을 구분할 수 있다.

4. 진행률은 채널 + 집계 고루틴 하나. 7-3의 tracker와 같은 구조다. Progress 콜백이 단일 고루틴에서만 호출되므로 사용자가 그 안에서 동기화를 신경 쓸 필요가 없다. atomic.Int64로 세면 카운트는 맞지만 출력 순서가 뒤엉킨다.

5. close(ticks) 다음에 <-reported. 집계 고루틴이 실제로 끝났는지 확인한다. 이 한 줄이 6번 요구사항을 지킨다.

4단계 — 배선

main은 조립과 출력만 한다(6-6).

examples/07-concurrency/09-collector/main.go

func run(name string, timeout time.Duration, list []string) {
	fmt.Printf("== %s (워커 4, 제한시간 %v, URL %d개) ==\n", name, timeout, len(list))

	c := collect.New(fetch.Sim{Unit: 5 * time.Millisecond}, 4)
	c.Progress = func(done, total int) {
		fmt.Printf("  진행 %d/%d (%d%%)\n", done, total, done*100/total)
	}

	ctx, cancel := context.WithTimeout(context.Background(), timeout)
	defer cancel()

	rs, err := c.Run(ctx, list)
	report(rs)

	s := collect.Summarize(rs)
	fmt.Printf("  성공 %d, 실패 %d, 합계 %d bytes\n", s.OK, s.Failed, s.TotalBytes)
	fmt.Println("  Run 반환 에러:", err)
	fmt.Println()
}

func main() {
	before := runtime.NumGoroutine()

	run("여유 있는 예산", 2*time.Second, urls())
	run("빠듯한 예산", 150*time.Millisecond, urls("https://example.com/slow"))

	// 고루틴 누수 검사. 모든 워커와 집계 고루틴이 정리됐으면 시작값으로 돌아온다.
	after := runtime.NumGoroutine()
	fmt.Printf("고루틴: 시작 %d → 종료 %d (누수 %d)\n", before, after, after-before)
}

에러를 분류해서 출력하는 부분에서 4-7의 errors.Is가 쓰인다.

examples/07-concurrency/09-collector/main.go

func report(rs []collect.Result) {
	for _, r := range rs {
		switch {
		case r.Err == nil:
			fmt.Printf("  OK       %-32s %4d bytes\n", r.URL, r.Bytes)
		case errors.Is(r.Err, fetch.ErrNotFound):
			fmt.Printf("  NOTFOUND %-32s\n", r.URL)
		case errors.Is(r.Err, context.DeadlineExceeded):
			fmt.Printf("  TIMEOUT  %-32s\n", r.URL)
		default:
			fmt.Printf("  ERROR    %-32s %v\n", r.URL, r.Err)
		}
	}
}

실행

cd examples/07-concurrency
go run ./09-collector

== 여유 있는 예산 (워커 4, 제한시간 2s, URL 8개) ==
  진행 2/8 (25%)
  진행 4/8 (50%)
  진행 6/8 (75%)
  진행 8/8 (100%)
  OK       https://example.com/a             210 bytes
  OK       https://example.com/bb            220 bytes
  OK       https://example.com/ccc           230 bytes
  OK       https://example.com/dddd          240 bytes
  OK       https://example.com/eeeee         250 bytes
  NOTFOUND https://example.com/missing     
  OK       https://example.com/ffffff        260 bytes
  OK       https://example.com/ggggggg       270 bytes
  성공 7, 실패 1, 합계 1680 bytes
  Run 반환 에러: <nil>

== 빠듯한 예산 (워커 4, 제한시간 150ms, URL 9개) ==
  진행 3/9 (33%)
  진행 5/9 (55%)
  진행 7/9 (77%)
  진행 9/9 (100%)
  OK       https://example.com/a             210 bytes
  OK       https://example.com/bb            220 bytes
  OK       https://example.com/ccc           230 bytes
  OK       https://example.com/dddd          240 bytes
  OK       https://example.com/eeeee         250 bytes
  NOTFOUND https://example.com/missing     
  OK       https://example.com/ffffff        260 bytes
  OK       https://example.com/ggggggg       270 bytes
  TIMEOUT  https://example.com/slow        
  성공 7, 실패 2, 합계 1680 bytes
  Run 반환 에러: context deadline exceeded

고루틴: 시작 1 → 종료 1 (누수 0)

이 출력은 완전히 결정적이다. 다섯 번 돌려 해시가 전부 같았다. 동시성 프로그램인데도 그런 이유는 지금까지 쌓아 온 세 가지 덕분이다.

1. 결과를 입력 인덱스에 쓴다 — 도착 순서가 출력에 영향을 주지 않는다.
2. 진행률을 완료 개수로만 보고한다 — 어느 URL이 먼저인지는 출력에 없다.
3. 가짜 fetcher의 지연이 URL로부터 결정되고, 타임아웃(150ms)이 빠른 작업 최대치(25ms × 4워커)와 느린 작업(500ms) 사이에 넉넉히 놓여 있다.

3번은 타이밍에 의존한다는 점을 기억한다. 극단적으로 부하가 걸린 기계에서는 빠른 작업도 잘릴 수 있다. 두 번째 실행의 33/55/77% 같은 어색한 숫자는 URL이 9개라서 25% 구간이 정수로 안 떨어지는 것이고, 이것도 결정적이다.

검증

1. race detector

go run -race ./09-collector

경고 없이 끝나고 종료 코드는 0이다. results[i]를 잠금 없이 쓰는데도 조용한 이유는 각 고루틴이 서로 다른 인덱스에만 쓰기 때문이다. 같은 슬라이스라도 다른 원소는 다른 메모리 위치이므로 7-6의 정의상 경합이 아니다.

2. 고루틴 누수

마지막 줄의 누수 0이 그 검사다. 실제로 새는지 확인하려면 Run의 마지막 <-reported를 빼고 돌려 보면 된다.

고루틴: 시작 1 → 종료 3 (누수 2)

두 개가 남는다. run을 두 번 불렀으므로 집계 고루틴이 하나씩 남은 것이다. close(ticks)로 for range ticks는 끝나지만, 끝났는지 확인하지 않고 main이 진행했기 때문에 그 시점의 스냅숏에 잡힌다.

호출 한 번에 고루틴 하나. 서버라면 요청 하나에 하나씩 쌓인다. 이런 종류가 실제 프로덕션 누수의 대부분이다.

3. 데드라인이 진짜 지켜지는가

빠듯한 예산 실행의 /slow는 500ms가 필요하지만 150ms에 잘렸다. Sim.Fetch의 select가 없었다면 Run이 500ms 넘게 걸렸을 것이다.

여기서 쓰인 것들

챕터	무엇
7-1	인덱스로 결과 자리 잡기, NumGoroutine으로 누수 확인
7-2	chan struct{} 신호, close로 range 끝내기
7-3	소유권 — ticks를 만든 쪽이 닫는다, 상태를 독점하는 집계 고루틴
7-4	타임아웃 select, NewTimer + defer Stop
7-5	뮤텍스를 쓰지 않기로 한 결정
7-6	-race 검증
7-7	예산 전파, ctx.Done() 존중, defer cancel()
7-8	errgroup.SetLimit, 결과 순서 보존, 의존성 주입
6-5	소비자 쪽 인터페이스, internal/

정리

• 경계를 인터페이스로 긋는다. collect가 fetch를 모르기 때문에 나중에 진짜 HTTP 구현으로 갈아 끼울 수 있다. 그것이 예제를 오프라인으로 유지하면서도 현실적으로 만드는 방법이다.
• 동시 실행 제한 + 전체 예산 + 개별 실패 허용은 서로 다른 세 가지 정책이고, 각각 다른 장치로 구현된다. SetLimit, context.WithTimeout, Result.Err.
• 결과 순서 보존은 fan-in을 쓰지 않는 것으로 얻는다. 인덱스에 쓰면 공짜다.
• 끝났음을 확인하는 코드가 없으면 그것이 누수다. close는 신호일 뿐, 상대가 정리를 마쳤다는 뜻이 아니다.
• 동시성 프로그램도 결정적인 출력을 낼 수 있다. 그렇게 설계해야 디버깅과 테스트가 가능하다.

연습문제

1. 재시도. Fetch가 ErrNotFound가 아닌 에러를 냈을 때 최대 2번까지 다시 시도하도록 collect를 고쳐 보자. 재시도 사이에 대기를 넣는다면 그 대기도 ctx를 존중해야 한다. time.Sleep을 쓰면 왜 안 되는가?

2. 부분 결과. 지금은 예산이 끝나면 Run이 context.DeadlineExceeded를 반환하지만 결과 슬라이스는 그대로 준다. 호출자가 "성공한 것만 쓰고 나머지는 나중에 다시" 하려면 어떤 정보가 더 필요한가? Result에 무엇을 추가하겠는가?

3. errgroup 없이. errgroup을 걷어내고 sync.WaitGroup + 세마포어 (7-3)로 같은 동작을 만들어 보자. 몇 줄이 늘어나는가? errgroup이 대신해 주던 것이 정확히 무엇이었는지 목록으로 적어 보자.

4. 누수 만들기. Sim.Fetch의 select에서 case <-ctx.Done()을 지우고 빠듯한 예산 실행만 남겨 보자. 프로그램은 언제 끝나는가? NumGoroutine은 무엇을 보고하는가? Run이 반환한 뒤에도 워커가 살아 있을 수 있다는 사실이 왜 위험한지 설명해 보자.

5. 워커 수 실험. New(..., workers)를 1, 2, 4, 8, 16으로 바꿔 가며 여유 있는 예산 실행의 전체 소요 시간을 재 보자 (time go run ./09-collector). 어디서부터 나아지지 않는가? URL이 8개인데 워커 16이 의미가 있는가?

6. 진짜 HTTP로 (파트 9 이후). net/http를 배운 뒤 Fetcher를 구현하는 httpfetch 패키지를 추가해 보자. collect 패키지는 한 줄도 바뀌지 않아야 한다. 바뀐다면 경계를 잘못 그은 것이다.