TCP 통신 구현하기 - 동기 blocking, 비동기 non-blocking

오늘 부터 오픈소스 웹 통신 프레임 워크를 만들기 위한 여정을 떠나보려고 한다.

일단 제일 먼저 TCP 통신을 직접 구현해보며, 네트워크 지식을 더 깊게 경험해보는 게 첫번째 목표다.

 

# TCP

TCP(Transmission Control Protocol)는 인터넷 프로토콜 스위트의 핵심 프로토콜로, 신뢰성 있는 데이터 전송을 보장한다. 신뢰성 있는 데이터 전송이라 하면, 보내고 받았다는 게 확인 된다는 것이다.

연결할 때는 3-way handshake, 연결을 종료할 때는 4-way handshake가 일어난다.

sequenceDiagram
    participant C as Client
    participant S as Server
    
    Note over C,S: 3-Way Handshake (연결 수립)
    C->>S: SYN (seq=x)
    Note right of C: 클라이언트: 연결 요청
    S->>C: SYN + ACK (seq=y, ack=x+1)
    Note left of S: 서버: 요청 수락 & 동기화
    C->>S: ACK (ack=y+1)
    Note right of C: 클라이언트: 연결 확인
    
    Note over C,S: 데이터 전송
    C->>S: Data Transfer
    S->>C: ACK
    S->>C: Data Transfer
    C->>S: ACK
    
    Note over C,S: 4-Way Handshake (연결 종료)
    C->>S: FIN
    Note right of C: 클라이언트: 연결 종료 요청
    S->>C: ACK
    Note left of S: 서버: 종료 요청 확인
    S->>C: FIN
    Note left of S: 서버: 서버도 종료 준비 완료
    C->>S: ACK
    Note right of C: 클라이언트: 종료 확인

3way handshake의 경우에는 연결 시도, 연결 확인 및 승인 요청, 최종 확인 단계로 이루어진다.

SYN을 C-> S로 전달하고, 시퀀스 번호랑 같이 담아서 보낸다. 서버는 SYN과 받은 시퀀스 번호를 가지고, 서버 쪽 시퀀스 번호와 클라이언트 가 준 시퀀스 번호에 +1 해서 ACK으로 보낸다. 만들어진 ACK + 클라이언트 가 준 SYN을 다시 클라이언트에게 보내면 클라이언트 쪽에서 이번에는 서버의 시퀀스 번호에 1을 더해 ACK로 응답해준다. 이게 서버에 도착하면 연결이 성립된다.

 

4way handshake는 조금 복잡한데 신뢰성이라는 개념을 들고가면 똑같다.

시작은 여전히 클라이언트다. 연결을 종료하겠다고 FIN을 보낸다. 그러면 서버가 요청을 받았음을 알려주는 ACK를 보내고, 서버 자체도 종료하겠다고 FIN을 보낸다. 그러면 클라이언트에서 서버의 종료 응답을 받고 ACK를 다시 서버에 전해주며 연결이 종료된다. 양쪽 모두 더이상 보낼 데이터가 없으니 리소스를 해제해야하는데, 상태확인을 한다음에 끊는 방식으로 안정성을 확보하게 되는 것이다.

 

시퀀스 번호를 사용하니 순서가 보장되기 때문에, 이점이 UDP 통신과 다른 것이다.

흐름제어도 가능하고, 혼잡 제어도 가능하다. 패킷의 체크섬을 사용해 오류 검출과 손상 패킷 재전송 기능도 있기 때문에 신뢰성 있는 통신이라고 할 수 있게 된다. 

interface TcpServer { // 라이프사이클 관리용 인터페이스
    fun start(port: Int)
    fun stop()
    fun isRunning(): Boolean
}

시작, 종료, 상태확인용 함수를 인터페이스로 만들어 시작했다.

private val isRunning = AtomicBoolean(false)
private var serverSocket: ServerSocket? = null
private val threadPool = Executors.newCachedThreadPool()

소켓 통신의 구현체로는 ServerSocket을 사용할 것이다. AtomicBoolean가 멀티스레트 환경에서 race condition 발생을 방지해준다고 해서 이걸 상태 확인 변수로 만들었다.

그리고 클라이언트 연결을 처리할 쓰레드 풀을 만들어 주면 끝난다.

cacheThreadPool이라 스레드를 생성하고, 60초간 사용되지않는 스레드는 제거해준다.

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());
}

start부터 구현한다.

