싱글 스레드 서버에서 멀티스레드 서버로
싱글 스레드 서버에서 멀티스레드 서버로 (From a Single-Threaded to a Multithreaded Server)
지금 서버는 각 요청을 차례로 처리해요. 즉 첫 연결의 처리가 끝나기 전에는 두 번째 연결을 처리하지 않는다는 뜻이에요. 서버가 점점 더 많은 요청을 받으면, 이 직렬 실행은 점점 더 비효율적이 돼요. 서버가 처리하는 데 오래 걸리는 요청을 받으면, 뒤따르는 요청들이 빨리 처리될 수 있더라도 그 긴 요청이 끝날 때까지 기다려야 하죠. 이걸 고쳐야 하는데, 먼저 문제가 실제로 어떻게 생기는지 살펴볼게요.
출처: The Rust Book
느린 요청 시뮬레이션하기 (Simulating a Slow Request)
천천히 처리되는 요청이 현재 서버 구현에 들어온 다른 요청들에 어떻게 영향을 줄 수 있는지 살펴볼게요. Listing 21-10은 /sleep 요청 처리를 구현하는데, 서버가 응답하기 전에 5초 동안 잠들게 하는 시뮬레이션된 느린 응답을 만들어요.
use std::{
fs,
io::{BufReader, prelude::*},
net::{TcpListener, TcpStream},
thread,
time::Duration,
};
// --snip--
fn main() {
let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
for stream in listener.incoming() {
let stream = stream.unwrap();
handle_connection(stream);
}
}
fn handle_connection(mut stream: TcpStream) {
// --snip--
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"),
};
// --snip--
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();
}
이제 경우가 세 개라서 if에서 match로 바꿨어요. 문자열 리터럴 값에 패턴-매칭하려면 request_line의 슬라이스에 명시적으로 매치해야 해요. match는 동등성(equality) 메서드처럼 자동 참조·역참조를 하지 않거든요.
첫 번째 팔은 Listing 21-9의 if 블록과 같아요. 두 번째 팔은 /sleep 요청과 일치해요. 그 요청을 받으면 서버는 성공 HTML 페이지를 렌더링하기 전에 5초 동안 잠들어요. 세 번째 팔은 Listing 21-9의 else 블록과 같아요.
우리 서버가 얼마나 원시적인지 볼 수 있죠. 실제 라이브러리들은 여러 요청을 인식하는 것을 훨씬 덜 장황한 방식으로 처리할 거예요!
cargo run으로 서버를 시작하세요. 그다음 브라우저 창 두 개를 여세요. 하나는 _http://127.0.0.1:7878_용이고 다른 하나는 _http://127.0.0.1:7878/sleep_용이에요. 이전처럼 / URI를 몇 번 입력하면 빠르게 응답하는 걸 볼 수 있어요. 하지만 _/sleep_을 입력한 다음 _/_을 로드하면, sleep이 5초를 완전히 잔 뒤에야 _/_이 로드되는 걸 볼 거예요.
느린 요청 뒤에 요청들이 밀리는 것을 피하기 위해 쓸 수 있는 기법이 여러 가지 있어요. 17장에서 했던 것처럼 async를 쓰는 것도 있죠. 우리가 구현할 것은 스레드 풀이에요.
스레드 풀로 처리량 개선하기 (Improving Throughput with a Thread Pool)
_스레드 풀(thread pool)_은 작업을 처리할 준비가 되어 기다리고 있는 스폰된(spawned) 스레드들의 집단이에요. 프로그램이 새 작업을 받으면 풀의 스레드 중 하나를 그 작업에 할당하고, 그 스레드가 작업을 처리해요. 풀의 나머지 스레드들은 첫 스레드가 처리하는 동안 들어오는 다른 작업들을 처리할 수 있어요. 첫 스레드가 자기 작업 처리를 마치면 유휴 스레드 풀로 돌아와 새 작업을 처리할 준비를 해요. 스레드 풀은 연결을 동시에 처리하게 해주므로, 서버의 처리량을 높여줘요.
풀의 스레드 수는 DoS 공격으로부터 우리를 보호하기 위해 작은 수로 제한할 거예요. 요청이 올 때마다 프로그램이 새 스레드를 만든다면, 우리 서버에 1천만 개의 요청을 하는 사람이 우리 서버의 모든 리소스를 사용해 버려 요청 처리를 마비시킬 수 있거든요.
그래서 무제한의 스레드를 스폰하는 대신, 고정된 수의 스레드가 풀에서 기다리게 할 거예요. 들어오는 요청은 처리를 위해 풀로 보내져요. 풀은 들어오는 요청의 큐(queue)를 유지해요. 풀의 각 스레드는 이 큐에서 요청을 하나 꺼내(pop), 요청을 처리하고, 큐에 또 다른 요청을 요청해요. 이 설계로 우리는 _N_개까지 요청을 동시에 처리할 수 있는데, 여기서 _N_은 스레드 수예요. 각 스레드가 오래 걸리는 요청에 응답하고 있으면, 뒤따르는 요청은 여전히 큐에서 밀리지만, 그 지점에 도달하기 전에 처리할 수 있는 오래 걸리는 요청의 수는 늘었어요.
이 기법은 웹 서버의 처리량을 개선하는 여러 방법 중 하나일 뿐이에요. 탐구해 볼 수 있는 다른 옵션으로는 fork/join 모델, 싱글 스레드 async I/O 모델, 멀티스레드 async I/O 모델이 있어요. 이 주제에 관심이 있다면 다른 해결책에 대해 더 읽어 보고 구현을 시도해 볼 수 있어요. Rust 같은 저수준 언어에서는 이 모든 옵션이 가능하답니다.
스레드 풀 구현을 시작하기 전에, 풀을 사용하는 것이 어떻게 생겨야 하는지 이야기해 볼게요. 코드를 설계하려고 할 때 클라이언트 인터페이스를 먼저 쓰는 것이 설계를 안내하는 데 도움이 될 수 있어요. 코드의 API를, 여러분이 호출하고 싶은 방식으로 구조화되게 작성하세요. 그다음 그 구조 안에서 기능을 구현하세요. 기능을 먼저 구현하고 나서 공개 API를 설계하지 말고요.
12장의 프로젝트에서 테스트 주도 개발(test-driven development)을 썼던 것과 비슷하게, 여기서는 컴파일러 주도 개발(compiler-driven development)을 쓸 거예요. 우리가 원하는 함수를 호출하는 코드를 작성한 다음, 컴파일러의 오류를 보고 코드가 동작하게 하려면 무엇을 바꿔야 할지 결정할 거예요. 하지만 그 전에, 우리가 출발점으로 쓰지 않을 기법을 먼저 탐구해 볼게요.
요청마다 스레드 스폰하기 (Spawning a Thread for Each Request)
먼저, 모든 연결에 새 스레드를 만들 경우 우리 코드가 어떻게 생길지 살펴볼게요. 앞서 언급했듯이 잠재적으로 무제한의 스레드를 스폰하는 문제 때문에 이것이 최종 계획은 아니지만, 먼저 동작하는 멀티스레드 서버를 얻기 위한 출발점이에요. 그다음 개선으로 스레드 풀을 추가할 텐데, 두 해결책을 대조하는 게 더 쉬워질 거예요.
Listing 21-11은 for 루프 안에서 각 스트림을 처리할 새 스레드를 스폰하도록 main에 가할 변경을 보여줘요.
use std::{
fs,
io::{BufReader, prelude::*},
net::{TcpListener, TcpStream},
thread,
time::Duration,
};
fn main() {
let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
for stream in listener.incoming() {
let stream = stream.unwrap();
thread::spawn(|| {
handle_connection(stream);
});
}
}
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();
}
16장에서 배웠듯이 thread::spawn은 새 스레드를 만들고 클로저의 코드를 새 스레드에서 실행해요. 이 코드를 실행하고 브라우저에서 _/sleep_을 로드한 다음 다른 브라우저 탭 두 개에서 _/_을 로드하면, / 요청이 _/sleep_이 끝나기를 기다릴 필요가 없다는 걸 확실히 볼 수 있어요. 하지만 앞서 언급했듯이, 제한 없이 새 스레드를 만들게 되므로 결국 시스템을 압도할 거예요.
17장에서 이것이 바로 async와 await가 진짜 빛을 발하는 종류의 상황이라는 걸 떠올릴 수도 있어요. 스레드 풀을 만들 때 그 점을 염두에 두고, async를 쓰면 사물이 어떻게 달라지거나 같을지 생각해 보세요.
유한한 수의 스레드 만들기 (Creating a Finite Number of Threads)
우리 스레드 풀이 비슷하고 익숙한 방식으로 동작하길 원해요. 그래야 스레드에서 스레드 풀로 전환할 때 우리 API를 쓰는 코드에 큰 변경이 필요 없거든요. Listing 21-12는 thread::spawn 대신 사용하고 싶은 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() {
let stream = stream.unwrap();
pool.execute(|| {
handle_connection(stream);
});
}
}
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();
}
ThreadPool::new로 설정 가능한 수의 스레드(이 경우 네 개)를 가진 새 스레드 풀을 만들어요. 그다음 for 루프에서 pool.execute는 각 스트림에 대해 풀이 실행해야 할 클로저를 받는다는 점에서 thread::spawn과 비슷한 인터페이스를 가져요. pool.execute가 클로저를 받아 풀의 스레드에 넘겨 실행하도록 구현해야 해요. 이 코드는 아직 컴파일되지 않지만, 시도해서 컴파일러가 우리를 어떻게 고칠지 안내하게 할 거예요.
컴파일러 주도 개발로 ThreadPool 만들기 (Building ThreadPool Using Compiler-Driven Development)
Listing 21-12의 변경을 _src/main.rs_에 적용한 다음, cargo check의 컴파일러 오류로 개발을 이끌어 볼게요. 여기 첫 번째로 얻는 오류가 있어요.
$ cargo check
Checking hello v0.1.0 (file:///projects/hello)
error[E0433]: failed to resolve: use of undeclared type `ThreadPool`
--> src/main.rs:11:16
|
11 | let pool = ThreadPool::new(4);
| ^^^^^^^^^^ use of undeclared type `ThreadPool`
For more information about this error, try `rustc --explain E0433`.
error: could not compile `hello` (bin "hello") due to 1 previous error
좋아요! 이 오류는 ThreadPool 타입이나 모듈이 필요하다고 알려줘요. 이제 하나 만들어 볼게요. ThreadPool 구현은 우리 웹 서버가 하는 작업의 종류와 독립적일 거예요. 그래서 hello 크레이트를 바이너리 크레이트에서 라이브러리 크레이트로 바꿔 ThreadPool 구현을 담을게요. 라이브러리 크레이트로 바꾸고 나면, 웹 요청을 서빙하는 것에만 국한하지 않고 스레드 풀을 사용하려는 어떤 작업이든 별도의 스레드 풀 라이브러리를 쓸 수도 있어요.
다음을 담은 src/lib.rs 파일을 만드세요. 이건 지금 가질 수 있는 가장 단순한 ThreadPool 구조체 정의예요.
pub struct ThreadPool;
그다음 라이브러리 크레이트에서 ThreadPool을 스코프로 가져오기 위해 main.rs 파일을 편집하세요. src/main.rs 맨 위에 다음 코드를 추가하면 돼요.
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() {
let stream = stream.unwrap();
pool.execute(|| {
handle_connection(stream);
});
}
}
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();
}
이 코드는 여전히 동작하지 않지만, 다시 확인해서 다음에 다뤄야 할 오류를 얻어 볼게요.
$ cargo check
Checking hello v0.1.0 (file:///projects/hello)
error[E0599]: no function or associated item named `new` found for struct `ThreadPool` in the current scope
--> src/main.rs:12:28
|
12 | let pool = ThreadPool::new(4);
| ^^^ function or associated item not found in `ThreadPool`
For more information about this error, try `rustc --explain E0599`.
error: could not compile `hello` (bin "hello") due to 1 previous error
이 오류는 ThreadPool에 new라는 연관 함수를 만들어야 한다는 뜻이에요. 그리고 new는 4를 인자로 받을 수 있는 파라미터 하나를 가져야 하고 ThreadPool 인스턴스를 반환해야 한다는 것도 알아요. 이런 특성을 가진 가장 단순한 new 함수를 구현해 볼게요.
pub struct ThreadPool;
impl ThreadPool {
pub fn new(size: usize) -> ThreadPool {
ThreadPool
}
}
size 파라미터의 타입으로 usize를 고른 이유는 음수의 스레드 수는 말이 안 되다는 걸 알고 있기 때문이에요. 또 이 4를 스레드 컬렉션의 요소 수로 쓸 거라는 것도 아는데, 3장의 "Integer Types" 절에서 다뤘듯이 그것이 바로 usize 타입의 용도예요.
코드를 다시 확인해 볼게요.
$ cargo check
Checking hello v0.1.0 (file:///projects/hello)
error[E0599]: no method named `execute` found for struct `ThreadPool` in the current scope
--> src/main.rs:17:14
|
17 | pool.execute(|| {
| -----^^^^^^^ method not found in `ThreadPool`
For more information about this error, try `rustc --explain E0599`.
error: could not compile `hello` (bin "hello") due to 1 previous error
이제 ThreadPool에 execute 메서드가 없어서 오류가 발생해요. 앞선 "Creating a Finite Number of Threads" 절에서 스레드 풀이 thread::spawn과 비슷한 인터페이스를 가져야 한다고 결정했던 걸 떠올려 보세요. 추가로 execute 함수를, 받은 클로저를 풀의 유휴 스레드에 넘겨 실행하도록 구현할 거예요.
ThreadPool에 execute 메서드를 정의해 클로저를 파라미터로 받게 할 거예요. 13장의 "Moving Captured Values Out of Closures"에서 세 가지 서로 다른 트레이트 Fn, FnMut, FnOnce로 클로저를 파라미터로 받을 수 있다는 걸 떠올려 보세요. 여기서 어떤 종류의 클로저를 쓸지 결정해야 해요. 결국 표준 라이브러리 thread::spawn 구현과 비슷한 일을 하게 될 거라는 걸 알므로, thread::spawn 시그니처가 파라미터에 어떤 바운드를 걸고 있는지 살펴볼 수 있어요. 문서가 다음을 보여줘요.
pub fn spawn<F, T>(f: F) -> JoinHandle<T>
where
F: FnOnce() -> T,
F: Send + 'static,
T: Send + 'static,
F 타입 파라미터가 여기서 우리가 관심을 갖는 것이에요. T 타입 파라미터는 반환 값과 관련되므로 신경 쓰지 않아요. spawn이 F에 FnOnce를 트레이트 바운드로 쓴다는 걸 볼 수 있어요. 우리도 아마 그걸 원할 거예요. 결국 execute에서 받은 인자를 spawn에 넘길 것이기 때문이에요. 또 요청을 실행하는 스레드가 그 요청의 클로저를 한 번만 실행할 것이라는 점이 FnOnce의 Once와 일치하므로, FnOnce가 우리가 쓰고 싶은 트레이트라는 확신이 더 들어요.
F 타입 파라미터는 또한 Send 트레이트 바운드와 'static 라이프타임 바운드를 가지는데, 이는 우리 상황에서 유용해요. 클로저를 한 스레드에서 다른 스레드로 옮기려면 Send가 필요하고, 스레드가 실행하는 데 얼마나 걸릴지 모르므로 'static이 필요해요. 이런 바운드를 가진 F 타입의 제네릭 파라미터를 받는 execute 메서드를 ThreadPool에 만들어 볼게요.
pub struct ThreadPool;
impl ThreadPool {
// --snip--
pub fn new(size: usize) -> ThreadPool {
ThreadPool
}
pub fn execute<F>(&self, f: F)
where
F: FnOnce() + Send + 'static,
{
}
}
FnOnce 뒤에 여전히 ()를 쓰는 이유는 이 FnOnce가 파라미터를 받지 않고 유닛 타입 ()을 반환하는 클로저를 나타내기 때문이에요. 함수 정의처럼 반환 타입은 시그니처에서 생략할 수 있지만, 파라미터가 없어도 괄호는 여전히 필요해요.
다시 말하지만 이것이 execute 메서드의 가장 단순한 구현이에요. 아무것도 하지 않지만, 우리는 코드가 컴파일되도록 만들려고만 하고 있어요. 다시 확인해 볼게요.
$ cargo check
Checking hello v0.1.0 (file:///projects/hello)
Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.24s
컴파일돼요! 하지만 cargo run을 시도하고 브라우저에서 요청을 만들면, 이번 장의 시작에서 봤던 오류를 브라우저에서 보게 될 거예요. 우리 라이브러리가 아직 execute에 전달된 클로저를 실제로 호출하고 있지 않거든요!
참고: Haskell이나 Rust 같은 엄격한 컴파일러를 가진 언어에 대해 들을 수 있는 말이 있어요. "코드가 컴파일되면 동작한다." 하지만 이 말이 보편적으로 참이지는 않아요. 우리 프로젝트는 컴파일되지만, 전혀 아무것도 하지 않아요! 실제로 완전한 프로젝트를 만들고 있다면, 지금이 코드가 컴파일되는 것 그리고 우리가 원하는 동작을 가지는지 확인하는 단위 테스트를 작성하기 시작할 좋은 시기일 거예요.
생각해 보세요. 여기서 클로저 대신 future를 실행한다면 무엇이 달라질까요?
new에서 스레드 수 검증하기 (Validating the Number of Threads in new)
우리는 new와 execute의 파라미터로 아무것도 하지 않고 있어요. 원하는 동작으로 이 함수들의 본문을 구현해 볼게요. 먼저 new에 대해 생각해 보죠. 앞서 음수의 스레드 수를 가진 풀은 말이 안 되므로 size 파라미터에 부호 없는 타입을 골랐어요. 하지만 스레드가 0개인 풀도 말이 안 되는데, 0은 완벽히 유효한 usize예요. ThreadPool 인스턴스를 반환하기 전에 size가 0보다 큰지 확인하는 코드를 추가하고, Listing 21-13처럼 assert! 매크로를 사용해 0을 받으면 프로그램이 패닉하게 만들 거예요.
pub struct ThreadPool;
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);
ThreadPool
}
// --snip--
pub fn execute<F>(&self, f: F)
where
F: FnOnce() + Send + 'static,
{
}
}
또한 문서 주석(doc comments)으로 ThreadPool에 대한 문서를 조금 추가했어요. 14장에서 다뤘듯이, 함수가 패닉할 수 있는 상황을 집어내는 섹션을 추가하는 좋은 문서 관행을 따랐다는 점을 주목하세요. cargo doc --open을 실행하고 ThreadPool 구조체를 클릭해 new에 대해 생성된 문서가 어떻게 보이는지 확인해 보세요!
여기서 한 것처럼 assert! 매크로를 추가하는 대신, new를 build로 바꾸고 12장의 I/O 프로젝트에서 Config::build로 했던 것처럼 Result를 반환할 수도 있어요. 하지만 이 경우 스레드가 하나도 없는 스레드 풀을 만들려는 것은 복구 불가능한 오류여야 한다고 결정했어요. 야심이 있다면, 다음 시그니처를 가진 build라는 함수를 작성해 new 함수와 비교해 보세요.
pub fn build(size: usize) -> Result<ThreadPool, PoolCreationError> {
스레드를 저장할 공간 만들기 (Creating Space to Store the Threads)
이제 풀에 저장할 유효한 스레드 수를 알 수 있는 방법이 생겼으니, 그 스레드들을 만들고 구조체를 반환하기 전에 ThreadPool 구조체에 저장할 수 있어요. 하지만 스레드를 어떻게 "저장"할까요? thread::spawn 시그니처를 다시 살펴볼게요.
pub fn spawn<F, T>(f: F) -> JoinHandle<T>
where
F: FnOnce() -> T,
F: Send + 'static,
T: Send + 'static,
spawn 함수는 JoinHandle<T>를 반환하는데, 여기서 T는 클로저가 반환하는 타입이에요. JoinHandle도 써 보고 어떻게 되는지 볼게요. 우리 경우 스레드 풀에 넘기는 클로저는 연결을 처리하고 아무것도 반환하지 않으므로, T는 유닛 타입 ()이 될 거예요.
Listing 21-14의 코드는 컴파일되지만 아직 스레드를 만들지는 않아요. ThreadPool의 정의를 thread::JoinHandle<()> 인스턴스의 벡터를 담도록 바꾸고, 벡터를 size 용량으로 초기화하고, 스레드를 만들 코드를 실행할 for 루프를 설정하고, 이들을 담은 ThreadPool 인스턴스를 반환했어요.
use std::thread;
pub struct ThreadPool {
threads: Vec<thread::JoinHandle<()>>,
}
impl ThreadPool {
// --snip--
/// 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 mut threads = Vec::with_capacity(size);
for _ in 0..size {
// create some threads and store them in the vector
}
ThreadPool { threads }
}
// --snip--
pub fn execute<F>(&self, f: F)
where
F: FnOnce() + Send + 'static,
{
}
}
ThreadPool의 벡터 항목의 타입으로 thread::JoinHandle을 쓰므로, 라이브러리 크레이트에 std::thread를 스코프로 가져왔어요.
유효한 크기를 받으면 우리 ThreadPool은 size개의 항목을 담을 수 있는 새 벡터를 만들어요. with_capacity 함수는 Vec::new와 같은 작업을 수행하지만 중요한 차이가 있어요. 벡터에 공간을 미리 할당(pre-allocate)한다는 거죠. 벡터에 size개의 요소를 저장해야 한다는 걸 알고 있으므로, 요소가 삽입될 때 스스로 크기를 조정하는 Vec::new를 쓰는 것보다 이 할당을 미리 하는 것이 약간 더 효율적이에요.
cargo check를 다시 실행하면 성공해야 해요.
ThreadPool에서 스레드로 코드 보내기 (Sending Code from the ThreadPool to a Thread)
Listing 21-14의 for 루프에 스레드 생성에 관한 주석을 남겨 두었어요. 여기서 실제로 스레드를 만드는 방법을 살펴볼게요. 표준 라이브러리는 스레드를 만드는 방법으로 thread::spawn을 제공하는데, thread::spawn은 스레드가 생성되는 즉시 실행해야 할 코드를 기대해요. 하지만 우리 경우에는 스레드를 만들고, 나중에 보낼 코드를 기다리게 하려고 해요. 표준 라이브러리의 스레드 구현에는 그렇게 할 수 있는 방법이 없어요. 우리가 직접 구현해야 하죠.
이 동작은 ThreadPool과 스레드 사이에, 이 새 동작을 관리할 새 데이터 구조를 도입해 구현할 거예요. 이 데이터 구조를 _Worker_라 부를 텐데, 풀링 구현에서 흔한 용어예요. Worker는 실행해야 할 코드를 집어 들어 그 코드를 자기 스레드에서 실행해요.
레스토랑 주방에서 일하는 사람들을 생각해 보세요. 주방 직원들은 고객의 주문이 들어올 때까지 기다렸다가, 그 주문을 받아 처리하는 책임을 져요.
스레드 풀에 JoinHandle<()> 인스턴스의 벡터를 저장하는 대신, Worker 구조체의 인스턴스를 저장할 거예요. 각 Worker는 단일 JoinHandle<()> 인스턴스를 저장해요. 그다음 Worker에, 실행할 코드의 클로저를 받아 이미 실행 중인 스레드로 보내 실행하는 메서드를 구현할 거예요. 또 각 Worker에 id를 줘서, 로깅이나 디버깅할 때 풀의 서로 다른 Worker 인스턴스를 구분할 수 있게 할 거예요.
ThreadPool을 만들 때 일어날 새 과정은 이러해요. Worker를 이렇게 설정한 뒤 클로저를 스레드로 보내는 코드를 구현할 거예요.
id와JoinHandle<()>를 담는Worker구조체 정의하기ThreadPool이Worker인스턴스의 벡터를 담도록 바꾸기id번호를 받아,id와 빈 클로저로 스폰된 스레드를 담는Worker인스턴스를 반환하는Worker::new함수 정의하기ThreadPool::new에서for루프 카운터로id를 생성하고, 그id로 새Worker를 만들어 벡터에 저장하기
도전하고 싶다면 Listing 21-15의 코드를 보기 전에 이 변경들을 직접 구현해 보세요.
준비됐나요? 앞선 수정을 하는 한 가지 방법이 담긴 Listing 21-15가 여기 있어요.
use std::thread;
pub struct ThreadPool {
workers: Vec<Worker>,
}
impl ThreadPool {
// --snip--
/// 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 mut workers = Vec::with_capacity(size);
for id in 0..size {
workers.push(Worker::new(id));
}
ThreadPool { workers }
}
// --snip--
pub fn execute<F>(&self, f: F)
where
F: FnOnce() + Send + 'static,
{
}
}
struct Worker {
id: usize,
thread: thread::JoinHandle<()>,
}
impl Worker {
fn new(id: usize) -> Worker {
let thread = thread::spawn(|| {});
Worker { id, thread }
}
}
ThreadPool의 필드 이름을 threads에서 workers로 바꿨는데, 이제 JoinHandle<()> 인스턴스 대신 Worker 인스턴스를 담고 있기 때문이에요. for 루프의 카운터를 Worker::new의 인자로 사용하고, 각 새 Worker를 workers라는 벡터에 저장해요.
외부 코드(우리 서버인 src/main.rs 같은)는 ThreadPool 안에서 Worker 구조체를 사용하는 구현 세부사항을 알 필요가 없으므로, Worker 구조체와 그 new 함수를 비공개(private)로 만들어요. Worker::new 함수는 우리가 준 id를 사용하고, 빈 클로저로 새 스레드를 스폰해 만든 JoinHandle<()> 인스턴스를 저장해요.
참고: 시스템 리소스가 충분하지 않아 운영 체제가 스레드를 만들 수 없다면
thread::spawn은 패닉할 거예요. 그러면 일부 스레드의 생성은 성공할지라도 우리 서버 전체가 패닉하게 되죠. 단순함을 위해 이 동작도 괜찮지만, 프로덕션 스레드 풀 구현에서는std::thread::Builder와 그spawn메서드(이것은Result를 반환해요)를 쓰는 편이 좋을 거예요.
이 코드는 컴파일되고 ThreadPool::new의 인자로 지정한 수의 Worker 인스턴스를 저장할 거예요. 하지만 execute에서 받는 클로저를 우리는 여전히 처리하지 않고 있어요. 이제 그 방법을 살펴볼게요.
채널로 스레드에 요청 보내기 (Sending Requests to Threads via Channels)
다음으로 다룰 문제는 thread::spawn에 주어지는 클로저가 전혀 아무것도 하지 않는다는 거예요. 현재 우리는 execute 메서드에서 실행하고 싶은 클로저를 받아요. 하지만 ThreadPool을 만드는 동안 각 Worker를 만들 때 thread::spawn에 실행할 클로저를 줘야 해요.
방금 만든 Worker 구조체들이 ThreadPool에 있는 큐에서 실행할 코드를 가져와 그 코드를 자기 스레드로 보내 실행하길 원해요.
16장에서 배운 채널—두 스레드 간에 통신하는 간단한 방법—이 이 사용 사례에 완벽할 거예요. 채널을 작업의 큐로 기능하게 하고, execute가 ThreadPool에서 Worker 인스턴스들로 작업을 보내면, Worker가 그 작업을 자기 스레드로 보낼 거예요. 계획은 이래요.
ThreadPool이 채널을 만들고 송신자(sender)를 붙잡아 둔다- 각
Worker가 수신자(receiver)를 붙잡아 둔다 - 채널로 보내고 싶은 클로저를 담을 새
Job구조체를 만든다 execute메서드가 실행하고 싶은 작업을 송신자로 보낸다Worker는 자기 스레드에서 수신자를 반복하고, 받은 어떤 작업의 클로저든 실행한다
Listing 21-16처럼 ThreadPool::new에서 채널을 만들고 송신자를 ThreadPool 인스턴스에 보관하는 것부터 시작할게요. Job 구조체는 지금 아무것도 담지 않지만, 채널로 보내는 항목의 타입이 될 거예요.
use std::{sync::mpsc, thread};
pub struct ThreadPool {
workers: Vec<Worker>,
sender: mpsc::Sender<Job>,
}
struct Job;
impl ThreadPool {
// --snip--
/// 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 mut workers = Vec::with_capacity(size);
for id in 0..size {
workers.push(Worker::new(id));
}
ThreadPool { workers, sender }
}
// --snip--
pub fn execute<F>(&self, f: F)
where
F: FnOnce() + Send + 'static,
{
}
}
struct Worker {
id: usize,
thread: thread::JoinHandle<()>,
}
impl Worker {
fn new(id: usize) -> Worker {
let thread = thread::spawn(|| {});
Worker { id, thread }
}
}
ThreadPool::new에서 새 채널을 만들고 풀이 송신자를 보관하게 해요. 이건 성공적으로 컴파일돼요.
스레드 풀이 채널을 만들 때 채널의 수신자를 각 Worker에 넘겨 보겠어요. Worker 인스턴스들이 스폰하는 스레드에서 수신자를 쓰고 싶다는 걸 알므로, 클로저 안에서 receiver 파라미터를 참조할 거예요. Listing 21-17의 코드는 아직 완전히 컴파일되지는 않아요.
use std::{sync::mpsc, thread};
pub struct ThreadPool {
workers: Vec<Worker>,
sender: mpsc::Sender<Job>,
}
struct Job;
impl ThreadPool {
// --snip--
/// 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 mut workers = Vec::with_capacity(size);
for id in 0..size {
workers.push(Worker::new(id, receiver));
}
ThreadPool { workers, sender }
}
// --snip--
pub fn execute<F>(&self, f: F)
where
F: FnOnce() + Send + 'static,
{
}
}
// --snip--
struct Worker {
id: usize,
thread: thread::JoinHandle<()>,
}
impl Worker {
fn new(id: usize, receiver: mpsc::Receiver<Job>) -> Worker {
let thread = thread::spawn(|| {
receiver;
});
Worker { id, thread }
}
}
작고 단순한 변경을 몇 가지 했어요. receiver를 Worker::new에 넘기고, 클로저 안에서 그것을 사용했죠.
이 코드를 확인하려 하면 이런 오류를 얻어요.
$ cargo check
Checking hello v0.1.0 (file:///projects/hello)
error[E0382]: use of moved value: `receiver`
--> src/lib.rs:26:42
|
21 | let (sender, receiver) = mpsc::channel();
| -------- move occurs because `receiver` has type `std::sync::mpsc::Receiver<Job>`, which does not implement the `Copy` trait
...
25 | for id in 0..size {
| ----------------- inside of this loop
26 | workers.push(Worker::new(id, receiver));
| ^^^^^^^^ value moved here, in previous iteration of loop
|
note: consider changing this parameter type in method `new` to borrow instead if owning the value isn't necessary
--> src/lib.rs:47:33
|
47 | fn new(id: usize, receiver: mpsc::Receiver<Job>) -> Worker {
| --- in this method ^^^^^^^^^^^^^^^^^^^ this parameter takes ownership of the value
help: consider moving the expression out of the loop so it is only moved once
|
25 ~ let mut value = Worker::new(id, receiver);
26 ~ for id in 0..size {
27 ~ workers.push(value);
|
For more information about this error, try `rustc --explain E0382`.
error: could not compile `hello` (lib) due to 1 previous error
코드가 receiver를 여러 Worker 인스턴스에 넘기려 하고 있어요. 16장에서 떠올리듯 이건 동작하지 않아요. Rust가 제공하는 채널 구현은 다중 생산자(producer), 단일 _소비자(consumer)_이기 때문이에요. 즉 이 코드를 고치려고 채널의 소비 끝을 그냥 복제할 수는 없어요. 또 여러 소비자에게 메시지를 여러 번 보내고 싶지도 않아요. 각 메시지가 한 번 처리되는, 여러 Worker 인스턴스가 있는 단일 메시지 목록을 원하죠.
추가로, 채널 큐에서 작업을 꺼내는 것은 receiver를 변경(mutate)하는 것을 포함하므로, 스레드들은 receiver를 안전하게 공유하고 수정할 방법이 필요해요. 그렇지 않으면 16장에서 다뤘던 것처럼 경합 조건(race condition)이 생길 수 있어요.
16장에서 다룬 스레드 안전 스마트 포인터를 떠올려 보세요. 여러 스레드에 걸쳐 소유권을 공유하고 스레드들이 값을 변경하게 하려면 Arc<Mutex<T>>를 써야 해요. Arc 타입은 여러 Worker 인스턴스가 수신자를 소유하게 하고, Mutex는 한 번에 한 Worker만 수신자에서 작업을 얻도록 보장해요. Listing 21-18이 우리가 해야 할 변경을 보여줘요.
use std::{
sync::{Arc, Mutex, mpsc},
thread,
};
// --snip--
pub struct ThreadPool {
workers: Vec<Worker>,
sender: mpsc::Sender<Job>,
}
struct Job;
impl ThreadPool {
// --snip--
/// 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 }
}
// --snip--
pub fn execute<F>(&self, f: F)
where
F: FnOnce() + Send + 'static,
{
}
}
// --snip--
struct Worker {
id: usize,
thread: thread::JoinHandle<()>,
}
impl Worker {
fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
// --snip--
let thread = thread::spawn(|| {
receiver;
});
Worker { id, thread }
}
}
ThreadPool::new에서 수신자를 Arc와 Mutex에 넣어요. 각 새 Worker마다 Arc를 복제해 참조 카운트를 올려, Worker 인스턴스들이 수신자의 소유권을 공유하게 해요.
이 변경들로 코드가 컴파일돼요! 거의 다 왔어요!
execute 메서드 구현하기 (Implementing the execute Method)
드디어 ThreadPool에 execute 메서드를 구현해 볼게요. 또 Job을 구조체에서, execute가 받는 클로저의 타입을 담는 트레이트 객체의 타입 별칭으로 바꿀 거예요. 20장의 "Type Synonyms and Type Aliases" 절에서 다뤘듯이, 타입 별칭은 긴 타입을 쓰기 쉽게 짧게 만들어줘요. Listing 21-19를 보세요.
use std::{
sync::{Arc, Mutex, mpsc},
thread,
};
pub struct ThreadPool {
workers: Vec<Worker>,
sender: mpsc::Sender<Job>,
}
// --snip--
type Job = Box<dyn FnOnce() + Send + 'static>;
impl ThreadPool {
// --snip--
/// 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 }
}
pub fn execute<F>(&self, f: F)
where
F: FnOnce() + Send + 'static,
{
let job = Box::new(f);
self.sender.send(job).unwrap();
}
}
// --snip--
struct Worker {
id: usize,
thread: thread::JoinHandle<()>,
}
impl Worker {
fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
let thread = thread::spawn(|| {
receiver;
});
Worker { id, thread }
}
}
execute에서 받은 클로저로 새 Job 인스턴스를 만든 뒤, 그 작업을 채널의 송신 끝으로 보내요. send에 unwrap을 호출하는 것은 전송이 실패하는 경우 때문이에요. 이는 예를 들어 실행하는 스레드를 모두 멈추면 수신 끝이 새 메시지 수신을 멈추게 되어 생길 수 있어요. 지금은 우리 스레드가 실행을 멈추게 할 수 없어요. 스레드들은 풀이 존재하는 한 계속 실행되거든요. unwrap을 쓰는 이유는 실패 경우가 일어나지 않을 거라는 걸 알지만 컴파일러는 모르기 때문이에요.
하지만 아직 끝나지 않았어요! Worker에서 thread::spawn에 넘기는 클로저는 여전히 채널의 수신 끝을 _참조_만 하고 있어요. 대신 클로저가 영원히 반복하면서 채널 수신 끝에 작업을 요청하고, 작업을 받으면 실행하도록 해야 해요. Listing 21-20에 나온 변경을 Worker::new에 적용해 볼게요.
use std::{
sync::{Arc, Mutex, mpsc},
thread,
};
pub struct ThreadPool {
workers: Vec<Worker>,
sender: 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 }
}
pub fn execute<F>(&self, f: F)
where
F: FnOnce() + Send + 'static,
{
let job = Box::new(f);
self.sender.send(job).unwrap();
}
}
struct Worker {
id: usize,
thread: thread::JoinHandle<()>,
}
// --snip--
impl Worker {
fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
let thread = thread::spawn(move || {
loop {
let job = receiver.lock().unwrap().recv().unwrap();
println!("Worker {id} got a job; executing.");
job();
}
});
Worker { id, thread }
}
}
여기서 먼저 receiver에 lock을 호출해 뮤텍스를 획득하고, 어떤 오류에서는 패닉하도록 unwrap을 호출해요. 뮤텍스가 poisoned 상태에 있으면 잠금 획득이 실패할 수 있는데, 다른 스레드가 잠금을 해제하기보다 잠금을 쥔 채로 패닉했을 때 생길 수 있어요. 이 상황에서 unwrap을 호출해 이 스레드가 패닉하게 하는 게 올바른 조치예요. 이 unwrap을, 여러분에게 의미 있는 오류 메시지를 가진 expect로 바꿔도 좋아요.
뮤텍스에 대한 잠금을 얻으면 recv를 호출해 채널에서 Job을 받아요. 마지막 unwrap도 여기서 어떤 오류든 넘어가게 하는데, send 메서드가 수신자가 꺼지면 Err를 반환하는 것과 비슷하게, 송신자를 쥔 스레드가 종료되면 오류가 발생할 수 있어요.
recv 호출은 블로킹이므로, 아직 작업이 없으면 현재 스레드는 작업이 생길 때까지 기다려요. Mutex<T>는 한 번에 한 Worker 스레드만 작업을 요청하려 하도록 보장해요.
이제 스레드 풀이 동작하는 상태가 됐어요! cargo run을 해서 요청 몇 개를 만들어 보세요.
$ cargo run
Compiling hello v0.1.0 (file:///projects/hello)
warning: field `workers` is never read
--> src/lib.rs:7:5
|
6 | pub struct ThreadPool {
| ---------- field in this struct
7 | workers: Vec<Worker>,
| ^^^^^^^
|
= note: `#[warn(dead_code)]` on by default
warning: fields `id` and `thread` are never read
--> src/lib.rs:48:5
|
47 | struct Worker {
| ------ fields in this struct
48 | id: usize,
| ^^
49 | thread: thread::JoinHandle<()>,
| ^^^^^^
warning: `hello` (lib) generated 2 warnings
Finished `dev` profile [unoptimized + debuginfo] target(s) in 4.91s
Running `target/debug/hello`
Worker 0 got a job; executing.
Worker 2 got a job; executing.
Worker 1 got a job; executing.
Worker 3 got a job; executing.
Worker 0 got a job; executing.
Worker 2 got a job; executing.
Worker 1 got a job; executing.
Worker 3 got a job; executing.
Worker 0 got a job; executing.
Worker 2 got a job; executing.
성공이에요! 이제 연결을 비동기적으로 실행하는 스레드 풀이 생겼어요. 만들어지는 스레드는 네 개를 넘지 않으므로, 서버가 많은 요청을 받아도 우리 시스템이 과부하되지 않아요. /sleep 에 요청을 만들면 서버는 다른 스레드가 그 요청을 실행하게 해서 다른 요청도 처리할 수 있어요.
참고: 여러 브라우저 창에서 동시에 _/sleep_을 열면, 5초 간격으로 한 번에 하나씩 로드될 수도 있어요. 일부 웹 브라우저는 캐싱 이유로 같은 요청의 여러 인스턴스를 순차적으로 실행해요. 이 제한은 우리 웹 서버가 만든 게 아니에요.
이제 멈춰서 Listing 21-18, 21-19, 21-20의 코드가, 해야 할 작업에 클로저 대신 future를 쓴다면 어떻게 달라질지 생각하기 좋은 때예요. 어떤 타입이 바뀔까요? 메서드 시그니처는 어떻게 달라질까요(달라진다면)? 코드의 어느 부분이 그대로 남을까요?
17장과 19장에서 while let 루프를 배운 뒤, 왜 Worker 스레드 코드를 Listing 21-21처럼 쓰지 않았는지 궁금할 수 있어요.
use std::{
sync::{Arc, Mutex, mpsc},
thread,
};
pub struct ThreadPool {
workers: Vec<Worker>,
sender: 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 }
}
pub fn execute<F>(&self, f: F)
where
F: FnOnce() + Send + 'static,
{
let job = Box::new(f);
self.sender.send(job).unwrap();
}
}
struct Worker {
id: usize,
thread: thread::JoinHandle<()>,
}
// --snip--
impl Worker {
fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
let thread = thread::spawn(move || {
while let Ok(job) = receiver.lock().unwrap().recv() {
println!("Worker {id} got a job; executing.");
job();
}
});
Worker { id, thread }
}
}
이 코드는 컴파일되고 실행되지만, 원하는 스레딩 동작을 내지 못해요. 느린 요청이 여전히 다른 요청들이 처리를 기다리게 만들 거예요. 그 이유는 다소 미묘해요. Mutex 구조체에는 공개 unlock 메서드가 없는데, 잠금의 소유권이 lock 메서드가 반환하는 LockResult<MutexGuard<T>> 안의 MutexGuard<T>의 라이프타임에 기반하기 때문이에요. 컴파일 시점에 대여 검사기는 Mutex로 보호되는 리소스는 잠금을 쥐고 있지 않으면 접근할 수 없다는 규칙을 강제할 수 있어요. 하지만 이 구현은 MutexGuard<T>의 라이프타임을 유의하지 않으면 잠금을 의도보다 더 오래 쥘 수도 있어요.
let job = receiver.lock().unwrap().recv().unwrap();을 쓰는 Listing 21-20의 코드가 동작하는 이유는, let을 쓰면 등호 오른쪽 표현식에 사용된 임시 값들이 let 문이 끝날 때 즉시 드롭되기 때문이에요. 하지만 while let(그리고 if let과 match)은 연관된 블록이 끝날 때까지 임시 값을 드롭하지 않아요. Listing 21-21에서 잠금은 job() 호출 기간 동안 유지되는데, 이는 다른 Worker 인스턴스들이 작업을 받을 수 없게 만들어요.