Корректное завершение и очистка

Код из листинга 21-20, как мы и хотели, отвечает на запросы асинхронно с помощью пула потоков. Компилятор выдаёт предупреждения о полях workers, id и thread, которые мы напрямую не используем, — это напоминание о том, что мы ничего за собой не убираем. Когда мы останавливаем главный поток не самым изящным способом — сочетанием ctrl-C, — все остальные потоки тоже немедленно останавливаются, даже если они как раз обслуживают запрос.

Поэтому дальше мы реализуем трейт Drop, чтобы вызывать join для каждого потока пула: так потоки смогут закончить запросы, над которыми работают, прежде чем закрыться. Затем реализуем способ сказать потокам, что им пора перестать принимать новые запросы и завершиться. Чтобы увидеть этот код в действии, изменим сервер так, чтобы он принимал только два запроса, а затем корректно останавливал свой пул потоков.

Обратите внимание по ходу дела: ничто из этого не затрагивает код, который выполняет замыкания, так что всё здесь было бы точно так же, если бы мы использовали пул потоков для асинхронной среды выполнения.

Реализуем трейт `Drop` для `ThreadPool`

Начнём с реализации Drop для нашего пула потоков. Когда пул уничтожается, все его потоки должны быть присоединены (join), чтобы гарантированно закончить свою работу. В листинге 21-22 — первая попытка реализовать Drop; этот код пока не совсем работает.

Файл: 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 {
    /// Создаёт новый ThreadPool.
    ///
    /// size — число потоков в пуле.
    ///
    /// # Паника
    ///
    /// Функция `new` паникует, если size равен нулю.
    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();
    }
}

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

            worker.thread.join().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 }
    }
}

Листинг 21-22: Присоединяем каждый поток, когда пул потоков выходит из области видимости

Сначала перебираем в цикле все workers пула. Используем для этого &mut, потому что self — изменяемая ссылка, а ещё нам нужно иметь возможность изменять worker. Для каждого worker выводим сообщение о том, что этот экземпляр Worker завершает работу, а затем вызываем join у потока этого экземпляра. Если вызов join не удастся, с помощью unwrap заставляем Rust запаниковать и завершиться уже некорректно.

При компиляции этого кода получаем ошибку:

$ cargo check
    Checking hello v0.1.0 (file:///projects/hello)
error[E0507]: cannot move out of `worker.thread` which is behind a mutable reference
  --> src/lib.rs:52:13
   |
52 |             worker.thread.join().unwrap();
   |             ^^^^^^^^^^^^^ ------ `worker.thread` moved due to this method call
   |             |
   |             move occurs because `worker.thread` has type `JoinHandle<()>`, which does not implement the `Copy` trait
   |
note: `JoinHandle::<T>::join` takes ownership of the receiver `self`, which moves `worker.thread`
  --> /rustc/88d9e12ae178fab0fb5cc050a94da85685d449ea/library/std/src/thread/join_handle.rs:149:16

For more information about this error, try `rustc --explain E0507`.
error: could not compile `hello` (lib) due to 1 previous error

Ошибка говорит, что вызвать join нельзя: у нас есть лишь изменяемое заимствование каждого worker, а join забирает владение своим аргументом. Чтобы решить проблему, нужно переместить поток из экземпляра Worker, который владеет thread, — тогда join сможет поглотить поток. Один из способов — поступить так же, как в листинге 18-15. Если бы Worker хранил Option<thread::JoinHandle<()>>, можно было бы вызвать у Option метод take: он забирает значение из варианта Some и оставляет на его месте None. Иначе говоря, у работающего Worker в thread был бы Some, а когда понадобилось бы убрать за Worker, мы бы заменили Some на None, и потока для выполнения у Worker не осталось бы.

Но понадобилось бы это только при уничтожении Worker. А взамен пришлось бы возиться с Option<thread::JoinHandle<()>> везде, где мы обращаемся к worker.thread. В идиоматичном Rust Option используется довольно часто, но если вы ловите себя на том, что в качестве обходного пути заворачиваете в Option то, что, как вы знаете, всегда будет на месте, — стоит поискать другой подход, чтобы код был чище и в нём было меньше места для ошибок.

В данном случае есть вариант получше — метод Vec::drain. Он принимает диапазон, указывающий, какие элементы удалить из вектора, и возвращает итератор по этим элементам. Если передать диапазон .., из вектора будут удалены все значения.

Итак, нужно обновить реализацию drop у ThreadPool вот так:

Файл: src/lib.rs

#![allow(unused)]
fn main() {
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 {
    /// Создаёт новый ThreadPool.
    ///
    /// size — число потоков в пуле.
    ///
    /// # Паника
    ///
    /// Функция `new` паникует, если size равен нулю.
    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();
    }
}

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

            worker.thread.join().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 }
    }
}
}

