[Go] Concurrency: Pipeline pattern

파이프라인 패턴은 List 형태의 벡터 데이터를 연속적으로 가공할 때 사용하기 적당한 동시성 처리 패턴이다.

구현 방식은 조금씩 다르지만 이건 다른 언어들에서도 범용적으로 쓰이는 개념이다. Rust의 async Stream, Node.js의 Stream 등과도 유사하다.

기본적인 원리는 이렇다.
배열이란 데이터의 개념을 "채널"로서 감싸고, 그걸 통해서 연쇄적으로 chaining 호출을 할 수 있도록 하는 것이다.

먼저 파이프라인 채널을 생성하는 generator 함수가 있어야 한다.

가변인자를 받아서 그걸 기반으로 채널을 만들고 반환해주도록 했다.
물론 이 채널의 완성은 비동기적으로 이루어진다. 그래도 괜찮다.

다음과 같이 파이프라인 스트림을 초기화할 수 있다.

이 다음에는 파이프라인 연산을 구현해야 한다.
여기서는 덧셈과 곱셈 연산 2가지를 만들어보겠다.

둘이 구조나 원리는 같다.
기존 스트림 채널을 파라미터로 전달받고, 그걸 뺑뺑 돌면서 새 스트림을 생성해서 반환한다.

이 또한 처리가 비동기적으로 완성되나, 어차피 채널을 통해 핸들링하므로 신경쓸 것이 없다.

그러면 이제 아래와 같은 형태로 스트림을 넣고, 다시 반환받고, 넣고 반환받는 식으로 연쇄적으로 병렬 호출을 이어나갈 수 있다.

이러한 형태로 파이프라이닝 비스무리하게 할 수 있기 때문에 파이프라인 패턴이라 부르는 것이다.
각 step은 완벽하게 동시적으로 실행되며 각 요소의 순서 또한 보장된다.

실행해보면

기대한 대로 동작한다.
곱셈 2가 수행된 이후에 덧셈이 잘 붙어있다.

물론 이 예제처럼 간단한 연산만을 한다면 이걸 쓰는 의미가 없다. 그냥 채널 없이 동기적으로 실행하는 것이 훨씬 빠를 것이다. 이건 각 요소별 생성시간이나 연산시간이 길어져서 대기시간이 늘어질 수 있을 때 유용하다.

아래는 전체 코드다.

package main

func gererateStream[T any](done <-chan struct{}, values ...T) <-chan T {
	stream := make(chan T)
	go func() {
		defer close(stream)
		for _, value := range values {
			select {
			case <-done:
				return
			case stream <- value:
			}
		}
	}()
	return stream
}

func multiply(done <-chan struct{}, valueStream <-chan int, multiplier int) <-chan int {
	multipliedStream := make(chan int)
	go func() {
		defer close(multipliedStream)
		for value := range valueStream {
			select {
			case <-done:
				return
			case multipliedStream <- value * multiplier:
			}
		}
	}()
	return multipliedStream
}

func add(done <-chan struct{}, valueStream <-chan int, additive int) <-chan int {
	addedStream := make(chan int)
	go func() {
		defer close(addedStream)
		for value := range valueStream {
			select {
			case <-done:
				return
			case addedStream <- value + additive:
			}
		}
	}()
	return addedStream
}

func main() {
	done := make(chan struct{})
	defer close(done)

	intStream := gererateStream(done, 1, 2, 3, 4, 5)

	multipliedStream := multiply(done, intStream, 2)
	addedStream := add(done, multipliedStream, 1)

	for result := range addedStream {
		println(result)
	}
}