ThreadPool과 Worker 구현

싱글스레드 서버를 멀티스레드 서버로 바꾸기에서 ThreadPool::new(4)와 pool.execute를 쓰는 main을 먼저 썼습니다. 이 글에서는 그 코드가 내는 컴파일러 오류를 하나씩 따라가며 ThreadPool을 만들고, 스레드를 담을 Worker 구조체를 더합니다.

컴파일러 주도 개발을 사용하여 ThreadPool 구현하기

지금의 src/main.rs 전체는 다음과 같습니다. handle_connection은 앞 글의 /sleep을 처리하는 판입니다.

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

// --생략--

src/main.rs를 이렇게 바꾼 다음, 개발을 이끌도록 cargo check의 컴파일러 오류를 봅니다. 처음 받는 오류는 다음과 같습니다.

cargo check

error[E0433]: cannot find type `ThreadPool` in this scope
  --> 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`.

좋습니다. 이 오류는 ThreadPool 타입이나 모듈이 필요하다고 알려 주므로, 이제 하나를 만듭니다. ThreadPool의 구현은 웹 서버가 하는 일의 종류와 상관없을 것입니다. 그러니 hello 크레이트를 바이너리 크레이트에서 ThreadPool 구현을 담는 라이브러리 크레이트로 바꿉니다. 라이브러리 크레이트로 바꾸고 나면, 웹 요청을 처리하는 것뿐 아니라 스레드 풀로 하고 싶은 어떤 일에든 이 별도의 스레드 풀 라이브러리를 쓸 수도 있습니다.

지금 가질 수 있는 가장 단순한 ThreadPool 구조체의 정의를 담은 src/lib.rs 파일을 만듭니다.

pub struct ThreadPool;

그다음 src/main.rs의 맨 위에 다음 코드를 더해, 라이브러리 크레이트의 ThreadPool을 스코프로 가져옵니다. hello는 패키지 이름이자 src/lib.rs로 만든 라이브러리 크레이트의 이름이므로, 바이너리 크레이트인 src/main.rs는 바깥 크레이트를 쓸 때처럼 hello::로 시작하는 경로를 씁니다.

use hello::ThreadPool;

이 코드는 아직 동작하지 않지만, 다시 확인해 다음에 다뤄야 할 오류를 얻습니다. 이제 ThreadPool이라는 이름은 찾지만 그 타입에 아직 아무 함수도 없으므로, 오류가 가리키는 대상이 타입의 이름에서 new로 옮겨 갑니다.

error[E0599]: no associated function or constant named `new` found for struct `ThreadPool` in the current scope
  --> src/main.rs:12:28
   |
12 |     let pool = ThreadPool::new(4);
   |                            ^^^ associated function or constant not found in `ThreadPool`

For more information about this error, try `rustc --explain E0599`.

맨 위에 use 줄을 하나 더했으므로 줄 번호가 하나 늘었습니다. 이 오류는 다음으로 ThreadPool에 new라는 연관 함수를 만들어야 한다는 것을 나타냅니다. new에는 인수로 4를 받을 수 있는 매개변수 하나가 있어야 하고, ThreadPool 인스턴스를 돌려주어야 한다는 것도 압니다. 그런 특성을 가진 가장 단순한 new 함수를 구현합니다.

pub struct ThreadPool;

impl ThreadPool {
    pub fn new(size: usize) -> ThreadPool {
        ThreadPool
    }
}

size 매개변수의 타입으로 usize를 고른 까닭은, 음수 개수의 스레드는 말이 되지 않기 때문입니다. 또 이 4를 스레드 컬렉션의 원소 수로 쓸 것이며, 데이터 타입에서 이야기했듯 그것이 usize 타입의 쓰임새입니다. 코드를 다시 확인합니다.

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`.

이제 ThreadPool에 execute 메서드가 없어서 오류가 납니다. 앞 글의 유한한 개수의 스레드 생성하기에서, 스레드 풀이 thread::spawn과 비슷한 인터페이스를 가지게 하기로 정했던 것을 떠올려 보세요. 또 execute 함수가 받은 클로저를 풀의 쉬고 있는 스레드에 주어 실행하게 하도록 구현할 것입니다.

