Сервис перестал отвечать. CPU — ноль, память ровная, паник нет, последняя строка в логе: «беру блокировку». Таймаут, которым эта блокировка была обёрнута, тоже не сработал: таймер — задача, а исполнять её некому. Один рабочий поток встал в ожидании мьютекса, отпустить который должна задача, которой для продолжения нужен ровно этот поток.

В синхронном коде такое кончилось бы иначе. Поток, повисший на мьютексе, мешает себе и тому, кто его ждёт, а остальные потоки живут дальше. В async всё иначе, потому что поток здесь не единица работы, а исполнитель для сотен задач. Заблокировали поток — заморозили всех, кто на нём стоял в очереди, включая таймеры, ввод-вывод и ту самую задачу, чьего завершения вы ждёте.

В статье рассмотрим пять мест, где блокировки в async Rust ведут себя не так, как подсказывает опыт синхронного кода.

Компилятор ловит гард через .await только у Send-футур

Правило «не держи std::sync::MutexGuard через await» знают многие, и обычно добавляют: компилятор всё равно не даст. Вообще-то даст, просто не там, где вы ждёте.

Вот кэш, который лезет в сеть, не отпуская блокировку:

async fn broken(cache: &Mutex<HashMap<String, String>>, k: &str) -> String {
    let mut guard = cache.lock().unwrap();
    if let Some(v) = guard.get(k) {
        return v.clone();
    }
    let v = fetch(k).await;              // гард жив через await
    guard.insert(k.to_string(), v.clone());
    v
}

Отдаём в tokio::spawn — компилятор ругается:

error: future cannot be sent between threads safely
    = help: within `{async block@src/main.rs:19:18}`, the trait `Send`
      is not implemented for `std::sync::MutexGuard<'_, HashMap<...>>`
note: future is not `Send` as this value is used across an await

Требование пришло из границы F: Future + Send + 'static у tokio::spawn. Не из языка, не из проверки на «правильный async». Убираем spawn и просто ждём эту функцию в main — всё собирается и работает:

#[tokio::main]
async fn main() {
    let cache = Mutex::new(HashMap::new());
    println!("{}", broken(&cache, "a").await);
    println!("скомпилировалось и отработало");
}

То же самое с spawn_local под LocalSet, любой функцией на однопоточном рантайме — везде, где задача не переезжает между потоками. Компилятор молчит не потому, что код корректен, а потому, что его никто не просил проверять.

Молчание обходится дорого.

Однопоточный рантайм, две задачи: первая берёт std-мьютекс и уходит в sleep, вторая через 50 мс пробует взять тот же мьютекс.

local.spawn_local(async move {
    let mut g = a.lock().unwrap();
    tokio::time::sleep(Duration::from_millis(200)).await;
    *g += 1;
    println!("задача A завершилась");
});

local.spawn_local(async move {
    tokio::time::sleep(Duration::from_millis(50)).await;
    println!("задача B пробует взять блокировку");
    let _g = b.lock().unwrap();          // блокирует единственный поток
    println!("задача B взяла блокировку");
});

let res = tokio::time::timeout(Duration::from_secs(3), local).await;

Вывод:

задача B пробует взять блокировку

И всё. Задача A не завершилась, задача B блокировку не взяла, трёхсекундный таймаут не сработал — процесс пришлось закрывать снаружи. Поток застрял внутри lock(), задача A ждёт очереди на этом же потоке, таймер ждёт там же. Компилятор не помешает держать std-гард через .await там, где задача не переезжает между потоками. К адекватному конкурентному коду это практически никогда не приводит.

Насколько уверенно вы ориентируетесь в Rust, можно проверить на вступительном тесте.

tokio::sync::Mutex по умолчанию: 50 наносекунд вместо 16

После предыдущего раздела хочется заменить все std::sync::Mutex на tokio::sync::Mutex и забыть. Понятная реакция, но в большинстве мест неверная. Обычный мьютекс из стандартной библиотеки в асинхронном коде использовать нормально и часто предпочтительнее.

Асинхронный мьютекс умеет только одно сверх обычного — жить через точку .await. За это он берёт с каждого захвата. Миллион захватов и отпусканий без конкуренции, один поток, релизная сборка:

let s = StdMutex::new(0u64);
let t0 = Instant::now();
for _ in 0..N { *s.lock().unwrap() += 1; }
let std_el = t0.elapsed();

