## 题目一：多级反馈队列调度算法模拟 ### 代码概述 该代码实现了多级反馈队列调度算法，包括进程建模、调度队列设计、进程状态转换、阻塞模拟以及输出保存等功能。代码使用 Python 实现，并通过类和函数组织。 ### 代码解析 ```python import random import json from collections import deque from dataclasses import dataclass, field @dataclass class Process: pid: int arrive: int cpu_total: int io_gap: int cpu_since_io: int = 0 state: str = 'NEW' queue_level: int = 0 start: int = -1 finish: int = -1 wait_time: int = 0 cpu_need: int = field(init=False) def __post_init__(self): self.cpu_need = self.cpu_total class Scheduler: def __init__(self, processes): self.processes = processes self.time = 0 self.ready_queues = [deque(), deque(), deque()] self.running_process = None self.gantt_chart = [] self.cpu_utilization = 0 self.throughput = 0 self.total_cpu_time = 0 self.completed_processes = [] def simulate(self): while self.running_process or any(self.ready_queues) or self.processes: self._advance_time() self._check_arrivals() self._check_unblock() self._schedule_process() if self.running_process: self._run_process() else: self.gantt_chart.append((self.time, None)) self._update_metrics() self._calculate_performance_metrics() def _advance_time(self): self.time += 1 def _check_arrivals(self): while self.processes and self.processes[0].arrive <= self.time: process = self.processes.pop(0) process.state = 'READY' self.ready_queues[process.queue_level].append(process) if process.start == -1: process.start = self.time def _check_unblock(self): for queue in self.ready_queues: for process in list(queue): if process.state == 'BLOCK': process.state = 'READY' queue.remove(process) self.ready_queues[process.queue_level].append(process) def _schedule_process(self): for i in range(3): if self.ready_queues[i]: self.running_process = self.ready_queues[i].popleft() self.running_process.state = 'RUN' return self.running_process = None def _run_process(self): self.total_cpu_time += 1 self.running_process.cpu_need -= 1 self.running_process.cpu_since_io += 1 if self.running_process.cpu_need == 0: self.running_process.finish = self.time self.running_process.wait_time = self.running_process.finish - self.running_process.arrive - self.running_process.cpu_total self.completed_processes.append(self.running_process) self.running_process = None elif self.running_process.cpu_since_io >= self.running_process.io_gap: self.running_process.state = 'BLOCK' self.running_process.cpu_since_io = 0 self.ready_queues[self.running_process.queue_level].append(self.running_process) self.running_process = None elif self.running_process.queue_level == 0 and self.time % 2 == 0: self.running_process.queue_level = min(self.running_process.queue_level + 1, 2) self.ready_queues[self.running_process.queue_level].append(self.running_process) self.running_process = None elif self.running_process.queue_level == 1 and self.time % 4 == 0: self.running_process.queue_level = min(self.running_process.queue_level + 1, 2) self.ready_queues[self.running_process.queue_level].append(self.running_process) self.running_process = None def _update_metrics(self): self.cpu_utilization = (self.total_cpu_time / self.time) * 100 if self.time > 0 else 0 self.throughput = len(self.completed_processes) / self.time if self.time > 0 else 0 def _calculate_performance_metrics(self): total_turnaround = sum(p.finish - p.arrive for p in self.completed_processes) total_weighted_turnaround = sum((p.finish - p.arrive) / p.cpu_total for p in self.completed_processes) total_wait = sum(p.wait_time for p in self.completed_processes) avg_turnaround = total_turnaround / len(self.completed_processes) if self.completed_processes else 0 avg_weighted_turnaround = total_weighted_turnaround / len(self.completed_processes) if self.completed_processes else 0 avg_wait = total_wait / len(self.completed_processes) if self.completed_processes else 0 print(f"平均周转时间: {avg_turnaround:.2f} ms") print(f"平均带权周转时间: {avg_weighted_turnaround:.2f}") print(f"平均等待时间: {avg_wait:.2f} ms") print(f"CPU利用率: {self.cpu_utilization:.2f}%") print(f"吞吐量: {self.throughput:.4f} 进程/ms") def generate_processes(n=20): return [ Process(pid=i, arrive=random.randint(0, 10), cpu_total=random.randint(5, 20), io_gap=random.randint(2, 8)) for i in range(n) ] if __name__ == "__main__": processes = generate_processes() scheduler = Scheduler(processes) scheduler.simulate() ``` ### 知识点 1. **多级反馈队列调度算法**：通过多个就绪队列实现进程调度，根据时间片用尽情况调整进程优先级。 2. **进程状态转换**：进程在不同状态间转换的机制，如从就绪到运行再到阻塞或结束。 3. **性能指标计算**：计算周转时间、带权周转时间、等待时间等性能指标。 --- ## 题目二：简易 inode-位示图 文件系统模拟 ### 代码概述 该代码模拟了一个简易的类 UNIX 文件系统，包括文件创建、读写、删除等基本操作，以及磁盘块管理和位示图管理。代码使用 Python 实现，并通过类和函数组织。 ### 代码解析 ```python class FileSystem: DISK_SIZE = 128 # 总磁盘块数，每块 1KB MAX_FILE_SIZE = 12 # 单文件最大大小 12KB class Inode: def __init__(self, file_name): self.file_name = file_name self.size = 0 self.blocks = [] def __init__(self): self.inodes = {} self.bitmap = [0] * FileSystem.DISK_SIZE def mkfs(self): self.__init__() print("格式化完成") def alloc_block(self): for i in range(FileSystem.DISK_SIZE): if not self.bitmap[i]: self.bitmap[i] = 1 return i print("磁盘已满") return None def free_block(self, block): if 0 <= block < FileSystem.DISK_SIZE: self.bitmap[block] = 0 def touch(self, file_name): if file_name in self.inodes: print(f"{file_name} 已存在") return self.inodes[file_name] = FileSystem.Inode(file_name) print(f"{file_name} 创建成功") def write(self, file_name, size_kb): if file_name not in self.inodes: print("无此文件") return inode = self.inodes[file_name] remaining_size = min(FileSystem.MAX_FILE_SIZE - inode.size, size_kb) if remaining_size <= 0: print("超出直接块上限") return while remaining_size > 0: block = self.alloc_block() if block is None: print("磁盘已满") return inode.blocks.append(block) inode.size += 1 remaining_size -= 1 print(f"写入完成, 文件大小 {inode.size} KB") def read(self, file_name): if file_name not in self.inodes: print("无此文件") return inode = self.inodes[file_name] print(f"{file_name}: {inode.size} KB") def rm(self, file_name): if file_name not in self.inodes: print("无此文件") return inode = self.inodes.pop(file_name) for block in inode.blocks: self.free_block(block) print(f"{file_name} 已删除") def ls(self): print("===== inode 表 =====") for name, inode in self.inodes.items(): print(f"{name:<10} {inode.size}KB blocks:{inode.blocks}") print("===== 位示图(1=占用) =====") for i in range(0, FileSystem.DISK_SIZE, 32): chunk = ''.join(str(bit) for bit in self.bitmap[i:i+32]) print(chunk) if __name__ == "__main__": fs = FileSystem() commands = [ ("mkfs", lambda: fs.mkfs()), ("touch test.txt", lambda: fs.touch("test.txt")), ("write test.txt 5", lambda: fs.write("test.txt", 5)), ("read test.txt", lambda: fs.read("test.txt")), ("write test.txt 8", lambda: fs.write("test.txt", 8)), ("write test.txt 1", lambda: fs.write("test.txt", 1)), ("ls", lambda: fs.ls()), ("rm test.txt", lambda: fs.rm("test.txt")), ("read test.txt", lambda: fs.read("test.txt")), ("write test.txt 1", lambda: fs.write("test.txt", 1)), ] for command, func in commands: print(f"fs> {command}") func() print() ``` ### 知识点 1. **inode 索引结构**：用于存储文件元数据的结构，不包含文件名，只包含文件大小和块指针。 2. **位示图（Bitmap）**：用于管理磁盘块的占用状态，便于快速查找空闲块。 3. **文件操作流程**：包括文件创建、读写、删除及块分配回收的具体实现逻辑。