Task와 gen_tcp

Task와 gen_tcp (Task and gen_tcp)

이번 장에서는 Erlang의 :gen_tcp 모듈로 요청을 처리하는 서버를 만드는 법을 배워요. 그러면서 Elixir의 Task 모듈을 깊이 있게 다뤄요. 이후 장에서 서버가 실제로 버킷과 상호작용하도록 확장할 거예요.

출처: Task and gen_tcp

본문

에코 서버(Echo server)

우리는 TCP 서버를 에코 서버(echo server) 부터 시작해요. 에코 서버는 요청에서 받은 텍스트를 그대로 응답으로 보내요. 이 서버를 조금씩 개선해서, 감독되고 여러 연결을 처리할 수 있게 만들 거예요.

TCP 서버는 대략 다음 단계를 수행해요.

  1. 소켓을 얻을 때까지 포트를 리슨(listen)
  2. 그 포트에서 클라이언트 연결을 기다렸다가 수락(accept)
  3. 클라이언트 요청을 읽고 응답을 씀

lib/kv/server.ex를 만들고 다음 함수들을 추가해요.

defmodule KV.Server do
  require Logger

  def accept(port) do
    # The options below mean:
    #
    # 1. `:binary` - receives data as binaries (instead of lists)
    # 2. `packet: :line` - receives data line by line
    # 3. `active: false` - blocks on `:gen_tcp.recv/2` until data is available
    # 4. `reuseaddr: true` - allows us to reuse the address if the listener crashes
    #
    {:ok, socket} =
      :gen_tcp.listen(port, [:binary, packet: :line, active: false, reuseaddr: true])
    Logger.info("Accepting connections on port #{port}")
    loop_acceptor(socket)
  end

  defp loop_acceptor(socket) do
    {:ok, client} = :gen_tcp.accept(socket)
    serve(client)
    loop_acceptor(socket)
  end

  defp serve(socket) do
    socket
    |> read_line()
    |> write_line(socket)

    serve(socket)
  end

  defp read_line(socket) do
    {:ok, data} = :gen_tcp.recv(socket, 0)
    data
  end

  defp write_line(line, socket) do
    :gen_tcp.send(socket, line)
  end
end

서버는 KV.Server.accept(4040)(4040 포트)으로 시작해요. accept/1의 첫 단계는 :gen_tcp.listen/2로 포트를 리슨하고 loop_acceptor/1을 호출하는 거예요. loop_acceptor/1은 클라이언트 연결을 계속 받아들이는 루프이고, 수락한 연결마다 serve/1을 호출해요.

serve/1은 소켓에서 한 줄을 읽고 그 줄을 다시 소켓에 쓰는 루프예요. 이 함수는 파이프 연산자 |>/2 를 사용해서 데이터 흐름을 표현해요. 파이프 연산자는 왼쪽을 평가해서 그 결과를 오른쪽 함수의 첫 인자로 넘겨요.

socket |> read_line() |> write_line(socket)

는 다음과 같아요.

write_line(read_line(socket), socket)

read_line/1은 :gen_tcp.recv/2로 소켓에서 데이터를 받고, write_line/2는 :gen_tcp.send/2로 소켓에 써요.

주의: serve/1은 loop_acceptor/1 안에서 순차적으로 호출되는 무한 루프라서, 뒤의 loop_acceptor/1 꼬리 호출은 사실 도달하지 않아요. 하지만 곧 serve/1을 별도 프로세스에서 실행해야 할 거라서 그 꼬리 호출이 필요해질 거예요.

이제 iex -S mix로 세션을 시작하고 KV.Server.accept(4040)을 실행하면 서버가 돌아가고 콘솔이 블로킹되는 걸 볼 수 있어요. telnet 127.0.0.1 4040으로 접속해서 "hello"를 치면 "hello"가 그대로 돌아와요.

telnet을 종료하면 IEx 세션에 오류가 보일 거예요.

** (MatchError) no match of right hand side value: {:error, :closed}

이건 :gen_tcp.recv/2에서 데이터를 기대했는데 클라이언트가 연결을 닫았기 때문이에요. 이런 케이스를 더 잘 처리해야 해요.

