이 시리즈는 Rust 공식문서를 통해 공부한 흔적임을 밝힙니다.

우리는 지난 시간, 잘 작동하는 쓰레드 풀 웹 서버를 만들어보았으나
다음과 같은 세 개의 경고 메시지가 남아 있었다.

warning: field is never read: `workers`
 --> src/lib.rs:4:5
  |
4 |     workers: Vec<Worker>,
  |     ^^^^^^^^^^^^^^^^^^^^
  |
  = note: `#[warn(dead_code)]` on by default

warning: field is never read: `id`
  --> src/lib.rs:44:5
   |
44 |     id: usize,
   |     ^^^^^^^^^

warning: field is never read: `thread`
  --> src/lib.rs:45:5
   |
45 |     thread: thread::JoinHandle<()>,
   |     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

warning: 3 warnings emitted

우리가 이 녀석들을 직접 읽지 않기 때문에 이에 대한 해제 작업이 제대로 이루어지지 않을 수 있다.
ctrl + c를 통해 강제 종료를 하게 될 경우 요청을 처리하는 도중이라고 해도
모든 쓰레드는 메인 쓰레드와 함께 종료된다.
그런 의미에서 ThreadPool에 Drop 트레이트를 구현하여
쓰레드 종료 전에 요청에 대한 처리를 완료하도록 수정해보자.

Drop 트레이트 구현

ThreadPool에 Drop 트레이트를 구현해보자.
우리가 이번 웹 서버 프로젝트에서 계속 그래왔듯이
이번에도 일단 작성해본 후 컴파일러의 오류 메시지에 따른 판단을 할 것이다.

peter@hp-laptop:~/rust-practice/chapter20/hello$ vi src/lib.rs

src/lib.rs

// snip

impl Drop for ThreadPool {
    fn drop(&mut self) {
        for worker in &mut self.workers {
            println!("Shutting down worker {}", worker.id);

            worker.thread.join().unwrap();
        }
    }
}

// snip

peter@hp-laptop:~/rust-practice/chapter20/hello$ cargo check
    Checking hello v0.1.0 (/home/peter/rust-practice/chapter20/hello)
error[E0507]: cannot move out of `worker.thread` which is behind a mutable reference
  --> src/lib.rs:48:13
   |
48 |             worker.thread.join().unwrap();
   |             ^^^^^^^^^^^^^ move occurs because `worker.thread` has type `std::thread::JoinHandle<()>`, which does not implement the `Copy` trait

error: aborting due to previous error

For more information about this error, try `rustc --explain E0507`.
error: could not compile `hello`.

To learn more, run the command again with --verbose.
peter@hp-laptop:~/rust-practice/chapter20/hello$ 

이것은 worker에 대한 소유권을 갖지 못한 상태에서 그것의 복사도 할 수 없어 발생한다.

이 문제를 해결하기 위해 우리는 Worker의 thread를
thread::JoinHandle<()>에서 Option<thread::JoinHandle<()>>으로 변경할 수 있다.
이로써 Some에 있는 값을 가져가고 None으로 채워 넣을 수 있다.
물론 이 필드를 사용하는 모든 곳에서 이 수정사항을 반영해주어야 한다.

peter@hp-laptop:~/rust-practice/chapter20/hello$ vi src/lib.rs

src/lib.rs

// snip

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();
            }
        }
    }
}

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 job = receiver.lock().unwrap().recv().unwrap();

                println!("Worker {} got a job; executing.", id);

                job();
            }
        );

        Worker { 
            id, 
            thread: Some(thread)
        }
    }
}

drop에서 Option이 None이면 이미 그것은 해제되었다는 것이므로
Some인 경우에 대해서만 꺼내서 join하고
Worker::new에서 생성할 때 Some으로 감싸 생성하도록 수정하였다.

그런데 문제가 있다.
우리의 drop은 종료된 쓰레드를 join하고 싶어하지만
우리의 쓰레드는 종료되지 않는다.
계속 무한 루프를 돌고 있을 뿐이다, 세상에...

수신 중단

우리는 이 문제를 해결하기 위해 코드를 수정할 것이다.
기존에는 쓰레드가 작업 큐에서 받아온 작업을 수행하는 게 다였지만
더이상 작업을 기다리지 않고 무한 루프를 빠져 나오는 선택지도 줄 것이다.

이를 위해 Job 인스턴스가 아닌 Message 열거형을 사용하도록 하며
그 열거값을 통해 Job 인스턴스를 받아오거나 무한 루프를 벗어나도록 한다.

peter@hp-laptop:~/rust-practice/chapter20/hello$ vi src/lib.rs

src/lib.rs

use std::thread;
use std::sync::{mpsc, Arc, Mutex};

pub struct ThreadPool {
    workers: Vec<Worker>,
    sender: mpsc::Sender<Message>,
}

type Job = Box<dyn FnOnce() + Send + 'static>;

enum Message {
    NewJob(Job),
    Terminate,
}

impl ThreadPool {
	// snip

    pub fn execute<F>(&self, f: F)
        where
            F: FnOnce() + Send + 'static
    {
        let job = Box::new(f);

        self.sender.send(Message::NewJob(job)).unwrap();
    }
}

impl Drop for ThreadPool {
    fn drop(&mut self) {
        println!("Sending terminate message to all workers.");

        for _ in &mut self.workers {
            self.sender.send(Message::Terminate).unwrap();
        }

        for worker in &mut self.workers {
            println!("Shutting down worker {}", worker.id);
            if let Some(thread) = worker.thread.take() {
                thread.join().unwrap();
            }
        }
    }
}

// snip

impl Worker {
    fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Message>>>) -> Worker {
        let thread = thread::spawn( move ||
            loop {
                let message = receiver.lock().unwrap().recv().unwrap();

                match message {
                    Message::NewJob(job) => {
                        println!("Worker {} got a job; executing.", id);

                        job();
                    },
                    Message::Terminate => {
                        println!("Worker {} was told to terminate.", id);

                        break;
                    }
                }
            }
        );

        // snip
    }
}

이제 요청을 처리하던 도중에 종료 시그널을 받아도
처리 중인 작업을 모두 끝낸 후 join될 것이다.

그리고 이제 Rust 공식 문서의 모든 학습을 마쳤다.
엄청난 Rust 마스터까지는 아니더라도
어디 가서 Rustacean이라고 주장할 수 있는 실력이 되었으리라.

공부 과정에서 작성된 코드를 모아놓은 저장소의 커밋 수가 181개로,
당연한 얘기지만(?) 내 저장소 중 가장 많은 커밋 수를 가진 저장소가 되었으며
어디 가서 Rustacean이라고 주장할 만한 언어 구성이 되었다.

두 달 동안 함께 공부해준(?) 페리스에게 감사 인사를 남긴다...☆

이 포스트의 내용은 공식문서의 20장 3절 Graceful Shutdown and Cleanup에 해당합니다.