第12课：WAL预写日志

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

📖 课程概述

WAL(Write-Ahead Logging)预写日志是保证事务原子性和持久性的核心技术。核心原则：在修改数据页之前，必须先将对应的日志记录写入持久存储。本课深入实现WAL的日志格式、刷盘策略、检查点机制和崩溃恢复。

📝 WAL核心原理

💻 C语言实现：WAL与ARIES恢复

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

#define PAGE_SIZE     4096
#define MAX_PAGES     32
#define MAX_LOG_ENTRY 512
#define MAX_TXN       64
#define MAX_DATA      128

// 日志记录类型
typedef enum {
    LOG_BEGIN, LOG_INSERT, LOG_UPDATE, LOG_DELETE,
    LOG_COMMIT, LOG_ABORT, LOG_CHECKPOINT, LOG_CLR  // Compensation Log Record
} LogType;

// LSN类型
typedef uint32_t LSN;

// 日志记录
typedef struct {
    LSN      lsn;
    uint32_t txn_id;
    LogType  type;
    uint32_t page_id;
    char     undo_data[MAX_DATA]; // 用于undo的数据
    char     redo_data[MAX_DATA]; // 用于redo的数据
    LSN      prev_lsn;            // 同一事务的前一条LSN
} LogRecord;

// 页面(带LSN)
typedef struct {
    uint32_t page_id;
    LSN      page_lsn;  // 页面上最后的LSN(用于Redo判断)
    char     data[PAGE_SIZE - 8];
    int      dirty;
} Page;

// WAL管理器
typedef struct {
    LogRecord log[MAX_LOG_ENTRY];
    int       log_count;
    LSN       next_lsn;
    Page*     pages[MAX_PAGES];
    int       num_pages;
    LSN       last_checkpoint_lsn;
    uint32_t  active_txns[MAX_TXN]; // 活跃事务
    int       num_active;
} WALSystem;

WALSystem* wal_system_create() {
    WALSystem* sys = calloc(1, sizeof(WALSystem));
    sys->next_lsn = 1;
    sys->last_checkpoint_lsn = 0;
    printf("[WAL] 系统初始化\n");
    return sys;
}

Page* wal_get_page(WALSystem* sys, uint32_t page_id) {
    for (int i = 0; i < sys->num_pages; i++) {
        if (sys->pages[i]->page_id == page_id) return sys->pages[i];
    }
    Page* p = calloc(1, sizeof(Page));
    p->page_id = page_id;
    sys->pages[sys->num_pages++] = p;
    return p;
}

// 写日志
LSN wal_log(WALSystem* sys, uint32_t txn_id, LogType type,
            uint32_t page_id, const char* undo, const char* redo,
            LSN prev_lsn) {
    LogRecord* rec = &sys->log[sys->log_count];
    rec->lsn = sys->next_lsn++;
    rec->txn_id = txn_id;
    rec->type = type;
    rec->page_id = page_id;
    if (undo) strncpy(rec->undo_data, undo, MAX_DATA - 1);
    if (redo) strncpy(rec->redo_data, redo, MAX_DATA - 1);
    rec->prev_lsn = prev_lsn;
    sys->log_count++;

    const char* type_str[] = {"BEGIN","INSERT","UPDATE","DELETE",
                              "COMMIT","ABORT","CHECKPOINT","CLR"};
    printf("  [WAL] LSN=%u TXN=%u %s", rec->lsn, txn_id, type_str[type]);
    if (page_id != UINT32_MAX) printf(" PAGE=%u", page_id);
    printf("\n");
    return rec->lsn;
}

// 检查点
void wal_checkpoint(WALSystem* sys) {
    printf("[Checkpoint] 创建检查点...\n");
    // 记录当前活跃事务和脏页
    printf("  活跃事务: ");
    for (int i = 0; i < sys->num_active; i++)
        printf("%u ", sys->active_txns[i]);
    printf("\n  脏页: ");
    for (int i = 0; i < sys->num_pages; i++)
        if (sys->pages[i]->dirty) printf("%u ", sys->pages[i]->page_id);
    printf("\n");

    // 刷所有脏页
    for (int i = 0; i < sys->num_pages; i++) {
        if (sys->pages[i]->dirty) {
            printf("  Flush page %u (LSN=%u)\n",
                   sys->pages[i]->page_id, sys->pages[i]->page_lsn);
            sys->pages[i]->dirty = 0;
        }
    }
    sys->last_checkpoint_lsn = sys->next_lsn - 1;
    wal_log(sys, 0, LOG_CHECKPOINT, UINT32_MAX, NULL, NULL, 0);
    printf("[Checkpoint] 完成, LSN=%u\n", sys->last_checkpoint_lsn);
}

