第13课：MVCC多版本并发

事务与并发 第13课 / 共25课

📖 课程概述

MVCC(多版本并发控制)是现代数据库实现高并发读写的核心技术。通过为每行数据保留多个版本，读操作不需要加锁，写操作不阻塞读操作，极大地提高了并发性能。本课深入实现MVCC的可见性判断、快照隔离和版本清理。

👁️ MVCC可见性规则

隔离级别与MVCC

💻 C语言实现：完整MVCC引擎

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <stdint.h>

#define MAX_ROWS     64
#define MAX_VERSIONS 16
#define MAX_KEY      64
#define MAX_VAL      255
#define MAX_ACTIVE   32
#define MAX_TXN      128

typedef uint32_t TxnID;

// 事务状态
typedef enum { TXN_RUNNING, TXN_COMMITTED, TXN_ABORTED } TxnState;

// 事务信息
typedef struct {
    TxnID    id;
    TxnState state;
    TxnID    snapshot[MAX_ACTIVE]; // 快照中的活跃事务
    int      snap_size;
} TxnInfo;

// 行版本(PostgreSQL风格)
typedef struct {
    TxnID   xmin;       // 创建该版本的事务
    TxnID   xmax;       // 删除/更新该版本的事务(0=有效)
    char    key[MAX_KEY];
    char    value[MAX_VAL];
    int     active;     // 是否仍可见
} RowVersion;

// 行的多版本链
typedef struct {
    RowVersion versions[MAX_VERSIONS];
    int        num_versions;
} Row;

// MVCC引擎
typedef struct {
    Row       rows[MAX_ROWS];
    int       num_rows;
    TxnInfo   txns[MAX_TXN];
    int       num_txns;
    TxnID     next_txn;
    TxnID     active_list[MAX_ACTIVE];
    int       active_count;
    int       reads, writes, conflicts;
} MVCCEngine;

MVCCEngine* mvcc_create() {
    MVCCEngine* e = calloc(1, sizeof(MVCCEngine));
    e->next_txn = 1;
    printf("[MVCC] 引擎创建\n");
    return e;
}

// 创建快照
void take_snapshot(MVCCEngine* e, TxnInfo* txn) {
    txn->snap_size = e->active_count;
    memcpy(txn->snapshot, e->active_list, e->active_count * sizeof(TxnID));
}

// 可见性判断
int is_visible(MVCCEngine* e, RowVersion* v, TxnInfo* reader) {
    // 规则1: 创建者是自己 → 可见
    if (v->xmin == reader->id) {
        if (v->xmax == 0) return 1;
        if (v->xmax == reader->id) return 0; // 自己删除的
        // xmax是否在快照中
        for (int i = 0; i < reader->snap_size; i++) {
            if (v->xmax == reader->snapshot[i]) return 1; // 删除未提交
        }
        // xmax已提交 → 不可见
        for (int i = 0; i < e->num_txns; i++) {
            if (e->txns[i].id == v->xmax && e->txns[i].state == TXN_COMMITTED)
                return 0;
        }
        return 1;
    }

    // 规则2: 创建者是否已提交
    int xmin_committed = 0;
    for (int i = 0; i < e->num_txns; i++) {
        if (e->txns[i].id == v->xmin && e->txns[i].state == TXN_COMMITTED) {
            xmin_committed = 1;
            break;
        }
    }
    if (!xmin_committed) return 0; // 创建者未提交

    // 规则3: 创建者是否在快照活跃列表中
    for (int i = 0; i < reader->snap_size; i++) {
        if (v->xmin == reader->snapshot[i]) return 0; // 创建者活跃→不可见
    }

    // 创建者已提交且不在活跃列表 → 创建可见
    if (v->xmax == 0) return 1; // 未被删除

    // 规则4: 删除者判断
    if (v->xmax == reader->id) return 0; // 自己删除的
    for (int i = 0; i < reader->snap_size; i++) {
        if (v->xmax == reader->snapshot[i]) return 1; // 删除未提交
    }
    // xmax已提交且不在活跃列表 → 删除可见→版本不可见
    for (int i = 0; i < e->num_txns; i++) {
        if (e->txns[i].id == v->xmax && e->txns[i].state == TXN_COMMITTED)
            return 0;
    }
    return 1; // 删除未提交→版本仍可见
}