let t = TokioMutex::new(0u64);
let t0 = Instant::now();
for _ in 0..N { *t.lock().await += 1; }
let tok_el = t0.elapsed();

std::sync::Mutex     15.6ms -> 16 нс на операцию
tokio::sync::Mutex   50.1ms -> 50 нс на операцию
отношение 3.2x

Тридцать четыре наносекунды разницы на захват. На горячем пути, который дёргают миллион раз в секунду, — это 34 мс CPU в секунду на ровном месте. Направление от железа не зависит. Асинхронный мьютекс устроен через семафор с очередью пробуждений, а std-мьютекс на неконкурентном пути укладывается в атомарную операцию.

К тому же мьютекс tokio гарантирует строгий FIFO: порядок захвата совпадает с порядком вызовов lock.

Рабочее правило в том, что если под замком данные — счётчик, карта, структура состояния — берите std::sync::Mutex или parking_lot::Mutex и не удерживайте его через .await. Если под замком ресурс, работа с которым сама асинхронна — соединение с базой, сокет, — тогда tokio::sync::Mutex.

Десять задач по сто миллисекунд занимают секунду

Допустим, асинхронный мьютекс взят осознанно, и гард живёт через .await. Тогда всё между захватом и отпусканием исполняется строго по одному. Само по себе это смысл мьютекса, но в async под замок легко заезжает ввод-вывод, который никакой защиты не требует.

Десять задач, каждая делает стомиллисекундный запрос и увеличивает счётчик. Разница — только в том, где стоит захват:

// гард живёт через await
let mut g = m.lock().await;
io().await;
*g += 1;

// блокировка только вокруг мутации
io().await;
*m.lock().await += 1;

гард через await:  1.013060726s, счётчик 10
блокировка после:  101.38357ms, счётчик 10

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

Разбирается по одному вопросу на каждый .await внутри критической секции: что сломается, если два таких вызова пойдут одновременно? Если ответ «ничего» — вызову место снаружи.

Если ответ есть, скорее всего вам нужен не мьютекс, а отдельная задача-владелец, которая принимает запросы каналом и исполняет их по очереди.

Второй read() не берётся, потому что впереди писатель

RwLock в async приносит ловушку, которой в синхронном мире у многих не было. Политика tokio — write-preferring, она же честная FIFO. Новый читатель не получит блокировку, пока не отработают все писатели, вставшие в очередь раньше него. Сделано, чтобы читатели не заморили писателей голодом.

Следствие на трёх шагах: берём чтение, ставим писателя в очередь, пробуем взять второе чтение, не отпустив первое:

let r1 = lock.read().await;
println!("read #1 взят");

let l = lock.clone();
tokio::spawn(async move {
    let _w = l.write().await;
    println!("писатель вошёл");
});
tokio::time::sleep(Duration::from_millis(50)).await;

let r2 = tokio::time::timeout(Duration::from_millis(500), lock.read()).await;
match r2 {
    Ok(_)  => println!("read #2 взят — очередь читательская"),
    Err(_) => println!("read #2 НЕ взят за 500 мс — писатель впереди"),
}

read #1 взят
писатель в очереди
read #2 НЕ взят за 500 мс — писатель впереди
писатель вошёл

Второе чтение не проходит. Первое чтение держит блокировку и ждёт второго, второе ждёт писателя, писатель ждёт первого — круг замкнулся. В реальном сервисе два read() разбросаны по разным функциям, и между ними десяток кадров стека.

Проверка одна: пройти по всем местам, где под живым read-гардом вызывается что-то, что тоже может взять этот же RwLock.

Отмена посреди критической секции и отравление, которого нет

Это специфично именно для async. Задачу в tokio можно отменить: abort, проигравшая ветка select, истёкший timeout, брошенный JoinHandle. Отмена наступает в точке .await, и если такая точка оказалась внутри критической секции, задача умирает посередине изменения данных.

Счёт, с которого списывают деньги, и журнал, куда пишут списание. Между ними — асинхронная проверка:

let task = tokio::spawn(async move {
    let mut g = a.lock().await;
    g.balance -= 50;                  // деньги сняли
    slow_audit().await;               // точка отмены
    g.log.push("списание 50".into()); // запись в журнал
});

tokio::time::sleep(Duration::from_millis(50)).await;
task.abort();

после отмены: Account { balance: 50, log: [] }

