Библиотека crossbeam для Rust: инструменты конкурентного программирования

Добро пожаловать на PythonLib! Сегодня мы погрузимся в мир конкурентного программирования на Rust с помощью библиотеки crossbeam. Это одна из ключевых экосистемных библиотек, которая предоставляет мощные и безопасные инструменты для работы с многопоточностью, каналами, эпохальной сборкой мусора и безблокировочными структурами данных.

1. Что это и зачем нужна

Crossbeam — это коллекция инструментов для конкурентного и параллельного программирования на Rust. Она создана для того, чтобы сделать многопоточное программирование безопасным, эффективным и удобным. В отличие от стандартной библиотеки Rust, которая предоставляет базовые примитивы синхронизации (Mutex, Arc, каналы mpsc), crossbeam предлагает более продвинутые и высокопроизводительные альтернативы.

Зачем нужна crossbeam? Во-первых, она решает проблему ложного совместного использования (false sharing) в кэшах процессора через свои структуры данных. Во-вторых, она предоставляет эпохальную сборку мусора (epoch-based reclamation) для безопасного управления памятью в безблокировочных структурах. В-третьих, её каналы (channels) являются многопотребительскими и многопроизводительными (MPMC), в отличие от однопотребительских каналов из std. Наконец, crossbeam содержит множество полезных утилит, таких как WaitGroup, Barrier и ShardedLock, которые упрощают написание сложных конкурентных алгоритмов.

Эта библиотека активно используется в промышленных проектах, таких как Tokio, Rayon и TiKV. Если вы пишете высоконагруженные серверы, базы данных или любые другие системы, требующие максимальной производительности на многоядерных процессорах, crossbeam станет вашим незаменимым помощником.

2. Установка

Crossbeam — это библиотека для языка Rust, поэтому она устанавливается через менеджер пакетов Cargo. Добавьте следующую строку в ваш файл Cargo.toml:

[dependencies]
crossbeam = "0.8"

Затем выполните команду:

cargo build

Если вам нужны только определённые компоненты (например, только каналы или только эпохальный сборщик), вы можете использовать под-крейты:

[dependencies]
crossbeam-channel = "0.5"
crossbeam-epoch = "0.9"
crossbeam-utils = "0.8"
crossbeam-deque = "0.8"
crossbeam-queue = "0.3"

Однако для большинства случаев достаточно просто добавить crossbeam = "0.8".

3. Быстрый старт — минимальный рабочий пример

Давайте начнём с простейшего примера: создадим два потока, которые обмениваются данными через многопотребительский канал crossbeam.

use crossbeam::channel;
use std::thread;

fn main() {
    // Создаём канал с неограниченным буфером
    let (sender, receiver) = channel::unbounded();

    // Поток-отправитель
    thread::spawn(move || {
        sender.send("Привет из crossbeam!").unwrap();
        println!("Сообщение отправлено");
    });

    // Поток-получатель
    thread::spawn(move || {
        let msg = receiver.recv().unwrap();
        println!("Получено: {}", msg);
    });

    // Даём потокам время выполниться
    thread::sleep(std::time::Duration::from_millis(100));
}

Этот пример демонстрирует базовую механику: channel::unbounded() создаёт пару (отправитель, получатель). Отправитель посылает строку, получатель её принимает. В отличие от std::sync::mpsc, здесь может быть несколько отправителей и несколько получателей.

4. Основные методы/функции/классы

Рассмотрим 8 ключевых компонентов библиотеки crossbeam с подробными примерами.

4.1. channel::unbounded()

Сигнатура: pub fn unbounded<T>() -> (Sender<T>, Receiver<T>)

Что делает: Создаёт неограниченный канал (буфер может расти бесконечно). Возвращает пару отправитель/получатель. Поддерживает множественных отправителей и получателей (MPMC).

use crossbeam::channel;

fn main() {
    let (s, r) = channel::unbounded();
    s.send(42).unwrap();
    s.send(43).unwrap();
    println!("{}", r.recv().unwrap()); // 42
    println!("{}", r.recv().unwrap()); // 43
}

4.2. channel::bounded(cap: usize)

Сигнатура: pub fn bounded<T>(cap: usize) -> (Sender<T>, Receiver<T>)

Что делает: Создаёт ограниченный канал с фиксированной ёмкостью буфера. Если буфер заполнен, отправка будет блокироваться до появления места. Полезно для ограничения потребления памяти и создания backpressure.

use crossbeam::channel;
use std::thread;

fn main() {
    let (s, r) = channel::bounded(2);
    
    thread::spawn(move || {
        for i in 0..5 {
            s.send(i).unwrap();
            println!("Отправлено: {}", i);
        }
    });

    thread::sleep(std::time::Duration::from_millis(100));
    for _ in 0..5 {
        println!("Получено: {}", r.recv().unwrap());
    }
}

4.3. Receiver::recv()

Сигнатура: pub fn recv(&self) -> Result<T, RecvError>

Что делает: Блокирует текущий поток до тех пор, пока не будет получено сообщение из канала. Возвращает Err(RecvError), если все отправители были уничтожены.

use crossbeam::channel;

fn main() {
    let (s, r) = channel::unbounded();
    s.send("hello").unwrap();
    drop(s); // Уничтожаем отправителя
    match r.recv() {
        Ok(msg) => println!("Получено: {}", msg),
        Err(_) => println!("Канал закрыт"),
    }
}

4.4. Sender::send()

Сигнатура: pub fn send(&self, msg: T) -> Result<(), SendError<T>>

Что делает: Отправляет сообщение в канал. Для ограниченных каналов может блокироваться. Возвращает ошибку, если все получатели уничтожены.

