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

От однопоточного сервера к многопоточному

Сейчас сервер обрабатывает запросы по очереди: второе соединение он не начнёт обрабатывать, пока не закончит с первым. Чем больше запросов приходит на сервер, тем хуже такая последовательная обработка. Если на сервер придёт запрос, обработка которого занимает много времени, всем следующим запросам придётся ждать, пока он не завершится, — даже если сами они обрабатываются быстро. Это нужно исправить, но сначала посмотрим на проблему в действии.

Имитируем медленный запрос

Посмотрим, как медленный запрос влияет на остальные запросы к нашему серверу в его текущем виде. В листинге 21-10 добавлена обработка запроса к /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();

    for stream in listener.incoming() {
        let stream = stream.unwrap();

        handle_connection(stream);
    }
}

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

Случаев теперь три, поэтому мы перешли с if на match. Чтобы сопоставлять с образцами-строковыми литералами, нужно явно сопоставлять срез request_line: match, в отличие от метода проверки на равенство, не выполняет автоматическое взятие ссылки и разыменование.

Первая ветка такая же, как блок if из листинга 21-9. Вторая ветка соответствует запросу к /sleep: получив его, сервер поспит пять секунд, а потом отрисует страницу успешного ответа. Третья ветка такая же, как блок else из листинга 21-9.

Видно, насколько примитивен наш сервер: настоящие библиотеки распознают разные запросы гораздо лаконичнее!

Запустите сервер командой cargo run. Затем откройте два окна браузера: одно для http://127.0.0.1:7878, другое для http://127.0.0.1:7878/sleep. Если несколько раз открыть URI /, как раньше, он будет отвечать быстро. Но если открыть /sleep, а затем загрузить /, то / будет ждать, пока sleep не проспит свои пять секунд, и только потом загрузится.

Чтобы запросы не скапливались за медленным, есть несколько приёмов — например, асинхронность, как в главе 17. Мы же реализуем пул потоков.

Увеличиваем пропускную способность с помощью пула потоков

Пул потоков (thread pool) — это группа заранее запущенных потоков, которые ждут, когда им дадут задачу. Получив новую задачу, программа назначает на неё один из потоков пула, и этот поток её обрабатывает. Остальные потоки пула свободны и могут взять другие задачи, пришедшие, пока первый поток занят. Закончив свою задачу, первый поток возвращается в пул свободных потоков и готов к новой. Пул потоков позволяет обрабатывать соединения конкурентно и тем самым увеличивает пропускную способность сервера.

Число потоков в пуле мы ограничим небольшим значением, чтобы защититься от DoS-атак: если бы программа создавала новый поток на каждый пришедший запрос, кто-нибудь мог бы отправить на сервер 10 миллионов запросов, исчерпать все его ресурсы и полностью остановить обработку запросов.

Поэтому вместо неограниченного числа потоков в пуле будет ждать фиксированное их количество. Пришедшие запросы отправляются в пул на обработку. Пул ведёт очередь входящих запросов. Каждый поток пула берёт запрос из этой очереди, обрабатывает его и просит у очереди следующий. При таком устройстве можно обрабатывать до N запросов конкурентно, где N — число потоков. Если каждый поток занят долгим запросом, следующие запросы всё равно могут скапливаться в очереди, но мы увеличили число долгих запросов, которые сервер выдержит, прежде чем до этого дойдёт.

Это лишь один из многих способов увеличить пропускную способность веб-сервера. Можно изучить и другие варианты: модель fork/join, однопоточную модель асинхронного ввода-вывода, многопоточную модель асинхронного ввода-вывода. Если тема вам интересна, почитайте о других решениях и попробуйте их реализовать: в низкоуровневом языке вроде Rust возможны все эти варианты.

Прежде чем реализовывать пул потоков, поговорим о том, как им должно быть удобно пользоваться. Когда проектируете код, полезно сначала написать клиентский интерфейс — он подскажет, каким должно быть устройство. Напишите API так, как вам хотелось бы его вызывать, а затем реализуйте функциональность внутри этой структуры, а не наоборот — сначала функциональность, а потом публичный API.

