임의 개수의 Future 다루기
임의 개수의 Future 다루기 (Working With Any Number of Futures)
런타임에 제어를 양보하기 (Yielding Control to the Runtime)
"첫 Async 프로그램" 절에서 기억해 보세요. 각 await 지점에서 Rust는 await되는 future가 준비되지 않았다면 작업을 일시정지하고 다른 작업으로 전환할 기회를 런타임에 줍니다. 그 반대도 참이에요. Rust는 await 지점에서만 async 블록을 일시정지하고 제어를 런타임에 돌려줍니다. await 지점 사이의 모든 것은 동기적(synchronous)이에요.
즉 await 지점 없이 async 블록 안에서 많은 작업을 한다면, 그 future는 다른 future들이 진행되는 것을 막아요. 어떤 경우에는 이를 한 future가 다른 future들을 굶기는(starving) 것이라고 부르기도 합니다. 어떤 상황에서는 그게 큰 문제가 아닐 수 있어요. 하지만 비싼 설정이나 오래 걸리는 작업을 한다거나, 특정 작업을 무한정 계속하는 future가 있다면, 언제 어디에서 제어를 런타임에 돌려줄지 생각해야 합니다.
굶김 문제를 예시로 들기 위해 오래 걸리는 연산을 시뮬레이션해 보고, 그다음 해결 방법을 살펴볼게요. Listing 17-14는 slow 함수를 소개합니다.
extern crate trpl; // required for mdbook test
use std::{thread, time::Duration};
fn main() {
trpl::block_on(async {
// We will call `slow` here later
});
}
fn slow(name: &str, ms: u64) {
thread::sleep(Duration::from_millis(ms));
println!("'{name}' ran for {ms}ms");
}
이 코드는 trpl::sleep 대신 std::thread::sleep을 사용해서, slow를 호출하면 몇 밀리초 동안 현재 스레드를 차단하게 했어요. slow를 오래 실행되고 차단적인 실제 세계의 연산을 대신하는 것으로 쓸 수 있습니다.
Listing 17-15에서 slow를 사용해 이런 CPU 바운드 작업을 future 두 개에서 수행하는 것을 흉내 냅니다.
extern crate trpl; // required for mdbook test
use std::{thread, time::Duration};
fn main() {
trpl::block_on(async {
let a = async {
println!("'a' started.");
slow("a", 30);
slow("a", 10);
slow("a", 20);
trpl::sleep(Duration::from_millis(50)).await;
println!("'a' finished.");
};
let b = async {
println!("'b' started.");
slow("b", 75);
slow("b", 10);
slow("b", 15);
slow("b", 350);
trpl::sleep(Duration::from_millis(50)).await;
println!("'b' finished.");
};
trpl::select(a, b).await;
});
}
fn slow(name: &str, ms: u64) {
thread::sleep(Duration::from_millis(ms));
println!("'{name}' ran for {ms}ms");
}
각 future는 많은 slow 연산을 수행한 뒤에야 제어를 런타임에 돌려줘요. 이 코드를 실행하면 이런 출력이 보일 거예요:
'a' started.
'a' ran for 30ms
'a' ran for 10ms
'a' ran for 20ms
'b' started.
'b' ran for 75ms
'b' ran for 10ms
'b' ran for 15ms
'b' ran for 350ms
'a' finished.
Listing 17-5에서 두 URL을 가져오는 future들을 경쟁시키는 데 trpl::select를 썼던 것처럼, 여기서도 select는 a가 끝나는 즉시 끝나요. 다만 두 future의 slow 호출 사이에는 교차(interleaving)가 없습니다. a future는 trpl::sleep 호출이 await될 때까지 모든 작업을 하고, b future는 자기 trpl::sleep 호출이 await될 때까지 모든 작업을 하며, 마지막으로 a future가 완료되죠. 두 future가 자신의 slow 작업 사이에도 모두 진행되게 하려면, 제어를 런타임에 돌려줄 수 있는 await 지점이 필요해요. 즉 우리가 await할 수 있는 무언가가 필요하다는 뜻입니다!
이런 종류의 인계(handoff)가 Listing 17-15에서 이미 일어나는 걸 볼 수 있어요. a future 끝의 trpl::sleep을 제거하면 b future를 전혀 실행하지 않고 완료될 거예요. 연산들이 번갈아 진행되게 하는 출발점으로 trpl::sleep 함수를 사용해 봅시다. Listing 17-16처럼요.
extern crate trpl; // required for mdbook test
use std::{thread, time::Duration};
fn main() {
trpl::block_on(async {
let one_ms = Duration::from_millis(1);
let a = async {
println!("'a' started.");
slow("a", 30);
trpl::sleep(one_ms).await;
slow("a", 10);
trpl::sleep(one_ms).await;
slow("a", 20);
trpl::sleep(one_ms).await;
println!("'a' finished.");
};
let b = async {
println!("'b' started.");
slow("b", 75);
trpl::sleep(one_ms).await;
slow("b", 10);
trpl::sleep(one_ms).await;
slow("b", 15);
trpl::sleep(one_ms).await;
slow("b", 350);
trpl::sleep(one_ms).await;
println!("'b' finished.");
};
trpl::select(a, b).await;
});
}
fn slow(name: &str, ms: u64) {
thread::sleep(Duration::from_millis(ms));
println!("'{name}' ran for {ms}ms");
}
각 slow 호출 사이에 await 지점이 있는 trpl::sleep 호출을 추가했어요. 이제 두 future의 작업이 교차됩니다:
'a' started.
'a' ran for 30ms
'b' started.
'b' ran for 75ms
'a' ran for 10ms
'b' ran for 10ms
'a' ran for 20ms
'b' ran for 15ms
'a' finished.
a future는 여전히 trpl::sleep을 호출하기 전에 slow를 호출하므로 잠시 동안 실행된 뒤 제어를 b에 넘겨요. 하지만 그 후로는 future들이 그중 하나가 await 지점에 도달할 때마다 서로 오가며 교환합니다. 이 경우에는 매 slow 호출마다 그렇게 했지만, 우리에게 가장 합리적인 방식으로 작업을 나눌 수 있어요.
하지만 여기서 우리는 실제로 잠들고 싶은 게 아니에요. 가능한 한 빨리 진행하고 싶으니까요. 제어를 런타임에 돌려주기만 하면 돼요. trpl::yield_now 함수를 사용하면 직접 할 수 있습니다. Listing 17-17에서는 그 모든 trpl::sleep 호출을 trpl::yield_now로 바꿉니다.
extern crate trpl; // required for mdbook test
use std::{thread, time::Duration};
fn main() {
trpl::block_on(async {
let a = async {
println!("'a' started.");
slow("a", 30);
trpl::yield_now().await;
slow("a", 10);
trpl::yield_now().await;
slow("a", 20);
trpl::yield_now().await;
println!("'a' finished.");
};
let b = async {
println!("'b' started.");
slow("b", 75);
trpl::yield_now().await;
slow("b", 10);
trpl::yield_now().await;
slow("b", 15);
trpl::yield_now().await;
slow("b", 350);
trpl::yield_now().await;
println!("'b' finished.");
};
trpl::select(a, b).await;
});
}
fn slow(name: &str, ms: u64) {
thread::sleep(Duration::from_millis(ms));
println!("'{name}' ran for {ms}ms");
}
이 코드는 실제 의도를 더 명확히 드러낼 뿐 아니라 sleep을 쓰는 것보다 상당히 빨라질 수 있어요. sleep이 사용하는 것 같은 타이머는 세분화할 수 있는 한계가 있는 경우가 많거든요. 예를 들어 우리가 쓰는 sleep 버전은 1나노초의 Duration을 넘겨도 항상 적어도 1밀리초는 잡니다. 다시 말하지만 현대 컴퓨터는 빠르죠. 1밀리초에도 아주 많은 일을 할 수 있어요!
이는 프로그램이 하는 다른 일에 따라 async가 연산 바운드(compute-bound) 작업에도 유용할 수 있다는 뜻이에요. async는 프로그램의 서로 다른 부분 사이의 관계를 구조화하는 유용한 도구를 제공하니까요(다만 async 상태 기계의 오버헤드라는 비용이 있긴 해요). 이것은 일종의 협력적 멀티태스킹(cooperative multitasking) 으로, 각 future가 await 지점을 통해 언제 제어를 넘길지 결정할 힘을 가집니다. 따라서 각 future는 너무 오래 차단하지 않을 책임도 지녀요. 일부 Rust 기반 임베디드 운영체제에서는 이것이 유일한 멀티태스킹 방식이에요!
물론 실제 세계의 코드에서 함수 호출과 await 지점을 매 줄마다 번갈아 쓰는 경우는 보통 없을 거예요. 이런 방식으로 제어를 양보하는 것은 비교적 저렴하지만 공짜는 아니에요. 많은 경우 연산 바운드 작업을 잘게 나누려는 시도가 오히려 훨씬 느려지게 만들 수도 있어서, 때로는 연산이 잠시 차단되도록 두는 것이 전반적인 성능에 더 나을 수 있어요. 코드의 실제 성능 병목이 무엇인지 항상 측정해 보세요. 하지만 직렬로 일어나는 작업이 많다는 걸 발견했는데 동시에 일어날 거라고 기대했다면, 그 밑에 깔린 역학을 기억하는 것이 중요해요!
우리만의 Async 추상화 만들기 (Building Our Own Async Abstractions)
future들을 조합해 새로운 패턴을 만들 수도 있어요. 예를 들어 이미 가진 async 빌딩 블록으로 타임아웃(timeout) 함수를 만들 수 있습니다. 완성되면 그 결과는 또 더 많은 async 추상화를 만드는 데 쓸 수 있는 또 하나의 빌딩 블록이 될 거예요.
Listing 17-18은 이 타임아웃이 느린 future와 함께 어떻게 동작할지 기대하는 모습을 보여줍니다.
extern crate trpl; // required for mdbook test
use std::time::Duration;
fn main() {
trpl::block_on(async {
let slow = async {
trpl::sleep(Duration::from_secs(5)).await;
"Finally finished"
};
match timeout(slow, Duration::from_secs(2)).await {
Ok(message) => println!("Succeeded with '{message}'"),
Err(duration) => {
println!("Failed after {} seconds", duration.as_secs())
}
}
});
}
이걸 구현해 봅시다! 먼저 timeout의 API에 대해 생각해 볼게요:
- 우리가 await할 수 있도록 그것 자체가 async 함수여야 한다.
- 첫 번째 매개변수는 실행할 future여야 한다. 어떤 future와도 작동하도록 제네릭으로 만들 수 있다.
- 두 번째 매개변수는 기다릴 최대 시간일 것이다.
Duration을 쓰면trpl::sleep에 넘기기 쉽다. - 그것은
Result를 반환해야 한다. future가 성공적으로 완료되면 그Result는 future가 만든 값을 담은Ok가 된다. 타임아웃이 먼저 경과하면Result는 타임아웃이 기다린 시간을 담은Err가 된다.
Listing 17-19는 이 선언을 보여줍니다.
extern crate trpl; // required for mdbook test
use std::time::Duration;
fn main() {
trpl::block_on(async {
let slow = async {
trpl::sleep(Duration::from_secs(5)).await;
"Finally finished"
};
match timeout(slow, Duration::from_secs(2)).await {
Ok(message) => println!("Succeeded with '{message}'"),
Err(duration) => {
println!("Failed after {} seconds", duration.as_secs())
}
}
});
}
async fn timeout<F: Future>(
future_to_try: F,
max_time: Duration,
) -> Result<F::Output, Duration> {
// Here is where our implementation will go!
}
이로써 타입에 대한 목표는 충족됐어요. 이제 필요한 동작을 생각해 봅시다. 전달받은 future를 그 시간(duration)과 경쟁시키고 싶어요. trpl::sleep으로 시간에서 타이머 future를 만들고, trpl::select로 그 타이머를 호출자가 넘기는 future와 함께 실행할 수 있습니다.
Listing 17-20에서 trpl::select를 await한 결과에 대해 match해서 timeout을 구현합니다.
extern crate trpl; // required for mdbook test
use std::time::Duration;
use trpl::Either;
// --snip--
fn main() {
trpl::block_on(async {
let slow = async {
trpl::sleep(Duration::from_secs(5)).await;
"Finally finished"
};
match timeout(slow, Duration::from_secs(2)).await {
Ok(message) => println!("Succeeded with '{message}'"),
Err(duration) => {
println!("Failed after {} seconds", duration.as_secs())
}
}
});
}
async fn timeout<F: Future>(
future_to_try: F,
max_time: Duration,
) -> Result<F::Output, Duration> {
match trpl::select(future_to_try, trpl::sleep(max_time)).await {
Either::Left(output) => Ok(output),
Either::Right(_) => Err(max_time),
}
}
trpl::select의 구현은 공정하지 않아요. 항상 전달된 순서대로 인자를 폴링합니다(다른 select 구현은 어느 인자를 먼저 폴링할지 무작위로 고르기도 해요). 따라서 future_to_try가 완료될 기회를 얻도록 우리는 그것을 select에 먼저 넘겨요. max_time이 아주 짧은 시간이어도 말이죠. future_to_try가 먼저 끝나면 select는 future_to_try의 출력을 담은 Left를 반환합니다. 타이머가 먼저 끝나면 select는 타이머의 출력인 ()를 담은 Right를 반환해요.
future_to_try가 성공해서 Left(output)을 얻으면 Ok(output)을 반환합니다. 대신 sleep 타이머가 경과해서 Right(())를 얻으면 _로 ()를 무시하고 Err(max_time)을 반환하죠.
이로써 우리는 다른 async 헬퍼 둘로 만들어진 동작하는 timeout을 가지게 됐어요. 코드를 실행하면 타임아웃 뒤에 실패 모드를 출력할 거예요:
Failed after 2 seconds
future는 다른 future와 합성되므로, 더 작은 async 빌딩 블록으로 정말 강력한 도구를 만들 수 있어요. 예를 들어 같은 접근법으로 타임아웃을 재시도(retry)와 결합하고, 그것들을 다시 네트워크 호출(예: Listing 17-5의 것) 같은 연산과 쓸 수 있습니다.
실제로는 보통 async·await를 직접 사용하고, 그다음에 select 같은 함수와 join! 매크로 같은 것들을 보조적으로 사용해 가장 바깥쪽 future들이 어떻게 실행될지 제어해요.
우리는 지금까지 여러 future를 동시에 다루는 여러 방법을 봤어요. 다음으로는 스트림(stream) 으로 여러 future를 시간에 걸친 순서로 다루는 방법을 살펴볼 거예요.