처음배우는엘릭서프로그래밍

10.2 Stream: 지연 계산하기

엘릭서에서 Enum 모듈은 소위 ‘탐욕스러운’ 연산을 한다. 즉, 컬렉션을 전달하면 그 컬렉셔느이 모든 값들에 연산을 수행한다. 그리고 수행한 결과는 일반적으로 새로운 컬렉션이 된다. 다음 파이프라인 예제를 보자.

[1, 2, 3, 4, 5]
|> Enum.map(&(&1 * &1))
# [1, 4, 9, 16, 25]
|> Enum.with_index()
# [{1, 0}, {4, 1}, {9, 2}, {16, 3}, {25, 4}]
|> Enum.map(fn {value, index} -> value - index end)
# [1, 3, 7, 13, 21]
|> IO.inspect()

처음 나오는 map 함수는 원본 리스트를 받아 각 값을 제곱한 새로운 리스트를 만든다. with_index는 이 리스트를 받아 {값, 인덱스} 튜플 리스트를 반환하며, 이어지는 map 함수는 각 항목의 값에서 인덱스를 빼어 IO.inspect에 전달한다. 즉 이 파이프라인은 최종 결과물을 만들어내기까지 리스트를 총 4개 생성한다. 다른 예를 하나 들어보자. 다음 파일로부터 내용을 읽어 가장 길이가 긴 줄의 내용을 반환하는 코드다.

IO.inspect("#{__DIR__}/files/71k-English-uncleaned-wordlist/71k English wordlist.txt")

IO.inspect(
  File.read!("#{__DIR__}/files/71k-English-uncleaned-wordlist/71k English wordlist.txt")
  |> String.split()
  |> Enum.max_by(&String.length/1)
)

~예시에서는 사전에 수록된 단어 목록을 사용했는데.~ 현재 예시는 huggingface.co/datasets에서 가져온 데이터로 테스트했다. 가장 긴 단어를 찾기 위해 데이터 전체를 메모리에 올리고 리스트로 만들었다.

앞서 . 본두가지 예시 모두 답을 찾을 . 수있으나 이상적인 풀이는 아니다. Enum 모듈의 함수들은 스스로 완결성을 가지기 때문이다. 각 호출이 컬렉션을 받아 다른 완성된 컬렉션을 반환한다. 대신 컬렉션의 각 항목을 필요할 때만 가져와 처리할 . 수있다면 좋을 것이다. 그러면 연산 도중 결과를 완성된 컬렉션으로 갖고 있을 필요 없이, 원본 컬렉션의 각 값에 함수를 차례로 적용하기만 하면 된다. 스트림이 바로 . 그 역활을 한다.

10.2.1 스트림 조합하기

스트림을 만드는 간단한 예를 하나 보자.

Stream.map([1, 3, 5, 7], &(&1 + 1))

같은 리스트에 Enum.map을 호출하면 [2, 4, 6, 8]이라는 결과를 즉시 얻을 . 수있다. 위 예제에서는 대신 스트림 값이 반환되는데, 이. 스트림에는 컬렉션에 수행해야 할 연산의 명세가 포함된다. 그렇다면 스트림에서 실제 결과를 받아오려면 어떻게 해야할까? 이 스트림을 컬렉션으로 생각하고 Enum 모듈의 함수에 넣으면 된다.

s = Stream.map([1, 3, 5, 7], &(&1 + 1))
Enum.to_list(s)

스트림 역시 Enummerable 프로토콜을 구현한 타입이므로 Stream 모듈의 함수에 컬렉션 대신 스트릠을 전달할 수도 있다. 따라서 스트림을 조합해 사용할 수 있다.

squares = Stream.map([1, 2, 3, 4], &(&1 * &1))
plus_ones = Stream.map(squares, &(&1 + 1))
odds = Stream.filter(plus_ones, fn x -> rem(x, 2) == 1 end)
Enum.to_list(odds)

물론 이코드는 다음과 같이 정리할 수 있다.

[1, 2, 3, 4]
|> Stream.map(&(&1 * &1))
|> Stream.map(&(&1 + 1))
|> Stream.filter(fn x -> rem(x, 2) == 1 end)
|> Enum.to_list()

스트림은 중간 결과를 담은 리스트를 만들지 않는다. 그저 원본 컬렉션의 각 값을 연속된 연산에 차례로 태울 뿐이다. 앞에 출력된 스트림 값이 스트림의 동작을 이해하는 실마리가 된다. 조합된 스트림은 함수의 리스트로 표현되어, 실제로 연산이 수행될 때컬렉션의 각 항목에 차례로 적용된다.

