Взаимоблокировки мьютексов

Взаимные блокировки возникают, когда потоки бесконечно ждут друг друга, чтобы снять блокировки мьютекса, что приводит к остановке. Избежание взаимоблокировок имеет важное значение в многопоточном программировании.

Рассмотрим следующий пример, который демонстрирует потенциальную взаимоблокировку

use std::sync::{Mutex, Arc};
use std::thread;
use std::time::Duration;

fn main() {

    let data1 = Arc::new(Mutex::new(0)); // определяем первый общий ресурс
    let data2 = Arc::new(Mutex::new(0)); // определяем второй общий ресурс

    // создаем 1-й поток
    let thread1 = thread::spawn({
        // копируем оба ресурса
        let data1_clone = Arc::clone(&data1);
        let data2_clone = Arc::clone(&data2);
        move || {

            let mut guard1 = data1_clone.lock().unwrap(); // получаем блокировку на 1-й ресурс
            *guard1 += 1;
            thread::sleep(Duration::from_millis(200));
            let mut guard2 = data2_clone.lock().unwrap(); // получаем блокировку на 2-й ресурс
            *guard2 += 1;
        }
    });

    // создаем 2-й поток
    let thread2 = thread::spawn({

        // копируем оба ресурса
        let data1_clone = Arc::clone(&data1);
        let data2_clone = Arc::clone(&data2);

        move || {
            let mut guard2 = data2_clone.lock().unwrap(); // получаем блокировку на 2-й ресурс
            *guard2 += 1;
            thread::sleep(Duration::from_millis(200));
            let mut guard1 = data1_clone.lock().unwrap(); // получаем блокировку на 1-й ресурс
            *guard1 += 1;
        }
    });
    thread1.join().unwrap();
    thread2.join().unwrap();

    println!("data1 = {}, data2 = {}",
        data1.lock().unwrap(),
        data2.lock().unwrap()
    );
}

Здесь в фукции main определяется два ресурса. И также создается два потока. Первый поток сначала блокирует первый ресурс, изменяет его и после небольшой задержки пытается блокировать второй ресурс. Второй поток, наборот, сначала блокирует второй ресурс, изменяет его и потом пытается блокировать первый ресурс. С высокой вероятностью пока первый поток будет ждать разблокировки второго ресурса, первый поток будет ждать разблокировки первого ресурса. В итоге взаимное ожидание приведет к блокировке работы программы.

Чтобы выйти из этой ситуации для получения блокировки мы можем использовать метод try_lock() - если с его помощью невозможно получить блокировку, то поток продолжает работу без блокировки. Это может помочь предотвратить взаимоблокировки в ситуациях, когда получение блокировки не является критическим. Например, перепишем предыдущий пример с помощью try_lock():

use std::sync::{Mutex, Arc};
use std::thread;
use std::time::Duration;

fn check_mutex_and_increase(data_clone: Arc<Mutex<i32>>){
    let mut guard = data_clone.try_lock();
    match guard {
        Ok(ref mut value) => { **value += 1; }
        Err(_) => { println!("Failed to acquire lock, continuing…"); }
    }
}
fn main() {

    let data1 = Arc::new(Mutex::new(0)); // определяем первый общий ресурс
    let data2 = Arc::new(Mutex::new(0)); // определяем второй общий ресурс

    // создаем 1-й поток
    let thread1 = thread::spawn({
        // копируем оба ресурса
        let data1_clone = Arc::clone(&data1);
        let data2_clone = Arc::clone(&data2);
        move || {

            let mut guard1 = data1_clone.lock().unwrap(); // получаем блокировку на 1-й ресурс
            *guard1 += 1;
            thread::sleep(Duration::from_millis(200));
            // получаем блокировку на 2-й ресурс
            check_mutex_and_increase(data2_clone);
        }
    });

    // создаем 2-й поток
    let thread2 = thread::spawn({

        // копируем оба ресурса
        let data1_clone = Arc::clone(&data1);
        let data2_clone = Arc::clone(&data2);

        move || {
            // получаем блокировку на 2-й ресурс
            let mut guard2 = data2_clone.lock().unwrap(); // получаем блокировку на 1-й ресурс
            *guard2 += 1;
            thread::sleep(Duration::from_millis(200));
             // получаем блокировку на 1-й ресурс
            check_mutex_and_increase(data1_clone);
        }
    });
    thread1.join().unwrap();
    thread2.join().unwrap();


    println!("data1 = {}, data2 = {}",
        data1.lock().unwrap(),
        data2.lock().unwrap()
    );
}

Здесь также первую блокировку потоки получают с помощью метода lock(). Код получения второй блокировки я вынес в отдельную функцию:

fn check_mutex_and_increase(data_clone: Arc<Mutex<i32>>){
    let mut guard = data_clone.try_lock();
    match guard {
        Ok(ref mut value) => { **value += 1; }
        Err(_) => { println!("Failed to acquire lock, continuing…"); }
    }
}

Здесь сначала с помощью вызова data_clone.try_lock() пытаемся получить блокировку (метод возвращает объект Result). Далее с помощью match проверяем полученный результат. Если ошибки нет, то в value мы получаем мьютекс и через него обращаемся к самому значению, увеличивая его на 1. Обратите внимание на двойное разыменование - **value

Если же произошла ошибка, то есть блокировку невозможно получить, то просто выводим соответствующее сообщение.

Два потока также пытаются получить блокировку для двух одних и тех же ресурсов:

let mut guard1 = data1_clone.lock().unwrap(); // получаем блокировку на 1-й ресурс
*guard1 += 1;
thread::sleep(Duration::from_millis(200));
check_mutex_and_increase(data2_clone);

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

Failed to acquire lock, continuing…
Failed to acquire lock, continuing…
data1 = 1, data2 = 1