Rust async 깊이 알아보기: select!, 타임아웃, 스트림, Pin
async/await 페이지에서 기초를 배웠어요. async fn은 게으른 퓨처(future)를 돌려주고, .await가 그걸 실행하고, #[tokio::main]이 런타임을 시작하고, tokio::join!과 tokio::spawn으로 여러 일을 동시에 돌릴 수 있었죠. 이 중에 낯선 게 있다면 그 페이지를 먼저 읽어 보세요. 이 페이지는 거기서 이어지는 내용이에요.
실제 비동기 프로그램에는 도구가 몇 가지 더 필요해요. 바쁜 주방을 다시 떠올려 볼게요. 어떤 때는 모든 요리가 나올 때까지 기다려요(join!). 어떤 때는 먼저 일어나는 일 하나만 중요해요. 음식이 나오든지, 손님이 가 버리든지요(select!). 어떤 때는 5분 기다리다 포기해요(timeout). 그리고 어떤 때는 저녁 내내 주문이 하나씩 계속 들어와요(스트림). 이 페이지에서는 이걸 전부 다루고, 언젠가 에러 메시지에서 꼭 마주치게 될 낯선 단어 Pin도 쉬운 말로 설명해요.
모든 예제는 기초 페이지와 똑같이 features = ["full"]을 켠 tokio를 써요. Rust Playground에 tokio가 기본으로 들어 있어서 “직접 실행해 보기” 버튼도 잘 동작해요.
여러 일을 한꺼번에 실행하기
섹션 제목: “여러 일을 한꺼번에 실행하기”tokio::join!은 이미 알고 있죠. 모든 퓨처가 끝날 때까지 기다렸다가 결과를 튜플로 돌려줘요. 그 형제인 tokio::try_join!은 Result를 돌려주는 퓨처를 위한 거예요. 모든 값이 담긴 Ok를 주거나, 가장 먼저 나온 Err를 줘요. 그리고 하나라도 실패하면 나머지는 더 기다리지 않아요.
use std::time::Duration;use tokio::time::sleep;
async fn load_profile(id: u32) -> Result<String, String> { sleep(Duration::from_millis(20)).await; Ok(format!("프로필 #{id}"))}
async fn load_settings(id: u32) -> Result<String, String> { sleep(Duration::from_millis(10)).await; if id == 2 { return Err(format!("사용자 {id}의 설정이 없어요")); } Ok(String::from("다크 모드"))}
#[tokio::main]async fn main() { for id in [1, 2] { match tokio::try_join!(load_profile(id), load_settings(id)) { Ok((profile, settings)) => println!("사용자 {id}: {profile}, {settings}"), Err(e) => println!("사용자 {id} 불러오기 실패: {e}"), } }}실행 결과:
사용자 1: 프로필 #1, 다크 모드사용자 2 불러오기 실패: 사용자 2의 설정이 없어요join!과 try_join!은 퓨처들을 현재 태스크 안에서 번갈아 가며 실행해요. 미리 알고 있는 몇 개의 퓨처에는 딱 좋아요. 하지만 작업 목록의 크기를 실행 중에야 알 수 있다면, 각각을 별도의 태스크로 spawn하세요. 기초 페이지에서처럼 JoinHandle을 Vec에 하나씩 넣어도 돼요. 그런데 tokio에는 이럴 때 쓰기 좋은 도구가 있어요. 바로 JoinSet이에요. 태스크를 담는 주머니 같은 건데, 태스크가 끝나는 대로 결과를 하나씩 건네줘요.
use std::time::Duration;use tokio::task::JoinSet;use tokio::time::sleep;
async fn check_site(site: &'static str, ms: u64) -> (&'static str, u64) { sleep(Duration::from_millis(ms)).await; // 네트워크 호출이라고 생각해 주세요 (site, ms)}
#[tokio::main]async fn main() { let mut set = JoinSet::new(); for (site, ms) in [("rust-lang.org", 30), ("docs.rs", 10), ("crates.io", 20)] { set.spawn(check_site(site, ms)); }
let mut results = Vec::new(); while let Some(res) = set.join_next().await { results.push(res.unwrap()); // 태스크가 패닉했을 때만 Err }
results.sort(); // 끝나는 순서는 실행할 때마다 달라질 수 있어요 for (site, ms) in results { println!("{site} 응답 시간: {ms}ms"); }}실행 결과:
crates.io 응답 시간: 20msdocs.rs 응답 시간: 10msrust-lang.org 응답 시간: 30msjoin_next()는 주머니가 비면 None을 돌려주니까, while let 반복문은 알아서 끝나요. 결과는 spawn한 순서가 아니라 끝난 순서대로 와요. 걸리는 시간이 비슷한 태스크끼리는 그 순서가 실행할 때마다 바뀔 수 있어서, 출력하기 전에 정렬했어요. 좋은 점이 하나 더 있어요. JoinSet을 드롭하면 그 안에 남아 있던 태스크가 모두 취소돼요. 실수로 뭔가가 계속 돌아가는 일이 없어요.
select!로 퓨처끼리 경주시키기
섹션 제목: “select!로 퓨처끼리 경주시키기”join!은 전부를 기다려요. tokio::select!는 가장 먼저 끝나는 하나를 기다려요. 각 분기는 패턴, 퓨처, 코드 블록으로 이루어져요. 패턴 = 퓨처 => 코드 모양이에요. 어떤 퓨처든 가장 먼저 끝난 쪽의 코드가 실행되고, 나머지 분기는 취소돼요.
Rust에서 취소된다는 건 무슨 뜻일까요? 진 퓨처는 그냥 드롭돼요. 스코프를 벗어나는 다른 값과 똑같아요. 다시는 실행되지 않아요. 그 안의 지역 변수도 함께 드롭되니까, 정리 코드도 실행돼요. 이 예제는 작은 Drop 타입으로 그걸 눈에 보이게 해요.
use std::time::Duration;use tokio::time::sleep;
struct Download(&'static str);
impl Drop for Download { fn drop(&mut self) { println!("{} 다운로드 정리 중", self.0); }}
async fn download(mirror: &'static str, ms: u64) -> &'static str { let _download = Download(mirror); sleep(Duration::from_millis(ms)).await; println!("{mirror} 다운로드 완료"); mirror}
#[tokio::main]async fn main() { tokio::select! { winner = download("유럽", 50) => println!("승자: {winner}"), winner = download("아시아", 1000) => println!("승자: {winner}"), } println!("main은 계속 진행돼요");}실행 결과:
유럽 다운로드 완료유럽 다운로드 정리 중아시아 다운로드 정리 중승자: 유럽main은 계속 진행돼요빠진 줄이 보이나요? “아시아 다운로드 완료”는 한 번도 출력되지 않아요. 아시아 퓨처는 .await에서 멈춘 채 드롭됐고, 그 안의 Download 값도 곧바로 정리됐어요. 비동기 Rust에서 취소는 어디서나 이렇게 동작해요. 퓨처를 더 이상 폴링(poll)하지 않고 드롭하는 것이에요. 따로 “취소” 함수를 부를 필요가 없어요.
select!의 대표적인 쓰임새는 타임아웃이에요. 하려는 작업과 타이머를 경주시키는 거죠.
tokio::select! { report = build_report() => println!("{report}"), _ = tokio::time::sleep(Duration::from_secs(5)) => println!("너무 느려서 포기할게요"),}이것도 잘 동작하지만, 더 간단한 방법이 있어요. 다음 섹션에서 다룰 tokio::time::timeout이에요.
종료 신호가 올 때까지 일하기
섹션 제목: “종료 신호가 올 때까지 일하기”select!는 loop 안에서 진가를 발휘해요. 여기서는 워커(worker)가 채널에서 작업을 받으면서, 동시에 두 번째 채널로 “이제 그만” 메시지가 오는지도 지켜봐요. oneshot 채널은 메시지를 딱 하나만 실어 나르기 때문에, 종료 신호로 쓰기에 안성맞춤이에요.
use std::time::Duration;use tokio::sync::{mpsc, oneshot};use tokio::time::sleep;
#[tokio::main]async fn main() { let (job_tx, mut job_rx) = mpsc::channel::<String>(10); let (stop_tx, mut stop_rx) = oneshot::channel::<()>();
let worker = tokio::spawn(async move { let mut done = 0; loop { tokio::select! { Some(job) = job_rx.recv() => { println!("작업 중: {job}"); done += 1; } _ = &mut stop_rx => { println!("종료 신호를 받았어요"); break; } } } done });
for job in ["사진 크기 조정", "이메일 보내기", "검색 색인 갱신"] { job_tx.send(job.to_string()).await.unwrap(); } sleep(Duration::from_millis(100)).await; // 워커가 일을 마칠 시간을 줘요 stop_tx.send(()).unwrap();
let done = worker.await.unwrap(); println!("워커가 작업 {done}개를 끝냈어요");}실행 결과:
작업 중: 사진 크기 조정작업 중: 이메일 보내기작업 중: 검색 색인 갱신종료 신호를 받았어요워커가 작업 3개를 끝냈어요알아 두면 좋은 점이 몇 가지 있어요.
stop_rx가 아니라&mut stop_rx라고 썼어요. 반복문을 돌 때마다select!에는 같은 수신자가 다시 필요해요. 그래서 첫 번째 바퀴에 move해 버리지 않고 빌려주는 거예요.- 반복문을 돌 때마다
job_rx.recv()는 새 퓨처를 만들고, 진 쪽은 드롭돼요. 그래도 괜찮아요.recv()는 취소 안전(cancel-safe)하거든요. 드롭해도 메시지를 잃어버리지 않아요. - 패턴이 맞지 않으면(여기서는 송신자가 모두 사라져서
recv()가None을 돌려줄 때),select!는 그 분기를 끄고 나머지 분기를 계속 기다려요. - 여러 분기가 동시에 준비되면
select!는 그중 하나를 무작위로 골라요. 그래서 한 분기가 반복문을 독차지할 수 없어요. 위에서부터 차례대로 확인하게 하려면select!안의 첫 줄에biased;라고 쓰세요.
취소 안전성, 한 문단으로 정리. 진 분기는 드롭되니까, 그 분기가 하다 만 일은 버려져요. 채널의 recv()나 sleep()이라면 문제없어요. 아직 아무것도 가져가지 않았으니까요. 그런데 네트워크 소켓에서 100바이트를 읽는 퓨처가 60바이트까지 읽은 상태에서 드롭된다고 생각해 보세요. 그 60바이트는 사라져요. 이런 퓨처는 취소 안전하지 않아요. tokio 문서에는 메서드마다 취소 안전한지가 적혀 있어요. 반복문 안에서 select!를 쓸 때는 호출하는 메서드를 확인하거나, 퓨처를 반복문 밖에서 한 번만 만들고 그 퓨처를 계속 폴링하세요(방법은 Pin 섹션에서 볼 거예요).
타임아웃과 타이머
섹션 제목: “타임아웃과 타이머”tokio::time::timeout(duration, future)는 어떤 퓨처든 감쌀 수 있어요. 퓨처가 제시간에 끝나면 Ok(value)를 받아요. 시간이 먼저 다 되면 퓨처는 드롭되고(select!에서 진 분기처럼 취소돼요) Err(Elapsed)를 받아요.
use std::time::Duration;use tokio::time::{sleep, timeout};
async fn fetch_price(server_delay_ms: u64) -> u32 { sleep(Duration::from_millis(server_delay_ms)).await; 42}
#[tokio::main]async fn main() { let limit = Duration::from_millis(200);
for (server, delay) in [("빠른 서버", 20), ("느린 서버", 2000)] { match timeout(limit, fetch_price(delay)).await { Ok(price) => println!("{server}: 가격은 {price}"), Err(e) => println!("{server}: {e}"), } }}실행 결과:
빠른 서버: 가격은 42느린 서버: deadline has elapseddeadline has elapsed는 tokio가 Elapsed 에러를 출력할 때 쓰는 메시지예요. “제한 시간이 지났다”는 뜻이고, tokio가 만든 문장이라 영어로 나와요.
바깥 세상과 통신하는 일에는 모두 타임아웃을 거세요. HTTP 호출, 데이터베이스 쿼리, 소켓 읽기 같은 것들이요. 응답하지 않는 서버 하나 때문에 프로그램 전체가 영원히 멈춰서는 안 되니까요.
sleep과 interval의 차이. sleep은 한 번 기다려요. interval은 메트로놈처럼 일정한 박자를 만들어 줘요. 반복문 안에서 .tick().await를 부르면, 틱이 예정된 시간마다 와요. 첫 번째 틱은 바로 와요.
use std::time::Duration;use tokio::time::interval;
#[tokio::main]async fn main() { let mut heartbeat = interval(Duration::from_millis(50));
for beat in 1..=3 { heartbeat.tick().await; // 첫 번째 틱은 곧바로 끝나요 println!("하트비트 {beat}"); }}실행 결과:
하트비트 1하트비트 2하트비트 3그냥 반복문에서 50ms씩 sleep하면 안 될까요? 그러면 한 바퀴가 “하는 일 + 50ms”만큼 걸려서, 일정이 조금씩 밀려요. interval은 각 틱의 예정된 시작 시각을 기준으로 재요. 그래서 하는 일이 10ms 걸리면 남은 40ms만 기다려요. 하트비트, 폴링, “1분마다 저장” 같은 작업에는 interval을, 한 번 쉬어 갈 때는 sleep을 쓰세요.
스트림: 비동기 반복자
섹션 제목: “스트림: 비동기 반복자”반복자는 next()로 값을 하나씩 줘요. 스트림(stream)은 그 비동기 버전이에요. next().await는 다음 값이 준비되면 그 값을 주고, 더 없으면 None을 줘요. 주식 시세 알림, 데이터베이스에서 도착하는 행, 채팅 서버에서 오는 메시지를 떠올려 보세요.
사실 스트림 비슷한 걸 이미 써 봤어요. tokio 채널의 수신자예요. rx.recv().await는 다음 메시지를 기다리고, 송신자가 모두 사라지면 None을 돌려줘요.
use std::time::Duration;use tokio::sync::mpsc;use tokio::time::sleep;
#[tokio::main]async fn main() { let (tx, mut rx) = mpsc::channel(8);
tokio::spawn(async move { for temp in [21, 22, 24, 23] { tx.send(temp).await.unwrap(); sleep(Duration::from_millis(10)).await; // 10ms마다 한 번씩 측정해요 } // 여기서 tx가 드롭되면서 채널이 닫혀요 });
while let Some(temp) = rx.recv().await { println!("온도: {temp}°C"); } println!("센서가 멈췄어요");}실행 결과:
온도: 21°C온도: 22°C온도: 24°C온도: 23°C센서가 멈췄어요표준 라이브러리에는 아직 Stream 트레이트가 없어요(아직 설계 중이에요). 생태계에서는 futures 크레이트의 트레이트를 쓰고, tokio의 도우미 크레이트인 tokio-stream이 반복자에서 익숙한 메서드를 담은 StreamExt 트레이트를 더해 줘요. next(), map(), filter(), take() 같은 것들이요. 프로젝트에 추가해 볼게요.
[dependencies]tokio = { version = "1", features = ["full"] }tokio-stream = "0.1"use tokio_stream::{self as stream, StreamExt};
#[tokio::main]async fn main() { let prices = stream::iter(vec![100, 102, 98, 105, 110]);
let mut doubled = prices.map(|p| p * 2).take(3);
while let Some(p) = doubled.next().await { println!("{p}"); }}이 코드는 200, 204, 196을 출력해요. next() 뒤에 .await가 붙는 것만 빼면 반복자 코드와 똑같이 읽히죠. tokio 채널 수신자를 진짜 스트림으로 바꾸려면 tokio_stream::wrappers::ReceiverStream::new(rx)로 감싸면 돼요. (futures 크레이트에도 같은 아이디어의 StreamExt가 따로 있어요. 두 트레이트 모두 next() 메서드가 있어서 충돌하니까, 파일 하나에서는 한쪽만 쓰세요.)
Pin, 쉬운 말로
섹션 제목: “Pin, 쉬운 말로”언젠가는 타입이나 에러 메시지에서 Pin을 보게 될 거예요. 어려운 이론은 빼고 핵심 아이디어만 살펴볼게요.
Rust는 async fn을 컴파일할 때 숨겨진 구조체, 즉 상태 기계(state machine)로 바꿔요. .await 뒤에도 필요한 지역 변수는 모두 그 구조체의 필드가 돼요. 이제 지역 변수 하나가 다른 지역 변수를 빌린다고 해 볼게요. let name = String::from("Ana"); let r = &name; 사이에 .await가 있는 경우처럼요. 그러면 구조체가 자기 자신의 필드를 가리키는 포인터를 갖게 돼요. 이 구조체를 메모리의 다른 곳으로 옮기면, 그 포인터는 예전의 빈 자리를 가리키게 되겠죠.
Pin이 바로 그 해결책이에요. 고정(pin)된 값은 “이 값은 이제 메모리에서 움직이지 않는다“라는 약속이에요. 퓨처가 고정되고 나면 안전하게 폴링할 수 있어요. 대부분은 신경 쓸 일이 없어요. .await가 알아서 퓨처를 고정해 주거든요.
Pin은 다음 세 가지 상황에서 만나게 돼요.
- 같은 퓨처를 여러 번 폴링할 때. 예를 들어
select!반복문 안의 마감 시간(deadline)이요. - 퓨처를 저장할 때.
Vec이나 구조체 필드에 넣는 경우로, 보통Pin<Box<dyn Future<Output = T>>>모양이에요. async fn안에서 재귀를 할 때.
첫 번째 상황부터 볼게요. 반복문에서 sleep을 고정하지 않고 다시 쓰려고 하면 이렇게 돼요.
let deadline = sleep(Duration::from_millis(300));
loop { tokio::select! { _ = &mut deadline => break, Some(n) = rx.recv() => println!("{n}"), }}error[E0277]: `PhantomPinned` cannot be unpinned | = note: consider using the `pin!` macro consider using `Box::pin` if you need to access the pinned value outside of the current scope = note: required for `Sleep` to implement `Unpin`“PhantomPinned은 고정을 풀 수 없다”, 즉 Sleep이 Unpin이 아니라는 에러예요. 컴파일러가 해결책까지 알려 주고 있어요. std::pin::pin!(또는 tokio::pin!)은 퓨처를 스택 위 그 자리에 고정해요. 이제 반복문의 모든 바퀴가 같은 마감 시간을 함께 써요.
use std::pin::pin;use std::time::Duration;use tokio::sync::mpsc;use tokio::time::sleep;
#[tokio::main]async fn main() { let (tx, mut rx) = mpsc::channel(8); tokio::spawn(async move { for order in 1..=3 { tx.send(order).await.unwrap(); sleep(Duration::from_millis(10)).await; } sleep(Duration::from_secs(5)).await; // 그다음엔 가게가 조용해져요 });
let mut deadline = pin!(sleep(Duration::from_millis(300)));
loop { tokio::select! { _ = &mut deadline => { println!("영업 종료 시간이에요"); break; } Some(order) = rx.recv() => println!("주문 #{order} 접수"), } }}실행 결과:
주문 #1 접수주문 #2 접수주문 #3 접수영업 종료 시간이에요sleep(...)을 select! 안에 썼다면, 주문이 새로 들어올 때마다 300ms 타이머가 새로 시작돼서 바쁜 가게는 영영 문을 닫지 못했을 거예요. 마감 시간 하나를 반복문 밖에서 고정해 두면 이 문제가 해결돼요.
재귀의 경우, 컴파일러는 퓨처의 크기를 알아야 하는데, 자기 자신을 품은 퓨처는 크기가 무한히 커져요. async fn countdown 안에서 countdown(n - 1).await를 부르면 이런 에러가 나요.
error[E0733]: recursion in an async fn requires boxing | = note: a recursive `async fn` call must introduce indirection such as `Box::pin` to avoid an infinitely sized future“async fn의 재귀에는 박싱이 필요하다”는 뜻이에요. Box::pin은 안쪽 퓨처를 힙으로 옮기고 거기에 고정해요. 그러면 바깥 퓨처는 크기가 정해진 포인터 하나만 갖게 돼요.
async fn countdown(n: u32) { if n == 0 { println!("발사!"); return; } println!("{n}..."); Box::pin(countdown(n - 1)).await;}
#[tokio::main]async fn main() { countdown(3).await;}실행 결과:
3...2...1...발사!솔직히 대부분의 사람에게는 이 정도면 충분해요. 여러 번 폴링하는 퓨처에는 pin!, 퓨처를 저장하거나 재귀할 때는 Box::pin이에요. 전체 이야기가 궁금하다면 ’더 알아보기’에 있는 Rust Book 장이 차근차근 설명해 줘요.
트레이트의 async fn과 비동기 클로저
섹션 제목: “트레이트의 async fn과 비동기 클로저”Rust 1.75부터는 트레이트 안에 async fn을 바로 쓸 수 있어요. Rust 1.85부터는 비동기 클로저도 생겼어요. async |x| ...라고 쓰고, 짝이 되는 트레이트 AsyncFn, AsyncFnMut, AsyncFnOnce도 있어요(클로저 트레이트의 비동기 버전이에요). 둘 다 한 예제에 담아 볼게요.
use std::time::Duration;use tokio::time::sleep;
trait Notifier { async fn notify(&self, user: &str) -> String;}
struct Email;
impl Notifier for Email { async fn notify(&self, user: &str) -> String { sleep(Duration::from_millis(10)).await; // 메일 서버와 통신하는 척해요 format!("{user}님에게 이메일을 보냈어요") }}
async fn for_each_user(users: &[&str], job: impl AsyncFn(&str) -> String) { for user in users { println!("{}", job(user).await); }}
#[tokio::main]async fn main() { let email = Email; let users = ["민지", "준호"];
for_each_user(&users, async |user: &str| email.notify(user).await).await;}실행 결과:
민지님에게 이메일을 보냈어요준호님에게 이메일을 보냈어요트레이트의 async fn에는 함정이 하나 있어요. 트레이트는 돌려주는 퓨처가 Send라고 약속하지 않는데, tokio::spawn은 Send인 퓨처가 필요해요(스레드 사이를 옮겨 다닐 수 있으니까요). 그래서 제네릭 N: Notifier를 통한 호출을 spawn하면 실패해요.
fn send_in_background<N: Notifier + Send + Sync + 'static>(notifier: N) { tokio::spawn(async move { let msg = notifier.notify("ana").await; println!("{msg}"); });}error: future cannot be sent between threads safely |7 | let msg = notifier.notify("ana").await; | ^^^^^^^^^^^^^^^^^^^^^^ await occurs here on type `impl Future<Output = String>`, which is not `Send` |help: `Send` can be made part of the associated future's guarantees for all implementations of `Notifier::notify` |2 - async fn notify(&self, user: &str) -> String;2 + fn notify(&self, user: &str) -> impl std::future::Future<Output = String> + Send;“퓨처를 스레드 사이로 안전하게 보낼 수 없다”는 에러예요. 보통은 컴파일러가 제안하는 대로 고치면 돼요. 트레이트 메서드를 fn ... -> impl Future<Output = String> + Send로 쓰는 거예요. 구현하는 쪽은 여전히 평범한 async fn으로 쓸 수 있어요. 그리고 async fn이 있는 트레이트는 아직 dyn Notifier 같은 트레이트 객체로 쓸 수 없다는 점도 기억해 두세요. 그게 필요하다면 많이 쓰이는 async-trait 크레이트가 퓨처를 대신 박싱해 줘요.
자주 하는 실수
섹션 제목: “자주 하는 실수”런타임을 블로킹하기. 비동기 태스크 안에서 std::thread::sleep이나 무거운 계산, 큰 std::fs 읽기를 하면 워커 스레드 전체가 멈추고, 그 스레드 위의 다른 태스크가 모두 기다리게 돼요. 기초 페이지에서는 sleep 버전을 봤죠. CPU를 많이 쓰거나 블로킹하는 일은 tokio::task::spawn_blocking에 맡기세요. 그런 일을 위해 따로 마련된 스레드 풀에서 실행해 줘요.
fn count_primes(limit: u64) -> usize { (2..limit) .filter(|&n| (2..).take_while(|d| d * d <= n).all(|d| n % d != 0)) .count()}
#[tokio::main]async fn main() { let handle = tokio::task::spawn_blocking(|| count_primes(50_000));
println!("소수를 세는 동안에도 요청에 계속 응답하고 있어요...");
let primes = handle.await.unwrap(); println!("50000보다 작은 소수는 {primes}개예요");}실행 결과:
소수를 세는 동안에도 요청에 계속 응답하고 있어요...50000보다 작은 소수는 5133개예요.await를 넘어서까지 std::sync::Mutex 가드를 쥐고 있기. 이 코드는 별문제 없어 보여요.
let handle = tokio::spawn(async move { let mut count = visits.lock().unwrap(); *count += 1; save_to_disk(*count).await;});error: future cannot be sent between threads safely | = help: within `{async block@src/main.rs:13:31: 13:41}`, the trait `Send` is not implemented for `std::sync::MutexGuard<'_, u32>`note: future is not `Send` as this value is used across an await |14 | let mut count = visits.lock().unwrap(); | --------- has type `std::sync::MutexGuard<'_, u32>` which is not `Send`15 | *count += 1;16 | save_to_disk(*count).await; | ^^^^^ await occurs here, with `mut count` maybe used later컴파일러가 여러분을 지켜 주는 거예요. 이 태스크가 .await에서 쉬는 동안에도 락을 계속 쥐고 있게 되고, 그 락을 원하는 다른 태스크는 자기 스레드 전체를 블로킹하게 돼요. 해결책은 .await 전에 가드를 드롭하는 거예요. 락을 거는 부분을 별도의 블록에 넣으면 돼요.
let count = { let mut count = visits.lock().unwrap(); *count += 1; *count}; // 여기서 가드가 드롭돼요save_to_disk(count).await;정말로 락을 쥔 채 기다려야 한다면 tokio::sync::Mutex를 쓰세요. 이 뮤텍스의 lock().await는 스레드를 블로킹하지 않고 기다려요. (락에 대해 더 알고 싶다면 공유 상태를 보세요.)
.await 빠뜨리기. 퓨처는 게으르기 때문에, .await 없이 save_file();이라고만 쓰면 아무 일도 일어나지 않아요. 컴파일러가 “futures do nothing unless you .await or poll them”(퓨처는 .await하거나 폴링하지 않으면 아무것도 하지 않는다)이라고 경고해 줘요. 자세한 내용은 async/await를 보고, 이 경고는 절대 무시하지 마세요.
태스크를 spawn해 놓고 기다리지 않기. main이 반환되면 런타임이 종료되고, 끝나지 않은 태스크는 모두 드롭돼요.
use std::time::Duration;use tokio::time::sleep;
#[tokio::main]async fn main() { tokio::spawn(async { sleep(Duration::from_millis(100)).await; println!("보고서를 저장했어요"); // 절대 출력되지 않아요! });
println!("main이 끝났어요");}실행 결과:
main이 끝났어요보고서는 끝내 저장되지 않아요. 중요한 작업이라면 JoinHandle을 보관해 두었다가 main이 끝나기 전에 .await하세요(또는 태스크들을 JoinSet에 넣고 모두 꺼낼 때까지 기다리세요).
연습 문제
섹션 제목: “연습 문제”1. 제한 시간이 있는 가격 비교
섹션 제목: “1. 제한 시간이 있는 가격 비교”가게 네 곳이 가격 문의에 각각 20, 40, 300, 60밀리초 뒤에 답해요. JoinSet으로 네 곳에 동시에 물어보되, 가게마다 150ms만 기다려 주세요. 도착한 가격을 가게 이름순으로 정렬해서 출력하고, 시간 초과된 가게가 몇 곳인지도 출력하세요. 힌트: timeout(limit, ask_shop(name, delay))를 주머니에 spawn하세요. 끝난 태스크마다 Ok나 Err(Elapsed)가 나오니까, 그 결과와 함께 가게 이름도 돌려주세요.
정답 보기
use std::time::Duration;use tokio::task::JoinSet;use tokio::time::{sleep, timeout};
async fn ask_shop(delay_ms: u64) -> u32 { sleep(Duration::from_millis(delay_ms)).await; 100 + delay_ms as u32 // 지어낸 가격이에요}
#[tokio::main]async fn main() { let shops = [("동네 슈퍼", 20), ("메가마트", 40), ("느림보 상회", 300), ("잡화점", 60)]; let limit = Duration::from_millis(150);
let mut set = JoinSet::new(); for (name, delay) in shops { set.spawn(async move { (name, timeout(limit, ask_shop(delay)).await) }); }
let mut prices = Vec::new(); let mut timed_out = 0; while let Some(res) = set.join_next().await { match res.unwrap() { (name, Ok(price)) => prices.push((name, price)), (_, Err(_)) => timed_out += 1, } }
prices.sort(); for (name, price) in prices { println!("{name}: {price}"); } println!("시간 초과된 가게: {timed_out}곳");}실행 결과:
동네 슈퍼: 120메가마트: 140잡화점: 160시간 초과된 가게: 1곳모든 가게가 동시에 기다리기 때문에, 느림보 상회가 아무리 느려도 전체 확인은 150ms 정도면 끝나요. 타임아웃이 걸리면 그 퓨처는 드롭되니까, 뒤에서 계속 돌아갈 수도 없어요.
2. 비동기 재귀로 폴더 크기 구하기
섹션 제목: “2. 비동기 재귀로 폴더 크기 구하기”Folder에는 이름, KB 단위 파일 크기 목록, 하위 폴더 목록이 있어요. async fn total_size(folder: &Folder) -> u64를 작성해서, 폴더 자신의 파일 크기와 모든 하위 폴더의 크기를 더하세요. 하위 폴더마다 자기 자신을 호출하면 돼요. 각 폴더는 먼저 tokio::task::yield_now().await로 짧은 비동기 “디스크 읽기”를 한다고 칠게요. 힌트: 재귀 호출을 Box::pin(...)으로 감싸지 않으면 E0733 에러가 나요.
정답 보기
struct Folder { name: &'static str, files_kb: Vec<u64>, subfolders: Vec<Folder>,}
async fn total_size(folder: &Folder) -> u64 { tokio::task::yield_now().await; // 디스크를 읽는 척해요
let mut total: u64 = folder.files_kb.iter().sum(); for sub in &folder.subfolders { total += Box::pin(total_size(sub)).await; } println!("{}: {} KB", folder.name, total); total}
#[tokio::main]async fn main() { let home = Folder { name: "home", files_kb: vec![4], subfolders: vec![ Folder { name: "photos", files_kb: vec![2048, 1024], subfolders: vec![] }, Folder { name: "code", files_kb: vec![12, 8], subfolders: vec![Folder { name: "target", files_kb: vec![500], subfolders: vec![] }], }, ], };
let total = total_size(&home).await; println!("합계: {total} KB");}실행 결과:
photos: 3072 KBtarget: 500 KBcode: 520 KBhome: 3596 KB합계: 3596 KB폴더 이름은 실제 디렉터리 이름처럼 보이도록 영어로 두었어요. 각 폴더는 하위 폴더의 크기를 먼저 기다려야 하니까, 하위 폴더가 모두 출력된 다음에 출력돼요. Box::pin이 중첩된 호출의 퓨처를 각각 힙에 올려 두기 때문에, 바깥 퓨처의 크기는 일정하게 유지돼요.
3. 큐를 다 비운 다음 멈추기
섹션 제목: “3. 큐를 다 비운 다음 멈추기”종료 신호를 보내기 전의 sleep(100ms)가 필요 없도록 종료 예제를 고쳐 보세요. 작업 세 개와 종료 신호를 곧바로 보내되, 워커가 멈추기 전에 대기 중인 작업을 모두 끝내도록 해야 해요. 힌트: 그냥 두면 select!는 두 분기가 모두 준비됐을 때 무작위로 하나를 골라요. biased;를 추가하고 작업 분기를 맨 위에 두세요.
정답 보기
use tokio::sync::{mpsc, oneshot};
#[tokio::main]async fn main() { let (job_tx, mut job_rx) = mpsc::channel::<String>(10); let (stop_tx, mut stop_rx) = oneshot::channel::<()>();
for job in ["사진 크기 조정", "이메일 보내기", "검색 색인 갱신"] { job_tx.send(job.to_string()).await.unwrap(); } stop_tx.send(()).unwrap();
let worker = tokio::spawn(async move { let mut done = 0; loop { tokio::select! { biased; Some(job) = job_rx.recv() => { println!("작업 중: {job}"); done += 1; } _ = &mut stop_rx => { println!("큐가 비었으니 멈출게요"); break; } } } done });
let done = worker.await.unwrap(); println!("워커가 작업 {done}개를 끝냈어요");}실행 결과:
작업 중: 사진 크기 조정작업 중: 이메일 보내기작업 중: 검색 색인 갱신큐가 비었으니 멈출게요워커가 작업 3개를 끝냈어요biased;가 있으면 select!는 항상 작업 분기부터 확인해요. 그래서 recv()에 준비된 게 없을 때만 종료 신호를 봐요. 이게 없으면 워커가 작업을 한두 개만 하고 멈출 수도 있어요. 가끔씩만 나타나는, 찾기 힘든 버그가 되는 거죠.
join!은 모든 퓨처를 기다리고,try_join!은 첫Err에서 멈추고,JoinSet은 spawn한 여러 태스크를 실행하면서 끝나는 대로 결과를 돌려줘요(순서가 중요하면 정렬하세요).select!는 가장 먼저 끝나는 퓨처를 기다려요. 나머지는 드롭되는데, 이게 비동기 Rust에서 취소가 동작하는 방식이에요. 반복문 안에서는 취소 안전한 퓨처를 쓰고, 순서가 중요하면biased;를 쓰세요.timeout은 느린 퓨처를 포기해요.interval은 일정한 박자로 틱을 주고,sleep은 한 번 기다려요.- 스트림은 비동기 반복자예요. 채널 수신자도 스트림처럼 동작하고,
tokio-stream이map,take같은 메서드를 더해 줘요. Pin은 퓨처가 메모리에서 움직이지 않겠다는 약속이에요. 한 퓨처를 여러 번 폴링할 때는pin!을, 퓨처를 저장하거나 비동기 재귀를 할 때는Box::pin을 쓰세요.- 런타임을 블로킹하지 말고(
spawn_blocking),std의 락을.await너머까지 쥐고 있지 말고, spawn한 태스크는 꼭 기다리세요.
이해도 점검
섹션 제목: “이해도 점검”이해도 점검
1. tokio::select!에서 먼저 끝나지 못한 분기는 어떻게 되나요?
2. 시간이 다 되면 tokio::time::timeout은 무엇을 돌려주나요?
3. async fn이 자기 자신을 호출할 때 왜 Box::pin이 필요할까요?
4. tokio 프로그램 안에서 CPU를 많이 쓰는 2초짜리 작업을 해야 해요. 가장 좋은 도구는 무엇일까요?
점수: 0 / 4
더 알아보기
섹션 제목: “더 알아보기”- Select — Tokio 튜토리얼 (영어):
select!를 자세히 다뤄요. 분기가 동작하는 방식, 취소의 의미, 반복문 안에서 쓰는 법까지요. - Streams — Tokio 튜토리얼 (영어):
tokio-stream으로 스트림을 쓰고, 변환하고, 직접 만드는 방법이에요. - A Closer Look at the Traits for Async — The Rust Book (영어):
Future,Pin,Unpin,Stream을 차근차근 친절하게 살펴봐요. 한국어판에는 아직 비동기 장이 없어요. - std::pin — Rust 표준 라이브러리 (영어): 고정(pinning),
Pin,Unpin,pin!매크로에 대한 공식 설명이에요. - docs.rs의 tokio::time (영어):
sleep,timeout,interval과 놓친 틱을 어떻게 처리하는지 다뤄요.