Многопоточность в Rust: Fearless concurrency и практические примеры

let data = vec![1, 2, 3];
 
std::thread::spawn(move || {
    // Теперь поток владеет data
    println!("Данные в потоке: {:?}", data);
});
 
// Эта строка вызовет ошибку компиляции, так как data перемещена в поток
// println!("Данные в главном потоке: {:?}", data);

let mut counter = 0;
 
// Попытка передать ссылку на counter в поток
let handle = std::thread::spawn(|| {
    // Эта строка вызовет ошибку компиляции
    counter += 1;
});
 
// Главный поток продолжает работать с counter
counter += 1;
 
handle.join().unwrap();

let data = vec![1, 2, 3];
 
let handle = std::thread::spawn(|| {
    println!("Данные: {:?}", data); // Ошибка: data не принадлежит потоку
});
 
println!("Оригинальные данные: {:?}", data);
handle.join().unwrap();

let data = vec![1, 2, 3];
 
let handle = std::thread::spawn(move || {
    println!("Данные: {:?}", data); // Теперь всё в порядке
});
 
// println!("Оригинальные данные: {:?}", data); // Теперь эта строка вызовет ошибку
handle.join().unwrap();

// Примеры Send и Sync типов
// Vec<T> является Send и Sync, если T является Send и Sync
let v = vec![1, 2, 3]; // Vec<i32> - Send и Sync
 
// Rc<T> не является Send или Sync, поскольку его счетчик ссылок не потокобезопасен
use std::rc::Rc;
let rc = Rc::new(42); // Rc<i32> - не Send и не Sync

fn main() {
    let handle;
    {
        let data = vec![1, 2, 3];
        handle = std::thread::spawn(|| {
            // Ошибка: data может быть освобождена до выполнения потока
            println!("Данные: {:?}", data);
        });
    } // data уничтожается здесь
    handle.join().unwrap(); // Поток может попытаться использовать уже уничтоженные данные
}

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
 
struct Counter {
    count: AtomicUsize,
}
 
impl Counter {
    fn new() -> Self {
        Counter {
            count: AtomicUsize::new(0),
        }
    }
 
    fn increment(&self) {
        self.count.fetch_add(1, Ordering::SeqCst);
    }
 
    fn get(&self) -> usize {
        self.count.load(Ordering::SeqCst)
    }
}

use std::sync::mpsc;
use std::thread;
 
fn main() {
    // Создаем канал
    let (tx, rx) = mpsc::channel();
    
    // Клонируем отправителя для использования в нескольких потоках
    let tx1 = tx.clone();
    
    // Первый поток
    thread::spawn(move || {
        tx.send("Сообщение из первого потока").unwrap();
    });
    
    // Второй поток
    thread::spawn(move || {
        tx1.send("Сообщение из второго потока").unwrap();
    });
    
    // Получаем сообщения в основном потоке
    for _ in 0..2 {
        println!("Получено: {}", rx.recv().unwrap());
    }
}

// Стандартные типы вроде i32, bool, String и т.д. являются Send и Sync
struct MyData {
    id: i32,
    name: String,
    active: bool,
}
// MyData автоматически получает реализации Send и Sync

use std::rc::Rc;
use std::cell::RefCell;
 
// Rc не является Send или Sync
let shared_data = Rc::new(42);
 
// RefCell не является Sync
let cell = RefCell::new(42);

use std::marker::PhantomData;
 
// Тип, который не может быть отправлен между потоками
struct NotSendable {
    _marker: PhantomData<*const ()>, // сырой указатель не является Send
}

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::thread;
 
fn main() {
    // Создаем разделяемый атомарный счетчик
    let counter = Arc::new(AtomicUsize::new(0));
    let mut handles = vec![];
    
    for _ in 0..10 {
        let counter_clone = Arc::clone(&counter);
        
        // Инкрементируем счетчик в 10 параллельных потоках
        let handle = thread::spawn(move || {
            for _ in 0..100 {
                counter_clone.fetch_add(1, Ordering::SeqCst);
            }
        });
        
        handles.push(handle);
    }
    
    // Дожидаемся завершения всех потоков
    for handle in handles {
        handle.join().unwrap();
    }
    
    println!("Итоговое значение: {}", counter.load(Ordering::SeqCst));
}

use std::cell::Cell;
 
struct Counter {
    value: Cell<usize>,
}
 
impl Counter {
    fn new() -> Self {
        Counter { value: Cell::new(0) }
    }
    