// 事务操作
TxnID mvcc_begin(MVCCEngine* e) {
    TxnID id = e->next_txn++;
    TxnInfo* txn = &e->txns[e->num_txns++];
    txn->id = id;
    txn->state = TXN_RUNNING;
    take_snapshot(e, txn);
    e->active_list[e->active_count++] = id;
    printf("[MVCC] BEGIN txn=%u (快照: %d个活跃事务)\n", id, txn->snap_size);
    return id;
}

void mvcc_commit(MVCCEngine* e, TxnID txn_id) {
    for (int i = 0; i < e->num_txns; i++) {
        if (e->txns[i].id == txn_id) {
            e->txns[i].state = TXN_COMMITTED;
            // 从活跃列表移除
            for (int j = 0; j < e->active_count; j++) {
                if (e->active_list[j] == txn_id) {
                    e->active_list[j] = e->active_list[e->active_count - 1];
                    e->active_count--;
                    break;
                }
            }
            printf("[MVCC] COMMIT txn=%u\n", txn_id);
            return;
        }
    }
}

void mvcc_abort(MVCCEngine* e, TxnID txn_id) {
    for (int i = 0; i < e->num_txns; i++) {
        if (e->txns[i].id == txn_id) {
            e->txns[i].state = TXN_ABORTED;
            for (int j = 0; j < e->active_count; j++) {
                if (e->active_list[j] == txn_id) {
                    e->active_list[j] = e->active_list[e->active_count - 1];
                    e->active_count--;
                    break;
                }
            }
            printf("[MVCC] ABORT txn=%u\n", txn_id);
            return;
        }
    }
}

// 插入
int mvcc_insert(MVCCEngine* e, TxnID txn, const char* key, const char* val) {
    if (e->num_rows >= MAX_ROWS) return -1;
    Row* row = &e->rows[e->num_rows++];
    RowVersion* v = &row->versions[0];
    v->xmin = txn;
    v->xmax = 0;
    strcpy(v->key, key);
    strcpy(v->value, val);
    v->active = 1;
    row->num_versions = 1;
    e->writes++;
    printf("  [MVCC] INSERT %s=%s (txn=%u)\n", key, val, txn);
    return 0;
}

// 更新
int mvcc_update(MVCCEngine* e, TxnID txn, const char* key, const char* new_val) {
    for (int i = 0; i < e->num_rows; i++) {
        Row* row = &e->rows[i];
        for (int j = 0; j < row->num_versions; j++) {
            if (strcmp(row->versions[j].key, key) == 0 && row->versions[j].xmax == 0) {
                if (row->num_versions >= MAX_VERSIONS) return -1;
                // 标记旧版本
                row->versions[j].xmax = txn;
                row->versions[j].active = 0;
                // 创建新版本
                RowVersion* nv = &row->versions[row->num_versions];
                nv->xmin = txn;
                nv->xmax = 0;
                strcpy(nv->key, key);
                strcpy(nv->value, new_val);
                nv->active = 1;
                row->num_versions++;
                e->writes++;
                printf("  [MVCC] UPDATE %s=%s→%s (txn=%u)\n", key,
                       row->versions[j].value, new_val, txn);
                return 0;
            }
        }
    }
    return -1;
}

// 删除
int mvcc_delete(MVCCEngine* e, TxnID txn, const char* key) {
    for (int i = 0; i < e->num_rows; i++) {
        Row* row = &e->rows[i];
        for (int j = 0; j < row->num_versions; j++) {
            if (strcmp(row->versions[j].key, key) == 0 && row->versions[j].xmax == 0) {
                row->versions[j].xmax = txn;
                row->versions[j].active = 0;
                e->writes++;
                printf("  [MVCC] DELETE %s (txn=%u)\n", key, txn);
                return 0;
            }
        }
    }
    return -1;
}