리스트에만 스트림을 사용할 수 있는 것은 아니다. 많은 엘릭서 모듈이 스트림을 지원한다. 가장 긴 단어를 찾는 예제를 스트림을 이용해 다음처럼 고칠 수 있다.

File.open!("#{__DIR__}/files/71k-English-uncleaned-wordlist/71k English wordlist.txt")
|> IO.stream(:line)
|> Enum.max_by(&String.length/1)

이 코드의 핵심은 IO.stream 함수다. 이 함수는 IO 디바이스(열린 파일)를 줄 단위로 값을 제공하는 스트림으로 바꿔준다. 이런 구현 방식은 대단히 유용해서, 두 함수를 줄인 단축 함수도 제공된다.

File.stream!("#{__DIR__}/files/71k-English-uncleaned-wordlist/71k English wordlist.txt")
|> Enum.max_by(&String.length/1)

중간 결괏값을 저장하지 않는다는 점은 장점이지만 실행 시간은 원래 구현보다 두 배 정도 느려진다. 하지만 데이터가 다른 서버나 외부 센서(예를 들면 온도계)로 부터 올 때는 괜찮은 구현이다. 데이터가 천천히 들어오거나 끝없이 들어올 수도 있기 때문이다. Enum으로 구현하면 데이터가 모두 들어올 때까지 기다려야 하지만, 스트림을 이용하면 데이터가 들어올 때마다 처리할 수 있다.

10.2.2 무한 스트림

스트림은 지연 계산을 하므로 입력이 미리 준비되어 있지 않아도 괜찮다. 다음 예를 보자.

Enum.map(1..10_000_000, &(&1 + 1)) |> Enum.take(5)

이 구현은 결과가 출력될 때까지 약 8초가 걸린다(내 서버 기준 4초). 값 천만 개짜리 리스트를 만들고 거기에서 처음 다섯 개의 값을 가져오기 때문이다. 코드를 다음과 같이 고쳐보자.

Stream.map(1..10_000_000, &(&1 + 1)) |> Enum.take(5)

이 경우 결괏값이 즉시 반환된다. take는 5개의 값만 있으면 되므로, 스트림에서 5개의 값만 받아오면 그 뒤를 추가로 계산할 필요가 없다.

이 예제에서는 스트림이 받는 컬렉션이 유한했지만, 무한히 많은 값을 받을 수도 있다. 무한히 큰 컬렉션을 만들려면 함수를 이용해 스트림을 만들어주저야 한다.

10.2.3 직접 스트림 만들기

스트림은 엘릭서 라이브러리 중에서 홀로 독립되어 있는 편인데, 런타임에서 특별히 지원해주는 부분이 없기 때문이다. 그렇다고 스트림 타입을 직접 만들기 위해 아주 낮은 수준까지 내려가야 하는 것은 아니다. 물론 실제 구현은 복잡하지만(문자열이나 날짜 처리 이론이 복잡한 것과 마찬가지다) 이를 추상화한 몇 가지 함수가 수고를 덜어준다. 그 함수는 cycle repeatedly iterate unfold resource등이 있다(함수 이름만 봐서는 마지막 두 함수가 어떤 역할을 하는지 도저히 알기 어렵듯이 내부 구현은 매우 복잡하다). 일단 가장 간단한 함수 세 개부터 시작해보자. 바로 cycle repeatedly iterate이다.

Stream.resource

스트림과 외부에 있는 리소스가 어떻게 상호작용하는지 궁금해질 즘이다. 10.2.1 스트림 조합하기에서 파일 내용을 줄 단위 스트림으로 만드는 코드를 봤는데, 같은 것을 직접 구현해서 사용하려면 어떻게 해야 할까? 스트림이 시작될 때 파일을 열고, 내용을 줄 단위로 반환하고, 다 읽어들이고 나면 파일을 닫아야 한다. 데이터베이스를 사용한다면 커서 방식 페이징을 이용해 값을 차례대로 가져와 스트림으로 바꿔야 한다. 스트림이 시작될 떄 쿼리를 수행해 각 레코드를 스트림으로 내보내고, 데이터를 모두 가져오고 나면 쿼리를 종료하게 된다. 바로 이 작업에 Stream.resource를 사용한다.

Stream.resource는 Stream.unfold를 기반으로 하지만 두 가지 차이가 있다. unfold의 첫 번째 인자는 반복 함수를 처음 호출하는할 때 전달할 값이 었다. 하지만 그 값이 어떤 자원이라면 스트림이 실제로 값을 내보내기 전까지는 그 자원에 접근하지 않는 것이 좋다. 스트림이 만들어지더라도 실제로 값을 내보내기까지 시간이 오래 걸릴 수도 있다. 따라서 resource는 값이 아니라 값을 반환하는 함수를 받는다. 이것이 첫 번째 차이점이다.

