채널과 execute 구현
채널로 작업을 Worker에 전달하고 execute 메서드로 요청을 병렬 실행합니다.
채널을 통해 스레드에 요청 보내기
다음으로 다룰 문제는 thread::spawn에 주어진 클로저가 아무 일도 하지 않는다는 것입니다.
현재는 execute 메서드에서 실행하고자 하는 클로저를 얻습니다.
그러나 ThreadPool의 생성 중에 각 Worker를 생성할 때 실행할 클로저를 thread::spawn에 제공해야 합니다.
앞에서 만들어 둔 Worker 구조체가 ThreadPool에 보관된 대기열에서 실행할
코드를 가져온 다음 그 코드를 자신의 스레드로 전송하여 실행하기를 원합니다.
16장에서 배운 채널, 즉 두 개의 스레드 간에 통신하는 간단한 방법은 지금의 사용 사례에 완벽히 들어맞을 것입니다.
채널을 사용하여 작업의 대기열로 작동하도록 하고, execute가 ThreadPool에서 Worker 인스턴스로 작업을 보내면, 이 인스턴스는 자신의 스레드로 작업을 보내게 됩니다.
계획은 다음과 같습니다.
ThreadPool은 채널을 생성하고 송신자를 대기시킵니다.- 각
Worker는 수신자를 보관합니다. - 채널을 통해 보내려는 클로저를 가진 새로운 구조체
Job을 만듭니다. execute메서드는 송신자를 통하여 실행하려는 작업을 보냅니다.Worker는 자신의 스레드에서 수신자에 대한 반복을 수행하고 자신이 받은 작업의 클로저를 실행합니다.
예제 19-16에 나온 것처럼 ThreadPool::new에 채널을 생성하고 ThreadPool 인스턴스가 송신자를 갖도록 하는 것으로 시작하겠습니다.
지금은 Job 구조체에 아무것도 없지만 이것이 채널을 통해 전송될 아이템 타입이 될 것입니다.
Job 인스턴스를 보내는 채널의
송신자를 저장하도록 ThreadPool 수정하기
use std::{sync::mpsc, thread};
pub struct ThreadPool {
workers: Vec<Worker>,
sender: mpsc::Sender<Job>,
}
struct Job;
impl ThreadPool {
// --생략--
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 }
}
// --생략--
}
ThreadPool::new에서 새 채널을 생성하고 풀이 송신자를 보유하도록 합니다.
이 코드는 성공적으로 컴파일됩니다.
스레드 풀이 채널을 생성할 때 채널의 수신자를 각 워커에 전달해 봅시다.
워커가 생성하는 스레드에서 수신자를 사용하고자 함을 알고 있으므로, 클로저에서 receiver 매개변수를 참조하겠습니다.
예제 19-17의 코드는 아직 컴파일되지 않습니다.
예제 19-17: 워커에게 수신자 넘기기impl ThreadPool {
// --생략--
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 }
}
// --생략--
}
// --생략--
impl Worker {
fn new(id: usize, receiver: mpsc::Receiver<Job>) -> Worker {
let thread = thread::spawn(|| {
receiver;
});
Worker { id, thread }
}
}
약간의 직관적인 변경 사항을 적용했습니다.
수신자를 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
...
26 | workers.push(Worker::new(id, receiver));
| ^^^^^^^^ value moved here, in previous iteration of loop
For more information about this error, try `rustc --explain E0382`.
error: could not compile `hello` due to previous error
이 코드는 여러 개의 Worker 인스턴스에게 receiver를 전달하는 시도를 하고 있습니다.
16장에서 상기하실 수 있듯, 이는 동작하지 않습니다.
러스트가 제공하는 채널 구현체는 여러 개의 생산자, 하나의 소비자입니다.
즉, 이 코드를 수정하기 위해 채널의 소비자 쪽만 복제할 수는 없습니다.
또한 여러 소비자에게 메시지를 여러 번 보내고 싶지도 않습니다; 각 메시지가 한 번씩 처리되도록 여러 워커가 있는 하나의 메시지 리스트가 되기를 원합니다.
또한, 채널 대기열에서 작업을 빼내려면 receiver를 변경해야 하므로,
스레드가 receiver를 안전하게 공유하고 수정할 수 있는 방법이 필요합니다;
그렇지 않으면(16장에서 다룬 것처럼) 데이터 경합이 발생할 수 있습니다.
16장에서 설명한 스레드 안전 스마트 포인터를 상기해봅시다.
여러 스레드에서 소유권을 공유하고 스레드가 값을 변경할 수 있도록 하려면, Arc<Mutex<T>>를 사용할 필요가 있습니다.
Arc 타입은 여러 워커가 수신자를 소유할 수 있도록 하고, Mutex는 한 번에 한 워커만 수신자로부터 작업을 가져올 수 있도록 합니다.
예제 19-18은 변경해야 할 사항을 보여줍니다.
예제 19-18:Arc와 Mutex를 사용하여
여러 워커 간에 수신자 공유하기
use std::{
sync::{mpsc, Arc, Mutex},
thread,
};
// --생략--
impl ThreadPool {
// --생략--
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 }
}
// --생략--
}
// --생략--
impl Worker {
fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
// --생략--
}
}
ThreadPool::new에서 수신자를 Arc와 Mutex에 넣습니다.
각각의 새 워커에 대해 Arc를 복제하여 참조 카운트를 늘려 워커가 수신자의 소유권을 공유할 수 있도록 합니다.
이 변경 사항을 적용하면 코드가 컴파일됩니다!
이제 끝이 보입니다!
execute 메서드 구현하기
마지막으로 ThreadPool에서 execute 메서드를 구현해봅시다.
또한 Job을 구조체가 아니라 execute가 수신하는 클로저 타입을 갖는 트레이트 객체의 타입 별칭으로 변경하겠습니다.
18장의 ‘타입 별칭으로 타입의 동의어 만들기’ 절에서 설명한 것처럼, 타입 별칭을 사용하면 긴 타입을 사용하기 쉽도록 짧게 만들 수 있습니다.
예제 19-19를 봅시다.
예제 19-19: 각 클로저를 담는Box에 대한 Job
타입 별칭을 만들어서 이 작업을 채널로 보내기
// --생략--
type Job = Box<dyn FnOnce() + Send + 'static>;
impl ThreadPool {
// --생략--
pub fn execute<F>(&self, f: F)
where
F: FnOnce() + Send + 'static,
{
let job = Box::new(f);
self.sender.send(job).unwrap();
}
}
// --생략--
execute에서 얻은 클로저를 사용하여 새 Job 인스턴스를 생성한 후, 해당 작업을 채널 단말로 보냅니다.
전송이 실패할 경우를 위하여 send에서 unwrap을 호출하고 있습니다.
이를테면 모든 스레드의 실행이 중지되어 수신 측에서 새 메시지의 수신을 중단한 경우에는 전송이 실패할 수 있습니다.
현재로서는 스레드 실행을 중지할 수 없습니다.
풀이 존재하는 한 스레드는 계속 실행됩니다.
여기서 unwrap을 사용하는 이유는 실패 사례가 발생하지 않을 것이라는 것을 우리는 알고 있지만, 컴파일러는 이를 알지 못하기 때문입니다.
하지만 아직 끝나지 않았습니다!
워커에서 thread::spawn으로 전달된 클로저는 여전히 채널의 수신 단말을 참조만 하고 있습니다.
그 대신 클로저가 영원히 반복되도록 하여, 채널의 수신 단말에 작업을 요청하고 작업을 받으면 해당 작업을 실행하도록 할 필요가 있습니다.
예제 19-20에 나온 변경 사항을 Worker::new에 적용해봅시다.
// --생략--
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 is never read: `workers`
--> src/lib.rs:7:5
|
7 | workers: Vec<Worker>,
| ^^^^^^^^^^^^^^^^^^^^
|
= note: `#[warn(dead_code)]` on by default
warning: field is never read: `id`
--> src/lib.rs:48:5
|
48 | id: usize,
| ^^^^^^^^^
warning: field is never read: `thread`
--> src/lib.rs:49:5
|
49 | thread: thread::JoinHandle<()>,
| ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
warning: `hello` (lib) generated 3 warnings
Finished dev [unoptimized + debuginfo] target(s) in 1.40s
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.성공이군요!
이제 연결을 비동기적으로 실행하는 스레드 풀이 생겼습니다.
스레드가 4개 이상 생성되지 않으므로 서버가 많은 요청을 받더라도 시스템에 과부하가 발생하지 않습니다.
/sleep 요청을 보내면, 서버는 다른 요청에 대해 다른 스레드에서 이를 처리하는 방식으로 요청을 처리할 수 있습니다.
Note: 여러 브라우저 창에서 /sleep을 동시에 열면, 5초 간격으로 한 번에 하나씩 로드될 수도 있습니다. 몇몇 웹 브라우저는 캐싱을 위해 동일한 요청의 여러 인스턴스를 순차적으로 실행합니다. 이 제한 사항은 우리의 웹 서버에 의한 것이 아닙니다.
18장에서 while let 루프에 대해 배운 후이니, 예제 19-21에 나온 것처럼
워커 스레드 코드를 작성하지 않은 이유가 궁금할 수도 있습니다.
while let을 사용한 Worker::new의
대체 구현
// --생략--
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>의 수명을 염두에 두지 않은 경우 락이 의도한 것보다 더 오래 유지될 수 있습니다.
예제 19-20의 let job = receiver.lock().unwrap().recv().unwrap();을 사용하는 코드는 작동하는데, let을 사용하면 등호 기호 오른쪽의 표현식에 사용된 모든 임시 값이 let 문이 끝날 때 즉시 버려지기 때문입니다.
그러나 while let(그리고 if let과 match)은 연관된 블록이 끝날 때까지 임시 값을 버리지 않습니다.
예제 19-21에서는 job()을 호출하는 동안 락이 유지되므로 다른 워커가 작업을 받을 수 없습니다.
다음 다이어그램은 스레드 풀이 느린 요청을 여러 워커로 나누되 Receiver 락 범위를 조심해야 하는 이유를 보여줍니다.
싱글스레드 서버를 멀티스레드 서버로 바꾸기 학습 판단 기준표
싱글스레드 서버를 멀티스레드 서버로 바꾸기 마지막 점검은 실제 호출 모양과 실패 처리 경계가 자연스럽게 이어지는지에 맞춥니다.