pkglog
EN

C++ Asio Composed Operation | 비동기 함수 설계 [#7]

작성 · · 18분 읽기 · 수정

이 글의 핵심

비동기 읽기를 콜백 안에서 또 호출하는 식으로 이어 붙이면, 에러 처리와 수명 관리가 호출하는 쪽 코드마다 흩어집니다. Composed Operation은 여러 단계의 비동기 작업을 하나의 초기화 함수로 감싸 Asio의 기본 연산처럼 쓸 수 있게 하는 패턴으로, 헤더와 바디 프로토콜 예제로 구현 흐름을 따라가고 디버깅, 성능 측정, 흔한 실수까지 짚습니다.

들어가며: “헤더 읽고 → 바디 읽고”를 한 번에

실제 프로토콜은 “먼저 N바이트 헤더를 읽으며, 그 다음 헤더에 적힌 길이만큼 바디를 읽는다” 같은 여러 단계의 비동기 I/O로 이루어집니다. 이를 매번 콜백을 중첩해 작성하면 반복되고 지저분해집니다. Composed Operation은 이런 여러 개의 비동기 연산을 하나의 비동기 연산처럼 묶어서, async_read_header_then_body 같은 “나만의 비동기 함수”로 만드는 Asio의 설계 패턴입니다. 이렇게 만들면 호출 측에서는 한 번의 비동기 호출로 “헤더+바디 읽기 완료”를 기다릴 수 있으며, 코루틴이면 한 번의 co_await로 처리할 수 있습니다. 언제 쓰면 좋을까요? Echo나 단순 라인 프로토콜은 async_read_until만으로 충분합니다. 고정 헤더 + 가변 바디, 프레임 단위 읽기처럼 “여러 번의 async_read를 한 단위로 묶고 싶을 때” Composed Operation을 만들면, 프로토콜 계층이 깔끔해지고 #6의 코루틴과도 co_await async_read_packet(…) 한 줄로 맞물리게 할 수 있습니다. 목표:

  • Composed Operation의 개념 — 여러 비동기 단계를 하나로 묶기
  • async_initiate와 async_compose (또는 수동 초기화)로 구현하는 흐름
  • 실전: async_read_packet (헤더 4바이트 + 바디) 예시

Composed Operation이란

개념

  • 단일 비동기 연산: async_read_some, async_write처럼 “한 번 시작하면 한 번의 완료 콜백”으로 끝나는 연산.
  • Composed Operation: 내부적으로 여러 번의 비동기 연산을 순서대로 시작하며, 각 완료 시 다음 단계를 시작하는 상태 머신을 구현한 비동기 연산. 외부에서는 “하나의 비동기 연산”처럼 보입니다. 예: async_read_until은 내부적으로 “버퍼에 구분자가 올 때까지 async_read_some을 반복”하는 Composed Operation입니다. 사실 이 글에서 계속 쓰는 async_read 자체도 “요청한 바이트가 다 찰 때까지 async_read_some을 반복”하는 Composed Operation입니다. 우리도 async_read_packet처럼 “헤더 읽기 → 바디 읽기”를 한 번에 하는 연산을 만들 수 있습니다.

그냥 콜백을 중첩하면 되는데 왜 이런 형식을 갖춰야 할까요? Asio의 비동기 연산에는 호출자가 기대하는 보이지 않는 계약이 있기 때문입니다. 완료 핸들러는 시작 함수 안에서 바로 호출되지 않고, 핸들러에 연결된 executor(예: strand)에서 실행되며, 연산이 진행되는 동안 io_context의 작업 카운트가 유지되어 run()이 먼저 끝나지 않고, 핸들러에 붙은 할당자와 취소 슬롯이 내부 단계까지 전달되어야 합니다. 중첩 람다로 직접 짜면 이 규칙들을 하나씩 손으로 지켜야 하고, 하나라도 빠뜨리면 strand를 쓴 호출자에게서만 데이터 경쟁이 생기는 식의 찾기 어려운 버그가 됩니다. async_compose는 이 계약을 대신 지켜 주는 도구입니다.

설계 목표: async_read_packet

시그니처 (개념)

  • async_read_packet(socket, header_buf, body_buf, token)
  • 헤더: 고정 4바이트 (바디 길이).
  • 바디: 헤더에 적힌 길이만큼 읽기.
  • 완료 시: 에러 코드와 읽은 바이트 수(또는 헤더+바디 성공 여부). token이 콜백이면 (error_code, size_t) 형태의 완료 핸들러, use_awaitable이면 awaitable<std::size_t> 등으로 완료를 전달.

구현 흐름: 상태 머신과 연쇄 호출

단계

  1. 1단계: async_read로 정확히 4바이트(헤더)를 읽습니다. 완료 핸들러에서:
    • 에러면 최종 완료 콜백/awaitable에 에러 전달.
    • 성공이면 헤더에서 바디 길이를 파싱하며, 2단계로 진행.
  2. 2단계: async_read로 바디 길이만큼 읽습니다. 완료 핸들러에서:
    • 에러/성공을 최종 완료에 전달. 이 “1단계 → 2단계”를 한 번의 비동기 시작으로 감싸는 래퍼가 Composed Operation입니다. 구현 시에는 작업 객체(operation state) 가 자신을 비동기 연산의 완료 핸들러로 넘기면서, 완료 시 “다음 단계를 시작”하거나 “최종 완료를 호출”하는 식으로 작성합니다.

의사 코드

struct read_packet_op {
    void start() {
        async_read(socket, buffer(header), [this](ec, n) {
            if (ec) { complete(ec, 0); return; }
            size_t body_len = parse_header(header);
            async_read(socket, buffer(body, body_len), [this](ec, n) {
                complete(ec, 4 + n);
            });
        });
    }
    void complete(error_code ec, size_t total) {
        // token에 따라 콜백 호출 또는 awaitable 재개
    }
};

실제로는 완료 토큰(token) 에 따라 “콜백 호출” vs “awaitable 재개”를 async_initiate 등으로 통일해 처리합니다.

이 의사 코드의 [this] 캡처가 보여 주듯, 연산 상태 객체는 마지막 완료 핸들러가 실행될 때까지 살아 있어야 합니다. 지역 변수로 만든 read_packet_op의 start()를 부르고 함수가 반환하면, 첫 번째 async_read가 끝났을 때 this는 이미 사라진 뒤입니다. 수동 구현에서는 상태를 shared_ptr로 힙에 두고 각 단계 핸들러가 그것을 캡처하게 해야 하고, 아래의 async_compose는 상태를 핸들러 안에 담아 단계마다 이동시키는 방식으로 이 문제를 없앱니다.

async_initiate와 완료 토큰

Asio의 완료 토큰

  • 콜백: void(error_code, size_t) 형태의 핸들러.
  • use_awaitable: 코루틴에서 co_await할 때 쓰는 토큰. 완료 시 awaitable이 재개되도록 함. async_initiate는 “어떤 토큰이든 받아서, 해당 토큰에 맞게 비동기 연산을 시작하며, 완료 시 그 토큰에 맞게 결과를 전달”하도록 도와줍니다. Composed Operation의 시작 함수에서 async_initiate를 호출하며, 내부에서 1단계 비동기를 시작할 때 “완료 시 이 작업 객체의 다음 단계를 호출”하는 식으로 바인딩하면, 콜백/awaitable 둘 다 지원하는 async_read_packet을 만들 수 있습니다.

async_compose로 구현한 async_read_packet

Boost 1.70부터 제공되는 async_compose는 async_initiate 위에 “여러 단계를 가진 연산”을 쉽게 만들도록 얹은 헬퍼입니다(C++20 전용 기능이 아니라 C++11 이상에서 동작합니다). 상태 머신을 람다(또는 함수 객체) 하나로 쓰고, 다음 단계를 시작할 때 self를 완료 핸들러로 넘기면 됩니다.

#include <boost/asio.hpp>
#include <array>
#include <cstddef>
#include <vector>
namespace asio = boost::asio;
using asio::ip::tcp;

template <typename CompletionToken>
auto async_read_packet(tcp::socket& socket,
                       std::array<std::byte, 4>& header,
                       std::vector<std::byte>& body,
                       CompletionToken&& token)
{
    enum class state { start, header_done, body_done };
    return asio::async_compose<CompletionToken,
                               void(boost::system::error_code, std::size_t)>(
        [&socket, &header, &body, st = state::start]
        (auto& self, boost::system::error_code ec = {}, std::size_t n = 0) mutable {
            switch (st) {
            case state::start:                       // 1단계: 헤더 4바이트
                st = state::header_done;
                asio::async_read(socket, asio::buffer(header), std::move(self));
                return;
            case state::header_done: {               // 2단계: 길이 파싱 후 바디
                if (ec) { self.complete(ec, 0); return; }
                std::uint32_t len = 0;
                for (std::byte b : header)
                    len = (len << 8) | std::to_integer<std::uint32_t>(b);  // 빅엔디안
                if (len > 64u * 1024 * 1024) {
                    self.complete(asio::error::message_size, 0);
                    return;
                }
                body.resize(len);
                st = state::body_done;
                asio::async_read(socket, asio::buffer(body), std::move(self));
                return;
            }
            case state::body_done:                   // 완료: 토큰에 결과 전달
                self.complete(ec, ec ? 0 : 4 + n);
                return;
            }
        },
        token, socket);   // socket: 작업 추적과 기본 executor를 얻을 I/O 객체
}

같은 함수를 세 가지 방식으로 부를 수 있다는 것이 이 설계의 핵심 이점입니다.

// 1) 콜백
async_read_packet(sock, hdr, body, [](boost::system::error_code ec, std::size_t n) { /* ... */ });
// 2) 코루틴 (에러는 예외)
std::size_t n = co_await async_read_packet(sock, hdr, body, asio::use_awaitable);
// 3) 코루틴 (에러를 값으로)
auto [ec, m] = co_await async_read_packet(sock, hdr, body, asio::as_tuple(asio::use_awaitable));

구현에서 눈여겨볼 곳은 세 군데입니다. 첫째, 람다는 처음에 self 하나만 받아 호출되고 이후에는 내부 async_read의 결과 (ec, n)과 함께 다시 호출되므로, 기본 인자로 두 경우를 한 함수에 담습니다. 둘째, st는 람다 안에 값으로 들어 있어 단계마다 self와 함께 이동하므로 힙 할당이나 shared_ptr 없이도 상태가 유지됩니다. 셋째, 마지막 인자 socket은 연산이 진행되는 동안 소켓의 executor에 작업이 남아 있다고 알려 run()이 먼저 끝나지 않게 합니다.

반대로 이 구현이 보장하지 않는 것도 있습니다. header와 body는 참조로 캡처했으므로 호출자가 완료 때까지 살려 둬야 하고, 같은 소켓에 이 연산을 동시에 두 번 걸면 두 읽기가 바이트를 나눠 가져 경계가 깨집니다. 또 start 단계에서 곧바로 self.complete()를 호출해야 하는 경우(예: 소켓이 이미 닫혀 있음을 미리 알고 즉시 실패하는 경우)에는 완료 핸들러가 시작 함수 안에서 실행되지 않도록 asio::post(socket.get_executor(), ...)로 한 번 미뤄야 한다는 점을 기억하십시오. 제가 composed operation을 처음 만들 때 가장 오래 헤맨 버그가 이것이었는데, 콜백으로 부를 때는 멀쩡하다가 호출자가 뮤텍스를 잡은 상태에서 호출하면 핸들러가 같은 뮤텍스를 다시 잡으려다 교착되는 형태로만 나타났습니다.

문서 참고

구체적인 async_initiate 서명과 연산 상태 라이프타임 관리는 Boost.Asio 문서 - Composed Operations와 예제를 참고하는 것이 좋습니다. 여러 단계를 가진 연산을 co_await로 더 읽기 쉽게 쓰고 싶다면, 최신 Boost.Asio의 co_composed(실험적 기능)도 살펴볼 만합니다.

정리

  • Composed Operation은 여러 비동기 단계(헤더 읽기 → 바디 읽기)를 하나의 비동기 연산처럼 묶는 패턴.
  • 내부는 상태 머신: 1단계 완료 핸들러에서 2단계를 시작하며, 최종 단계에서 완료 토큰(콜백 또는 awaitable)에 결과를 전달.
  • async_initiate와 완료 토큰을 사용하면 콜백과 co_await 둘 다 지원하는 나만의 async_read_packet 같은 API를 우아하게 설계할 수 있습니다. 이렇게 만든 비동기 프로토콜 함수는 Echo나 채팅이 아닌 “헤더+바디” 프로토콜을 다루는 고성능 서버의 기본 단위가 됩니다.

보강: Composed Operation 실전 예제 — HTTP 스타일 헤더 + 바디

실제 HTTP는 더 복잡하지만, “먼저 고정 헤더(또는 헤더 블록)를 읽으며, 그다음 Content-Length만큼 바디를 읽는다”는 흐름은 아래와 같이 모델링할 수 있습니다.

프로토콜 가정

  • 4바이트 빅엔디안 길이 필드(바디 바이트 수).
  • 그 다음 바디를 정확히 그 길이만큼 읽습니다.

콜백 스타일 연쇄 (핵심만)

void read_length_then_body(
    boost::asio::ip::tcp::socket& socket,
    std::array<std::byte, 4>& len_buf,
    std::vector<std::byte>& body_buf,
    std::function<void(boost::system::error_code, std::size_t)> done)
{
    boost::asio::async_read(socket, boost::asio::buffer(len_buf),
        [&socket, &len_buf, &body_buf, done = std::move(done)]
        (const boost::system::error_code& ec, std::size_t) {
            if (ec) { done(ec, 0); return; }
            std::uint32_t n = 0;
            for (int i = 0; i < 4; ++i)
                n = (n << 8) | static_cast<unsigned char>(len_buf[static_cast<std::size_t>(i)]);
            if (n > 64 * 1024 * 1024) {  // 예: 상한으로 DoS 완화
                done(boost::asio::error::message_size, 0);
                return;
            }
            body_buf.resize(n);
            boost::asio::async_read(socket, boost::asio::buffer(body_buf),
                [done = std::move(done)](const boost::system::error_code& ec2, std::size_t m) {
                    done(ec2, m);
                });
        });
}

이 콜백 버전은 짧지만 앞에서 말한 “보이지 않는 계약”을 몇 가지 어깁니다. 바깥 람다는 mutable이 아니라서 안쪽의 std::move(done)이 실제로는 복사가 되고, std::function은 호출자가 strand에 묶어 둔 핸들러의 executor 정보를 지워 버리므로 호출자가 bind_executor(strand, ...)로 넘겨도 done은 strand 밖에서 실행될 수 있습니다. 또 소켓과 버퍼를 참조로 잡고 있어 호출자가 세션 객체를 먼저 파괴하면 댕글링입니다. 이런 문제들이 앞 절의 async_compose 버전에서는 구조적으로 해결된다는 점이 두 방식의 실질적인 차이입니다.

코루틴과 결합

throw std::system_error를 쓰려면 <system_error>를 포함합니다.

boost::asio::awaitable<std::vector<std::byte>> read_packet(boost::asio::ip::tcp::socket& socket) {
    std::array<std::byte, 4> len_buf{};
    co_await boost::asio::async_read(socket, boost::asio::buffer(len_buf), boost::asio::use_awaitable);
    std::uint32_t n = 0;
    for (int i = 0; i < 4; ++i)
        n = (n << 8u) | static_cast<unsigned char>(len_buf[static_cast<std::size_t>(i)]);
    const std::size_t max_body = 64 * 1024 * 1024;
    if (n > max_body)
        throw std::system_error(std::make_error_code(std::errc::message_too_long));
    std::vector<std::byte> body(n);
    co_await boost::asio::async_read(socket, boost::asio::buffer(body), boost::asio::use_awaitable);
    co_return body;
}

위 두 단계를 하나의 async_read_packet으로 묶으면, 호출부는 read_packet(socket, token) 한 번으로 끝나고, 내부 상태 머신·에러 전달은 Composed Operation으로 캡슐화할 수 있습니다.

보안·운영 체크

  • 최대 바디 길이를 반드시 제한합니다(메모리 고갈 방지).
  • 부분 헤더·부분 바디는 async_read가 “정확히 N바이트”를 채워 줄 때까지 반복하거나, 한 번의 Composed Operation 안에서 처리합니다.

코루틴 버전의 read_packet은 코드가 가장 읽기 쉽지만, 헤더를 받은 뒤 바디가 오지 않으면 co_await가 영원히 끝나지 않는다는 점은 콜백 버전과 똑같습니다. 연결이 끊기면 End of file(asio::error::eof)로 예외가 나지만, 상대가 연결만 유지한 채 아무것도 보내지 않으면 아무 일도 일어나지 않습니다.

보강: 디버깅 팁

  • 단계별로 에러 코드를 로그에 남기고, 어느 단계에서 끊겼는지(헤더 / 바디)를 구분합니다. async_compose 구현이라면 st 값을 함께 기록하면 됩니다.
  • 타임아웃은 별도 steady_timer를 같은 Strand에 묶어, 헤더만 오고 바디가 안 오는 경우를 처리합니다. 타이머가 만료되면 socket.cancel()을 호출하고, 진행 중인 단계의 async_read는 operation_aborted로 완료되어 연산 전체가 그 에러로 끝납니다. 최신 Boost.Asio(1.86 이상)에서는 asio::cancel_after(std::chrono::seconds(10), token)처럼 완료 토큰에 타임아웃을 붙이는 방법도 있습니다.
  • BOOST_ASIO_ENABLE_HANDLER_TRACKING을 정의하고 빌드하면 핸들러가 언제 만들어지고 실행되는지 표준 에러로 출력되어, 어느 단계에서 연산이 멈췄는지 추적하기 쉽습니다.

보강: 성능 측정 방법

  • 동일 크기 패킷을 초당 N개 전송할 때, Composed 전후로 처리량·CPU·할당 횟수를 비교합니다. async_compose는 단계마다 핸들러를 이동하므로 상태 객체가 크면 이동 비용이 생기고, 핸들러를 힙에 저장하는 할당이 단계마다 일어날 수 있습니다. #5 핸들러 할당자에서 다룬 연관 할당자를 붙이면 이 비용을 줄일 수 있습니다.
  • 한 단계짜리 async_read_some 루프와 비교해 프레임 경계가 맞는지 검증한 뒤, 최적화는 프로파일 기준으로 합니다. 작은 패킷이 많다면 헤더와 바디를 따로 읽는 대신 큰 버퍼로 async_read_some을 한 번 하고 그 안에서 여러 프레임을 파싱하는 방식이 시스템 콜 수를 크게 줄입니다.

보강: 흔한 실수와 해결책

실수해결
async_read_some만으로 “헤더 4바이트”를 기대버퍼에 쪼개 들어옴 → 정확히 4바이트는 async_read.
길이 필드를 신뢰만 하고 상한 없음DoS·OOM → 최대 길이·연결당 버퍼 제한.
Composed 내부에서 소켓 생명주기 끊김shared_ptr로 세션 유지 또는 취소 토큰 사용.
시작 함수 안에서 바로 self.complete() 호출완료를 post로 미뤄 “핸들러는 시작 함수 반환 후 실행” 규칙 유지.
같은 소켓에 읽기 연산을 동시에 두 개한 소켓당 읽기 체인은 하나만, 쓰기는 송신 큐로 직렬화.

자주 묻는 질문

Q. Composed Operation은 언제 쓰는 게 좋나요? A. Echo나 단순 라인 프로토콜은 async_read_until만으로 충분합니다. 고정 헤더 + 가변 바디, 프레임 단위 읽기처럼 여러 번의 async_read를 한 단위로 묶고 싶을 때 Composed Operation을 만들면, 호출부가 co_await async_read_packet(...) 한 줄로 정리됩니다. Q. async_initiate 없이 콜백만 써도 되나요? A. 콜백만 쓴다면 수동으로 연쇄 호출을 구현해도 됩니다. async_initiate와 완료 토큰을 쓰면 콜백과 use_awaitable(코루틴) 둘 다 같은 API로 지원할 수 있어, 나중에 코루틴으로 옮길 때 호출부를 바꿀 필요가 없습니다.

시리즈 마무리

C++ 고성능 네트워크 가이드 시리즈는 여기까지입니다.

  1. #1 — io_context, run/poll, Proactor, work_guard
  2. #2 — 멀티스레드 Asio, Data Race, Mutex 한계
  3. #3 — Strand, make_strand, bind_executor
  4. #4 — post, dispatch, defer
  5. #5 — 핸들러 메모리, 커스텀 할당자
  6. #6 — C++20 코루틴, awaitable
  7. #7 — Composed Operation
    더 깊이 보고 싶다면 C++ 실전 가이드 #29: Asio와 #30: WebSocket·프로토콜을 이어서 읽어 보시면 좋습니다.

async_read_packet 상태 전이

flowchart TD
    S[start: 헤더 async_read 시작] --> H{header_done: 에러?}
    H -->|예| C1[complete ec, 0]
    H -->|아니오| L{길이 상한 초과?}
    L -->|예| C2[complete message_size, 0]
    L -->|아니오| B[바디 async_read 시작]
    B --> D[body_done: complete ec, 4 + n]

설명: async_compose 람다가 호출될 때마다 st 값에 따라 한 칸씩 진행합니다. 어느 경로로 끝나든 self.complete()는 정확히 한 번만 호출되어야 하며, 두 번 호출하거나 한 번도 호출하지 않으면 호출자는 결과를 두 번 받거나 영원히 기다리게 됩니다.

같이 보면 좋은 글