Подобно тому как в проекте из главы 12 мы разрабатывали через тесты, здесь мы будем разрабатывать «через компилятор». Напишем код, вызывающий нужные нам функции, а потом по ошибкам компилятора будем решать, что поменять дальше, чтобы код заработал. Но сначала в качестве отправной точки посмотрим на приём, которым мы пользоваться не будем.

По потоку на каждый запрос

Сначала посмотрим, как выглядел бы код, если бы он создавал новый поток на каждое соединение. Как уже говорилось, это не окончательный план — из-за проблем с потенциально неограниченным числом потоков, — но это отправная точка, чтобы сначала получить работающий многопоточный сервер. Потом мы добавим пул потоков как улучшение, и сравнивать два решения будет проще.

В листинге 21-11 показано, что нужно изменить в main, чтобы в цикле for для обработки каждого потока создавался новый поток выполнения.

use std::{
    fs,
    io::{BufReader, prelude::*},
    net::{TcpListener, TcpStream},
    thread,
    time::Duration,
};

fn main() {
    let listener = TcpListener::bind("127.0.0.1:7878").unwrap();

    for stream in listener.incoming() {
        let stream = stream.unwrap();

        thread::spawn(|| {
            handle_connection(stream);
        });
    }
}

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

Как вы узнали в главе 16, thread::spawn создаёт новый поток и выполняет в нём код замыкания. Если запустить этот код, открыть в браузере /sleep, а затем / ещё в двух вкладках, вы увидите, что запросам к / действительно не нужно ждать окончания /sleep. Но, как мы уже говорили, рано или поздно это перегрузит систему, ведь новые потоки создаются без всякого ограничения.

Возможно, вы помните из главы 17, что это именно тот случай, где async и await по-настоящему блистают! Держите это в голове, пока мы строим пул потоков, и думайте, что было бы иначе, а что так же, если бы мы использовали async.

Создаём ограниченное число потоков

Мы хотим, чтобы пул потоков работал похожим, привычным образом, — тогда переход с потоков на пул не потребует больших изменений в коде, который использует наш API. В листинге 21-12 показан гипотетический интерфейс структуры ThreadPool, которую мы хотим использовать вместо thread::spawn.

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

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

С помощью ThreadPool::new мы создаём пул с настраиваемым числом потоков — в данном случае с четырьмя. Затем в цикле for вызываем pool.execute, интерфейс которого похож на thread::spawn: он принимает замыкание, которое пул должен выполнить для каждого потока данных. Нужно реализовать pool.execute так, чтобы он брал замыкание и отдавал его на выполнение одному из потоков пула. Этот код пока не скомпилируется, но мы всё равно попробуем — пусть компилятор подскажет, как его исправить.

Строим ThreadPool, следуя подсказкам компилятора

Внесите в src/main.rs изменения из листинга 21-12 и будем двигаться вперёд по ошибкам компилятора из cargo check. Вот первая ошибка:

