自己写:一页多大,一共几页,一共多少字节
外面一整张,里面一页页
你看到的
里面其实
把 TODO 补完,打出一页多大、一共几页、两个乘起来是多少(三段,用 / 隔开;库用 2000 行的)。
这条路线两样工具都在题面里备好了
① 真的 sqlite3(页这一层是能内省出来的)
mk(rows, idx, w) → 现造一个库:rows 行、每行 w 个字符、idx=True 再建一份索引
pages(db) 一共几页|page_size(db) 一页多大
rootpage(db, 名字) 树根在第几页|pagesof(db, 名字) 它占了几页
pagetypes(db, 名字) → 像 [("internal", 1), ("leaf", 14)] 这样
② 三个小模型(这三样在 sqlite 上看不见,但机制正是考点)
Pool(cap) 缓冲池:read(页号) 返回"命中"或"读盘",p.hit / p.miss 记着次数
WAL(use_log) 预写日志:crash_after(n, pairs) / recover(pairs)
MVCC() 多版本:write / snapshot / read(键, 快照号) / versions(键)import sqlite3
# ── 真的 sqlite3:页这一层是能内省出来的 ──────────────────────────────
def mk(rows=2000, idx=False, w=20):
"""现造一个库:rows 行,每行 w 个字符;idx=True 时再建一份索引。"""
db = sqlite3.connect(":memory:")
db.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT NOT NULL)")
db.executemany("INSERT INTO t VALUES (?,?)",
[(i, "x" * w) for i in range(1, rows + 1)])
if idx:
db.execute("CREATE INDEX ix ON t(v)")
db.commit()
return db
def pages(db):
"""这个库一共用了多少页。"""
return db.execute("PRAGMA page_count").fetchone()[0]
def page_size(db):
"""一页多大(字节)。"""
return db.execute("PRAGMA page_size").fetchone()[0]
def rootpage(db, name):
"""某张表或某份索引,它的树根在第几页。"""
r = db.execute("SELECT rootpage FROM sqlite_master WHERE name = ?", (name,)).fetchone()
return r[0] if r else None
def pagesof(db, name):
"""某张表或某份索引,一共占了几页(dbstat 是 sqlite 自带的内省表)。"""
return db.execute("SELECT COUNT(*) FROM dbstat WHERE name = ?", (name,)).fetchone()[0]
def pagetypes(db, name):
"""它的页分成哪几类、各几页。返回像 [("internal", 1), ("leaf", 14)] 这样。"""
return db.execute("SELECT pagetype, COUNT(*) FROM dbstat WHERE name = ? "
"GROUP BY pagetype ORDER BY pagetype", (name,)).fetchall()
# ── 缓冲池:内存里只放得下几页,读一页先看在不在池子里 ────────────────
class Pool:
"""cap 页的缓冲池,满了就把**最久没碰过**的那页挤出去。"""
def __init__(self, cap):
self.cap = cap
self.inpool = [] # 池子里有哪几页,越靠后越新碰过
self.hit = 0
self.miss = 0
def read(self, pg):
if pg in self.inpool:
self.hit = self.hit + 1
self.inpool.remove(pg)
self.inpool.append(pg)
return "命中"
self.miss = self.miss + 1
self.inpool.append(pg)
if len(self.inpool) > self.cap:
self.inpool.pop(0)
return "读盘"
# ── 预写日志:先把要做的事记下来,再去动数据 ──────────────────────────
class WAL:
"""预写日志的要点是**顺序**:先把整笔事情记进日志,再去动数据。
这样断电断在哪儿都没关系——日志里有全部,照着补完就行。"""
def __init__(self, use_log=True):
self.data = {}
self.log = []
self.use_log = use_log
def crash_after(self, n, pairs):
"""一笔事情要改 len(pairs) 处。
有日志的话:**先把这 len(pairs) 条全部写进日志**,再一处一处改数据;
没日志的话:直接一处一处改。
两种都在改到第 n 处的时候断电。返回断电时数据里已经改了几处。"""
self.data = {}
self.log = []
if self.use_log:
for k, v in pairs: # ← 先记全,再动数据
self.log.append((k, v))
for i in range(n):
k, v = pairs[i]
self.data[k] = v
return len(self.data)
def recover(self):
"""照日志把这笔事情补完。没有日志就补不了。"""
if not self.use_log:
return "补不了"
for k, v in self.log:
self.data[k] = v
return "补上了"
def applied(self):
"""数据里现在改了几处。"""
return len(self.data)
# ── 多版本:每次写不覆盖旧的,而是**再存一版** ────────────────────────
class MVCC:
"""每个键留着历次的版本。读的人拿着一个快照号,只看得见那之前的版本。"""
def __init__(self):
self.ver = 0
self.rows = {} # 键 → [(版本号, 值), ...]
self.blocked = 0 # 读被挡住的次数(这个模型里永远是 0)
def write(self, k, v):
self.ver = self.ver + 1
self.rows.setdefault(k, []).append((self.ver, v))
return self.ver
def snapshot(self):
"""开一个快照:从这一刻起,只看得见到目前为止的版本。"""
return self.ver
def read(self, k, snap):
"""按快照读:拿版本号不超过 snap 的**最新那一版**。"""
best = None
for ver, val in self.rows.get(k, []):
if ver <= snap:
best = val
return best
def versions(self, k):
"""这个键一共留了几版。"""
return len(self.rows.get(k, []))
d = mk(2000)
# TODO:按 一页多大/一共几页/乘起来 打出来
print("")
全部评论