    fn increment(&self) {
        let old_value = self.value.get();
        self.value.set(old_value + 1);
    }
    
    fn get(&self) -> usize {
        self.value.get()
    }
}

use crossbeam::queue::ArrayQueue;
use std::sync::Arc;
use std::thread;
 
fn main() {
    let queue = Arc::new(ArrayQueue::new(100));
    let mut handles = vec![];
    
    // Потоки-производители
    for i in 0..5 {
        let q = Arc::clone(&queue);
        let handle = thread::spawn(move || {
            for j in 0..10 {
                q.push(i * 100 + j).unwrap();
            }
        });
        handles.push(handle);
    }
    
    // Поток-потребитель
    let consumer = {
        let q = Arc::clone(&queue);
        thread::spawn(move || {
            let mut sum = 0;
            for _ in 0..50 {
                if let Some(item) = q.pop() {
                    sum += item;
                }
            }
            sum
        })
    };
    
    // Дожидаемся завершения производителей
    for handle in handles {
        handle.join().unwrap();
    }
    
    // Получаем результат от потребителя
    let sum = consumer.join().unwrap();
    println!("Сумма: {}", sum);
}

use std::thread;
 
fn main() {
    // Создаем новый поток
    let handle = thread::spawn(|| {
        println!("Привет из параллельного мира!");
    });
    
    println!("Основной поток продолжает работу");
    
    // Ждем завершения созданного потока
    handle.join().unwrap();
}

let builder = thread::Builder::new()
    .name("важный_вычислитель".to_string())
    .stack_size(4 * 1024 * 1024); // 4 МБ стека
 
let handle = builder.spawn(|| {
    // Выполняем сложные вычисления...
}).unwrap();

use std::thread; // Начиная с Rust 1.63 scoped threads доступны в стандартной библиотеке
 
fn main() {
    let data = vec![1, 2, 3, 4, 5];
    
    thread::scope(|s| {
        // Создаем поток, который может безопасно заимствовать data
        s.spawn(|| {
            println!("Первый элемент: {}", data[0]);
        });
        
        // Можно создать несколько потоков
        s.spawn(|| {
            println!("Длина вектора: {}", data.len());
        });
    }); // Здесь все потоки гарантированно завершены
    
    // data все еще доступна здесь
    println!("Данные после обработки: {:?}", data);
}

use rayon::prelude::*;
 
fn main() {
    let numbers = vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 10];
    
    // Обычный последовательный итератор
    let sum_sequential: i32 = numbers.iter().map(|&n| n * n).sum();
    
    // Параллельный итератор (просто заменяем iter() на par_iter())
    let sum_parallel: i32 = numbers.par_iter().map(|&n| n * n).sum();
    
    assert_eq!(sum_sequential, sum_parallel);
}

use rayon::prelude::*;
 
fn fibonacci(n: u32) -> u64 {
    if n <= 1 {
        return n as u64;
    }
    
    // Для больших чисел используем параллельное вычисление
    if n > 20 {
        let (a, b) = rayon::join(
            || fibonacci(n - 1),
            || fibonacci(n - 2)
        );
        return a + b;
    }
    
    // Для маленьких чисел последовательное вычисление эффективнее
    fibonacci(n - 1) + fibonacci(n - 2)
}

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::thread;
 
fn main() {
    let counter = Arc::new(AtomicUsize::new(0));
    let mut handles = Vec::new();
    
    for _ in 0..10 {
        let counter_clone = Arc::clone(&counter);
        let handle = thread::spawn(move || {
            for _ in 0..1000 {
                // Атомарно увеличиваем счетчик на 1
                counter_clone.fetch_add(1, Ordering::SeqCst);
            }
        });
        handles.push(handle);
    }
    
    for handle in handles {
        handle.join().unwrap();
    }
    
    println!("Финальное значение: {}", counter.load(Ordering::SeqCst));
    assert_eq!(counter.load(Ordering::SeqCst), 10_000);
}

use crossbeam_channel::{bounded, select};
use std::time::Duration;
 
