Своя СУБД на Rust · урок 9: конкуренция, блокировки и дедлоки

Кому это. В прошлом уроке появились транзакции и MVCC. DELETE стал журналируемой пометкой xmax, видимость решает is_visible по снимку, COMMIT форсит журнал, ROLLBACK идёт по undo-логу через замыкание. Хэндл Txn уже позволял открыть несколько транзакций, но пока никто не проверял, что будет, если две из них тронут одну строку.

Зачем. Сегодня транзакции начинают сосуществовать по-честному. Заработает snapshot isolation: транзакция A видит стабильный снимок, пока B рядом коммитит. Чтобы запись осталась корректной, добавим менеджер блокировок, проверку write-write конфликтов и поиск дедлока.

Зачем именно Rust. Здесь Rust-порт принимает решение, о котором надо сказать прямо. Менеджер блокировок не ждёт занятого лока. Конфликт без цикла возвращает Err(Error::LockWait) с SQLSTATE 55P03, тем же кодом, что Postgres отдаёт на SELECT ... FOR UPDATE NOWAIT. Это паритет с TS-эталоном, а не лень: движок однопоточный, BufferPool держит состояние в RefCell, а в уроке 11 сервер обслуживает все соединения через один Mutex<Database>. Если бы транзакция A ждала лок внутри этого мьютекса, транзакция B, которая держит лок, никогда не получила бы мьютекс, чтобы зафиксироваться. Сервер встал бы целиком. Ошибки конкуренции становятся вариантами одного enum Error, а match в sqlstate() превращает их в коды для клиента.

Важно про объём. Фундамент готов: версии, видимость, снимок, менеджер транзакций. Сегодня новый файл lock.rs, пара функций в операторах и транзакционные методы Database. Тесты сами чередуют операторы нескольких транзакций в одном потоке.

Где мы в движке

┌────────────────────────────────────────────────┐
│  Бинарный протокол + TCP-сервер                  │  урок 11
├────────────────────────────────────────────────┤
│  Исполнитель (Volcano) + EXPLAIN                 │  урок 6
│  Планировщик, binder, каталог                    │  урок 6
│  Парсер SQL (рекурсивный спуск + Pratt)          │  урок 5
├────────────────────────────────────────────────┤
│  Vacuum                                          │  урок 10
│  MVCC (видимость, версии кортежа)                │  урок 8
│  Блокировки (2PL), конкуренция                   │  урок 9  ← мы здесь
│  Менеджер транзакций (xid, состояния)            │  урок 8
│  WAL (журнал, redo, recovery)                    │  урок 7
├────────────────────────────────────────────────┤
│  B+Tree индекс                                   │  урок 4
│  Куча (slotted pages, seq scan)                  │  урок 3
├────────────────────────────────────────────────┤
│  Буферный пул (кэш страниц, вытеснение)           │  урок 2
│  DiskManager (страницы поверх файла)              │  урок 1
└────────────────────────────────────────────────┘

Две стороны конкуренции

У конкуренции под MVCC две стороны, и решаются они разными механизмами.

Чтение против записи. Тут MVCC уже всё сделал. Читатель смотрит на свой снимок и не видит чужих незафиксированных версий, а писатель создаёт новую версию, не трогая ту, что видит читатель. Ни одному не нужно ждать другого. Сегодня нужно лишь убедиться, что снимок честно наполняется активными транзакциями.

Запись против записи. Здесь версий мало. Две транзакции, которые одновременно меняют одну строку, обе поставили бы ей xmax, и одно изменение потерялось бы. Нужны два инструмента. Эксклюзивный лок строки по строгому 2PL не даёт двум писателям работать над строкой одновременно. А проверка xmax ловит случай, когда писатель опоздал: строку уже изменила и зафиксировала транзакция, которой нет в его снимке.

Снимок наконец наполняется

capture_snapshot из прошлого урока уже кладёт в active все открытые транзакции кроме себя. Сегодня это начинает влиять на результат. Видимость проверяется без базы, на руками собранных кортежах:

// tests/lesson_09_concurrency.rs (видимость на непустом снимке)

fn status() -> CommitStatus {
    CommitStatus::from_entries([
        (1, TxnState::Committed),
        (2, TxnState::Committed),
        (5, TxnState::InProgress),
        (6, TxnState::Aborted),
    ])
}

/// Снимок транзакции 10: активна транзакция 5, граница xmax = 10.
fn snap() -> Snapshot {
    Snapshot {
        xmin: 5,
        xmax: 10,
        active: HashSet::from([5]),
    }
}

fn tuple(xmin: u32, xmax: u32) -> Tuple {
    Tuple::with_header(vec![int(1)], xmin, xmax)
}

#[test]
fn insert_of_an_active_concurrent_transaction_is_invisible() {
    assert!(!is_visible(&tuple(5, 0), &snap(), 10, &status()));
}

