GenServer로 클라이언트-서버 만들기

GenServer로 클라이언트-서버 만들기 (Client-server with GenServer)

분산 키-값 저장소를 마무리하면서, 클라이언트가 버킷을 구독해 버킷에서 일어나는 모든 변경 사항을, 그 버킷이 클러스터의 어디에 있든 실시간으로 알림 받는 기능을 구현해 볼게요. SUBSCRIBE라는 새 명령을 추가해서 이렇게 사용할 거예요.

SUBSCRIBE shopping
milk SET TO 1
eggs SET TO 10
milk DELETED

이걸 동작시키려면 KV.Bucket 구현을 바꿔 구독을 추적하고 브로드캐스트를 내보내야 해요. 하지만 곧 보겠지만, 이건 agent 위에서는 구현할 수 없어요. 버킷 구현을 GenServer로 다시 작성해야 하죠.

출처: Client-server with GenServer

본문

링크와 모니터 (Links and monitors)

Elixir의 프로세스는 서로 격리돼 있어요. 통신이 필요하면 메시지를 주고받는 방식으로 하죠. 그런데 프로세스가 완료됐든 크래시했든, 언제 종료됐는지 어떻게 알 수 있을까요? 두 가지 선택지가 있어요. **링크(link)**와 **모니터(monitor)**예요.

우리는 링크를 많이 사용해 왔어요. 프로세스를 시작할 때 보통 start_link나 비슷한 방식으로 시작했죠. 링크의 개념은, 어느 한 프로세스가 크래시하면 링크 때문에 다른 쪽도 크래시한다는 거예요. Getting Started 가이드의 Process 장에서 다뤘죠. 복습해 볼게요.

iex> self()
#PID<0.115.0>
iex> spawn_link(fn -> :nothing_bad_will_happen end)
#PID<0.116.0>
iex> self()
#PID<0.115.0>

iex> spawn_link(fn -> raise "oops" end)
#PID<0.117.0>

12:37:33.229 [error] Process #PID<0.117.0> raised an exception
Interactive Elixir (1.18.4) - press Ctrl+C to exit (type h() ENTER for help)
iex> self()
#PID<0.118.0>

링크가 이렇게 널리 쓰이는 이유는, 감독자(supervisor) 안에서 프로세스를 시작할 때 감독자가 종료되면 우리 프로세스도 크래시하게 만들고 싶기 때문이에요. 반대로 자식이 종료될 때 감독자가 크래시하는 건 원하지 않죠. 그래서 감독자는 Process.flag(:trap_exit, true)를 호출해 링크에서 오는 exit 신호를 잡아둬요(trap).

다시 말해 링크는 프로세스 사이에 본질적인 관계를 만들어요. 그냥 프로세스가 언제 죽는지만 추적하고 싶고, exit 신호를 서로 묶고 싶지 않다면, 모니터가 더 나은 해결책이에요. 모니터링 중인 프로세스가 종료되면, 이유가 무엇이든 우리 메일박스(inbox)에 메시지가 도착해요.

