2. MESI 总线事务验证

任务目标

构造一个最小多核缓存模型，验证两条总线事务的触发条件：

• BusRd：本地缓存未命中且只需读取时触发，本地状态由 I 变为 S（若其他核有副本）或 E（若无人持有）。
• BusRdX：本地缓存未命中且需要写入时触发，本地状态由 I 变为 M，同时使其他核的副本失效（S/E/M → I）。

验证点：事务类型、发起核的最终状态、其他核是否被失效。

核心模型

每个核维护一个状态（I/S/E/M）。总线是共享的：一次事务会广播给所有核，被命中的核按 MESI 规则响应。下面代码用两个核覆盖三种关键场景：读未命中（他核有副本）、读未命中（无人持有）、写未命中（他核有副本）。

完整代码

from enum import Enum

class S(Enum):
    I = "Invalid"
    S = "Shared"
    E = "Exclusive"
    M = "Modified"

class Bus:
    def __init__(self):
        self.log = []
    def transaction(self, kind, requester, caches):
        # kind: "BusRd" or "BusRdX"
        others = [c for i, c in enumerate(caches) if i != requester]
        hit = any(c.state in (S.S, S.E, S.M) for c in others)
        if kind == "BusRdX":
            for c in others:
                if c.state != S.I:
                    c.state = S.I          # 写未命中：他核副本失效
            caches[requester].state = S.M
        else:  # BusRd
            caches[requester].state = S.S if hit else S.E
        self.log.append((kind, requester, hit,
                         [c.state.value for c in caches]))

class Cache:
    def __init__(self, name):
        self.name = name
        self.state = S.I

def run(scenario):
    caches = [Cache("C0"), Cache("C1")]
    bus = Bus()
    for kind, requester in scenario:
        bus.transaction(kind, requester, caches)
    return bus.log

# 场景1：C0 先读（无人持有 -> E），C1 再读（他核有 -> S）
print("S1:", run([("BusRd", 0), ("BusRd", 1)]))
# 场景2：C0 读（E），C1 写未命中（BusRdX，C0 失效）
print("S2:", run([("BusRd", 0), ("BusRdX", 1)]))
# 场景3：C0 写未命中（BusRdX，无人持有 -> M）
print("S3:", run([("BusRdX", 0)]))

运行与验证

直接执行 python mesi_bus.py。日志每项为 (事务, 发起核, 是否命中他核, 各核状态)。

• S1 期望：第一次 BusRd 后 C0=E；第二次 BusRd 命中他核，C0=S、C1=S。
• S2 期望：BusRd 后 C0=E；BusRdX 后 C1=M，C0 被失效为 I。
• S3 期望：BusRdX 后 C0=M，无他核失效。

通过条件：三条日志的事务类型、发起核状态、他核失效行为与上述一致，即证明 BusRd/BusRdX 的触发条件与本地状态迁移正确。

关键点

BusRd 只在读未命中时触发，命中他核则进 S，否则进 E；BusRdX 在写未命中时触发，无论他核是否持有都进 M，并强制他核失效。若需扩展，可在 transaction 中增加对本地已命中（非 I）的短路判断，避免重复发起总线事务。