ThreadPool과 Worker 구현
컴파일러 오류를 따라 ThreadPool과 Worker의 소유권 및 작업 스레드 구조를 구현합니다.
컴파일러 주도 개발을 사용하여 ThreadPool 구현하기
예제 19-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` due to previous error
훌륭합니다!
이 에러는 ThreadPool 타입 또는 모듈이 필요하다는 것을 알려주므로, 지금 구축해보겠습니다.
ThreadPool 구현은 웹 서버가 수행하는 작업의 종류와 무관합니다.
따라서, hello 크레이트를 바이너리 크레이트에서 라이브러리 크레이트로 전환하여 ThreadPool 구현을 담아봅시다.
라이브러리 크레이트로 변경한 후에는 웹 요청을 처리하는 것뿐만 아니라, 스레드 풀을 사용하여 수행하려는 어떤 작업에 대해서라도 분리된 스레드 풀 라이브러리를 사용할 수 있습니다.
현재 우리가 가질 수 있는 가장 간단한 ThreadPool 구조체의 정의에
해당하는 다음의 내용이 포함된 src/lib.rs를 생성하세요.
pub struct ThreadPool;
그런 다음 src/main.rs의 상단에 아래 코드를 추가하도록 main.rs 파일을 수정하여
라이브러리 크레이트에서 ThreadPool을 스코프에 가져오도록 합니다.
use hello::ThreadPool;
이 코드는 여전히 작동하지 않겠지만, 다음으로 해결해야 할 에러를 확인해봅시다.
$ 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` due to previous error
이 에러는 다음으로 ThreadPool에 대해 new라는 이름의 연관 함수를 만들어야 함을 나타냅니다.
또한 new에는 4를 인수로 받을 수 있는 하나의 매개변수가 있어야 하며 ThreadPool 인스턴스를 반환해야 한다는 것을 알고 있습니다.
이러한 특성을 갖는 가장 간단한 new 함수를 구현해 봅시다.
pub struct ThreadPool;
impl ThreadPool {
pub fn new(size: usize) -> ThreadPool {
ThreadPool
}
}
size 매개변수의 타입으로 usize를 선택한 이유는 음수 개수의 스레드가 의미 없음을 알고 있기 때문입니다.
또한 2장의 ‘정수형’절에서 설명한 것처럼, 이 4를 스레드 컬렉션의 요소 개수로 사용할 것임을 알고 있으며, 이것이 바로 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` due to previous error
이번에는 ThreadPool에 execute 메서드가 없어서 에러가 발생합니다.
‘유한한 개수의 스레드 생성하기’절에서 스레드 풀이 thread::spawn과 유사한 인터페이스를 가져야 한다고 결정했던 것을 상기하세요.
또한, 주어진 클로저를 받아 풀의 유휴 스레드에 전달하여 실행되도록 execute 함수를 구현하겠습니다.
클로저를 매개변수로 받도록 ThreadPool에 execute 메서드를 정의하겠습니다.
12장의 ‘캡처된 값을 클로저 밖으로 이동하기와 Fn 트레이트’절에서 클로저를 매개변수로 받기 위해 세 가지 트레이트를 사용할 수 있다고 한 것을 상기하세요.
바로 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가 여기서 사용하고자 하는 트레이트라는 것을 더욱 확신할 수 있으며, 이는 FnOnce의 Once와도 일치합니다.
F 타입 매개변수에는 또한 트레이트 바운드 Send와 라이프타임 바운드 'static이 있는데, 이는 지금의 상황에서 유용합니다.
한 스레드에서 다른 스레드로 클로저를 전송하기 위해서는 Send가 필요하고, 스레드가 실행되는 데 얼마나 오래 걸릴지 모르므로 'static이 필요합니다.
이러한 바운드를 사용하여 ThreadPool에 F 타입의 일반 매개변수를 받는 execute 메서드를 만들어 보겠습니다.
impl 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 [unoptimized + debuginfo] target(s) in 0.24s
컴파일 되는군요!
하지만 cargo run을 실행한 다음 브라우저에서 요청을 날리면, 이 장의 시작 부분에서 보았던 에러가 브라우저에 표시될 것입니다.
우리 라이브러리는 아직 실제로 execute로 전달된 클로저를 호출하지 않았거든요!
Note: 하스켈이나 러스트와 같은 엄격한 컴파일러가 있는 언어에 대해서 ‘코드가 컴파일되면 작동한다’는 말을 흔히 들을 수 있습니다. 하지만 이 말이 보편적인 사실은 아닙니다. 우리 프로젝트는 컴파일은 되지만, 아무것도 하지 않습니다! 실제 완전한 프로젝트를 구축하고 있다면, 코드가 컴파일되고 그러면서도 원하는 동작을 하는지 확인하기 위해서 유닛 테스트의 작성을 시작하기 좋은 시점입니다.
new에서 스레드 개수 검증하기
아직 매개변수 new와 execute로 아무것도 하지 않고 있습니다.
이제 원하는 동작이 되도록 이 함수들의 본문을 구현해봅시다.
먼저 new에 대해 생각해봅시다.
앞서 size 매개변수에 부호 없는 타입을 선택했는데, 그 이유는 스레드 수가 음수인 풀은 의미가 없기 때문입니다.
그러나 스레드 수가 0인 풀도 의미가 없지만, 0은 완벽하게 유효한 usize입니다.
예제 19-13에 나온 것처럼 ThreadPool 인스턴스를 반환하기 전에 size가 0보다 큰지 확인하는 코드를 추가하고, assert! 매크로를 사용하여 0을 수신하면 프로그램이 패닉 상태에 빠지도록 하겠습니다.
size가 0이면 패닉을 일으키도록
ThreadPool::new 구현하기
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
}
// --생략--
}
문서화 주석을 써서 ThreadPool에 대한 문서도 약간 추가했습니다.
14장에서 설명한 것처럼 함수가 패닉에 빠질 수 있는 상황을 설명하는 구절을 추가하여 좋은 문서화 관행을 따랐음을 주목하세요.
cargo doc --open을 실행하고 ThreadPool 구조체를 클릭하여 new에 대해 생성된 문서가 어떻게 보이는지 확인해보세요!
여기처럼 assert! 매크로를 추가하는 대신, 예제 11-9의 I/O 프로젝트에서 Config::build를 구현할 때처럼 new를 build로 변경하고 Result를 반환할 수도 있습니다.
하지만 지금의 경우에는 스레드 없이 스레드 풀을 생성하려고 하면 복구할 수 없는 에러가 발생한다고 결정했습니다.
도전할 마음이 있다면 다음과 같은 시그니처를 가진 build라는 이름의 함수를 작성해서 new 함수와 비교해보세요.
pub fn build(size: usize) -> Result<ThreadPool, PoolCreationError> {스레드를 저장할 공간 만들기
이제 풀에 저장할 스레드의 유효한 개수가 입력된 것을 알 방법이 생겼으므로, 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는 유닛 타입 ()가 됩니다.
예제 19-14의 코드는 컴파일되지만 아직 스레드를 생성하지 않습니다.
ThreadPool의 정의를 변경하여 thread::JoinHandle<()> 인스턴스의 벡터를 보관하고, size 용량으로 벡터를 초기화하고, 스레드를 생성하기 위해 어떤 코드를 실행하는 for 루프를 설정한 다음, 이들을 담고 있는 ThreadPool 인스턴스를 반환했습니다.
ThreadPool가 스레드를 담아둘
벡터 생성하기
use std::thread;
pub struct ThreadPool {
threads: Vec<thread::JoinHandle<()>>,
}
impl ThreadPool {
// --생략--
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 }
}
// --생략--
}
ThreadPool 내 벡터의 아이템 타입으로 thread::JoinHandle을
사용하기 때문에, 라이브러리 크레이트로부터 std::thread를 스코프로
가져왔습니다.
일단 유효한 크기를 받으면 ThreadPool은 아이템을 size만큼 담을 수 있는 새 벡터를 생성합니다.
with_capacity 함수는 Vec::new와 동일한 작업을 수행하지만 중요한 차이점이 있습니다.
벡터에 공간을 미리 할당한다는 것입니다.
벡터에 size 개의 요소들을 저장해야 한다는 것을 알고 있기 때문에, 요소가 삽입될 때 크기가 조정되는 Vec::new를 사용하는 것보다는, 이렇게 할당을 미리 수행하는 것이 약간 더 효율적입니다.
cargo check를 다시 실행해 보면 성공할 것입니다.
ThreadPool에서 스레드로 코드의 전송을 담당하는 Worker 구조체
예제 19-14의 for 루프에 스레드 생성에 관한 주석을 남겼습니다.
여기서는 실제로 스레드를 생성하는 방법을 살펴보겠습니다.
표준 라이브러리는 스레드를 생성하는 방법으로 thread::spawn을 제공하며, thread::spawn은 스레드가 생성되는 즉시 스레드가 실행해야 하는 코드를 가져올 것으로 예상합니다.
그러나 지금의 경우는 스레드를 생성한 후 나중에 전송될 코드를 대기하도록 하고 싶습니다.
표준 라이브러리의 스레드 구현에는 이를 수행하는 방법이 포함되어 있지 않습니다; 우리가 수동으로 구현해야 합니다.
ThreadPool과 스레드 사이에 이러한 새로운 동작을 관리하게 될 새로운 데이터 구조를 도입하여 이 동작을 구현하겠습니다.
이 데이터 구조를 워커(worker) 라고 부를 건데, 이는 풀링 구현에서 일반적으로 사용되는 용어입니다.
워커는 실행해야 하는 코드를 집어 들어서 이 코드를 워커의 스레드에서 실행합니다.
식당의 주방에서 일하는 사람들을 생각해보세요.
워커는 고객으로부터 주문이 들어올 때까지 기다렸다가, 주문을 받고 주문을 이행하는 일을 담당합니다.
스레드 풀에 JoinHandle<()> 인스턴스의 벡터를 저장하는 대신, Worker 구조체의 인스턴스를 저장하겠습니다.
각 Worker는 하나의 JoinHandle<()> 인스턴스를 저장하게 될 겁니다.
그런 다음 실행할 코드의 클로저를 가져와서 이미 실행 중인 스레드로 전송하여 이를 실행하는 메서드를 Worker에 구현하겠습니다.
또한 각 워커에 id를 부여하여 로깅이나 디버깅을 할 때 풀의 워커들을 서로 구별할 수 있도록 하겠습니다.
다음은 ThreadPool을 생성할 때 일어날 새로운 과정입니다.
이런 식으로 Worker를 설정한 다음 클로저를 스레드로 보내는 코드를 구현하겠습니다.
id와JoinHandle<()>를 가지고 있는Worker구조체를 정의합니다.ThreadPool이Worker인스턴스의 벡터를 갖도록 변경합니다.id숫자를 받아서id와 빈 클로저로 생성된 스레드를 가진Worker인스턴스를 반환하는Worker::new함수를 정의합니다.ThreadPool::new에서for루프 카운터를 사용하여id를 생성하고, 해당id로 새Worker를 생성한 다음 벡터에 워커를 저장합니다.
도전해 보고 싶다면 예제 19-15의 코드를 살펴보기 전에 이러한 변경 사항을 직접 구현해보세요.
준비됐나요?
여기 예제 19-15가 앞서 설명한 변경 사항을 구현한 방법 중 하나를 보여줍니다.
예제 19-15: 스레드를 직접 가지는 대신Worker
인스턴스를 가지도록 ThreadPool 수정하기
use std::thread;
pub struct ThreadPool {
workers: Vec<Worker>,
}
impl ThreadPool {
// --생략--
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 }
}
// --생략--
}
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 함수를 비공개로 설정합니다.
Worker::new 함수는 우리가 제공한 id를 써서 빈 클로저를 사용하는 새 스레드를 생성하는 방식으로 만들어진 JoinHandle<()> 인스턴스를 저장합니다.
Note: 시스템 리소스가 충분하지 않아 운영체제가 스레드를 생성할 수 없는 경우
thread::spawn은 패닉을 일으키게 됩니다. 그러면 스레드 생성이 일부 성공하더라도 전체 서버가 패닉에 빠지게 됩니다. 단순화를 위해서라면 이 동작은 괜찮지만, 프로덕션에서의 스레드 풀 구현이라면 패닉 대신Result를 반환하는std::thread::Builder와 여기서 제공하는spawn메서드를 사용하고 싶을 수도 있습니다.
이 코드는 컴파일되고 우리가 ThreadPool::new에 인수로 지정한 개수만큼 Worker 인스턴스를 저장합니다.
하지만 execute에서 얻은 클로저는 여전히 처리하지 않고 있습니다.
다음에는 이를 처리하는 방법을 살펴보겠습니다.