// 快照读
void mvcc_read(MVCCEngine* e, TxnID txn) {
    TxnInfo* reader = NULL;
    for (int i = 0; i < e->num_txns; i++) {
        if (e->txns[i].id == txn) { reader = &e->txns[i]; break; }
    }
    if (!reader) return;
    printf("  [MVCC] 快照读 txn=%u:\n", txn);
    for (int i = 0; i < e->num_rows; i++) {
        Row* row = &e->rows[i];
        for (int j = row->num_versions - 1; j >= 0; j--) {
            if (is_visible(e, &row->versions[j], reader)) {
                printf("    %s = %s (xmin=%u, xmax=%u)\n",
                       row->versions[j].key, row->versions[j].value,
                       row->versions[j].xmin, row->versions[j].xmax);
                break;
            }
        }
    }
    e->reads++;
}

int main() {
    printf("╔══════════════════════════════════════╗\n");
    printf("║   MVCC多版本并发控制                 ║\n");
    printf("╚══════════════════════════════════════╝\n\n");

    MVCCEngine* e = mvcc_create();

    // 事务1: 初始数据
    TxnID t1 = mvcc_begin(e);
    mvcc_insert(e, t1, "alice", "Beijing");
    mvcc_insert(e, t1, "bob", "Shanghai");
    mvcc_commit(e, t1);

    // 事务2: 读(看到初始数据)
    printf("\n--- 事务2: 快照读 ---\n");
    TxnID t2 = mvcc_begin(e);
    mvcc_read(e, t2);

    // 事务3: 更新(不阻塞事务2的读)
    printf("\n--- 事务3: 更新(并发) ---\n");
    TxnID t3 = mvcc_begin(e);
    mvcc_update(e, t3, "alice", "Hangzhou");

    // 事务2再次读(应看到旧版本 - REPEATABLE READ)
    printf("\n--- 事务2: 再次读(应看到旧版本) ---\n");
    mvcc_read(e, t2);

    mvcc_commit(e, t3);

    // 事务4: 读(看到新版本)
    printf("\n--- 事务4: 读(已提交后) ---\n");
    TxnID t4 = mvcc_begin(e);
    mvcc_read(e, t4);
    mvcc_commit(e, t4);

    mvcc_commit(e, t2);

    // 事务5: 删除+读
    printf("\n--- 事务5: 删除+并发读 ---\n");
    TxnID t5 = mvcc_begin(e);
    TxnID t6 = mvcc_begin(e);
    mvcc_delete(e, t5, "bob");
    printf("  事务6读(应仍看到bob):\n");
    mvcc_read(e, t6);
    mvcc_commit(e, t5);
    mvcc_commit(e, t6);

    printf("\n--- 最终统计 ---\n");
    printf("读取: %d  写入: %d  冲突: %d\n", e->reads, e->writes, e->conflicts);

    printf("\n✅ MVCC引擎运行完成\n");
    return 0;
}

🐍 Python实现：快照隔离演示

"""
快照隔离(Snapshot Isolation)完整演示
展示不同隔离级别下的并发行为
"""
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Set
from copy import deepcopy

@dataclass
class Version:
    xmin: int
    xmax: int  # 0=valid
    key: str
    value: str

@dataclass
class TxnInfo:
    id: int
    state: str = "running"
    snapshot: Set[int] = field(default_factory=set)
    write_set: Set[str] = field(default_factory=set)

