I/O 的三种方式 -- 轮询、中断、DMA
上一讲我们了解了 CPU 和内存之间如何通过总线通信。但计算机不只是 CPU 加内存——它还需要和键盘、鼠标、硬盘、网卡等各种外部设备（I/O 设备）交换数据。
CPU 与外部设备交换数据有三种基本方式：轮询（Polling）、中断（Interrupt）、DMA（直接存储器访问）。三者的效率天差地别——选择哪种方式，往往决定了整个系统的性能表现。
本讲用三种生活场景帮你建立直觉，再用 Python 代码直观对比三种方式的效率差异。
三个生活场景类比
场景一：轮询——每隔 5 秒去看看水开了没
你在厨房烧水，但没有水壶哨子。你每隔 5 秒就从客厅跑到厨房看一眼：「水开了吗？水开了吗？水开了吗？」
在此期间，你完全没法安心看电视，因为每隔几秒就要跑一趟。你几乎把所有时间都花在了「检查状态」上。
这就是轮询：CPU 不断循环检查设备是否就绪，把时间都浪费在无意义的查询上。
场景二：中断——水壶哨子响了再去关火
这回水壶带哨子。你把水放上炉子后，回到客厅安心看电视。水烧开了，哨子尖锐地响起来。你暂停看电视，去厨房关火，然后继续看电视。
你绝大多数时间都在做有意义的事情（看电视），只在关键时刻被打断一小会儿。
这就是中断：CPU 正常执行程序，设备就绪时主动发信号通知 CPU。CPU 暂停手头工作，处理设备数据，然后恢复之前的工作。
场景三：DMA——直接找搬家公司，你根本不用动手
你要搬家，有一整车家具要从旧房子运到新房子。如果一件一件自己搬（中断方式），你得来回跑几十趟。
你雇了一家搬家公司（DMA 控制器），告诉他们：「旧房子地址是 X，新房子地址是 Y，一共有 Z 件家具。」然后你继续忙自己的事。搬家公司搞定一切后，给你打个电话说「搬完了」。
这就是DMA：专门的 DMA 控制器直接管理内存和外设之间的数据传输，CPU 完全不用参与。传输完成后，DMA 控制器才给 CPU 发一个中断通知。
三种方式的本质区别
| 特性 | 轮询 (Polling) | 中断 (Interrupt) | DMA |
|---|---|---|---|
| CPU 参与度 | 全程忙碌，不停检查 | 传输时短暂参与 | 几乎不参与，只收完成通知 |
| 数据谁搬 | CPU 亲自搬每一个字节 | CPU 亲自搬每一个字节 | DMA 控制器搬，CPU 不管 |
| CPU 空闲时间 | 几乎没有 | 大部分时间空闲 | 几乎全部时间空闲 |
| 实现复杂度 | 最简单 | 需要中断控制器硬件 | 需要 DMA 控制器硬件 |
| 响应延迟 | 取决于轮询间隔 | 微秒级 | 取决于 DMA 传输总时长 |
| 适用数据量 | 极少量 | 中小量（KB 级） | 大量（MB/GB 级） |
| 典型场景 | 简单嵌入式、极短等待 | 键盘输入、鼠标移动、网络包到达 | 磁盘读写、显卡帧缓冲、网卡大包 |
你每次敲键盘触发的是中断，你拷贝大文件时用的是 DMA。现代操作系统会根据数据量和场景，自动在三种方式之间选择最优方案。
时间轴可视化：CPU 忙碌与空闲的对比
假设需要从硬盘读取 4 个数据块到内存。我们用 ASCII 时间轴来直观对比三种方式下 CPU 的状态：
任务: 从硬盘读取 4 个数据块到内存
【轮询 - CPU 几乎全程忙碌】
时间: 0----1----2----3----4----5----6----7----8----9----10--->
CPU: [查][查][搬][查][查][搬][查][查][搬][查][查][搬]
忙 忙 忙 忙 忙 忙 忙 忙 忙 忙 忙 忙
↑ 几乎没有任何空闲
【中断 - CPU 大部分时间空闲】
时间: 0----1----2----3----4----5----6----7----8----9----10--->
CPU: [工作][等待...][搬][工作] [等待...][搬][工作]
忙 闲 忙 忙 闲 忙 忙
↑ 中断触发 ↑ 中断触发
【DMA - CPU 几乎全程空闲】
时间: 0----1----2----3----4----5----6----7----8----9----10--->
CPU: [设置DMA][工作][工作][工作] [被通知完成]
忙 忙 忙 忙 忙
DMA: [搬][搬][搬][搬] ↑ DMA 完成后中断
忙 忙 忙 忙
↑ CPU 完全不管传输过程
一目了然：轮询下 CPU 被绑死，中断下 CPU 只被短暂打断，DMA 下 CPU 几乎完全解放。
代码演示：Python 模拟三种 I/O 方式并打印时间轴
下面这段代码分别用轮询、中断、DMA 三种方式模拟同一批数据的传输过程，统计 CPU 忙碌/空闲的时间占比，并用字符画时间轴直观展示差异。
实例
"""
I/O 三种方式对比：轮询 vs 中断 vs DMA（runoob 演示）
本程序模拟 CPU 从「外部设备」传输 10 个数据块到内存的场景，
分别用轮询、中断、DMA 三种方式完成，打印时间轴直观对比 CPU 状态。
核心指标: CPU 忙碌时间占比。占比越高，说明 CPU 被 I/O 拖累越严重。
"""
import time
import random
class IODevice:
"""模拟外部 I/O 设备（如硬盘、键盘等）"""
def __init__(self, name="硬盘", response_time=4):
"""
name: 设备名称
response_time: 设备响应所需的「时间单位」数
"""
self.name = name
self.response_time = response_time # 每个数据块的设备准备时间
self.data_buffer = [] # 设备内部数据缓冲区
self.ready = False # 设备是否就绪
def prepare_data(self, block_id):
"""
设备准备数据——模拟设备内部处理过程。
实际硬件中这一步是硬盘寻道、网卡收包等操作。
"""
self.ready = False
data = f"DATA_BLOCK_{block_id:02d}"
self.data_buffer.append(data)
# 模拟设备在不同时间单位后准备好
return data
def is_ready(self):
"""CPU 轮询时检查此状态"""
return self.ready
def mark_ready(self):
"""设备完成准备，发送中断信号"""
self.ready = True
def get_data(self):
"""从设备缓冲区取走一个数据块"""
if self.data_buffer:
return self.data_buffer.pop(0)
return None
class Timeline:
"""
时间轴记录器——记录每个时间单位 CPU 和 DMA 的状态
用于最终绘制 ASCII 时间轴
"""
def __init__(self):
self.cpu_states = [] # 每个时间单位 CPU 的状态: 'BUSY' 或 'IDLE'
self.dma_states = [] # 每个时间单位 DMA 的状态
self.events = [] # 事件日志: (时间, 描述)
def record_cpu(self, state):
self.cpu_states.append(state)
def record_dma(self, state):
self.dma_states.append(state)
def log_event(self, time_unit, desc):
self.events.append((time_unit, desc))
def draw(self, title="时间轴"):
"""绘制 ASCII 时间轴"""
total = len(self.cpu_states)
if total == 0:
print(" (无时间数据)")
return
print(f"\n ╔══ {title} ══╗")
print(f" ║ 总时间单位: {total}")
# CPU 状态行
cpu_line = "CPU: "
for s in self.cpu_states:
cpu_line += "[" + ("忙" if s == "BUSY" else "闲") + "]"
print(f" ║ {cpu_line}")
# DMA 状态行（如果有 DMA 活动）
if any(s == "BUSY" for s in self.dma_states):
dma_line = "DMA: "
for s in self.dma_states:
dma_line += "[" + ("忙" if s == "BUSY" else "闲") + "]"
print(f" ║ {dma_line}")
# 事件日志
if self.events:
print(f" ╠══ 事件日志 ══╣")
for t, desc in self.events:
print(f" ║ T={t}: {desc}")
# 统计汇总
busy_count = sum(1 for s in self.cpu_states if s == "BUSY")
idle_count = total - busy_count
busy_pct = busy_count / total * 100 if total > 0 else 0
idle_pct = idle_count / total * 100 if total > 0 else 0
print(f" ╠══ 统计 ══╣")
print(f" ║ CPU 忙碌: {busy_count}/{total} ({busy_pct:.0f}%)")
print(f" ║ CPU 空闲: {idle_count}/{total} ({idle_pct:.0f}%)")
print(f" ╚{'═' * 20}╝")
return busy_pct
class Memory:
"""模拟主内存"""
def __init__(self):
self.data = {}
def write(self, addr, data):
self.data[addr] = data
def __repr__(self):
return f"Memory({len(self.data)} 条数据)"
class Controller:
"""CPU 控制器——执行不同 I/O 方式的模拟"""
def __init__(self):
self.memory = Memory()
self.total_transferred = 0
def simulate_polling(self, device, block_count):
"""
轮询方式模拟:
CPU 在每个时间单位都去检查设备是否就绪。
如果就绪，搬一个数据块；否则继续检查。
"""
print(f"\n{'#'*55}")
print(f"# 方式一: 轮询 (Polling)")
print(f"# 设备: {device.name} | 数据块数: {block_count}")
print(f"# 策略: CPU 每个时间单位都检查设备状态")
print(f"{'#'*55}")
timeline = Timeline()
time_unit = 0
blocks_done = 0
device_ready_at = 0 # 设备在哪个时间单位准备好下一块
# 初始化: 设备开始准备第一块数据
device.prepare_data(0)
device_ready_at = time_unit + device.response_time
while blocks_done < block_count:
# CPU 检查设备状态（轮询）
timeline.record_cpu("BUSY")
timeline.record_dma("IDLE")
if time_unit >= device_ready_at and device.is_ready():
# 设备就绪 -> CPU 搬运数据
data = device.get_data()
self.memory.write(blocks_done, data)
timeline.log_event(time_unit, f"CPU 搬运数据块 {blocks_done}: {data}")
blocks_done += 1
self.total_transferred += 1
# 准备下一块数据（如果还有）
if blocks_done < block_count:
device.prepare_data(blocks_done)
device_ready_at = time_unit + device.response_time
else:
# 设备未就绪 -> CPU 白跑一趟
timeline.log_event(time_unit,
f"CPU 检查设备... {'就绪' if device.is_ready() else '未就绪'}")
# 设备后台工作（如果是准备阶段）
if time_unit == device_ready_at:
device.mark_ready()
timeline.log_event(time_unit, "设备准备完成，数据就绪")
time_unit += 1
print(f"\n 传输完成: {self.total_transferred} 块数据")
busy_pct = timeline.draw("轮询方式时间轴")
return busy_pct
def simulate_interrupt(self, device, block_count):
"""
中断方式模拟:
CPU 大部分时间在执行其他任务（记为 IDLE），
只在收到中断信号后暂停当前工作，处理数据传输，然后继续。
"""
print(f"\n{'#'*55}")
print(f"# 方式二: 中断 (Interrupt)")
print(f"# 设备: {device.name} | 数据块数: {block_count}")
print(f"# 策略: CPU 正常执行任务，设备就绪时发中断通知")
print(f"{'#'*55}")
timeline = Timeline()
time_unit = 0
blocks_done = 0
transferred_in_interrupt = 0
device_ready_at = 0
# 初始化
device.prepare_data(0)
device_ready_at = time_unit + device.response_time
while blocks_done < block_count:
# 默认：CPU 在执行其他任务（空闲）
if time_unit == device_ready_at:
# 设备就绪，发中断信号
device.mark_ready()
timeline.log_event(time_unit, "中断触发！设备数据就绪")
# CPU 响应中断，搬运数据（这段时间 CPU 忙碌）
data = device.get_data()
for substep in range(2): # 中断处理需要 2 个时间单位
if substep == 0:
timeline.record_cpu("BUSY") # 保存上下文，进入中断处理
timeline.record_dma("IDLE")
timeline.log_event(time_unit + substep, f"CPU 进入中断处理")
else:
timeline.record_cpu("BUSY") # 搬运数据
timeline.record_dma("IDLE")
self.memory.write(blocks_done, data)
transferred_in_interrupt += 1
timeline.log_event(time_unit + substep, f"CPU 搬运数据块 {blocks_done}: {data}")
blocks_done += 1
# 中断处理完成，CPU 恢复之前的工作
timeline.record_cpu("BUSY")
timeline.record_dma("IDLE")
timeline.log_event(time_unit + 2, "CPU 恢复之前的工作")
time_unit += 3 # 跳过中断处理占用的时间
# 准备下一块
if blocks_done < block_count:
device.prepare_data(blocks_done)
device_ready_at = time_unit + device.response_time
continue
# 正常执行阶段——CPU 处理自己的任务
timeline.record_cpu("IDLE")
timeline.record_dma("IDLE")
timeline.log_event(time_unit, "CPU 执行其他任务 (空闲/用户程序)")
time_unit += 1
print(f"\n 传输完成: {self.total_transferred} 块数据")
busy_pct = timeline.draw("中断方式时间轴")
return busy_pct
def simulate_dma(self, device, block_count):
"""
DMA 方式模拟:
CPU 仅在一开始配置 DMA 控制器，随后完全不管数据传输。
DMA 控制器自主搬运所有数据块。
搬运完成后，DMA 发中断通知 CPU。
"""
print(f"\n{'#'*55}")
print(f"# 方式三: DMA (直接存储器访问)")
print(f"# 设备: {device.name} | 数据块数: {block_count}")
print(f"# 策略: CPU 配置 DMA 后继续工作，DMA 控制器自动搬运数据")
print(f"{'#'*55}")
timeline = Timeline()
time_unit = 0
dma_busy_until = 0
setup_done = False
blocks_to_transfer = list(range(block_count))
dma_buffer = []
# 准备所有数据块
for i in range(block_count):
device.prepare_data(i)
while blocks_to_transfer or dma_buffer:
# 第一阶段: CPU 配置 DMA 控制器（仅一次，占用 2 个时间单位）
if not setup_done:
for substep in range(2):
timeline.record_cpu("BUSY")
timeline.record_dma("IDLE")
if substep == 0:
timeline.log_event(time_unit,
"CPU 配置 DMA: 设置源地址、目标地址、传输字节数")
else:
timeline.log_event(time_unit,
"CPU 启动 DMA 传输，然后继续自己的工作")
time_unit += 1
setup_done = True
dma_busy_until = time_unit + block_count * device.response_time
# 标记设备就绪
device.mark_ready()
# 预填充 DMA 缓冲区
while device.data_buffer:
dma_buffer.append(device.data_buffer.pop(0))
continue
# 第二阶段: DMA 控制器自主搬运数据，CPU 空闲
if time_unit < dma_busy_until:
timeline.record_cpu("IDLE")
timeline.record_dma("BUSY")
if dma_buffer:
data = dma_buffer.pop(0)
blocks_to_transfer.pop(0)
mem_addr = block_count - len(blocks_to_transfer) - 1
self.memory.write(mem_addr, data)
self.total_transferred += 1
timeline.log_event(time_unit, f"DMA 搬运数据块 {mem_addr}: {data}")
else:
timeline.log_event(time_unit, "DMA 控制器工作中...")
else:
# 第三阶段: DMA 完成，发中断通知 CPU
timeline.record_cpu("BUSY")
timeline.record_dma("BUSY")
timeline.log_event(time_unit, "DMA 传输完成，中断通知 CPU")
timeline.record_cpu("BUSY")
timeline.record_dma("IDLE")
timeline.log_event(time_unit + 1, "CPU 确认传输完成")
time_unit += 2
break
time_unit += 1
print(f"\n 传输完成: {self.total_transferred} 块数据")
busy_pct = timeline.draw("DMA 方式时间轴")
return busy_pct
# ===== 主程序: 三种方式对比 =====
if __name__ == "__main__":
print("=" * 60)
print(" I/O 三种方式效率对比 —— RUNOOB 计算机组成原理演示")
print("=" * 60)
print()
print("场景: 从外部设备传输 5 个数据块到内存")
print("设备响应时间（每块数据准备所需时间单位）: 4")
print()
BLOCK_COUNT = 5
RESPONSE_TIME = 4
results = {}
# ---- 方式一: 轮询 ----
dev1 = IODevice("硬盘", RESPONSE_TIME)
ctrl1 = Controller()
pct1 = ctrl1.simulate_polling(dev1, BLOCK_COUNT)
results["轮询"] = pct1
# ---- 方式二: 中断 ----
dev2 = IODevice("硬盘", RESPONSE_TIME)
ctrl2 = Controller()
pct2 = ctrl2.simulate_interrupt(dev2, BLOCK_COUNT)
results["中断"] = pct2
# ---- 方式三: DMA ----
dev3 = IODevice("硬盘", RESPONSE_TIME)
ctrl3 = Controller()
pct3 = ctrl3.simulate_dma(dev3, BLOCK_COUNT)
results["DMA"] = pct3
# ---- 最终对比 ----
print("\n" + "=" * 60)
print(" 三种方式 CPU 忙碌时间占比对比")
print("=" * 60)
print(f" 轮询 (Polling): CPU 忙碌 {results['轮询']:.0f}% 的时间")
print(f" 中断 (Interrupt): CPU 忙碌 {results['中断']:.0f}% 的时间")
print(f" DMA: CPU 忙碌 {results['DMA']:.0f}% 的时间")
print()
best = min(results, key=results.get)
worst = max(results, key=results.get)
print(f" 最优方式: {best} (CPU 占用最少)")
print(f" 最差方式: {worst} (CPU 占用最多)")
print()
print(f" 结论: 对于大量数据传输，DMA 相比轮询的 CPU 占用降低了")
print(f" 约 {results['轮询'] - results['DMA']:.0f} 个百分点。")
print(f" 这就是为什么现代计算机传输大文件时使用 DMA。")
print("=" * 60)
I/O 三种方式对比：轮询 vs 中断 vs DMA（runoob 演示）
本程序模拟 CPU 从「外部设备」传输 10 个数据块到内存的场景，
分别用轮询、中断、DMA 三种方式完成，打印时间轴直观对比 CPU 状态。
核心指标: CPU 忙碌时间占比。占比越高，说明 CPU 被 I/O 拖累越严重。
"""
import time
import random
class IODevice:
"""模拟外部 I/O 设备（如硬盘、键盘等）"""
def __init__(self, name="硬盘", response_time=4):
"""
name: 设备名称
response_time: 设备响应所需的「时间单位」数
"""
self.name = name
self.response_time = response_time # 每个数据块的设备准备时间
self.data_buffer = [] # 设备内部数据缓冲区
self.ready = False # 设备是否就绪
def prepare_data(self, block_id):
"""
设备准备数据——模拟设备内部处理过程。
实际硬件中这一步是硬盘寻道、网卡收包等操作。
"""
self.ready = False
data = f"DATA_BLOCK_{block_id:02d}"
self.data_buffer.append(data)
# 模拟设备在不同时间单位后准备好
return data
def is_ready(self):
"""CPU 轮询时检查此状态"""
return self.ready
def mark_ready(self):
"""设备完成准备，发送中断信号"""
self.ready = True
def get_data(self):
"""从设备缓冲区取走一个数据块"""
if self.data_buffer:
return self.data_buffer.pop(0)
return None
class Timeline:
"""
时间轴记录器——记录每个时间单位 CPU 和 DMA 的状态
用于最终绘制 ASCII 时间轴
"""
def __init__(self):
self.cpu_states = [] # 每个时间单位 CPU 的状态: 'BUSY' 或 'IDLE'
self.dma_states = [] # 每个时间单位 DMA 的状态
self.events = [] # 事件日志: (时间, 描述)
def record_cpu(self, state):
self.cpu_states.append(state)
def record_dma(self, state):
self.dma_states.append(state)
def log_event(self, time_unit, desc):
self.events.append((time_unit, desc))
def draw(self, title="时间轴"):
"""绘制 ASCII 时间轴"""
total = len(self.cpu_states)
if total == 0:
print(" (无时间数据)")
return
print(f"\n ╔══ {title} ══╗")
print(f" ║ 总时间单位: {total}")
# CPU 状态行
cpu_line = "CPU: "
for s in self.cpu_states:
cpu_line += "[" + ("忙" if s == "BUSY" else "闲") + "]"
print(f" ║ {cpu_line}")
# DMA 状态行（如果有 DMA 活动）
if any(s == "BUSY" for s in self.dma_states):
dma_line = "DMA: "
for s in self.dma_states:
dma_line += "[" + ("忙" if s == "BUSY" else "闲") + "]"
print(f" ║ {dma_line}")
# 事件日志
if self.events:
print(f" ╠══ 事件日志 ══╣")
for t, desc in self.events:
print(f" ║ T={t}: {desc}")
# 统计汇总
busy_count = sum(1 for s in self.cpu_states if s == "BUSY")
idle_count = total - busy_count
busy_pct = busy_count / total * 100 if total > 0 else 0
idle_pct = idle_count / total * 100 if total > 0 else 0
print(f" ╠══ 统计 ══╣")
print(f" ║ CPU 忙碌: {busy_count}/{total} ({busy_pct:.0f}%)")
print(f" ║ CPU 空闲: {idle_count}/{total} ({idle_pct:.0f}%)")
print(f" ╚{'═' * 20}╝")
return busy_pct
class Memory:
"""模拟主内存"""
def __init__(self):
self.data = {}
def write(self, addr, data):
self.data[addr] = data
def __repr__(self):
return f"Memory({len(self.data)} 条数据)"
class Controller:
"""CPU 控制器——执行不同 I/O 方式的模拟"""
def __init__(self):
self.memory = Memory()
self.total_transferred = 0
def simulate_polling(self, device, block_count):
"""
轮询方式模拟:
CPU 在每个时间单位都去检查设备是否就绪。
如果就绪，搬一个数据块；否则继续检查。
"""
print(f"\n{'#'*55}")
print(f"# 方式一: 轮询 (Polling)")
print(f"# 设备: {device.name} | 数据块数: {block_count}")
print(f"# 策略: CPU 每个时间单位都检查设备状态")
print(f"{'#'*55}")
timeline = Timeline()
time_unit = 0
blocks_done = 0
device_ready_at = 0 # 设备在哪个时间单位准备好下一块
# 初始化: 设备开始准备第一块数据
device.prepare_data(0)
device_ready_at = time_unit + device.response_time
while blocks_done < block_count:
# CPU 检查设备状态（轮询）
timeline.record_cpu("BUSY")
timeline.record_dma("IDLE")
if time_unit >= device_ready_at and device.is_ready():
# 设备就绪 -> CPU 搬运数据
data = device.get_data()
self.memory.write(blocks_done, data)
timeline.log_event(time_unit, f"CPU 搬运数据块 {blocks_done}: {data}")
blocks_done += 1
self.total_transferred += 1
# 准备下一块数据（如果还有）
if blocks_done < block_count:
device.prepare_data(blocks_done)
device_ready_at = time_unit + device.response_time
else:
# 设备未就绪 -> CPU 白跑一趟
timeline.log_event(time_unit,
f"CPU 检查设备... {'就绪' if device.is_ready() else '未就绪'}")
# 设备后台工作（如果是准备阶段）
if time_unit == device_ready_at:
device.mark_ready()
timeline.log_event(time_unit, "设备准备完成，数据就绪")
time_unit += 1
print(f"\n 传输完成: {self.total_transferred} 块数据")
busy_pct = timeline.draw("轮询方式时间轴")
return busy_pct
def simulate_interrupt(self, device, block_count):
"""
中断方式模拟:
CPU 大部分时间在执行其他任务（记为 IDLE），
只在收到中断信号后暂停当前工作，处理数据传输，然后继续。
"""
print(f"\n{'#'*55}")
print(f"# 方式二: 中断 (Interrupt)")
print(f"# 设备: {device.name} | 数据块数: {block_count}")
print(f"# 策略: CPU 正常执行任务，设备就绪时发中断通知")
print(f"{'#'*55}")
timeline = Timeline()
time_unit = 0
blocks_done = 0
transferred_in_interrupt = 0
device_ready_at = 0
# 初始化
device.prepare_data(0)
device_ready_at = time_unit + device.response_time
while blocks_done < block_count:
# 默认：CPU 在执行其他任务（空闲）
if time_unit == device_ready_at:
# 设备就绪，发中断信号
device.mark_ready()
timeline.log_event(time_unit, "中断触发！设备数据就绪")
# CPU 响应中断，搬运数据（这段时间 CPU 忙碌）
data = device.get_data()
for substep in range(2): # 中断处理需要 2 个时间单位
if substep == 0:
timeline.record_cpu("BUSY") # 保存上下文，进入中断处理
timeline.record_dma("IDLE")
timeline.log_event(time_unit + substep, f"CPU 进入中断处理")
else:
timeline.record_cpu("BUSY") # 搬运数据
timeline.record_dma("IDLE")
self.memory.write(blocks_done, data)
transferred_in_interrupt += 1
timeline.log_event(time_unit + substep, f"CPU 搬运数据块 {blocks_done}: {data}")
blocks_done += 1
# 中断处理完成，CPU 恢复之前的工作
timeline.record_cpu("BUSY")
timeline.record_dma("IDLE")
timeline.log_event(time_unit + 2, "CPU 恢复之前的工作")
time_unit += 3 # 跳过中断处理占用的时间
# 准备下一块
if blocks_done < block_count:
device.prepare_data(blocks_done)
device_ready_at = time_unit + device.response_time
continue
# 正常执行阶段——CPU 处理自己的任务
timeline.record_cpu("IDLE")
timeline.record_dma("IDLE")
timeline.log_event(time_unit, "CPU 执行其他任务 (空闲/用户程序)")
time_unit += 1
print(f"\n 传输完成: {self.total_transferred} 块数据")
busy_pct = timeline.draw("中断方式时间轴")
return busy_pct
def simulate_dma(self, device, block_count):
"""
DMA 方式模拟:
CPU 仅在一开始配置 DMA 控制器，随后完全不管数据传输。
DMA 控制器自主搬运所有数据块。
搬运完成后，DMA 发中断通知 CPU。
"""
print(f"\n{'#'*55}")
print(f"# 方式三: DMA (直接存储器访问)")
print(f"# 设备: {device.name} | 数据块数: {block_count}")
print(f"# 策略: CPU 配置 DMA 后继续工作，DMA 控制器自动搬运数据")
print(f"{'#'*55}")
timeline = Timeline()
time_unit = 0
dma_busy_until = 0
setup_done = False
blocks_to_transfer = list(range(block_count))
dma_buffer = []
# 准备所有数据块
for i in range(block_count):
device.prepare_data(i)
while blocks_to_transfer or dma_buffer:
# 第一阶段: CPU 配置 DMA 控制器（仅一次，占用 2 个时间单位）
if not setup_done:
for substep in range(2):
timeline.record_cpu("BUSY")
timeline.record_dma("IDLE")
if substep == 0:
timeline.log_event(time_unit,
"CPU 配置 DMA: 设置源地址、目标地址、传输字节数")
else:
timeline.log_event(time_unit,
"CPU 启动 DMA 传输，然后继续自己的工作")
time_unit += 1
setup_done = True
dma_busy_until = time_unit + block_count * device.response_time
# 标记设备就绪
device.mark_ready()
# 预填充 DMA 缓冲区
while device.data_buffer:
dma_buffer.append(device.data_buffer.pop(0))
continue
# 第二阶段: DMA 控制器自主搬运数据，CPU 空闲
if time_unit < dma_busy_until:
timeline.record_cpu("IDLE")
timeline.record_dma("BUSY")
if dma_buffer:
data = dma_buffer.pop(0)
blocks_to_transfer.pop(0)
mem_addr = block_count - len(blocks_to_transfer) - 1
self.memory.write(mem_addr, data)
self.total_transferred += 1
timeline.log_event(time_unit, f"DMA 搬运数据块 {mem_addr}: {data}")
else:
timeline.log_event(time_unit, "DMA 控制器工作中...")
else:
# 第三阶段: DMA 完成，发中断通知 CPU
timeline.record_cpu("BUSY")
timeline.record_dma("BUSY")
timeline.log_event(time_unit, "DMA 传输完成，中断通知 CPU")
timeline.record_cpu("BUSY")
timeline.record_dma("IDLE")
timeline.log_event(time_unit + 1, "CPU 确认传输完成")
time_unit += 2
break
time_unit += 1
print(f"\n 传输完成: {self.total_transferred} 块数据")
busy_pct = timeline.draw("DMA 方式时间轴")
return busy_pct
# ===== 主程序: 三种方式对比 =====
if __name__ == "__main__":
print("=" * 60)
print(" I/O 三种方式效率对比 —— RUNOOB 计算机组成原理演示")
print("=" * 60)
print()
print("场景: 从外部设备传输 5 个数据块到内存")
print("设备响应时间（每块数据准备所需时间单位）: 4")
print()
BLOCK_COUNT = 5
RESPONSE_TIME = 4
results = {}
# ---- 方式一: 轮询 ----
dev1 = IODevice("硬盘", RESPONSE_TIME)
ctrl1 = Controller()
pct1 = ctrl1.simulate_polling(dev1, BLOCK_COUNT)
results["轮询"] = pct1
# ---- 方式二: 中断 ----
dev2 = IODevice("硬盘", RESPONSE_TIME)
ctrl2 = Controller()
pct2 = ctrl2.simulate_interrupt(dev2, BLOCK_COUNT)
results["中断"] = pct2
# ---- 方式三: DMA ----
dev3 = IODevice("硬盘", RESPONSE_TIME)
ctrl3 = Controller()
pct3 = ctrl3.simulate_dma(dev3, BLOCK_COUNT)
results["DMA"] = pct3
# ---- 最终对比 ----
print("\n" + "=" * 60)
print(" 三种方式 CPU 忙碌时间占比对比")
print("=" * 60)
print(f" 轮询 (Polling): CPU 忙碌 {results['轮询']:.0f}% 的时间")
print(f" 中断 (Interrupt): CPU 忙碌 {results['中断']:.0f}% 的时间")
print(f" DMA: CPU 忙碌 {results['DMA']:.0f}% 的时间")
print()
best = min(results, key=results.get)
worst = max(results, key=results.get)
print(f" 最优方式: {best} (CPU 占用最少)")
print(f" 最差方式: {worst} (CPU 占用最多)")
print()
print(f" 结论: 对于大量数据传输，DMA 相比轮询的 CPU 占用降低了")
print(f" 约 {results['轮询'] - results['DMA']:.0f} 个百分点。")
print(f" 这就是为什么现代计算机传输大文件时使用 DMA。")
print("=" * 60)
现实世界的实际应用场景
键盘输入——中断的典型例子
你每按下一个键，键盘控制器就向 CPU 发一个中断信号。CPU 暂停当前程序，读取按键码，然后恢复。这个过程快到让你觉察不到任何延迟。
如果键盘用轮询方式——CPU 每秒钟检查 1000 次「用户按键了吗？」——你的电脑就会卡死，什么也干不了。
大文件复制——DMA 的典型例子
你从 C 盘复制一个 2GB 的电影到 D 盘。硬盘控制器通过 DMA 直接向内存搬运数据，整个过程中 CPU 几乎不参与。你还可以同时刷网页、写代码，完全不受影响。
如果没有 DMA——CPU 要亲自搬运 2GB 的数据——在此期间 CPU 基本无法处理任何其他任务。
为什么轮询没有被淘汰？
轮询虽然效率最低，但它在极简单的嵌入式系统中仍然有用：
- 硬件成本最低——不需要中断控制器和 DMA 控制器。
- 对于「设备几乎瞬间就绪」的场景，轮询的开销与中断相当，但实现简单得多。
- 某些实时系统中，轮询的确定性（精确知道何时检查）比中断的随机性更有优势。
小结与检验
一句话总结：轮询是 CPU 不停地问「好了没」，中断是设备主动喊「我好了」，DMA 是找个帮手替你干活——数据量越大，DMA 的优势越明显。
自测题
- 你正在用电脑写文档，同时后台在复制一个大文件。复制文件用的是哪种 I/O 方式？你敲键盘用的又是哪种？
- 如果一个设备的数据准备时间极短（比如只有 1 微秒），用轮询还会不会浪费 CPU？为什么？
- DMA 控制器完成数据传输后，怎样通知 CPU？它用的是什么机制？
参考答案：1. 文件复制用 DMA，键盘输入用中断。2. 仍然会浪费——因为轮询期间 CPU 无法做其他事，即使每次轮询很快，被占用的时间片也无法用于执行用户程序。但如果设备确实极快，轮询的实现简单性可能更有优势。3. DMA 完成后通过中断通知 CPU——这是 DMA 唯一需要 CPU 参与的时刻。