태스크 에서

태스크 소개

태스크는 하나의 특정 작업을 실행하기 위해 만들어진 프로세스예요. 보통 다른 프로세스와 통신하지 않지만, 태스크를 시작한 프로세스에는 결과를 반환할 수 있어요.

태스크는 작업을 병렬로 처리할 때 흔히 사용해요.

async/await

태스크를 시작하려면 Task.async/1을 사용해요. 이 함수는 익명 함수를 인자로 받아서, 호출한 프로세스에 연결된 새로운 프로세스에서 실행해요. 그리고 %Task{} 구조체를 반환해요.

실행 결과를 받으려면 %Task{} 구조체를 Task.await/2에 넘겨요. 그러면 태스크가 끝날 때까지 기다렸다가 결과를 반환해요. 두 번째 인자는 밀리초 단위의 타임아웃이고, 기본값은 5000이에요. 타임아웃으로 :infinity를 넘기면 태스크가 타임아웃 없이 실행돼요.

task = Task.async(fn -> Enum.sum(0..999_999) end)
# => %Task{
#      owner: #PID<0.110.0>,
#      pid: #PID<0.119.0>,
#      ref: #Reference<0.3715954046.815529991.83334>
#    }

Task.await(task)
# => 499999500000

태스크를 시작한 뒤 기다리기 전까지는, 태스크를 시작한 프로세스가 블록되지 않고 다른 작업을 할 수도 있다는 점을 기억해 둬요.

Task.async/1로 시작한 태스크는 반드시 기다려야 해요. 그 태스크가 호출한 프로세스로 메시지를 보내기 때문이에요. Task.await/2는 태스크마다 한 번만 호출할 수 있어요.

start/start_link

부수 효과만을 위해 태스크를 시작하고 싶다면 Task.start/1이나 Task.start_link/1을 사용해요. Task.start/1은 호출한 프로세스에 연결되지 않은 태스크를 시작하고, Task.start_link/1은 호출한 프로세스에 연결된 태스크를 시작해요. 두 함수 모두 {:ok, pid} 튜플을 반환해요.

타임아웃

비동기 태스크를 시작한 다음 Enum.map/2로 기다리는 것은 흔한 패턴이에요. 예를 들어 이렇게요:

inputs
|> Enum.map(&Task.async(fn -> function.(&1) end))
|> Enum.map(&Task.await(&1, 1_000))

Task.await는 동기 방식이라는 점을 알아 두는 게 중요해요. 호출될 때마다 타임아웃 시간을 0부터 세기 시작하기 때문이에요. 즉, 이렇게 순차적으로 사용하면 타임아웃이 서로 더해진다는 뜻이에요.

처음 기다리는 태스크는 타임아웃보다 오래 실행되지 않지만, 두 번째로 기다리는 태스크는 첫 번째 태스크가 실행된 시간에 타임아웃을 더한 만큼 실행될 수 있어요.

각 태스크가 이전 태스크보다 조금씩 오래 걸리지만, 태스크 실행 시간의 차이는 기다리는 타임아웃보다 짧은 상황을 생각해 봐요.

inputs = 1..10
function = fn n -> :timer.sleep(n * 900) && n end

이 태스크들이 얼마나 오래 실행되는지는 Erlang 함수인 :timer.tc/1로 측정할 수 있어요. 단위는 마이크로초예요.

:timer.tc(fn ->
  inputs
  |> Enum.map(&Task.async(fn -> function.(&1) end))
  |> Enum.map(&Task.await(&1, 1_000))
end)

# => {9004517, [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]}

9004517마이크로초는 9초예요. 주어진 타임아웃인 1초(1000밀리초)보다 훨씬 길죠.

이런 동작을 원한다면, 좋아요! 그대로 해봐요. 아니라면 Task.async_stream/3이나 Task.await_many/2를 사용하는 방법이 있어요.

Task.async_stream/3

Task.async_stream/3을 사용하면 같은 함수를 여러 입력에 대해 각각 별도의 태스크에서 실행할 수 있어요. 이때 태스크들은 Task.async/1로 실행할 때와 마찬가지로 현재 프로세스에 연결돼요. 스트림을 반환해요.

각 태스크에 개별적으로 적용되는 타임아웃을 받아요. 앞과 같은 입력과 함수를 Task.async_stream/3에 사용한 경우를 생각해 봐요:

inputs
|> Task.async_stream(function, timeout: 1_000)
|> Enum.to_list()

# => ** (exit) exited in: Task.Supervised.stream(1000)
#       ** (EXIT) time out
#       (elixir 1.11.3) lib/task/supervised.ex:304: Task.Supervised.stream_reduce/7
#       (elixir 1.11.3) lib/enum.ex:3473: Enum.reverse/1
#       (elixir 1.11.3) lib/enum.ex:3066: Enum.to_list/1

하나의 함수를 여러 입력에 대해 실행해야 한다면 Task.async_stream/3이 아주 좋아요. 동시에 실행할 태스크 수나 태스크가 타임아웃됐을 때 어떻게 할지 같은 유용한 옵션을 받아서 세밀하게 제어할 수 있어요.

서로 다른 작업을 수행하는 태스크를 실행해야 한다면, 대신 Task.await_many/2가 필요할 수도 있어요.

Task.await_many/2

Task.await_many/2는 Elixir 1.11에서 도입됐어요. 여러 태스크를 하나의 공유된 타임아웃으로 기다릴 수 있게 해줘요. 앞과 같은 입력과 함수를 Task.await_many/2에 사용한 경우를 생각해 봐요:

inputs
|> Enum.map(&Task.async(fn -> function.(&1) end))
|> Task.await_many(1_000)

# => ** (exit) exited in: Task.await_many([...], 1000)
#        ** (EXIT) time out
#        (elixir 1.11.3) lib/task.ex:725: Task.await_many/5
#        (elixir 1.11.3) lib/task.ex:709: Task.await_many/2