#[test]
fn delete_by_a_concurrent_transaction_does_not_hide_the_row() {
    assert!(is_visible(&tuple(1, 5), &snap(), 10, &status()));
}

#[test]
fn insert_from_the_future_is_invisible() {
    assert!(!is_visible(&tuple(11, 0), &snap(), 10, &status()));
}

Три теста, три правила. Вставка транзакции 5, открытой в момент снимка, не видна. Удаление той же транзакции 5 ещё не скрывает строку: для меня оно не случилось. Вставка из «будущего», с xid не меньше xmax снимка, тоже не видна, даже если когда-нибудь зафиксируется. HashSet::from([5]) собирает множество из массива одним вызовом, без цепочки insert.

Менеджер блокировок

// src/lock.rs
//! LockManager (урок 9): строгий двухфазный протокол. Транзакция берёт эксклюзивный лок
//! на ресурс (строку) при записи и держит до commit/abort. На конфликте добавляется ребро
//! графа ожидания и ищется цикл: цикл это дедлок (`Error::Deadlock`, запросившая жертва),
//! иначе would-block (`Error::LockWait`).

use std::collections::{HashMap, HashSet};

use crate::error::{Error, Result};
use crate::tuple::Xid;

#[derive(Debug, Default)]
pub struct LockManager {
    /// Ресурс -> владелец (эксклюзивно).
    owners: HashMap<String, Xid>,
    /// Транзакция -> ресурсы, которыми она владеет.
    held: HashMap<Xid, HashSet<String>>,
    /// Граф ожидания: ожидающий -> те, кого он ждёт.
    waits_for: HashMap<Xid, HashSet<Xid>>,
}

impl LockManager {
    pub fn new() -> LockManager {
        LockManager::default()
    }

    /// Взять эксклюзивный лок. Повторный захват своим владельцем это no-op.
    pub fn acquire(&mut self, xid: Xid, resource: &str) -> Result<()> {
        let owner = match self.owners.get(resource) {
            None => None,
            Some(&owner) if owner == xid => None,
            Some(&owner) => Some(owner),
        };
        let Some(owner) = owner else {
            self.owners.insert(resource.to_owned(), xid);
            self.held
                .entry(xid)
                .or_default()
                .insert(resource.to_owned());
            return Ok(());
        };
        self.waits_for.entry(xid).or_default().insert(owner);
        if self.has_cycle(xid) {
            if let Some(targets) = self.waits_for.get_mut(&xid) {
                targets.remove(&owner);
            }
            return Err(Error::Deadlock(format!(
                "deadlock: транзакция {xid} откатывается (ждала {owner})"
            )));
        }
        Err(Error::LockWait(format!(
            "ресурс {resource} занят транзакцией {owner}"
        )))
    }

    /// Освободить все локи и рёбра ожидания транзакции (на commit/abort).
    pub fn release(&mut self, xid: Xid) {
        for resource in self.held.remove(&xid).unwrap_or_default() {
            if self.owners.get(&resource) == Some(&xid) {
                self.owners.remove(&resource);
            }
        }
        self.waits_for.remove(&xid);
        for waiters in self.waits_for.values_mut() {
            waiters.remove(&xid);
        }
    }

    /// Достижим ли start из самого себя по рёбрам ожидания (поиск в глубину).
    fn has_cycle(&self, start: Xid) -> bool {
        let mut visited = HashSet::new();
        let mut stack = vec![start];
        while let Some(node) = stack.pop() {
            for &next in self.waits_for.get(&node).into_iter().flatten() {
                if next == start {
                    return true;
                }
                if visited.insert(next) {
                    stack.push(next);
                }
            }
        }
        false
    }
}

Три карты держат всё состояние. owners отвечает «кто держит ресурс», held отвечает «что держит транзакция» и нужна для быстрого release, waits_for это граф ожидания.

Разберём acquire по ходу чтения.

1. Первый match сводит три случая к одному Option<Xid>: ресурс свободен или уже мой (это None, конфликта нет) либо занят другим (Some(owner)). Защитное условие Some(&owner) if owner == xid делает повторный захват своим лок-ом бесплатным.
2. let Some(owner) = owner else { … } это let-else. Ветка else здесь это счастливый путь «конфликта нет»: записать владельца, запомнить ресурс у транзакции и вернуть Ok. Ниже остаётся только конфликт, без лишнего уровня вложенности.
3. entry(xid).or_default() достаёт множество транзакции или создаёт пустое. Один поиск в карте вместо пары get и insert.
4. На конфликте сначала добавляем ребро ожидания, потом ищем цикл. Нашёлся цикл: ребро убираем (жертва откатится и ждать не будет) и возвращаем Deadlock. Не нашёлся: возвращаем LockWait, а ребро оставляем. Транзакция всё ещё «хочет» этот ресурс, и если позже кто-то замкнёт круг, has_cycle его увидит.