fn main() {
    let (s1, r1) = bounded(1); // Ограниченный канал с буфером размером 1
    let (s2, r2) = bounded(1);
    
    // Отправляем сообщения в отдельных потоках
    std::thread::spawn(move || {
        std::thread::sleep(Duration::from_millis(100));
        s1.send("сообщение 1").unwrap();
    });
    
    std::thread::spawn(move || {
        std::thread::sleep(Duration::from_millis(50));
        s2.send("сообщение 2").unwrap();
    });
    
    // Получаем сообщение из канала, который первым будет готов
    select! {
        recv(r1) -> msg => println!("Получено: {}", msg.unwrap()),
        recv(r2) -> msg => println!("Получено: {}", msg.unwrap()),
    }
}

use crossbeam_skiplist::SkipSet;
 
fn main() {
    let set = SkipSet::new();
    
    // Вставляем элементы
    set.insert(1);
    set.insert(2);
    
    // Проверяем наличие элементов
    assert!(set.contains(&1));
    assert!(!set.contains(&3));
}

#[repr(align(64))] // Выравнивание на границу кеш-линии (обычно 64 байта)
struct PaddedCounter {
    value: AtomicUsize,
    // Padding для заполнения полной кеш-линии
    _padding: [u8; 56], // 64 - размер AtomicUsize
}

use std::thread;
 
fn process_chunks(data: Vec<u32>) -> u32 {
    let chunk_size = data.len() / 4; // Для 4 потоков
    let mut handles = Vec::new();
    
    // Разбиваем данные на части и обрабатываем каждую в отдельном потоке
    for chunk in data.chunks(chunk_size) {
        let chunk_vec = chunk.to_vec();
        let handle = thread::spawn(move || {
            chunk_vec.iter().sum::<u32>()
        });
        handles.push(handle);
    }
    
    // Собираем результаты
    handles.into_iter()
        .map(|h| h.join().unwrap())
        .sum()
}

use std::sync::mpsc;
use std::thread;
 
fn pipeline_processing(data: Vec<u32>) -> Vec<u32> {
    let (tx1, rx1) = mpsc::channel();
    let (tx2, rx2) = mpsc::channel();
    
    // Поток 1: квадраты чисел
    thread::spawn(move || {
        for value in data {
            tx1.send(value * value).unwrap();
        }
    });
    
    // Поток 2: прибавление 10
    thread::spawn(move || {
        for value in rx1 {
            tx2.send(value + 10).unwrap();
        }
    });
    
    // Собираем результаты в основном потоке
    rx2.iter().collect()
}

use threadpool::ThreadPool;
use std::sync::mpsc::channel;
 
fn main() {
    let pool = ThreadPool::new(4); // Пул из 4 потоков
    let (tx, rx) = channel();
    
    // Отправляем 10 задач
    for i in 0..10 {
        let tx = tx.clone();
        pool.execute(move || {
            // Выполняем работу
            let result = i * i;
            tx.send(result).unwrap();
        });
    }
    
    // Получаем результаты
    let results: Vec<_> = (0..10).map(|_| rx.recv().unwrap()).collect();
    println!("Результаты: {:?}", results);
}

use std::sync::{Arc, Barrier};
use std::thread;
 
fn main() {
    let mut handles = Vec::with_capacity(10);
    let barrier = Arc::new(Barrier::new(10));
    
    for i in 0..10 {
        let barrier = Arc::clone(&barrier);
        let handle = thread::spawn(move || {
            println!("Поток {} готовится", i);
            
            // Все потоки будут ждать здесь, пока все 10 не дойдут до этой точки
            let wait_result = barrier.wait();
            
            // Теперь все потоки продолжат выполнение одновременно
            if wait_result.is_leader() {
                println!("Все потоки синхронизированы!");
            }
            
            println!("Поток {} продолжает работу", i);
        });
        handles.push(handle);
    }
    
    for handle in handles {
        handle.join().unwrap();
    }
}

use tracy_client::span;
 
fn expensive_operation() {
    let _span = span!("expensive_operation");
    // Код, который нужно профилировать
}

use std::sync::{Arc, Mutex};
use std::thread;
 
fn main() {
    let results = Arc::new(Mutex::new(Vec::new()));
    let errors = Arc::new(Mutex::new(Vec::new()));
    let mut handles = Vec::new();
    
    for i in 0..10 {
        let results = Arc::clone(&results);
        let errors = Arc::clone(&errors);
        
        let handle = thread::spawn(move || {
            if i % 3 == 0 {
                // Имитируем ошибку для некоторых значений
                errors.lock().unwrap().push(format!("Ошибка при обработке {}", i));
                return;
            }
            
            // Успешная обработка
            results.lock().unwrap().push(i * 2);
        });
        
        handles.push(handle);
    }
    
    for handle in handles {
        handle.join().unwrap();
    }
    
    println!("Успешные результаты: {:?}", results.lock().unwrap());
    println!("Ошибки: {:?}", errors.lock().unwrap());
}