iex> pid = spawn(fn -> Process.sleep(5000) end)
#PID<0.119.0>
iex> Process.monitor(pid)
#Reference<0.1076459145.2159017989.118674>
iex> flush()
:ok
# Wait five seconds
iex> flush()
{:DOWN, #Reference<0.1076459145.2159017989.118674>, :process, #PID<0.119.0>, :normal}
:ok

프로세스가 종료되면 "DOWN 메시지"를 받는데, 다섯 요소짜리 튜플로 표현돼요. 마지막 요소가 크래시한 이유(:normal은 정상 종료를 의미)예요.

모니터는 구독 기능에서 아주 중요한 역할을 해요. 클라이언트가 버킷을 구독하면, 버킷은 그 클라이언트 PID를 저장하고 변경이 있을 때마다 메시지를 보내요. 하지만 클라이언트가 종료되면(예: 연결이 끊긴 경우), 버킷은 구독자 목록에서 그 클라이언트를 제거해야 해요. 그렇지 않으면 클라이언트가 연결·해제를 반복할 때마다 목록이 계속 커지겠죠.

우리는 KV.Bucket을 구현할 때 Agent 모듈을 골랐는데, 아쉽게도 agent는 메시지를 받을 수 없어요. 그래서 첫 단계는 KV.Bucket을 GenServer로 다시 작성하는 거예요. GenServer 모듈 문서에 GenServer가 무엇이고 어떻게 구현하는지 좋은 개요가 있어요. 한번 읽어 보고 진행할게요.

GenServer 콜백 (GenServer callbacks)

GenServer는 특정 조건에서 제한된 함수 집합을 호출하는 프로세스예요. Agent를 쓸 때는 클라이언트 코드와 서버 코드를 나란히 두곤 했어요.

def put(bucket, key, value) do
  Agent.update(bucket, &Map.put(&1, key, value))
end

그 코드를 조금 나눠 볼게요.

def put(bucket, key, value) do
  # Here is the client code
  Agent.update(bucket, fn state ->
    # Here is the server code
    Map.put(state, key, value)
  end)
  # Back to the client code
end

위 코드에서는 "클라이언트"라고 부르는 프로세스가 agent("서버")에 요청을 보내요. 요청에는 서버가 실행해야 할 익명 함수가 담겨 있죠. GenServer에서는 위 코드가 두 개의 별도 함수가 돼요. 대략 이렇게요.

def put(bucket, key, value) do
  # Send the server a :put "instruction"
  GenServer.call(bucket, {:put, key, value})
end

# Server callback

def handle_call({:put, key, value}, _from, state) do
  {:reply, :ok, Map.put(state, key, value)}
end

KV.Bucket을 한 번에 다시 작성해 볼게요. lib/kv/bucket.ex를 열고 내용을 이 새 버전으로 바꾸세요.

defmodule KV.Bucket do
  use GenServer

  @doc """
  Starts a new bucket.
  """
  def start_link(opts) do
    GenServer.start_link(__MODULE__, %{}, opts)
  end

  @doc """
  Gets a value from the `bucket` by `key`.
  """
  def get(bucket, key) do
    GenServer.call(bucket, {:get, key})
  end

  @doc """
  Puts the `value` for the given `key` in the `bucket`.
  """
  def put(bucket, key, value) do
    GenServer.call(bucket, {:put, key, value})
  end

  @doc """
  Deletes `key` from `bucket`.

  Returns the current value of `key`, if `key` exists.
  """
  def delete(bucket, key) do
    GenServer.call(bucket, {:delete, key})
  end

  ### Callbacks

  @impl true
  def init(bucket) do
    state = %{
      bucket: bucket
    }

    {:ok, state}
  end

  @impl true
  def handle_call({:get, key}, _from, state) do
    value = get_in(state.bucket[key])
    {:reply, value, state}
  end

  def handle_call({:put, key, value}, _from, state) do
    state = put_in(state.bucket[key], value)
    {:reply, :ok, state}
  end

  def handle_call({:delete, key}, _from, state) do
    {value, state} = pop_in(state.bucket[key])
    {:reply, value, state}
  end
end

첫 번째 함수는 start_link/1로, 옵션 목록을 넘겨 새 GenServer를 시작해요. GenServer.start_link/3은 인자 세 개를 받죠.

• 서버 콜백이 구현된 모듈. 여기서는 __MODULE__(현재 모듈을 의미)이에요.
• 초기화 인자. 여기서는 빈 버킷 %{}이에요.
• 서버 이름 같은 것을 지정할 때 사용하는 옵션 목록. agent에서처럼 옵션 목록을 다시 GenServer.start_link/3로 전달해요.

시작되면 GenServer는 init/1 콜백을 호출해요. 이 콜백은 GenServer.start_link/3에 준 두 번째 인자를 받아 {:ok, state}를 반환하는데, 여기서 state는 새 맵이에요. GenServer API가 클라이언트/서버 구분을 얼마나 뚜렷하게 만드는지 벌써 알 수 있어요. start_link/3은 클라이언트에서 일어나고, init/1은 서버에서 실행되는 대응 콜백이죠.

GenServer에 보낼 수 있는 요청에는 두 종류가 있어요. call과 cast예요. call은 동기적이고 서버는 그런 요청에 반드시 응답을 보내야 해요. 서버가 응답을 계산하는 동안 클라이언트는 기다리고 있죠. cast는 비동기적이라 서버가 응답을 보내지 않고, 클라이언트도 기다리지 않아요. 두 요청 모두 서버에 보내는 메시지이고 순서대로 처리돼요. 지금까지는 Agent와 동일한 의미를 유지하려고 GenServer.call/2만 사용했지만, 구독을 구현할 때 cast를 시험해 볼 거예요. 동작을 그대로 유지했으니 모든 테스트가 여전히 통과해요.

각 요청은 특정 콜백으로 구현돼야 해요. call/2 요청에는 handle_call/3 콜백을 구현하는데, 이 콜백이 request, 요청을 받은 프로세스(_from), 현재 서버 상태(state)를 받아요. handle_call/3 콜백은 {:reply, reply, updated_state} 형식의 튜플을 반환해요. 튜플의 첫 요소 :reply는 서버가 클라이언트에 응답을 보내야 함을 나타내요. 두 번째 요소 reply는 클라이언트에 보내질 값이고, 세 번째 updated_state는 새 서버 상태예요.

앞서 구현에서 쓴 또 다른 Elixir 기능은 중첩 순회 함수, get_in/1·put_in/2·pop_in/1이에요. bucket을 GenServer 상태로 유지하는 대신, 안에 bucket 키가 있는 상태 맵을 정의했어요. GenServer 상태의 일부로 구독자(subscribers)도 추적해야 하므로 중요하죠. 이 새 함수들은 다른 자료 구조에 중첩된 자료 구조를 다루기 쉽게 해줘요.

GenServer를 마련했으니 구독 작업을 시작할게요. 테스트부터 시작하죠.

구독 구현하기 (Implementing subscriptions)

새 테스트는 버킷에 구독하고, 버킷에 연산이 수행될 때 그 이벤트에 대한 메시지를 받는지 단언(assert)할 거예요. test/kv/bucket_test.exs를 열고 이걸 입력하세요.

  test "subscribes to puts and deletes" do
    {:ok, bucket} = start_supervised(KV.Bucket)
    KV.Bucket.subscribe(bucket)

    KV.Bucket.put(bucket, "milk", 3)
    assert_receive {:put, "milk", 3}

    # Also check it works even from another process
    spawn(fn -> KV.Bucket.delete(bucket, "milk") end)
    assert_receive {:delete, "milk"}
  end

테스트를 통과시키려면 KV.Bucket.subscribe/1을 구현해야 해요. KV.Bucket에 이 새 함수 세 개를 추가해 볼게요.

  @doc """
  Subscribes the current process to the bucket.
  """
  def subscribe(bucket) do
    GenServer.cast(bucket, {:subscribe, self()})
  end

  @impl true
  def handle_cast({:subscribe, pid}, state) do
    Process.monitor(pid)
    state = update_in(state.subscribers, &MapSet.put(&1, pid))
    {:noreply, state}
  end

  @impl true
  def handle_info({:DOWN, _ref, _type, pid, _reason}, state) do
    state = update_in(state.subscribers, &MapSet.delete(&1, pid))
    {:noreply, state}
  end

구독하면 현재 프로세스 식별자와 함께 cast/2 요청을 보내고, request와 현재 서버 상태를 받는 handle_cast/2 콜백을 구현해요. 그다음 주어진 pid를 모니터링하고 MapSet으로 구현한 구독자 목록에 추가해요. handle_cast/2 콜백은 {:noreply, updated_state} 형식의 튜플을 반환해요. 실제 애플리케이션이라면 back pressure를 제공하므로 비동기 cast 대신 동기 call로 구현했을 거라는 점을 기억하세요. 여기서는 cast 콜백을 어떻게 구현하는지 보여주려고 이렇게 하는 거예요.

그리고 프로세스를 모니터링했으니, 그 프로세스가 종료되면 "DOWN 메시지"를 받아요. GenServer는 일반 메시지를 handle_info/2 콜백으로 처리하는데, 이것도 보통 {:noreply, updated_state}를 반환해요. 이 콜백에서 종료된 PID를 구독자 목록에서 제거해요.

거의 다 왔어요. handle_cast/2와 handle_info/2 콜백이 모두 상태에 MapSet이 담긴 subscribers 키가 있다고 가정하는 걸 볼 수 있어요. 기존 init/1을 아래처럼 바꿔서 추가해 줄게요.

  @impl true
  def init(bucket) do
    state = %{
      bucket: bucket,
      subscribers: MapSet.new()
    }

    {:ok, state}
  end

마지막으로 put/3과 delete/2의 콜백을, 호출될 때마다 메시지를 브로드캐스트하도록 바꿀게요.

  def handle_call({:put, key, value}, _from, state) do
    state = put_in(state.bucket[key], value)
    broadcast(state, {:put, key, value})
    {:reply, :ok, state}
  end

  def handle_call({:delete, key}, _from, state) do
    {value, state} = pop_in(state.bucket[key])
    broadcast(state, {:delete, key})
    {:reply, value, state}
  end

  defp broadcast(state, message) do
    for pid <- state.subscribers do
      send(pid, message)
    end
  end

get/2의 콜백은 수정할 필요 없어요. 이제 테스트를 다시 실행하면 새 테스트가 통과할 거예요!

모두 연결하기 (Wiring it all up)

이제 버킷이 구독을 처리하니, 서버에서 이 새 기능을 노출해야 해요. 다시 테스트부터 시작할게요. test/kv/server_test.exs를 열고 이 새 테스트를 추가하세요.

  test "subscribes to buckets", %{socket: socket, name: name} do
    assert send_and_recv(socket, "CREATE #{name}\r\n") == "OK\r\n"
    :gen_tcp.send(socket, "SUBSCRIBE #{name}\r\n")

    {:ok, other} = :gen_tcp.connect(~c"localhost", 4040, @socket_options)

    assert send_and_recv(other, "PUT #{name} milk 3\r\n") == "OK\r\n"
    assert :gen_tcp.recv(socket, 0, 1000) == {:ok, "milk SET TO 3\r\n"}

    assert send_and_recv(other, "DELETE #{name} milk\r\n") == "OK\r\n"
    assert :gen_tcp.recv(socket, 0, 1000) == {:ok, "milk DELETED\r\n"}
  end

테스트는 버킷을 만들고 구독해요. 그다음 명령을 보내기 위해 TCP 연결을 하나 더 열어요. 명령을 보낼 때마다 구독한 소켓이 메시지를 받을 것으로 기대하죠. 테스트를 통과시키려면 KV.Command가 새 SUBSCRIBE 명령을 파싱하고 실행하도록 바꿔야 해요. lib/kv/commands.ex를 열고 먼저 parse/1 정의를 아래처럼 바꾸세요.

  def parse(line) do
    case String.split(line) do
      ["SUBSCRIBE", bucket] -> {:ok, {:subscribe, bucket}}
      ["CREATE", bucket] -> {:ok, {:create, bucket}}
      ["GET", bucket, key] -> {:ok, {:get, bucket, key}}
      ["PUT", bucket, key, value] -> {:ok, {:put, bucket, key, value}}
      ["DELETE", bucket, key] -> {:ok, {:delete, bucket, key}}
      _ -> {:error, :unknown_command}
    end
  end

"SUBSCRIBE"를 튜플로 바꾸는 절을 하나 추가했어요. 이제 run/1 안에서 이 튜플을 매치해야 해요. run/1의 맨 아래에 아래 코드로 절 하나를 추가하면 돼요.

  def run({:subscribe, bucket}, socket) do
    lookup(bucket, fn pid ->
      KV.Bucket.subscribe(pid)
      :inet.setopts(socket, active: true)
      receive_messages(socket)
    end)
  end

  defp receive_messages(socket) do
    receive do
      {:put, key, value} ->
        :gen_tcp.send(socket, "#{key} SET TO #{value}\r\n")
        receive_messages(socket)

      {:delete, key} ->
        :gen_tcp.send(socket, "#{key} DELETED\r\n")
        receive_messages(socket)

      {:tcp_closed, ^socket} ->
        {:error, :closed}

      # If we receive any message, including socket writes, we discard them
      _ ->
        receive_messages(socket)
    end
  end

부분별로 살펴볼게요. 기존 lookup/2 비공개 함수로 버킷을 찾아요. 찾으면 현재 프로세스를 버킷에 구독해요. 그다음 :inet.setopts(socket, active: true)(곧 설명할게요)와 receive_messages/1을 호출해요.

receive_messages/1은 버킷에서 메시지를 기다렸다가 자기 자신을 다시 호출해 루프가 돼요. {:put, key, value}와 {:delete, key}를 매치해 그 이벤트를 소켓에 써요. 그리고 TCP 소켓이 닫히면 전달되는 {:tcp_closed, ^socket} 메시지도 매치해서 루프를 끝내는 데 사용해요. 나머지 메시지는 버려요.

여기서 ":tcp_closed, ^socket은 어디서 오는 거지?" 하는 의문이 들 수도 있어요. 지금까지 소켓에서 메시지를 받을 때는 :gen_tcp.recv/3을 사용해, 콘텐츠가 올 때까지 현재 프로세스를 블로킹하는 호출을 했어요. 이걸 "수동 모드(passive mode)"라고 불러요. 하지만 :gen_tcp에 메시지가 도착할 때마다 현재 프로세스의 메일박스로 스트리밍하라고 요청할 수도 있는데, 이걸 "활성 모드(active mode)"라고 해요. :inet.setopts(socket, active: true)를 호출해 설정한 게 바로 그거예요. 그런 메시지는 {:tcp, socket, data} 형태예요. 소켓이 활성 모드에서 닫히면 {:tcp_closed, socket} 메시지를 전달해요. 이 메시지를 받으면 루프를 빠져나오고, 연결 프로세스가 종료되죠. 버킷이 프로세스를 모니터링하고 있으므로 구독도 자동으로 제거돼요. 실제로 확인하고 싶다면 주어진 버킷의 구독자 수를 반환하는 COUNT SUBSCRIPTIONS 명령을 추가해 보면 돼요.

실제 시스템에서는 메시지 큐가 넘치지 않도록 TCP 메시지 하나만 전달되게 :inet.setopts(socket, active: :once)를 호출하는 걸 선호하는 경우가 많아요. 메시지를 받으면 :inet.setopts/2를 다시 호출하죠. 우리 경우엔 소켓으로 도착하는 걸 그냥 버리면 되므로 active: true로 설정해도 똑같이 무방해요. 어떤 시나리오든 활성 모드의 장점은 :gen_tcp.recv/3에 블로킹하는 대신, 프로세스가 TCP 메시지와 다른 프로세스의 메시지를 동시에 받을 수 있다는 거예요.

마무리로, 이 새 기능을 분산 환경에서도 시험해 보세요. NODES=... PORT=... iex --sname ... -S mix 인스턴스를 두 개 시작해요. 하나에서 버킷을 만들고, 다른 하나에서 같은 버킷에 구독해요. 첫 번째 셸로 돌아가 보면, 한 머신에서 버킷에 명령을 보내도 메시지가 다른 쪽으로 스트리밍되는 걸 볼 수 있어요. 즉 우리 구독 시스템도 분산돼 있고, 메시지를 보내기만 하면 되는 거예요!

call, cast 아니면 info?

지금까지 세 가지 콜백 handle_call/3, handle_cast/2, handle_info/2를 사용했어요. 각각을 언제 써야 하는지 고려할 점은 다음과 같아요.

• handle_call/3은 동기 요청에 사용해야 해요. 서버 응답을 기다리는 건 유용한 back-pressure 메커니즘이므로 기본 선택이 돼요.
• handle_cast/2는 응답을 신경 쓰지 않는 비동기 요청에 사용해야 해요. cast는 서버가 메시지를 받았음을 보장하지 않으므로 아껴 써야 해요. 예를 들어 이번 장에서 정의한 subscribe/1은 call/2를 썼어야 해요. 여기서는 학습 목적으로 cast/2를 쓴 거예요.
• handle_info/2는 GenServer.call/2나 GenServer.cast/2로 보내지지 않은, 서버가 받는 다른 모든 메시지에 사용해야 해요. send/2로 보낸 일반 메시지도 포함되고요. 모니터링 :DOWN 메시지가 그 예시예요.

call·cast·info의 차이와 지원되는 반환 값 등을 기억하기 쉽게, 작은 GenServer 치트 시트를 제공하고 있어요.

Agent 아니면 GenServer?

마지막 장으로 넘어가기 전에 궁금할 수 있어요. 앞으로 Agent를 써야 할까요, GenServer을 써야 할까요? 이 가이드에서 봤듯이 agent는 시작하기 쉽지만 할 수 있는 일이 제한적이에요. Agent는 사실상 GenServer의 부분 집합이에요. 실제로 agent는 GenServer 위에 구현돼 있어요. 감독자, Registry 모듈, Erlang과 Elixir에서 찾을 수 있는 다른 많은 기능도 마찬가지예요.

즉 GenServer는 Elixir에서 동시성·결함 허용 시스템을 구축하는 가장 핵심적인 구성 요소예요. 상태를 관리하고 프로세스 간 상호작용을 조정하는 견고하고 유연한 프레임워크를 제공하죠. 이런 이유로 agent를 절대 쓰지 말고 바로 GenServer로 뛰어드는 걸 규칙으로 삼는 사람도 많아요. 반면 어디에 조금씩 상태를 저장하는 데 agent를 쓰는 것으로 충분하다고 여기는 사람도 있어요. 어느 쪽이든 문제없어요!

이것이 분산 키-값 저장소를 위해 구현한 마지막 기능이에요. 다음 장에서는 애플리케이션을 프로덕션에 배포하기 전에 패키징하는 방법을 배울 거예요.