🌐 Multiplexing NonBlocking IO 네트워크 아키텍처 조사 - YangJJune/U-Compass GitHub Wiki

ThreadPool의 효율적인 개수를 연구하려 한 이유?

• 실시간 위치 공유의 빠른 속도와 효율을 위한 7주간이었다.
• 지난 시간에 ThreadPool의 개수가 무작정 많다고 좋은 것이 아니라는 것을 확인할 수 있었고, 이에 따라 효율적인 Thread의 개수에 대해 공부를 하던 중 흥미로운 내용을 확인할 수 있었다.
• 바로 Multi-Process / Multi-Thread / Multi-Plex 프로그래밍이었다

Multi-Processing

• 여러 개의 독립된 프로세스를 동시에 실행하는 것이다
• 따라서 시스템 자원을 독립적으로 할당받는다
  • CPU 자원 분할은 OS에 의해 이루어진다
  • 따라서 프로세스 간의 데이터 전달에는 IPC (파이프, 큐, 소켓 등)가 필요하다
• 컨텍스트 스위칭 발생 시 굉장히 일이 크다
  • Memory Address Space 의 변경이 필요하기 때문(메모리 주소, 맵핑, 페이지 테이블 등)

Multi-Threading

• 모든 쓰레드가 프로세스의 메모리 공간을 공유한다
  • 따라서 모든 쓰레드는 같은 메모리 영역에 접근할 수 있다
  • 메모리 공간을 공유하기 때문에 데이터 전달이 용이하다
• I/O 작업이 진행되는 동안 다른 쓰레드에서는 작업을 진행하는 등의 효율을 발휘할 수 있다
• 컨텍스트 스위칭 발생 시 일이 프로세스에 비해 비교적 적다

Multi-Plexing

• 위 도식도는 select()를 사용하여 IO Multiplexing을 하는 플로우를 나타낸 것이다.
• 엄밀히 이야기하면 위의 Multi-Processing, Multi-Threading과 직접적으로 엮어서 표현하기 보다는 IO에서 Multiplexing을 할 수 있고, 이 때 위의 2개의 기법과 비교하여 사용할 수 있다고 표현할 수 있다.
• 제대로 된 Multiplexing 개념으로 들어가기 전에 Synchronous, Asynchronous, Blocking IO와 Non-blocking IO에 대해 알아보도록 한다.

Synchronous IO - 동기

• 모든 작업이 일련의 순서를 따른다 → 작업의 순서 보장
• 작업 완료를 user space에서 판단한다
• 일련의 pipeline을 유지하는 구조에서 효율적이다

Asynchronous IO

• kernel에 I/O 작업을 요청해두고 다른 작업을 진행한다 → 작업의 순서 보장 X
• 작업 완료를 kernel space에서 통보한다
• 각 작업이 독립적이거나, 작업별 지연이 큰 경우 효율적이다

Blocking

• 요청한 작업이 모두 완료될 때까지 기다렸다가 완료될 때 응답과 결과를 반환받는다.
• 요청한 작업 결과를 기다린다.

Non-blocking

• 작업 요청 이후 결과는 나중에 필요할 때 전달받는다.
• 요청한 작업 결과를 기다리지 않는다.

IBM I/O Model

• 표를 보는 법은 “Blocking 하면서 Asynchronous 한 모델은 I/O 멀티플렉싱이다”는 식으로 보면 된다.
• Synchronous Blocking I/O → Blocking I/O
• Synchronous Non-Blocking I/O → Non-Blocking I/O
• Asynchronous Blocking I/O → I/O Multiplexing
• Asynchronous Non-Blocking I/O → Asynchronous I/O

Blocking I/O

Non-Blocking I/O

I/O Multiplexing

• 프로세스에서 특정 파일에 접근할 때는 파일 디스크립터(이하 FD, File Descriptor)라는 추상적인 값을 사용한다. 이때, 이 FD들을 감시하여 I/O Multiplexing을 할 수 있다

• 위 구조는 kernel은 I/O 요청을 받으며 처리를 시작함과 동시에 user process에게 미완료 상태를 반환하고 user process는 데이터가 준비됐다는 알람이 올 때까지 대기하는 모습을 보여주고 있다

• 이어서, kernel의 응답을 기다리다 보면 kernel에서 결과 값이 준비되었다는 callback 신호가 오고 user process는 자신의 buffer로 데이터를 복사해오는 모습을 보여주는데, 이는 select 호출 결과가 유의미한 값으로 나올 때까지 user process에서 loop를 돌리며 대기하는 방식이다.

• select / poll / epoll(Linux전용) / kqueue(maxOS전용)

• Select

  • 최대 1024개의 FD를 감시한다
    • FD 개수 제한
    • 감시 대상이 늘어날수록 느려짐 O(N)

