우아한 종료와 정리
앞 절의 요청 처리를 유지하면서 새 작업 전송을 끝내고, ThreadPool의 Drop에서 남은 작업과 Worker 스레드 종료를 기다립니다.
채널과 execute 구현까지의 코드는 의도대로 스레드 풀로 요청에 비동기로 응답합니다. workers, id, thread 필드를 직접 쓰지 않는다는 경고가 나오는데, 이것은 아무것도 정리하지 않고 있다는 것을 일깨워 줍니다. 덜 우아한 Ctrl+C 방법으로 메인 스레드를 멈추면, 다른 스레드들도 요청을 처리하던 중이더라도 곧바로 멈춥니다.
그다음 풀의 각 스레드에 join을 불러, 끝나기 전에 처리하던 요청을 마칠 수 있도록 Drop 트레이트를 구현합니다. 그다음 스레드들에게 새 요청을 받지 말고 끝내라고 알리는 방법을 구현합니다. 이 코드가 동작하는 것을 보려고, 서버를 고쳐 요청 두 개만 받은 뒤 스레드 풀을 우아하게 끝내게 합니다.
진행하면서 알아 둘 점이 하나 있습니다. 이 가운데 어느 것도 클로저의 실행을 처리하는 코드 부분에는 영향을 주지 않으므로, 비동기 런타임에 스레드 풀을 쓰더라도 여기서의 모든 것은 같을 것입니다.
ThreadPool에 대한 Drop 트레이트 구현하기
스레드 풀에 Drop을 구현하는 것으로 시작합니다. 풀이 버려질 때 스레드들이 모두 합류(join)해 작업을 마치게 해야 합니다. 다음 코드는 Drop 구현의 첫 시도입니다. 이 코드는 아직 제대로 동작하지 않습니다.
impl Drop for ThreadPool {
fn drop(&mut self) {
for worker in &mut self.workers {
println!("Shutting down worker {}", worker.id);
worker.thread.join().unwrap();
}
}
}먼저 스레드 풀의 각 workers를 돕니다. self가 가변 참조이고 worker를 바꿀 수도 있어야 하므로 &mut를 씁니다. 각 worker마다 이 특정 Worker 인스턴스가 끝나고 있다는 메시지를 출력하고, 그 Worker 인스턴스의 스레드에 join을 부릅니다. join 호출이 실패하면, unwrap으로 러스트가 패닉을 내고 우아하지 않게 끝나게 합니다.
이 코드를 채널과 execute 구현의 src/lib.rs 전체에서 ThreadPool의 impl 블록 바로 뒤에 더하고 컴파일하면 다음 오류가 납니다.
error[E0507]: cannot move out of `worker.thread` which is behind a mutable reference
--> src/lib.rs:52:13
|
52 | worker.thread.join().unwrap();
| ^^^^^^^^^^^^^ ------ `worker.thread` moved due to this method call
| |
| move occurs because `worker.thread` has type `JoinHandle<()>`, which does not implement the `Copy` trait
|
note: `JoinHandle::<T>::join` takes ownership of the receiver `self`, which moves `worker.thread`
--생략--
For more information about this error, try `rustc --explain E0507`.마지막 note가 가리키는 표준 라이브러리의 경로는 컴파일러 버전마다 달라 생략했습니다. 이 오류는 각 worker의 가변 대여만 가지고 있고, join은 인수의 소유권을 가져가므로 join을 부를 수 없다고 알려 줍니다. 이 문제를 풀려면, join이 스레드를 소비할 수 있도록 스레드를 가진 Worker 인스턴스에서 스레드를 꺼내야 합니다. Worker가 Option<thread::JoinHandle<()>>를 담게 하면, Option의 take 메서드로 Some 배리언트에서 값을 꺼내고 그 자리에 None을 남길 수 있습니다. 다시 말해 실행 중인 Worker는 thread에 Some 배리언트를 가지고, Worker를 정리하고 싶을 때는 Some을 None으로 바꿔 Worker에 실행할 스레드가 없게 합니다.
그래서 Worker의 정의를 다음처럼 바꿉니다.
struct Worker {
id: usize,
thread: Option<thread::JoinHandle<()>>,
}새 Worker를 만들 때는 thread 값을 Some으로 감쌉니다. 필드의 타입이 Option<thread::JoinHandle<()>>로 바뀌었으므로, 만든 직후에는 스레드가 있다는 뜻으로 Some에 담아 둡니다.
impl Worker {
fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
// --생략--
Worker {
id,
thread: Some(thread),
}
}
}그리고 Drop 구현에서는 take로 Worker의 스레드를 꺼내, Some이면 join을 부릅니다. take는 Option에서 값을 꺼내고 그 자리에 None을 남기므로, 가변 참조만 가진 drop 안에서도 JoinHandle의 소유권을 얻어 join에 넘길 수 있습니다.
impl Drop for ThreadPool {
fn drop(&mut self) {
for worker in &mut self.workers {
println!("Shutting down worker {}", worker.id);
if let Some(thread) = worker.thread.take() {
thread.join().unwrap();
}
}
}
}Worker의 thread가 이미 None이라면 그 Worker는 이미 스레드가 정리된 것이므로, 그때는 아무 일도 하지 않습니다. Option을 쓰는 대신 for worker in self.workers.drain(..)처럼 벡터에서 Worker의 소유권을 통째로 꺼내 worker.thread.join()을 부르는 방법도 있습니다.
이제 코드는 경고 없이 컴파일됩니다. 하지만 나쁜 소식은 이 코드가 아직 원하는 대로 동작하지 않는다는 것입니다. 열쇠는 Worker 인스턴스의 스레드가 실행하는 클로저의 로직입니다. 지금은 join을 부르지만, 스레드들은 작업을 찾으며 영원히 돌고 있으므로 끝나지 않습니다. 지금의 drop 구현으로 ThreadPool을 버리려 하면, 메인 스레드는 첫 스레드가 끝나기를 영원히 기다리며 막힐 것입니다.
이 문제를 고치려면 ThreadPool의 drop 구현을 바꾸고, 그다음 Worker의 루프도 바꿔야 합니다.
작업을 기다리는 스레드에게 정지 신호 보내기
먼저 ThreadPool의 drop 구현에서, 스레드가 끝나기를 기다리기 전에 송신자를 명시적으로 버리도록 바꿉니다. 다음 코드는 ThreadPool이 송신자를 명시적으로 버리도록 바꾼 것입니다. 스레드를 꺼낼 때처럼, 여기서도 Option::take로 ThreadPool에서 송신자를 꺼낼 수 있도록 Option을 씁니다.
pub struct ThreadPool {
workers: Vec<Worker>,
sender: Option<mpsc::Sender<Job>>,
}
// --생략--
impl ThreadPool {
pub fn new(size: usize) -> ThreadPool {
// --생략--
ThreadPool {
workers,
sender: Some(sender),
}
}
pub fn execute<F>(&self, f: F)
where
F: FnOnce() + Send + 'static,
{
let job = Box::new(f);
self.sender.as_ref().unwrap().send(job).unwrap();
}
}
impl Drop for ThreadPool {
fn drop(&mut self) {
drop(self.sender.take());
for worker in &mut self.workers {
println!("Shutting down worker {}", worker.id);
if let Some(thread) = worker.thread.take() {
thread.join().unwrap();
}
}
}
}sender를 버리면 채널이 닫히며, 그것은 더 이상 메시지가 보내지지 않는다는 뜻입니다. 그러면 Worker 인스턴스들이 무한 루프에서 하는 recv 호출이 모두 오류를 돌려줄 것입니다. 다음 코드에서는 Worker의 루프를 바꿔, 그 경우에 루프를 우아하게 빠져나오게 합니다. 그러면 ThreadPool의 drop 구현이 스레드에 join을 부를 때 스레드가 끝날 것입니다.
impl Worker {
fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
let thread = thread::spawn(move || {
loop {
let message = receiver.lock().unwrap().recv();
match message {
Ok(job) => {
println!("Worker {id} got a job; executing.");
job();
}
Err(_) => {
println!("Worker {id} disconnected; shutting down.");
break;
}
}
}
});
Worker {
id,
thread: Some(thread),
}
}
}이 코드가 동작하는 것을 보기 위해, 다음 코드처럼 main을 고쳐 서버를 우아하게 끝내기 전에 요청 두 개만 받게 합니다. take(2)는 incoming이 내주는 연결 가운데 앞의 두 개만 꺼내는 반복자 어댑터이므로, 두 요청을 넘기고 나면 반복이 끝나고 main의 끝에서 pool이 버려집니다.
fn main() {
let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
let pool = ThreadPool::new(4);
for stream in listener.incoming().take(2) {
let stream = stream.unwrap();
pool.execute(|| {
handle_connection(stream);
});
}
println!("Shutting down.");
}실제 웹 서버가 요청 두 개만 처리하고 끝나기를 바라지는 않을 것입니다. 이 코드는 우아한 종료와 정리가 동작한다는 것을 보여 줄 뿐입니다.
take 메서드는 Iterator 트레이트에 정의되어 있으며, 반복을 최대 처음 두 항목으로 제한합니다. main이 끝날 때 ThreadPool은 스코프를 벗어나고, drop 구현이 실행됩니다.
cargo run으로 서버를 시작하고 요청 세 개를 보냅니다. 세 번째 요청은 오류가 나야 하며, 터미널에서는 다음과 비슷한 출력을 볼 것입니다.
Worker 0 got a job; executing.
Shutting down.
Shutting down worker 0
Worker 3 got a job; executing.
Worker 1 disconnected; shutting down.
Worker 2 disconnected; shutting down.
Worker 3 disconnected; shutting down.
Worker 0 disconnected; shutting down.
Shutting down worker 1
Shutting down worker 2
Shutting down worker 3Worker의 ID와 출력되는 메시지의 순서는 다를 수 있습니다. 메시지를 보면 이 코드가 어떻게 동작하는지 알 수 있습니다. Worker 인스턴스 0과 3이 처음 두 요청을 받았습니다. 서버는 두 번째 연결 뒤 연결을 받는 것을 멈추고, ThreadPool의 Drop 구현은 Worker 3가 작업을 시작하기도 전에 실행되기 시작합니다. sender를 버리면 모든 Worker 인스턴스의 연결이 끊어지고 끝내라는 신호가 됩니다. Worker 인스턴스들은 연결이 끊길 때 각각 메시지를 출력하고, 그다음 스레드 풀이 join을 불러 각 Worker 스레드가 끝나기를 기다립니다.
이 특정 실행의 흥미로운 면 하나를 보세요. ThreadPool은 sender를 버렸고, 어느 Worker가 오류를 받기 전에 Worker 0에 합류하려 했습니다. Worker 0은 아직 recv에서 오류를 받지 않았으므로, 메인 스레드는 Worker 0이 끝나기를 기다리며 막혔습니다. 그동안 Worker 3은 작업을 받았고, 그다음 모든 스레드가 오류를 받았습니다. Worker 0이 끝나자 메인 스레드는 나머지 Worker 인스턴스가 끝나기를 기다렸습니다. 그 시점에 그것들은 모두 루프를 빠져나와 멈췄습니다. 아래 그림은 이 종료 과정을 정리한 것입니다.
스레드 풀 우아한 종료 순서
Drop에서 Sender를 take·drop하고 각 JoinHandle을 take·join하는 메인 흐름과 워커의 잔여 Job 처리→recv Err→break가 겹친다. 채널 닫기와 강제 중단, join 대기와 종료 신호를 혼동하지 않는 상태 추적 가치가 높다.
RUST · 두 스레드의 종료 흐름
채널을 닫고, 남은 작업이 끝날 때까지 join으로 기다린다
메인 스레드의 정리와 워커의 마지막 작업 처리는 함께 진행됩니다. join은 종료 신호를 보내는 연산이 아니라 종료를 기다리는 연산입니다.
메인의 join 대기와 다른 워커의 잔여 작업 실행은 겹친다. 로그 출력 순서는 달라질 수 있다.
코드·조건·비교표와 보충 설명
메인 스레드 · ThreadPool::drop
Sender를 먼저 꺼내 버린다
drop(self.sender.take()). 이 풀의 Sender가 사라져 새 작업 전송이 끝납니다. 다른 Sender 복제본도 모두 사라져야 채널이 닫힙니다.
워커 핸들의 소유권을 꺼낸다
worker.thread.take()는 Some(handle)을 반환하고 필드에는 None을 남깁니다.
join으로 종료를 기다린다
handle.join()은 핸들을 소비하며 해당 워커가 끝날 때까지 기다립니다. 남은 워커도 차례대로 기다립니다.
워커 스레드 · 수신 루프
이미 보낸 Job은 계속 받는다
recv()가 Ok(job)을 반환하면 수신 잠금을 해제하고 job()을 실행합니다.
모든 Sender와 대기 작업이 사라진다
이미 실행 중인 작업은 완료합니다. 닫힌 채널의 버퍼가 비면 다음 recv()는 오류를 반환합니다.
Err 갈래에서 break한다
루프를 벗어나 워커 스레드가 끝납니다. 해당 워커를 기다리던 join()도 반환할 수 있습니다.
겹쳐 진행되는 부분
메인 스레드가 첫 워커의 join()에서 기다리는 동안, 다른 워커는 대기 작업을 받거나 실행할 수 있습니다. 로그 순서는 실행마다 달라질 수 있습니다. 모든 join().unwrap()이 성공하면 풀의 Drop이 완료됩니다.
Sender를 버렸다고 워커가 즉시 중단되지는 않습니다. 종료 순서는 채널 닫기 → 남은 작업 완료 → 각 워커 종료 확인입니다.
이 예의 조건
- 본문과 원 도표에 제시된 사례의 입력·조건을 사용합니다. 설명용 수치나 독립 비교 경로는 각 단계에 표시합니다.
- 현재 상태·값과 전환 근거의 위치는 모든 단계에서 같습니다. 조건이 맞지 않거나 실행되지 않은 경로를 성공한 동작으로 해석하지 않습니다.
| 관찰·상황 | 값·보충 정보 | 변화·이유 |
|---|---|---|
| 송신 닫기 pool Drop이 sender.take 값을 drop한다. |
|
|
| 잔여 작업·join 메인이 handle을 take·join하는 동안 worker가 남은 job을 처리한다. |
|
|
| 채널 소진·완료 버퍼가 비어 recv가 Err가 되면 worker가 break하고 join이 반환한다. |
|
|
스레드 풀 우아한 종료 순서
Drop에서 Sender를 take·drop하고 각 JoinHandle을 take·join하는 메인 흐름과 워커의 잔여 Job 처리→recv Err→break가 겹친다. 채널 닫기와 강제 중단, join 대기와 종료 신호를 혼동하지 않는 상태 추적 가치가 높다.
RUST · 두 스레드의 종료 흐름
채널을 닫고, 남은 작업이 끝날 때까지 join으로 기다린다
메인 스레드의 정리와 워커의 마지막 작업 처리는 함께 진행됩니다. join은 종료 신호를 보내는 연산이 아니라 종료를 기다리는 연산입니다.
메인의 join 대기와 다른 워커의 잔여 작업 실행은 겹친다. 로그 출력 순서는 달라질 수 있다.
코드·조건·비교표와 보충 설명
메인 스레드 · ThreadPool::drop
Sender를 먼저 꺼내 버린다
drop(self.sender.take()). 이 풀의 Sender가 사라져 새 작업 전송이 끝납니다. 다른 Sender 복제본도 모두 사라져야 채널이 닫힙니다.
워커 핸들의 소유권을 꺼낸다
worker.thread.take()는 Some(handle)을 반환하고 필드에는 None을 남깁니다.
join으로 종료를 기다린다
handle.join()은 핸들을 소비하며 해당 워커가 끝날 때까지 기다립니다. 남은 워커도 차례대로 기다립니다.
워커 스레드 · 수신 루프
이미 보낸 Job은 계속 받는다
recv()가 Ok(job)을 반환하면 수신 잠금을 해제하고 job()을 실행합니다.
모든 Sender와 대기 작업이 사라진다
이미 실행 중인 작업은 완료합니다. 닫힌 채널의 버퍼가 비면 다음 recv()는 오류를 반환합니다.
Err 갈래에서 break한다
루프를 벗어나 워커 스레드가 끝납니다. 해당 워커를 기다리던 join()도 반환할 수 있습니다.
겹쳐 진행되는 부분
메인 스레드가 첫 워커의 join()에서 기다리는 동안, 다른 워커는 대기 작업을 받거나 실행할 수 있습니다. 로그 순서는 실행마다 달라질 수 있습니다. 모든 join().unwrap()이 성공하면 풀의 Drop이 완료됩니다.
Sender를 버렸다고 워커가 즉시 중단되지는 않습니다. 종료 순서는 채널 닫기 → 남은 작업 완료 → 각 워커 종료 확인입니다.
이 예의 조건
- 본문과 원 도표에 제시된 사례의 입력·조건을 사용합니다. 설명용 수치나 독립 비교 경로는 각 단계에 표시합니다.
- 현재 상태·값과 전환 근거의 위치는 모든 단계에서 같습니다. 조건이 맞지 않거나 실행되지 않은 경로를 성공한 동작으로 해석하지 않습니다.
| 관찰·상황 | 값·보충 정보 | 변화·이유 |
|---|---|---|
| 송신 닫기 pool Drop이 sender.take 값을 drop한다. |
|
|
| 잔여 작업·join 메인이 handle을 take·join하는 동안 worker가 남은 job을 처리한다. |
|
|
| 채널 소진·완료 버퍼가 비어 recv가 Err가 되면 worker가 break하고 join이 반환한다. |
|
|
메인 스레드의 ThreadPool::drop은 먼저 drop(self.sender.take())로 Sender를 꺼내 버려 새 작업의 전송을 끝냅니다. 다른 Sender 복제본이 있다면 그것도 모두 사라져야 채널이 닫힙니다. 그다음 worker.thread.take()가 Some(handle)을 돌려주고 필드에는 None을 남기며, handle.join()은 핸들을 소비해 그 워커가 끝날 때까지 기다립니다. join은 끝내라는 신호를 보내는 연산이 아니라 끝나기를 기다리는 연산입니다. 워커 스레드 쪽에서는 이미 보낸 Job을 계속 받아, recv()가 Ok(job)을 돌려주면 수신 잠금을 풀고 job()을 실행합니다. 실행 중인 작업은 끝까지 마치고, 닫힌 채널의 버퍼가 비면 다음 recv()가 오류를 돌려주어 Err 갈래에서 break하며 워커 스레드가 끝나고, 그 워커를 기다리던 join()도 돌아옵니다. 메인 스레드가 첫 워커의 join()에서 기다리는 동안 다른 워커는 대기 중인 작업을 받거나 실행할 수 있으므로, 로그의 순서는 실행할 때마다 다를 수 있습니다. 모든 join().unwrap()이 성공하면 풀의 Drop이 끝납니다. Sender를 버렸다고 워커가 곧바로 멈추는 것이 아니라, 채널 닫기, 남은 작업 마치기, 각 워커의 종료 확인의 순서로 끝납니다.
축하합니다. 프로젝트를 끝냈습니다. 스레드 풀을 써서 비동기로 응답하는 기본 웹 서버가 생겼습니다. 서버를 우아하게 끝낼 수 있으며, 풀의 모든 스레드를 정리합니다. 다음은 참고용 전체 코드입니다.
use hello::ThreadPool;
use std::{
fs,
io::{BufReader, prelude::*},
net::{TcpListener, TcpStream},
thread,
time::Duration,
};
fn main() {
let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
let pool = ThreadPool::new(4);
for stream in listener.incoming().take(2) {
let stream = stream.unwrap();
pool.execute(|| {
handle_connection(stream);
});
}
println!("Shutting down.");
}
fn handle_connection(mut stream: TcpStream) {
let buf_reader = BufReader::new(&stream);
let request_line = buf_reader.lines().next().unwrap().unwrap();
let (status_line, filename) = match &request_line[..] {
"GET / HTTP/1.1" => ("HTTP/1.1 200 OK", "hello.html"),
"GET /sleep HTTP/1.1" => {
thread::sleep(Duration::from_secs(5));
("HTTP/1.1 200 OK", "hello.html")
}
_ => ("HTTP/1.1 404 NOT FOUND", "404.html"),
};
let contents = fs::read_to_string(filename).unwrap();
let length = contents.len();
let response =
format!("{status_line}\r\nContent-Length: {length}\r\n\r\n{contents}");
stream.write_all(response.as_bytes()).unwrap();
}use std::{
sync::{Arc, Mutex, mpsc},
thread,
};
pub struct ThreadPool {
workers: Vec<Worker>,
sender: Option<mpsc::Sender<Job>>,
}
type Job = Box<dyn FnOnce() + Send + 'static>;
impl ThreadPool {
/// Create a new ThreadPool.
///
/// The size is the number of threads in the pool.
///
/// # Panics
///
/// The `new` function will panic if the size is zero.
pub fn new(size: usize) -> ThreadPool {
assert!(size > 0);
let (sender, receiver) = mpsc::channel();
let receiver = Arc::new(Mutex::new(receiver));
let mut workers = Vec::with_capacity(size);
for id in 0..size {
workers.push(Worker::new(id, Arc::clone(&receiver)));
}
ThreadPool {
workers,
sender: Some(sender),
}
}
pub fn execute<F>(&self, f: F)
where
F: FnOnce() + Send + 'static,
{
let job = Box::new(f);
self.sender.as_ref().unwrap().send(job).unwrap();
}
}
impl Drop for ThreadPool {
fn drop(&mut self) {
drop(self.sender.take());
for worker in &mut self.workers {
println!("Shutting down worker {}", worker.id);
if let Some(thread) = worker.thread.take() {
thread.join().unwrap();
}
}
}
}
struct Worker {
id: usize,
thread: Option<thread::JoinHandle<()>>,
}
impl Worker {
fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
let thread = thread::spawn(move || {
loop {
let message = receiver.lock().unwrap().recv();
match message {
Ok(job) => {
println!("Worker {id} got a job; executing.");
job();
}
Err(_) => {
println!("Worker {id} disconnected; shutting down.");
break;
}
}
}
});
Worker {
id,
thread: Some(thread),
}
}
}여기서 더 할 수 있는 일이 있습니다. 이 프로젝트를 계속 다듬고 싶다면, 다음 아이디어를 해 보세요.
| 아이디어 |
|---|
ThreadPool과 그 공개 메서드에 문서를 더 단다. |
| 라이브러리의 기능에 테스트를 더한다. |
unwrap 호출을 더 견고한 오류 처리로 바꾼다. |
웹 요청을 처리하는 것 말고 다른 일에 ThreadPool을 쓴다. |
| crates.io에서 스레드 풀 크레이트를 찾아 그것으로 비슷한 웹 서버를 구현하고, 그 API와 견고함을 우리가 구현한 스레드 풀과 비교한다. |
더 깊이: 작업 하나가 패닉을 내면 풀은 어떻게 될까
job() 안에서 패닉이 나면, 그 패닉은 작업을 실행하던 워커의 스레드를 되감아 끝냅니다. std::thread 문서는 잡히지 않은 패닉이 그 스레드만 끝내고, 다른 스레드는 join으로 그것을 알아낼 수 있다고 설명합니다. 이 풀의 워커는 작업을 받는 let 문이 끝날 때 이미 락 가드를 버렸으므로, 작업이 패닉을 내도 수신자를 감싼 Mutex가 오염되지는 않습니다. 대신 패닉이 난 워커는 루프로 돌아오지 않으므로, 풀에서 작업을 받는 스레드가 그만큼 줄어듭니다.
Drop 구현도 이 경우를 만납니다. JoinHandle::join 문서에 따르면 합류할 스레드가 패닉으로 끝났다면 join은 panic!에 넘긴 값을 담은 Err를 돌려줍니다. 그래서 thread.join().unwrap()은 패닉으로 끝난 워커를 만나면 ThreadPool의 drop 안에서 다시 패닉을 내고, 그 뒤에 남은 워커에는 합류하지 못합니다. 남은 워커까지 기다리고 싶다면 join의 Err를 기록만 하고 다음 워커로 넘어가도록 고칩니다. 앞의 표에 있는 unwrap을 더 견고한 오류 처리로 바꾸는 일의 한 예가 이것입니다.
Tokio 런타임으로 확장하기
이 절은 선택 사항입니다. 지금까지는 정해진 수의 운영체제 워커가 동기 I/O 작업을 처리했습니다. 비동기 태스크가 실행되고 기다리는 흐름은 Sync와 Send 트레이트를 이용한 확장 가능한 동시성의 첫 Tokio 예제에서 보았습니다. 같은 서버를 Tokio로 쓴다면 연결마다 태스크를 만들고, 소켓을 읽고 쓸 때 .await로 기다립니다.
아래 코드는 비동기 소켓과 tokio::spawn의 모양만 보여 주는 별도의 예입니다. 요청을 한 번 읽고 OK만 쓰며, 읽은 길이와 I/O 오류는 처리하지 않습니다. 이 글의 스레드 풀 프로젝트에 이어 붙이는 코드가 아니며, Cargo.toml에 tokio = { version = "1.54.1", features = ["full"] }를 더한 별도 프로젝트에서 씁니다.
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;
#[tokio::main]
async fn main() -> std::io::Result<()> {
let listener = TcpListener::bind("127.0.0.1:7878").await?;
loop {
let (mut socket, _addr) = listener.accept().await?;
tokio::spawn(async move {
let mut buf = [0; 1024];
let _ = socket.read(&mut buf).await;
let _ = socket
.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nOK")
.await;
});
}
}#[tokio::main]은 기본으로 여러 스레드를 쓰는 런타임을 준비하고, listener.accept().await는 연결이 올 때까지 스레드를 막지 않고 기다립니다. 연결마다 tokio::spawn으로 태스크를 만들지만, 태스크는 운영체제 스레드가 아니라 런타임의 몇몇 스레드 위에서 번갈아 실행되는 작은 작업 단위입니다. 연결마다 스레드를 만들던 방식과 달리 태스크를 아주 많이 만들어도 스레드가 그만큼 늘어나지 않습니다.
어떤 도구를 고를지는 일의 성격에 따라 다릅니다.
| 일의 성격 | 알맞은 도구 |
|---|---|
| CPU를 많이 쓰는 계산 | std::thread나 rayon 같은 스레드와 스레드 풀 |
| 기다리는 시간이 많은 I/O | Tokio 태스크와 비동기 소켓, 채널 |
락을 쥔 채 .await를 넘겨야 하는 공유 상태 | tokio::sync::Mutex |
| 메시지 전달 | 동기 코드 사이에는 std::sync::mpsc, 비동기 파이프라인에는 tokio::sync::mpsc |
Send와 Sync의 규칙은 Sync와 Send 트레이트를 이용한 확장 가능한 동시성에서 본 그대로 적용됩니다. 이 글의 스레드 풀 흐름과 비교해 어느 모델이 요구 사항에 맞는지 판단하면 됩니다.
최종 정리
여기까지 읽었다면 러스트로 프로젝트를 직접 구현하고, 다른 사람의 러스트 코드도 더 수월하게 읽을 수 있습니다. 앞으로는 공식 문서, 크레이트 문서, 커뮤니티의 예제를 함께 보며 필요한 주제를 채워 가면 됩니다. 마지막으로 스레드 풀 모델에서 Tokio 런타임으로 넓힐 때 무엇이 달라지는지 아래 그림으로 정리해 봅니다.
스레드 풀과 async 런타임 책임 경계
Future의 poll→Pending→실행 양보→wake→다시 poll→Ready라는 대기 상태 변화가 blocking job의 워커 점유와 대비된다. std/tokio Mutex 선택 표는 전체 관계도와 표로 보존하고 async도 OS 스레드 위에서 실행한다는 조건을 유지한다.
RUST · 실행 자원과 대기
스레드는 작업을 실행하고, 런타임은 준비된 태스크를 poll한다
스레드 풀의 blocking 작업과 Tokio의 async 태스크는 대기할 때 실행 자원을 쓰는 방식이 다릅니다.
async 안의 blocking 호출도 OS 스레드를 막는다. std Mutex는 await 전 해제하고, tokio Mutex의 async 잠금과는 별도로 선택한다.
코드·조건·비교표와 보충 설명
OS 스레드 풀
작업 대기열 → 워커
고정 워커가 blocking job을 하나씩 받아 실행합니다.
작업이 끝날 때까지 스레드 사용
동기 I/O로 막히거나 sleep하면 그 워커도 기다립니다. 워커 수가 실행 중인 작업 수를 제한합니다.
Async 런타임
준비된 태스크 → poll
스케줄러가 준비된 태스크의 Future를 poll합니다. 태스크는 런타임의 OS 스레드 위에서 실행됩니다.
Pending → 양보 → wake → 다시 poll
await한 Future가 Pending이면 실행을 양보합니다. 준비되면 wake로 다시 poll 대상이 됩니다. Ready면 await에서 계속 진행할 수 있습니다.
짧은 데이터 임계 구역
std::sync::Mutex를 검토합니다. async 코드에서도 await 전에 잠금을 풀고 짧게 데이터를 다룬다면 사용할 수 있습니다.
await 너머 잠금이 필요한 I/O
tokio::sync::Mutex를 검토합니다. lock 획득은 async이며 guard를 await 동안 유지하도록 설계되어 있습니다.
async 태스크 안에서도 blocking 호출은 런타임 스레드를 막습니다. 태스크 수를 늘리는 것과 OS 워커 수를 늘리는 것은 같은 변화가 아닙니다.
이 예의 조건
- 본문과 원 도표에 제시된 사례의 입력·조건을 사용합니다. 설명용 수치나 독립 비교 경로는 각 단계에 표시합니다.
- 현재 상태·값과 전환 근거의 위치는 모든 단계에서 같습니다. 조건이 맞지 않거나 실행되지 않은 경로를 성공한 동작으로 해석하지 않습니다.
| 관찰·상황 | 값·보충 정보 | 변화·이유 |
|---|---|---|
| poll 호출 준비된 task의 Future를 poll한다. |
|
|
| Pending 양보 준비되지 않은 Future가 Pending을 돌려준다. |
|
|
| wake·재 poll 깨움 뒤 다시 poll해 Ready이면 await 이후로 간다. |
|
|
스레드 풀과 async 런타임 책임 경계
Future의 poll→Pending→실행 양보→wake→다시 poll→Ready라는 대기 상태 변화가 blocking job의 워커 점유와 대비된다. std/tokio Mutex 선택 표는 전체 관계도와 표로 보존하고 async도 OS 스레드 위에서 실행한다는 조건을 유지한다.
RUST · 실행 자원과 대기
스레드는 작업을 실행하고, 런타임은 준비된 태스크를 poll한다
스레드 풀의 blocking 작업과 Tokio의 async 태스크는 대기할 때 실행 자원을 쓰는 방식이 다릅니다.
async 안의 blocking 호출도 OS 스레드를 막는다. std Mutex는 await 전 해제하고, tokio Mutex의 async 잠금과는 별도로 선택한다.
코드·조건·비교표와 보충 설명
OS 스레드 풀
작업 대기열 → 워커
고정 워커가 blocking job을 하나씩 받아 실행합니다.
작업이 끝날 때까지 스레드 사용
동기 I/O로 막히거나 sleep하면 그 워커도 기다립니다. 워커 수가 실행 중인 작업 수를 제한합니다.
Async 런타임
준비된 태스크 → poll
스케줄러가 준비된 태스크의 Future를 poll합니다. 태스크는 런타임의 OS 스레드 위에서 실행됩니다.
Pending → 양보 → wake → 다시 poll
await한 Future가 Pending이면 실행을 양보합니다. 준비되면 wake로 다시 poll 대상이 됩니다. Ready면 await에서 계속 진행할 수 있습니다.
짧은 데이터 임계 구역
std::sync::Mutex를 검토합니다. async 코드에서도 await 전에 잠금을 풀고 짧게 데이터를 다룬다면 사용할 수 있습니다.
await 너머 잠금이 필요한 I/O
tokio::sync::Mutex를 검토합니다. lock 획득은 async이며 guard를 await 동안 유지하도록 설계되어 있습니다.
async 태스크 안에서도 blocking 호출은 런타임 스레드를 막습니다. 태스크 수를 늘리는 것과 OS 워커 수를 늘리는 것은 같은 변화가 아닙니다.
이 예의 조건
- 본문과 원 도표에 제시된 사례의 입력·조건을 사용합니다. 설명용 수치나 독립 비교 경로는 각 단계에 표시합니다.
- 현재 상태·값과 전환 근거의 위치는 모든 단계에서 같습니다. 조건이 맞지 않거나 실행되지 않은 경로를 성공한 동작으로 해석하지 않습니다.
| 관찰·상황 | 값·보충 정보 | 변화·이유 |
|---|---|---|
| poll 호출 준비된 task의 Future를 poll한다. |
|
|
| Pending 양보 준비되지 않은 Future가 Pending을 돌려준다. |
|
|
| wake·재 poll 깨움 뒤 다시 poll해 Ready이면 await 이후로 간다. |
|
|
스레드 풀에서는 정해진 워커가 대기열에서 막히는(blocking) 작업을 하나씩 받아 실행하며, 작업이 동기 I/O로 막히거나 잠들면 그 워커도 함께 기다립니다. 그래서 워커의 수가 실행 중인 작업의 수를 제한합니다. 비동기 런타임에서는 스케줄러가 준비된 태스크의 Future를 실행(poll)하며, 태스크는 런타임이 가진 운영체제 스레드 위에서 실행됩니다. .await한 Future가 Pending이면 실행을 양보하고, 준비되면 깨워져(wake) 다시 실행 대상이 되며, 깨운다는 것이 곧바로 끝난다는 뜻은 아닙니다. Ready가 되면 .await 뒤의 코드로 이어집니다. 비동기 태스크 안에서도 막히는 호출은 런타임 스레드를 막으므로, 태스크의 수를 늘리는 것과 운영체제 워커의 수를 늘리는 것은 같은 일이 아닙니다. 짧게 데이터를 다루고 .await 전에 잠금을 푼다면 std::sync::Mutex도 쓸 수 있고, .await를 넘어 잠금을 쥐어야 한다면 잠금을 비동기로 얻는 tokio::sync::Mutex를 검토합니다.
정리
공식 Book을 바탕으로 한 이 교재를 끝까지 읽어 주셔서 감사합니다. 이제 자신만의 러스트 프로젝트를 구현하고 다른 사람의 프로젝트를 도울 준비가 되었습니다. 러스트로 여정을 이어 가며 만나는 어떤 어려움에든 기꺼이 도와줄 다른 러스타시안들의 따뜻한 커뮤니티가 있다는 것을 기억하세요.
ThreadPool의Drop에서 각Worker의 스레드에join을 불러 처리 중인 작업이 끝나기를 기다리며,join이 핸들의 소유권을 가져가므로Option과take로 핸들을 꺼냅니다.- 송신자도
Option으로 담아drop(self.sender.take())로 먼저 버리면 채널이 닫혀, 버퍼가 빈 뒤 워커의recv가 오류를 돌려줍니다. - 워커의 루프는
recv의 결과를match해Ok(job)이면 실행하고,Err이면break해 스레드를 끝냅니다. join은 끝내라는 신호가 아니라 끝나기를 기다리는 연산이며, 종료는 채널 닫기, 남은 작업 마치기, 각 워커의 종료 확인의 순서로 이루어집니다.- Tokio 같은 비동기 런타임에서는 연결마다 태스크를 만들고
.await로 기다리며, 막히는 호출은 여전히 런타임 스레드를 막습니다.
연습 문제
1. 송신자를 버리지 않으면
Drop 구현에서 drop(self.sender.take()); 줄을 지우면 서버가 요청 두 개를 처리한 뒤 어떻게 되나요?
해설 보기
ThreadPool의 필드는 Drop::drop이 끝난 뒤에야 버려지므로, drop 안에서는 송신자가 아직 살아 있어 채널이 닫히지 않습니다. 그래서 워커들은 recv에서 계속 새 작업을 기다리고, 메인 스레드는 첫 워커의 join에서 영원히 막혀 프로그램이 끝나지 않습니다. 송신자를 먼저 버려야 워커들이 Err를 받고 루프를 빠져나옵니다.
2. 처리 중인 요청이 있을 때
/sleep 요청 하나를 처리하는 중에 main의 for 루프가 끝나면, /sleep의 응답은 클라이언트에 도착하나요? Ctrl+C로 프로세스를 멈추면 어떻게 되나요?
해설 보기
for 루프가 끝나 ThreadPool이 버려지면, Drop은 송신자를 버린 뒤 각 워커의 join을 기다리므로, /sleep을 처리하던 워커는 5초를 다 채우고 응답을 쓴 다음 끝납니다. 그래서 응답은 도착합니다. 반면 Ctrl+C로 프로세스를 멈추면 Drop이 실행되지 않고 모든 스레드가 곧바로 멈추므로, 처리 중이던 응답은 끝나지 않습니다. Ctrl+C에도 우아하게 끝내려면 신호를 받아 루프를 끝내는 코드를 따로 더해야 합니다.
이것으로 19장과 본문을 모두 마칩니다. 부록의 A - 키워드부터는 러스트를 쓰며 참고할 키워드, 연산자, 파생 가능한 트레이트, 개발 도구 같은 자료를 정리합니다.