Материал: Язык программирования Rust

Листинг 20-13: Реализация
ThreadPool::new
с паникой, если
size
равен нулю
Мы добавили документации в
ThreadPool с помощью комментариев. Обратите внимание, мы следовали хорошим практикам документирования, добавив раздел, в котором указывается ситуация при которой функция может паниковать как обсуждалось в главе 14. Попробуйте запустить cargo doc --open и кликнуть структуру
ThreadPool
,
чтобы увидеть как выглядит сгенерированная документация для new
!
Вместо добавления макроса assert!
, как мы здесь сделали, мы могли бы описать у new возвращать
Result как мы делали в
Config::new проекта ввода/вывода в коде 12-9. Но сейчас мы решили, что попытка создания пула потоков без любого указания количества потоков должно быть не восстанавливаемой ошибкой. Если вы чувствуете себя честолюбивым, попробуйте написать версию new со следующей сигнатурой, чтобы сравнить обе версии:
Создание места для хранения потоков
Теперь у нас есть способ узнать, что задано допустимое число потоков для хранения в пуле и мы можем создать эти потоки и сохранить их в структуре
ThreadPool перед её
возвратом. Но как мы "храним" поток? Давайте ещё раз посмотрим на сигнатуру thread::spawn
:
Функция spawn возвращает тип
JoinHandle
, где
T
является типом, который возвращает замыкание. Давайте попробуем использовать
JoinHandle и посмотрим, что 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
}
// --snip--
} pub fn new
(size: usize
) ->
Result
{ pub fn spawn
(f: F) -> JoinHandle where
F:
FnOnce
() -> T,
F:
Send
+
'static
,
T:
Send
+
'static
,

произойдёт. В нашем случае замыкания, которые мы передаём пулу потоков, будут обрабатывать соединение и ничего не будут возвращать, поэтому
T
будет единичным
(unit) типом
()
Листинг 20-14 скомпилируется, но пока не создаёт потоков. Мы изменили объявление
ThreadPool
, чтобы оно содержало вектор экземпляров thread::JoinHandle<()>
,
инициализировали вектор с размером size
, установили цикл for
, который будет запускать некоторый код для создания потоков и вернули экземпляр
ThreadPool содержащий потоки.
Файл: src/lib.rs
Листинг 20-14: Создание вектора в
ThreadPool
для хранения потоков
Мы добавили std::thread в область видимости библиотечного крейта, потому что мы используем thread::JoinHandle в качестве типа элементов вектора в
ThreadPool
После получения корректного значения size, наш
ThreadPool создаёт новый вектор,
который может содержать size элементов. В этой книге мы ещё не использовали функцию with_capacity
, которая выполняет ту же задачу что и
Vec::new
, но с важным отличием: она заранее выделяет указанную память в векторе. Поскольку мы знаем, что нам нужно хранить size элементов в векторе, выполнение этого выделения немного более эффективно, чем использование
Vec::new
, который изменяет размеры при вставке элементов.
Когда вы снова запустите cargo check
, вы получите ещё несколько предупреждений, но все должно завершится успехом.
use std::thread; pub struct
ThreadPool
{ threads:
Vec
<:joinhandle>>,
} impl
ThreadPool {
// --snip-- pub fn new
(size: usize
) -> ThreadPool { assert!
(size >
0
); let mut threads =
Vec
::with_capacity(size); for
_ in
0
..size {
// create some threads and store them in the vector
}
ThreadPool { threads }
}
// --snip--
}

Структура Worker ответственная за отправку кода из ThreadPool в поток
Мы оставили комментарий относительно создания потоков в цикле for кода 20-14.
Здесь мы рассмотрим, как мы на самом деле создаём потоки. Стандартная библиотека предоставляет thread::spawn как способ создания потоков, а thread::spawn ожидает получить некоторый код, который поток должен запустить как только поток создан.
Однако в нашем случае мы хотим создать потоки и заставить их ждать код, который мы отправим им позже. Реализация потоков в стандартной библиотеке не имеет какого то способа это сделать, мы должны реализовать это вручную.
Мы будем реализовывать это поведение с помощью новой структуры данных между
ThreadPool и потоками, которая будет управлять этим новым поведением. Мы назовём эту структуру данных
Worker
, что является общим термином в реализации пулов.
Подумайте о людях, работающих на кухне в ресторане: рабочие ждут пока не поступят заказы от клиентов, а затем они несут ответственность за принятие этих заказов и их выполнение.
Вместо хранения вектора
JoinHandle<()>
в пуле потоков, мы будем сохранять экземпляры структуры
Worker
. Каждый
Worker будет хранить один экземпляр
JoinHandle<()>
. Затем мы реализуем метод у
Worker
, который берет код замыкания для запуска и отправляет его в уже запущенный поток для выполнения. Мы также назначим каждому работнику id
, чтобы мы могли различать разных работников в пуле при ведении журнала или отладке.
Давайте внесём изменения в последовательность действий, которая выполняется при создании
ThreadPool
. Мы реализуем код, который отправляет замыкание в поток после того, как мы настроили
Worker следующим образом:
1. Определим структуру
Worker
(работник), которая содержит id и
JoinHandle<()>
2. Изменим
ThreadPool
, чтобы он содержал вектор экземпляров
Worker
3. Определим функцию
Worker::new
, которая принимает номер id и возвращает экземпляр
Worker
, который содержит id и поток, порождённый пустым замыканием.
4. В
ThreadPool::new используем счётчик цикла for для генерации id
, создаём новый
Worker с этим id и сохраняем экземпляр "работника" в вектор.
Если вы готовы принять вызов, попробуйте реализовать эти изменения самостоятельно,
прежде чем смотреть код листинге 20-15.
Готовы? Вот листинг 20-15 с одним из способов сделать предыдущие модификации.
Файл: src/lib.rs

Листинг 20-15: Изменение
ThreadPool
для хранения экземпляров
Worker
вместо непосредственного
хранения потоков
Мы изменили имя поля в
ThreadPool с threads на workers
, потому что теперь оно содержит экземпляры
Worker вместо экземпляров
JoinHandle<()>
. Мы используем счётчик в цикле for в качестве аргумента для
Worker::new и сохраняем каждый новый
Worker в векторе с именем workers
Внешний код (вроде нашего сервера в src/bin/main.rs) не должен знать подробности реализации касательно использования структуры
Worker внутри
ThreadPool
, поэтому мы делаем структуру
Worker и её новую функцию new приватными. Функция
Worker::new использует заданный нами id и сохраняет экземпляр
JoinHandle<()>
,
который создаётся путём порождение нового потока с пустым замыканием.
Этот код скомпилируется и будет хранить количество экземпляров
Worker
, которое мы указали в качестве аргумента функции
ThreadPool::new
. Но мы все ещё не обрабатываем use std::thread; pub struct
ThreadPool
{ workers:
Vec
,
} impl
ThreadPool {
// --snip-- 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 }
}
// --snip--
} struct
Worker
{ id: usize
, thread: thread::JoinHandle<()>,
} impl
Worker { fn new
(id: usize
) -> Worker { let thread = thread::spawn(|| {});
Worker { id, thread }
}
}