use crossbeam::channel;

fn main() {
    let (s, r) = channel::bounded(1);
    s.send("data").unwrap();
    // Если раскомментировать, send заблокируется, так как буфер полон
    // s.send("data2").unwrap();
    drop(r);
    // Теперь send вернёт ошибку
    match s.send("lost") {
        Ok(_) => println!("Отправлено"),
        Err(e) => println!("Ошибка: {:?}", e),
    }
}

4.5. Receiver::try_recv()

Сигнатура: pub fn try_recv(&self) -> Result<T, TryRecvError>

Что делает: Неблокирующая попытка получить сообщение. Возвращает Err(TryRecvError::Empty), если канал пуст, или Err(TryRecvError::Disconnected), если канал закрыт.

use crossbeam::channel;

fn main() {
    let (s, r) = channel::unbounded();
    s.send(10).unwrap();
    println!("{:?}", r.try_recv()); // Ok(10)
    println!("{:?}", r.try_recv()); // Err(Empty)
    drop(s);
    println!("{:?}", r.try_recv()); // Err(Disconnected)
}

4.6. crossbeam::sync::WaitGroup

Сигнатура: pub struct WaitGroup { /* fields omitted */ }

Методы: new(), add(count: usize), done(), wait()

Что делает: Аналог sync.WaitGroup из Go. Позволяет дождаться завершения группы потоков. Каждый поток вызывает done() при завершении, а основной поток вызывает wait() для блокировки до завершения всех.

use crossbeam::sync::WaitGroup;
use std::thread;

fn main() {
    let wg = WaitGroup::new();
    let workers = 5;

    for i in 0..workers {
        let wg = wg.clone();
        thread::spawn(move || {
            println!("Работник {} начал", i);
            thread::sleep(std::time::Duration::from_millis(100));
            println!("Работник {} закончил", i);
            drop(wg); // эквивалент done()
        });
    }

    wg.wait();
    println!("Все работники завершили работу");
}

4.7. crossbeam::sync::Barrier

Сигнатура: pub fn new(n: usize) -> Barrier

Методы: wait()

Что делает: Барьер синхронизации. Блокирует n потоков до тех пор, пока все они не вызовут wait(). После этого все потоки продолжают выполнение одновременно. Полезно для синхронизации фаз вычислений.

use crossbeam::sync::Barrier;
use std::thread;

fn main() {
    let barrier = Barrier::new(3);
    let mut handles = vec![];

    for i in 0..3 {
        let b = barrier.clone();
        handles.push(thread::spawn(move || {
            println!("Поток {} готовится", i);
            thread::sleep(std::time::Duration::from_millis(i * 50));
            b.wait();
            println!("Поток {} продолжает после барьера", i);
        }));
    }

    for h in handles {
        h.join().unwrap();
    }
}

4.8. crossbeam::atomic::AtomicCell

Сигнатура: pub struct AtomicCell<T> { /* fields omitted */ }

Методы: new(val: T), load(&self, order: Ordering) -> T, store(&self, val: T, order: Ordering), swap(&self, val: T, order: Ordering) -> T, compare_exchange(...)

Что делает: Атомарная ячейка для любого типа, который поддерживает Copy. В отличие от std::sync::atomic, который поддерживает только целые числа и указатели, AtomicCell работает с любыми копируемыми типами, включая пользовательские структуры.

use crossbeam::atomic::AtomicCell;
use std::sync::atomic::Ordering;

#[derive(Copy, Clone, Debug)]
struct Point {
    x: i32,
    y: i32,
}

fn main() {
    let cell = AtomicCell::new(Point { x: 0, y: 0 });
    
    // Атомарная загрузка
    let p = cell.load(Ordering::SeqCst);
    println!("Точка: {:?}", p);
    
    // Атомарная замена
    let old = cell.swap(Point { x: 10, y: 20 }, Ordering::SeqCst);
    println!("Старая точка: {:?}", old);
    println!("Новая точка: {:?}", cell.load(Ordering::SeqCst));
}

5. Реальный практический пример

Давайте напишем полноценный пример: пул воркеров для параллельной обработки чисел. Мы будем использовать crossbeam-channel для распределения задач и WaitGroup для синхронизации.

use crossbeam::channel;
use crossbeam::sync::WaitGroup;
use std::thread;
use std::time::Duration;

/// Функция-воркер: получает задачу, обрабатывает её и отправляет результат
fn worker(
    id: usize,
    tasks: channel::Receiver,
    results: channel::Sender,
    wg: WaitGroup,
) {
    // Обрабатываем задачи, пока канал не закроется
    while let Ok(task) = tasks.recv() {
        // Симуляция сложной обработки
        let result = task * task; // возводим в квадрат
        println!("Воркер {}: {}^2 = {}", id, task, result);
        thread::sleep(Duration::from_millis(50));
        
        // Отправляем результат
        if results.send(result).is_err() {
            println!("Воркер {}: канал результатов закрыт", id);
            break;
        }
    }
    println!("Воркер {} завершил работу", id);
    drop(wg); // Уменьшаем счётчик WaitGroup
}

fn main() {
    // Создаём каналы для задач и результатов
    let (task_sender, task_receiver) = channel::bounded(10);
    let (result_sender, result_receiver) = channel::unbounded();
    
    // Создаём WaitGroup для ожидания всех воркеров
    let wg = WaitGroup::new();
    let num_workers = 4;
    
    // Запускаем воркеров
    for i in 0..num_workers {
        let wg = wg.clone();
        let tasks = task_receiver.clone();
        let results = result_sender.clone();
        
        thread::spawn(move || {
            worker(i, tasks, results, wg);
        });
    }
    
    // Отправляем задачи