第02课：磁盘与缓冲区管理

存储引擎 第2课 / 共25课

📖 课程概述

磁盘I/O是数据库性能的最大瓶颈。理解磁盘结构和缓冲区管理，是优化数据库性能的基础。本课深入探讨磁盘的物理特性、操作系统I/O栈、以及数据库缓冲池的设计原理。

💿 磁盘物理结构

关键性能参数

🗄️ 缓冲池架构

LRU淘汰算法

最近最少使用(Least Recently Used)：淘汰最长时间未被访问的页面。

Clock淘汰算法（LRU的近似）

将缓冲池页面排成环形，每个页面有一个引用位(reference bit)。时钟指针扫描，跳过引用位为1的页面（清除为0），淘汰引用位为0的页面。

💻 C语言实现：完整缓冲池管理器

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

#define PAGE_SIZE     4096
#define BUFFER_SIZE   8    // 缓冲池大小（页数）
#define DISK_PAGES    32   // 磁盘总页数

typedef struct {
    uint32_t page_id;
    uint8_t  data[PAGE_SIZE];
    int      dirty;
    int      pin_count;
    int      ref_bit;       // Clock算法引用位
    int      valid;         // 是否有有效数据
} Frame;

typedef struct {
    Frame    frames[BUFFER_SIZE];
    int      clock_hand;    // Clock指针
    int      hits;
    int      misses;
    FILE*    disk_file;     // 模拟磁盘文件
} BufferPool;

// 模拟磁盘延迟(微秒)
void simulate_disk_delay() {
    struct timespec ts = {0, 100000}; // 0.1ms
    nanosleep(&ts, NULL);
}

BufferPool* buffer_pool_init() {
    BufferPool* bp = calloc(1, sizeof(BufferPool));
    bp->disk_file = tmpfile();
    // 初始化磁盘数据
    for (int i = 0; i < DISK_PAGES; i++) {
        fseek(bp->disk_file, i * PAGE_SIZE, SEEK_SET);
        char header[32];
        snprintf(header, sizeof(header), "PAGE_%d_DATA", i);
        fwrite(header, 1, strlen(header), bp->disk_file);
    }
    printf("[BufferPool] 初始化完成，%d帧，磁盘%d页\n",
           BUFFER_SIZE, DISK_PAGES);
    return bp;
}

// Clock算法淘汰一帧
int clock_evict(BufferPool* bp) {
    int start = bp->clock_hand;
    int scanned = 0;
    while (scanned < BUFFER_SIZE * 2) {
        Frame* f = &bp->frames[bp->clock_hand];
        if (!f->valid) {
            // 空帧，直接使用
            int chosen = bp->clock_hand;
            bp->clock_hand = (bp->clock_hand + 1) % BUFFER_SIZE;
            return chosen;
        }
        if (f->pin_count > 0) {
            // 被钉住，跳过
            bp->clock_hand = (bp->clock_hand + 1) % BUFFER_SIZE;
            scanned++;
            continue;
        }
        if (f->ref_bit) {
            // 引用位为1，清零，给第二次机会
            f->ref_bit = 0;
            bp->clock_hand = (bp->clock_hand + 1) % BUFFER_SIZE;
            scanned++;
            continue;
        }
        // 引用位为0，未钉住 → 淘汰
        int chosen = bp->clock_hand;
        if (f->dirty) {
            // 写回脏页
            fseek(bp->disk_file, f->page_id * PAGE_SIZE, SEEK_SET);
            fwrite(f->data, 1, PAGE_SIZE, bp->disk_file);
            printf("  [Clock] 淘汰脏页 %u → 写回磁盘\n", f->page_id);
            f->dirty = 0;
        } else {
            printf("  [Clock] 淘汰干净页 %u\n", f->page_id);
        }
        bp->clock_hand = (bp->clock_hand + 1) % BUFFER_SIZE;
        return chosen;
    }
    return -1; // 所有帧都被钉住
}

// 查找页面是否在缓冲池
int find_in_pool(BufferPool* bp, uint32_t page_id) {
    for (int i = 0; i < BUFFER_SIZE; i++) {
        if (bp->frames[i].valid && bp->frames[i].page_id == page_id) {
            return i;
        }
    }
    return -1;
}

