2024-10-05

HSPACE Rust 특강 #11 — 비동기 프로그래밍 Part 1

RustHSPACE 2024AsyncFutures

동시성 회차에서 스레드를 배웠는데, 왜 또 비동기인가? 이번 회차는 그 질문에서 시작한다.

동기 → 비동기

동기 코드에서 소켓을 읽는다고 하자.

let mut buf = [0u8; 1024];
let n = socket.read(&mut buf)?;   // 데이터가 올 때까지 이 스레드는 멈춘다

이 스레드는 데이터가 도착할 때까지 아무것도 못 한다. 연결이 하나면 상관없다. 그런데 동시 접속이 만 개라면?

스레드당 하나: 스레드마다 스택이 필요하다. 기본 8MB × 10,000 = 80GB. 실제로는 스택이 다 안 쓰이지만, 그래도 커널 자원과 컨텍스트 스위칭 비용이 든다
컨텍스트 스위칭은 커널을 거치므로 마이크로초 단위 비용이 든다. 스레드 만 개가 서로 전환하면 그 자체로 CPU를 태운다

비동기의 발상은 이렇다. 기다리는 동안 스레드를 놔주자. 소켓이 아직 준비되지 않았으면 그 작업을 잠시 접어두고, 같은 스레드로 다른 작업을 진행한다. 데이터가 도착하면 접어둔 작업을 다시 편다.

이게 "I/O 바운드에는 비동기, CPU 바운드에는 스레드"라는 구분의 근거다. CPU를 계속 쓰는 작업은 접어둘 타이밍이 없으므로 비동기로 얻을 게 없다.

퓨처

Rust의 비동기 계산은 퓨처(Future) 트레잇으로 표현된다.

trait Future {
    type Output;
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>;
}

enum Poll<T> {
    Ready(T),
    Pending,
}

poll 을 호출하면 두 가지 중 하나가 온다.

Ready(value) — 끝났다. 값이 여기 있다
Pending — 아직이다. 나중에 다시 물어봐 달라

여기서 결정적인 특징이 나온다. Rust의 퓨처는 게으르다(lazy). 만들기만 해서는 아무 일도 일어나지 않고, 누군가 poll 을 불러야 진행된다. JavaScript의 Promise 가 생성 즉시 실행되는 것과 정반대다.

let fut = do_something();   // 아직 아무것도 실행되지 않았다
fut.await;                  // 이제 실행된다

처음에는 이게 불편해 보였는데, 이 성질 덕에 취소가 공짜다. 퓨처를 그냥 드롭하면 그 작업은 없던 일이 된다. 실행 중인 작업을 중간에 안전하게 멈추는 건 어려운 문제인데, 게으른 퓨처는 이걸 "poll을 그만 부르면 됨"으로 바꿔놓는다.

그리고 Pending 을 받은 쪽이 계속 물어보면(busy polling) 낭비다. 그래서 Context 안에 웨이커(Waker) 가 들어 있다. 퓨처는 "준비되면 이 웨이커를 호출해 달라"고 등록해두고 Pending 을 반환한다. 이 부분은 Part 2에서 직접 구현한다.

async 함수와 await 표현식

퓨처를 손으로 구현하는 건 고통스럽다. async 문법이 그걸 대신 해준다.

async fn read_line(socket: &mut TcpStream) -> io::Result<String> {
    let mut line = String::new();
    socket.read_line(&mut line).await?;
    Ok(line)
}

async fn 은 함수 본문을 상태 기계로 컴파일한다. 반환 타입은 io::Result<String> 이 아니라 impl Future<Output = io::Result<String>> 이다. .await 마다 상태 기계의 중단점이 하나씩 생기고, 지역 변수 중 중단점을 가로지르는 것들은 상태 기계의 필드가 된다.

즉, async fn 은 코루틴을 만드는 문법 설탕이다. 콜백 지옥 없이 순차적인 코드를 쓰면 컴파일러가 상태 기계로 바꿔준다. C#의 async/await와 같은 발상이지만, Rust는 힙 할당 없이 이걸 해낸다. 상태 기계 하나가 하나의 구조체이고, 크기가 컴파일 시점에 정해진다.