ThreadPool의 execute 메서드가 클로저를 매개변수로 받도록 정의합니다. 클로저: 자신의 환경을 캡처하는 익명 함수에서, 클로저를 매개변수로 받을 때 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에 만듭니다.

impl ThreadPool {
    // --생략--
    pub fn execute<F>(&self, f: F)
    where
        F: FnOnce() + Send + 'static,
    {
    }
}

FnOnce 뒤에도 ()를 씁니다. 이 FnOnce는 매개변수를 받지 않고 유닛 타입 ()를 돌려주는 클로저를 나타내기 때문입니다. 함수 정의처럼 시그니처에서 반환 타입을 생략할 수 있지만, 매개변수가 없더라도 괄호는 필요합니다.

다시 말하지만 이것은 execute 메서드의 가장 단순한 구현이며, 아무것도 하지 않지만 코드가 컴파일되게만 하려는 것입니다. 다시 확인합니다.

    Checking hello v0.1.0 (file:///projects/hello)
    Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.24s

컴파일됩니다. 하지만 cargo run을 실행하고 브라우저에서 요청을 보내면, 이 장의 시작에서 본 오류를 브라우저에서 보게 됩니다. 라이브러리가 아직 execute에 넘긴 클로저를 실제로 부르지 않기 때문입니다.

엄격한 컴파일러를 가진 하스켈(Haskell)과 러스트 같은 언어에 대해 "코드가 컴파일되면 동작한다"라는 말을 들을 수 있습니다. 하지만 이 말이 언제나 맞지는 않습니다. 이 프로젝트는 컴파일되지만 아무것도 하지 않습니다. 실제의 완전한 프로젝트를 만든다면, 지금이 코드가 컴파일되면서도 원하는 동작을 하는지 확인하는 유닛 테스트를 쓰기 시작할 좋은 때일 것입니다.

생각해 볼 것이 있습니다. 클로저 대신 퓨처(future)를 실행한다면, 여기서 무엇이 달라질까요?

new에서 스레드 개수 검증하기

new와 execute의 매개변수로 아무것도 하지 않고 있습니다. 원하는 동작으로 이 함수들의 본문을 구현합니다. 먼저 new를 생각해 봅니다. 앞에서 size 매개변수에 부호 없는 타입을 골랐습니다. 스레드의 개수가 음수인 풀은 말이 되지 않기 때문입니다. 하지만 스레드가 0개인 풀도 말이 되지 않는데, 0은 완전히 유효한 usize입니다. 다음처럼 ThreadPool 인스턴스를 돌려주기 전에 size가 0보다 큰지 확인하고, 0을 받으면 assert! 매크로로 프로그램이 패닉을 내게 하는 코드를 더합니다.

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에 문서화 주석으로 문서도 더했습니다. Crates.io 배포에서 이야기했듯, 함수가 패닉을 낼 수 있는 상황을 밝히는 절을 더해 좋은 문서화 관행을 따랐다는 점을 보세요. cargo doc --open을 실행하고 ThreadPool 구조체를 눌러, new에 대해 만들어진 문서가 어떻게 보이는지 확인해 보세요.

여기서처럼 assert! 매크로를 더하는 대신, 다음처럼 new를 build로 바꿔 Result를 돌려주게 할 수도 있습니다. 모듈성과 에러 처리 향상을 위한 리팩터링의 I/O 프로젝트에서 Config::build로 한 것과 같습니다.

pub fn build(size: usize) -> Result<ThreadPool, PoolCreationError> {

하지만 이 경우에는 스레드 없이 스레드 풀을 만들려는 것을 복구할 수 없는 오류로 보기로 했습니다. 의욕이 있다면 build라는 함수를 이 시그니처로 써서 new 함수와 비교해 보세요.

스레드를 저장할 공간 만들기

풀에 저장할 유효한 스레드 수를 알 수 있게 되었으니, 그 스레드들을 만들어 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는 유닛 타입 ()가 될 것입니다.

다음 코드는 컴파일되지만 아직 스레드를 만들지는 않습니다. ThreadPool의 정의를 thread::JoinHandle<()> 인스턴스의 벡터를 담도록 바꾸고, 그 벡터를 size의 용량으로 초기화하고, 스레드를 만드는 어떤 코드를 실행할 for 루프를 마련하고, 그것들을 담은 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 {
            // 스레드를 만들어 벡터에 저장합니다
        }

        ThreadPool { threads }
    }
    // --생략--
}

