실전: 센더/리시버로 비동기 파이프라인 짜보기

시나리오: 여러 URL을 받아 다운로드하고 합치기

앞 두 장에서 익힌 just, then, when_all, scheduler를 엮어서 실제로 있을 법한 작업을 짜보겠습니다. 여러 URL에서 데이터를 내려받아, 각각을 파싱하고, 실패한 것은 로그로 남기고, 성공한 것들만 합치는 파이프라인입니다. 콜백으로 짰다면 앞서 본 것처럼 중첩이 심해졌을 코드입니다.

먼저 개별 조각부터 준비합니다.

#include <execution>
#include <exec/static_thread_pool.hpp>
#include <string>
#include <vector>
#include <iostream>
namespace ex = std::execution;

struct Document { std::string url; std::string body; };

// 실제로는 네트워크 I/O 지만, 예제이므로 동기 함수로 단순화한다
std::string download(const std::string& url) {
    if (url.find("bad") != std::string::npos) {
        throw std::runtime_error("download failed: " + url);
    }
    return "<content of " + url + ">";
}

Document parse(const std::string& url, std::string body) {
    return Document{url, body};
}

이제 URL 하나를 처리하는 센더를 만듭니다. 다운로드가 실패하면 예외를 던지므로, then 체인 안에서 던져진 예외는 자동으로 그 센더의 에러 채널(set_error)로 흘러갑니다 — 콜백 코드에서처럼 직접 에러 콜백을 호출할 필요가 없습니다.

ex::sender auto fetch_one(exec::static_thread_pool::scheduler sched,
                           std::string url) {
    return ex::schedule(sched)
         | ex::then([url] { return download(url); })
         | ex::then([url](std::string body) { return parse(url, std::move(body)); });
}

여러 다운로드를 동시에 실행하고 합치기

when_all은 인자로 받은 센더 개수가 고정돼 있어야 해서, URL 목록처럼 개수가 동적인 경우엔 when_all_vector 계열(구현체마다 이름이 다를 수 있습니다 — 여기서는 개념을 보여주기 위해 반복 조합으로 풀어씁니다)이 필요합니다. 이 책에서는 URL 세 개로 고정한 예제로 핵심 흐름만 보여드립니다.

void run_pipeline() {
    exec::static_thread_pool pool(4);
    auto sched = pool.get_scheduler();

    std::vector<std::string> urls = {
        "https://a.example/data",
        "https://bad.example/data",   // 일부러 실패시킬 URL
        "https://c.example/data",
    };

    auto combined = ex::when_all(
        fetch_one(sched, urls[0]),
        fetch_one(sched, urls[1]),
        fetch_one(sched, urls[2])
    );

    try {
        auto [docs] = *ex::sync_wait(std::move(combined));
        // ⚠️ when_all 은 하나라도 실패하면 전체가 실패한다 — 아래 절 참고
    } catch (const std::exception& e) {
        std::cerr << "pipeline failed: " << e.what() << "\n";
    }
}

when_all의 함정 — "하나가 실패하면 전부 실패"

여기서 반드시 짚어야 할 게 있습니다. when_all은 넘겨받은 센더 중 하나라도 에러로 끝나면 전체 결과가 에러가 됩니다. 이건 버그가 아니라 설계이지만, "URL 하나쯤 실패해도 나머지는 살리고 싶다"는 요구와는 정면으로 부딪힙니다. 위 코드에서 urls[1]이 실패하면 urls[0], urls[2]가 이미 성공했더라도 그 결과는 버려집니다.

이 문제를 풀려면 실패할 수 있는 개별 센더를 then이 아니라 에러를 값으로 흡수하는 조합기(예: upon_error, 또는 std::expected로 감싸는 방식)로 감싸서, "실패도 하나의 정상적인 결과"로 바꿔줘야 합니다.

ex::sender auto fetch_one_safe(exec::static_thread_pool::scheduler sched,
                                std::string url) {
    return fetch_one(sched, url)
         | ex::then([](Document d) -> std::optional<Document> { return d; })
         | ex::upon_error([url](std::exception_ptr) -> std::optional<Document> {
               std::cerr << "skip failed url: " << url << "\n";
               return std::nullopt;
           });
}

이렇게 감싸면 when_all에 넘겼을 때 개별 실패가 전체를 무너뜨리지 않고, 결과는 std::optional<Document> 목록으로 모입니다. 실패한 항목만 걸러내는 건 이후 평범한 로직으로 처리하면 됩니다.

이 장에서 제일 하고 싶었던 말은 이겁니다 — 센더/리시버는 콜백 지옥을 없애주지만, "일부 실패를 허용할지"는 여전히 개발자가 명시적으로 설계해야 하는 문제라는 것입니다. 프레임워크가 대신 판단해주지 않습니다. when_all의 기본 동작(전부 아니면 전무)을 모르고 썼다가, 하나의 네트워크 타임아웃 때문에 이미 받아온 나머지 데이터까지 통째로 날리는 실수는 실무에서 충분히 나올 법합니다.