RU
EN
ES
ZH
FR
DE
PT
AR
HI
parking_lot для Rust: быстрые примитивы синхронизации
Библиотека parking_lot — это одна из самых популярных и производительных реализаций примитивов синхронизации для языка Rust. Она предоставляет альтернативу стандартным мьютексам (std::sync::Mutex), блокировкам чтения-записи (std::sync::RwLock) и условным переменным (std::sync::Condvar), но с существенно лучшей производительностью и меньшим размером кода.
Зачем нужна эта библиотека? Стандартные примитивы синхронизации в Rust, хотя и безопасны, имеют несколько недостатков: они используют системные вызовы (futex на Linux) даже в простых сценариях, требуют дополнительной памяти под флаги и имеют более сложную внутреннюю реализацию. parking_lot решает эти проблемы, используя более эффективные алгоритмы (например, оптимистичные блокировки и спин-лупы) и минимизируя накладные расходы. В результате код работает быстрее, особенно в многопоточных приложениях с высокой конкуренцией.
Библиотека полностью совместима с Rust и не требует unsafe-кода со стороны пользователя. Она предлагает те же интерфейсы, что и стандартная библиотека, но с дополнительными возможностями: Mutex с поддержкой таймаутов, RwLock с честной очередью, Once для ленивой инициализации и ReentrantMutex для реентерабельных блокировок. parking_lot — это стандарт де-факто для высокопроизводительных Rust-приложений, от веб-серверов до игровых движков.
Установка
Добавьте библиотеку в файл Cargo.toml вашего проекта:
[dependencies]
parking_lot = "0.12"Затем выполните команду:
cargo buildИли, если вы используете cargo add (начиная с Rust 1.62):
cargo add parking_lotПосле этого вы можете импортировать примитивы из библиотеки:
use parking_lot::{Mutex, RwLock, Condvar, Once, ReentrantMutex};Быстрый старт: минимальный рабочий пример
Вот простейший пример использования Mutex из parking_lot для защиты общего счётчика в многопоточной среде:
use parking_lot::Mutex;
use std::thread;
fn main() {
// Создаём мьютекс, оборачивающий значение
let counter = Mutex::new(0);
// Создаём 10 потоков, каждый увеличивает счётчик на 1
let mut handles = vec![];
for _ in 0..10 {
let handle = thread::spawn(|| {
// Захватываем блокировку — .lock() возвращает RAII-гард
let mut guard = counter.lock();
*guard += 1;
// guard автоматически освобождается при выходе из области видимости
});
handles.push(handle);
}
// Ждём завершения всех потоков
for handle in handles {
handle.join().unwrap();
}
// Читаем финальное значение
let final_value = *counter.lock();
println!("Финальное значение: {}", final_value); // Должно быть 10
}В этом примере counter.lock() возвращает MutexGuard, который автоматически освобождает блокировку при разрушении. В отличие от стандартного std::sync::Mutex, здесь не нужно обрабатывать ошибки (poisoning) — parking_lot по умолчанию не отравляет мьютекс при панике.
Основные методы и типы
Ниже приведены 8 ключевых методов/типов библиотеки parking_lot с сигнатурами, описанием и примерами.
1. Mutex::lock()
Сигнатура: pub fn lock(&self) -> MutexGuard<'_, T>
Описание: Захватывает мьютекс, блокируя текущий поток до тех пор, пока блокировка не будет получена. Возвращает RAII-гард, который автоматически освобождает блокировку при выходе из области видимости.
use parking_lot::Mutex;
let data = Mutex::new(42);
{
let mut guard = data.lock();
*guard += 1;
} // guard освобождается здесь
println!("{}", *data.lock()); // 432. Mutex::try_lock()
Сигнатура: pub fn try_lock(&self) -> Option<MutexGuard<'_, T>>
Описание: Пытается захватить мьютекс без блокировки. Возвращает Some(guard), если блокировка успешно получена, или None, если мьютекс уже захвачен другим потоком.
use parking_lot::Mutex;
let data = Mutex::new(0);
// Попытка захватить без блокировки
if let Some(mut guard) = data.try_lock() {
*guard = 1;
println!("Блокировка получена");
} else {
println!("Блокировка не получена (занята)");
}3. RwLock::read()
Сигнатура: pub fn read(&self) -> RwLockReadGuard<'_, T>
Описание: Захватывает блокировку на чтение. Множество потоков могут одновременно удерживать блокировку на чтение, но блокировка на запись будет ждать, пока все читатели не освободят её.
use parking_lot::RwLock;
let data = RwLock::new(10);
// Чтение
let guard = data.read();
println!("Значение: {}", *guard); // 10
// guard освобождается автоматически4. RwLock::write()
Сигнатура: pub fn write(&self) -> RwLockWriteGuard<'_, T>
Описание: Захватывает блокировку на запись. Только один поток может удерживать блокировку на запись, и никакие читатели не могут одновременно работать.
use parking_lot::RwLock;
let data = RwLock::new(10);
{
let mut guard = data.write();
*guard += 5;
} // блокировка освобождена
println!("{}", *data.read()); // 155. Condvar::wait()
Сигнатура: pub fn wait<T>(&self, mutex_guard: &mut MutexGuard<'_, T>)
Описание: Блокирует текущий поток до тех пор, пока не будет вызван notify_one() или notify_all() на той же условной переменной. При вызове автоматически освобождает мьютекс и повторно захватывает его после пробуждения.
use parking_lot::{Mutex, Condvar};
let mutex = Mutex::new(false);
let condvar = Condvar::new();
let mut guard = mutex.lock();
// Ждём, пока условие не станет true
while !*guard {
condvar.wait(&mut guard);
}
println!("Условие выполнено!");6. Condvar::notify_one()
Сигнатура: pub fn notify_one(&self)
Описание: Пробуждает один поток, ожидающий на этой условной переменной. Если ни один поток не ожидает, вызов игнорируется.
use parking_lot::{Mutex, Condvar};
use std::thread;
let mutex = Mutex::new(false);
let condvar = Condvar::new();
thread::spawn(move || {
thread::sleep(std::time::Duration::from_millis(100));
let mut guard = mutex.lock();
*guard = true;
condvar.notify_one(); // пробуждаем ожидающий поток
});
let mut guard = mutex.lock();
while !*guard {
condvar.wait(&mut guard);
}
println!("Поток пробуждён!");7. Once::call_once()
Сигнатура: pub fn call_once<F>(&self, f: F) where F: FnOnce()
Описание: Выполняет переданное замыкание ровно один раз, независимо от того, сколько потоков вызывают этот метод. Используется для ленивой инициализации глобальных данных.
use parking_lot::Once;
use std::sync::atomic::{AtomicUsize, Ordering};
static INIT: Once = Once::new();
static COUNTER: AtomicUsize = AtomicUsize::new(0);
fn init() {
COUNTER.fetch_add(1, Ordering::SeqCst);
}
fn main() {
// 10 потоков пытаются вызвать init, но он выполнится только один раз
let mut handles = vec![];
for _ in 0..10 {
handles.push(std::thread::spawn(|| {
INIT.call_once(|| {
init();
});
}));
}
for h in handles { h.join().unwrap(); }
println!("COUNTER = {}", COUNTER.load(Ordering::SeqCst)); // 1
}8. ReentrantMutex::lock()
Сигнатура: pub fn lock(&self) -> ReentrantMutexGuard<'_, T>
Описание: Захватывает реентерабельный мьютекс. В отличие от обычного Mutex, один и тот же поток может захватить этот мьютекс несколько раз без взаимоблокировки. Полезно для рекурсивных функций, работающих с общими данными.
use parking_lot::ReentrantMutex;
let mutex = ReentrantMutex::new(0);
fn recursive_increment(mutex: &ReentrantMutex, depth: u32) {
if depth == 0 { return; }
let mut guard = mutex.lock();
*guard += 1;
recursive_increment(mutex, depth - 1); // рекурсивный вызов — блокировка уже удержана
}
recursive_increment(&mutex, 5);
println!("{}", *mutex.lock()); // 5 Реальный практический пример: пул потоков с балансировкой нагрузки
В этом примере мы создадим простой пул потоков, который распределяет задачи между рабочими потоками, используя Mutex, Condvar и RwLock из parking_lot. Пул будет принимать замыкания и выполнять их в отдельных потоках, с возможностью динамической настройки количества рабочих.
use parking_lot::{Mutex, Condvar, RwLock};
use std::sync::Arc;
use std::thread;
use std::time::Duration;
// Тип задачи — замыкание, которое можно отправить между потоками
type Job = Box;
struct ThreadPool {
// Очередь задач, защищённая мьютексом
jobs: Arc>>,
// Условная переменная для уведомления рабочих о новых задачах
condvar: Arc,
// Количество рабочих потоков (защищено RwLock для параллельного чтения)
num_workers: Arc>,
// Флаг остановки
stop: Arc>,
}
impl ThreadPool {
fn new(initial_workers: usize) -> Self {
let pool = ThreadPool {
jobs: Arc::new(Mutex::new(Vec::new())),
condvar: Arc::new(Condvar::new()),
num_workers: Arc::new(RwLock::new(initial_workers)),
stop: Arc::new(Mutex::new(false)),
};
// Запускаем начальное количество рабочих потоков
for _ in 0..initial_workers {
pool.spawn_worker();
}
pool
}
fn spawn_worker(&self) {
let jobs = Arc::clone(&self.jobs);
let condvar = Arc::clone(&self.condvar);
let stop = Arc::clone(&self.stop);
let num_workers = Arc::clone(&self.num_workers);
thread::spawn(move || {
loop {
// Проверяем флаг остановки
if *stop.lock() {
break;
}
// Пытаемся получить задачу из очереди
let job = {
let mut queue = jobs.lock();
// Если очередь пуста и не остановлено — ждём
while queue.is_empty() && !*stop.lock() {
condvar.wait(&mut queue);
}
// Если остановлено и очередь пуста — выходим
if *stop.lock() && queue.is_empty() {
break;
}
queue.pop()
};
// Выполняем задачу, если она есть
if let Some(job) = job {
job();
}
}
// Уменьшаем счётчик рабочих при завершении
let mut workers = num_workers.write();
*workers -= 1;
});
}
fn execute(&self, f: F)
where
F: FnOnce() + Send + 'static,
{
let mut queue = self.jobs.lock();
queue.push(Box::new(f));
// Уведомляем один ожидающий поток
self.condvar.notify_one();
}
fn set_num_workers(&self, new_count: usize) {
let current = *self.num_workers.read();
if new_count > current {
// Добавляем новых рабочих
for _ in 0..(new_count - current) {
self.spawn_worker();
}
} else if new_count < current {
// Уменьшаем количество — просто устанавливаем флаг остановки для лишних
// (в реальном проекте нужна более сложная логика)
println!("Уменьшение рабочих потоков не реализовано полностью");
}
*self.num_workers.write() = new_count;
}
fn stop(&self) {
*self.stop.lock() = true;
// Пробуждаем все ожи 