Баланс уменьшился, журнал пуст. Гард освободился при разворачивании задачи, мьютекс исправен, следующий, кто его возьмёт, увидит структуру, согласованную по типам и бессмысленную по смыслу. Никакого сигнала об этом он не получит: у tokio::sync::Mutex нет понятия отравления. У std::sync::Mutex оно есть — после паники под замком is_poisoned() вернёт true. Но отмена задачи паникой не считается, так что даже будь отравление у tokio, здесь бы оно не сработало.

Это можно исправить формой кода. Все изменения, которые обязаны произойти вместе, собираются в один синхронный кусок без .await внутри: посчитали заранее, взяли замок, применили целиком, отпустили. Асинхронный аудит уезжает до захвата или после отпускания.

Когда разнести не получается — явный откат. Обёртка, которая в Drop возвращает баланс на место, если до отметки об успехе дело не дошло:

struct Rollback<'a> { acc: &'a mut Account, amount: i64, done: bool }

impl Drop for Rollback<'_> {
    fn drop(&mut self) {
        if !self.done {
            self.acc.balance += self.amount;
        }
    }
}

Операция идёт через эту обёртку, done выставляется последней строкой, когда состояние снова согласовано:

let mut rb = Rollback { acc: &mut *g, amount: 50, done: false };
rb.acc.balance -= 50;
slow_audit().await;              // сняли задачу здесь — сработает Drop
rb.acc.log.push("списание 50".into());
rb.done = true;

Тот же прогон с abort через 50 мс теперь даёт Account { balance: 100, log: [] }. Работает потому, что при отмене задачи её кадр разворачивается штатно. Деструкторы локальных переменных вызываются в обычном порядке — тот же механизм, что освобождает и сам гард.

Чем заменить замок там, где он не нужен

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

thread 'main' panicked at src/main.rs:6:16:
Cannot block the current thread from within a runtime.

Паника вместо тихого зависания — предпочтительнее, её видно сразу. Место blocking_lock — внутри spawn_blocking или в обычном потоке, который забирает данные из асинхронного мира.

Оберните Arc<Mutex<...>> в структуру, которая наружу отдаёт обычные синхронные методы, а замок берёт внутри:

#[derive(Clone, Default)]
struct Cache(Arc<Mutex<HashMap<String, String>>>);

impl Cache {
    fn get(&self, k: &str) -> Option<String> {
        self.0.lock().unwrap().get(k).cloned()
    }
    fn put(&self, k: String, v: String) {
        self.0.lock().unwrap().insert(k, v);
    }
}

async fn lookup(cache: &Cache, k: &str) -> String {
    if let Some(v) = cache.get(k) {   // гард умирает внутри get
        return v;
    }
    let v = fetch(k).await;
    cache.put(k.to_string(), v.clone());
    v
}

Это тот же кэш из первого раздела, но он спокойно уходит в tokio::spawn и не требует асинхронного мьютекса. Гард физически не может пережить .await: в синхронном методе таких точек нет.

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

Что со всем этим делать

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

Порядок разбора:

1. Грепом по проекту найти все .await внутри областей, где жив какой-нибудь гард. Это одно движение, и оно закрывает первые три пункта.

2. Посмотреть, что под замком лежит. Данные — значит std::sync::Mutex и синхронные методы-обёртки. Ресурс ввода-вывода — значит либо tokio::sync::Mutex, либо задача-владелец с каналом.

3. Пройти по всем RwLock и проверить, не берётся ли чтение под живым чтением.

4. Пройти по всем местам, куда может прилететь отмена: select, timeout, abort. Убедиться, что изменение состояния там атомарно относительно точек .await.

К тому же подозреваю, что под конкуренцией разрыв между std и tokio ведёт себя нелинейно и в какой-то точке разворачивается в пользу tokio за счёт очереди

Разобраться с конкурентностью в Rust проще, когда хорошо понимаешь базовые механизмы языка: владение, заимствование и работу со ссылками. А дальше можно посмотреть, как эти принципы применяются уже в прикладных задачах — например, при разработке кроссплатформенных приложений.

На ближайших бесплатных уроках можно разобрать эти темы вместе с преподавателями, задать вопросы по своим кейсам и заодно посмотреть, как устроено обучение на практике.

• 9 сентября в 20:00. «Владение, заимствование и ссылки в Rust: как компилятор делает ваш код безопасным». Записаться

• 23 сентября в 20:00. «Создание кроссплатформенного приложения с GUI на Rust: от идеи до реальности». Записаться

Полный список бесплатных уроков сентября смотрите в дайджесте.