use std::thread;
use std::arch::x86_64::{__m256, _mm256_add_ps, _mm256_set1_ps, _mm256_storeu_ps};
use std::mem::transmute;
 
unsafe fn simd_add(a: &[f32], b: &[f32], c: &mut [f32], start: usize, end: usize) {
    for i in (start..end).step_by(8) {
        if i + 8 <= end {
            let a_ptr = a.as_ptr().add(i);
            let b_ptr = b.as_ptr().add(i);
            let c_ptr = c.as_mut_ptr().add(i);
            
            let a_vec: __m256 = _mm256_loadu_ps(a_ptr);
            let b_vec: __m256 = _mm256_loadu_ps(b_ptr);
            let result = _mm256_add_ps(a_vec, b_vec);
            
            _mm256_storeu_ps(c_ptr, result);
        } else {
            // Обрабатываем оставшиеся элементы
            for j in i..end {
                c[j] = a[j] + b[j];
            }
        }
    }
}
 
// Функция _mm256_loadu_ps отсутствует в скомпилированном коде, но она должна быть
// в реальной программе с использованием инструкций AVX

use std::sync::{Arc, Mutex};
use std::thread;
 
fn main() {
    let counter = Arc::new(Mutex::new(0));
    let mut handles = vec![];
 
    for _ in 0..10 {
        let counter = Arc::clone(&counter);
        let handle = thread::spawn(move || {
            let mut num = counter.lock().unwrap();
            *num += 1;
        });
        handles.push(handle);
    }
 
    for handle in handles {
        handle.join().unwrap();
    }
 
    println!("Результат: {}", *counter.lock().unwrap());
}

use std::sync::{Arc, RwLock};
use std::thread;
 
fn main() {
    let data = Arc::new(RwLock::new(vec![1, 2, 3]));
    
    // Несколько потоков читателей
    let mut read_handles = vec![];
    for i in 0..3 {
        let data = Arc::clone(&data);
        let handle = thread::spawn(move || {
            let values = data.read().unwrap();
            println!("Поток {} видит: {:?}", i, *values);
        });
        read_handles.push(handle);
    }
    
    // Один поток писатель
    let data_clone = Arc::clone(&data);
    let write_handle = thread::spawn(move || {
        let mut values = data_clone.write().unwrap();
        values.push(4);
        println!("Поток записи изменил данные: {:?}", *values);
    });
    
    // Дожидаемся завершения всех потоков
    for handle in read_handles {
        handle.join().unwrap();
    }
    write_handle.join().unwrap();
}

use std::sync::Arc;
use std::thread;
 
fn main() {
    let shared_data = Arc::new(vec![1, 2, 3, 4]);
    let mut handles = vec![];
    
    for i in 0..4 {
        let data = Arc::clone(&shared_data);
        let handle = thread::spawn(move || {
            println!("Поток {} видит элемент {}: {}", i, i, data[i]);
        });
        handles.push(handle);
    }
    
    for handle in handles {
        handle.join().unwrap();
    }
}

use std::sync::{Arc, Mutex};
use std::thread;
 
fn process_data(data: &Vec<i32>) -> i32 {
    data.iter().sum()
}
 
fn main() {
    let data = Arc::new(vec![1, 2, 3, 4, 5]);
    let results = Arc::new(Mutex::new(vec![]));
    let mut handles = vec![];
    
    for _ in 0..4 {
        // Клонируем Arc перед передачей в поток
        let data = Arc::clone(&data);
        let results = Arc::clone(&results);
        
        let handle = thread::spawn(move || {
            // Вычисления без блокировок
            let result = process_data(&data);
            
            // Блокировка только для краткой операции вставки
            results.lock().unwrap().push(result);
        });
        
        handles.push(handle);
    }
    
    for handle in handles {
        handle.join().unwrap();
    }
    
    println!("Результаты: {:?}", *results.lock().unwrap());
}

use std::sync::{Arc, Mutex, Condvar};
use std::thread;
 