class SIVersion:
    """快照隔离版本管理器"""
    def __init__(self):
        self.data: Dict[str, List[Version]] = {}
        self.txns: Dict[int, TxnInfo] = {}
        self.next_txn = 1
        self.committed: Set[int] = set()
        self.aborted: Set[int] = set()

    def begin(self) -> int:
        txn_id = self.next_txn
        self.next_txn += 1
        snapshot = set(t for t, info in self.txns.items() if info.state == "running")
        self.txns[txn_id] = TxnInfo(id=txn_id, snapshot=snapshot)
        print(f"  [Txn {txn_id}] BEGIN (快照活跃事务: {snapshot})")
        return txn_id

    def _is_visible(self, v: Version, txn_id: int) -> bool:
        txn = self.txns[txn_id]
        # xmin可见?
        if v.xmin in txn.snapshot: return False  # 创建者活跃
        if v.xmin not in self.committed: return False  # 创建者未提交
        # 创建者已提交且不在活跃快照
        if v.xmax == 0: return True
        if v.xmax == txn_id: return False
        if v.xmax in txn.snapshot: return True  # 删除者活跃→仍可见
        if v.xmax in self.committed: return False  # 删除者已提交→不可见
        return True

    def read(self, txn_id: int, key: str) -> Optional[str]:
        if key not in self.data: return None
        for v in reversed(self.data[key]):
            if self._is_visible(v, txn_id):
                print(f"  [Txn {txn_id}] READ {key}={v.value} (xmin={v.xmin}, xmax={v.xmax})")
                return v.value
        print(f"  [Txn {txn_id}] READ {key}=None (无可见版本)")
        return None

    def write(self, txn_id: int, key: str, value: str):
        txn = self.txns[txn_id]
        # 检查写-写冲突(first-committer-wins)
        for other_id, other in self.txns.items():
            if other_id != txn_id and other.state == "running" and key in other.write_set:
                print(f"  [Txn {txn_id}] WRITE CONFLICT on {key} with Txn {other_id}!")
                return False
        # 标记旧版本
        if key in self.data:
            for v in self.data[key]:
                if v.xmax == 0 and self._is_visible(v, txn_id):
                    v.xmax = txn_id
        # 创建新版本
        new_v = Version(xmin=txn_id, xmax=0, key=key, value=value)
        if key not in self.data:
            self.data[key] = []
        self.data[key].append(new_v)
        txn.write_set.add(key)
        print(f"  [Txn {txn_id}] WRITE {key}={value}")
        return True

    def commit(self, txn_id: int) -> bool:
        txn = self.txns[txn_id]
        # 验证: first-committer-wins
        for key in txn.write_set:
            for other_id, other in self.txns.items():
                if other_id != txn_id and other.state == "committed" and key in other.write_set:
                    if other.id not in txn.snapshot:  # other在快照之后提交
                        print(f"  [Txn {txn_id}] 验证失败: {key}被Txn{other_id}先提交")
                        self.abort(txn_id)
                        return False
        txn.state = "committed"
        self.committed.add(txn_id)
        print(f"  [Txn {txn_id}] COMMIT ✓")
        return True

    def abort(self, txn_id: int):
        txn = self.txns[txn_id]
        # 回滚写操作
        for key in txn.write_set:
            if key in self.data:
                for v in self.data[key]:
                    if v.xmin == txn_id: v.xmax = txn_id  # 标记为已删除
                    if v.xmax == txn_id: v.xmax = 0  # 恢复旧版本
        txn.state = "aborted"
        self.aborted.add(txn_id)
        print(f"  [Txn {txn_id}] ABORT ✗")

# 演示: 写偏序(write skew)问题
print("=== 快照隔离演示 ===\n")

si = SIVersion()
# 初始数据
t0 = si.begin()
si.write(t0, "alice_balance", "1000")
si.write(t0, "bob_balance", "1000")
si.commit(t0)

# 两个事务同时读取并更新(写偏序)
print("\n--- 写偏序问题 ---")
t1 = si.begin()
t2 = si.begin()

# 两个事务都读取两个余额
si.read(t1, "alice_balance")
si.read(t1, "bob_balance")
si.read(t2, "alice_balance")
si.read(t2, "bob_balance")

# 约束: 总余额 >= 1000
# t1: 从alice转出500 (alice=500, bob=1000, 总=1500 ≥ 1000 ✓)
# t2: 从bob转出500 (alice=1000, bob=500, 总=1500 ≥ 1000 ✓)
# 但两个都执行后: alice=500, bob=500, 总=1000 (刚好满足)
# 如果转出更多，就可能违反约束

si.write(t1, "alice_balance", "500")  # alice - 500
si.write(t2, "bob_balance", "500")    # bob - 500

si.commit(t1)
result = si.commit(t2)  # t2可能提交成功(写偏序!)

print(f"\n最终状态:")
for key in si.data:
    for v in reversed(si.data[key]):
        if v.xmax == 0:
            print(f"  {key} = {v.value}")
print(f"\n⚠️ 写偏序: 两个事务都基于过时的快照做了决策!")
print("✅ MVCC快照隔离演示完成")

🔑 关键概念总结