.await 는 async 문맥 안에서만 쓸 수 있다. 그래서 최상위에서는 다른 방법이 필요하다.

block_on

동기 세계와 비동기 세계의 다리다.

use async_std::task;

fn main() {
    let result = task::block_on(async {
        let socket = TcpStream::connect("127.0.0.1:8080").await?;
        // ...
        Ok(())
    });
}

block_on 은 퓨처가 완료될 때까지 현재 스레드를 블로킹하며 poll 을 반복한다. main 이 async 일 수 없으니 진입점에서 한 번 쓴다. #[tokio::main] 같은 속성 매크로도 결국 이걸 감싼 것이다.

주의할 게 있다. async 함수 안에서 block_on 을 부르면 안 된다. 그 스레드가 막히면서 같은 실행기 위의 다른 태스크들이 전부 멈춘다. 실제로 겪어보면 "왜 프로그램이 아무 반응이 없지?"로 나타난다.

비동기 태스크 생성하기

여러 작업을 동시에 진행하려면 태스크로 띄운다.

use async_std::task;

let handle = task::spawn(async {
    expensive_io().await
});

let result = handle.await;   // join과 비슷하지만 블로킹하지 않는다

thread::spawn 과 이름이 비슷하지만 완전히 다르다. OS 스레드를 만드는 게 아니라 실행기의 큐에 퓨처를 등록하는 것이다. 그래서 태스크 하나의 비용이 스레드보다 훨씬 싸다. 수십만 개를 띄워도 된다.

예제로 나온 HTTP 클라이언트가 이 구조를 잘 보여준다.

pub async fn many_requests(urls: &[String]) -> Vec<Result<String, surf::Error>> {
    let client = surf::Client::new();

    let mut handles = vec![];
    for url in urls {
        let request = client.get(&url).recv_string();
        handles.push(async_std::task::spawn(request));   // 요청들을 동시에 시작
    }

    let mut results = vec![];
    for handle in handles {
        results.push(handle.await);                      // 결과를 순서대로 수거
    }

    results
}

fn main() {
    let requests = &[
        "http://example.com".to_string(),
        "https://www.red-bean.com".to_string(),
        "https://en.wikipedia.org/wiki/Main_Page".to_string(),
    ];

    let results = async_std::task::block_on(many_requests(requests));
    for result in results {
        match result {
            Ok(response) => println!("*** {response}\n"),
            Err(err) => eprintln!("error: {err}\n"),
        }
    }
}

첫 루프에서 세 요청을 전부 시작하고, 둘째 루프에서 결과를 모은다. 세 요청이 동시에 날아가므로 총 소요 시간은 셋 중 가장 느린 것 하나다.

여기서 spawn 없이 그냥 request.await 를 첫 루프에 넣으면 어떻게 될까? 하나씩 순차적으로 기다리게 되어 시간이 다 더해진다. .await 는 동시성을 만들지 않는다. spawn (또는 join! 같은 조합자)이 만든다. 이걸 헷갈려서 "비동기로 바꿨는데 왜 안 빨라지지?"가 되는 경우가 많다.

async 블록

함수 전체가 아니라 일부만 비동기로 만들고 싶으면 async 블록을 쓴다.

let fut = async {
    let data = fetch().await?;
    process(data).await
};

블록도 퓨처를 만든다. spawn 에 넘길 때 특히 유용하다.

값을 소유해야 하면 async move 를 쓴다. 클로저의 move 와 같은 이유다 — 태스크가 주변 스코프보다 오래 살 수 있기 때문이다.

let name = String::from("Chris");
task::spawn(async move {
    println!("{name}");
});

async 함수 만들기

async fn 의 반환 타입을 직접 적어야 할 때가 있다. 트레잇 안에서 쓰거나, 조건에 따라 다른 퓨처를 반환할 때다.

// async fn과 대략 동등하다
fn read_line(socket: &mut TcpStream) -> impl Future<Output = io::Result<String>> + '_ {
    async move {
        // ...
    }
}

'_ 수명이 붙는 게 포인트다. 퓨처가 socket 을 빌리고 있으므로, 퓨처가 살아 있는 동안 socket 도 살아 있어야 한다. async fn 을 쓰면 컴파일러가 이걸 자동으로 처리해준다.