override fun start(port: Int) {
    if (isRunning.get()) {
        throw IllegalStateException("Server is already running")
    }

    serverSocket = ServerSocket().apply {
        bind(InetSocketAddress(port), backlog)
    }
    isRunning.set(true)

    threadPool.submit {
        while (isRunning.get()) {
            try {
                val clientSocket = serverSocket?.accept() ?: break
                handleNewConnection(clientSocket)
            } catch (e: Exception) {
                if (isRunning.get()) {
                    e.printStackTrace()
                }
            }
        }
    }
}

private fun handleNewConnection(clientSocket: Socket) {
    threadPool.submit {
        try {
            connectionHandler.handle(clientSocket)
        } catch (e: Exception) {
            e.printStackTrace()
        } finally {
            try {
                clientSocket.close()
            } catch (e: Exception) {
                e.printStackTrace()
            }
        }
    }
}

ServerSocket객체를 할당하는 게 TCP/IP계층에서 LISTEN상태로 진입하는 TCP 연결의 시작점이다. OS가 저기서 받는 port에 대한 TCP 리스너를 생성하고 baklog에는 대기열을 처리할 TCP의 SYN 큐 크기를 지정해준다. 이걸 백프레셔라고 하는 것 같다.

SYN 큐에 완료되지않은 연결요청들이 쌓여있을 예정이다. 

public Socket accept() throws IOException {
    if (isClosed())
        throw new SocketException("Socket is closed");
    if (!isBound())
        throw new SocketException("Socket is not bound yet");
    Socket s = new Socket((SocketImpl) null);
    implAccept(s);
    return s;
}

실행중이라면 계속 요청을 보내고 응답을 받을 수 있다. 이때 `serverSocket?.accept()`부분이 3way handshake가 일어나는 부분이다. accept 자체가 연결완료되었을 때 반환된다.

 

TCP연결에 성공했다면 그 연결한 소켓에서 통신을 이룬다. 커넥션 핸들러를 구현하기 나름인걸로 해두었는데 일단 단순 에코용으로만 작성했다. TCP이 스트림으로 동작하니까 inputstream, outputstream으로 관리한다.

class EchoConnectionHandler : ConnectionHandler {
    override fun handle(socket: Socket) {
        val input = socket.getInputStream().bufferedReader()
        val output = socket.getOutputStream().bufferedWriter()

        input.lineSequence().forEach { line ->
            output.write("$line\n")
            output.flush()
        }
    }
}

커넥션 메서드에서의 `clientSocket.close()` 는 accept와 반대로 4way handshake를 발동시킨다.

telnet으로 테스트하면, 한글은 깨져서 오지만 일단 입력한게 응답으로 돌아오는 걸 볼 수 있다.

 

지금 구현해둔건 쓰레드 풀을 사용하긴 했지만 blocking I/O 방식이라 각 연결마다 하나의 스레드를 필요로한다. 커넥션이 열리는 만큼 스레드를 열어야하기 때문에 생성 비용이 1MB 스택 메모리인걸 감안했을 때 매우 비효율적임을 알 수 있다.

 

# 블로킹, 논블로킹, 동기, 비동기

일단 블로킹 I/O랑 논블로킹 I/O의 차이점을 살펴보자. 매우 헷갈리는 개념들인데, 블로킹/논블로킹과 동기/비동기는 서로 다른 개념인걸 알고 들어가야된다.

 

블로킹, 논블로킹은 호출 함수가 제어권을 바로 돌려주는 지, 아니면 대기 시키는 지에 대한 차이고, 동기 비동기는 작업완료의 통지 방식 차이를 나타낸다. 그래서 동기는 직접 완료가 됐다고 통지를 해주고, 비동기는 콜백형식으로 통지를 해서 다른 작업을 수행할 수 있게 분리시켜주는 것이다.

 

예시를 카페로 들어보면 좀 더 이해가 잘 된다. 커피를 주문하는 게 함수를 호출하는 것이라고 가정하겠다.

• 블로킹은 주문 받았으니 카운터에서 커피 기다리라고 시키는 것
• 논블로킹은 진동벨을 주고 기다리라고 하는 것
• 동기는 커피가 나왔을 때 사람이 직접 알려주는 것(알려줄 때 까지 기다려야된다)
• 비동기는 진동벨이 울렸을 때 찾아가라고 하는 것

이렇게 4가지로 정리 하겠다.

 

위에서 내가 Tcp서버를 구현한 방식은 블로킹 방식이다. 각 커넥션 마다 스레드를 할당해서 호출을 기다린다.

 

 

도움이 됐다면 댓글이나 공감 버튼 한 번씩 누르고 가주세요!