• Poll

  • Select의 개선 버전
    • FD 개수 제한 없음
    • 여전히 매번 FD 리스트를 커널에 전달해야하므로 감시 대상이 늘어날수록 느려짐 O(N)

• epoll

  • Linux 2.6 이후 도입된 고성능 이벤트 기반 I/O 멀티플렉싱 메커니즘
  • 이벤트 기반으로 동작하며, 비동기적으로 연결을 처리한다. O(1)
    • 높은 성능
    • 효율적인 메모리 사용
    • 비동기 처리 → I/O가 완료된 소켓만 처리

• kqueue

  • 커널 내의 이벤트 큐를 관리하며, 이벤트 기반으로 동작
  • epoll과 유사하게 비동기 I/O 모델을 구현하여, 연결된 소켓의 이벤트를 감시하고 해당 이벤트가 발생할 때 처리한다
  • BSD 계열 운영체제에서만 사용 가능

Asynchronous I/O

• Asynchronous I/O 환경에서 user process는 system call 이후 I/O 처리에 신경 쓰지 않고 있다가 작업이 완료되면 kernel로부터 signal, thread 기반 callback 등으로 결과를 마치 event처럼 전달받게 됩니다. 그렇기에 응답이 오기 전까지 user process는 I/O와 독립적인 다른 processing이 가능한 구조입니다.

구현

select

import socket
import select
import time

server = socket.socket()
server.bind(('0.0.0.0', 3388))
server.listen()
server.setblocking(False)

# 서버 소켓 저장
sockets = [server]

while True:
   # 변화가 있을 때 까지 대기 (무한루프)
   readable, _, _ = select.select(sockets, [], [])
   print(readable)
   for sock in readable:
      # 서버 소켓과 동일한 객체인가? (소켓인가?)
      if sock is server:
         # 새로운 클라이언트 소켓 추가
         client, _ = server.accept()
         client.setblocking(False)
         sockets.append(client)
      # 데이터인가? (이미 연결된 소켓으로부터 온 데이터)
      else:
         data = sock.recv(1024)
         if data:
               sock.sendall(data)
         else:
               sockets.remove(sock)
               sock.close()
   time.sleep(1)

poll

import socket
import select
import time

server = socket.socket()
server.bind(('0.0.0.0', 3388))
server.listen()
server.setblocking(False)
# poll 객체 생성
poller = select.poll()

# 서버소켓의 fd -> socket 매핑
fd_to_socket = {server.fileno(): server}

# 서버 소켓을 읽기 이벤트에 등록
poller.register(server, select.POLLIN)

print(f"Server listening")

while True:
    events = poller.poll(1000)  # timeout: 1000ms
    print(events)

    for fd, event in events:
        sock = fd_to_socket[fd]
        
        # 서버 소켓과 동일한가? (소켓 객체인가?)
        if sock is server:
            # 새 클라이언트 연결
            client_sock, addr = server.accept()
            print(f"New connection from {addr}")
            client_sock.setblocking(False)
            #클라 소켓 fd와 객체 연결
            fd_to_socket[client_sock.fileno()] = client_sock
            poller.register(client_sock, select.POLLIN)
        elif event & select.POLLIN:
            # 데이터 수신
            data = sock.recv(1024)
            if data:
                print(f"Received: {data.decode()}")
                sock.sendall(data)  # 에코
            else:
                # 연결 종료
                print("Client disconnected")
                poller.unregister(fd)
                fd_to_socket.pop(fd)
                sock.close()

epoll

import socket
import select
import time

server = socket.socket()
server.bind(('0.0.0.0', 3388))
server.listen()
server.setblocking(False)
# epoll 객체 생성
epoll = select.epoll()

# epoll에 서버 소켓 등록 (읽기 이벤트)
epoll.register(server.fileno(), select.EPOLLIN)

# 파일 디스크립터 → 소켓 매핑
fd_to_socket = {server.fileno(): server}

print(f"Server listening...")

try:
    while True:
        # 이벤트 감지 (timeout: 1초)
        events = epoll.poll(1)

        for fd, event in events:
            sock = fd_to_socket[fd]

            # 서버 소켓에 이벤트 발생 → 새 연결
            if sock is server:
                client_sock, addr = server.accept()
                print(f"New connection from {addr}")
                client_sock.setblocking(False)
                fd_to_socket[client_sock.fileno()] = client_sock
                epoll.register(client_sock.fileno(), select.EPOLLIN)

            # 클라이언트 소켓에 읽기 이벤트
            elif event & select.EPOLLIN:
                data = sock.recv(1024)
                if data:
                    print(f"Received: {data.decode().strip()}")
                    sock.sendall(data)  # 에코
                else:
                    print("Client disconnected")
                    epoll.unregister(fd)
                    sock.close()
                    del fd_to_socket[fd]

            # 클라이언트 소켓 에러 or 종료
            elif event & (select.EPOLLHUP | select.EPOLLERR):
                print("Client connection closed (error)")
                epoll.unregister(fd)
                sock.close()
                del fd_to_socket[fd]