// ARIES恢复算法
void aries_recovery(WALSystem* sys) {
    printf("\n╔══════════════════════════════════╗\n");
    printf("║   ARIES 崩溃恢复                 ║\n");
    printf("╚══════════════════════════════════╝\n");

    // Phase 1: Analysis (分析)
    printf("\n--- Phase 1: Analysis ---\n");
    uint32_t active[MAX_TXN];
    int num_active = 0;
    LSN min_lsn = sys->last_checkpoint_lsn > 0 ? sys->last_checkpoint_lsn : 1;
    printf("从LSN %u开始扫描\n", min_lsn);

    for (int i = 0; i < sys->log_count; i++) {
        LogRecord* r = &sys->log[i];
        if (r->lsn < min_lsn) continue;
        if (r->type == LOG_BEGIN) {
            active[num_active++] = r->txn_id;
            printf("  发现活跃事务: %u\n", r->txn_id);
        } else if (r->type == LOG_COMMIT || r->type == LOG_ABORT) {
            printf("  事务 %u 已%s\n", r->txn_id,
                   r->type == LOG_COMMIT ? "提交" : "中止");
        }
    }

    // Phase 2: Redo (重做)
    printf("\n--- Phase 2: Redo ---\n");
    for (int i = 0; i < sys->log_count; i++) {
        LogRecord* r = &sys->log[i];
        if (r->lsn < min_lsn) continue;
        if (r->type != LOG_INSERT && r->type != LOG_UPDATE && r->type != LOG_DELETE)
            continue;
        // 检查页面LSN是否需要Redo
        Page* page = wal_get_page(sys, r->page_id);
        if (page->page_lsn < r->lsn) {
            printf("  Redo LSN=%u: PAGE %u (page_lsn=%u < log_lsn=%u)\n",
                   r->lsn, r->page_id, page->page_lsn, r->lsn);
            // 应用redo
            if (r->redo_data[0]) {
                snprintf(page->data, PAGE_SIZE - 8, "%s", r->redo_data);
            }
            page->page_lsn = r->lsn;
        } else {
            printf("  Skip LSN=%u: PAGE %u (已是最新)\n", r->lsn, r->page_id);
        }
    }
    printf("Redo阶段完成\n");

    // Phase 3: Undo (撤销)
    printf("\n--- Phase 3: Undo ---\n");
    // 找出需要undo的事务(活跃但未提交/中止的)
    for (int i = sys->log_count - 1; i >= 0; i--) {
        LogRecord* r = &sys->log[i];
        if (r->lsn < min_lsn) continue;
        // 只undo未提交事务的数据操作
        int is_committed = 0;
        for (int j = 0; j < sys->log_count; j++) {
            if (sys->log[j].txn_id == r->txn_id &&
                (sys->log[j].type == LOG_COMMIT || sys->log[j].type == LOG_ABORT)) {
                is_committed = 1;
                break;
            }
        }
        if (!is_committed && (r->type == LOG_INSERT || r->type == LOG_UPDATE ||
                              r->type == LOG_DELETE)) {
            printf("  Undo LSN=%u: TXN %u %s\n", r->lsn, r->txn_id,
                   r->type == LOG_INSERT ? "INSERT" :
                   r->type == LOG_UPDATE ? "UPDATE" : "DELETE");
            // 写CLR(补偿日志)
            wal_log(sys, r->txn_id, LOG_CLR, r->page_id,
                    r->redo_data, r->undo_data, r->prev_lsn);
            // 应用undo
            if (r->undo_data[0]) {
                Page* page = wal_get_page(sys, r->page_id);
                snprintf(page->data, PAGE_SIZE - 8, "%s", r->undo_data);
            }
        }
    }
    printf("Undo阶段完成\n");
    printf("\n[ARIES] 恢复完成!\n");
}