fn main() {
    let pair = Arc::new((Mutex::new(false), Condvar::new()));
    let pair_clone = Arc::clone(&pair);
 
    // Поток, который будет ждать сигнала
    let handle = thread::spawn(move || {
        let (lock, cvar) = &*pair_clone;
        let mut started = lock.lock().unwrap();
        
        // Пока флаг не установлен, ждём
        while !*started {
            started = cvar.wait(started).unwrap();
        }
        
        println!("Поток получил сигнал и продолжает работу");
    });
 
    // Даём потоку время для запуска
    thread::sleep(std::time::Duration::from_millis(100));
    
    // Отправляем сигнал
    let (lock, cvar) = &*pair;
    let mut started = lock.lock().unwrap();
    *started = true;
    cvar.notify_one();
    
    handle.join().unwrap();
}

// Правильно: всегда сначала lock1, потом lock2
let guard1 = lock1.lock().unwrap();
let guard2 = lock2.lock().unwrap();

match mutex.try_lock() {
    Ok(guard) => {
        // Используем защищенные данные
    },
    Err(_) => {
        // Блокировка занята, делаем что-то другое
    }
}

// Ждем блокировку не более 100 мс
if let Ok(guard) = mutex.try_lock_for(Duration::from_millis(100)) {
    // Работаем с защищенными данными
} else {
    // Превышен тайм-аут, принимаем меры
}

use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::Arc;
use std::thread;
 
struct AtomicFlag {
    flag: AtomicBool,
    counter: AtomicUsize,
}
 
impl AtomicFlag {
    fn new() -> Self {
        AtomicFlag {
            flag: AtomicBool::new(false),
            counter: AtomicUsize::new(0),
        }
    }
    
    fn set(&self) -> bool {
        if !self.flag.swap(true, Ordering::AcqRel) {
            self.counter.fetch_add(1, Ordering::Relaxed);
            true
        } else {
            false
        }
    }
    
    fn clear(&self) {
        self.flag.store(false, Ordering::Release);
    }
    
    fn count(&self) -> usize {
        self.counter.load(Ordering::Relaxed)
    }
}
 
fn main() {
    let flag = Arc::new(AtomicFlag::new());
    let mut handles = vec![];
    
    for _ in 0..10 {
        let flag = Arc::clone(&flag);
        let handle = thread::spawn(move || {
            if flag.set() {
                // Критическая секция
                thread::sleep(std::time::Duration::from_millis(10));
                flag.clear();
            }
        });
        handles.push(handle);
    }
    
    for handle in handles {
        handle.join().unwrap();
    }
    
    println!("Флаг был установлен {} раз", flag.count());
}

use std::sync::mpsc;
use std::thread;
use std::time::Duration;
 
fn main() {
    // Неограниченный канал
    let (tx, rx) = mpsc::channel();
    
    // Запускаем отдельный поток-отправитель
    thread::spawn(move || {
        for i in 1..=5 {
            tx.send(i).unwrap();
            thread::sleep(Duration::from_millis(100));
        }
    });
    
    // Основной поток как получатель
    for received in rx {
        println!("Получено: {}", received);
    }
}

use std::sync::mpsc;
use std::thread;
 
fn main() {
    // Ограниченный канал с буфером на 2 сообщения
    let (tx, rx) = mpsc::sync_channel(2);
    
    let sender = thread::spawn(move || {
        for i in 1..=5 {
            println!("Отправляем: {}", i);
            tx.send(i).unwrap(); // Блокируется после заполнения буфера
            println!("Отправлено: {}", i);
        }
    });
    
    thread::sleep(std::time::Duration::from_secs(1)); // Имитация задержки получателя
    
    for received in rx {
        println!("Получено: {}", received);
        thread::sleep(std::time::Duration::from_millis(500));
    }
    
    sender.join().unwrap();
}

use std::sync::mpsc;
use std::thread;
 
fn main() {
    let (tx, rx) = mpsc::channel();
    let rx = std::sync::Arc::new(std::sync::Mutex::new(rx));
    
    // Создаём пул рабочих потоков
    let mut workers = Vec::new();
    for id in 0..4 {
        let rx = std::sync::Arc::clone(&rx);
        let worker = thread::spawn(move || {
            loop {
                let message = {
                    let rx = rx.lock().unwrap();
                    match rx.try_recv() {
                        Ok(msg) => msg,
                        Err(_) => break,
                    }
                };
                println!("Рабочий {} обрабатывает: {}", id, message);
                thread::sleep(std::time::Duration::from_millis(100));
            }
        });
        workers.push(worker);
    }
    
    // Отправляем задания
    for i in 0..10 {
        tx.send(i).unwrap();
    }
    
    drop(tx); // Закрываем канал
    
    // Ждём завершения всех рабочих
    for worker in workers {
        worker.join().unwrap();
    }
}