$ cargo check
    Checking hello v0.1.0 (file:///projects/hello)
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`.
error: could not compile `hello` (bin "hello") due to 1 previous error

Отлично! Ошибка говорит, что нужен тип или модуль ThreadPool, — его мы сейчас и сделаем. Реализация ThreadPool не будет зависеть от того, какую работу выполняет веб-сервер. Поэтому превратим крейт hello из бинарного в библиотечный, и в нём будет жить реализация ThreadPool. После этого отдельную библиотеку пула потоков можно будет использовать для любой работы, которой нужен пул потоков, а не только для обслуживания веб-запросов.

Создайте файл src/lib.rs со следующим содержимым — это самое простое определение структуры ThreadPool, какое пока возможно:

pub struct ThreadPool;

Затем отредактируйте main.rs: подключите ThreadPool из библиотечного крейта, добавив в начало 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() {
        let stream = stream.unwrap();

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

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

Код всё ещё не работает, но проверим его снова и получим следующую ошибку, с которой надо разобраться:

$ cargo check
    Checking hello v0.1.0 (file:///projects/hello)
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`.
error: could not compile `hello` (bin "hello") due to 1 previous error

Эта ошибка говорит, что дальше нужно создать для ThreadPool ассоциированную функцию new. Мы также знаем, что у new должен быть один параметр, способный принять аргумент 4, и что она должна возвращать экземпляр ThreadPool. Реализуем самую простую функцию new с такими свойствами:

pub struct ThreadPool;

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

Для параметра size мы выбрали тип usize, потому что отрицательное число потоков не имеет смысла. К тому же эта 4 будет числом элементов в коллекции потоков, а именно для этого и нужен тип usize, как говорилось в разделе «Целочисленные типы» главы 3.

Проверим код ещё раз:

$ cargo check
    Checking hello v0.1.0 (file:///projects/hello)
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`.
error: could not compile `hello` (bin "hello") due to 1 previous error

Теперь ошибка в том, что у ThreadPool нет метода execute. Вспомните: в разделе «Создаём ограниченное число потоков» мы решили, что интерфейс пула потоков должен быть похож на thread::spawn. Кроме того, мы реализуем функцию execute так, чтобы она брала переданное замыкание и отдавала его на выполнение свободному потоку пула.

Определим у ThreadPool метод execute, принимающий замыкание в качестве параметра. Как говорилось в разделе «Перемещение захваченных значений из замыкания» главы 13, принимать замыкания в параметрах можно с тремя разными трейтами: 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 — потому что мы не знаем, сколько времени поток будет его выполнять. Создадим у ThreadPool метод execute, принимающий обобщённый параметр типа F с этими ограничениями:

pub struct ThreadPool;

impl ThreadPool {
    // --пропуск--
    pub fn new(size: usize) -> ThreadPool {
        ThreadPool
    }

    pub fn execute<F>(&self, f: F)
    where
        F: FnOnce() + Send + 'static,
    {
    }
}

После FnOnce по-прежнему стоят (), потому что этот FnOnce описывает замыкание без параметров, возвращающее единичный тип (). Как и в определениях функций, возвращаемый тип можно опустить из сигнатуры, но скобки нужны, даже если параметров нет.

Опять же, это самая простая реализация метода execute: он ничего не делает, мы лишь добиваемся, чтобы код скомпилировался. Проверим ещё раз:

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

Компилируется! Но учтите: если выполнить cargo run и сделать запрос в браузере, вы увидите те же ошибки, что и в начале главы. Наша библиотека пока не вызывает замыкание, переданное в execute!

Подумайте: что было бы иначе, если бы мы выполняли не замыкание, а футуру?

Проверяем число потоков в new

Мы пока никак не используем параметры new и execute. Реализуем тела этих функций с нужным поведением. Начнём с new. Раньше мы выбрали для параметра size беззнаковый тип, потому что пул с отрицательным числом потоков не имеет смысла. Но и пул с нулём потоков не имеет смысла, а ноль — вполне допустимое значение usize. Добавим проверку, что size больше нуля, прежде чем вернуть экземпляр ThreadPool, и с помощью макроса assert! заставим программу паниковать, если пришёл ноль, как в листинге 21-13.

pub struct ThreadPool;

impl ThreadPool {
    /// Создаёт новый ThreadPool.
    ///
    /// size — число потоков в пуле.
    ///
    /// # Паника
    ///
    /// Функция `new` паникует, если size равен нулю.
    pub fn new(size: usize) -> ThreadPool {
        assert!(size > 0);

        ThreadPool
    }

    // --пропуск--
    pub fn execute<F>(&self, f: F)
    where
        F: FnOnce() + Send + 'static,
    {
    }
}

Мы также добавили к ThreadPool документационные комментарии. Обратите внимание: следуя хорошим практикам документирования, мы добавили раздел, описывающий ситуации, в которых функция может паниковать, — как обсуждалось в главе 14. Попробуйте выполнить cargo doc --open и щёлкнуть по структуре ThreadPool, чтобы увидеть, как выглядит сгенерированная документация для new!

Вместо макроса assert! можно было бы переименовать new в build и возвращать Result, как мы сделали с Config::build в проекте с вводом-выводом в листинге 12-9. Но здесь мы решили, что попытка создать пул потоков вообще без потоков должна быть неисправимой ошибкой. Если чувствуете в себе силы, попробуйте написать функцию build с такой сигнатурой и сравните её с функцией new:

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

Готовим место для хранения потоков

Теперь, когда мы знаем, что число потоков для пула допустимо, можно создать эти потоки и сохранить их в структуре 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 будет единичным типом ().

Код из листинга 21-14 скомпилируется, но потоков пока не создаёт. Мы изменили определение ThreadPool, чтобы оно хранило вектор экземпляров thread::JoinHandle<()>, инициализировали вектор ёмкостью size, завели цикл for, в котором будет код создания потоков, и возвращаем экземпляр ThreadPool, который их содержит.

use std::thread;

pub struct ThreadPool {
    threads: Vec<thread::JoinHandle<()>>,
}

impl ThreadPool {
    // --пропуск--
    /// Создаёт новый ThreadPool.
    ///
    /// size — число потоков в пуле.
    ///
    /// # Паника
    ///
    /// Функция `new` паникует, если size равен нулю.
    pub fn new(size: usize) -> ThreadPool {
        assert!(size > 0);

        let mut threads = Vec::with_capacity(size);

        for _ in 0..size {
            // создаём потоки и сохраняем их в векторе
        }

        ThreadPool { threads }
    }
    // --пропуск--

    pub fn execute<F>(&self, f: F)
    where
        F: FnOnce() + Send + 'static,
    {
    }
}

Мы подключили std::thread в библиотечном крейте, потому что используем thread::JoinHandle как тип элементов вектора в ThreadPool.

Получив допустимый размер, ThreadPool создаёт новый вектор, способный вместить size элементов. Функция with_capacity делает то же, что Vec::new, но с важным отличием: она заранее выделяет место в векторе. Мы знаем, что нам нужно хранить в векторе size элементов, поэтому выделить память сразу немного эффективнее, чем использовать Vec::new, который меняет размер по мере добавления элементов.

Если снова запустить cargo check, проверка должна пройти успешно.

Передаём код из ThreadPool в поток

В цикле for из листинга 21-14 мы оставили комментарий о создании потоков. Теперь разберёмся, как на самом деле их создавать. Стандартная библиотека предоставляет для создания потоков thread::spawn, и thread::spawn ожидает получить код, который поток должен выполнить сразу после создания. Но нам нужно создать потоки и заставить их ждать код, который мы отправим позже. В стандартной реализации потоков такой возможности нет — придётся реализовать её вручную.

Для этого введём между ThreadPool и потоками новую структуру данных, которая и будет отвечать за это поведение. Назовём её Worker («работник») — это распространённый термин в реализациях пулов. Worker забирает код, который нужно выполнить, и выполняет его в своём потоке.

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

Вместо вектора экземпляров JoinHandle<()> будем хранить в пуле экземпляры структуры Worker. Каждый Worker будет хранить один экземпляр JoinHandle<()>. Затем реализуем у Worker метод, который будет принимать замыкание с кодом и отправлять его на выполнение в уже работающий поток. Ещё дадим каждому Worker идентификатор id, чтобы различать экземпляры Worker в пуле при логировании и отладке.

Вот как будет выглядеть создание ThreadPool теперь. Код, отправляющий замыкание в поток, мы напишем, когда Worker будет устроен так:

1. Определить структуру Worker, которая хранит id и JoinHandle<()>.
2. Изменить ThreadPool, чтобы он хранил вектор экземпляров Worker.
3. Определить функцию Worker::new, которая принимает число id и возвращает экземпляр Worker с этим id и потоком, запущенным с пустым замыканием.
4. В ThreadPool::new с помощью счётчика цикла for генерировать id, создавать новый Worker с этим id и сохранять его в векторе.

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

Готовы? Вот листинг 21-15 с одним из вариантов этих изменений.

use std::thread;

pub struct ThreadPool {
    workers: Vec<Worker>,
}

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

    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) -> Worker {
        let thread = thread::spawn(|| {});

        Worker { id, thread }
    }
}

Мы переименовали поле ThreadPool из threads в workers, потому что теперь в нём хранятся экземпляры Worker, а не JoinHandle<()>. Счётчик цикла for мы передаём аргументом в Worker::new, а каждый новый Worker сохраняем в векторе workers.

Внешнему коду (например, нашему серверу в src/main.rs) не нужно знать подробности реализации — то, что внутри ThreadPool используется структура Worker, — поэтому мы делаем структуру Worker и её функцию new приватными. Функция Worker::new использует переданный ей id и сохраняет экземпляр JoinHandle<()>, полученный при запуске нового потока с пустым замыканием.

Этот код скомпилируется и сохранит столько экземпляров Worker, сколько мы указали аргументом ThreadPool::new. Но замыкание, которое мы получаем в execute, мы всё ещё не обрабатываем. Разберёмся с этим дальше.

Отправляем запросы в потоки через каналы

Следующая проблема: замыкания, переданные в thread::spawn, не делают абсолютно ничего. Сейчас замыкание, которое нужно выполнить, мы получаем в методе execute. Но передать замыкание в thread::spawn нужно при создании каждого Worker, то есть при создании ThreadPool.

Мы хотим, чтобы только что созданные структуры Worker забирали код из очереди, которую хранит ThreadPool, и отправляли его на выполнение в свой поток.

Каналы из главы 16 — простой способ связи между двумя потоками — идеально подходят для этой задачи. Канал будет играть роль очереди заданий, а execute будет отправлять задание из ThreadPool экземплярам Worker, а те — в свой поток. План такой:

1. ThreadPool создаёт канал и хранит у себя отправителя.
2. Каждый Worker хранит у себя получателя.
3. Создаём новую структуру Job, которая будет хранить замыкания, отправляемые по каналу.
4. Метод execute отправляет задание, которое нужно выполнить, через отправителя.
5. Worker в своём потоке в цикле читает получателя и выполняет замыкания всех полученных заданий.

Начнём с того, что создадим канал в ThreadPool::new и сохраним отправителя в экземпляре ThreadPool, как в листинге 21-16. Структура Job пока ничего не хранит, но это будет тип элементов, которые мы отправляем по каналу.

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

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

struct Job;

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

    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) -> Worker {
        let thread = thread::spawn(|| {});

        Worker { id, thread }
    }
}

В ThreadPool::new мы создаём новый канал, и пул хранит отправителя. Это успешно скомпилируется.

Попробуем при создании канала передать получателя в каждый Worker. Мы знаем, что получатель нужен в потоке, который запускает Worker, поэтому обратимся к параметру receiver в замыкании. Код из листинга 21-17 пока не совсем компилируется.

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

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

struct Job;

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

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

$ 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
...
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:47:33
   |
47 |     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`.
error: could not compile `hello` (lib) due to 1 previous error

Код пытается передать receiver нескольким экземплярам Worker. Так не выйдет — вспомните главу 16: каналы в Rust устроены по схеме «много производителей, один потребитель». Значит, просто клонировать принимающий конец канала, чтобы исправить код, нельзя. К тому же мы не хотим отправлять одно сообщение нескольким потребителям много раз: нам нужен один список сообщений и несколько экземпляров Worker, причём так, чтобы каждое сообщение обрабатывалось один раз.

Кроме того, чтобы взять задание из очереди канала, нужно изменить receiver, поэтому потокам нужен безопасный способ совместно владеть receiver и изменять его — иначе возможны состояния гонки (см. главу 16).

Вспомните потокобезопасные умные указатели из главы 16: чтобы разделить владение между несколькими потоками и позволить им изменять значение, нужен Arc<Mutex<T>>. Arc позволит нескольким экземплярам Worker владеть получателем, а Mutex гарантирует, что в каждый момент задание из получателя берёт только один Worker. Нужные изменения — в листинге 21-18.

use std::{
    sync::{Arc, Mutex, mpsc},
    thread,
};
// --пропуск--

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

struct Job;

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,
    {
    }
}

// --пропуск--

struct Worker {
    id: usize,
    thread: thread::JoinHandle<()>,
}

impl Worker {
    fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
        // --пропуск--
        let thread = thread::spawn(|| {
            receiver;
        });

        Worker { id, thread }
    }
}

