Computer >> 컴퓨터 >  >> 프로그래밍 >> Ruby

동시성 심층 분석: 이벤트 루프로 구현하는 채팅 서버

Ruby Magic 동시성 시리즈의 마지막 아티클에 오신 것을 환영합니다. 지금까지 우리는 여러 프로세스와 여러 스레드를 사용해 채팅 서버를 구현해 보았습니다. 이번에는 같은 기능을 이벤트 루프(event loop) 방식으로 구현해 보겠습니다.

복습

이전 글에서 사용한 것과 동일한 클라이언트와 서버 설정을 그대로 사용합니다. 목표는 다음과 같은 채팅 시스템을 만드는 것입니다.

기본 설정에 대한 자세한 내용은 이전 아티클을 참고해 주세요. 이 글의 예제에서 사용된 전체 소스 코드는 GitHub에서 확인할 수 있으니 직접 실험해 보셔도 좋습니다.

이벤트 루프 기반 채팅 서버

채팅 서버에 이벤트 루프를 적용하려면 스레드나 프로세스를 사용할 때와는 다른 사고방식이 필요합니다. 전통적인 방식에서는 하나의 스레드 또는 프로세스가 하나의 연결을 담당합니다. 반면 이벤트 루프를 사용하면 단일 프로세스의 단일 스레드가 여러 연결을 동시에 처리합니다. 이것이 어떻게 동작하는지 하나씩 살펴보겠습니다.

이벤트 루프란?

EventMachine이나 Node.js에서 사용하는 이벤트 루프는 대략 다음과 같이 동작합니다. 먼저 운영체제에 특정 이벤트, 예를 들어 소켓 연결이 열리는 상황에 관심이 있다고 알립니다. 이를 위해 연결이나 소켓 같은 IO 객체에 관심을 등록하는 함수를 호출합니다.

이 IO 객체에서 무슨 일이 일어나면 운영체제가 프로그램에 이벤트를 보냅니다. 이 이벤트들을 큐(queue)에 넣고, 이벤트 루프는 큐에서 이벤트를 하나씩 꺼내 순서대로 처리합니다.

어떤 의미에서 이벤트 루프는 진정한 동시성(concurrency)은 아닙니다. 매우 작은 단위의 작업을 순차적으로 빠르게 수행하면서 마치 동시에 실행되는 것처럼 보이게 만드는 것입니다.

관심사를 등록하고 운영체제로부터 IO 이벤트를 전달받으려면 원래 C 확장(extension)을 작성해야 합니다. Ruby 표준 라이브러리에는 해당 API가 없기 때문입니다. 하지만 C 확장 작성은 이 글의 범위를 벗어나므로, 대신 IO.select를 사용해 이벤트를 생성하겠습니다. IO.select는 감시할 IO 객체 배열을 인자로 받아, 배열 중 하나 이상의 객체가 읽기 또는 쓰기 준비가 될 때까지 기다린 후 준비된 IO 객체만 담은 배열을 반환합니다.

연결과 관련된 모든 처리 로직은 Fiber로 구현합니다. 앞으로 이 코드를 "핸들러(handler)"라고 부르겠습니다. Fiber는 실행을 일시 중지했다가 재개할 수 있는 코드 블록입니다. Ruby VM이 이 과정을 자동으로 처리해 주지 않기 때문에, 개발자가 직접 resume과 yield를 호출해야 합니다. 우리는 IO.select의 결과를 활용해 각 핸들러에게 자신의 연결이 읽기 또는 쓰기 가능한 상태임을 알려줄 것입니다.

이전 글의 멀티 스레드 및 멀티 프로세스 예제와 마찬가지로, 클라이언트와 주고받은 메시지를 추적하기 위한 저장소가 필요합니다. 하지만 이번에는 Mutex가 필요 없습니다. 이벤트 루프는 단일 스레드에서 실행되므로 여러 스레드가 동시에 객체를 변경할 위험이 없기 때문입니다.

client_handlers = {}
messages = []

클라이언트 핸들러는 다음 Fiber로 구현됩니다. 소켓을 읽거나 쓸 수 있는 상태가 되면 이벤트가 발생하고, Fiber가 이에 반응합니다. 상태가 :readable이면 소켓에서 한 줄을 읽어 messages 배열에 추가하고, 상태가 :writable이면 마지막 쓰기 이후 다른 클라이언트로부터 수신된 메시지를 클라이언트에 전송합니다. 이벤트 처리가 끝나면 Fiber.yield를 호출해 일시 중지하고 다음 이벤트를 기다립니다.

def create_client_handler(nickname, socket)
  Fiber.new do
    last_write = Time.now
    loop do
      state = Fiber.yield
 
      if state == :readable
        # 소켓에서 메시지를 읽습니다
        incoming = read_line_from(socket)
        # 문제가 없으면 쓰기 목록에 추가합니다
        $messages.push(
          :time => Time.now,
          :nickname => nickname,
          :text => incoming
        )
      elsif state == :writable
        # 소켓에 메시지를 씁니다
        get_messages_to_send(last_write, nickname, $messages).each do |message|
          socket.puts "#{message[:nickname]}: #{message[:text]}"
        end
        last_write = Time.now
      end
    end
  end