스레드 풀에서 실행하기

async_std::task::spawn 은 태스크를 스레드 풀에 올린다. 즉 태스크가 어느 스레드에서 실행될지 모르고, .await 마다 다른 스레드로 옮겨갈 수도 있다.

CPU를 오래 쓰는 작업을 비동기 태스크 안에서 하면 그 스레드가 막히므로, 그런 건 spawn_blocking 으로 따로 뺀다.

let result = task::spawn_blocking(|| {
    heavy_computation()      // 전용 스레드에서 실행
}).await;

Send를 구현해야 하는 이유

슬라이드에서 가장 많은 분량을 차지한 주제다. 그리고 실제로 비동기 Rust에서 제일 자주 막히는 지점이기도 하다.

태스크가 스레드 풀에서 돌고, .await 지점에서 다른 스레드로 옮겨갈 수 있다면, 태스크가 담고 있는 모든 것이 Send 여야 한다.

use std::rc::Rc;

task::spawn(async {
    let rc = Rc::new(42);
    do_something().await;      // 여기서 스레드가 바뀔 수 있다
    println!("{rc}");          // 에러: Rc는 Send가 아니다
});

핵심은 어떤 값이 .await 를 가로질러 살아 있는가다. .await 이전에 드롭되는 값은 상태 기계에 저장되지 않으므로 문제가 없다.

task::spawn(async {
    {
        let rc = Rc::new(42);
        println!("{rc}");
    }                          // rc가 여기서 드롭됨
    do_something().await;      // OK — rc는 이 지점을 넘지 않는다
});

에러 메시지가 future cannot be sent between threads safely 라고 나오는데, 어떤 값이 문제인지 컴파일러가 짚어준다. 처음엔 이 메시지가 무섭게 느껴졌지만, ".await 를 넘는 값 중에 Send 가 아닌 게 있다"는 한 문장으로 정리하고 나니 읽을 만해졌다.

MutexGuard 도 같은 이유로 자주 걸린다.

let guard = mutex.lock().unwrap();
do_something().await;              // 락을 잡은 채로 await → 위험

이건 Send 문제이기도 하고, 그보다 설계 문제다. 락을 잡은 채 .await 하면 그동안 다른 태스크가 락을 못 잡는다. tokio::sync::Mutex 처럼 비동기용 뮤텍스를 쓰거나, 락 스코프를 .await 전에 끝내야 한다.

오래 걸리는 계산과 비동기 설계 전략

정리하면 이렇다.

작업 종류	도구
I/O 대기 (네트워크, 파일)	`async` / `.await`
짧은 CPU 작업	그냥 `async` 안에서
긴 CPU 작업	`spawn_blocking` 또는 별도 스레드
데이터 병렬 처리	Rayon

비동기 실행기의 스레드를 CPU 작업으로 막으면 다른 모든 태스크가 굶는다. 이건 비동기 런타임에서 가장 흔한 성능 사고다.

진짜 비동기 채팅 클라이언트

회차 후반은 채팅 애플리케이션을 만든다. 클라이언트가 이번 회차, 서버가 Part 2다.

프로토콜부터. 클라이언트와 서버가 주고받는 메시지를 enum 으로 정의하고 serde로 JSON 직렬화한다.

#[derive(Debug, Serialize, Deserialize, PartialEq)]
pub enum FromClient {
    Join { group_name: Arc<String> },
    Post { group_name: Arc<String>, message: Arc<String> },
}

#[derive(Debug, Serialize, Deserialize, PartialEq)]
pub enum FromServer {
    Message { group_name: Arc<String>, message: Arc<String> },
    Error(String),
}

Arc<String> 을 쓴 게 눈에 띈다. 서버가 한 메시지를 그룹의 모든 구독자에게 보낼 때, 문자열을 복사하지 않고 Arc 를 복제해서 나눠주기 위해서다. serde의 rc 기능을 켜면 Arc 도 직렬화된다.

오류 타입은 Box<dyn Error + Send + Sync + 'static> 으로 뭉뚱그린다.

pub type ChatError = Box<dyn Error + Send + Sync + 'static>;
pub type ChatResult<T> = Result<T, ChatError>;