has_cycle это обход в глубину с явным стеком, без рекурсии: глубина графа ограничена только числом транзакций, а явный Vec не упрётся в стек потока. Выражение self.waits_for.get(&node).into_iter().flatten() превращает Option<&HashSet<Xid>> в итератор по соседям: None даёт пустой итератор, Some(set) даёт элементы множества. Никаких if let для вершины без рёбер. visited.insert(next) одновременно проверяет и отмечает вершину: insert возвращает false, если элемент уже был.

release выглядит избыточно аккуратным: зачем проверять owners.get(&resource) == Some(&xid), если ресурс записан в held этой транзакции? Это защита инварианта на случай будущих правок, например разделяемых локов в упражнении: освобождение не должно снять чужой лок. А цикл по values_mut() убирает транзакцию из всех списков ожидания: тот, кто её ждал, больше не ждёт.

Ключ ресурса это строка "relation:page:slot", как в TS-эталоне. Типизированный ключ (u32, PageId, u16) был бы дешевле, без аллокации format! на каждую запись. Порт выбрал строку ради паритета с другими языками и простоты печати в сообщении об ошибке.

Почему мы не ждём

В настоящей СУБД транзакция, наткнувшаяся на чужой лок, засыпает до его освобождения. В нашем движке она получает ошибку. Это решение стоит разобрать до конца, потому что Rust делает его последствия видимыми.

Движок однопоточный по устройству. BufferPool хранит состояние в RefCell, а latch страницы это try_lock, который паникует, если latch уже занят. Database из-за этого Send, но не Sync: её можно передать другому потоку, но не делить по &. В уроке 11 сервер положит её в Arc<Mutex<Database>>, и каждый запрос будет выполняться целиком под этим мьютексом. Теперь представь ожидание лока внутри acquire. Транзакция A держит мьютекс базы и спит, ожидая строку. Транзакция B держит эту строку и пришла по другому соединению зафиксироваться. Ей нужен мьютекс базы, а его держит спящая A. Две транзакции ждут друг друга через два разных механизма, и ни граф ожидания, ни мьютекс этого не видят.

Чтобы ждать честно, acquire должен отпустить мьютекс базы на время сна и снова захватить его после пробуждения. В Rust для этого есть Condvar::wait, который делает ровно это с MutexGuard. Но тогда другие запросы будут менять базу, пока A спит, и после пробуждения A обязана заново проверить и лок, и xmax строки. Это перестройка исполнителя, поэтому она вынесена в упражнение со звёздочкой.

Fail-fast на практике значит: LockWait валит только текущий оператор. Транзакция остаётся открытой, её ребро ожидания живёт в графе, и клиент может повторить оператор позже или откатиться. Ровно так ведёт себя Postgres с NOWAIT, поэтому код ошибки 55P03 (lock_not_available) взят оттуда.

Ошибки конкуренции

В TS-эталоне каждая ошибка это свой класс, а сессия выбирает код через instanceof. В Rust это варианты одного enum:

// src/error.rs (фрагмент)

#[derive(Debug)]
pub enum Error {
    /// Ошибка ввода-вывода (диск, сокет).
    Io(io::Error),
    /// Нарушение уникальности PRIMARY KEY.
    Uniqueness(String),
    /// Write-write конфликт под snapshot isolation.
    Serialization(String),
    /// Лок занят другой транзакцией без цикла ожидания.
    LockWait(String),
    /// Цикл в графе ожидания: запросившая транзакция это жертва.
    Deadlock(String),
    // ... Sql, Engine, Invalid, Wire
}

impl Error {
    /// SQLSTATE-подобный код для ErrorResponse (см. session.ts errorCode).
    pub fn sqlstate(&self) -> &str {
        match self {
            Error::Uniqueness(_) => "23505",
            Error::Serialization(_) => "40001",
            Error::Deadlock(_) => "40P01",
            Error::LockWait(_) => "55P03",
            Error::Sql { .. } => "42601",
            Error::Engine(_) => "42000",
            Error::Wire { code, .. } => code,
            Error::Io(_) | Error::Invalid(_) => "XX000",
        }
    }
}

match без ветки _ здесь принципиален. Добавь новый вариант ошибки, и sqlstate перестанет компилироваться, пока ты не назначишь ему код. В TS забытый instanceof молча дал бы общий код. Коды взяты из приложения Postgres: 40001 это serialization_failure, 40P01 это deadlock_detected, 23505 это unique_violation. Клиент, который умеет повторять транзакцию на 40001 и 40P01, сможет делать это и с нашим сервером. Тестам вариант удобно проверять через matches!(error, Error::LockWait(_)).