В ThreadPool::new мы помещаем получателя в Arc и Mutex. Для каждого нового Worker клонируем Arc, увеличивая счётчик ссылок, чтобы экземпляры Worker могли совместно владеть получателем.

С этими изменениями код компилируется! Мы почти у цели!

Реализуем метод execute

Наконец реализуем метод execute у ThreadPool. Заодно превратим Job из структуры в псевдоним типа для трейт-объекта, хранящего тип замыкания, которое получает execute. Как говорилось в разделе «Синонимы и псевдонимы типов» главы 20, псевдонимы типов позволяют сокращать длинные типы для удобства. Посмотрите на листинг 21-19.

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

// --пропуск--

struct Worker {
    id: usize,
    thread: thread::JoinHandle<()>,
}

impl Worker {
    fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
        let thread = thread::spawn(|| {
            receiver;
        });

        Worker { id, thread }
    }
}

Создав новый экземпляр Job из замыкания, полученного в execute, мы отправляем это задание в передающий конец канала. На send вызываем unwrap на случай, если отправка не удастся. Такое может случиться, например, если мы остановим все потоки, — тогда принимающий конец перестанет получать новые сообщения. Пока что остановить потоки мы не можем: они работают, пока существует пул. unwrap мы используем потому, что знаем: этот сбой не произойдёт, — но компилятор этого не знает.