하지만 지금은 더 중요한 버그가 있어요. TCP acceptor가 죽으면 어떻게 될까요? 감독이 없으니 서버가 죽고, 재시작되지 않아서 더는 요청을 처리하지 못해요. 그래서 서버를 감독 트리로 옮겨야 해요.

태스크(Tasks)

이미 있는 함수를 애플리케이션 시작 시점에 그냥 실행하고 싶을 때 딱 맞는 모듈이 Task예요. Task.start_link/1은 익명 함수를 받아서 새 프로세스에서 실행하고, 그 프로세스는 감독 트리의 일부가 돼요.

lib/kv.ex의 start/2에 새 자식을 추가해요.

def start(_type, _args) do
  children = [
    {Registry, name: KV, keys: :unique},
    {DynamicSupervisor, name: KV.BucketSupervisor, strategy: :one_for_one},
    {Task, fn -> KV.Server.accept(4040) end}
  ]
  Supervisor.start_link(children, strategy: :one_for_one)
end

KV.Server.accept(4040)을 태스크로 실행하겠다는 뜻이에요. 포트는 지금 하드코딩하고 나중에 설정으로 만들 거예요. 자식 스펙으로 2-튜플을 넘기면 Task.start_link/1이 호출돼요.

이제 서버가 감독 트리의 일부라 애플리케이션 실행 시 자동으로 시작돼요. 그런데 한 번에 하나의 클라이언트만 처리할 수 있어요. 두 telnet 클라이언트를 동시에 열면 두 번째가 에코되지 않아요. 그 이유는 연결을 수락하는 같은 프로세스에서 요청을 처리하기 때문이에요 — 한 클라이언트가 연결되어 있으면 다른 클라이언트를 수락할 수 없거든요.

(결함 있는) 동시성 추가하기

동시 연결을 처리하려면, 한 프로세스가 acceptor로 동작하면서 다른 프로세스를 스폰해 요청을 처리해야 해요. loop_acceptor/1을 Task.start_link/1을 쓰도록 바꿔요.

defp loop_acceptor(socket) do
  {:ok, client} = :gen_tcp.accept(socket)
  {:ok, pid} = Task.start_link(fn -> serve(client) end)
  :ok = :gen_tcp.controlling_process(client, pid)
  loop_acceptor(socket)
end

이제 새 클라이언트가 올 때마다 새 태스크를 시작해요. 그런데 한 telnet 세션을 종료하면 다른 세션도 죽어요! 이유는 두 가지예요.

  1. :gen_tcp.recv/2가 {:error, :closed} 튜플을 반환할 거라 예상하지 못하는 서버 버그
  2. 각 서버 태스크가 acceptor 프로세스와 링크되어 있어서, 한 태스크가 죽으면 acceptor 프로세스도 죽고 모든 다른 태스크와 클라이언트까지 끌어내려요

가이드 전체의 중요한 규칙은 항상 프로세스를 슈퍼바이저의 자식으로 시작하라는 거예요. 시스템의 각 부분을 격리하지 않으면 장애가 시스템 전체로 연쇄(cascade)돼요. 이를 고치려면 DynamicSupervisor를 쓸 수도 있지만, 태스크에는 태스크 감독에 최적화된 Task.Supervisor 가 있어요.

태스크 슈퍼바이저 추가하기

lib/kv.ex의 start/2를 다시 바꿔 감독 트리에 태스크 슈퍼바이저를 추가해요.

def start(_type, _args) do
  children = [
    {Registry, name: KV, keys: :unique},
    {DynamicSupervisor, name: KV.BucketSupervisor, strategy: :one_for_one},
    {Task.Supervisor, name: KV.ServerSupervisor},
    {Task, fn -> KV.Server.accept(4040) end}
  ]
  Supervisor.start_link(children, strategy: :one_for_one)
end

이름은 KV.ServerSupervisor로 지정했어요. 자식이 시작되는 순서가 중요해요. acceptor는 마지막에 와야 해요. 앞에 오면 Task.Supervisor가 돌기 전이나 버킷을 찾기 전에 요청을 받아들이기 시작할 수 있거든요.

이제 loop_acceptor/1을 Task.Supervisor를 쓰도록 바꿔요.