Это устраняет ошибку компиляции и не требует никаких других изменений в коде. Учтите: поскольку drop может вызываться во время паники, unwrap тоже может запаниковать и вызвать двойную панику, которая немедленно обрушит программу и прервёт любую незавершённую очистку. Для учебной программы это допустимо, но в продакшен-коде так делать не рекомендуется.

Сигнализируем потокам, что пора перестать ждать задания

После всех изменений код компилируется без единого предупреждения. Но плохая новость в том, что работает он пока не так, как мы хотим. Всё дело в логике замыканий, которые выполняют потоки экземпляров Worker: сейчас мы вызываем join, но потоки от этого не завершатся, потому что они бесконечно крутятся в loop в поисках заданий. Если попытаться уничтожить ThreadPool с текущей реализацией drop, главный поток навсегда заблокируется в ожидании, пока завершится первый поток.

Чтобы это исправить, нужно изменить реализацию drop у ThreadPool, а затем цикл в Worker.

Сначала изменим реализацию drop у ThreadPool, чтобы она явно уничтожала sender, прежде чем ждать завершения потоков. В листинге 21-23 показаны изменения в ThreadPool, позволяющие явно уничтожить sender. В отличие от ситуации с потоком, здесь нам действительно нужен Option, чтобы с помощью Option::take переместить sender из ThreadPool.

Файл: src/lib.rs

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

pub struct ThreadPool {
    workers: Vec<Worker>,
    sender: Option<mpsc::Sender<Job>>,
}
// --пропуск--

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

impl ThreadPool {
    /// Создаёт новый ThreadPool.
    ///
    /// size — число потоков в пуле.
    ///
    /// # Паника
    ///
    /// Функция `new` паникует, если size равен нулю.
    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: Some(sender),
        }
    }

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

        self.sender.as_ref().unwrap().send(job).unwrap();
    }
}

impl Drop for ThreadPool {
    fn drop(&mut self) {
        drop(self.sender.take());

        for worker in self.workers.drain(..) {
            println!("Shutting down worker {}", worker.id);

            worker.thread.join().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 }
    }
}

Листинг 21-23: Явно уничтожаем sender, прежде чем присоединять потоки Worker

Уничтожение sender закрывает канал — это означает, что сообщений больше не будет. Когда это происходит, все вызовы recv, которые экземпляры Worker делают в бесконечном цикле, начинают возвращать ошибку. В листинге 21-24 мы меняем цикл в Worker так, чтобы в этом случае он корректно завершался, — а значит, потоки закончат работу, когда реализация drop у ThreadPool вызовет для них join.

Файл: src/lib.rs

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

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

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

impl ThreadPool {
    /// Создаёт новый ThreadPool.
    ///
    /// size — число потоков в пуле.
    ///
    /// # Паника
    ///
    /// Функция `new` паникует, если size равен нулю.
    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: Some(sender),
        }
    }

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

        self.sender.as_ref().unwrap().send(job).unwrap();
    }
}

impl Drop for ThreadPool {
    fn drop(&mut self) {
        drop(self.sender.take());

        for worker in self.workers.drain(..) {
            println!("Shutting down worker {}", worker.id);

            worker.thread.join().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 message = receiver.lock().unwrap().recv();

                match message {
                    Ok(job) => {
                        println!("Worker {id} got a job; executing.");

                        job();
                    }
                    Err(_) => {
                        println!("Worker {id} disconnected; shutting down.");
                        break;
                    }
                }
            }
        });

        Worker { id, thread }
    }
}

Листинг 21-24: Явно выходим из цикла, когда recv возвращает ошибку

Чтобы увидеть этот код в действии, изменим main так, чтобы сервер принимал только два запроса, а затем корректно завершался, как в листинге 21-25.

Файл: src/main.rs

use hello::ThreadPool;
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().take(2) {
        let stream = stream.unwrap();

        pool.execute(|| {
            handle_connection(stream);
        });
    }

    println!("Shutting down.");
}

fn handle_connection(mut stream: TcpStream) {
    let buf_reader = BufReader::new(&stream);
    let request_line = buf_reader.lines().next().unwrap().unwrap();

    let (status_line, filename) = match &request_line[..] {
        "GET / HTTP/1.1" => ("HTTP/1.1 200 OK", "hello.html"),
        "GET /sleep HTTP/1.1" => {
            thread::sleep(Duration::from_secs(5));
            ("HTTP/1.1 200 OK", "hello.html")
        }
        _ => ("HTTP/1.1 404 NOT FOUND", "404.html"),
    };

    let contents = fs::read_to_string(filename).unwrap();
    let length = contents.len();

    let response =
        format!("{status_line}\r\nContent-Length: {length}\r\n\r\n{contents}");

    stream.write_all(response.as_bytes()).unwrap();
}

Листинг 21-25: Завершаем работу сервера после обслуживания двух запросов, выходя из цикла

Настоящему веб-серверу, конечно, незачем останавливаться после двух запросов. Этот код лишь показывает, что корректное завершение и очистка работают.