end

그렇다면 소켓이 준비되었을 때 어떻게 정확한 타이밍에 Fiber가 읽거나 쓰도록 트리거할 수 있을까요? 우리는 네 단계로 구성된 이벤트 루프를 사용합니다.

loop do
  # 1단계: 새 연결 요청 수락
  accept_incoming_connections
 
  # 2단계: 읽기/쓰기 준비가 된 연결 조회
  get_ready_connections
 
  # 3단계: 읽기 가능한 연결에서 데이터 읽기
  read_from_readable_connections
 
  # 4단계: 쓰기 가능한 연결에 데이터 쓰기
  write_to_writable_connections
end

여기에는 특별한 마법 같은 것이 없습니다. 평범한 Ruby 루프일 뿐입니다.

1단계: 새 연결 요청 수락

새로 들어온 연결이 있는지 확인합니다. 여기서는 클라이언트 접속을 기다리지 않는 accept_nonblock을 사용합니다. 새 클라이언트가 없으면 예외를 발생시키며, 우리는 이 예외를 잡아 다음 단계로 넘어갑니다. 새 클라이언트가 있다면 해당 클라이언트의 핸들러를 생성해 clients 저장소에 등록합니다. 나중에 핸들러를 찾을 수 있도록 소켓 객체를 해시의 키(key)로 사용합니다.

begin
  socket = server.accept_nonblock
  nickname = socket.gets.chomp
  $client_handlers[socket] = create_client_handler(nickname, socket)
  puts "Accepted connection from #{nickname}"
rescue IO::WaitReadable, Errno::EINTR
  # 현재 새로 들어온 연결이 없습니다
end

2단계: 읽기/쓰기 준비가 된 연결 조회

다음으로, 연결이 준비되면 운영체제가 알려달라고 요청합니다. 읽기, 쓰기, 에러 감시 대상으로 client_handlers 저장소의 키, 즉 1단계에서 수락한 소켓 객체들을 전달하고, 최대 10밀리초 동안 기다립니다.

readable, writable = IO.select(
  $client_handlers.keys,
  $client_handlers.keys,
  $client_handlers.keys,
  0.01
)

3단계: 읽기 가능한 연결에서 데이터 읽기

읽기 가능한 연결이 있다면 해당 클라이언트 핸들러를 :readable 상태로 resume하여 트리거합니다. IO.select가 반환한 Socket 객체가 핸들러 저장소의 키로 사용되고 있기 때문에, 손쉽게 클라이언트 핸들러를 조회할 수 있습니다.

if readable
  readable.each do |ready_socket|
    # 저장소에서 클라이언트를 가져옵니다
    client = $client_handlers[ready_socket]
 
    client.resume(:readable)
  end
end

4단계: 쓰기 가능한 연결에 데이터 쓰기

쓰기 가능한 연결이 있다면 해당 클라이언트 핸들러를 :writable 상태로 resume하여 트리거합니다.

if writable
  writable.each do |ready_socket|
    # 저장소에서 클라이언트를 가져옵니다
    client = $client_handlers[ready_socket]
    next unless client
 
    client.resume(:writable)
  end
end

이렇게 네 단계를 루프로 반복하며 핸들러를 생성하고, 적절한 시점에 핸들러를 :readable 또는 :writable 상태로 resume하면 완전히 동작하는 이벤트 기반 채팅 서버가 완성됩니다. 연결당 오버헤드가 매우 적기 때문에 대규모 동시 클라이언트까지도 충분히 확장(scale)할 수 있습니다.

이 방식은 루프의 한 번의 tick당 처리하는 작업량을 작게 유지하는 한 매우 잘 동작합니다. 특히 계산이 많은 작업에서는 더욱 중요합니다. 이벤트 루프는 단일 스레드로 실행되기 때문에 CPU를 하나만 활용할 수 있습니다. 실제 운영 환경에서는 이 한계를 극복하기 위해 여러 프로세스가 각각 이벤트 루프를 실행하는 구조를 흔히 사용합니다.

마무리: 어떤 방식을 선택해야 할까?

여기까지 세 가지 방식을 모두 살펴보았습니다. 그렇다면 어떤 방법을 사용해야 할까요?

  • 대부분의 애플리케이션: 스레딩(threading)이 가장 적합합니다. 다루기 가장 간단한 방식입니다.
  • 장시간 스트림을 다루는 고동시성 애플리케이션: 이벤트 루프를 사용하면 뛰어난 확장성을 얻을 수 있습니다.
  • 프로세스 크래시가 예상되는 경우: 검증된 멀티 프로세스 방식이 가장 견고합니다.

이것으로 동시성 시리즈를 마무리합니다. 전체 내용을 복습하고 싶다면 동시성 마스터하기 원문 아티클과 멀티 프로세스, 멀티 스레드를 다룬 상세 아티클도 함께 확인해 보세요.