defp loop_acceptor(socket) do
  {:ok, client} = :gen_tcp.accept(socket)
  {:ok, pid} = Task.Supervisor.start_child(KV.ServerSupervisor, fn -> serve(client) end)
  :ok = :gen_tcp.controlling_process(client, pid)
  loop_acceptor(socket)
end

추가된 줄 :ok = :gen_tcp.controlling_process(client, pid)은 자식 프로세스를 클라이언트 소켓의 제어 프로세스(controlling process) 로 만드는 역할을 해요. 이걸 안 하면 소켓이 기본적으로 수락한 프로세스(acceptor)에 묶이므로, acceptor가 죽으면 모든 클라이언트가 함께 죽어요.

이제 여러 동시 telnet 클라이언트를 열어도, 한 클라이언트가 나가도 acceptor가 죽지 않아요. (아직 :gen_tcp.recv/2 버그는 다음 장에서 고칠 거예요.)

재시작 전략(Restart strategies)

아직 깊이 다루지 않은 중요한 주제가 있어요. 감독되는 프로세스가 죽으면 어떻게 될까요?

프로세스가 종료될 때 정확히 무슨 일이 일어나는지는 자식 스펙(child specification) 에 담겨 있어요. KV.Bucket.child_spec/1을 보면 restart 키가 없어서 기본값인 :permanent예요. 반면 Task.child_spec/1은 restart: :temporary예요.

  • :permanent — 종료 이유와 무관하게 항상 재시작
  • :temporary — 죽은 이유와 무관하게 절대 재시작하지 않음
  • :transient — 정상 종료하면 재시작하지 않고, 비정상 종료 시에만 재시작

KV.Bucket에 :permanent는 논리적이에요 — 사용자가 만든 버킷을 다시 만들도록 요구하지 않아야 하니까요. 하지만 태스크는 이번 장에서 정반대 방식으로 두 번 사용했으니, 적어도 하나는 틀렸을 거예요.

  • acceptor 태스크 — 인프라의 핵심 구성 요소예요. 죽으면 더 이상 요청을 받지 못하고 서버가 무용지물이 돼요. → :permanent로 돌아야 해요.
  • Task.Supervisor로 시작하는 연결 처리 태스크 — 죽은 이유가 그냥 연결 문제일 수 있고, 같은 연결 위에서 재시작하면 또 실패할 뿐이에요. → :temporary 유지가 맞아요.

Elixir는 기존 자식 스펙을 바꾸는 API를 제공해요.

def start(_type, _args) do
  children = [
    {Registry, name: KV, keys: :unique},
    {DynamicSupervisor, name: KV.BucketSupervisor, strategy: :one_for_one},
    {Task.Supervisor, name: KV.ServerSupervisor},
    Supervisor.child_spec({Task, fn -> KV.Server.accept(4040) end}, restart: :permanent)
  ]
  Supervisor.start_link(children, strategy: :one_for_one)
end

이제 항상 실행되는 acceptor와, 그 아래에서 시작되는 임시 태스크가, 항상 실행되는 태스크 슈퍼바이저 아래에서 돌아요.

생태계 활용하기(Leveraging the ecosystem)

이번 장에서 동시성과 장애 허용을 탐구하면서 기본적인 TCP acceptor를 구현했어요. 하지만 아직 프로덕션 준비는 아니에요. 프로덕션급 TCP 서버는 각자 슈퍼바이저를 가진 acceptor 풀(pool) 을 돌려요. Elixir의 PartitionSupervisor로 acceptor를 파티셔닝·확장할 수도 있지만 이 가이드 범위 밖이에요. 실전에서는 이 용도에 맞는 기존 패키지(Erlang의 Ranch, Elixir의 Thousand Island)를 사용해요.

다음 장에서는 클라이언트 요청을 파싱하고 응답을 보내는 작업을 시작하며 서버를 마무리해요.

더 알아보기

  • 태스크: Task 모듈, Task.start_link/1, Task.child_spec/1
  • 태스크 감독: Task.Supervisor 모듈, Task.Supervisor.start_child/2
  • 감독: Supervisor 모듈, Supervisor.child_spec/2, 재시작 전략(:permanent/:temporary/:transient)
  • 네트워크: :gen_tcp (Erlang), :gen_tcp.controlling_process/2
  • 이전/다음 장: "Supervising dynamic children" → "Doctests, patterns, and with"