채널과 execute 구현
채널로 작업을 Worker에 전달하고 execute 메서드로 요청을 병렬 실행합니다.
ThreadPool과 Worker 구현에서 Worker를 담은 ThreadPool을 만들었지만, Worker의 스레드는 빈 클로저를 실행할 뿐이고 execute도 받은 클로저로 아무것도 하지 않습니다. 이 글에서는 채널로 작업을 Worker에 보내고, execute를 완성해 서버가 요청을 동시에 처리하게 합니다.
채널을 통해 스레드에 요청 보내기
다음으로 다룰 문제는, thread::spawn에 준 클로저가 아무것도 하지 않는다는 것입니다. 지금은 execute 메서드에서 실행하고 싶은 클로저를 받습니다. 하지만 ThreadPool을 만드는 동안 각 Worker를 만들 때, thread::spawn에 실행할 클로저를 주어야 합니다.
방금 만든 Worker 구조체가 ThreadPool이 가진 대기열에서 실행할 코드를 가져와, 그 코드를 자기 스레드에 보내 실행하게 하고 싶습니다.
스레드 간 메시지 전달에서 배운 채널, 곧 두 스레드 사이에서 소통하는 간단한 방법이 이 경우에 딱 맞을 것입니다. 채널을 작업의 대기열로 쓰고, execute는 ThreadPool에서 Worker 인스턴스로 작업을 보내며, Worker는 그 작업을 자기 스레드에 보냅니다. 계획은 다음과 같습니다.
| 단계 | 할 일 |
|---|---|
| 1 | ThreadPool이 채널을 만들고 송신자를 가집니다. |
| 2 | 각 Worker가 수신자를 가집니다. |
| 3 | 채널로 보낼 클로저를 담을 새 Job 구조체를 만듭니다. |
| 4 | execute 메서드가 실행하고 싶은 작업을 송신자로 보냅니다. |
| 5 | Worker는 자기 스레드에서 수신자를 돌며, 받은 작업의 클로저를 실행합니다. |
아래 그림은 이 계획대로 만든 뒤 작업이 어떻게 전달되고 실행되는지 보여 줍니다.
스레드 풀 작업 채널 경계
execute의 FnOnce Job을 Box로 감싸 send한 뒤 워커가 공유 Receiver를 잠그고 recv하고 guard를 해제한 다음 job을 실행하는 실제 코드의 자원 경계다. recv 직렬화와 job 병렬 실행을 단계마다 구분할 이득이 있다.
RUST · ARCHITECTURE
수신할 때만 잠그고, 작업은 잠금 밖에서 실행한다
execute가 Box로 감싼 작업을 채널에 넣습니다. 워커는 공유 Receiver에서 작업 하나를 받은 뒤 잠금을 풀고 실행합니다.
단계별 조건과 보충 설명
호출자 → execute
FnOnce + Send + 'static closure를 Job (Box<dyn FnOnce() + Send + 'static>)로 포장합니다. non-Send나 non-'static borrow를 캡처한 작업은 이 경계를 통과하지 못합니다.
Sender → unbounded queue
mpsc::channel()은 multi-producer, single-consumer이며 개념상 무제한 buffer입니다. worker 수는 queue admission·memory backpressure를 제한하지 않습니다.
공유 Receiver
Arc<Mutex<Receiver<Job>>>가 여러 worker의 수신 접근을 조정합니다. lock과 recv만 직렬화합니다.
Worker A · B
let 문이 끝나면서 recv에 쓰인 임시 MutexGuard가 사라집니다. 각 워커는 잠금 없이 job을 실행하므로 서로 다른 작업이 동시에 실행될 수 있습니다.
종료 조건
모든 Sender가 사라지고 대기열도 비면 recv가 오류를 반환합니다. 이 절의 recv().unwrap()은 그때 패닉합니다. 다음 절에서 오류를 받아 정상 종료하도록 바꿉니다.
세미콜론이 수신 잠금의 끝이다
let job = receiver.lock()
.unwrap().recv().unwrap();
job();Job = Box<dyn FnOnce() + Send + 'static>. 위 코드의 잠금은 let 문 끝에서 해제됩니다. while let Ok(job) = receiver.lock().unwrap().recv()로 바꾸면 임시 guard가 루프 본문까지 살아서 job()도 잠금 안에서 실행됩니다.
- job 전달과 recv의 동기화 경계
- 고정 worker가 active concurrency를 제한
- unbounded queue에는 backpressure 없음
- 모든 Sender clone drop → Receiver가 채널 종료 관찰
호출자, execute, Sender, unbounded queue, 공유 Receiver, 두 worker 사이의 작업 전달과 lock 해제 경계입니다.
이 예의 조건
- 본문과 원 도표에 제시된 사례의 입력·조건을 사용합니다. 설명용 수치나 독립 비교 경로는 각 단계에 표시합니다.
- 현재 상태·값과 전환 근거의 위치는 모든 단계에서 같습니다. 조건이 맞지 않거나 실행되지 않은 경로를 성공한 동작으로 해석하지 않습니다.
| 관찰·상황 | 값·보충 정보 | 변화·이유 |
|---|---|---|
| 작업 전송 execute가 클로저를 Box Job으로 send한다. |
|
|
| 수신 경계 worker가 공유 Receiver 잠금으로 job 하나를 받는다. |
|
|
| 잠금 밖 실행 guard를 끝내고 job을 호출한다. |
|
|
스레드 풀 작업 채널 경계
execute의 FnOnce Job을 Box로 감싸 send한 뒤 워커가 공유 Receiver를 잠그고 recv하고 guard를 해제한 다음 job을 실행하는 실제 코드의 자원 경계다. recv 직렬화와 job 병렬 실행을 단계마다 구분할 이득이 있다.
RUST · ARCHITECTURE
수신할 때만 잠그고, 작업은 잠금 밖에서 실행한다
execute가 Box로 감싼 작업을 채널에 넣습니다. 워커는 공유 Receiver에서 작업 하나를 받은 뒤 잠금을 풀고 실행합니다.
단계별 조건과 보충 설명
호출자 → execute
FnOnce + Send + 'static closure를 Job (Box<dyn FnOnce() + Send + 'static>)로 포장합니다. non-Send나 non-'static borrow를 캡처한 작업은 이 경계를 통과하지 못합니다.
Sender → unbounded queue
mpsc::channel()은 multi-producer, single-consumer이며 개념상 무제한 buffer입니다. worker 수는 queue admission·memory backpressure를 제한하지 않습니다.
공유 Receiver
Arc<Mutex<Receiver<Job>>>가 여러 worker의 수신 접근을 조정합니다. lock과 recv만 직렬화합니다.
Worker A · B
let 문이 끝나면서 recv에 쓰인 임시 MutexGuard가 사라집니다. 각 워커는 잠금 없이 job을 실행하므로 서로 다른 작업이 동시에 실행될 수 있습니다.
종료 조건
모든 Sender가 사라지고 대기열도 비면 recv가 오류를 반환합니다. 이 절의 recv().unwrap()은 그때 패닉합니다. 다음 절에서 오류를 받아 정상 종료하도록 바꿉니다.
세미콜론이 수신 잠금의 끝이다
let job = receiver.lock()
.unwrap().recv().unwrap();
job();Job = Box<dyn FnOnce() + Send + 'static>. 위 코드의 잠금은 let 문 끝에서 해제됩니다. while let Ok(job) = receiver.lock().unwrap().recv()로 바꾸면 임시 guard가 루프 본문까지 살아서 job()도 잠금 안에서 실행됩니다.
- job 전달과 recv의 동기화 경계
- 고정 worker가 active concurrency를 제한
- unbounded queue에는 backpressure 없음
- 모든 Sender clone drop → Receiver가 채널 종료 관찰
호출자, execute, Sender, unbounded queue, 공유 Receiver, 두 worker 사이의 작업 전달과 lock 해제 경계입니다.
이 예의 조건
- 본문과 원 도표에 제시된 사례의 입력·조건을 사용합니다. 설명용 수치나 독립 비교 경로는 각 단계에 표시합니다.
- 현재 상태·값과 전환 근거의 위치는 모든 단계에서 같습니다. 조건이 맞지 않거나 실행되지 않은 경로를 성공한 동작으로 해석하지 않습니다.
| 관찰·상황 | 값·보충 정보 | 변화·이유 |
|---|---|---|
| 작업 전송 execute가 클로저를 Box Job으로 send한다. |
|
|
| 수신 경계 worker가 공유 Receiver 잠금으로 job 하나를 받는다. |
|
|
| 잠금 밖 실행 guard를 끝내고 job을 호출한다. |
|
|
호출하는 쪽이 넘긴 클로저는 FnOnce + Send + 'static이어야 하며, execute가 그것을 Job, 곧 Box<dyn FnOnce() + Send + 'static>으로 감쌉니다. Send가 아니거나 'static이 아닌 참조를 캡처한 작업은 이 단계를 통과하지 못합니다. Job은 Sender를 거쳐 대기열에 들어가며, mpsc::channel()은 생산자가 여럿이고 소비자가 하나인 채널이고 개념상 크기 제한이 없는 버퍼라서, 워커의 수가 대기열에 들어오는 양이나 메모리를 제한하지는 않습니다. 여러 워커는 Arc<Mutex<Receiver<Job>>>로 수신자를 함께 쓰며, 잠금은 lock과 recv만 하나씩 처리합니다. let job = receiver.lock().unwrap().recv().unwrap(); 문이 끝나면 recv에 쓰인 임시 MutexGuard가 사라지므로, 각 워커는 잠금 없이 job()을 실행하고 서로 다른 작업이 동시에 실행될 수 있습니다. 모든 Sender가 사라지고 대기열도 비면 recv가 오류를 돌려주며, 이 글의 recv().unwrap()은 그때 패닉을 냅니다. 우아한 종료와 정리에서 그 오류를 받아 정상적으로 끝나도록 고칩니다.
먼저 ThreadPool::new에서 채널을 만들고, 다음 코드처럼 ThreadPool 인스턴스에 송신자를 담습니다. Job 구조체는 아직 아무것도 담지 않지만, 채널로 보낼 항목의 타입이 될 것입니다.
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에서 새 채널을 만들고, 풀이 송신자를 가지게 합니다. 이 코드는 성공적으로 컴파일됩니다.
스레드 풀이 채널을 만들 때, 각 Worker에 채널의 수신자를 넘겨 봅니다. Worker 인스턴스가 만드는 스레드에서 수신자를 쓰고 싶으므로, 클로저 안에서 receiver 매개변수를 참조합니다. 다음 코드는 아직 컴파일되지 않습니다. 지금까지의 src/lib.rs 전체입니다.
use std::{sync::mpsc, thread};
pub struct ThreadPool {
workers: Vec<Worker>,
sender: mpsc::Sender<Job>,
}
struct Job;
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 mut workers = Vec::with_capacity(size);
for id in 0..size {
workers.push(Worker::new(id, receiver));
}
ThreadPool { workers, sender }
}
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, receiver: mpsc::Receiver<Job>) -> Worker {
let thread = thread::spawn(|| {
receiver;
});
Worker { id, thread }
}
}작고 단순한 변경을 했습니다. 수신자를 Worker::new에 넘기고, 그것을 클로저 안에서 씁니다. 이 코드를 확인하면 다음 오류가 납니다.
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:45:33
|
45 | fn new(id: usize, receiver: mpsc::Receiver<Job>) -> Worker {
| --- in this method ^^^^^^^^^^^^^^^^^^^ this parameter takes ownership of the value
For more information about this error, try `rustc --explain E0382`.이 코드는 receiver를 여러 Worker 인스턴스에 넘기려 합니다. 스레드 간 메시지 전달에서 배웠듯 이것은 동작하지 않습니다. 러스트가 제공하는 채널의 구현은 다중 생산자, 단일 소비자입니다. 그러므로 채널의 소비하는 쪽 끝을 복제해 이 코드를 고칠 수는 없습니다. 또 같은 메시지를 여러 소비자에게 여러 번 보내고 싶은 것도 아닙니다. 여러 Worker 인스턴스가 있는 메시지 목록 하나를 두고, 각 메시지가 한 번만 처리되기를 원합니다.
또 채널 대기열에서 작업을 꺼내려면 receiver를 바꿔야 하므로, 스레드들은 receiver를 함께 쓰고 바꿀 수 있는 안전한 방법이 필요합니다. 그렇지 않으면 경쟁 조건이 생길 수 있습니다. 공유 상태 동시성에서 이야기한 스레드 안전한 스마트 포인터를 떠올려 보세요. 여러 스레드가 소유권을 함께 가지고 스레드들이 값을 바꿀 수 있게 하려면 Arc<Mutex<T>>를 써야 합니다. Arc 타입은 여러 Worker 인스턴스가 수신자를 함께 소유하게 하고, Mutex는 한 번에 한 Worker만 수신자에서 작업을 얻게 합니다. 다음 코드가 바꿔야 할 내용입니다.
use std::{
sync::{Arc, Mutex, mpsc},
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에 넣습니다. 새 Worker마다 Arc를 복제해 참조 카운트를 늘리므로, Worker 인스턴스들이 수신자의 소유권을 함께 가질 수 있습니다. 이렇게 바꾸면 코드가 컴파일됩니다. 거의 다 왔습니다.
execute 메서드 구현하기
마지막으로 ThreadPool의 execute 메서드를 구현합니다. Job도 구조체에서, execute가 받는 클로저의 타입을 담는 트레이트 객체의 타입 별칭으로 바꿉니다. 고급 타입에서 이야기했듯, 타입 별칭으로 긴 타입을 짧게 쓸 수 있습니다. 다음 코드를 보세요.
// --생략--
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을 쓰는 까닭은, 실패하는 경우가 일어나지 않는다는 것을 우리는 알지만 컴파일러는 모르기 때문입니다.
하지만 아직 완전히 끝나지 않았습니다. Worker에서 thread::spawn에 넘기는 클로저는 여전히 채널의 수신하는 끝을 참조할 뿐입니다. 대신 클로저가 영원히 돌며, 채널의 수신하는 끝에 작업을 요청하고 작업을 받으면 실행하게 해야 합니다. 다음 코드처럼 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을 불러 이 스레드가 패닉을 내게 하는 것이 알맞은 동작입니다. 의미 있는 오류 메시지를 담은 expect로 바꿔도 좋습니다.
뮤텍스의 락을 얻으면, recv를 불러 채널에서 Job을 받습니다. 마지막 unwrap은 여기서도 오류를 지나쳐 패닉을 내며, 송신자를 가진 스레드가 멈췄다면 오류가 날 수 있습니다. 수신자가 멈추면 send 메서드가 Err를 돌려주는 것과 비슷합니다.
recv를 부르면 막히므로, 아직 작업이 없다면 지금의 스레드는 작업을 쓸 수 있게 될 때까지 기다립니다. Mutex<T>는 한 번에 한 Worker 스레드만 작업을 요청하도록 보장합니다.
이제 스레드 풀이 동작하는 상태가 되었습니다. cargo run을 실행하고 요청을 몇 개 보내 봅니다. 지금까지의 src/lib.rs 전체는 다음과 같습니다.
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<()>,
}
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 }
}
}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)]` (part of `#[warn(unused)]`) 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에 요청을 보내면, 서버는 다른 스레드가 다른 요청을 실행하게 해 그 요청들을 처리할 수 있습니다. 출력에서 워커의 순서는 실행할 때마다 다를 수 있습니다.
패턴이 사용될 수 있는 모든 곳에서 while let 루프를 배웠으니, 다음 코드처럼 Worker 스레드의 코드를 쓰지 않은 까닭이 궁금할 수 있습니다.
// --생략--
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();을 쓰는 앞의 코드가 동작하는 까닭은, let을 쓰면 등호 오른쪽의 표현식에서 쓰인 임시 값은 let 문이 끝날 때 곧바로 버려지기 때문입니다. 하지만 while let(과 if let, match)은 연결된 블록이 끝날 때까지 임시 값을 버리지 않습니다. while let을 쓴 코드에서는 job()을 부르는 동안에도 락을 쥐고 있게 되므로, 다른 Worker 인스턴스가 작업을 받을 수 없습니다.
정리
ThreadPool은mpsc::channel()로 채널을 만들어 송신자를 가지고, 수신자는Arc<Mutex<Receiver<Job>>>로 감싸 모든Worker가 함께 씁니다.- 수신자를 그냥 여러
Worker에 넘기면 루프의 이전 반복에서 이미 옮겨졌다는 E0382 오류가 나며, 채널의 수신하는 끝은 복제할 수 없습니다. Job은Box<dyn FnOnce() + Send + 'static>의 타입 별칭이고,execute는 받은 클로저를Box에 넣어sender.send로 보냅니다.Worker의 스레드는loop안에서let job = receiver.lock().unwrap().recv().unwrap();으로 작업을 받아job()을 실행하며,let문이 끝날 때 락이 풀립니다.while let Ok(job) = receiver.lock().unwrap().recv()로 쓰면 블록이 끝날 때까지 락을 쥐고 있어 작업들이 하나씩만 실행됩니다.
연습 문제
1. 락은 언제 풀릴까
다음 두 Worker 루프 가운데 작업을 동시에 실행할 수 있는 것은 무엇인가요? 까닭은 무엇인가요?
// (가)
loop {
let guard = receiver.lock().unwrap();
let job = guard.recv().unwrap();
job();
}
// (나)
loop {
let job = {
let guard = receiver.lock().unwrap();
guard.recv().unwrap()
};
job();
}해설 보기
(나)만 동시에 실행할 수 있습니다. (가)는 guard가 루프 본문 끝까지 살아 있으므로 job()을 실행하는 동안에도 락을 쥐고 있어, 다른 워커가 작업을 받지 못합니다. (나)는 안쪽 블록이 끝나며 guard가 버려져 락이 풀린 뒤에 job()을 실행하므로, 그동안 다른 워커가 다음 작업을 받을 수 있습니다.
2. 대기열이 무한히 쌓이면
/sleep 요청이 초당 100개씩 계속 들어오면, 워커 네 개인 이 서버에서 무슨 일이 일어나나요? 어떻게 막을 수 있나요?
해설 보기
워커 네 개는 5초마다 작업 네 개씩만 끝내므로, 처리하지 못한 작업이 크기 제한이 없는 채널의 대기열에 계속 쌓여 메모리 사용량이 끝없이 늘어나고, 응답 시간도 계속 길어집니다. 대기열의 크기를 정한 mpsc::sync_channel(capacity)를 쓰면 대기열이 가득 찼을 때 send가 기다리게 되어, 받아들이는 연결의 속도를 처리 속도에 맞출 수 있습니다. 가득 찼을 때 곧바로 거절하려면 try_send로 실패를 받아 503 같은 응답을 보낼 수도 있습니다.
다음 글 우아한 종료와 정리에서는 ThreadPool이 버려질 때 남은 작업을 마치고 Worker 스레드가 정상적으로 끝나도록 정리하는 코드를 더합니다.