2024-10-06
HSPACE Rust 특강 #12 — 비동기 프로그래밍 Part 2
특강의 마지막 회차. Part 1에서 만든 채팅 클라이언트의 짝인 서버를 만들고,
그다음 지금까지 라이브러리에 맡겼던 block_on 과 spawn_blocking 을 직접
구현한다. 마지막 부분이 특히 좋았다. 퓨처가 어떻게 굴러가는지 안에서 보게 된다.
서버의 메인 함수
fn main() -> ChatResult<()> {
let address = std::env::args().nth(1).expect("Usage: server ADDRESS");
let chat_group_table = Arc::new(group_table::GroupTable::new());
async_std::task::block_on(async {
let listener = net::TcpListener::bind(address).await?;
let mut new_connections = listener.incoming();
while let Some(socket_result) = new_connections.next().await {
let socket = socket_result?;
let groups = chat_group_table.clone();
task::spawn(async {
log_error(serve(socket, groups).await);
});
}
Ok(())
})
}
전형적인 서버 루프인데, 비동기 버전이다.
listener.incoming()이 스트림이다.while let ... .next().await로 연결을 하나씩 받는다- 연결마다
task::spawn으로 태스크를 띄운다. 연결당 스레드가 아니라 연결당 태스크다. 이게 비동기 서버의 핵심 구조다 Arc<GroupTable>을 복제해 각 태스크에 넘긴다. 여러 태스크가 그룹 테이블을 공유해야 하기 때문이다
스레드 기반 서버였다면 만 개 연결에 만 개 스레드가 필요했을 텐데, 여기서는 태스크 만 개가 스레드 풀 위에서 돈다.
채팅 연결 처리하기
각 연결이 하는 일은 단순하다. 클라이언트에게서 오는 요청을 읽고 처리한다.
pub async fn serve(socket: TcpStream, groups: Arc<GroupTable>) -> ChatResult<()> {
let outbound = Arc::new(Outbound::new(socket.clone()));
let buffered = BufReader::new(socket);
let mut from_client = utils::receive_as_json(buffered);
while let Some(request_result) = from_client.next().await {
let request = request_result?;
let result = match request {
FromClient::Join { group_name } => {
let group = groups.get_or_create(group_name);
group.join(outbound.clone());
Ok(())
}
FromClient::Post { group_name, message } => match groups.get(&group_name) {
Some(group) => {
group.post(message);
Ok(())
}
None => Err(format!("Group '{group_name}' does not exist")),
},
};
if let Err(message) = result {
outbound.send(FromServer::Error(message)).await?;
}
}
Ok(())
}
socket.clone() 이 눈에 띈다. async_std 의 TcpStream 은 복제하면 같은 소켓을
가리키는 핸들이 하나 더 생긴다. 그래서 읽기용과 쓰기용을 따로 들고 있을 수 있다.
나가는 쪽은 Outbound 로 감싼다.
pub struct Outbound(Mutex<TcpStream>);
impl Outbound {
pub async fn send(&self, packet: FromServer) -> ChatResult<()> {
let mut guard = self.0.lock().await;
utils::send_as_json(&mut *guard, &packet).await?;
guard.flush().await?;
Ok(())
}
}
왜 뮤텍스가 필요한가? 한 클라이언트가 여러 그룹에 가입할 수 있고, 각 그룹이 별도 태스크에서 메시지를 보내기 때문이다. 락이 없으면 두 메시지의 JSON이 뒤섞여서 나갈 수 있다.
여기서 중요한 게 async_std::sync::Mutex 를 쓴다는 점이다. .lock().await 이지
.lock().unwrap() 이 아니다. Part 1에서 "락을 잡은 채 .await 하지 말라"고 했는데,
여기서는 락 안에서 .await 를 해야만 한다(쓰기가 비동기니까). 이럴 때 쓰는 게
비동기 뮤텍스다. 락 대기 중에 스레드를 블로킹하지 않고 태스크를 접어둔다.
동기 뮤텍스와 비동기 뮤텍스의 구분이 이렇게 실제 코드에서 나뉘는 걸 보니 명확해졌다.
- 동기
std::sync::Mutex— 잠깐 데이터를 만지고 바로 놓을 때. 빠르다 - 비동기
async_std::sync::Mutex— 락을 잡은 채.await해야 할 때
그룹 테이블
동기 뮤텍스가 맞는 자리도 있다.
pub struct GroupTable(Mutex<HashMap<Arc<String>, Arc<Group>>>);
impl GroupTable {
pub fn get(&self, name: &String) -> Option<Arc<Group>> {
self.0.lock().unwrap().get(name).cloned()
}
pub fn get_or_create(&self, name: Arc<String>) -> Arc<Group> {
self.0
.lock()
.unwrap()
.entry(name.clone())
.or_insert_with(|| Arc::new(Group::new(name)))
.clone()
}
}
이쪽은 std::sync::Mutex 다. 해시맵 조회는 순식간에 끝나고 .await 가 없으므로
동기 락으로 충분하다. 두 메서드 모두 Arc 를 복제해서 반환하고 락을 즉시
놓는다는 점이 중요하다. 락 안에서 오래 머무르지 않는다.
entry().or_insert_with().clone() 이 "있으면 가져오고 없으면 만들기"를 한 번의
해시 조회로 처리하는 관용구다.
채팅 그룹
그룹은 브로드캐스트 채널로 구현된다.
pub struct Group {
name: Arc<String>,
sender: broadcast::Sender<Arc<String>>,
}
impl Group {
pub fn new(name: Arc<String>) -> Group {
let (sender, _receiver) = broadcast::channel(1000);
Group { name, sender }
}
pub fn join(&self, outbound: Arc<Outbound>) {
let receiver = self.sender.subscribe();
task::spawn(handle_subscriber(self.name.clone(), receiver, outbound));
}
pub fn post(&self, message: Arc<String>) {
// 구독자가 없을 때만 에러가 난다.
let _ignored = self.sender.send(message);
}
}
동시성 회차에서 본 mpsc 채널과 다르다. mpsc 는 여러 생산자, 하나의 소비자다.
채팅 그룹은 반대로 여러 소비자가 같은 메시지를 모두 받아야 하므로 tokio 의
broadcast 채널을 쓴다.
가입할 때마다 구독자용 태스크를 하나 띄운다.
async fn handle_subscriber(
group_name: Arc<String>,
mut receiver: broadcast::Receiver<Arc<String>>,
outbound: Arc<Outbound>,
) {
loop {
let packet = match receiver.recv().await {
Ok(message) => FromServer::Message {
group_name: group_name.clone(),
message: message.clone(),
},
Err(RecvError::Lagged(n)) => {
FromServer::Error(format!("Dropped {n} messages from {group_name}."))
}
Err(RecvError::Closed) => break,
};
if outbound.send(packet).await.is_err() {
break;
}
}
}
RecvError::Lagged(n) 처리가 이 설계에서 가장 배울 점이었다. 브로드캐스트 채널의
버퍼는 1000개인데, 느린 클라이언트가 그보다 뒤처지면 오래된 메시지가 버려진다.
그때 Lagged(n) 이 오고, 서버는 "메시지 n개를 놓쳤다"라고 알려준다.
느린 소비자 하나 때문에 전체가 멈추지 않는다. 채팅에서는 최신 메시지를 보는 게 중요하지 모든 메시지를 보장하는 게 중요한 게 아니므로, 이게 맞는 트레이드오프다. 백프레셔를 어떻게 처리할지가 설계 결정이라는 걸 보여주는 좋은 예다.
post 에서 에러를 무시하는 것도 주석에 이유가 적혀 있다. 연결의 나가는 쪽이 들어오는
쪽보다 조금 먼저 끝날 수 있어서, 빈 그룹에 보내는 상황이 정상적으로 발생한다.
기본 제공 퓨처와 이그제큐터
여기서부터가 이번 회차의 진짜 주제다. 지금까지 async_std::task::block_on 을
그냥 썼는데, 그게 안에서 뭘 하는지 직접 만들어본다.
퓨처를 진행시키려면 누군가 poll 을 불러야 한다. 그 "누군가"가 이그제큐터
(executor) 다. 그리고 Pending 을 받았을 때 언제 다시 부를지 알려주는 게
웨이커(Waker) 다.
웨이커 호출하기: spawn_blocking 구현
spawn_blocking 을 직접 만들어보면 웨이커의 역할이 명확해진다. 하는 일은
"블로킹 작업을 별도 스레드에서 돌리고, 그 결과를 기다리는 퓨처를 준다"이다.
pub struct SpawnBlocking<T>(Arc<Mutex<Shared<T>>>);
struct Shared<T> {
value: Option<T>, // 결과 (아직 없으면 None)
waker: Option<Waker>, // 결과가 나오면 깨울 대상
}
공유 상태에 값과 웨이커 두 개를 둔다. 이게 전부다.
pub fn spawn_blocking<T, F>(closure: F) -> SpawnBlocking<T>
where
F: FnOnce() -> T + Send + 'static,
T: Send + 'static,
{
let inner = Arc::new(Mutex::new(Shared { value: None, waker: None }));
std::thread::spawn({
let inner = inner.clone();
move || {
let value = closure(); // 블로킹 작업 실행
let maybe_waker = {
let mut guard = inner.lock().unwrap();
guard.value = Some(value); // 결과 저장
guard.waker.take() // 등록된 웨이커 회수
};
if let Some(waker) = maybe_waker {
waker.wake(); // 이그제큐터에게 알림
}
}
});
SpawnBlocking(inner)
}
그리고 퓨처 구현이 놀랍도록 짧다.
impl<T: Send> Future for SpawnBlocking<T> {
type Output = T;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<T> {
let mut guard = self.0.lock().unwrap();
if let Some(value) = guard.value.take() {
return Poll::Ready(value); // 결과가 있으면 바로 준다
}
guard.waker = Some(cx.waker().clone()); // 없으면 웨이커를 남겨두고
Poll::Pending // 나중에 깨워달라고 한다
}
}
비동기의 전체 프로토콜이 이 스무 줄에 다 있다.
- 이그제큐터가
poll을 부른다 - 준비 안 됐으면 퓨처가
Context에서 웨이커를 꺼내 저장하고Pending반환 - 이그제큐터는 이 퓨처를 잠시 잊는다 (바쁜 대기 없음)
- 작업이 끝나면 워커 스레드가
waker.wake()호출 - 이그제큐터가 그 태스크를 실행 큐에 다시 넣는다
- 다시
poll→Ready(value)
guard.waker.take() 를 락 안에서 하고 waker.wake() 는 락 밖에서 하는 것도
눈여겨볼 점이다. 락을 잡은 채 남의 코드(웨이커)를 부르면 데드락 위험이 있다.
block_on 구현하기
이그제큐터 쪽이다. 놀랍게도 더 짧다.
use crossbeam::sync::Parker;
use futures_lite::pin;
use waker_fn::waker_fn;
fn block_on<F: Future>(future: F) -> F::Output {
let parker = Parker::new();
let unparker = parker.unparker().clone();
let waker = waker_fn(move || unparker.unpark()); // 깨우면 park를 푼다
let mut context = Context::from_waker(&waker);
pin!(future);
loop {
match future.as_mut().poll(&mut context) {
Poll::Ready(value) => return value,
Poll::Pending => parker.park(), // 깨울 때까지 스레드를 재운다
}
}
}
읽어보면 정말 단순하다.
- 웨이커를 만든다. 하는 일은 "이 스레드를 깨우기"뿐이다
- 루프를 돌며
poll한다 Ready면 값을 반환하고 끝Pending이면park()로 스레드를 재운다. 바쁜 대기가 아니다. 웨이커가unpark를 부를 때까지 이 스레드는 CPU를 전혀 쓰지 않는다
"비동기 런타임"이라고 하면 대단히 복잡한 물건일 것 같은데, 단일 퓨처를 돌리는
최소 이그제큐터는 이 정도다. async_std 나 tokio 가 복잡한 건 여기에 다중
태스크 스케줄링, 작업 훔치기, I/O 이벤트 루프(epoll/kqueue), 타이머를 얹었기
때문이다. 뼈대 자체는 이 루프다.
핀 설정과 퓨처의 두 가지 생애 단계
pin!(future) 가 왜 필요한지가 이 회차 마지막 개념이다. Pin 은 비동기 Rust에서
제일 어렵다고들 하는데, 왜 필요한지부터 보면 이해가 된다.
async 블록은 상태 기계로 컴파일된다고 했다. 그런데 이런 코드를 보자.
async fn f() {
let value = compute();
let r = &value; // 지역 변수를 가리키는 레퍼런스
something().await; // 중단점 — value와 r 둘 다 상태 기계에 저장된다
println!("{r}");
}
r 은 value 를 가리키는데, 둘 다 같은 상태 기계 구조체의 필드다. 즉
구조체가 자기 자신의 다른 필드를 가리킨다(자기 참조 구조체).
이제 이 구조체를 메모리 상에서 옮기면 어떻게 될까? value 의 주소가 바뀌는데
r 은 옛 주소를 그대로 들고 있다. 댕글링 포인터다.
그래서 퓨처는 두 가지 생애 단계를 가진다.
- 생성 직후 — 아직 poll된 적이 없다. 자기 참조가 생기지 않았으므로 자유롭게 옮길 수 있다
- 첫 poll 이후 — 자기 참조가 생겼을 수 있다. 더 이상 옮기면 안 된다
이 "옮기면 안 됨"을 타입으로 표현한 게 Pin<P> 다. Pin<&mut F> 를 들고 있으면
F 를 밖으로 꺼낼 수 없고, 따라서 옮길 수 없다. Future::poll 의 수신자가
self: Pin<&mut Self> 인 이유가 정확히 이것이다. poll하려면 먼저 핀으로 고정해야
한다.
핀을 만드는 방법은 두 가지다.
Box::pin(future)— 힙에 놓는다. 힙 주소는 안 변하므로 안전하다.Pin<Box<F>>를 얻는다pin!(future)— 스택에 놓고 매크로가 원래 변수를 가려서(shadow) 접근을 막는다. 할당이 없어서 빠르다
block_on 에서 pin! 을 쓰는 이유가 이거다. 할당 없이 스택에서 고정한다.
Unpin 트레잇
그런데 모든 타입이 이 걱정을 할 필요는 없다. i32 나 String 은 자기 참조가
없으니 옮겨도 아무 문제가 없다. 이런 타입은 Unpin 을 구현한다.
- 대부분의 타입은 자동으로
Unpin이다 async블록이 만드는 퓨처는Unpin이 아니다
T: Unpin 이면 Pin<&mut T> 에서 자유롭게 &mut T 를 꺼낼 수 있다. 즉 핀이
아무 제약도 걸지 않는다. Part 1의 send_as_json 시그니처에 S: ... + Unpin 이
붙어 있던 이유가 이거다. 스트림을 자유롭게 다루려면 Unpin 이어야 편하다.
Pin 은 결국 "제자리 초기화된 자기 참조 구조체를 안전하게 다루기 위한 최소한의
장치" 다. 이름이 무섭지만, 하는 일은 "옮기지 마"라는 표시 하나다.
비동기 코드는 언제 써야 좋을까
마지막 슬라이드들이 이 질문을 다룬다. 실전 조언에 가까운 부분이라 정리해둔다.
비동기가 맞는 경우
- 동시 연결이 아주 많고 대부분 I/O 대기인 서버
- 대기 시간이 지배적인 클라이언트 (여러 API를 동시에 호출)
- 태스크는 많은데 각각은 대부분 놀고 있는 경우
스레드가 맞는 경우
- CPU 바운드 계산 (Rayon이 더 낫다)
- 동시 작업 개수가 적을 때 (수십 개면 스레드로 충분하다)
- 코드가 단순하기를 바랄 때
비동기의 비용
이걸 강의에서 솔직하게 짚어준 게 좋았다.
- 함수 색깔 문제 —
async함수는async문맥에서만 부를 수 있다. 한 곳을 비동기로 만들면 호출 사슬 전체가 비동기가 되는 전염성이 있다 - 런타임 선택 —
tokio냐async-std냐를 골라야 하고, 생태계가 갈린다 - 디버깅 — 스택 트레이스가 상태 기계를 거치면서 읽기 어려워진다
Send바운드와Pin— 컴파일 에러가 동기 코드보다 훨씬 난해하다
그래서 결론은 "필요할 때만"이었다. 만 개 연결을 감당해야 하는 게 아니라면 스레드가 더 단순하고 더 빠르게 짜인다. 비동기는 성능이 필요해서 복잡도를 사는 것이지, 더 좋은 코드를 쓰는 방법이 아니다.
특강을 마치며
12편에 걸쳐 정리한 내용을 돌아보면, 결국 이 언어는 하나의 아이디어에서 전부 파생된다.
- 소유권 — 값의 책임자를 하나로 정한다
- 빌림 — 소유권을 넘기지 않고 쓰는 법. 공유는 여럿, 변경은 하나
- 수명 — 레퍼런스가 대상보다 오래 살지 않게 검사
- 이 규칙이 스레드 경계로 확장되면
Send/Sync가 되고, 데이터 경합이 사라진다 - 이 규칙이 비동기로 확장되면
.await를 가로지르는 값에 대한 제약이 된다 Option/Result는 널과 예외를 타입으로 대체하고, 패턴 매칭이 그걸 꺼낸다- 트레잇은 상속 없이 다형성을 주고, 정적/동적 디스패치를 고르게 한다
- 클로저·이터레이터·매크로는 그 위에서 런타임 비용 없이 추상화를 얹는다
컴파일러와 싸우는 시간이 길었던 건 사실이다. 하지만 그 싸움의 대부분은 "이 값의 소유권이 지금 누구에게 있는가"를 내가 명확히 정하지 않은 탓이었다. 컴파일러가 까다로운 게 아니라, 원래 답해야 했던 질문을 미뤄뒀던 것이다.
특강 이름에 있던 Linux Kernel 파트는 이 언어 기반 위에 올라가는 내용이다.
unsafe, FFI, no_std, 그리고 Rust for Linux 프로젝트가 그 방향인데,
그건 다음 공부거리로 남겨둔다.
정리
- 비동기 서버는 연결당 태스크. 스레드 풀 위에서 수만 개가 돈다
- 락 안에서
.await해야 하면 비동기 뮤텍스, 아니면 동기 뮤텍스 - 브로드캐스트 채널의
Lagged처리처럼, 백프레셔 정책은 설계 결정이다 - 퓨처의 프로토콜은 "준비 안 됐으면 웨이커를 남기고
Pending" 한 줄이다 - 최소
block_on은 poll →park루프 하나면 된다 Pin은 자기 참조 상태 기계를 옮기지 못하게 막는 표시다- 비동기는 공짜가 아니다. 필요할 때만 쓴다