"""The universal constructor as a netlist stored in the interpreter's state. Section 7 of the paper interprets stored netlists; this module stores the paper's own construction. A microprogram of records implements the SUBLEQ transducer of Definitions 2.9 and 3.1 over a 128-byte memory, with the tape, the output word, the head and the write pointer all held in signals of the interpreter, and runs the constructor P on a recipe. One pass of the record region is one SUBLEQ step: fetch, subtract, write back, apply the device, branch. Nothing outside U takes part, not even U's own output device. The program is P with its variable and device cells moved from the top of a 256-byte memory to the top of a 128-byte one, which is the same program: every address it names is a datum, and the branch targets are unchanged. python src/hosted_constructor.py [--target-bytes N] """ from __future__ import annotations import argparse import json import os import sys import time from typing import Dict, List, Tuple sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import torch from reflect import (Cfg, Leveled, build_net, encode_netlist, state_to_vec, to_bits) from selfrep import P, describe, decode REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) MEM_BYTES = 128 ADDR = 7 # address bits of the hosted machine def remap(x: int) -> int: """Move a cell of the 256-byte memory to the same offset from the top of a 128-byte one.""" return x - 0x80 if x >= 0xF0 else x P128 = [(remap(a), remap(b), remap(c)) for a, b, c in P] Z, ONE, T1, T2, NEG1, EOK = 0x70, 0x71, 0x72, 0x73, 0x74, 0x75 C_WR, C_EOT, C_RW, C_RD, R_IN, R_OUT = 0x79, 0x7A, 0x7B, 0x7C, 0x7D, 0x7E HALT_PC = 0x7F def image128() -> List[int]: mem = [0] * MEM_BYTES for idx, (a, b, c) in enumerate(P128): mem[idx * 3] = a mem[idx * 3 + 1] = b mem[idx * 3 + 2] = c mem[ONE] = 1 mem[NEG1] = 0xFF mem[EOK] = 1 return mem def reference128(mem0: List[int], tape: bytes, max_steps: int = 1 << 22): """The transducer of Definitions 2.9 and 3.1 on a 128-byte memory.""" mem = list(mem0) out = bytearray() h = 0 pc = 0 n = 0 while pc != HALT_PC and n < max_steps: A = mem[pc % MEM_BYTES] B = mem[(pc + 1) % MEM_BYTES] C = mem[(pc + 2) % MEM_BYTES] r = (mem[B % MEM_BYTES] - mem[A % MEM_BYTES]) & 0xFF mem[B % MEM_BYTES] = r pc = C % MEM_BYTES if (r == 0 or r >= 0x80) else (pc + 3) % MEM_BYTES if mem[C_WR] == 1: out.append(mem[R_OUT]) mem[R_OUT] = 0 if mem[C_RD] == 1: if h < len(tape): mem[R_IN] = tape[h] mem[C_EOT] = 0 h += 1 else: mem[C_EOT] = 1 if mem[C_RW] == 1: h = 0 mem[C_WR] = mem[C_RD] = mem[C_RW] = 0 n += 1 return bytes(out), n, mem, h # ============================================================================= # the microprogram # ============================================================================= class Micro: """Records, with work signals allocated as they are named.""" def __init__(self, cfg, work_base: int, work_max: int): self.cfg = cfg self.recs: List[Tuple[List[Tuple[int, int, int]], int, Tuple[int, int]]] = [] self.next = work_base self.limit = work_base + work_max self.named: Dict[str, int] = {} def sig(self, name: str = None) -> int: s = self.next self.next += 1 if self.next > self.limit: raise ValueError("work signals exhausted") if name: self.named[name] = s return s def reg(self, name: str, width: int) -> List[int]: r = [self.sig() for _ in range(width)] self.named[name] = r[0] return r def emit(self, slots, bias, out): self.recs.append((slots, bias, out)) # --- one-record primitives ------------------------------------------ def const(self, dst, v): self.emit([], 0 if v else -1, (dst, 0)) def copy(self, dst, src): self.emit([(src, 1, 0)], -1, (dst, 0)) def not_(self, dst, src): self.emit([(src, -1, 0)], 0, (dst, 0)) def and2(self, dst, a, b, na=False, nb=False): wa, wb = (-1 if na else 1), (-1 if nb else 1) bias = -((0 if na else 1) + (0 if nb else 1)) self.emit([(a, wa, 0), (b, wb, 0)], bias, (dst, 0)) def or2(self, dst, a, b): self.emit([(a, 1, 0), (b, 1, 0)], -1, (dst, 0)) def read_idx(self, dst, base): """dst <- sig[base + PTR].""" self.emit([(base, 1, 1)], -1, (dst, 0)) def write_idx(self, base, src): """sig[base + PTR] <- src.""" self.emit([(src, 1, 0)], -1, (base, 1)) def write_idx_and(self, base, src, gate, ngate=False): """sig[base + PTR] <- src AND gate (or src AND NOT gate).""" wg = -1 if ngate else 1 bias = -1 if ngate else -2 self.emit([(src, 1, 0), (gate, wg, 0)], bias, (base, 1)) # --- small cells ------------------------------------------------------ def xor(self, dst, a, b): t1, t2 = self.sig(), self.sig() self.or2(t1, a, b) self.emit([(a, -1, 0), (b, -1, 0)], 1, (t2, 0)) # NAND self.and2(dst, t1, t2) def mux(self, dst, sel, x1, x0): ns, a1, a0 = self.sig(), self.sig(), self.sig() self.not_(ns, sel) self.and2(a1, x1, sel) self.and2(a0, x0, ns) self.or2(dst, a1, a0) def and_lits(self, dst, lits): """dst <- conjunction of literals (signal, want) via a binary tree.""" level = [] i = 0 while i < len(lits): if i + 1 < len(lits): (sa, wa), (sb, wb) = lits[i], lits[i + 1] t = self.sig() self.and2(t, sa, sb, na=not wa, nb=not wb) level.append(t) i += 2 else: sa, wa = lits[i] t = self.sig() if wa: self.copy(t, sa) else: self.not_(t, sa) level.append(t) i += 1 while len(level) > 1: nxt = [] for j in range(0, len(level) - 1, 2): last = len(level) == 2 t = dst if last else self.sig() self.and2(t, level[j], level[j + 1]) nxt.append(t) if len(level) % 2: nxt.append(level[-1]) level = nxt if level[0] != dst: self.copy(dst, level[0]) def incr_gated(self, reg, gate): """reg <- reg + gate, on len(reg) bits, least significant first.""" carries = [gate] for k in range(len(reg) - 1): c = self.sig() self.and2(c, reg[k], carries[k]) carries.append(c) for k in range(len(reg)): self.xor(reg[k], reg[k], carries[k]) def add_const(self, dst, src, c): """dst <- src + c on len(src) bits, least significant first.""" carry = None for k in range(len(src)): bit = (c >> k) & 1 if bit and carry is None: self.not_(dst[k], src[k]) carry = src[k] elif bit: t = self.sig() self.xor(t, src[k], carry) self.not_(dst[k], t) nc = self.sig() self.or2(nc, src[k], carry) carry = nc elif carry is None: self.copy(dst[k], src[k]) else: nc = self.sig() self.and2(nc, src[k], carry) self.xor(dst[k], src[k], carry) carry = nc def build_microprogram(cfg, mem_base, tape_base, out_base, tape_len, work_base, work_max): """The records implementing one SUBLEQ step with its device.""" m = Micro(cfg, work_base, work_max) PC = m.reg("PC", ADDR) # least significant first P1 = m.reg("P1", ADDR) P2 = m.reg("P2", ADDR) P3 = m.reg("P3", ADDR) Ab = m.reg("A", 8) Bb = m.reg("B", 8) Cb = m.reg("C", 8) X = m.reg("X", 8) Y = m.reg("Y", 8) Rb = m.reg("R", 8) NPC = m.reg("NPC", ADDR) head = m.reg("head", ADDR) wp = m.reg("wp", ADDR) def cell(addr, bit): """The signal holding bit `bit` of memory cell `addr`.""" return mem_base + bit * MEM_BYTES + addr def ptr_set(src): """PTR <- src, on ADDR bits; the pointer is stored most significant first, so bit k of the value is coordinate PTR_BASE + A - 1 - k.""" for k in range(ADDR): m.copy(cfg.PTR_BASE + cfg.A - 1 - k, src[k]) # the pointer's high bits stay zero for k in range(ADDR, cfg.A): m.const(cfg.PTR_BASE + cfg.A - 1 - k, 0) m.add_const(P1, PC, 1) m.add_const(P2, PC, 2) m.add_const(P3, PC, 3) for reg, src in ((Ab, PC), (Bb, P1), (Cb, P2)): ptr_set(src) for b in range(8): m.read_idx(reg[b], mem_base + b * MEM_BYTES) for reg, src in ((X, Ab), (Y, Bb)): ptr_set(src[:ADDR]) for b in range(8): m.read_idx(reg[b], mem_base + b * MEM_BYTES) # R = Y - X, and the branch decision nx = [m.sig() for _ in range(8)] for b in range(8): m.not_(nx[b], X[b]) carry = None for b in range(8): t = m.sig() if carry is None: # carry in is 1 m.xor(t, Y[b], nx[b]) m.not_(Rb[b], t) c = m.sig() m.or2(c, Y[b], nx[b]) else: m.xor(t, Y[b], nx[b]) m.xor(Rb[b], t, carry) c1, c2, c = m.sig(), m.sig(), m.sig() m.and2(c1, Y[b], nx[b]) m.and2(c2, t, carry) m.or2(c, c1, c2) carry = c nz = m.sig() tree = Rb[0] for b in range(1, 8): t = m.sig() m.or2(t, tree, Rb[b]) tree = t m.not_(nz, tree) leq = m.sig() m.or2(leq, Rb[7], nz) # write back to M[B] ptr_set(Bb[:ADDR]) for b in range(8): m.write_idx(mem_base + b * MEM_BYTES, Rb[b]) # --- the device ------------------------------------------------------ wr, rd, rw = m.sig(), m.sig(), m.sig() m.and_lits(wr, [(cell(C_WR, 0), 1)] + [(cell(C_WR, b), 0) for b in range(1, 8)]) m.and_lits(rd, [(cell(C_RD, 0), 1)] + [(cell(C_RD, b), 0) for b in range(1, 8)]) m.and_lits(rw, [(cell(C_RW, 0), 1)] + [(cell(C_RW, b), 0) for b in range(1, 8)]) # read: R_IN <- tape[head] when the head is inside the tape ateot = m.sig() m.and_lits(ateot, [(head[k], (tape_len >> k) & 1) for k in range(ADDR)]) inrange = m.sig() m.not_(inrange, ateot) g = m.sig() m.and2(g, rd, inrange) ptr_set(head) tb = [m.sig() for _ in range(8)] for b in range(8): m.read_idx(tb[b], tape_base + b * tape_len) for b in range(8): m.mux(cell(R_IN, b), g, tb[b], cell(R_IN, b)) # end-of-tape status: set by a read at the end, cleared by a read inside m.mux(cell(C_EOT, 0), rd, ateot, cell(C_EOT, 0)) for b in range(1, 8): m.and2(cell(C_EOT, b), cell(C_EOT, b), rd, nb=True) m.incr_gated(head, g) for k in range(ADDR): # rewind sets the head to zero m.and2(head[k], head[k], rw, nb=True) # write: append R_OUT to the output word, then clear R_OUT ptr_set(wp) for b in range(8): m.write_idx_and(out_base + b * tape_len, cell(R_OUT, b), wr) for b in range(8): m.and2(cell(R_OUT, b), cell(R_OUT, b), wr, nb=True) m.incr_gated(wp, wr) # the request cells are cleared at the end of every step for b in range(8): m.const(cell(C_WR, b), 0) m.const(cell(C_RD, b), 0) m.const(cell(C_RW, b), 0) # --- branch and halt -------------------------------------------------- for k in range(ADDR): m.mux(NPC[k], leq, Cb[k], P3[k]) m.and_lits(cfg.HALT_SIG, [(NPC[k], (HALT_PC >> k) & 1) for k in range(ADDR)]) for k in range(ADDR): m.copy(PC[k], NPC[k]) return m def run_records(recs, sig, cfg, max_passes): """Definition 7.1 applied record by record, in place. Identical in semantics to reflect.ref_gate, and used here so that the microprogram can be checked against the reference machine without evaluating the interpreter's netlist. """ S, PB, A = cfg.S, cfg.PTR_BASE, cfg.A for p in range(max_passes): for slots, bias, (oa, oidx) in recs: ptr = 0 for k in range(A): ptr = (ptr << 1) | sig[PB + k] acc = bias for (a, w, idx) in slots: acc += w * sig[(a + (ptr if idx else 0)) & (S - 1)] sig[(oa + (ptr if oidx else 0)) & (S - 1)] = 1 if acc >= 0 else 0 if sig[cfg.HALT_SIG]: return p + 1, True return max_passes, False def layout(tape_len: int, out_len: int, work_max: int = 700, min_records: int = 1): """The smallest configuration with room for the microprogram and the regions it addresses.""" for A in range(12, 22): S = 1 << A R = 3 * A + 11 work_base = A + 6 mem_base = work_base + work_max tape_base = mem_base + 8 * MEM_BYTES out_base = tape_base + 8 * tape_len top = out_base + 8 * out_len for G in range((S - top) // R, min_records - 1, -1): span = S - G * R if span >= top + 8: return A, G, span, mem_base, tape_base, out_base, work_base raise ValueError("no configuration is large enough") def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--target", default="ABCCCCCCCCDE") ap.add_argument("--device", default="cpu") args = ap.parse_args() target = args.target.encode() tape = describe(target) assert decode(tape) == target tape_len = out_len = 64 assert len(tape) < tape_len and len(target) < out_len A, G, span, mem_base, tape_base, out_base, work_base = layout( tape_len, out_len, min_records=520) cfg = Cfg(A=A, G=G, banks=1, self_mod=False) assert cfg.span == span, (cfg.span, span) m = build_microprogram(cfg, mem_base, tape_base, out_base, tape_len, work_base, 700) n = len(m.recs) assert n <= G, (n, G) print(f" hosted machine: SUBLEQ over {MEM_BYTES} bytes with the tape, the " f"output word, the head and the write pointer in signals", flush=True) print(f" microprogram: {n} records; interpreter U_S': S = {cfg.S}, " f"G = {cfg.G}, b = {cfg.R}, addressed span {cfg.span}", flush=True) print(f" regions: work {work_base}..{m.next - 1}, memory {mem_base}, " f"tape {tape_base}, output {out_base}", flush=True) exp_out, exp_steps, exp_mem, exp_head = reference128(image128(), tape) assert exp_out == target, (exp_out, target) print(f" the reference runs P on a {len(tape)}-byte recipe in " f"{exp_steps} instructions, emitting {len(exp_out)} bytes", flush=True) nl = encode_netlist(cfg, m.recs) def initial_signals(): sig = [0] * cfg.S sig[cfg.NET0:cfg.NET0 + len(nl)] = nl mem0 = image128() for j in range(MEM_BYTES): for b in range(8): sig[mem_base + b * MEM_BYTES + j] = (mem0[j] >> b) & 1 for j, byte in enumerate(tape): for b in range(8): sig[tape_base + b * tape_len + j] = (byte >> b) & 1 return sig def readout(sig): def byte_at(base, stride, j): return sum(int(sig[base + b * stride + j]) << b for b in range(8)) return (bytes(byte_at(out_base, tape_len, j) for j in range(len(target))), [byte_at(mem_base, MEM_BYTES, j) for j in range(MEM_BYTES)], sum(int(sig[m.named["wp"] + k]) << k for k in range(ADDR)), sum(int(sig[m.named["head"] + k]) << k for k in range(ADDR))) # first at the level of Definition 7.1, record by record t0 = time.perf_counter() rsig = initial_signals() rpasses, rhalted = run_records(m.recs, rsig, cfg, exp_steps + 3) r_out, r_mem, r_wp, r_head = readout(rsig) rec_ok = (rhalted and rpasses == exp_steps and r_out == target and r_mem == exp_mem and r_head == exp_head) print(f" record semantics: halted after {rpasses} passes, emitted " f"{r_wp} bytes {r_out!r}, memory and head match the reference: " f"{'yes' if rec_ok else 'NO'} ({time.perf_counter() - t0:.1f}s)", flush=True) t0 = time.perf_counter() unet, uin, uout = build_net(cfg) lev = Leveled(unet, uin, uout, device=args.device) print(f" U_S' netlist: {len(unet.gates):,} units, {len(lev.plan)} layers " f"({time.perf_counter() - t0:.0f}s)", flush=True) V = state_to_vec(cfg, {"sig": initial_signals(), "gp": 0, "halt": 0} ).unsqueeze(0).to(args.device) t0 = time.perf_counter() halted_at = None for inst in range(exp_steps + 2): for _ in range(cfg.G): V = lev.step(V) if int(V[0, cfg.S + cfg.GPW]) == 1: halted_at = inst + 1 break if inst and inst % 20 == 0: print(f" {inst}/{exp_steps} instructions " f"({inst * cfg.G / (time.perf_counter() - t0):.0f} steps/s)", flush=True) dt = time.perf_counter() - t0 got, got_mem, wp_val, head_val = readout(V[0].cpu()) exact = got == target mem_same = got_mem == exp_mem print(f" ran to a halt after {halted_at} instructions " f"({halted_at * cfg.G:,} interpreter steps, {dt / 60:.1f} min, " f"{halted_at * cfg.G / dt:.0f} steps/s)", flush=True) print(f" emitted {wp_val} bytes: {got!r}") print(f" equals the target: {'yes' if exact else 'NO'}; final memory " f"equals the reference: {'yes' if mem_same else 'NO'}; head " f"{head_val} of {len(tape)}: " f"{'yes' if head_val == exp_head else 'NO'}") ok = (rec_ok and exact and mem_same and halted_at == exp_steps and head_val == exp_head) d = os.path.join(REPO, "paper", "runs") os.makedirs(d, exist_ok=True) json.dump({"records": n, "record_semantics_ok": rec_ok, "S": cfg.S, "G": cfg.G, "b": cfg.R, "span": cfg.span, "units": len(unet.gates), "layers": len(lev.plan), "memory_bytes": MEM_BYTES, "tape_bytes": len(tape), "target_bytes": len(target), "target": target.decode(), "instructions": halted_at, "reference_instructions": exp_steps, "interpreter_steps": halted_at * cfg.G, "emitted": got.decode(errors="replace"), "exact": exact, "memory_matches_reference": mem_same, "head": head_val, "seconds": dt, "ok": ok}, open(os.path.join(d, "paper_hosted.json"), "w"), indent=1) print("HOSTED CONSTRUCTOR:", "PASS" if ok else "FAIL") return 0 if ok else 1 if __name__ == "__main__": sys.exit(main())