// 获取页面
uint8_t* buffer_get_page(BufferPool* bp, uint32_t page_id) {
    int idx = find_in_pool(bp, page_id);
    if (idx >= 0) {
        bp->hits++;
        bp->frames[idx].ref_bit = 1;
        bp->frames[idx].pin_count++;
        printf("  [Hit] 页 %u 在帧 %d\n", page_id, idx);
        return bp->frames[idx].data;
    }
    // 缺页
    bp->misses++;
    int frame_idx = clock_evict(bp);
    if (frame_idx < 0) {
        printf("  [Error] 无法分配帧!\n");
        return NULL;
    }
    Frame* f = &bp->frames[frame_idx];
    // 从磁盘读取
    simulate_disk_delay();
    fseek(bp->disk_file, page_id * PAGE_SIZE, SEEK_SET);
    fread(f->data, 1, PAGE_SIZE, bp->disk_file);
    f->page_id = page_id;
    f->valid = 1;
    f->dirty = 0;
    f->pin_count = 1;
    f->ref_bit = 1;
    printf("  [Miss] 从磁盘加载页 %u → 帧 %d\n", page_id, frame_idx);
    return f->data;
}

// 释放页面
void buffer_unpin(BufferPool* bp, uint32_t page_id) {
    int idx = find_in_pool(bp, page_id);
    if (idx >= 0 && bp->frames[idx].pin_count > 0) {
        bp->frames[idx].pin_count--;
    }
}

// 标记脏页
void buffer_mark_dirty(BufferPool* bp, uint32_t page_id) {
    int idx = find_in_pool(bp, page_id);
    if (idx >= 0) {
        bp->frames[idx].dirty = 1;
    }
}

// 强制刷脏
void buffer_flush_all(BufferPool* bp) {
    for (int i = 0; i < BUFFER_SIZE; i++) {
        if (bp->frames[i].valid && bp->frames[i].dirty) {
            fseek(bp->disk_file,
                  bp->frames[i].page_id * PAGE_SIZE, SEEK_SET);
            fwrite(bp->frames[i].data, 1, PAGE_SIZE, bp->disk_file);
            bp->frames[i].dirty = 0;
            printf("  [Flush] 帧 %d (页 %u) 写回磁盘\n",
                   i, bp->frames[i].page_id);
        }
    }
}

// 统计
void buffer_stats(BufferPool* bp) {
    int total = bp->hits + bp->misses;
    double hit_rate = total > 0 ? (double)bp->hits / total * 100 : 0;
    printf("\n=== 缓冲池统计 ===\n");
    printf("命中: %d  缺页: %d  命中率: %.1f%%\n",
           bp->hits, bp->misses, hit_rate);
    printf("帧状态:\n");
    for (int i = 0; i < BUFFER_SIZE; i++) {
        Frame* f = &bp->frames[i];
        if (f->valid) {
            printf("  帧%d: 页%u dirty=%d pin=%d ref=%d\n",
                   i, f->page_id, f->dirty, f->pin_count, f->ref_bit);
        } else {
            printf("  帧%d: (空)\n", i);
        }
    }
}

// ========== 工作负载模拟 ==========
void simulate_oltp(BufferPool* bp, int ops) {
    printf("\n--- 模拟OLTP工作负载 (%d次操作) ---\n", ops);
    // 80/20法则：80%的请求访问20%的热点页
    uint32_t hot_pages[] = {0, 1, 2, 3, 4, 5, 6}; // 7个热点页
    for (int i = 0; i < ops; i++) {
        uint32_t page_id;
        if (rand() % 100 < 80) {
            // 80%访问热点页
            page_id = hot_pages[rand() % 7];
        } else {
            // 20%访问冷门页
            page_id = 7 + rand() % (DISK_PAGES - 7);
        }
        uint8_t* page = buffer_get_page(bp, page_id);
        if (page && rand() % 10 < 3) {
            // 30%写操作
            buffer_mark_dirty(bp, page_id);
            snprintf((char*)page, 32, "MODIFIED_%d", i);
        }
        buffer_unpin(bp, page_id);
    }
}

void simulate_sequential_scan(BufferPool* bp) {
    printf("\n--- 模拟顺序扫描 ---\n");
    for (int i = 0; i < DISK_PAGES; i++) {
        uint8_t* page = buffer_get_page(bp, i);
        buffer_unpin(bp, i);
    }
}