Send + Sync 가 붙은 이유가 이제 명확하다. 오류가 태스크 경계를 넘어야 하기 때문이다. 오류 처리 회차의 Box<dyn Error> 에 비동기 요구사항이 더해진 형태다.

패킷 보내기와 받기.

pub async fn send_as_json<S, P>(outbound: &mut S, packet: &P) -> ChatResult<()>
where
    S: async_std::io::Write + Unpin,
    P: Serialize,
{
    let mut json = serde_json::to_string(&packet)?;
    json.push('\n');
    outbound.write_all(json.as_bytes()).await?;
    Ok(())
}

pub fn receive_as_json<S, P>(inbound: S) -> impl Stream<Item = ChatResult<P>>
where
    S: async_std::io::BufRead + Unpin,
    P: DeserializeOwned,
{
    inbound.lines().map(|line_result| -> ChatResult<P> {
        let line = line_result?;
        let parsed = serde_json::from_str::<P>(&line)?;
        Ok(parsed)
    })
}

receive_as_json 이 반환하는 Stream 이 중요한 개념이다. Future 가 값 하나를 비동기적으로 내놓는다면, Stream 은 값을 여러 개 비동기적으로 내놓는다. 이터레이터의 비동기 버전이라고 보면 된다. .next().await 로 하나씩 꺼낸다.

한 줄에 JSON 하나(개행 구분)라는 단순한 프레이밍을 쓰는 것도 눈여겨볼 만하다. lines() 스트림이 프레이밍을 그대로 처리해준다.

클라이언트 메인.

fn main() -> ChatResult<()> {
    let address = std::env::args().nth(1).expect("Usage: client ADDRESS:PORT");

    task::block_on(async {
        let socket = net::TcpStream::connect(address).await?;
        socket.set_nodelay(true)?;

        let to_server = send_commands(socket.clone());
        let from_server = handle_replies(socket);

        from_server.race(to_server).await?;

        Ok(())
    })
}

이 구조가 깔끔하다. 클라이언트는 두 가지를 동시에 해야 한다.

표준 입력을 읽어 서버로 보내기 (send_commands)
서버에서 오는 메시지를 화면에 출력하기 (handle_replies)

race 는 둘 중 먼저 끝나는 쪽의 결과를 취하고 나머지는 버린다. 사용자가 Ctrl-D를 눌러 입력이 끝나거나 서버 연결이 끊기거나, 어느 쪽이든 하나가 끝나면 클라이언트도 끝난다. 스레드로 짰다면 종료 신호를 주고받는 코드가 필요했을 텐데, 비동기에서는 조합자 하나로 표현된다.

send_commands 안쪽도 보자.

async fn send_commands(mut to_server: net::TcpStream) -> ChatResult<()> {
    let mut command_lines = io::BufReader::new(io::stdin()).lines();

    while let Some(command_result) = command_lines.next().await {
        let command = command_result?;
        let request = match parse_command(&command) {
            Some(request) => request,
            None => continue,
        };

        utils::send_as_json(&mut to_server, &request).await?;
        to_server.flush().await?;
    }

    Ok(())
}

while let Some(x) = stream.next().await 가 비동기 스트림 순회의 관용구다. 동기 코드의 for 루프와 모양이 거의 같은데, 각 반복이 .await 에서 접힐 수 있다. 비동기 코드가 동기 코드처럼 읽힌다는 게 async/await의 핵심 가치라는 걸 여기서 실감했다.

정리

비동기는 I/O 대기 시간을 회수하는 기법이다. CPU 바운드에는 도움이 안 된다
Rust의 퓨처는 게을러서 poll하지 않으면 아무 일도 안 일어나고, 그래서 취소가 쉽다
async fn 은 함수를 상태 기계로 컴파일한다. 힙 할당 없음
.await 는 동시성을 만들지 않는다. spawn 이나 race/join 같은 조합자가 만든다
태스크가 스레드 풀에서 돌므로 .await 를 가로지르는 값은 Send 여야 한다
Stream 은 Future 의 다중 값 버전이자 이터레이터의 비동기 버전

다음 편은 비동기 프로그래밍 Part 2 — 채팅 서버, 그리고 block_on 과 spawn_blocking 을 직접 구현하며 웨이커와 Pin 을 파고든다.