Этот код немного неряшливый, но он достаточно хорошо подходит для целей имитации.
Мы создали второй запрос sleep
, данные которого распознает сервер. Мы добавили else if после блока if
, чтобы проверить запрос к /sleep. Когда этот запрос будет получен, сервер заснёт на 5 секунд, прежде чем отобразить HTML страницу успешного выполнения.
Можно увидеть, насколько примитивен наш сервер: реальные библиотеки будут обрабатывать распознавание нескольких запросов гораздо менее многословно!
Запустите сервер командой cargo run
. Затем откройте два окна браузера: одно с адресом http://127.0.0.1:7878/, другое с http://127.0.0.1:7878/sleep. Если вы несколько раз обратитесь к URI /, то как и раньше увидите, что сервер быстро ответит. Но если вы введёте URI /sleep, затем загрузите URI /, то увидите что / ждёт, пока
/sleep не отработает полные 5 секунд перед загрузкой страницы.
Есть несколько способов изменить работу нашего веб-сервера, чтобы избежать медленной обработки большого количества запросов из-за одного медленного; способ который мы реализуем является пулом потоков.
Улучшение пропускной способности с помощью пула потоков
Пул потоков является группой заранее порождённых потоков, ожидающих в пуле и готовых выполнить задачу. Когда программа получает новую задачу, она назначает задачу одному из потоков в пуле и этот поток будет обрабатывать задачу. Остальные потоки в пуле доступны для обработки любых других задач, возникающих во время обработки первого потока. Когда первый поток завершает обработку своей задачи, он возвращается в пул свободных потоков, готовых обработать новую задачу. Пул потоков позволяет обрабатывать соединения одновременно, увеличивая пропускную способность вашего сервера.
Мы ограничим число потоков в пуле небольшим числом, чтобы защитить нас от атак типа «отказ в обслуживании» (DoS - Denial of Service); если бы наша программа создавала новый поток в момент поступления каждого запроса, то кто-то сделавший 10 миллионов запросов к серверу, мог бы создать хаос, использовать все ресурсы нашего сервера и остановить обработку запросов.
Вместо порождения неограниченного количества потоков, у нас будет фиксированное количество потоков, ожидающих в пуле. По мере поступления запросов они будут отправляться в пул для обработки. Пул будет поддерживать очередь входящих запросов.
Каждый из потоков в пуле будет извлекать запрос из этой очереди, обрабатывать запрос и затем запрашивать в очереди следующий запрос. При таком дизайне мы можем обрабатывать
N
запросов одновременно, где
N
- количество потоков. Если каждый поток отвечает на длительный запрос, последующие запросы могут по-прежнему задержаться в очереди, но мы увеличили число долго играющих запросов, которые можно обработать до достижения этой точки.
Этот подход является лишь одним из многих способов улучшить пропускную способность веб-сервера. Другими вариантами, которые вы могли бы изучить являются модель fork/join и однопоточная модель асинхронного ввода-вывода. Если вам интересна эта тема, вы можете прочитать о других решениях больше и попробовать внедрить их в помощью Rust. С языком низкого уровня как Rust, возможны все эти варианты.
Прежде чем приступить к реализации пула потоков, давайте поговорим о том, как должно выглядеть использование пула. Когда вы пытаетесь проектировать код, сначала необходимо написать клиентский интерфейс. Напишите API кода, чтобы он был структурирован так, как вы хотите его вызывать, затем реализуйте функциональность данной структуры, вместо подхода реализовывать функционал, а затем разрабатывать общедоступный API.
Подобно тому, как мы использовали разработку через тестирование (test-driven) в проекте главы 12, мы будем использовать здесь разработку, управляемую компилятором
(compiler-driven). Мы напишем код, который вызывает нужные нам функции, а затем посмотрим на ошибки компилятора, чтобы определить, что мы должны изменить дальше, чтобы заставить код работать.
1 ... 54 55 56 57 58 59 60 61 62
Структура кода, если мы могли бы создавать поток для каждого запросаСначала давайте рассмотрим, как мог бы выглядеть код, если он создавал бы новый поток для каждого соединения. Как упоминалось ранее, это не окончательный план, а это отправная точка из-за проблем с возможным порождением неограниченного количества потоков. В листинге 20-11 показаны изменения, которые нужно внести в main, чтобы запускать новый поток для обработки каждого входящего потока соединения в цикле forФайл: src/main.rs
Листинг 20-11: Порождение нового потока для каждого потока соединенияКак вы изучили в главе 16, thread::spawn создаст новый поток и затем запустит код замыкания в этом новом потоке. Если вы запустите этот код и загрузите
/sleep в своём браузере, в затем загрузите
/ в двух других вкладках браузера, вы действительно увидите,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); }); } }
что запросы к / не должны ждать завершения /sleep. Но, как мы уже упоминали, это в конечном счёте перегрузит систему, потому что вы будете создавать новые потоки без каких-либо ограничений.
Создание аналогичного интерфейса для конечного числа потоков
Мы хотим, чтобы наш пул потоков работал аналогичным, знакомым образом, чтобы переключение с потоков на пул потоков не требовало больших изменений в коде использующем наш API. В листинге 20-12 показан гипотетический интерфейс для структуры
ThreadPool
, который мы хотим использовать вместо thread::spawn
Файл: src/main.rs
Листинг 20-12: Наш идеальный интерфейс
ThreadPool
Мы используем
ThreadPool::new
, чтобы создать новый пул потоков с конфигурируемым количеством потоков, в данном случае четыре. Затем в цикле for выполняем pool.execute имеющий интерфейс, аналогичный интерфейсу thread::spawn
, в котором выполняется замыкание, которое пул должен выполнить для каждого потока соединения. Нам нужно реализовать pool.execute
, чтобы он принимал замыкание и передавал его потоку из пула для выполнения. Этот код не компилируется, но мы постараемся, чтобы компилятор в его исправлении.
Создание структуры ThreadPool использованием разработки, управляемой
компилятором
Внесите изменения листинга 20-12 в файл src/main.rs, а затем давайте воспользуемся ошибками компилятора из команды cargo check для управления нашей разработкой.
Вот первая ошибка, которую мы получаем:
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);
});
}
}
Замечательно! Ошибка говорит о том, что нам нужен тип или модуль
ThreadPool
,
поэтому мы создадим его сейчас. Наша реализация
ThreadPool будет зависеть от того,
какую работу выполняет наш веб-сервер. Итак, давайте переделаем крейт hello из бинарного в библиотечный для хранения реализации
ThreadPool
. После того, как поменяем в библиотечный крейт, мы также сможем использовать отдельную библиотеку пула потоков для любой работы, которую мы хотим выполнить с его использованием, а не только для обслуживания веб-запросов.
Создайте файл src/lib.rs, который содержит следующее, что является простейшим определением структуры
ThreadPool
, которую мы можем иметь в данный момент:
Файл: src/lib.rs
Затем создайте новый каталог src/bin и переместите двоичный крейт с корнем в
src/main.rs в src/bin/main.rs. Это сделает библиотечный крейт основным крейтом в каталоге hello; мы все ещё можем запустить двоичный файл из src/bin/main.rs, используя cargo run
. Переместив файл main.rs, отредактируйте его, чтобы подключить крейт библиотеки и добавить тип
ThreadPool в область видимости, добавив следующий код в начало src/bin/main.rs:
Файл: src/bin/main.rs
Этот код по-прежнему не будет работать, но давайте проверим его ещё раз, чтобы получить следующую ошибку, которую нам нужно устранить:
$
cargo check
Checking hello v0.1.0 (file:///projects/hello) error[E0433]: failed to resolve: use of undeclared type `ThreadPool`
-->
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` due to previous error pub struct
ThreadPool
;
{{#rustdoc_include ../listings/ch20-web-server/no-listing-
01
-define-threadpool- struct
/
src
/bin/main.rs:here}}
Эта ошибка указывает, что далее нам нужно создать ассоциированную функцию с именем new для
ThreadPool
. Мы также знаем, что new должен иметь один параметр,
который может принимать
4
в качестве аргумента и должен возвращать экземпляр
ThreadPool
. Давайте реализуем простейшую функцию new
, которая будет иметь эти характеристики:
Файл: src/lib.rs
Мы выбираем usize в качестве типа параметра size
, потому что мы знаем, что отрицательное число потоков не имеет никакого смысла. Мы также знаем, что мы будем использовать число 4 в качестве количества элементов в коллекции потоков, для чего предназначен тип usize
, как обсуждалось в разделе "Целочисленные типы"
главы 3.
Давайте проверим код ещё раз:
Теперь мы получаем предупреждение и ошибку. Игнорируем предупреждение не надолго, ошибка происходит потому что у нас нет метода execute в структуре
ThreadPool
. Вспомните раздел "Создание подобного интерфейса для конечного числа
$
cargo check
Checking hello v0.1.0 (file:///projects/hello) error[E0599]: no function or associated item named `new` found for struct
`ThreadPool` in the current scope
-->
src/main.rs:12:28
|
12 | let pool = ThreadPool::new(4);
| ^^^ function or associated item not found in
`ThreadPool`
For more information about this error, try `rustc --explain E0599`. error: could not compile `hello` due to previous error pub struct
ThreadPool
; impl
ThreadPool { pub fn new
(size: usize
) -> ThreadPool {
ThreadPool
}
}
$
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` due to previous error
потоков"
, в котором мы решили, что наш пул потоков должен иметь интерфейс,
похожий на thread::spawn
. Кроме того, мы реализуем функцию execute
, чтобы она принимала замыкание и передавала его свободному потоку из пула для запуска.
Мы определим метод execute у
ThreadPool для приёма замыкания в качестве параметра. Вспомните раздел "Хранение замыканий с использованием общих параметров и типажей
Fn
"
главы 13 и о том, что мы можем принимать замыкания в качестве параметров с тремя различными типажами:
Fn
,
FnMut и
FnOnce
. Нам нужно решить, какой тип замыкания использовать здесь. Мы знаем, что в конечном счёте мы сделаем что-то похожее на реализацию стандартной библиотеки thread::spawn
, поэтому мы можем посмотреть, какие ограничения накладывает на его параметр в сигнатуре thread::spawn
. Документация показывает следующее:
Параметр типа
F
- это тот, который нас интересует; параметр типа
T
относится к возвращаемому значению и нам он не интересен. Можно увидеть, что spawn использует
FnOnce в качестве ограничения типажа у
F
. Это, вероятно то, чего мы хотим, потому что мы в конечном итоге передадим получаемый аргумент в execute для spawn
. Мы также можем быть ещё более уверены, что
FnOnce
- это тот типаж, который мы хотим использовать, поскольку поток для выполнения запроса будет выполнять этот запрос только один раз, что соответствует параметру
Once в типаже
FnOnce
Параметр типа
F
также имеет ограничение типажа
Send и ограничение времени жизни 'static
, которые полезны в нашей ситуации: нам нужен
Send для передачи замыкания из одного потока в другой и 'static
, потому что мы не знаем, сколько времени займёт выполнение потока. Давайте создадим метод execute для
ThreadPool
, который будет принимать обобщённый параметр типа
F
со следующими ограничениями:
Файл: src/lib.rs
Мы по-прежнему используем
()
после
FnOnce потому что типаж
FnOnce представляет замыкание, которое не принимает параметров и возвращает единичный тип
()
. Также как при определении функций, тип возвращаемого значения может быть опущен в сигнатуре, но даже если у нас нет параметров, нам все равно нужны скобки.
pub fn spawn
(f: F) -> JoinHandle where
F:
FnOnce
() -> T,
F:
Send
+
'static
,
T:
Send
+
'static
, impl
ThreadPool {
// --snip-- pub fn execute
(&
self
, f: F) where
F:
FnOnce
() +
Send
+
'static
,
{
}
}
Опять же, это самая простая реализация метода execute
: она ничего не делает, мы только пытаемся сделать код компилируемым. Давайте проверим снова:
Сейчас мы получаем только предупреждения, что означает, что код компилируется! Но обратите внимание, если вы попробуете cargo run и сделаете запрос в браузере, вы увидите ошибки в браузере, которые мы видели в начале главы. Наша библиотека на самом деле ещё не вызывает замыкание, переданное в execute
!
Примечание: вы возможно слышали высказывание о языках со строгими компиляторами, таких как Haskell и Rust, которое звучит так: «Если код компилируется, то он работает». Но это высказывание не всегда верно. Наш проект компилируется, но абсолютно ничего не делает! Если бы мы создавали реальный,
законченный проект, это был бы хороший момент начать писать модульные тесты,
чтобы проверять, что код компилируется и имеет желаемое поведение.
Проверка количества потоков в new
Мы продолжим получать предупреждения, потому что мы ничего не делаем с параметрами для new и execute
. Давайте реализуем тела этих функций в соответствии с желаемым поведением. Для начала давайте подумаем о new
. Ранее мы выбирали без знаковый тип для параметра size
, потому что пул с отрицательным числом потоков не имеет смысла. Тем не менее, пул с нулевым значением для потоков также не имеет смысла, но ноль является совершенно корректным для типа usize
. Мы добавим код,
чтобы проверить, что size больше нуля, перед возвращением экземпляра
ThreadPool и
будем паниковать, если программа получит ноль, используя макрос assert!
, как показано в листинге 20-13.
Файл: src/lib.rs
$
cargo check
Checking hello v0.1.0 (file:///projects/hello)
Finished dev [unoptimized + debuginfo] target(s) in 0.24s