Boost.asio 병렬 동기 TCP 서버

개요

병렬 동기 서버는 다음과 같은 사항을 만족시키는 분산 어플리케이션이다.

• 클라이언트 서버 통신 모델에서 서버로 동작한다.
• TCP 프로토콜을 사용해 클라이언트 프로그램과 통신한다.
• I/O 및 제어 연산을 하는 동안에는 연산이 끝나거나 오류가 발생할 때까지 실행 스레드를 멈춘다.
• 클라이언트를 한 번에 하나 이상 처리할 수 있다.

일반적인 병렬 동기 TCP 서버는 다음과 같은 알고리즘에 따라 동작한다.

1. 수용자 소켓을 할당하고 특정 TCP 포트에 묶는다.
2. 서버가 중단될 때까지 루프를 돈다.
3. 클라이언트로부터 연결 요청이 오기를 기다린다.
4. 연결 요청이 오면 받아들인다.
5. 스레드를 새로 생성하고 새 스레드에서 다음과 같은 작업을 한다.
6. 클라이언트로부터 요청 메시지가 오기를 기다린다.
7. 요청 메시지를 읽는다.
8. 요청을 처리한다.
9. 클라이언트에게 응답 메시지를 보낸다.
10. 클라이언트와의 연결을 닫고 소켓을 할당 해지한다.

 

병렬 동기 TCP 서버 구현

#include <boost/asio.hpp>
#include <thread>
#include <atomic>
#include <memory>
#include <iostream>

using namespace boost;

class Service {
public:
    Service() {}

    void StartHandlingClient(std::shared_ptr<asio::ip::tcp::socket> sock) {
        std::thread th([this, sock]() {
            HandleClient(sock);
            });

        th.detach();  // 스레드를 분리하여 백그라운드로 처리
    }

private:
    void HandleClient(std::shared_ptr<asio::ip::tcp::socket> sock) {
        try {
            asio::streambuf request;
            asio::read_until(*sock, request, '\n');  // 클라이언트로부터 데이터를 읽음

            // 요청 처리 시뮬레이션
            int i = 0;
            while (i != 1000000) i++;

            std::this_thread::sleep_for(std::chrono::milliseconds(500));  // 잠시 대기

            // 응답 전송
            std::string response = "Response\n";
            asio::write(*sock, asio::buffer(response));  // 응답을 클라이언트에게 전송
        }
        catch (system::system_error& e) {
            std::cout << "Error occurred! Error code = "
                << e.code() << ". Message: "
                << e.what() << std::endl;
        }

        // Clean-up
    }
};

class Acceptor {
public:
    Acceptor(asio::io_context& ios, unsigned short port_num) :
        m_ios(ios),
        m_acceptor(m_ios, asio::ip::tcp::endpoint(asio::ip::address_v4::any(), port_num)) {
        m_acceptor.listen();
    }

    void Accept() {
        auto sock = std::make_shared<asio::ip::tcp::socket>(m_ios);

        m_acceptor.accept(*sock);  // 연결 수락

        auto service = std::make_shared<Service>();
        service->StartHandlingClient(sock);  // 클라이언트 처리 시작
    }

private:
    asio::io_context& m_ios;
    asio::ip::tcp::acceptor m_acceptor;
};

class Server {
public:
    Server() : m_stop(false) {}

    void Start(unsigned short port_num) {
        m_thread.reset(new std::thread([this, port_num]() {
            Run(port_num);
            }));
    }

    void Stop() {
        m_stop.store(true);
        m_thread->join();  // 서버 종료 시 스레드 종료 대기
    }

private:
    void Run(unsigned short port_num) {
        Acceptor acc(m_ios, port_num);

        while (!m_stop.load()) {
            acc.Accept();  // 클라이언트의 연결을 기다림
        }
    }

    std::unique_ptr<std::thread> m_thread;
    std::atomic<bool> m_stop;
    asio::io_context m_ios;  // 최신 버전에서는 io_context를 사용
};

int main() {
    unsigned short port_num = 3333;

    try {
        Server srv;
        srv.Start(port_num);

        std::this_thread::sleep_for(std::chrono::seconds(60));  // 서버 60초 동안 실행

        srv.Stop();  // 서버 종료
    }
    catch (system::system_error& e) {
        std::cout << "Error occurred! Error code = "
            << e.code() << ". Message: "
            << e.what() << std::endl;
    }

    return 0;
}

 

Server 클래스는 서버의 실행을 관리한다.

Acceptor 클래스는 클라이언트의 연결을 수락한다.

Sevice 클래스는 클라이언트 연결을 처리하는 작업을 별도의 스레드에서 수행한다.

main 함수에서 서버를 시작한 후 60초간 sleep기간을 갖고 서버를 종료한다.

 

Service클래스가 이번 예제에서 핵심적인 역할을 맡는다.

다른 부분들은 구조적인 역할을 맡는 데 반해, 이 클래스는 서버가 클라이언트에게 제공하는 실제 기능을 구현한다.

StartHandlingClient() 메서드를 통해 매개변수로 받은 소켓을 인자로 스레드를 생성하여 HandleClient함수를 호출한다.

이 과정에서 스레드를 분리하여 해당 스레드가 알아서 연산 작업을 하도록 내버려 둔다.

 

클라이언트 요청을 처리하는 각 스레드는 독립적이므로, 요청과 응답이 동시에 여러 클라이언트에 대해 발생한다.

반복 TCP 서버와 다르게 각 클라이언트는 자신만의 스레드에서 작업이 수행되므로, 다른 클라이언트의 작업이 완료될 때까지 기다리지 않는다.

따라서 사용자는 서버가 비동기적으로 작동한다고 느낄 수 있다.

 

하지만 실제로는 클라이언트 작업 중 스레드는 블로킹 상태로 대기한다.

따라서 클라이언트 수가 많아지면 스레드 수가 증가하며, 이는 시스템 리소스를 빠르게 소모할 수 있다.

또한 반복 TCP 서버와 동일하게 악의 적인 사용자가 통신을 시작하고 아무런 요청을 하지 않은 경우 해당 스레드는 계속 블럭된 상태로 있게 되고, 해당 클라이언트가 통신을 종료할 때 까지 해결할 방법이 없다.