use std::sync::mpsc;
use std::thread;
 
fn main() {
    let (tx, rx) = mpsc::channel();
    
    // Запускаем несколько потоков-отправителей
    for id in 0..4 {
        let tx = tx.clone();
        thread::spawn(move || {
            for i in 0..3 {
                let message = format!("Поток {} сообщение {}", id, i);
                tx.send(message).unwrap();
                thread::sleep(std::time::Duration::from_millis(100));
            }
        });
    }
    
    // Закрываем оригинальный передатчик, чтобы receive знал, когда все передатчики завершились
    drop(tx);
    
    // Получаем все сообщения
    while let Ok(message) = rx.recv() {
        println!("Получено: {}", message);
    }
}

use crossbeam_queue::ArrayQueue;
use std::sync::Arc;
use std::thread;
 
fn main() {
    let queue = Arc::new(ArrayQueue::new(100));
    let queue_producer = Arc::clone(&queue);
    let queue_consumer = Arc::clone(&queue);
    
    // Поток-производитель
    let producer = thread::spawn(move || {
        for i in 0..50 {
            queue_producer.push(i).unwrap();
        }
    });
    
    // Поток-потребитель
    let consumer = thread::spawn(move || {
        let mut sum = 0;
        for _ in 0..50 {
            while let Some(item) = queue_consumer.pop() {
                sum += item;
            }
            thread::yield_now(); // Даём другим потокам возможность выполниться
        }
        sum
    });
    
    producer.join().unwrap();
    let sum = consumer.join().unwrap();
    println!("Сумма элементов: {}", sum);
}

use crossbeam_skiplist::SkipMap;
use std::sync::Arc;
use std::thread;
 
fn main() {
    let map = Arc::new(SkipMap::new());
    let mut handles = Vec::new();
    
    // Несколько потоков одновременно записывают в SkipMap
    for i in 0..4 {
        let map = Arc::clone(&map);
        let handle = thread::spawn(move || {
            for j in 0..25 {
                let key = i * 100 + j;
                map.insert(key, format!("value_{}", key));
            }
        });
        handles.push(handle);
    }
    
    // Поток, который читает из SkipMap
    let map_clone = Arc::clone(&map);
    let reader = thread::spawn(move || {
        thread::sleep(std::time::Duration::from_millis(50));
        let count = map_clone.len();
        println!("Текущее количество элементов: {}", count);
    });
    
    for handle in handles {
        handle.join().unwrap();
    }
    reader.join().unwrap();
    
    println!("Итоговое количество элементов: {}", map.len());
}

use dashmap::DashMap;
use std::sync::Arc;
use std::thread;
 
fn main() {
    let map = Arc::new(DashMap::new());
    
    // Заполняем карту в нескольких потоках
    let mut handles = Vec::new();
    for i in 0..4 {
        let map = Arc::clone(&map);
        let handle = thread::spawn(move || {
            for j in 0..100 {
                let key = format!("key_{}_{}", i, j);
                map.insert(key, i * j);
            }
        });
        handles.push(handle);
    }
    
    // Параллельно считаем сумму всех значений
    let map_sum = Arc::clone(&map);
    let sum_handle = thread::spawn(move || {
        let mut sum = 0;
        for item in map_sum.iter() {
            sum += *item.value();
        }
        sum
    });
    
    // Ждём завершения всех потоков
    for handle in handles {
        handle.join().unwrap();
    }
    
    let total_sum = sum_handle.join().unwrap();
    println!("Общее количество элементов: {}", map.len());
    println!("Сумма всех значений: {}", total_sum);
}

use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Instant;
 
fn parallel_sum(data: &[i32], num_threads: usize) -> i32 {
    let len = data.len();
    let chunk_size = (len + num_threads - 1) / num_threads;
    
    let result = Arc::new(Mutex::new(0));
    let mut handles = vec![];
    
    for i in 0..num_threads {
        let start = i * chunk_size;
        let end = std::cmp::min(start + chunk_size, len);
        
        if start >= len {
            break;
        }
        
        let thread_data = data[start..end].to_vec();
        let result = Arc::clone(&result);
        
        handles.push(thread::spawn(move || {
            let sum: i32 = thread_data.iter().sum();
            let mut global_sum = result.lock().unwrap();
            *global_sum += sum;
        }));
    }
    
    for handle in handles {
        handle.join().unwrap();
    }
    
    *result.lock().unwrap()
}
 