замыкание, которое мы получаем в методе execute
. Давайте взглянем на то, как это сделать.

Листинг 20-16: Модификация
ThreadPool
для хранения отправляющей части канала, который отправляет
экземпляры
Job
В
ThreadPool::new мы создаём наш новый канал и пул содержащий отправляющую сторону. Код успешно скомпилируется все ещё с предупреждениями.
Давайте попробуем передавать принимающую сторону канала каждому "работнику"
(структуре woker), когда пул потоков создаёт канал. Мы знаем, что хотим использовать получающую часть канала в потоке порождаемым "работником", поэтому мы будем ссылаться на параметр receiver в замыкании. Код 20-17 пока не компилируется.
Файл: src/lib.rs use std::{sync::mpsc, thread}; pub struct
ThreadPool
{ workers:
Vec
, sender: mpsc::Sender,
} struct
Job
; impl
ThreadPool {
// --snip-- 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 }
}
// --snip--
}

Листинг 20-17: Передача принимающей части канала "работнику"
Мы внесли несколько небольших и простых изменений: мы передаём принимающую часть канала в
Worker::new
, а затем используем его внутри замыкания.
При попытке проверить код, мы получаем ошибку:
Код пытается передать receiver в несколько экземпляров
Worker
. Это не будет работать, как вы помните из главы 16: реализация канала предоставляемая Rust,
impl
ThreadPool {
// --snip-- 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 }
}
// --snip--
}
// --snip-- impl
Worker { fn new
(id: usize
, receiver: mpsc::Receiver) -> Worker { let thread = thread::spawn(|| { receiver;
});
Worker { id, thread }
}
}
$
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`, which does not implement the `Copy` trait
26 | workers.push(Worker::new(id, receiver));
| ^^^^^^^^ value moved here, in previous iteration of loop
For more information about this error, try `rustc --explain E0382`. error: could not compile `hello` due to previous error

является моделью несколько производителей (multiple producer), один потребитель
(single consumer). Это означает, что мы не можем просто клонировать принимающую часть канала для исправления этого кода. Даже если бы мы это могли, это не техника которую мы хотели бы использовать; вместо этого мы хотим распределить задачи среди потоков, разделяя один receiver среди всех "работников".
Кроме того, удаление задачи из очереди канала включает изменение receiver
, поэтому потокам необходим безопасный способ делиться и изменять receiver
, в противном случае мы можем получить условия гонки (как описано в главе 16).
Вспомните умные указатели, которые обсуждались в главе 16: чтобы делиться владением между несколькими потоками и позволить потокам изменять значение, нам нужно использовать тип
Arc>
. Тип
Arc позволит нескольким "работникам" владеть получателем (receiver), а
Mutex гарантирует что только один "работник" получит задание
(job) от получателя в один момент времени. Листинг 20-18 показывает изменения,
которые мы должны сделать.
Файл: src/lib.rs use std::{ sync::{mpsc, Arc, Mutex}, thread,
};
// --snip-- impl
ThreadPool {
// --snip-- 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 }
}
// --snip--
}
// --snip-- impl
Worker { fn new
(id: usize
, receiver: Arc>>) -> Worker {
// --snip--
}
}