ThreadPool의 벡터 원소 타입으로 thread::JoinHandle을 쓰므로, 라이브러리 크레이트에서 std::thread를 스코프로 가져왔습니다.

유효한 크기를 받으면 ThreadPool은 size개의 항목을 담을 수 있는 새 벡터를 만듭니다. with_capacity 함수는 Vec::new와 같은 일을 하지만, 중요한 차이가 있습니다. 벡터의 공간을 미리 할당합니다. 벡터에 size개의 원소를 저장해야 한다는 것을 알므로, 이렇게 미리 할당하는 것은 원소가 들어올 때마다 크기를 바꾸는 Vec::new를 쓰는 것보다 조금 더 효율적입니다.

cargo check를 다시 실행하면 성공할 것입니다.

ThreadPool에서 스레드로 코드의 전송을 담당하는 Worker 구조체

앞의 코드의 for 루프에 스레드를 만드는 것에 관한 주석을 남겨 두었습니다. 여기서는 실제로 스레드를 어떻게 만드는지 봅니다. 표준 라이브러리는 스레드를 만드는 방법으로 thread::spawn을 제공하며, thread::spawn은 스레드가 만들어지자마자 실행할 어떤 코드를 받을 것으로 기대합니다. 하지만 우리의 경우에는 스레드를 만들고, 나중에 보낼 코드를 기다리게 하고 싶습니다. 표준 라이브러리의 스레드 구현에는 그렇게 하는 방법이 들어 있지 않으므로, 직접 구현해야 합니다.

이 동작은 ThreadPool과 스레드 사이에 새 자료 구조를 도입해 구현합니다. 이 자료 구조를 Worker라고 부르며, 이것은 풀 구현에서 흔히 쓰는 용어입니다. Worker는 실행해야 할 코드를 받아 그 코드를 자기 스레드에서 실행합니다.

음식점 주방에서 일하는 사람들을 생각해 보세요. 일하는 사람들은 손님의 주문이 들어올 때까지 기다리고, 그다음 그 주문을 받아 처리합니다.

스레드 풀에 JoinHandle<()> 인스턴스의 벡터를 저장하는 대신, Worker 구조체의 인스턴스를 저장합니다. 각 Worker는 JoinHandle<()> 인스턴스 하나를 저장합니다. 그다음 Worker에 실행할 코드의 클로저를 받아, 이미 실행 중인 스레드에 보내 실행하게 하는 메서드를 구현합니다. 또 로그를 남기거나 디버깅할 때 풀의 서로 다른 Worker 인스턴스를 구분할 수 있도록, 각 Worker에 id를 줍니다.

ThreadPool을 만들 때 일어날 새 과정은 이렇습니다. Worker를 이렇게 설정한 뒤, 스레드에 클로저를 보내는 코드를 구현합니다.

도전해 보고 싶다면, 다음 코드를 보기 전에 이 바꾸기를 직접 구현해 보세요.

준비되었나요? 다음은 앞의 바꾸기를 한 한 가지 방법입니다.

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<()> 인스턴스를 저장합니다.

운영체제가 시스템 자원이 모자라 스레드를 만들 수 없으면 thread::spawn은 패닉을 냅니다. 그러면 일부 스레드를 만드는 데 성공했더라도 서버 전체가 패닉을 낼 것입니다. 단순하게 하려고 이 동작을 그대로 두지만, 프로덕션 스레드 풀 구현이라면 Result를 돌려주는 std::thread::Builder와 그 spawn 메서드를 쓰고 싶을 것입니다.

이 코드는 컴파일되며, ThreadPool::new의 인수로 지정한 수만큼의 Worker 인스턴스를 저장합니다. 아래 그림은 지금 만든 구조의 소유 관계를 보여 줍니다. 그림에서 예제 19-15라고 적은 것이 바로 앞의 코드입니다.