Метод take определён в трейте Iterator и ограничивает перебор максимум двумя первыми элементами. В конце main ThreadPool выйдет из области видимости, и выполнится реализация drop.

Запустите сервер командой cargo run и сделайте три запроса. Третий запрос должен завершиться ошибкой, а в терминале вы увидите примерно такой вывод:

$ cargo run
   Compiling hello v0.1.0 (file:///projects/hello)
    Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.41s
     Running `target/debug/hello`
Worker 0 got a job; executing.
Shutting down.
Shutting down worker 0
Worker 3 got a job; executing.
Worker 1 disconnected; shutting down.
Worker 2 disconnected; shutting down.
Worker 3 disconnected; shutting down.
Worker 0 disconnected; shutting down.
Shutting down worker 1
Shutting down worker 2
Shutting down worker 3

Порядок идентификаторов Worker и сообщений у вас может быть другим. По сообщениям видно, как работает код: первые два запроса достались экземплярам Worker 0 и 3. После второго соединения сервер перестал принимать новые, и реализация Drop у ThreadPool начала выполняться ещё до того, как Worker 3 вообще взялся за своё задание. Уничтожение sender отключает все экземпляры Worker и велит им завершиться. Каждый Worker, отключаясь, выводит сообщение, а затем пул потоков вызывает join и ждёт завершения каждого потока Worker.

Обратите внимание на одну интересную деталь именно этого запуска: ThreadPool уничтожил sender, и ещё до того, как хоть один Worker получил ошибку, мы попытались присоединить Worker 0. Worker 0 пока не получил ошибку от recv, поэтому главный поток заблокировался, ожидая, пока Worker 0 закончит. Тем временем Worker 3 получил задание, а затем ошибку получили все потоки. Когда Worker 0 завершился, главный поток дождался завершения остальных экземпляров Worker. К этому моменту все они уже вышли из своих циклов и остановились.

Поздравляем! Проект готов: у нас есть простой веб-сервер, который с помощью пула потоков отвечает на запросы асинхронно. Сервер умеет корректно завершаться, убирая за собой все потоки пула.

Вот полный код для справки:

Файл: src/main.rs

use hello::ThreadPool;
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().take(2) {
        let stream = stream.unwrap();

        pool.execute(|| {
            handle_connection(stream);
        });
    }

    println!("Shutting down.");
}

fn handle_connection(mut stream: TcpStream) {
    let buf_reader = BufReader::new(&stream);
    let request_line = buf_reader.lines().next().unwrap().unwrap();

    let (status_line, filename) = match &request_line[..] {
        "GET / HTTP/1.1" => ("HTTP/1.1 200 OK", "hello.html"),
        "GET /sleep HTTP/1.1" => {
            thread::sleep(Duration::from_secs(5));
            ("HTTP/1.1 200 OK", "hello.html")
        }
        _ => ("HTTP/1.1 404 NOT FOUND", "404.html"),
    };

    let contents = fs::read_to_string(filename).unwrap();
    let length = contents.len();

    let response =
        format!("{status_line}\r\nContent-Length: {length}\r\n\r\n{contents}");

    stream.write_all(response.as_bytes()).unwrap();
}

Файл: src/lib.rs

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

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

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

impl ThreadPool {
    /// Создаёт новый ThreadPool.
    ///
    /// size — число потоков в пуле.
    ///
    /// # Паника
    ///
    /// Функция `new` паникует, если size равен нулю.
    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: Some(sender),
        }
    }

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

        self.sender.as_ref().unwrap().send(job).unwrap();
    }
}

impl Drop for ThreadPool {
    fn drop(&mut self) {
        drop(self.sender.take());

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

                match message {
                    Ok(job) => {
                        println!("Worker {id} got a job; executing.");

                        job();
                    }
                    Err(_) => {
                        println!("Worker {id} disconnected; shutting down.");
                        break;
                    }
                }
            }
        });

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

Здесь можно сделать ещё многое! Если хотите и дальше развивать проект, вот несколько идей:

Добавить больше документации к ThreadPool и его публичным методам.
Добавить тесты функциональности библиотеки.
Заменить вызовы unwrap более надёжной обработкой ошибок.
Использовать ThreadPool для какой-нибудь другой задачи, а не для обслуживания веб-запросов.
Найти крейт пула потоков на crates.io и написать похожий веб-сервер с его помощью. Затем сравнить его API и надёжность с нашим пулом потоков.

Итоги

Отличная работа! Вы дочитали книгу до конца! Спасибо, что прошли с нами это путешествие по Rust. Теперь вы готовы писать собственные проекты на Rust и помогать с чужими. Помните, что вокруг — дружелюбное сообщество растацеанов (Rustaceans), которые с радостью помогут вам с любыми трудностями на вашем пути в Rust.

Keyboard shortcuts

Язык программирования Rust