"""The two-head machine: the organism, the netlist N_{G,L}, and the host stored in it.""" import argparse import hashlib import heapq import json import os import random import struct import subprocess import sys import time import numpy as np import host as SR REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) ALPHA = 6 W = 1 << ALPHA KAPPA, RD, ARM, CPY, AR, AA, AC, SINK = range(8) B = 26 GATE = 8 ADDR = 18 class Asm: """Records over named cells.""" def __init__(self): self.recs, self.persist, self.names, self.uid = [], {}, {}, 0 def keep(self, name): self.persist[name] = 8 + len(self.persist) return name def reg(self, name): return name def fresh(self, base): self.uid += 1 return f'{base}#{self.uid}' def emit(self, a0, w0, a1, w1, beta, ao): assert -8 <= beta <= 7 and w0 in (-1, 0, 1) and w1 in (-1, 0, 1) self.recs.append((a0, w0, a1, w1, beta, ao)) def AND(self, x, y, dst): (a, pa), (c, pc) = x, y self.emit(a, 1 if pa else -1, c, 1 if pc else -1, -(pa + pc), dst) def OR(self, x, y, dst): (a, pa), (c, pc) = x, y self.emit(a, 1 if pa else -1, c, 1 if pc else -1, (2 - pa - pc) - 1, dst) def COPY(self, x, dst): a, pa = x self.emit(a, 1 if pa else -1, SINK, 0, -1 if pa else 0, dst) def MUX(self, s, a, b, dst): t1, t2 = self.fresh('m'), self.fresh('m') self.AND((s, True), a, t1) self.AND((s, False), b, t2) self.OR((t1, True), (t2, True), dst) def XOR(self, a, b, dst): t1, t2 = self.fresh('x'), self.fresh('x') self.OR(a, b, t1) self.OR((a[0], not a[1]), (b[0], not b[1]), t2) self.AND((t1, True), (t2, True), dst) def allocate(self): """Map every name to a window address; check each temporary is written before it is read.""" first_w, last_r = {}, {} for i, (a0, w0, a1, w1, be, ao) in enumerate(self.recs): for a, w in ((a0, w0), (a1, w1)): if isinstance(a, str) and a not in self.persist and w != 0: assert a in first_w, f'{a} read before written' last_r[a] = i if isinstance(ao, str) and ao not in self.persist: first_w.setdefault(ao, i) free_from = {} base = 8 + len(self.persist) addr = dict(self.persist) for name in sorted(first_w, key=lambda n: first_w[n]): end = last_r.get(name, first_w[name]) cand = [a for a, f in free_from.items() if f <= first_w[name]] a = min(cand) if cand else base + len(free_from) assert a < W, 'window full' addr[name] = a free_from[a] = end self.names = addr res = lambda a: addr[a] if isinstance(a, str) else a self.recs = [(res(a0) if w0 else SINK, w0, res(a1) if w1 else SINK, w1, be, res(ao)) for (a0, w0, a1, w1, be, ao) in self.recs] self.next_reg = base + len(free_from) return self def encode_record(r, alpha=ALPHA): """r = (a0, w0, a1, w1, beta, ao), laid out gate first, with addresses of alpha bits.""" a0, w0, a1, w1, beta, ao = r bits = [] put = lambda val, w: bits.extend((val >> (w - 1 - j)) & 1 for j in range(w)) wc = {1: 1, -1: 2, 0: 0} put(wc[w0], 2); put(wc[w1], 2); put(beta & 15, 4); put(a0, alpha); put(a1, alpha); put(ao, alpha) return bits def decode_record(bits, alpha=ALPHA): it = iter(bits) get = lambda w: sum(next(it) << (w - 1 - j) for j in range(w)) c0 = get(2); c1 = get(2); beta = get(4); a0 = get(alpha); a1 = get(alpha); ao = get(alpha) wd = {1: 1, 2: -1, 0: 0, 3: 0} return a0, wd[c0], a1, wd[c1], beta - 16 if beta >= 8 else beta, ao def gate_bits(kind, s0, s1): """The eight gate bits of a codon: w0 (2), w1 (2), beta (4); kind 0 AND, 1 OR, 2 COPY.""" sgn = {0: s0 | s1, 1: s0 & s1, 2: s0}[kind] odd = {0: s0 ^ s1, 1: 1 - (s0 ^ s1), 2: s0}[kind] w1 = (0, 0) if kind == 2 else (1 - s1, s1) return [1 - s0, s0, w1[0], w1[1], sgn, sgn, sgn, odd] def codon_of(gbits): """The gate codon whose gate bits are gbits, if any.""" for kind in (0, 1, 2): for s0 in (0, 1): for s1 in ((0, 1) if kind != 2 else (0,)): if gate_bits(kind, s0, s1) == gbits: return [kind >> 1, kind & 1, s0, s1] return None ESC = [1, 1, 0, 0] END = [1, 1, 1, 1] def describe(image): assert len(image) % B == 0 out = [] for r in range(len(image) // B): rec = image[r * B:(r + 1) * B] c = codon_of(rec[:GATE]) out += (c + rec[GATE:]) if c else (ESC + rec) return out + END def translate(desc): """The decoding the organism carries out, written plainly.""" out, i = [], 0 while True: c = desc[i:i + 4]; i += 4 kind, s0, s1 = c[0] * 2 + c[1], c[2], c[3] if kind == 3 and s0 and s1: return out, i if kind == 3: out += desc[i:i + GATE]; i += GATE else: out += gate_bits(kind, s0, s1) out += desc[i:i + ADDR]; i += ADDR class Machine: """The physics, transcribed: tapes listed from their heads; a window of 2^alpha cells.""" def __init__(self, G, L, Q0, Q1, D0, D1, window, alpha=ALPHA): self.alpha, self.B = alpha, 8 + 3 * alpha self.G, self.N, self.L = G, G * self.B, L self.Q = [list(Q0), list(Q1)] self.D = [list(D0), list(D1)] self.w = list(window) assert len(self.w) == 1 << alpha self.steps = 0 def cell(self, a): k = self.w[KAPPA] if a == RD: return self.D[k][0] if a == ARM: return self.Q[1 - k][0] if a == CPY: return self.D[1 - k][0] return self.w[a] def step(self): k = self.w[KAPPA] a0, w0, a1, w1, beta, ao = decode_record(self.Q[k][:self.B], self.alpha) val = 1 if w0 * self.cell(a0) + w1 * self.cell(a1) + beta >= 0 else 0 if ao == ARM: self.Q[1 - k][0] = val elif ao == CPY: self.D[1 - k][0] = val elif ao != RD: self.w[ao] = val rot = lambda t, s: t[s:] + t[:s] self.Q[k] = rot(self.Q[k], self.B) if self.w[AA]: self.Q[1 - k] = rot(self.Q[1 - k], 1) if self.w[AR]: self.D[k] = rot(self.D[k], 1) if self.w[AC]: self.D[1 - k] = rot(self.D[1 - k], 1) self.w[AA] = self.w[AR] = self.w[AC] = 0 self.steps += 1 def organism(): A = Asm() y, n = True, False m, C, Gp, Ad = (A.keep(s) for s in ('m', 'C', 'G', 'Ad')) c = [A.keep(f'c{i}') for i in range(4)] # c[3] is the first codon bit z = [A.keep(f'z{i}') for i in range(5)] # down counter, z[0] least significant T = lambda s: A.reg('T_' + s) # flags from the old state A.AND((z[0], n), (z[1], n), T('q')) A.AND((T('q'), y), (z[2], n), T('q')) A.AND((T('q'), y), (z[3], n), T('q')) A.AND((T('q'), y), (z[4], n), T('zero')) # the codon after this pass's shift is (c2, c1, c0, R); END and ESC at its completion A.AND((c[2], y), (c[1], y), T('k11')) A.AND((c[0], y), (RD, y), T('ss')) A.AND((T('k11'), y), (T('ss'), y), T('endc')) A.AND((T('k11'), y), (T('ss'), n), T('escn')) A.AND((C, y), (T('zero'), y), T('cdone')) A.AND((T('cdone'), y), (T('endc'), y), T('end')) A.AND((T('cdone'), y), (T('endc'), n), T('cgo')) # the escape flag of the codon held during G (the codon is complete then) A.AND((c[3], y), (c[2], y), T('k')) A.AND((c[1], y), (c[0], y), T('s')) A.AND((T('k'), y), (T('s'), n), T('esc')) # consume, emit, copy A.AND((Gp, y), (T('esc'), y), T('gesc')) A.OR((C, y), (Ad, y), T('ca')) A.OR((T('ca'), y), (T('gesc'), y), T('cons')) A.OR((Gp, y), (Ad, y), T('ga')) A.AND((T('ga'), y), (m, n), T('emit')) A.AND((Gp, y), (T('esc'), n), T('tsel')) A.AND((m, y), (T('cons'), y), T('cw')) # the table bit at position 7 - z of the gate field, from the held codon A.AND((c[3], y), (c[2], n), T('cpy')) # COPY kind A.XOR((c[1], y), (c[0], y), T('x')) A.AND((c[1], y), (c[0], y), T('a11')) A.OR((c[1], y), (c[0], y), T('o11')) A.MUX(c[2], (T('a11'), y), (T('o11'), y), T('sg0')) # OR: s0 and s1; AND: s0 or s1 A.MUX(c[3], (c[1], y), (T('sg0'), y), T('sign')) # COPY: s0 A.XOR((c[2], y), (T('x'), y), T('od0')) # OR: not x; AND: x A.MUX(c[3], (c[1], y), (T('od0'), y), T('odd')) A.AND((T('cpy'), n), (c[0], n), T('w1h')) A.AND((T('cpy'), n), (c[0], y), T('w1l')) A.MUX(z[0], (c[1], n), (c[1], y), T('t11')) # z2=1, z1=1: z0 ? not s0 : s0 A.MUX(z[0], (T('w1h'), y), (T('w1l'), y), T('t10')) # z2=1, z1=0: z0 ? w1h : w1l A.MUX(z[1], (T('t11'), y), (T('t10'), y), T('t1')) A.OR((z[1], y), (z[0], y), T('lo')) A.MUX(T('lo'), (T('sign'), y), (T('odd'), y), T('t0')) A.MUX(z[2], (T('t1'), y), (T('t0'), y), T('tbl')) A.MUX(T('tsel'), (T('tbl'), y), (RD, y), T('av')) # next phases A.AND((C, y), (T('zero'), n), T('n1')) A.OR((T('n1'), y), (T('end'), y), T('n2')) A.AND((Ad, y), (T('zero'), y), T('n3')) A.OR((T('n2'), y), (T('n3'), y), T('nC')) A.OR((m, n), (T('escn'), y), T('g1')) # construct mode, or an escape A.AND((T('cgo'), y), (T('g1'), y), T('g2')) A.AND((Gp, y), (T('zero'), n), T('g3')) A.OR((T('g2'), y), (T('g3'), y), T('nG')) A.AND((T('cgo'), y), (T('g1'), n), T('a1')) A.AND((Gp, y), (T('zero'), y), T('a2')) A.OR((T('a1'), y), (T('a2'), y), T('a3')) A.AND((Ad, y), (T('zero'), n), T('a4')) A.OR((T('a3'), y), (T('a4'), y), T('nA')) # the counter: zero loads 3, 7 or 17 for the next phase, otherwise it counts down A.COPY((z[0], n), T('d0')) A.COPY((z[0], n), T('b1')) A.XOR((z[1], y), (T('b1'), y), T('d1')) A.AND((T('b1'), y), (z[1], n), T('b2')) A.XOR((z[2], y), (T('b2'), y), T('d2')) A.AND((T('b2'), y), (z[2], n), T('b3')) A.XOR((z[3], y), (T('b3'), y), T('d3')) A.AND((T('b3'), y), (z[3], n), T('b4')) A.XOR((z[4], y), (T('b4'), y), T('d4')) A.OR((T('nC'), y), (T('nG'), y), T('l1')) A.OR((T('zero'), y), (T('d0'), y), z[0]) # every load is odd A.MUX(T('zero'), (T('l1'), y), (T('d1'), y), z[1]) # 3 and 7 A.MUX(T('zero'), (T('nG'), y), (T('d2'), y), z[2]) # 7 A.AND((T('zero'), n), (T('d3'), y), z[3]) A.MUX(T('zero'), (T('nA'), y), (T('d4'), y), z[4]) # 17 # the codon shifts during C A.MUX(C, (c[2], y), (c[3], y), c[3]) A.MUX(C, (c[1], y), (c[2], y), c[2]) A.MUX(C, (c[0], y), (c[1], y), c[1]) A.MUX(C, (RD, y), (c[0], y), c[0]) # under the arm and the copy head, from the old mode A.MUX(T('emit'), (T('av'), y), (ARM, y), ARM) A.MUX(T('cw'), (RD, y), (CPY, y), CPY) A.AND((m, y), (T('end'), y), T('hand')) A.COPY((T('nC'), y), C); A.COPY((T('nG'), y), Gp); A.COPY((T('nA'), y), Ad) A.XOR((m, y), (T('end'), y), m) # requests last A.COPY((T('emit'), y), AA) A.COPY((T('cw'), y), AC) A.COPY((T('cons'), y), AR) A.XOR((KAPPA, y), (T('hand'), y), KAPPA) return A.allocate() def start(A, rho, desc): """The state s0: construct mode, kappa 0, the organism in Q0 and its description on D0.""" window = [0] * W window[A.names['C']] = 1 for i, bit in enumerate((1, 1, 0, 0, 0)): # the counter starts at 3 window[A.names[f'z{i}']] = bit return Machine(len(A.recs), len(desc), rho, [0] * len(rho), desc, [0] * len(desc), window) def sha(bits): return hashlib.sha256(bytes(bits)).hexdigest() def run_organism(generations): A = organism() G = len(A.recs) rho = [b for r in A.recs for b in encode_record(r)] desc = describe(rho) out, used = translate(desc) assert out == rho and used == len(desc) M = start(A, rho, desc) gens, states = [], [] t0 = time.time() for g in range(1, generations + 1): first, k0 = M.steps, M.w[KAPPA] while M.w[KAPPA] == k0: for _ in range(G): M.step() steps = M.steps - first gens.append(dict(generation=g, steps=steps, passes=steps // G, kappa=M.w[KAPPA], banks_hold_organism=M.Q[0] == rho and M.Q[1] == rho, tapes_hold_description=M.D[0] == desc and M.D[1] == desc)) states.append((list(M.w), [list(t) for t in M.Q], [list(t) for t in M.D])) periodic = len(states) >= 3 and states[2] == states[0] esc = sum(1 for r in range(G) if codon_of(rho[r * B:r * B + GATE]) is None) return dict(records=G, bank_bits=G * B, description_bits=len(desc), description_over_bank=round(len(desc) / (G * B), 3), escaped_records=esc, window_cells_used=A.next_reg, generations=gens, state_after_generation_3_equals_after_1=periodic, sha256_organism=sha(rho), sha256_description=sha(desc), seconds=round(time.time() - t0, 1)) def rocq_files(out_dir): """Write Organism.v and SmokeData.v for the Rocq development.""" A = organism() z = lambda v: f'({v})%Z' if v < 0 else f'{v}%Z' recs = ';\n'.join(f' Rec {a0} {z(w0)} {a1} {z(w1)} {z(be)} {ao}' for a0, w0, a1, w1, be, ao in A.recs) names = ', '.join(f'{k} = {v}' for k, v in sorted(A.names.items(), key=lambda kv: kv[1]) if k in ('m', 'C', 'G', 'Ad') or k[0] in 'cz' and k[1:].isdigit()) lit = lambda bits: '[' + '; '.join('true' if b else 'false' for b in bits) + ']' with open(os.path.join(out_dir, 'Organism.v'), 'w') as f: f.write(f'(* The organism: {len(A.recs)} records, generated by src/twohead.py. *)\n') f.write('From Stdlib Require Import List Bool Arith ZArith.\nImport ListNotations.\n' 'From TH Require Import TwoHead.\n\n') f.write('Definition organism : list rec := [\n' + recs + '\n].\n\n') f.write(f'(* {names}; the pass starts in C with z = 3. *)\n') cells = [A.names['C'], A.names['z0'], A.names['z1']] f.write('Definition init_win : list bool :=\n map (fun i => ' + ' || '.join(f'Nat.eqb i {c}' for c in cells) + ') (seq 0 64).\n\n') f.write('Definition rho : list bool := concat (map enc_rec organism).\n' 'Definition desc : list bool := describe rho.\n\n' 'Definition s0 : st :=\n mkS init_win (tape_of rho) (tape_of (repeat false (length rho)))\n' ' (tape_of desc) (tape_of (repeat false (length desc))).\n') rho = [b for r in A.recs for b in encode_record(r)] desc = describe(rho) M = start(A, rho, desc) with open(os.path.join(out_dir, 'SmokeData.v'), 'w') as f: f.write('(* The organism and its description as the transcription computes them, and its\n' ' states from s0 after 1000 and 20000 steps; generated by src/twohead.py. *)\n') f.write('From Stdlib Require Import List.\nImport ListNotations.\n\n') f.write(f'Definition py_rho : list bool := {lit(rho)}.\n') f.write(f'Definition py_desc : list bool := {lit(desc)}.\n') for n in (1000, 20000): while M.steps < n: M.step() f.write(f'Definition py_win_{n} : list bool := {lit(M.w)}.\n') f.write(f'Definition py_q0_{n} : list bool := {lit(M.Q[0])}.\n') f.write(f'Definition py_q1_{n} : list bool := {lit(M.Q[1])}.\n') f.write(f'Definition py_d0_{n} : list bool := {lit(M.D[0])}.\n') f.write(f'Definition py_d1_{n} : list bool := {lit(M.D[1])}.\n') def main_organism(): ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) sub = ap.add_subparsers(dest='cmd', required=True) r = sub.add_parser('run') r.add_argument('--generations', type=int, default=3) q = sub.add_parser('rocq') q.add_argument('--out', default=os.path.join(REPO, 'rocq')) a = ap.parse_args() if a.cmd == 'run': res = run_organism(a.generations) path = os.path.join(REPO, 'paper', 'runs', 'paper_twohead.json') json.dump(res, open(path, 'w'), indent=1) print(json.dumps(res, indent=1)) else: rocq_files(a.out) NET_RECORD = os.path.join(REPO, 'paper', 'runs', 'paper_twohead_net.json') class Net: def __init__(self, n_state): self.n = n_state self.units = [] # (bias, [(signal, weight)]) def u(self, bias, ins): assert all(w in (-1, 1) for _, w in ins) and len({s for s, _ in ins}) == len(ins) self.units.append((bias, ins)) return self.n + len(self.units) - 1 # literals are (signal, positive) def AND(self, *lits): return self.u(-sum(1 for _, p in lits if p), [(s, 1 if p else -1) for s, p in lits]) def OR(self, *lits): return self.u(sum(1 for _, p in lits if not p) - 1, [(s, 1 if p else -1) for s, p in lits]) def MUX(self, s, a, b): # s ? a : b on signals return self.OR((self.AND((s, True), (a, True)), True), (self.AND((s, False), (b, True)), True)) def build_net(G, L, alpha=ALPHA): """The netlist N_{G,L} of the machine with a window of 2^alpha cells and records of 8 + 3 alpha bits.""" W, B = 1 << alpha, 8 + 3 * alpha N = G * B real = [0] + list(range(4, W)) # window cells held in the state wpos = {a: i for i, a in enumerate(real)} q0, q1 = len(real), len(real) + N d0, d1 = len(real) + 2 * N, len(real) + 2 * N + L n = len(real) + 2 * N + 2 * L net = Net(n) kap = wpos[KAPPA] # the record in front of the active bank f = [net.MUX(kap, q1 + j, q0 + j) for j in range(B)] c0, c1, be = f[0:2], f[2:4], f[4:8] a0, a1, ao = f[8:8 + alpha], f[8 + alpha:8 + 2 * alpha], f[8 + 2 * alpha:8 + 3 * alpha] # the cells under the three heads rd = net.MUX(kap, d1, d0) armv = net.MUX(kap, q0, q1) cpyv = net.MUX(kap, d0, d1) val = {a: wpos[a] for a in real} val[RD], val[ARM], val[CPY] = rd, armv, cpyv def decode(field): lines = [] for a in range(W): bits = [(a >> (alpha - 1 - j)) & 1 for j in range(alpha)] lines.append(net.u(-sum(bits), [(field[j], 1 if bits[j] else -1) for j in range(alpha)])) return lines l0, l1, lo = decode(a0), decode(a1), decode(ao) xs = [] for lines in (l0, l1): terms = [net.AND((lines[a], True), (val[a], True)) for a in range(W)] xs.append(net.u(-1, [(t, 1) for t in terms])) pos = [net.AND((c[0], False), (c[1], True)) for c in (c0, c1)] neg = [net.AND((c[0], True), (c[1], False)) for c in (c0, c1)] mterm = [net.AND((pos[s], True), (xs[s], True)) for s in range(2)] nterm = [net.AND((neg[s], True), (xs[s], True)) for s in range(2)] # [p >= c] for p = m0 + m1 - n0 - n1, and [beta >= t] on the four bits of beta pge = {c: net.u(-c, [(mterm[0], 1), (mterm[1], 1), (nterm[0], -1), (nterm[1], -1)]) for c in range(-2, 3)} b3, b2, b1, b0 = be or210 = net.OR((b2, True), (b1, True), (b0, True)) or21 = net.OR((b2, True), (b1, True)) and210 = net.AND((b2, True), (b1, True), (b0, True)) and21 = net.AND((b2, True), (b1, True)) bge = {0: net.AND((b3, False)), 1: net.AND((b3, False), (or210, True)), 2: net.AND((b3, False), (or21, True)), -1: net.OR((b3, False), (and210, True)), -2: net.OR((b3, False), (and21, True))} v = net.u(-1, [(net.AND((pge[c], True), (bge[-c], True)), 1) for c in range(-2, 3)]) # the write nxt = [None] * n xhat = {} for a in real: xhat[a] = net.MUX(lo[a], v, wpos[a]) zero = net.u(-1, []) for a in real: nxt[wpos[a]] = xhat[a] for a in (AR, AA, AC): nxt[wpos[a]] = zero advA, advR, advC = xhat[AA], xhat[AR], xhat[AC] # the banks: active turns by B, idle turns by one under the arm for j, base in ((0, q0), (1, q1)): act = (kap, j == 1) # kappa == j as a literal idle = (kap, j == 0) ia = net.AND(idle, (advA, True)) ih = net.AND(idle, (advA, False)) wa = net.AND((lo[ARM], True), idle) h0 = net.MUX(wa, v, base) hat = lambda i: h0 if i == 0 else base + i for i in range(N): nxt[base + i] = net.OR((net.AND(act, (base + (i + B) % N, True)), True), (net.AND((ia, True), (hat((i + 1) % N), True)), True), (net.AND((ih, True), (hat(i), True)), True)) # the description tapes: the active one under the read head, the idle one under the copy head for j, base in ((0, d0), (1, d1)): act = (kap, j == 1) idle = (kap, j == 0) rot = net.OR((net.AND(act, (advR, True)), True), (net.AND(idle, (advC, True)), True)) wc = net.AND((lo[CPY], True), idle) h0 = net.MUX(wc, v, base) hat = lambda i: h0 if i == 0 else base + i for i in range(L): nxt[base + i] = net.MUX(rot, hat((i + 1) % L), hat(i)) assert all(x is not None for x in nxt) return net, nxt, dict(real=real, q0=q0, q1=q1, d0=d0, d1=d1, n=n, N=N, L=L, alpha=alpha) def sizes(alpha, N, L): """The units, edges and state bits of N_{G,L} for a window of 2^alpha cells.""" W = 1 << alpha return dict(units=8 * W + 9 * alpha + 81 + 8 * N + 6 * L, edges=3 * alpha * W + 12 * W + 18 * alpha + 170 + 18 * N + 12 * L, state_bits=W - 3 + 2 * N + 2 * L) def stats(net, nxt): depth = np.zeros(net.n + len(net.units), dtype=np.int64) edges = 0 for i, (b, ins) in enumerate(net.units): depth[net.n + i] = 1 + max((depth[s] for s, _ in ins), default=0) edges += len(ins) return dict(units=len(net.units), edges=edges, depth=int(max(depth[s] for s in nxt)), state_bits=net.n) def evaluate(net, nxt, x): sig = np.concatenate([np.asarray(x, dtype=np.int64), np.zeros(len(net.units), dtype=np.int64)]) for i, (b, ins) in enumerate(net.units): sig[net.n + i] = 1 if b + sum(w * sig[s] for s, w in ins) >= 0 else 0 return [int(sig[s]) for s in nxt] def encode(M, lay): return [M.w[a] for a in lay['real']] + M.Q[0] + M.Q[1] + M.D[0] + M.D[1] def check_net(G, L, trials, seed=7, alpha=ALPHA): """One step of the netlist against the transcription from random full states.""" net, nxt, lay = build_net(G, L, alpha) rnd = random.Random(seed) bad = 0 N = G * (8 + 3 * alpha) for _ in range(trials): bits = lambda k: [rnd.randint(0, 1) for _ in range(k)] M = Machine(G, L, bits(N), bits(N), bits(L), bits(L), bits(1 << alpha), alpha) y = evaluate(net, nxt, encode(M, lay)) M.step() bad += y != encode(M, lay) st = stats(net, nxt) assert {k: st[k] for k in ('units', 'edges', 'state_bits')} == sizes(alpha, N, L) return bad, st def netfile_bytes(net, nxt): parts = [np.array([net.n, len(net.units), sum(len(i) for _, i in net.units)], dtype=' 1 else 'check' if cmd == 'check': t0 = time.time() small, _ = check_net(5, 40, 300) A, rho, desc, net, nxt, lay = organism_netlist() full, _ = check_net(len(A.recs), len(desc), 20) wide, st_wide = check_net(5, 40, 40, alpha=12) res = dict(small=dict(G=5, L=40, states=300, disagreements=small), full=dict(G=len(A.recs), L=len(desc), states=20, disagreements=full), wide=dict(alpha=12, G=5, L=40, states=40, disagreements=wide, netlist=st_wide), netlist=stats(net, nxt), seconds=round(time.time() - t0, 1)) net_update_record('random_states', res) print(json.dumps(res, indent=1)) elif cmd == 'export': path = sys.argv[2] if len(sys.argv) > 2 else os.path.join(REPO, 'variants', 'twohead_physics.bin') A, rho, desc, net, nxt, lay = organism_netlist() data = netfile_bytes(net, nxt) open(path, 'wb').write(data) res = dict(path=os.path.relpath(path, REPO), bytes=len(data), sha256=hashlib.sha256(data).hexdigest(), G=len(A.recs), L=len(desc), **stats(net, nxt)) net_update_record('netlist', res) print(json.dumps(res, indent=1)) elif cmd == 'lockstep': gens = int(sys.argv[2]) if len(sys.argv) > 2 else 2 A, rho, desc, net, nxt, lay = organism_netlist() tmp = os.path.join(REPO, 'variants') netpath, initpath = os.path.join(tmp, 'twohead_physics.bin'), os.path.join(tmp, '.twohead_init.bin') open(netpath, 'wb').write(netfile_bytes(net, nxt)) M = start(A, rho, desc) x = encode(M, lay) with open(initpath, 'wb') as fh: fh.write(np.array([len(A.recs), len(desc), len(lay['real'])] + lay['real'], dtype=' 0 else -1) for e in pred[p:p + fanin[j]]])) p += fanin[j] return len(nxt), units, nxt # state layout of sigma: the A counter bits most significant first, the halt bit at A, and bit k of # word j at A + 1 + A j + k, k = 0 the most significant def mem_bit(j, t, A=8): """State index of bit t (value 2^t) of memory word j.""" return A + 1 + A * j + (A - 1 - t) def pc_bit(t, A=8): """State index of bit t of the counter.""" return A - 1 - t # --------------------------------------------------------------------------- # records over named cells # --------------------------------------------------------------------------- class Prog: def __init__(self): self.recs = [] # (a0, w0, a1, w1, beta, ao), cells by name self.ends = {} # record -> the unit of the host whose value it writes self.uid = 0 def tmp(self): self.uid += 1 return ('t', self.uid) def emit(self, a0, w0, a1, w1, beta, ao): assert -8 <= beta <= 7 self.recs.append((a0 if w0 else ('c', SINK), w0, a1 if w1 else ('c', SINK), w1, beta, ao)) def AND(self, x, y, dst): (a, pa), (b, pb) = x, y self.emit(a, 1 if pa else -1, b, 1 if pb else -1, -(pa + pb), dst) def OR(self, x, y, dst): (a, pa), (b, pb) = x, y self.emit(a, 1 if pa else -1, b, 1 if pb else -1, 1 - pa - pb, dst) def COPY(self, x, dst): a, pa = x self.emit(a, 1 if pa else -1, None, 0, -1 if pa else 0, dst) def CONST(self, v, dst): # a gate of one literal and its negation: OR is 1, AND is 0 x = ('c', SINK) (self.OR if v else self.AND)((x, True), (x, False), dst) def XOR(self, x, y, dst): t1, t2 = self.tmp(), self.tmp() self.OR(x, y, t1) self.OR((x[0], not x[1]), (y[0], not y[1]), t2) self.AND((t1, True), (t2, True), dst) def MUX(self, s, a, b, dst): """dst <- s ? a : b, s a cell, a and b literals.""" t1, t2 = self.tmp(), self.tmp() self.AND((s, True), a, t1) self.AND((s, False), b, t2) self.OR((t1, True), (t2, True), dst) def chain(self, kind, lits, dst, before=None): """Chain the literals by 'and' or 'or' into dst, written once, last.""" assert lits if len(lits) == 1: if before: before(0) self.COPY(lits[0], dst) return g = self.AND if kind == 'and' else self.OR acc = self.tmp() if before: before(0); before(1) if len(lits) == 2: g(lits[0], lits[1], dst) return g(lits[0], lits[1], acc) for i in range(2, len(lits)): if before: before(i) g((acc, True), lits[i], dst if i == len(lits) - 1 else acc) def threshold(self, t, lits, dst): """at least t of the literals, by T_t(x1..xk) = T_t(x2..xk) or (x1 and T_{t-1}(x2..xk)).""" k = len(lits) if t <= 0: self.CONST(1, dst); return if t > k: self.CONST(0, dst); return if t == 1: self.chain('or', lits, dst); return if t == k: self.chain('and', lits, dst); return a, b = self.tmp(), self.tmp() self.threshold(t, lits[1:], a) self.threshold(t - 1, lits[1:], b) c = self.tmp() self.AND(lits[0], (b, True), c) self.OR((a, True), (c, True), dst) def compile_host(P, A=8): """The records of one step of H_A; state bits are cells ('s', i).""" n, units, nxt = host_units(A) cons = {} for u, (_, ins) in enumerate(units): for s, _w in ins: if s >= n: cons.setdefault(s - n, []).append(u) is_next = {nxt[i] - n: i for i in range(n)} assert len(is_next) == n and all(v >= n for v in nxt) fusable = {u for u, c in cons.items() if len(c) == 1} done, cell = set(), {} def cell_of(s): return ('s', s) if s < n else cell[s - n] def ensure_ancestors(u): for s, _w in units[u][1]: if s >= n: v = s - n if v in fusable: ensure_ancestors(v) else: need(v) def need(u): if u in done: return assert u not in is_next, 'a next value is read before the state is written' ensure_ancestors(u) cell[u] = P.tmp() emit_unit(u, cell[u]) done.add(u) def fused(v): if v not in done: cell[v] = P.tmp() emit_unit(v, cell[v]) done.add(v) def emit_unit(u, dst): b, ins = units[u] k = len(ins) t = sum(1 for _, w in ins if w < 0) - b lits = [None] * k def before(i): s = ins[i][0] if s >= n and (s - n) in fusable: fused(s - n) lits[i] = (cell_of(s), ins[i][1] > 0) if t <= 0 or t > k: P.CONST(1 if t <= 0 else 0, dst) elif 1 < t < k: for i in range(k): before(i) P.threshold(t, lits, dst) else: _chain_lazy(P, 'and' if t == k else 'or', lits, dst, before) P.ends[len(P.recs) - 1] = u order_next = [nxt[pc_bit(t, A)] - n for t in range(A)] + [nxt[A] - n] + \ [nxt[mem_bit(j, t, A)] - n for j in range(1 << A) for t in range(A)] for u in order_next: # stage A: everything the next values read ensure_ancestors(u) for u in order_next: # stage B: the next values, in place emit_unit(u, ('s', is_next[u])) done.add(u) return n, units, nxt def _chain_lazy(P, kind, lits, dst, before): k = len(lits) if k == 1: before(0) P.COPY(lits[0], dst) return g = P.AND if kind == 'and' else P.OR before(0); before(1) if k == 2: g(lits[0], lits[1], dst) return acc = P.tmp() g(lits[0], lits[1], acc) for i in range(2, k): before(i) g((acc, True), lits[i], dst if i == k - 1 else acc) # --------------------------------------------------------------------------- # the device, the startup block and the hand-over # --------------------------------------------------------------------------- def mem_cell(j, t, A=8): return ('s', mem_bit(j, t, A)) def word_is_one(P, j, A=8): """[memory word j == 1]""" dst = P.tmp() lits = [(mem_cell(j, 0, A), True)] + [(mem_cell(j, t, A), False) for t in range(1, A)] P.chain('and', lits, dst) return dst def compile_device(P, A=8, w=CW): io = SR.io_cells(A) cnt = [('d', 'cnt', i) for i in range(w)] lim = [('d', 'lim', i) for i in range(w)] dest = ('d', 'dest') wr = word_is_one(P, io['c_wr'], A) rd = word_is_one(P, io['c_rd'], A) rw = word_is_one(P, io['c_rw'], A) # [cnt == lim] eq = P.tmp() xs = [] for i in range(w): e = P.tmp() a1, a0 = P.tmp(), P.tmp() P.AND((cnt[i], True), (lim[i], True), a1) P.AND((cnt[i], False), (lim[i], False), a0) P.OR((a1, True), (a0, True), e) xs.append((e, True)) P.chain('and', xs, eq) eot = P.tmp() # the end of the tape is known after the rewind P.AND((eq, True), (dest, True), eot) # (i) the low byte of R_out, most significant bit first, under the arm or under the copy head ca, cc = P.tmp(), P.tmp() P.AND((wr, True), (dest, False), ca) P.AND((wr, True), (dest, True), cc) out, inp, ceot = io['r_out'], io['r_in'], io['c_eot'] for t in range(7, -1, -1): P.MUX(ca, (mem_cell(out, t, A), True), (('c', ARM), True), ('c', ARM)) P.COPY((ca, True), ('c', AA)) P.MUX(cc, (mem_cell(out, t, A), True), (('c', CPY), True), ('c', CPY)) P.COPY((cc, True), ('c', AC)) for t in range(A): P.AND((mem_cell(out, t, A), True), (wr, False), mem_cell(out, t, A)) g = P.tmp() P.AND((rd, True), (eot, False), g) # (ii) the byte under the read head into R_in, most significant bit first, each bit turning the tape, # and its sign bit into the bits above it for t in range(7, -1, -1): P.MUX(g, (('c', RD), True), (mem_cell(inp, t, A), True), mem_cell(inp, t, A)) P.COPY((g, True), ('c', AR)) for t in range(8, A): P.MUX(g, (mem_cell(inp, 7, A), True), (mem_cell(inp, t, A), True), mem_cell(inp, t, A)) # c_eot: bit 0 <- [cnt == lim] on a read, the other bits cleared on a read P.MUX(rd, (eot, True), (mem_cell(ceot, 0, A), True), mem_cell(ceot, 0, A)) for t in range(1, A): P.AND((mem_cell(ceot, t, A), True), (rd, False), mem_cell(ceot, t, A)) # the counter of bytes read: carries first, then the sum bits car = [g] for i in range(w - 1): c = P.tmp() P.AND((car[-1], True), (cnt[i], True), c) car.append(c) for i in range(w): P.XOR((cnt[i], True), (car[i], True), cnt[i]) # (iii) a rewind: the limit takes the count, the count returns to 0, the writes go to the copy tape for i in range(w): P.MUX(rw, (cnt[i], True), (lim[i], True), lim[i]) P.AND((cnt[i], True), (rw, False), cnt[i]) P.OR((dest, True), (rw, True), dest) # (iv) the request cells cleared for c in ('c_wr', 'c_rd', 'c_rw'): for t in range(A): P.CONST(0, mem_cell(io[c], t, A)) def compile_startup(P, mem_image, A=8, w=CW): """The startup records: while booted is 0, load the memory image and clear the counter, halt bit and registers.""" b = ('d', 'booted') for j in range(1 << A): for t in range(A): c = mem_cell(j, t, A) if (mem_image[j] >> t) & 1: P.OR((c, True), (b, False), c) # b ? c : 1 else: P.AND((c, True), (b, True), c) # b ? c : 0 for t in range(A): P.AND((('s', pc_bit(t, A)), True), (b, True), ('s', pc_bit(t, A))) P.AND((('s', A), True), (b, True), ('s', A)) for i in range(w): P.AND((('d', 'cnt', i), True), (b, True), ('d', 'cnt', i)) P.AND((('d', 'lim', i), True), (b, True), ('d', 'lim', i)) P.AND((('d', 'dest'), True), (b, True), ('d', 'dest')) P.CONST(1, b) def compile_handover(P, A=8): """The hand-over records: at the halt, clear booted and flip the bank bit at the end of the pass.""" h = ('s', A) P.AND((('d', 'booted'), True), (h, False), ('d', 'booted')) P.XOR((('c', KAPPA), True), (h, True), ('c', KAPPA)) # --------------------------------------------------------------------------- # allocation and encoding # --------------------------------------------------------------------------- def allocate(recs): """Window addresses for the special cells, the state, the registers and the temporaries.""" fixed = {} nxt_addr = 8 names = [] for r in recs: for a in (r[0], r[2], r[5]): if a is not None and a[0] in ('s', 'd') and a not in fixed: names.append(a) fixed[a] = None for a in sorted(names, key=lambda x: (x[0], x[1:] if x[0] == 'd' else x[1])) : fixed[a] = nxt_addr nxt_addr += 1 first_w, last_r = {}, {} for i, (a0, w0, a1, w1, be, ao) in enumerate(recs): for a, w in ((a0, w0), (a1, w1)): if w and a[0] == 't': assert a in first_w, ('read before written', a) last_r[a] = i if ao[0] == 't': first_w.setdefault(ao, i) base = nxt_addr busy, free, top = [], [], base # (index after which it is free, cell); free cells addr = dict(fixed) for name in sorted(first_w, key=lambda x: first_w[x]): start, end = first_w[name], last_r.get(name, first_w[name]) while busy and busy[0][0] < start: heapq.heappush(free, heapq.heappop(busy)[1]) if free: c = heapq.heappop(free) else: c, top = top, top + 1 addr[name] = c heapq.heappush(busy, (end, c)) def res(a): if a is None: return SINK if a[0] == 'c': return a[1] return addr[a] out = [(res(a0), w0, res(a1), w1, be, res(ao)) for (a0, w0, a1, w1, be, ao) in recs] return out, addr, top def bits_to_bytes(bits): assert len(bits) % 8 == 0 return bytes(sum(bits[i + k] << (7 - k) for k in range(8)) for i in range(0, len(bits), 8)) def bytes_to_bits(data): return [(b >> (7 - k)) & 1 for b in data for k in range(8)] def check_schedule(recs, lo, hi, n): """Check the in-place order of the host block and the live intervals of the temporaries.""" written = {} for i in range(lo, hi): ao = recs[i][5] if ao[0] == 's': assert ao not in written, ('state bit written twice', ao) written[ao] = i assert len(written) == n, 'a state bit without its next value' for i in range(lo, hi): a0, w0, a1, w1, _, _ = recs[i] for a, w in ((a0, w0), (a1, w1)): if w and a[0] == 's': assert i <= written[a], ('state bit read after it is written', a) last_r, last_w = {}, {} for i, (a0, w0, a1, w1, _, ao) in enumerate(recs): for a, w in ((a0, w0), (a1, w1)): if w and a[0] == 't': last_r[a] = i if ao[0] == 't': last_w[ao] = i assert all(last_w[t] < last_r.get(t, last_w[t] + 1) or t not in last_r for t in last_w), \ 'a temporary written after its last read' return dict(state_bits_written=len(written)) def build_stored(A=8): """The stored host of H_A, with the least counters from CW bits on and the least window that hold it.""" w = CW while True: P = Prog() compile_startup(P, SR.memory_image_a(SR.P_STAR, A), A, w) lo = len(P.recs) n, units, nxt = compile_host(P, A) n_host = len(P.recs) - lo check_schedule(P.recs, lo, len(P.recs), n) compile_device(P, A, w) n_device = len(P.recs) - lo - n_host compile_handover(P, A) recs, addr, top = allocate(P.recs) alpha = (top - 1).bit_length() b = 8 + 3 * alpha while b * len(recs) % 8: # an image of whole bytes recs.insert(len(recs) - 3, (SINK, 0, SINK, 0, -1, SINK)) if 2 * (b * len(recs) // 8) + 1 < 1 << w: break w += 1 image_bits = [x for r in recs for x in encode_record(r, alpha)] image = bits_to_bytes(image_bits) tau = SR.describe(image) assert SR.decode(tau) == image return dict(A=A, w=w, alpha=alpha, b=b, recs=recs, addr=addr, top=top, image=image, image_bits=image_bits, tau=tau, startup_records=lo, n_host_records=n_host, device_records=n_device, host=(n, units, nxt), ends=P.ends) # --------------------------------------------------------------------------- # the machine, transcribed # --------------------------------------------------------------------------- class HeadMachine: """T_{G,L} with a window of W cells; banks and tapes listed from their heads.""" def __init__(self, Q0, Q1, D0, D1, window, alpha=12): self.Q = [list(Q0), list(Q1)] self.D = [list(D0), list(D1)] self.h = [[0, 0], [0, 0]] # heads of the banks and of the tapes self.w = list(window) self.N, self.L = len(Q0), len(D0) self.alpha, self.b = alpha, 8 + 3 * alpha def cell(self, a): k = self.w[KAPPA] if a == RD: return self.D[k][self.h[1][k]] if a == ARM: return self.Q[1 - k][self.h[0][1 - k]] if a == CPY: return self.D[1 - k][self.h[1][1 - k]] return self.w[a] def step(self, rec=None): k = self.w[KAPPA] if rec is None: hq = self.h[0][k] rec = decode_record([self.Q[k][(hq + i) % self.N] for i in range(self.b)], self.alpha) a0, w0, a1, w1, beta, ao = rec v = 1 if w0 * self.cell(a0) + w1 * self.cell(a1) + beta >= 0 else 0 if ao == ARM: self.Q[1 - k][self.h[0][1 - k]] = v elif ao == CPY: self.D[1 - k][self.h[1][1 - k]] = v elif ao != RD: self.w[ao] = v self.h[0][k] = (self.h[0][k] + self.b) % self.N if self.w[AA]: self.h[0][1 - k] = (self.h[0][1 - k] + 1) % self.N if self.w[AR]: self.h[1][k] = (self.h[1][k] + 1) % self.L if self.w[AC]: self.h[1][1 - k] = (self.h[1][1 - k] + 1) % self.L self.w[AA] = self.w[AR] = self.w[AC] = 0 # --------------------------------------------------------------------------- # one pass against H_8 and the device # --------------------------------------------------------------------------- def host_step(n, units, nxt, x): sig = list(x) + [0] * len(units) for u, (b, ins) in enumerate(units): sig[n + u] = 1 if b + sum(w * sig[s] for s, w in ins) >= 0 else 0 return [sig[v] for v in nxt] def check_stored(trials=40, seed=3, verbose=False, A=8): S = build_stored(A) recs, addr, w, alpha = S['recs'], S['addr'], S['w'], S['alpha'] n, units, nxt = S['host'] io = SR.io_cells(A) G = len(recs) rng = random.Random(seed) bad = 0 image_bits = S['image_bits'] m0 = SR.memory_image_a(SR.P_STAR, A) x0 = [0] * n # the state x_0(m*) for j in range(1 << A): for t in range(A): x0[mem_bit(j, t, A)] = (m0[j] >> t) & 1 for trial in range(trials): # a random host state with random device registers, temporaries and tapes; every fifth # state not yet booted, and in every third the requests hold random words x = [rng.randrange(2) for _ in range(n)] x[A] = 0 if trial % 4 else 1 booted = 0 if trial % 5 == 0 else 1 for c in ('c_wr', 'c_rd', 'c_rw'): v = rng.randrange(1 << A) if trial % 3 == 0 else rng.choice((0, 1)) for t in range(A): x[mem_bit(io[c], t, A)] = (v >> t) & 1 L = 64 cnt, lim = rng.randrange(8), rng.randrange(8) dest = rng.randrange(2) win = [rng.randrange(2) for _ in range(1 << alpha)] win[KAPPA] = 0 for i in range(n): win[addr[('s', i)]] = x[i] for i in range(w): win[addr[('d', 'cnt', i)]] = (cnt >> i) & 1 win[addr[('d', 'lim', i)]] = (lim >> i) & 1 win[addr[('d', 'dest')]] = dest win[addr[('d', 'booted')]] = booted win[AR] = win[AA] = win[AC] = 0 D0 = [rng.randrange(2) for _ in range(L)] Q1 = [rng.randrange(2) for _ in range(len(image_bits))] D1 = [rng.randrange(2) for _ in range(L)] M = HeadMachine(image_bits, Q1, D0, D1, win, alpha) for g in range(G): M.step(recs[g]) # expected: the startup if not booted, one step of H_A, then the device on the ring tapes if not booted: x, cnt, lim, dest = list(x0), 0, 0, 0 y = host_step(n, units, nxt, x) get = lambda j: sum(y[mem_bit(j, t, A)] << t for t in range(A)) def put(j, v): for t in range(A): y[mem_bit(j, t, A)] = (v >> t) & 1 e_q1, e_d1, hq1, hd0, hd1 = list(Q1), list(D1), 0, 0, 0 e_cnt, e_lim, e_dest = cnt, lim, dest if get(io['c_wr']) == 1: byte = get(io['r_out']) & 0xFF for kk in range(8): bit = (byte >> (7 - kk)) & 1 if e_dest: e_d1[hd1] = bit; hd1 += 1 else: e_q1[hq1] = bit; hq1 += 1 put(io['r_out'], 0) if get(io['c_rd']) == 1: if not (e_dest and e_cnt == e_lim): byte = sum(D0[hd0 + kk] << (7 - kk) for kk in range(8)) put(io['r_in'], byte if byte < 128 else byte + (1 << A) - 256) hd0 += 8 put(io['c_eot'], 0) e_cnt = (e_cnt + 1) % (1 << w) else: put(io['c_eot'], 1) if get(io['c_rw']) == 1: e_lim, e_cnt, e_dest = e_cnt, 0, 1 for c in ('c_wr', 'c_rd', 'c_rw'): put(io[c], 0) e_kappa = 1 if y[A] else 0 got = [M.w[addr[('s', i)]] for i in range(n)] g_cnt = sum(M.w[addr[('d', 'cnt', i)]] << i for i in range(w)) g_lim = sum(M.w[addr[('d', 'lim', i)]] << i for i in range(w)) parts = dict(state=got == y, cnt=g_cnt == e_cnt, lim=g_lim == e_lim, dest=M.w[addr[('d', 'dest')]] == e_dest, bank=M.Q[1] == e_q1, copy=M.D[1] == e_d1, arm=M.h[0][1] == hq1, read=M.h[1][0] == hd0, copyhead=M.h[1][1] == hd1, kappa=M.w[KAPPA] == e_kappa, tape=M.D[0] == D0, booted=M.w[addr[('d', 'booted')]] == 1 - y[A], requests=M.w[AR] == M.w[AA] == M.w[AC] == 0, active=M.Q[0] == image_bits) ok = all(parts.values()) if not ok and verbose: print(trial, [k for k, v in parts.items() if not v], 'reqs', [get(io[c]) for c in ('c_wr', 'c_rd', 'c_rw')], 'dest', dest, 'cnt', cnt, 'lim', lim, 'halt', x[A], [i for i in range(n) if got[i] != y[i]][:10]) bad += not ok return dict(trials=trials, disagreements=bad, records=G, window_cells=S['top']) class BDD: """Reduced ordered decision diagrams; a node is an integer, 0 and 1 the leaves.""" def __init__(self): self.node = [None, None] self.uniq, self.memo = {}, {} def mk(self, v, lo, hi): if lo == hi: return lo r = self.uniq.get((v, lo, hi)) if r is None: r = self.uniq[(v, lo, hi)] = len(self.node) self.node.append((v, lo, hi)) return r def var(self, v): return self.mk(v, 0, 1) def ite(self, f, g, h): if f < 2: return g if f else h if g == h: return g if g == 1 and h == 0: return f key = (f, g, h) r = self.memo.get(key) if r is None: v = min(self.node[x][0] for x in key if x > 1) lo = [x if x < 2 or self.node[x][0] != v else self.node[x][1] for x in key] hi = [x if x < 2 or self.node[x][0] != v else self.node[x][2] for x in key] r = self.memo[key] = self.mk(v, self.ite(*lo), self.ite(*hi)) return r def record(self, w0, x, w1, y, beta): """H(w0 x + w1 y + beta) of the bits x and y.""" val = lambda a, b: 1 if w0 * a + w1 * b + beta >= 0 else 0 return self.ite(x, self.ite(y, val(1, 1), val(1, 0)), self.ite(y, val(0, 1), val(0, 0))) def atleast(self, t, lits): """At least t of the literals (x, positive) are true.""" if t <= 0: return 1 row = [1] + [0] * t for x, pos in reversed(lits): row = [1] + [self.ite(x, row[j - 1], row[j]) if pos else self.ite(x, row[j], row[j - 1]) for j in range(1, t + 1)] return row[t] def symbolic_check(A=8): """The startup and step records of the image of S_A on decision diagrams over the signals of H_A.""" sys.setrecursionlimit(100000) t0 = time.time() S = build_stored(A) n, units, nxt = host_units(A) b, alpha = S['b'], S['alpha'] bits = bytes_to_bits(S['image']) recs = [decode_record(bits[k * b:(k + 1) * b], alpha) for k in range(len(bits) // b)] lo = S['startup_records'] hi = lo + S['n_host_records'] addr = S['addr'] D = BDD() # the value of every signal on the current state: a unit that is a constant or a literal is that, # and the units of one function share the variable of the first; fn[u] is the function of unit u # of the values of its predecessors; later signals lie nearer the root var = lambda s: D.var((1 << 29) - s) top = lambda x: D.node[x][0] if x > 1 else 1 << 31 val, first, fn = [var(s) for s in range(n)], {}, [] for u, (bias, ins) in enumerate(units): lits = sorted(((val[s], wt > 0) for s, wt in ins), key=lambda l: top(l[0])) f = D.atleast(sum(1 for _, wt in ins if wt < 0) - bias, lits) fn.append(f) val.append(var(first.setdefault(f, n + u)) if f > 1 and D.node[f][1:] not in ((0, 1), (1, 0)) else f) unknown = lambda c: D.var(1 << 30 | c) state = [addr[('s', i)] for i in range(n)] ends = S['ends'] agree = [True] def run(window, a, z): # at the record that the compiler names as writing the value of unit u, the cell must hold # the function of u, and from then on it holds the value of u for k in range(a, z): a0, w0, a1, w1, beta, ao = recs[k] assert ao >= 7, ('a record writes a head or a request', k) g = D.record(w0, window[a0], w1, window[a1], beta) if k in ends: u = ends[k] agree[0] &= g == fn[u] g = val[n + u] window[ao] = g # the startup: x_0(m*) when booted is 0, the window unchanged when it is 1 boot = addr[('d', 'booted')] window = [unknown(c) for c in range(1 << alpha)] regs = [addr[('d', 'cnt', i)] for i in range(S['w'])] + [addr[('d', 'lim', i)] for i in range(S['w'])] + \ [addr[('d', 'dest')]] before = list(window) run(window, 0, lo) B0 = before[boot] m0 = SR.memory_image_a(SR.P_STAR, A) x0 = [0] * n for j in range(1 << A): for t in range(A): x0[mem_bit(j, t, A)] = (m0[j] >> t) & 1 startup = window[boot] == 1 and all(window[c] == D.ite(B0, before[c], 0) for c in regs) and \ all(window[state[i]] == D.ite(B0, before[state[i]], x0[i]) for i in range(n)) # the step: from every window, the state bits end holding the next state of H_A window = [unknown(c) for c in range(1 << alpha)] for i in range(n): window[state[i]] = val[i] run(window, lo, hi) step = agree[0] and len(ends) == len(units) and all(window[state[i]] == val[nxt[i]] for i in range(n)) return dict(A=A, startup_records=lo, step_records=hi - lo, startup_exact=startup, step_exact=step, sha256_image=hashlib.sha256(S['image']).hexdigest(), nodes=len(D.node), seconds=round(time.time() - t0, 1)) def stored_update_record(key, value): rec = json.load(open(STORED_RECORD)) if os.path.exists(STORED_RECORD) else {} rec[key] = value json.dump(rec, open(STORED_RECORD, 'w'), indent=1) def census(tau): """The literal and repeat tokens of a recipe and the bytes each kind gives.""" i = lit = rep = flit = frep = 0 while tau[i]: t = tau[i] if t < 128: lit, flit, i = lit + 1, flit + t, i + 1 + t else: rep, frep, i = rep + 1, frep + 256 - t, i + 2 return lit, rep, flit, frep def passes(tau): """The steps of P* on a recipe, Proposition 5.4.""" lit, rep, flit, frep = census(tau) return 5 * flit + 4 * frep + 7 * lit + 9 * rep + 7 + 1 + 6 * len(tau) + 2 def stored_key(name, A): return name if A == 8 else f'{name}_A{A}' def run_stored(generations, A=8): """Generations of the stored host on the machine in C, against the reference run of P*.""" S = build_stored(A) recs, image_bits, tau, alpha = S['recs'], S['image_bits'], S['tau'], S['alpha'] G = len(recs) out, steps = SR.run_reference_a(SR.memory_image_a(SR.P_STAR, A), tau, A) assert out == S['image'] + tau and steps == passes(tau) tau_bits = bytes_to_bits(tau) work = os.path.join(REPO, 'variants') inp, exe = os.path.join(work, f'.stored_host_{A}.bin'), os.path.join(work, f'.stored_host_run_{A}') with open(inp, 'wb') as f: f.write(struct.pack('<4i', 1 << alpha, alpha, G, len(tau_bits))) f.write(bytes(image_bits) + bytes(tau_bits) + bytes(1 << alpha)) subprocess.run(['gcc', '-O3', '-march=native', '-o', exe, os.path.join(REPO, 'src', 'stored_host_run.c')], check=True) t0 = time.time() p = subprocess.run([exe, inp, str(generations)], capture_output=True, text=True) os.remove(inp); os.remove(exe) lines = [json.loads(l) for l in p.stdout.split('\n') if l.strip()] gens = [l for l in lines if 'generation' in l] res = dict(A=A, records=G, window_cells=S['top'], record_bits=S['b'], counter_bits=S['w'], bank_bits=len(image_bits), image_bytes=len(S['image']), tau_bytes=len(tau), passes_expected=steps, generations=gens, seconds=round(time.time() - t0, 1), sha256_image=hashlib.sha256(S['image']).hexdigest(), sha256_tau=hashlib.sha256(tau).hexdigest()) for l in lines: if 'state_after_3_equals_after_1' in l: res['state_after_3_equals_after_1'] = l['state_after_3_equals_after_1'] res['every_generation_exact'] = all(g['passes'] == steps and g['steps'] == steps * G and g['bank_holds_image'] and g['tape_holds_tau'] for g in gens) and len(gens) == generations stored_update_record(stored_key('run', A), res) return res def stored_widths(widths): """The stored host of every word width: its records, window, counters, image and generation.""" rows = [] for A in widths: t0 = time.time() S = build_stored(A) G, tau = len(S['recs']), S['tau'] lit, rep, flit, frep = census(tau) P = passes(tau) rows.append(dict(A=A, records=G, startup=S['startup_records'], step=S['n_host_records'], device=S['device_records'], handover=G - S['startup_records'] - S['n_host_records'] - S['device_records'], window_cells=S['top'], alpha=S['alpha'], record_bits=S['b'], counter_bits=S['w'], image_bytes=len(S['image']), tau_bytes=len(tau), literal_tokens=lit, repeat_tokens=rep, passes=P, applications=P * G, seconds=round(time.time() - t0, 1))) print(json.dumps(rows[-1]), flush=True) stored_update_record('widths', rows) return rows def cmd_stored(): args = sys.argv[1:] A = 8 if '--width' in args: i = args.index('--width') A = int(args[i + 1]) del args[i:i + 2] cmd = args[0] if args else 'build' if cmd == 'check': t0 = time.time() res = check_stored(int(args[1]) if len(args) > 1 else 40, A=A) res['seconds'] = round(time.time() - t0, 1) stored_update_record(stored_key('pass_check', A), res) print(json.dumps(res)) sys.exit(0 if res['disagreements'] == 0 else 1) elif cmd == 'symbolic': res = symbolic_check(A) stored_update_record(stored_key('symbolic', A), res) print(json.dumps(res)) sys.exit(0 if res['startup_exact'] and res['step_exact'] else 1) elif cmd == 'run': gens = int(args[1]) if len(args) > 1 else 3 res = run_stored(gens, A) print(json.dumps(res, indent=1)) sys.exit(0 if res['every_generation_exact'] and (gens < 3 or res.get('state_after_3_equals_after_1') == 1) else 1) elif cmd == 'widths': stored_widths([int(a) for a in args[1:]] or list(range(8, 13))) elif cmd == 'build': t0 = time.time() S = build_stored(A) print(json.dumps(dict(A=A, records=len(S['recs']), host_records=S['n_host_records'], window_cells=S['top'], record_bits=S['b'], counter_bits=S['w'], bank_bits=len(S['image_bits']), image_bytes=len(S['image']), tau_bytes=len(S['tau']), sha256_image=hashlib.sha256(S['image']).hexdigest(), seconds=round(time.time() - t0, 1)))) COMMANDS = {'organism': main_organism, 'net': main_net, 'stored': cmd_stored} if __name__ == "__main__": if len(sys.argv) < 2 or sys.argv[1] not in COMMANDS: sys.exit('commands: ' + ', '.join(COMMANDS)) cmd = sys.argv.pop(1) sys.argv[0] += ' ' + cmd sys.exit(COMMANDS[cmd]())