两步的操作为什么也不安全
(每道题开头都有同一段:上面的模拟器。判题机要确定的答案,所以本条用「调度表」代替真线程的运气。)
贯穿全条的模拟器:
一条线程 = 一串「步」 inc_steps() = [("read",), ("add", 1), ("write",)] 读共享的 n 进寄存器、寄存器加一、写回
run_schedule(threads, schedule) schedule 是线程编号的序列,每个编号出现一次就让那条线程走一步;交回 (最终 n, 记录)
all_schedules(a, b) 两条线程(a 步、b 步)的全部交错
outcomes(threads) 各种最终值各出现几次
locked(steps) 把一串步包成一个原子步——模拟器遇到它一口气做完把「加」并进「写」(两步:读、写 n+1),窗口只剩一个缝。所有交错里错的还是不少:
def inc_steps(k=1):
return [("read",), ("add", k), ("write",)]
def check_act_steps(need=1):
return [("read",), ("check", need), ("add", -need), ("write",)]
def run_schedule(threads, schedule, start=0):
"""threads: 每条线程是「步」的列表;schedule: 线程编号的序列,每个编号出现一次就走一步。
共享变量 n;每条线程有自己的寄存器 reg。交回 (最终 n, 记录列表)。"""
n = start
pc = [0] * len(threads)
reg = [0] * len(threads)
log = []
for t in schedule:
if pc[t] >= len(threads[t]):
continue
step = threads[t][pc[t]]
pc[t] += 1
n, reg[t], note, go_on = do_step(step, n, reg[t])
log.append("T" + str(t) + ":" + note)
if not go_on:
pc[t] = len(threads[t]) # 检查没过:这条线程后面的步全部跳过
return n, log
def do_step(step, n, r):
if step[0] == "read":
return n, n, "read " + str(n), True
if step[0] == "add":
return n, r + step[1], "add→" + str(r + step[1]), True
if step[0] == "write":
return r, r, "write " + str(r), True
if step[0] == "check":
ok = r >= step[1]
return n, r, "check " + str(r) + (">=" if ok else "<") + str(step[1]), ok
if step[0] == "atomic":
notes = []
ok = True
for s in step[1]:
n, r, note, ok = do_step(s, n, r)
notes.append(note)
if not ok:
break
return n, r, "atomic(" + ",".join(notes) + ")", ok
raise ValueError(step[0])
def all_schedules(a, b):
"""两条线程(a 步和 b 步)的全部交错:每个交错是一串 0/1。"""
if a == 0:
return [[1] * b]
if b == 0:
return [[0] * a]
return [[0] + s for s in all_schedules(a - 1, b)] + [[1] + s for s in all_schedules(a, b - 1)]
def outcomes(threads, start=0):
counts = {}
for s in all_schedules(len(threads[0]), len(threads[1])):
n, _ = run_schedule(threads, s, start)
counts[n] = counts.get(n, 0) + 1
return counts
def locked(steps):
return [("atomic", steps)]
A = [("read",), ("add", 1), ("write",)]
S = [("read",), ("write_inc",)]
STEP0 = do_step
def do_step2(step, n, r):
if step[0] == "write_inc":
return r + 1, r + 1, "write " + str(r + 1), True
return STEP0(step, n, r)
do_step = do_step2
two = outcomes([S, list(S)])
three = outcomes([A, list(A)])
do_step = STEP0
print((str(two) + "/" + str(three)).replace(" ", ""))
全部评论