int main() {
    srand(42);
    printf("╔══════════════════════════════════════╗\n");
    printf("║  Buffer Pool Manager - Clock算法     ║\n");
    printf("╚══════════════════════════════════════╝\n\n");

    BufferPool* bp = buffer_pool_init();

    // 1. 基本操作测试
    printf("=== 基本操作测试 ===\n");
    uint8_t* p = buffer_get_page(bp, 0);
    buffer_mark_dirty(bp, 0);
    buffer_unpin(bp, 0);

    p = buffer_get_page(bp, 1);
    buffer_unpin(bp, 1);

    p = buffer_get_page(bp, 0);  // 应该命中
    buffer_unpin(bp, 0);

    // 2. 重新初始化，运行完整模拟
    BufferPool* bp2 = buffer_pool_init();
    simulate_oltp(bp2, 50);
    buffer_stats(bp2);

    // 3. 顺序扫描
    BufferPool* bp3 = buffer_pool_init();
    simulate_sequential_scan(bp3);
    buffer_stats(bp3);

    // 4. 刷脏测试
    printf("\n=== 刷脏测试 ===\n");
    buffer_flush_all(bp2);

    printf("\n✅ 缓冲池管理器运行完成\n");
    return 0;
}

🐍 Python实现：LRU vs Clock对比

"""
缓冲池淘汰算法对比：LRU vs Clock
"""
import random
from collections import OrderedDict
from dataclasses import dataclass
from typing import Optional

@dataclass
class Frame:
    page_id: int = -1
    dirty: bool = False
    ref_bit: bool = False

class LRUBufferPool:
    """精确LRU实现"""
    def __init__(self, capacity: int):
        self.capacity = capacity
        self.cache: OrderedDict = OrderedDict()
        self.hits = 0
        self.misses = 0
        self.disk_writes = 0

    def get(self, page_id: int, write: bool = False) -> bool:
        if page_id in self.cache:
            self.hits += 1
            self.cache.move_to_end(page_id)
            if write:
                self.cache[page_id] = True  # dirty
            return True
        self.misses += 1
        if len(self.cache) >= self.capacity:
            _, dirty = self.cache.popitem(last=False)
            if dirty:
                self.disk_writes += 1
        self.cache[page_id] = write
        return False

class ClockBufferPool:
    """Clock算法实现"""
    def __init__(self, capacity: int):
        self.capacity = capacity
        self.frames: list = [Frame() for _ in range(capacity)]
        self.page_to_frame: dict = {}
        self.hand = 0
        self.hits = 0
        self.misses = 0
        self.disk_writes = 0

    def get(self, page_id: int, write: bool = False) -> bool:
        if page_id in self.page_to_frame:
            self.hits += 1
            idx = self.page_to_frame[page_id]
            self.frames[idx].ref_bit = True
            if write:
                self.frames[idx].dirty = True
            return True
        self.misses += 1
        # 寻找可淘汰帧
        idx = self._evict()
        old_page = self.frames[idx].page_id
        if old_page >= 0:
            del self.page_to_frame[old_page]
            if self.frames[idx].dirty:
                self.disk_writes += 1
        self.frames[idx] = Frame(page_id=page_id, dirty=write, ref_bit=True)
        self.page_to_frame[page_id] = idx
        return False

    def _evict(self) -> int:
        # 找空帧
        for i in range(self.capacity):
            if self.frames[i].page_id < 0:
                return i
        # Clock扫描
        while True:
            f = self.frames[self.hand]
            if f.ref_bit:
                f.ref_bit = False
            else:
                chosen = self.hand
                self.hand = (self.hand + 1) % self.capacity
                return chosen
            self.hand = (self.hand + 1) % self.capacity

def generate_workload(n: int, hot_ratio: float = 0.8,
                      hot_pages: int = 5, total_pages: int = 50):
    """生成模拟工作负载"""
    ops = []
    for _ in range(n):
        if random.random() < hot_ratio:
            pid = random.randint(0, hot_pages - 1)
        else:
            pid = random.randint(hot_pages, total_pages - 1)
        write = random.random() < 0.3
        ops.append((pid, write))
    return ops

def run_comparison():
    random.seed(42)
    CAPACITIES = [4, 8, 16, 32]
    OPS = 10000

    print(f"{'容量':>6} | {'LRU命中%':>8} | {'Clock命中%':>9} | {'LRU写回':>6} | {'Clock写回':>8}")
    print("-" * 55)

    for cap in CAPACITIES:
        ops = generate_workload(OPS)
        lru = LRUBufferPool(cap)
        clk = ClockBufferPool(cap)
        for pid, write in ops:
            lru.get(pid, write)
            clk.get(pid, write)
        lru_total = lru.hits + lru.misses
        clk_total = clk.hits + clk.misses
        print(f"{cap:>6} | {lru.hits/lru_total*100:>7.1f}% | {clk.hits/clk_total*100:>8.1f}% | {lru.disk_writes:>6} | {clk.disk_writes:>8}")

run_comparison()

📊 性能分析

🔑 关键概念总结