Но это ещё не всё! В Worker замыкание, передаваемое в thread::spawn, по- прежнему лишь ссылается на принимающий конец канала. А нам нужно, чтобы замыкание бесконечно крутилось в цикле: просило у принимающего конца канала задание и выполняло его, когда получит. Внесём в Worker::new изменение из листинга 21-20.

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

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

Здесь мы сначала вызываем lock у receiver, чтобы захватить мьютекс, а затем 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 `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)]` 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, сервер сможет обслуживать другие запросы, выполняя их в другом потоке.

Самое время остановиться и подумать: чем отличался бы код из листингов 21-18, 21-19 и 21-20, если бы работа задавалась не замыканием, а футурами? Какие типы изменились бы? Изменились бы сигнатуры методов, и если да, то как? Какие части кода остались бы прежними?

Познакомившись с циклом while let в главах 17 и 19, вы, возможно, задаётесь вопросом: почему мы не написали код потока Worker так, как в листинге 21-21?

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

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 || {
            while let Ok(job) = receiver.lock().unwrap().recv() {
                println!("Worker {id} got a job; executing.");

                job();
            }
        });

        Worker { id, thread }
    }
}

Этот код компилируется и запускается, но нужного многопоточного поведения не даёт: медленный запрос всё так же заставляет остальные запросы ждать. Причина довольно тонкая. У структуры Mutex нет публичного метода unlock, потому что владение блокировкой определяется временем жизни MutexGuard<T> внутри LockResult<MutexGuard<T>>, который возвращает метод lock. Благодаря этому анализатор заимствований ещё на этапе компиляции следит за правилом: к ресурсу под защитой Mutex нельзя обратиться, не удерживая блокировку. Но из-за этого же блокировка может удерживаться дольше, чем задумано, если не следить за временем жизни MutexGuard<T>.

Код из листинга 21-20 с let job = receiver.lock().unwrap().recv().unwrap(); работает, потому что при let все временные значения из выражения справа от знака равенства уничтожаются сразу, как только заканчивается инструкция let. А вот while let (а также if let и match) не уничтожает временные значения до конца связанного с ним блока. В листинге 21-21 блокировка удерживается всё время, пока выполняется вызов job(), а значит, другие экземпляры Worker не могут получать задания.