Многопоточность

Кратко

Конкурентность/параллелизм можно реализовать через очередь сообщений (Message-passing) или общее состояние (Shared-state)

Thread

При работе с конкурентностью нужно учитывать следующее

гонки данных (data race)
Состояния гонки (race conditions): thread обращаются к данным или ресурсам в несогласованном порядке
взаимная блокировка (deadlock)
Взаимные блокировки (deadlocks): два thread ждут, пока другой закончит использовать ресурс, который уже есть у первого, и оба не могут продолжить
трудно воспроизводимые BUG
Ошибки, которые возникают только в определённых ситуациях и которые трудно воспроизвести и надёжно исправить

Green Thread — это thread на уровне языка программирования; его функциональность не полностью совпадает с thread, который предоставляет операционная система

use std::thread;
use std::time::Duration;

fn main() {
    let handle = thread::spawn(|| {  // учтите: если замыканию нужны аргументы, нужно явно указать move
        for i in 1..10 {
            println!("hi number {} from the spawned thread!", i);
            thread::sleep(Duration::from_millis(1));  // в этот момент у операционной системы есть возможность переключиться на другие thread
        }
    });

    handle.join().unwrap();  // join ждёт завершения всех thread (иначе дальнейший код не выполнится)

    for i in 1..5 {
        println!("hi number {} from the main thread!", i);
        thread::sleep(Duration::from_millis(1));
    }
}

Message-passing

базовый вариант с Channel

use std::sync::mpsc;  // multiple producer single consumer

fn main() {
    let (tx, rx) = mpsc::channel();  // transmitter, receiver

		thread::spawn(move || {
        let val = String::from("hi");
        tx.send(val).unwrap();  // в этот момент val перемещается (moved ownership)
    });

		let received = rx.recv().unwrap();  // при использовании try_recv блокировки не будет
    println!("Got: {}", received);
}

несколько producer/transmitter && приём нескольких сообщений

use std::sync::mpsc;
use std::thread;
use std::time::Duration;

fn main() {
    let (tx, rx) = mpsc::channel();

		// потому что дальше замыкание переместит ownership tx
    // поэтому здесь нужно заранее клонировать tx1, чтобы получить несколько producer
		let tx1 = tx.clone();
		thread::spawn(move || {
        let vals = vec![
            String::from("hi"),
            String::from("from"),
            String::from("the"),
            String::from("thread"),
        ];

        for val in vals {
            tx1.send(val).unwrap();
            thread::sleep(Duration::from_secs(1));
        }
    });

    thread::spawn(move || {
        let vals = vec![
            String::from("more"),
            String::from("messages"),
            String::from("for"),
            String::from("you"),
        ];

        for val in vals {
            tx.send(val).unwrap();
            thread::sleep(Duration::from_secs(1));
        }
    });

		// обратите внимание: в отличие от предыдущего примера, здесь используется for-in loop
    for received in rx {  
        println!("Got: {}", received);
    }
}

Shared-state

Mutex (Mutual Exclusion, взаимное исключение); одно из его проявлений: в один момент времени только один thread может обращаться к определённым данным

Thread должен получить lock, чтобы обратиться к данным в mutex. lock — тоже часть mutex: он постоянно отслеживает, у кого есть право на взаимоисключающий доступ к данным в mutex. Поэтому мьютекс описывают как структуру, защищающую свои данные через систему блокировок.

Mutexes трудно использовать из-за следующих двух правил:

перед использованием данных нужно попытаться получить блокировку
Вы должны попытаться получить блокировку до того, как начнёте использовать данные.
когда работа с данными под защитой mutex закончена, блокировку нужно освободить, чтобы другие thread могли её получить
Когда вы закончили с данными, которые охраняет mutex, вы должны разблокировать данные, чтобы другие thread могли получить блокировку.

Управлять mutexes очень трудно, поэтому многие предпочитают channels. Однако система типов и правила владения Rust гарантируют, что мы не получим и не освободим блокировку неправильно

базовый вариант с Mutex
```
use std::sync::Mutex;

fn main() {
    let m = Mutex::new(5);

    {
        let mut num = m.lock().unwrap();  // вызов lock получает блокировку
        *num = 6;
    }

    println!("m = {:?}", m);
}
```
Mutex<T> — это умный указатель; точнее, lock возвращает LockResult, а затем unwrap возвращает умный указатель MutexGuard; MutexGuard реализует трейт Deref, указывая на внутренние данные, и трейт Drop, автоматически освобождая блокировку, когда MutexGuard выходит из области видимости. Поэтому мы не рискуем забыть освободить блокировку и заблокировать мьютекс для других thread, поскольку освобождение происходит автоматически.

совместное использование Mutex несколькими thread

Atomic Reference Counted Type — можно понимать как Rc<T> с атомарными операциями

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!("Result: {}", *counter.lock().unwrap());
}

Send & Sync trait

Типы, реализующие трейт Send, могут передавать ownership между thread

Типы, реализующие трейт Sync, могут использоваться несколькими thread одновременно

Видно, что Sync включает в себя свойства Send

Источник

https://doc.rust-lang.org/book/ch16-00-concurrency.html

…

Атомарный подсчёт ссылок с `Arc<T>`

К счастью, Arc<T> — это тип, похожий на Rc<T>, но безопасный в конкурентных ситуациях. a означает atomic, то есть это тип с атомарным подсчётом ссылок. Атомарные операции — дополнительный вид примитивов конкурентности, который мы здесь подробно не разбираем: см. документацию стандартной библиотеки для std::sync::atomic . Сейчас достаточно знать, что атомарные типы работают как примитивные, но безопасны для совместного использования между thread.

Тогда может возникнуть вопрос: почему не все примитивные типы атомарны и почему типы стандартной библиотеки не реализованы по умолчанию через Arc<T>? Причина в том, что потокобезопасность сопряжена с потерей производительности, которую хочется оплачивать только тогда, когда это действительно нужно. Если вы выполняете операции со значениями в одном thread, код может работать быстрее, если не обязан обеспечивать гарантии атомарности.

Вернёмся к примеру: у Arc<T> и Rc<T> одинаковый API, поэтому мы исправим программу, изменив строку use, вызов new и вызов clone. Код из листинга 16-15 наконец скомпилируется и запустится:

…

Rust - Basic - 15 - Многопоточность