int main() {
    printf("╔══════════════════════════════════════╗\n");
    printf("║   WAL预写日志 + ARIES恢复            ║\n");
    printf("╚══════════════════════════════════════╝\n\n");

    WALSystem* sys = wal_system_create();

    // 事务1: 提交
    printf("--- 事务1 ---\n");
    sys->active_txns[sys->num_active++] = 1;
    LSN lsn1 = wal_log(sys, 1, LOG_BEGIN, UINT32_MAX, NULL, NULL, 0);
    LSN lsn2 = wal_log(sys, 1, LOG_INSERT, 0, NULL, "alice=Beijing", lsn1);
    LSN lsn3 = wal_log(sys, 1, LOG_UPDATE, 0, "alice=Beijing", "alice=Hangzhou", lsn2);
    Page* p0 = wal_get_page(sys, 0);
    snprintf(p0->data, PAGE_SIZE - 8, "alice=Hangzhou");
    p0->page_lsn = lsn3; p0->dirty = 1;
    LSN lsn4 = wal_log(sys, 1, LOG_COMMIT, UINT32_MAX, NULL, NULL, lsn3);

    // 事务2: 也提交
    printf("\n--- 事务2 ---\n");
    sys->active_txns[sys->num_active++] = 2;
    LSN lsn5 = wal_log(sys, 2, LOG_BEGIN, UINT32_MAX, NULL, NULL, 0);
    LSN lsn6 = wal_log(sys, 2, LOG_INSERT, 1, NULL, "bob=Shanghai", lsn5);
    Page* p1 = wal_get_page(sys, 1);
    snprintf(p1->data, PAGE_SIZE - 8, "bob=Shanghai");
    p1->page_lsn = lsn6; p1->dirty = 1;
    LSN lsn7 = wal_log(sys, 2, LOG_COMMIT, UINT32_MAX, NULL, NULL, lsn6);

    // 检查点
    printf("\n--- 检查点 ---\n");
    wal_checkpoint(sys);

    // 事务3: 未提交(模拟崩溃)
    printf("\n--- 事务3 (未提交) ---\n");
    sys->active_txns[sys->num_active++] = 3;
    LSN lsn8 = wal_log(sys, 3, LOG_BEGIN, UINT32_MAX, NULL, NULL, 0);
    LSN lsn9 = wal_log(sys, 3, LOG_UPDATE, 0, "alice=Hangzhou", "alice=Shenzhen", lsn8);
    p0->page_lsn = lsn9; p0->dirty = 1;
    snprintf(p0->data, PAGE_SIZE - 8, "alice=Shenzhen");
    // 崩溃! 事务3未提交

    printf("\n[模拟崩溃] 事务3未提交，系统崩溃!\n");
    // 清除内存状态(模拟重启)
    for (int i = 0; i < sys->num_pages; i++) {
        sys->pages[i]->dirty = 0;
        // 模拟页面可能不是最新的
    }

    // ARIES恢复
    aries_recovery(sys);

    printf("\n--- 恢复后数据 ---\n");
    for (int i = 0; i < sys->num_pages; i++) {
        printf("Page %u: %s (LSN=%u)\n",
               sys->pages[i]->page_id, sys->pages[i]->data, sys->pages[i]->page_lsn);
    }

    printf("\n✅ WAL+ARIES恢复运行完成\n");
    return 0;
}

🐍 Python实现：WAL性能优化

"""
WAL性能优化：组提交(Group Commit)
"""
import time, threading, queue
from dataclasses import dataclass, field
from typing import List, Dict

@dataclass
class LogRecord:
    lsn: int
    txn_id: int
    op: str
    data: str
    prev_lsn: int = 0

class WALWriter:
    def __init__(self, group_size=10, group_timeout_ms=5):
        self.log: List[LogRecord] = []
        self.next_lsn = 1
        self.group_size = group_size
        self.group_timeout_ms = group_timeout_ms
        self.pending = queue.Queue()
        self.fsync_count = 0
        self.total_records = 0
        self._running = True
        self.writer_thread = threading.Thread(target=self._writer_loop, daemon=True)
        self.writer_thread.start()

    def _writer_loop(self):
        batch = []
        while self._running:
            try:
                rec = self.pending.get(timeout=self.group_timeout_ms / 1000)
                batch.append(rec)
                # 收集一批
                while len(batch) < self.group_size:
                    try:
                        rec = self.pending.get_nowait()
                        batch.append(rec)
                    except queue.Empty:
                        break
                # 批量写入
                if batch:
                    self._flush_batch(batch)
                    batch = []
            except queue.Empty:
                if batch:
                    self._flush_batch(batch)
                    batch = []

    def _flush_batch(self, batch):
        for rec in batch:
            rec.lsn = self.next_lsn
            self.log.append(rec)
            self.next_lsn += 1
            self.total_records += 1
        self.fsync_count += 1
        # 模拟fsync
        time.sleep(0.001)

    def append(self, txn_id, op, data, prev_lsn=0):
        rec = LogRecord(lsn=0, txn_id=txn_id, op=op, data=data, prev_lsn=prev_lsn)
        self.pending.put(rec)
        return rec

    def stop(self):
        self._running = False
        self.writer_thread.join()

    def stats(self):
        return {
            "total_records": self.total_records,
            "fsync_count": self.fsync_count,
            "avg_batch_size": self.total_records / max(1, self.fsync_count)
        }

# 对比测试：单条提交 vs 组提交
def benchmark_wal(num_txns, group_size):
    wal = WALWriter(group_size=group_size)
    t0 = time.perf_counter()
    for i in range(num_txns):
        txn = i + 1
        wal.append(txn, "BEGIN", "")
        wal.append(txn, "INSERT", f"key_{i}=val_{i}")
        wal.append(txn, "COMMIT", "")
    time.sleep(0.5)  # 等待刷完
    elapsed = time.perf_counter() - t0
    s = wal.stats()
    wal.stop()
    return elapsed * 1000, s["fsync_count"], s["avg_batch_size"]

print("WAL组提交性能对比:")
print(f"{'模式':>15} | {'耗时ms':>8} | {'fsync次数':>8} | {'平均批次':>8}")
print("-" * 55)
for gs in [1, 5, 10, 20, 50]:
    ms, fsyncs, avg = benchmark_wal(200, gs)
    mode = f"组提交({gs})"
    print(f"{mode:>15} | {ms:>7.1f} | {fsyncs:>8} | {avg:>7.1f}")

print("\n✅ WAL性能优化演示完成")

🔑 关键概念总结