外观
os/models.py
配套源码，运行方法见同目录 README。返回实验总览。
python
"""Executable teaching models for xv6 concepts; these are not xv6 implementations.
Run from this directory: python3 -m unittest -v
Only the Python standard library is required.
"""
from dataclasses import dataclass
PAGE_SIZE = 4096
MAXVA = 1 << 38 # xv6 uses the lower half of Sv39 virtual addresses.
class MemoryFault(Exception):
"""Address, mapping, or access permission is invalid."""
class OutOfMemory(Exception):
"""The simulated physical page pool is exhausted."""
def split_va(va):
"""Return (VPN2, VPN1, VPN0, page offset), for xv6's supported range."""
if not isinstance(va, int) or not 0 <= va < MAXVA:
raise MemoryFault("virtual address outside xv6 MAXVA")
return ((va >> 30) & 0x1FF, (va >> 21) & 0x1FF,
(va >> 12) & 0x1FF, va & 0xFFF)
@dataclass(frozen=True)
class Leaf:
pa: int
read: bool = True
write: bool = False
execute: bool = False
user: bool = True
class Sv39:
"""Three-level page tables, including optional 2 MiB/1 GiB leaves.
Dictionary nodes stand for page-table pages. This omits TLBs, A/D bits,
physical address width, SUM/MXR, and supervisor instruction fetch rules.
"""
def __init__(self):
self.root = {}
def map(self, va, pa, *, level=0, read=True, write=False,
execute=False, user=True):
if level not in (0, 1, 2):
raise ValueError("level must be 0, 1, or 2")
indexes = split_va(va)[:3]
size = 1 << (12 + 9 * level)
if pa < 0 or va % size or pa % size:
raise MemoryFault("misaligned virtual or physical leaf address")
if not (read or execute) or (write and not read):
raise MemoryFault("invalid RISC-V leaf permission combination")
node = self.root
for depth, index in enumerate(indexes):
current_level = 2 - depth
if current_level == level:
if index in node:
raise MemoryFault("overlapping mapping")
node[index] = Leaf(pa, read, write, execute, user)
return
entry = node.setdefault(index, {})
if isinstance(entry, Leaf):
raise MemoryFault("mapping falls inside an existing huge page")
node = entry
def translate(self, va, access="read", *, from_user=True):
if access not in ("read", "write", "execute"):
raise ValueError("unknown access type")
indexes = split_va(va)[:3]
node = self.root
for depth, index in enumerate(indexes):
if index not in node:
raise MemoryFault("unmapped address")
entry = node[index]
if isinstance(entry, Leaf):
if from_user and not entry.user:
raise MemoryFault("user access to supervisor page")
if not getattr(entry, access):
raise MemoryFault("permission denied")
size = 1 << (12 + 9 * (2 - depth))
return entry.pa + (va & (size - 1))
node = entry
raise MemoryFault("non-leaf entry at level zero")
class PhysicalMemory:
"""Refcounts count mappings only; no temporary references are modeled."""
def __init__(self, capacity=16):
if capacity < 0:
raise ValueError("capacity must be non-negative")
self.capacity = capacity
self.pages = {}
self.refs = {}
self.next_id = 0
self.allocations = 0
self.copied_bytes = 0
def allocate(self, data=None):
if len(self.pages) >= self.capacity:
raise OutOfMemory("no free physical pages")
if data is not None and len(data) != PAGE_SIZE:
raise ValueError("page image must be exactly one page")
page = self.next_id
self.next_id += 1
self.pages[page] = bytearray(PAGE_SIZE) if data is None else bytearray(data)
self.refs[page] = 1
self.allocations += 1
if data is not None:
self.copied_bytes += PAGE_SIZE
return page
def retain(self, page):
self.refs[page] += 1
def release(self, page):
self.refs[page] -= 1
if self.refs[page] == 0:
del self.refs[page]
del self.pages[page]
@dataclass
class Mapping:
page: int
writable: bool
cow: bool = False
class Process:
"""Private mappings, fork sharing, COW faults, and kernel copyout.
Deliberate simplifications: no CPU execution, locks, TLB, page-table page
allocations, read permissions, executable pages, or lazy allocation.
Mapping mutation is serial. Cross-page writes may partially succeed.
"""
def __init__(self, memory):
self.memory = memory
self.mappings = {}
def map_zero(self, va, *, writable=True):
split_va(va)
if va % PAGE_SIZE:
raise MemoryFault("mapping must be page-aligned")
vpn = va // PAGE_SIZE
if vpn in self.mappings:
raise MemoryFault("overlapping mapping")
page = self.memory.allocate()
self.mappings[vpn] = Mapping(page, writable)
def fork(self):
child = Process(self.memory)
for vpn, mapping in self.mappings.items():
if mapping.writable:
mapping.writable = False
mapping.cow = True
self.memory.retain(mapping.page)
child.mappings[vpn] = Mapping(mapping.page, mapping.writable, mapping.cow)
return child
def _lookup(self, va):
split_va(va)
try:
return self.mappings[va // PAGE_SIZE]
except KeyError as exc:
raise MemoryFault("unmapped address") from exc
def _writable_page(self, va):
mapping = self._lookup(va)
if mapping.writable:
return mapping.page
if not mapping.cow:
raise MemoryFault("write to an originally read-only page")
old_page = mapping.page
if self.memory.refs[old_page] == 1:
mapping.writable = True
mapping.cow = False
return old_page
# Allocate before changing the old mapping: OOM preserves its content.
new_page = self.memory.allocate(self.memory.pages[old_page])
mapping.page = new_page
mapping.writable = True
mapping.cow = False
self.memory.release(old_page)
return new_page
def read(self, va, size):
if size < 0:
raise ValueError("negative read length")
result = bytearray()
for offset in range(size):
address = va + offset
mapping = self._lookup(address)
result.append(self.memory.pages[mapping.page][address % PAGE_SIZE])
return bytes(result)
def write(self, va, data):
for offset, byte in enumerate(data):
address = va + offset
page = self._writable_page(address)
self.memory.pages[page][address % PAGE_SIZE] = byte
def copyout(self, va, data):
"""Kernel writes must explicitly use the same COW resolution path."""
self.write(va, data)
def unmap(self, va):
split_va(va)
if va % PAGE_SIZE:
raise MemoryFault("unmap must be page-aligned")
mapping = self.mappings.pop(va // PAGE_SIZE)
self.memory.release(mapping.page)
def close(self):
for mapping in self.mappings.values():
self.memory.release(mapping.page)
self.mappings.clear()
def demo():
table = Sv39()
table.map(0x4000, 0x80002000, write=True)
print("Sv39:", hex(0x4123), "->", hex(table.translate(0x4123)))
memory = PhysicalMemory(capacity=4)
parent = Process(memory)
parent.map_zero(0)
parent.write(0, b"parent")
child = parent.fork()
print("After fork: physical pages =", len(memory.pages), "refs =", memory.refs)
child.write(0, b"CHILD!")
print("After child write:", parent.read(0, 6), child.read(0, 6))
print("Physical pages =", len(memory.pages), "copied bytes =", memory.copied_bytes)
child.close()
parent.close()
print("After exit: physical pages =", len(memory.pages))
if __name__ == "__main__":
demo()