3. 总线嗅探机制模拟

任务目标

本案例通过一个可运行的多线程模拟程序，演示多核处理器如何借助总线嗅探机制维护缓存一致性。程序模拟两个核心（Core0、Core1）各自持有本地缓存，当某个核心写入共享内存地址时，通过总线广播写失效（Invalidate）请求，另一个核心的总线嗅探器检测到该请求后，将本地缓存行标记为失效，从而保证后续读取能获取最新值。

核心机制说明

总线嗅探的核心是写失效（Write Invalidate）协议：

  • 每个核心的缓存行包含状态位（Modified/Shared/Invalid）；
  • 写操作前，核心向总线广播Invalidate消息；
  • 其他核心的总线嗅探器监听总线，若发现自身缓存了同一地址，则将该缓存行置为Invalid；
  • 发起写操作的核心完成写入后，其缓存行状态变为Modified。

完整实现代码

import threading
import time
from enum import Enum

class CacheState(Enum):
    INVALID = 0
    SHARED = 1
    MODIFIED = 2

class Bus:
    """模拟共享总线，负责广播嗅探消息"""
    def __init__(self):
        self.lock = threading.Lock()
        self.snoop_listeners = []

    def register_core(self, core):
        self.snoop_listeners.append(core)

    def broadcast_invalidate(self, addr, requester_id):
        """广播写失效请求，所有核心（除请求者外）进行嗅探响应"""
        with self.lock:
            print(f"[总线] 广播 Invalidate(addr={addr:#x}) 来自 Core{requester_id}")
            for core in self.snoop_listeners:
                if core.id != requester_id:
                    core.snoop_invalidate(addr)

class CacheLine:
    def __init__(self, addr, value):
        self.addr = addr
        self.value = value
        self.state = CacheState.INVALID

class Core:
    def __init__(self, core_id, bus):
        self.id = core_id
        self.bus = bus
        self.cache = {}  # addr -> CacheLine
        bus.register_core(self)

    def read(self, addr):
        """读取地址，若缓存失效则从内存加载（模拟）"""
        if addr in self.cache and self.cache[addr].state != CacheState.INVALID:
            line = self.cache[addr]
            print(f"Core{self.id} 读 {addr:#x} -> 缓存命中，值={line.value} (状态={line.state.name})")
            return line.value
        else:
            # 缓存未命中，从内存加载（模拟内存值为 addr * 10）
            mem_value = addr * 10
            self.cache[addr] = CacheLine(addr, mem_value)
            self.cache[addr].state = CacheState.SHARED
            print(f"Core{self.id} 读 {addr:#x} -> 内存加载，值={mem_value} (状态=SHARED)")
            return mem_value

    def write(self, addr, new_value):
        """写入地址，先广播失效，再本地更新"""
        # 步骤1：广播写失效请求
        self.bus.broadcast_invalidate(addr, self.id)

        # 步骤2：本地更新缓存
        if addr not in self.cache:
            self.cache[addr] = CacheLine(addr, 0)
        self.cache[addr].value = new_value
        self.cache[addr].state = CacheState.MODIFIED
        print(f"Core{self.id} 写 {addr:#x} = {new_value} (状态=MODIFIED)")

    def snoop_invalidate(self, addr):
        """总线嗅探回调：检测到失效请求，若本地有该地址则置为INVALID"""
        if addr in self.cache and self.cache[addr].state != CacheState.INVALID:
            old_state = self.cache[addr].state.name
            self.cache[addr].state = CacheState.INVALID
            print(f"  [嗅探] Core{self.id} 检测到失效，缓存行 {addr:#x} 状态 {old_state} -> INVALID")
        else:
            print(f"  [嗅探] Core{self.id} 无 {addr:#x} 缓存或已失效，忽略")

def run_scenario():
    """执行一致性场景：Core0写共享地址，Core1验证嗅探效果"""
    bus = Bus()
    core0 = Core(0, bus)
    core1 = Core(1, bus)

    addr = 0x1000

    print("=== 场景开始 ===")
    # 两个核心都读取同一地址，形成共享状态
    core0.read(addr)
    core1.read(addr)

    print("\n--- Core0 执行写操作 ---")
    core0.write(addr, 42)

    print("\n--- Core1 再次读取（应触发缓存失效重载）---")
    core1.read(addr)

    print("\n=== 场景结束 ===")

if __name__ == "__main__":
    run_scenario()

运行方式

python bus_snooping_sim.py

无需额外依赖，Python 3.6+ 即可运行。

验证方法与通过条件

运行程序后，观察输出是否符合以下预期执行序列（由代码逻辑唯一确定）：

  1. Core0 和 Core1 分别读取 0x1000，均从内存加载值 40960，状态为 SHARED；
  2. Core0 写入 42 时，总线广播 Invalidate；
  3. Core1 的嗅探器响应，将其缓存行状态从 SHARED 改为 INVALID；
  4. Core1 再次读取 0x1000 时，因缓存失效重新从内存加载，得到新值 42（而非旧值 40960）。

若输出中 Core1 第二次读取的值为 42，且嗅探日志显示状态从 SHARED 变为 INVALID，则说明总线嗅探机制正确维护了缓存一致性。若 Core1 仍读到旧值 40960，则说明嗅探失效未生效，实现有误。

关键实现要点

  • 总线广播：Bus.broadcast_invalidate 遍历所有已注册核心，调用其 snoop_invalidate 方法，模拟硬件广播与并行响应；
  • 嗅探回调：Core.snoop_invalidate 是核心对总线事件的响应入口，仅当本地缓存存在且非 INVALID 时才更新状态；
  • 状态转换：写操作前先广播失效，再本地置为 MODIFIED，保证其他核心的旧副本在写入完成前已失效，避免读到过期数据。

常见问题

  • 为什么 Core1 第二次读取会从内存加载？ 因为其缓存行已被嗅探置为 INVALID，读取时判定为未命中，故重新加载内存值（模拟中内存值已更新为 42）。
  • 若两个核心同时写同一地址怎么办？ 本模拟使用总线锁串行化广播，实际硬件依赖总线仲裁机制，本案例聚焦单次写失效的嗅探流程，不涉及写写竞争。
← 上一章 📖 返回目录 下一章 →