Async로 동시성 적용하기
Async로 동시성 적용하기 (Applying Concurrency with Async)
이번 절에서는 16장에서 스레드로 다뤘던 것과 같은 동시성 과제 일부에 async를 적용해 볼 거예요. 그 핵심 아이디어는 이미 많이 이야기했으니, 여기서는 스레드와 future 사이의 차이에 초점을 맞출게요.
동시성을 다루는 async API는 스레드를 쓸 때와 매우 비슷한 경우가 많아요. 어떤 경우에는 꽤 다르게 끝나기도 하고요. 스레드와 async의 API가 비슷해 보여도 동작이 다른 경우가 많고, 성능 특성은 거의 항상 달라요.
spawn_task로 새 작업 만들기
16장의 "spawn으로 새 스레드 만들기" 절에서 첫 번째로 다룬 연산은 두 개의 분리된 스레드에서 숫자를 세는 것이었어요. 이번에는 async로 똑같이 해 볼게요. trpl 크레이트는 thread::spawn API와 매우 비슷해 보이는 spawn_task 함수와, thread::sleep API의 async 버전인 sleep 함수를 제공합니다. Listing 17-6에서처럼 이 둘을 함께 써서 숫자 세기 예시를 구현할 수 있어요.
extern crate trpl; // required for mdbook test
use std::time::Duration;
fn main() {
trpl::block_on(async {
trpl::spawn_task(async {
for i in 1..10 {
println!("hi number {i} from the first task!");
trpl::sleep(Duration::from_millis(500)).await;
}
});
for i in 1..5 {
println!("hi number {i} from the second task!");
trpl::sleep(Duration::from_millis(500)).await;
}
});
}
시작점으로 trpl::block_on으로 main 함수를 설정해 최상위 함수가 async가 되게 했어요.
참고: 이번 장의 이 지점부터는 모든 예시에 main의 trpl::block_on이라는 이 똑같은 감싸는 코드가 포함되므로, main을 생략하듯이 이 코드도 자주 생략할 거예요. 코드에 포함하는 걸 잊지 마세요!
그런 다음 그 블록 안에 루프 두 개를 씁니다. 각 루프는 trpl::sleep 호출을 하나씩 포함하는데, 다음 메시지를 보내기 전에 반 초(500밀리초)씩 기다려요. 루프 하나는 trpl::spawn_task의 본문에 넣고, 다른 하나는 최상위 for 루프에 넣었습니다. 또한 sleep 호출 뒤에 await도 추가했어요.
이 코드는 스레드 기반 구현과 비슷하게 동작합니다. 실행하면 터미널에서 메시지가 다른 순서로 나타날 수도 있다는 점까지 포함해서요:
hi number 1 from the second task!
hi number 1 from the first task!
hi number 2 from the first task!
hi number 2 from the second task!
hi number 3 from the first task!
hi number 3 from the second task!
hi number 4 from the first task!
hi number 4 from the second task!
hi number 5 from the first task!
이 버전은 main async 블록 본문 안의 for 루프가 끝나는 즉시 멈춰요. spawn_task로 생성된 작업은 main 함수가 끝날 때 종료되거든요. 작업이 끝까지 돌기를 원한다면, 조인 핸들(join handle)을 사용해 첫 번째 작업이 완료되기를 기다려야 합니다. 스레드에서는 join 메서드로 스레드가 실행을 끝낼 때까지 "차단(block)"했죠. Listing 17-7에서는 await로 똑같이 할 수 있어요. 작업 핸들 자체가 future니까요. 그 Output 타입은 Result라서 await한 뒤에 unwrap도 해줘요.
extern crate trpl; // required for mdbook test
use std::time::Duration;
fn main() {
trpl::block_on(async {
let handle = trpl::spawn_task(async {
for i in 1..10 {
println!("hi number {i} from the first task!");
trpl::sleep(Duration::from_millis(500)).await;
}
});
for i in 1..5 {
println!("hi number {i} from the second task!");
trpl::sleep(Duration::from_millis(500)).await;
}
handle.await.unwrap();
});
}
이 갱신된 버전은 두 루프가 모두 끝날 때까지 실행됩니다:
hi number 1 from the second task!
hi number 1 from the first task!
hi number 2 from the first task!
hi number 2 from the second task!
hi number 3 from the first task!
hi number 3 from the second task!
hi number 4 from the first task!
hi number 4 from the second task!
hi number 5 from the first task!
hi number 6 from the first task!
hi number 7 from the first task!
hi number 8 from the first task!
hi number 9 from the first task!
지금까지 async와 스레드는 구문만 다를 뿐 비슷한 결과를 주는 것처럼 보여요. 조인 핸들에 join을 호출하는 대신 await를 쓰고, sleep 호출을 await하는 것뿐이죠.
더 큰 차이는 이 작업을 위해 운영체제 스레드를 하나 더 만들 필요가 없었다는 점이에요. 사실 여기서는 작업(task)조차 만들 필요가 없습니다. async 블록은 익명 future로 컴파일되니까, 각 루프를 async 블록에 넣고 trpl::join 함수로 런타임이 둘 다 끝까지 실행하게 할 수 있어요.
16장의 "모든 스레드가 끝나기를 기다리기" 절에서 우리는 std::thread::spawn을 호출할 때 반환되는 JoinHandle 타입의 join 메서드를 어떻게 쓰는지 보여줬어요. trpl::join 함수는 비슷하지만 future용이에요. 두 future를 넘기면, 넘긴 각 future의 출력을 담은 튜플을 그 둘이 모두 완료되면 출력으로 내놓는 단일 새 future를 만들어 냅니다. 따라서 Listing 17-8에서 우리는 trpl::join으로 fut1과 fut2 둘 다 끝나기를 기다려요. fut1과 fut2를 await하는 게 아니라, trpl::join이 만들어 낸 새 future를 await합니다. 그 출력은 단위 값 두 개를 담은 튜플일 뿐이라 무시해요.
extern crate trpl; // required for mdbook test
use std::time::Duration;
fn main() {
trpl::block_on(async {
let fut1 = async {
for i in 1..10 {
println!("hi number {i} from the first task!");
trpl::sleep(Duration::from_millis(500)).await;
}
};
let fut2 = async {
for i in 1..5 {
println!("hi number {i} from the second task!");
trpl::sleep(Duration::from_millis(500)).await;
}
};
trpl::join(fut1, fut2).await;
});
}
이걸 실행하면 두 future가 모두 끝까지 실행되는 걸 볼 수 있어요:
hi number 1 from the first task!
hi number 1 from the second task!
hi number 2 from the first task!
hi number 2 from the second task!
hi number 3 from the first task!
hi number 3 from the second task!
hi number 4 from the first task!
hi number 4 from the second task!
hi number 5 from the first task!
hi number 6 from the first task!
hi number 7 from the first task!
hi number 8 from the first task!
hi number 9 from the first task!
이제 매번 정확히 같은 순서를 보게 될 거예요. 이것은 스레드와 Listing 17-7의 trpl::spawn_task에서 본 것과는 아주 다릅니다. trpl::join 함수가 공정(fair) 하기 때문이에요. 즉 각 future를 동등하게 자주 검사해서 번갈아 가며, 다른 쪽이 준비되었을 때 한쪽이 앞서 나가도록 절대 놔두지 않는다는 뜻이죠. 스레드에서는 운영체제가 어느 스레드를 검사할지, 얼마나 오래 실행하게 할지를 결정합니다. async Rust에서는 런타임이 어느 작업을 검사할지 결정해요. (실제로는 async 런타임이 동시성을 관리하는 방식의 일부로 내부적으로 운영체제 스레드를 사용할 수 있어서 세부 사항이 복잡해지고, 공정성을 보장하는 것이 런타임에게 더 많은 작업일 수 있어요. 하지만 여전히 가능합니다!) 런타임은 어떤 주어진 연산에 대해 공정성을 보장할 필요는 없고, 공정성을 원하는지 여부를 고를 수 있는 서로 다른 API를 자주 제공해요.
이런 변형들을 시도해 보고 무엇을 하는지 살펴보세요:
- 두 루프 중 하나 또는 둘 다에서 async 블록을 제거해 보세요.
- 각 async 블록을 정의한 직후에 await해 보세요.
- 첫 번째 루프만 async 블록으로 감싸고, 두 번째 루프 본문 뒤에서 결과 future를 await해 보세요.
추가 도전으로, 코드를 실행하기 전에 각 경우의 출력이 무엇일지 알아맞혀 보세요!
메시지 전달로 두 작업 사이 데이터 보내기
future 사이에서 데이터를 공유하는 것도 익숙할 거예요. 다시 메시지 전달을 사용할 텐데, 이번에는 타입과 함수의 async 버전을 써요. 16장의 "메시지 전달로 스레드 사이 데이터 전송하기" 절에서와는 약간 다른 길을 택해, 스레드 기반 동시성과 future 기반 동시성의 핵심 차이 몇 가지를 보여줄 거예요. Listing 17-9에서는 별도 작업을 만들지 않고 단일 async 블록으로 시작합니다 — 16장에서 별도 스레드를 만들었던 것과 대조되죠.
extern crate trpl; // required for mdbook test
fn main() {
trpl::block_on(async {
let (tx, mut rx) = trpl::channel();
let val = String::from("hi");
tx.send(val).unwrap();
let received = rx.recv().await.unwrap();
println!("received '{received}'");
});
}
여기서는 16장에서 스레드와 함께 썼던 다중 생산자·단일 소비자 채널 API의 async 버전인 trpl::channel을 사용해요. 이 API의 async 버전은 스레드 기반 버전과 조금만 다릅니다. 불변 수신자 rx 대신 가변(mutable) 수신자를 쓰고, recv 메서드가 값을 직접 만드는 대신 await해야 하는 future를 만들어 내죠. 이제 송신자에서 수신자로 메시지를 보낼 수 있어요. 별도 스레드나 작업조차 만들 필요가 없다는 점을 보세요. rx.recv 호출을 await하기만 하면 됩니다.
std::mpsc::channel의 동기 Receiver::recv 메서드는 메시지를 받을 때까지 차단해요. trpl::Receiver::recv 메서드는 async라서 차단하지 않습니다. 차단하는 대신 메시지를 받거나 채널의 송신 측이 닫힐 때까지 제어를 런타임에 넘겨줘요. 반대로 우리는 send 호출을 await하지 않는데, 차단하지 않기 때문입니다. 그럴 필요도 없어요. 메시지를 보내는 채널이 무제한(unbounded)이거든요.
참고: 이 async 코드 전부가 trpl::block_on 호출 안의 async 블록에서 실행되므로, 그 안의 모든 것은 차단을 피할 수 있어요. 하지만 그 밖의 코드는 block_on 함수가 반환할 때 차단됩니다. 그게 바로 trpl::block_on 함수의 핵심 포인트예요. 일련의 async 코드에서 어디에서 차단할지, 따라서 동기 코드와 async 코드 사이를 어디에서 전환할지를 고를 수 있게 해 줍니다.
이 예시에서 두 가지를 눈여겨보세요. 첫째, 메시지는 즉시 도착할 거예요. 둘째, 여기서 future를 쓰긴 하지만 아직 동시성은 없어요. 이 목록의 모든 것은 future가 없어도 그러했을 것처럼 순서대로 일어납니다.
첫 번째 부분은 메시지 여러 개를 보내고 그 사이에 잠들어 처리해 봅시다. Listing 17-10처럼요.
extern crate trpl; // required for mdbook test
use std::time::Duration;
fn main() {
trpl::block_on(async {
let (tx, mut rx) = trpl::channel();
let vals = vec![
String::from("hi"),
String::from("from"),
String::from("the"),
String::from("future"),
];
for val in vals {
tx.send(val).unwrap();
trpl::sleep(Duration::from_millis(500)).await;
}
while let Some(value) = rx.recv().await {
println!("received '{value}'");
}
});
}
메시지를 보내는 것 말고도 받아야 해요. 이 경우 들어오는 메시지 수를 알고 있으니 rx.recv().await를 네 번 호출해 수동으로 할 수도 있어요. 하지만 실제 세계에서는 일반적으로 알 수 없는 수의 메시지를 기다리므로, 더 이상 메시지가 없다고 판단할 때까지 계속 기다려야 합니다.
Listing 16-10에서는 동기 채널에서 받은 모든 항목을 처리하기 위해 for 루프를 썼어요. 하지만 Rust에는 아직 비동기적으로 만들어지는 일련의 항목과 함께 for 루프를 쓸 방법이 없어서, 아직 보지 못한 루프를 써야 합니다. 바로 while let 조건 루프입니다. 이것은 6장의 "if let과 let...else로 간결한 제어 흐름" 절에서 봤던 if let 구조의 루프 버전이에요. 이 루프는 지정한 패턴이 값과 계속 일치하는 한 계속 실행됩니다.
rx.recv 호출은 future를 만들어 내는데, 우리는 그것을 await해요. 런타임은 그 future가 준비될 때까지 일시정지합니다. 메시지가 도착하면 future는 메시지가 도착할 때마다 Some(message)로 해석됩니다. 채널이 닫히면, 메시지가 도착했는지 여부와 관계없이 future는 대신 None으로 해석되어 더 이상 값이 없음을 나타내고, 따라서 폴링(polling)·즉 await를 멈춰야 함을 알려줘요.
while let 루프가 이 모든 것을 종합합니다. rx.recv().await의 결과가 Some(message)이면 메시지에 접근해 if let과 마찬가지로 루프 본문에서 사용할 수 있어요. 결과가 None이면 루프가 끝납니다. 루프가 매번 완료될 때마다 다시 await 지점에 도달하므로, 런타임은 또 다른 메시지가 도착할 때까지 그것을 다시 일시정지합니다.
이제 코드는 모든 메시지를 성공적으로 보내고 받아요. 안타깝게도 아직 몇 가지 문제가 있습니다. 우선 메시지가 반 초 간격으로 도착하지 않아요. 프로그램을 시작한 지 2초(2,000밀리초) 뒤에 한꺼번에 도착하죠. 또 이 프로그램은 끝나지도 않습니다! 대신 새 메시지를 영원히 기다려요. ctrl - C로 종료해야 합니다.
하나의 async 블록 안 코드는 선형으로 실행된다
먼저 왜 메시지가 각각 사이에 지연을 두고 도착하지 않고, 전체 지연이 지난 뒤에 한꺼번에 도착하는지 살펴볼게요. 주어진 async 블록 안에서 코드에 await 키워드가 나타나는 순서가, 프로그램이 실행될 때 실행되는 순서이기도 합니다.
Listing 17-10에는 async 블록이 하나뿐이라 그 안의 모든 것이 선형으로 실행돼요. 여전히 동시성이 없습니다. 모든 tx.send 호출이, 모든 trpl::sleep 호출 및 그와 관련된 await 지점과 섞여서 일어나죠. 그다음에야 while let 루프가 recv 호출의 await 지점들을 지나가기 시작합니다.
우리가 원하는 동작, 즉 각 메시지 사이에 잠자는 지연이 있는 동작을 얻으려면, Listing 17-11처럼 tx와 rx 연산을 각자의 async 블록에 넣어야 해요. 그러면 런타임이 Listing 17-8에서처럼 trpl::join으로 각각을 따로 실행할 수 있습니다. 다시 한 번, 개별 future가 아니라 trpl::join 호출의 결과를 await해요. 개별 future를 순서대로 await하면 결국 순차적인 흐름으로 돌아갈 뿐입니다 — 바로 우리가 피하려는 것이죠.
extern crate trpl; // required for mdbook test
use std::time::Duration;
fn main() {
trpl::block_on(async {
let (tx, mut rx) = trpl::channel();
let tx_fut = async {
let vals = vec![
String::from("hi"),
String::from("from"),
String::from("the"),
String::from("future"),
];
for val in vals {
tx.send(val).unwrap();
trpl::sleep(Duration::from_millis(500)).await;
}
};
let rx_fut = async {
while let Some(value) = rx.recv().await {
println!("received '{value}'");
}
};
trpl::join(tx_fut, rx_fut).await;
});
}
Listing 17-11의 갱신된 코드로 메시지는 2초 뒤에 한꺼번에 나오는 대신 500밀리초 간격으로 출력됩니다.
async 블록으로 소유권 옮기기
하지만 while let 루프가 trpl::join과 상호작용하는 방식 때문에 프로그램은 여전히 끝나지 않아요:
trpl::join이 반환하는 future는 넘긴 두 future가 모두 완료된 뒤에만 완료된다.tx_futfuture는vals의 마지막 메시지를 보낸 뒤 잠들기를 끝내면 완료된다.rx_futfuture는while let루프가 끝날 때까지 완료되지 않는다.while let루프는rx.recv를 await한 결과가None이 될 때까지 끝나지 않는다.rx.recv를 await하면 채널의 반대쪽 끝이 닫힌 후에만None이 반환된다.- 채널은
rx.close를 호출하거나 송신자 측인tx가 드롭될 때만 닫힌다. - 우리는 어디에서도
rx.close를 호출하지 않고,tx는trpl::block_on에 넘긴 가장 바깥쪽 async 블록이 끝날 때까지 드롭되지 않을 것이다. - 그 블록은
trpl::join완료에 막혀 있어서 끝날 수 없고, 그건 우리를 이 목록의 맨 위로 되돌려 보낸다.
지금 우리가 메시지를 보내는 async 블록은 tx를 빌리기만 해요. 메시지를 보내는 데 소유권이 필요하지 않기 때문이죠. 하지만 tx를 그 async 블록 안으로 옮길 수 있다면, 그 블록이 끝날 때 tx가 드롭될 거예요. 13장의 "참조자 포착 또는 소유권 이동" 절에서 클로저와 함께 move 키워드를 쓰는 법을 배웠고, 16장의 "스레드와 함께 move 클로저 사용하기" 절에서 말했듯이 스레드와 작업할 때는 종종 데이터를 클로저 안으로 옮겨야 합니다. 같은 기본 역학이 async 블록에도 적용되므로, move 키워드는 클로저에서처럼 async 블록에서도 동작해요.
Listing 17-12에서 메시지를 보내는 블록을 async에서 async move로 바꿉니다.
extern crate trpl; // required for mdbook test
use std::time::Duration;
fn main() {
trpl::block_on(async {
let (tx, mut rx) = trpl::channel();
let tx_fut = async move {
// --snip--
let vals = vec![
String::from("hi"),
String::from("from"),
String::from("the"),
String::from("future"),
];
for val in vals {
tx.send(val).unwrap();
trpl::sleep(Duration::from_millis(500)).await;
}
};
let rx_fut = async {
while let Some(value) = rx.recv().await {
println!("received '{value}'");
}
};
trpl::join(tx_fut, rx_fut).await;
});
}
이 버전의 코드를 실행하면 마지막 메시지를 보내고 받은 뒤에 우아하게 종료됩니다. 이제 하나 이상의 future에서 데이터를 보내려면 무엇을 바꿔야 하는지 살펴볼게요.
join! 매크로로 여러 future 합치기
이 async 채널은 다중 생산자 채널이기도 하므로, 여러 future에서 메시지를 보내려면 Listing 17-13처럼 tx에 clone을 호출할 수 있어요.
extern crate trpl; // required for mdbook test
use std::time::Duration;
fn main() {
trpl::block_on(async {
let (tx, mut rx) = trpl::channel();
let tx1 = tx.clone();
let tx1_fut = async move {
let vals = vec![
String::from("hi"),
String::from("from"),
String::from("the"),
String::from("future"),
];
for val in vals {
tx1.send(val).unwrap();
trpl::sleep(Duration::from_millis(500)).await;
}
};
let rx_fut = async {
while let Some(value) = rx.recv().await {
println!("received '{value}'");
}
};
let tx_fut = async move {
let vals = vec![
String::from("more"),
String::from("messages"),
String::from("for"),
String::from("you"),
];
for val in vals {
tx.send(val).unwrap();
trpl::sleep(Duration::from_millis(1500)).await;
}
};
trpl::join!(tx1_fut, tx_fut, rx_fut);
});
}
먼저 tx를 클론해 첫 번째 async 블록 바깥에서 tx1을 만들어요. 전에 tx로 했던 것처럼 tx1을 그 블록 안으로 옮깁니다. 그런 다음 나중에 원래 tx를 새 async 블록으로 옮겨, 약간 더 느린 지연으로 더 많은 메시지를 보내요. 우연히 이 새 async 블록을 메시지를 받는 async 블록 뒤에 놓았는데, 앞에 놓아도 상관없어요. 핵심은 future가 만들어지는 순서가 아니라 await되는 순서입니다.
메시지를 보내는 두 async 블록 모두 async move 블록이어야 해서, 두 블록이 끝날 때 tx와 tx1 둘 다 드롭되어야 해요. 그렇지 않으면 처음 시작했던 것과 같은 무한 루프로 돌아가게 됩니다.
마지막으로 추가 future를 처리하기 위해 trpl::join에서 trpl::join!으로 전환해요. join! 매크로는 컴파일 타임에 그 수를 아는 임의 개수의 future를 await합니다. 수를 모르는 future 컬렉션을 await하는 것은 이번 장에서 나중에 다룰 거예요.
이제 두 송신 future의 모든 메시지를 볼 수 있고, 송신 future들이 보낸 후 조금씩 다른 지연을 사용하므로 메시지도 그 다른 간격으로 수신됩니다:
received 'hi'
received 'more'
received 'from'
received 'the'
received 'messages'
received 'future'
received 'for'
received 'you'
우리는 메시지 전달로 future 사이에 데이터를 보내는 법, async 블록 안 코드가 순차적으로 실행되는 법, async 블록으로 소유권을 옮기는 법, 여러 future를 합치는 법을 살펴봤어요. 다음으로는 런타임에 다른 작업으로 전환할 수 있다고 어떻게·왜 알려주는지 논의해 봅시다.