두 번째 차이는 리소스를 모두 가져온 뒤 스트림을 종료할 때의 처리와 관련 있다. 이를 위해 resource 함수는 세 번째 인자를 받는다. 세 번째 인자로 전달되는 함수는 마지막 상태를 받아 자원 할당을 해제하는 데 필요한 처리를 수행한다.

라이브러리 문서에도 소개된 예제 코드는 다음과 같다.

Stream.resource(
  fn ->
    File.open!("#{__DIR__}/files/71k-English-uncleaned-wordlist/71k English wordlist.txt")
  end,
  fn file ->
    case IO.read(file, :line) do
      data when is_binary(data) -> {[data], file}
      _ -> {:halt, file}
    end
  end,
  fn file -> File.close(file) end
)

스트림이 활성화될 때 첫 번째 함수가 파일을 열어 파일 정보(파일에 접근하도록 해주는 IO 디바이스)를 두 번째 함수에 전달한다. 두 번째 함수는파일을 줄 단위로 읽어 그 내용과 파일 정보로 이루어진 튜플을 반환하고, 파일의 끝에 도달하면 :halt 튜플을 반환한다. 세 번째 함수에서 파일을 닫는다.

마지막으로 지금까지와는 다른 종류의 리소스인 ‘시간’을 생각해보자. Stream.resource를 사용해 분 단위가 바뀌기까지 남은 초를 단위로 세는 카운트다운 타이머를 구현해보려 한다. 자원을 할당받기 위한 첫 번째 함수에서는 남은 시간을 초 단위로 반환한다. 이 함수는 스트림이 실제로 계산될 때마다 실행되므로, 반환하는 카운트다운 값은 언제 호출되는지에 따라 다르다.

반복 호출되는 두 번째 함수에서는 남은 시간을 관리한다. 정각이면 {:halt, 0} 튜플을 반환하고, 정각이 아니면 1초간 기다린 후 최신 카운트다운 값을 담은 문자열과 갱신된 카운터를 반환한다. 이 예제에서는 리소스 할당을 해제할 필요가 없으므로 세 번째 함수는 아무 동작도 하지 않는다. 코드로 구현하면 다음과 같다.

defmodule Countdown do
  def sleep(seconds) do
    receive do
    after
      seconds * 1000 -> nil
    end
  end

  def say(text) do
    spawn(fn -> IO.puts(~c"say #{text}") end)
  end

  def timer do
    Stream.resource(
      # 분 단위가 바뀔 때까지 남은 초를 계산한다.
      fn ->
        {_h, _m, s} = :erlang.time()
        60 - s - 1
      end,
      # 1초 뒤 카운트다운 값을 반환한다.
      fn
        0 ->
          {:halt, 0}

        count ->
          sleep(1)
          {[inspect(count)], count - 1}
      end,
      # 할당 해제할 자원 없음
      fn _ -> nil end
    )
  end
end

counter = Countdown.timer()
# printer = counter |> Stream.each(&IO.puts/1)
# speaker = printer |> Stream.each(&Countdown.say/1)
speaker = counter |> Stream.each(&Countdown.say/1)
speaker |> Enum.take(5)

시간의 변화를 리소스로 만들어 카운트다운 값을 화면에 출력하고, 이것을 읽어주는 스트림을 만들었다. 잘 동작하낟. 시계가 22초를 가리킬 때 실행한 결과이므로 카운트다운이 37에서 시작한다. 같은 스트림을 몇 초후에 다시 실행해보자.

iex> speaker | Enum.to_list
6
5
4
3
2
1
["6", "5", "4", "3", "2", "1"]

이 코드는 좋은 구현은 아니다. 코드가 실행되면서 생기는 지연을 sleep 함수에서 보정하지 못하기 떄문이다. 하지만 스트림을 이용하면 비동기적인 리소스를 쉽게 다룰 수 있다는 점은 배울 만하다. 또 스트림은 사용할 때마다 초기화되므로 사이드 이펙트를 일으실 염려가 없다는 점 역시 알아두면 좋다. 스트림을 파이프로 Enum 함수와 연결할 때마다 새롭게 계산된 따끈따끈한 값을 받을 수 있다.

10.2.4 스트림의 실용성

함수형 프로그래밍이 문제를 새로운 관점에서 바라보게 했듯이, 스트림을 알고 나면 반복이나 컬렉션을 새로운 관점에서 보게 된다. 물론 반복이 필요한 상황에 언제나 스트림을 사용해야 하는 것은 아니다. 다만 실제 값이 필요해질 때까지 연산을 미루고 싶거나 큰 데이터를 굳이 한꺼번에 처리할 필요가 없다면 스트림 사용을 고려해보자.