fn main() {
    let data: Vec<i32> = (0..1_000_000).collect();
    
    // Тестируем с разным количеством потоков
    for threads in [1, 2, 4, 8, 16, 32] {
        let start = Instant::now();
        let sum = parallel_sum(&data, threads);
        let duration = start.elapsed();
        
        println!("Потоков: {}, Время: {:?}, Сумма: {}", 
                 threads, duration, sum);
    }
}

use criterion::{black_box, criterion_group, criterion_main, Criterion};
use std::sync::{Arc, Mutex};
use std::thread;
 
fn bench_mutex_vs_atomic(c: &mut Criterion) {
    let mut group = c.benchmark_group("Инкремент счётчика");
    
    group.bench_function("mutex", |b| {
        b.iter(|| {
            let counter = Arc::new(Mutex::new(0));
            let mut handles = vec![];
            
            for _ in 0..4 {
                let counter = Arc::clone(&counter);
                let handle = thread::spawn(move || {
                    for _ in 0..250 {
                        let mut num = counter.lock().unwrap();
                        *num += 1;
                    }
                });
                handles.push(handle);
            }
            
            for handle in handles {
                handle.join().unwrap();
            }
            
            black_box(*counter.lock().unwrap())
        })
    });
    
    group.bench_function("atomic", |b| {
        b.iter(|| {
            use std::sync::atomic::{AtomicUsize, Ordering};
            
            let counter = Arc::new(AtomicUsize::new(0));
            let mut handles = vec![];
            
            for _ in 0..4 {
                let counter = Arc::clone(&counter);
                let handle = thread::spawn(move || {
                    for _ in 0..250 {
                        counter.fetch_add(1, Ordering::SeqCst);
                    }
                });
                handles.push(handle);
            }
            
            for handle in handles {
                handle.join().unwrap();
            }
            
            black_box(counter.load(Ordering::SeqCst))
        })
    });
    
    group.finish();
}
 
criterion_group!(benches, bench_mutex_vs_atomic);
criterion_main!(benches);

// Неоптимально: блокировка удерживается дольше, чем необходимо
let mut data = mutex.lock().unwrap();
perform_lengthy_computation(&mut data); // Блокирует другие потоки

// Оптимальнее: блокировка только для получения данных
let local_copy = {
    let data = mutex.lock().unwrap();
    data.clone()
};
let result = perform_lengthy_computation(&local_copy);
mutex.lock().unwrap().update(result);

// Слишком крупная гранулярность: одна блокировка на всю структуру
struct Database {
    users: Vec<User>,
    posts: Vec<Post>,
    comments: Vec<Comment>,
}
 
// Лучше: отдельные блокировки для разных частей
struct Database {
    users: Mutex<Vec<User>>,
    posts: Mutex<Vec<Post>>,
    comments: Mutex<Vec<Comment>>,
}

use std::sync::mpsc;
use std::thread;
 
struct Actor {
    receiver: mpsc::Receiver<Message>,
    counter: usize,
}
 
enum Message {
    Increment,
    GetValue(mpsc::Sender<usize>),
    Stop,
}
 
impl Actor {
    fn new(receiver: mpsc::Receiver<Message>) -> Self {
        Actor {
            receiver,
            counter: 0,
        }
    }
    
    fn run(&mut self) {
        loop {
            match self.receiver.recv().unwrap() {
                Message::Increment => {
                    self.counter += 1;
                }
                Message::GetValue(sender) => {
                    sender.send(self.counter).unwrap();
                }
                Message::Stop => break,
            }
        }
    }
}
 
fn main() {
    let (sender, receiver) = mpsc::channel();
    
    let actor_thread = thread::spawn(move || {
        let mut actor = Actor::new(receiver);
        actor.run();
    });
    
    // Отправляем сообщения актору
    for _ in 0..5 {
        sender.send(Message::Increment).unwrap();
    }
    
    // Запрашиваем текущее значение
    let (response_tx, response_rx) = mpsc::channel();
    sender.send(Message::GetValue(response_tx)).unwrap();
    
    println!("Текущее значение: {}", response_rx.recv().unwrap());
    
    // Останавливаем актора
    sender.send(Message::Stop).unwrap();
    actor_thread.join().unwrap();
}

use tokio::sync::Mutex;
use std::sync::Arc;
 