let thread = thread::spawn(|| {});

F: FnOnce() + Send + 'static

ThreadPool::new(4)는 Vec<Worker>에 네 Worker를 값으로 넣어 소유하며, 각 Worker의 id는 0, 1, 2, 3입니다. 각 Worker는 자기 id와 JoinHandle<()>을 가지며, 이 핸들은 만든 스레드를 나타낼 뿐 보관한다고 작업이 실행되거나 기다리는 것은 아닙니다. 그림의 화살표는 소유 관계일 뿐, 작업을 보내는 방향이나 실행 순서가 아닙니다. 지금의 Worker::new는 thread::spawn(|| {})으로 빈 클로저를 실행하므로 스레드가 곧바로 끝날 수 있고, 아직 요청을 기다리는 루프가 없습니다. execute도 F: FnOnce() + Send + 'static 클로저를 받는 시그니처만 있고 본문은 비어 있어, 작업을 전달하거나 부르지 않습니다.

하지만 아직 execute에서 받는 클로저를 처리하지 않습니다. 다음 글에서 그것을 하는 방법을 봅니다.

더 깊이: thread::Builder로 정할 수 있는 것

앞에서 말한 Builder는 실패를 Result로 돌려주는 것 말고도, 스레드를 만들기 전에 이름과 스택 크기를 정하게 해 줍니다. std::thread 문서에 따르면 이름을 붙인 스레드에서 패닉이 나면 패닉 메시지에 그 이름이 나오고, 이름은 가능한 플랫폼에서 운영체제에도 전달됩니다. 그래서 워커마다 worker-0 같은 이름을 붙여 두면, 패닉 메시지나 운영체제의 도구에서 문제가 난 워커를 찾기 쉬워집니다. 기본으로 만든 스레드에는 이름이 없습니다.

새로 만드는 스레드의 기본 스택 크기는 플랫폼에 따라 다르고 앞으로 바뀔 수 있지만, 문서에 따르면 지금은 1등급(Tier-1) 플랫폼에서 모두 2MiB입니다. 깊은 재귀를 하는 작업을 맡긴다면 Builder::stack_size로 크기를 정하거나 RUST_MIN_STACK 환경 변수에 바이트 수를 줄 수 있으며, 둘 다 있으면 stack_size가 앞섭니다. 메인 스레드의 스택 크기는 러스트가 정하지 않는다는 점도 문서에 적혀 있습니다. 풀의 스레드 수를 정할 때는 std::thread::available_parallelism이 알려 주는 동시 실행 수를 참고할 수 있으며, cargo test의 테스트 실행기도 이 값으로 기본 스레드 수를 정합니다.

정리

• 아직 없는 ThreadPool을 쓰는 main을 컴파일하면 E0433, new가 없다는 E0599, execute가 없다는 E0599가 차례로 나오며, 각 오류에 맞춰 필요한 것만 구현합니다.
• hello를 라이브러리 크레이트로 바꿔 src/lib.rs에 ThreadPool을 두고, src/main.rs에서 use hello::ThreadPool;로 가져옵니다.
• execute는 thread::spawn처럼 F: FnOnce() + Send + 'static 클로저를 받으며, 컴파일된다고 동작하는 것은 아닙니다.
• new는 assert!(size > 0)으로 스레드가 0개인 풀을 막고, 패닉하는 조건을 문서화 주석의 Panics 절에 적습니다.
• ThreadPool은 Vec::with_capacity(size)로 Worker들을 담고, 각 Worker는 id와 JoinHandle<()>을 가지며, Worker는 비공개 구현 세부 사항입니다.

연습 문제

1. build로 바꿔 보기

ThreadPool::new를 패닉 대신 Result를 돌려주는 ThreadPool::build로 바꾸려 합니다. 시그니처와 main의 쓰임은 어떻게 되나요?

2. 왜 JoinHandle만으로는 부족할까

ThreadPool이 Vec<thread::JoinHandle<()>>를 직접 담는 대신 Worker를 거치게 한 까닭은 무엇인가요?

다음 글 채널과 execute 구현에서는 채널로 작업을 Worker에 보내고 execute 메서드를 완성합니다.