finally:
    epoll.unregister(server.fileno())
    epoll.close()
    server.close()

유사한 서비스의 구현 사례

Discord

• 실시간 음성/메시지 서비스
• 클라이언트-서버 네트워킹 아키텍처 사용
  • 1000명이 초과하면 P2P 아키텍처가 비용이 많이 들기 때문\

• 각 디스코드 커뮤니티를 Guild라 칭한다
• 그리고 Guild Process가 Guild의 조율과 라우팅을 담당한다
• 모든 사용자는 Session Process를 갖고, Guild Proccess는 새 메시지, 이벤트 또는 업데이트를 받으면 해당 정보를 관련 Session Process로 전달한다
  • Session Process는 해당 업데이트를 Discord 클라이언트로 푸시한다
  • 이는 많은 사용자의 부하가 걸릴 수록 Guild Process에 더 큰 부담을 주었다
    • BEAM이 효과적이라고는 하나 궁극적인 수단이 되지는 못하였다고 한다

• 그래서 최적화를 진행했다고 한다
  • 불필요한 작업을 줄이기
    • ex) 사용자가 길드를 열 때까지 처리 및 데이터 전송을 skip한다
  • 릴레이 최적화
    • Relay는 막대한 트래픽을 처리하기 위해 사용하는 BEAM 기반 프로세스
      • ex) 모든 유저한테 메시지 보내기
      • 이를 처리하기 위해 여러 프로세스를 띄워서 작업을 분산
      • 그러나 Relay 자체에 막대한 RAM이 소모되어 필요한 소수의 멤버만 추적하도록 릴레이 최적화
  • 서버 응답성 유지
    • Helper Process
      • BEAM 프로세스 간 빠른 데이터 공유를 위한 인메모리 데이터베이스 ETS 활용
      • 느린 작업의 경우 비동기 작업을 위해 Helper Process가 생성된다
        • ex) 길드 마이그레이션의 처리
  • Manifold Offload
    • 팬아웃을 길드에서 별도의 "sender" 프로세스로 분산하여 길드 작업 부하를 더욱 줄이고 길드 프로세스를 네트워크 백프레셔로부터 보호
    • Garbage Collection이 Tricky했다

Slack

• 실시간 메시지의 전송 / 수신을 위한 아키텍처 구조
• 크게 보면 이전 주차에서 진행한 Star / Tree 토폴로지임을 알 수 있다

• Client, Edge Region, Main Region으로 크게 구분된다.
  • Client
    • 실제 유저의 클라이언트 (PC, Web, App)이 구동되는 공간
    • 유저의 입력을 받거나 서버로부터의 메시지를 수신한다
  • Edge Region
    • Envoy : Lyft에서 개발한 고성능 L4/L7 Proxy 서버
      • 로드밸런싱
      • TLS 종료
      • 트래픽 관찰 및 제어 등
    • Gateway Server
      • 클라이언트의 모든 요청을 한 곳으로 받아, 적절한 내부 서비스로 전달하는 중앙 통제 지점
      • Slack 에서는 Channel Server로 가기 전에 모든 요청을 받는 역할
  • Main Region
    • 실질적인 기능이 구동되는 내부 서비스
    • 실질적인 데이터가 저장된다

참고자료

https://stackoverflow.com/questions/51635655/multiplexing-vs-multithreading-on-a-tcp-distributed-system

[https://velog.io/@euisuk-chung/파이썬-Multiprocessing-Multithreading-자원-분할-및-할당](https://velog.io/@euisuk-chung/%ED%8C%8C%EC%9D%B4%EC%8D%AC-Multiprocessing-Multithreading-%EC%9E%90%EC%9B%90-%EB%B6%84%ED%95%A0-%EB%B0%8F-%ED%95%A0%EB%8B%B9)

https://www.linkedin.com/pulse/choice-between-multithreading-multi-processing-when-use-deepak-kumar-uxyvf

https://blog.naver.com/n_cloudplatform/222189669084

https://medium.com/@rrakshith007/io-multiplexing-and-its-advantages-8c75584079d1

https://discord.com/blog/how-discord-handles-push-request-bursts-of-over-a-million-per-minute-with-elixirs-genstage

https://blog.quastor.org/p/slack-sends-millions-messages-real-time

https://slack.engineering/real-time-messaging/