async fn process_data(shared_data: Arc<Mutex<Vec<i32>>>) {
    // В асинхронном контексте используйте tokio::sync::Mutex вместо std::sync::Mutex
    let mut data = shared_data.lock().await;
    data.push(42);
}
 
#[tokio::main]
async fn main() {
    let shared_data = Arc::new(Mutex::new(vec![1, 2, 3]));
    
    let mut tasks = Vec::new();
    
    for _ in 0..10 {
        let data_clone = Arc::clone(&shared_data);
        let task = tokio::spawn(async move {
            process_data(data_clone).await;
        });
        tasks.push(task);
    }
    
    // Ожидаем завершения всех задач
    for task in tasks {
        task.await.unwrap();
    }
    
    println!("Результат: {:?}", *shared_data.lock().await);
}

use tokio_scoped::scope;
 
async fn process_data(data: &[i32]) -> i32 {
    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
    data.iter().sum()
}
 
#[tokio::main]
async fn main() {
    let data = vec![1, 2, 3, 4, 5];
    
    let result = scope(|s| {
        // Заимствуем data без перемещения владения
        s.spawn(async {
            process_data(&data).await
        })
    }).await;
    
    println!("Результат: {}", result);
}

use flume::{Sender, Receiver};
use std::thread;
 
fn main() {
    let (tx, rx): (Sender<i32>, Receiver<i32>) = flume::unbounded();
    
    thread::spawn(move || {
        for i in 0..10 {
            tx.send(i).unwrap();
        }
    });
    
    // Синхронное получение
    for i in rx.iter().take(5) {
        println!("Синхронно получено: {}", i);
    }
    
    // Можно использовать тот же приёмник асинхронно
    tokio::runtime::Runtime::new().unwrap().block_on(async {
        while let Ok(val) = rx.recv_async().await {
            println!("Асинхронно получено: {}", val);
        }
    });
}

// Гипотетический синтаксис будущих версий Rust
fn main() {
    let data = vec![1, 2, 3, 4, 5, 6, 7, 8];
    
    // Параллельное преобразование как часть языка
    let result = parallel for x in data {
        x * x
    };
    
    println!("Результат: {:?}", result);
}

use gpu_compute::prelude::*;
 
fn main() {
    let device = gpu_compute::Device::select_best().unwrap();
    let data = vec![1.0f32, 2.0, 3.0, 4.0, 5.0];
    
    // Перемещаем данные на GPU
    let gpu_buffer = device.create_buffer_from_slice(&data).unwrap();
    
    // Запускаем параллельные вычисления на GPU
    let result = device.execute_parallel(|idx| {
        let val = gpu_buffer.get(idx);
        val * val + 1.0
    }, data.len()).unwrap();
    
    println!("Результаты с GPU: {:?}", result.to_vec());
}

#[verify::concurrent]
fn increment_counter(counter: &AtomicUsize) {
    // Верификатор проверит атомарность и отсутствие гонок
    counter.fetch_add(1, Ordering::SeqCst);
}

use hybrid_parallel::{thread_pool, work_stealing, parallel_for};
 
fn main() {
    // Создаём иерархический пул потоков
    let pool = thread_pool::Builder::new()
        .num_threads(4)
        .work_stealing(true)
        .build();
    
    let data = vec![vec![1, 2, 3], vec![4, 5, 6], vec![7, 8, 9]];
    
    // Обрабатываем внешний массив параллельно
    let results = pool.scope(|s| {
        data.iter().map(|inner| {
            s.spawn(move || {
                // Для каждого внутреннего массива используем
                // параллельный итератор с кражей работы
                parallel_for(inner, |&x| x * x).sum::<i32>()
            })
        }).collect::<Vec<_>>()
    });
    
    println!("Результаты: {:?}", results);
}

use realtime_exec::{RealtimeThread, Priority, Scheduler};
 
fn main() {
    let scheduler = Scheduler::new();
    
    // Поток высокого приоритета
    let critical_task = RealtimeThread::new(Priority::High, || {
        // Критически важные вычисления
        // с гарантированным временем выполнения
    });
    
    // Поток среднего приоритета
    let normal_task = RealtimeThread::new(Priority::Normal, || {
        // Обычные вычисления
    });
    
    scheduler.add_thread(critical_task);
    scheduler.add_thread(normal_task);
    
    // Запускаем планировщик с предсказуемым поведением
    scheduler.run();
}