Prove self-reproduction by the dynamics for the stored host of every word width, with a sign-extending device read and the record count in closed form, check the stored hosts of widths 9 and 10 symbolically and on random states, check the extracted host family for widths 8 to 10, and remove the to-do note
dd2a421 | """The SUBLEQ host: its family, serialization, device, constructor, self-reproduction and evaluators.""" | |
| from __future__ import annotations | |
| import argparse | |
| import hashlib | |
| import json | |
| import os | |
| import struct | |
| import sys | |
| import time | |
| from typing import Dict, List, Tuple, Optional | |
| import numpy as np | |
| import torch | |
| from safetensors import safe_open | |
| from safetensors.torch import save_file | |
| from machines import Net, compile_net | |
| _DT = {torch.float32:"F32", torch.float64:"F64", torch.int64:"I64", | |
| torch.int32:"I32", torch.int16:"I16", torch.int8:"I8", | |
| torch.uint8:"U8", torch.bool:"BOOL"} | |
| def canonical_bytes(tensors, metadata_pairs): | |
| """Canonical safetensors bytes from tensors and metadata pairs in canonical order.""" | |
| header = {} | |
| if metadata_pairs is not None: | |
| header["__metadata__"] = {k: v for k, v in metadata_pairs} | |
| blobs, off = [], 0 | |
| for name, t in tensors: | |
| t = t.contiguous() | |
| b = t.numpy().tobytes() | |
| header[name] = {"dtype": _DT[t.dtype], "shape": list(t.shape), | |
| "data_offsets": [off, off + len(b)]} | |
| blobs.append(b); off += len(b) | |
| s = json.dumps(header, separators=(",", ":"), ensure_ascii=False).encode("utf-8") | |
| pad = (-(8 + len(s))) % 8 | |
| s = s + b" " * pad | |
| return struct.pack("<Q", len(s)) + s + b"".join(blobs) | |
| VERSION = 6 | |
| def io_cells(A: int) -> Dict[str, int]: | |
| S = 1 << A | |
| return {"c_wr": S - 7, "c_eot": S - 6, "c_rw": S - 5, "c_rd": S - 4, | |
| "r_in": S - 3, "r_out": S - 2, "halt_pc": S - 1} | |
| def layout(A: int) -> Dict[str, List[int]]: | |
| return {"pc": [0, A], "halt": [A, 1], "mem": [A + 1, 1 << A, A]} | |
| def state_names(A: int) -> List[str]: | |
| S = 1 << A | |
| return ([f"pc{i}" for i in range(A)] + ["halt"] | |
| + [f"m{j}b{t}" for j in range(S) for t in range(A)]) | |
| def build(A: int, fold: bool = False) -> Tuple[Net, List[str], List[str]]: | |
| """The step netlist of H_A, units in serialization order.""" | |
| S = 1 << A | |
| net = Net() | |
| g = net.g | |
| pc = [f"pc{i}" for i in range(A)] | |
| halt = "halt" | |
| mu = lambda j, t: f"m{j}b{t}" | |
| def lits(bits: List[str], value: int): | |
| """The literals asserting that the MSB-first `bits` spell `value`.""" | |
| return [(bits[i], 1 if (value >> (A - 1 - i)) & 1 else -1) for i in range(A)] | |
| # B1 decode lines of the counter | |
| for j in range(S): | |
| g(f"P{j}", lits(pc, j), -bin(j).count("1")) | |
| # B2 the counter held by the halt bit | |
| for i in range(A): | |
| g(f"HP{i}", [(pc[i], 1), (halt, 1)], -2) | |
| # B3 the fetch conjunctions: operand o, bit t, counter value j | |
| for o in range(3): | |
| for t in range(A): | |
| for j in range(S): | |
| if fold: | |
| g(f"FA{o}_{t}_{j}", [(mu((j + o) % S, t), 1)] + lits(pc, j), -1 - bin(j).count("1")) | |
| else: | |
| g(f"FA{o}_{t}_{j}", [(mu((j + o) % S, t), 1), (f"P{j}", 1)], -2) | |
| # B4 the bits of p + 3, read off the decode lines | |
| for i in range(A): | |
| g(f"Q{i}", [(f"P{j}", 1) for j in range(S) | |
| if (((j + 3) % S) >> (A - 1 - i)) & 1], -1) | |
| # B5 the three operand words A, B, C | |
| for o in range(3): | |
| for t in range(A): | |
| g(f"F{o}_{t}", [(f"FA{o}_{t}_{j}", 1) for j in range(S)], -1) | |
| Aop = [f"F0_{t}" for t in range(A)] | |
| Bop = [f"F1_{t}" for t in range(A)] | |
| Cop = [f"F2_{t}" for t in range(A)] | |
| # B6 decode lines of A and B | |
| if not fold: | |
| for j in range(S): | |
| g(f"PA{j}", lits(Aop, j), -bin(j).count("1")) | |
| for j in range(S): | |
| g(f"PB{j}", lits(Bop, j), -bin(j).count("1")) | |
| # B7 C is the halt address | |
| g("AC", [(c, 1) for c in Cop], -A) | |
| # B8 write enables | |
| for j in range(S): | |
| if fold: | |
| g(f"W{j}", [(halt, -1)] + lits(Bop, j), -bin(j).count("1")) | |
| else: | |
| g(f"W{j}", [(halt, -1), (f"PB{j}", 1)], -1) | |
| # B9 the read conjunctions of M[A] and M[B] | |
| for t in range(A): | |
| for j in range(S): | |
| if fold: | |
| g(f"RX{t}_{j}", [(mu(j, t), 1)] + lits(Aop, j), -1 - bin(j).count("1")) | |
| else: | |
| g(f"RX{t}_{j}", [(mu(j, t), 1), (f"PA{j}", 1)], -2) | |
| for t in range(A): | |
| for j in range(S): | |
| if fold: | |
| g(f"RY{t}_{j}", [(mu(j, t), 1)] + lits(Bop, j), -1 - bin(j).count("1")) | |
| else: | |
| g(f"RY{t}_{j}", [(mu(j, t), 1), (f"PB{j}", 1)], -2) | |
| # B10 the cells kept | |
| for j in range(S): | |
| for t in range(A): | |
| g(f"KP{j}_{t}", [(mu(j, t), 1), (f"W{j}", -1)], -1) | |
| # B11 X = M[A], Y = M[B] | |
| for t in range(A): | |
| g(f"X{t}", [(f"RX{t}_{j}", 1) for j in range(S)], -1) | |
| for t in range(A): | |
| g(f"Y{t}", [(f"RY{t}_{j}", 1) for j in range(S)], -1) | |
| xv = lambda i: f"X{A - 1 - i}" # value bit i, least significant first | |
| yv = lambda i: f"Y{A - 1 - i}" | |
| # B12 propagate signals of Y + not X | |
| for i in range(A - 1): | |
| g(f"PR{i}", [(xv(i), -1), (yv(i), 1)], 0) | |
| # B13 carry terms into bit k: the incoming carry through all lower | |
| # positions, and a generate at j propagated through j+1..k-1 | |
| for k in range(1, A): | |
| g(f"T{k}_n", [(f"PR{i}", 1) for i in range(k)], -k) | |
| for j in range(k): | |
| g(f"T{k}_{j}", [(yv(j), 1), (xv(j), -1)] | |
| + [(f"PR{i}", 1) for i in range(j + 1, k)], -(1 + (k - 1 - j))) | |
| # B14 carries | |
| for k in range(1, A): | |
| g(f"C{k}", [(f"T{k}_n", 1)] + [(f"T{k}_{j}", 1) for j in range(k)], -1) | |
| # B15 X = Y, bit by bit | |
| for i in range(A): | |
| g(f"AX{i}", [(xv(i), 1), (yv(i), 1)], -2) | |
| for i in range(A): | |
| g(f"NX{i}", [(xv(i), -1), (yv(i), -1)], 0) | |
| for i in range(A): | |
| g(f"XN{i}", [(f"AX{i}", 1), (f"NX{i}", 1)], -1) | |
| # B16 | |
| g("EQ", [(f"XN{i}", 1) for i in range(A)], -A) | |
| # B17 the sum bits: bit 0 is X xor Y, bit k the parity of y_k + not x_k + c_k | |
| g("O0", [(yv(0), 1), (xv(0), 1)], -1) | |
| g("N0", [(yv(0), -1), (xv(0), -1)], 1) | |
| g("R0", [("O0", 1), ("N0", 1)], -2) | |
| for k in range(1, A): | |
| for m, b in ((1, 0), (2, -1), (3, -2)): | |
| g(f"H{m}_{k}", [(yv(k), 1), (xv(k), -1), (f"C{k}", 1)], b) | |
| for k in range(1, A): | |
| g(f"R{k}", [(f"H1_{k}", 1), (f"H2_{k}", -1), (f"H3_{k}", 1)], -1) | |
| top = A - 1 | |
| g("LEQ", [(f"H1_{top}", 1), (f"H2_{top}", -1), (f"H3_{top}", 1), ("EQ", 1)], -1) | |
| # B18 the next counter | |
| for i in range(A): | |
| g(f"TA{i}", [(halt, -1), ("LEQ", 1), (Cop[i], 1)], -2) | |
| for i in range(A): | |
| g(f"TB{i}", [(halt, -1), ("LEQ", -1), (f"Q{i}", 1)], -1) | |
| for i in range(A): | |
| g(f"NP{i}", [(f"HP{i}", 1), (f"TA{i}", 1), (f"TB{i}", 1)], -1) | |
| # B19 the halt latch: the next counter is 2^A - 1 | |
| g("HA", [(halt, -1), ("LEQ", 1), ("AC", 1)], -2) | |
| g("HB", [(halt, -1), ("LEQ", -1), (f"P{S - 4}", 1)], -1) | |
| g("NH", [(halt, 1), ("HA", 1), ("HB", 1)], -1) | |
| # B20 the write: bit t of the word is value bit A-1-t of the difference | |
| for j in range(S): | |
| for t in range(A): | |
| g(f"WR{j}_{t}", [(f"R{A - 1 - t}", 1), (f"W{j}", 1)], -2) | |
| for j in range(S): | |
| for t in range(A): | |
| g(f"NM{j}_{t}", [(f"KP{j}_{t}", 1), (f"WR{j}_{t}", 1)], -1) | |
| outputs = ([f"NP{i}" for i in range(A)] + ["NH"] | |
| + [f"NM{j}_{t}" for j in range(S) for t in range(A)]) | |
| return net, state_names(A), outputs | |
| # --------------------------------------------------------------------------- | |
| # serialization | |
| # --------------------------------------------------------------------------- | |
| _DTORD_FAMILY = {"F64": 0, "F32": 1, "F16": 2, "I64": 3, "I32": 4, "I16": 5, "I8": 6} | |
| _NP = {"I8": "<i1", "I16": "<i2", "I32": "<i4"} | |
| def _itype(maxabs: int) -> str: | |
| return "I16" if maxabs <= 32767 else "I32" | |
| def encode(A: int, bias, fanin, pred, nxt, machine: str, lay, io) -> bytes: | |
| """The serialization of an ordered netlist from its four vectors.""" | |
| n, v = len(nxt), len(bias) | |
| t = {"net.bias": ("I8", np.asarray(bias, dtype=np.int64)), | |
| "net.fanin": (_itype(int(np.max(fanin))), np.asarray(fanin, dtype=np.int64)), | |
| "net.next": (_itype(int(np.max(np.abs(nxt)))), np.asarray(nxt, dtype=np.int64)), | |
| "net.pred": (_itype(int(np.max(np.abs(pred)))), np.asarray(pred, dtype=np.int64))} | |
| assert int(t["net.bias"][1].min()) >= -128 and int(t["net.bias"][1].max()) <= 127 | |
| for k, val in (("addr_bits", A), ("data_bits", A), ("memory_words", 1 << A), | |
| ("pc_width", A), ("state_bits", n), ("units", v), | |
| ("version", VERSION)): | |
| t[f"manifest.{k}"] = ("I32", np.asarray([val], dtype=np.int64)) | |
| order = sorted(t, key=lambda k: (_DTORD_FAMILY[t[k][0]], k)) | |
| header = {"__metadata__": { | |
| "machine": machine, "weight_quantization": "ternary", | |
| "signal_registry": json.dumps(lay, separators=(",", ":"), sort_keys=True), | |
| "io": json.dumps(io, separators=(",", ":"), sort_keys=True)}} | |
| blobs, off = [], 0 | |
| for k in order: | |
| dt, arr = t[k] | |
| b = arr.astype(_NP[dt]).tobytes() | |
| header[k] = {"dtype": dt, "shape": [int(arr.size)], "data_offsets": [off, off + len(b)]} | |
| blobs.append(b) | |
| off += len(b) | |
| s = json.dumps(header, separators=(",", ":"), ensure_ascii=False).encode("utf-8") | |
| s += b" " * ((-(8 + len(s))) % 8) | |
| return struct.pack("<Q", len(s)) + s + b"".join(blobs) | |
| def family_sigma_of_net(net: Net, inputs: List[str], outputs: List[str], A: int, | |
| machine: str = None, lay=None, io=None) -> bytes: | |
| """The serialization of a Net in the order its units were built.""" | |
| slot = {s: i for i, s in enumerate(inputs)} | |
| for gname in net.gates: | |
| slot[gname] = len(slot) | |
| bias, fanin, pred = [], [], [] | |
| for gname, (ins, b) in net.gates.items(): | |
| bias.append(b) | |
| fanin.append(len(ins)) | |
| for s, w in ins: | |
| assert w in (-1, 1) | |
| pred.append(w * (slot[s] + 1)) | |
| nxt = [slot[s] for s in outputs] | |
| return encode(A, bias, fanin, pred, nxt, machine or f"subleq{A}", | |
| lay or layout(A), io or io_cells(A)) | |
| def sigma(A: int) -> bytes: | |
| """The serialization of H_A, written block by block with numpy.""" | |
| S = 1 << A | |
| n = A + 1 + A * S | |
| bias, fanin, pred = [], [], [] | |
| base = {} | |
| count = [n] | |
| def block(name: str, size: int): | |
| base[name] = count[0] | |
| count[0] += size | |
| J = np.arange(S, dtype=np.int64) | |
| popc = np.array([bin(j).count("1") for j in range(S)], dtype=np.int64) | |
| bits = ((J[:, None] >> (A - 1 - np.arange(A))[None, :]) & 1) # [j, i] MSB first | |
| lsign = np.where(bits == 1, 1, -1) | |
| MU = lambda j, t: A + 1 + A * j + t # state index | |
| HALT, PC = A, np.arange(A, dtype=np.int64) | |
| def emit(b, f, p): | |
| bias.append(np.asarray(b, dtype=np.int64).ravel()) | |
| fanin.append(np.asarray(f, dtype=np.int64).ravel()) | |
| pred.append(np.asarray(p, dtype=np.int64).ravel()) | |
| # B1 | |
| block("P", S) | |
| emit(-popc, np.full(S, A), (lsign * (PC[None, :] + 1)).ravel()) | |
| # B2 | |
| block("HP", A) | |
| emit(np.full(A, -2), np.full(A, 2), np.stack([PC + 1, np.full(A, HALT + 1)], 1)) | |
| # B3 | |
| block("FA", 3 * A * S) | |
| o, t, j = np.meshgrid(np.arange(3), np.arange(A), J, indexing="ij") | |
| emit(np.full(3 * A * S, -2), np.full(3 * A * S, 2), | |
| np.stack([MU((j + o) % S, t) + 1, base["P"] + j + 1], -1)) | |
| # B4 | |
| block("Q", A) | |
| qf, qp = [], [] | |
| for i in range(A): | |
| sel = J[((((J + 3) % S) >> (A - 1 - i)) & 1) == 1] | |
| qf.append(len(sel)) | |
| qp.append(base["P"] + sel + 1) | |
| emit(np.full(A, -1), qf, np.concatenate(qp)) | |
| # B5 | |
| block("F", 3 * A) | |
| o, t, j = np.meshgrid(np.arange(3), np.arange(A), J, indexing="ij") | |
| emit(np.full(3 * A, -1), np.full(3 * A, S), | |
| base["FA"] + (o * A + t) * S + j + 1) | |
| F = lambda o, t: base["F"] + o * A + t | |
| # B6 | |
| block("PA", S) | |
| emit(-popc, np.full(S, A), (lsign * (F(0, np.arange(A))[None, :] + 1)).ravel()) | |
| block("PB", S) | |
| emit(-popc, np.full(S, A), (lsign * (F(1, np.arange(A))[None, :] + 1)).ravel()) | |
| # B7 | |
| block("AC", 1) | |
| emit([-A], [A], F(2, np.arange(A)) + 1) | |
| # B8 | |
| block("W", S) | |
| emit(np.full(S, -1), np.full(S, 2), | |
| np.stack([np.full(S, -(HALT + 1)), base["PB"] + J + 1], -1)) | |
| # B9 | |
| t, j = np.meshgrid(np.arange(A), J, indexing="ij") | |
| block("RX", A * S) | |
| emit(np.full(A * S, -2), np.full(A * S, 2), | |
| np.stack([MU(j, t) + 1, base["PA"] + j + 1], -1)) | |
| block("RY", A * S) | |
| emit(np.full(A * S, -2), np.full(A * S, 2), | |
| np.stack([MU(j, t) + 1, base["PB"] + j + 1], -1)) | |
| # B10 | |
| j, t = np.meshgrid(J, np.arange(A), indexing="ij") | |
| block("KP", S * A) | |
| emit(np.full(S * A, -1), np.full(S * A, 2), | |
| np.stack([MU(j, t) + 1, -(base["W"] + j + 1)], -1)) | |
| # B11 | |
| t, j = np.meshgrid(np.arange(A), J, indexing="ij") | |
| block("X", A) | |
| emit(np.full(A, -1), np.full(A, S), base["RX"] + t * S + j + 1) | |
| block("Y", A) | |
| emit(np.full(A, -1), np.full(A, S), base["RY"] + t * S + j + 1) | |
| X = lambda i: base["X"] + (A - 1 - i) | |
| Y = lambda i: base["Y"] + (A - 1 - i) | |
| # B12 | |
| block("PR", A - 1) | |
| ii = np.arange(A - 1) | |
| emit(np.zeros(A - 1), np.full(A - 1, 2), np.stack([-(X(ii) + 1), Y(ii) + 1], -1)) | |
| PR = lambda i: base["PR"] + i | |
| # B13 | |
| tb, tf, tp = [], [], [] | |
| tidx = {} | |
| block("T", sum(k + 1 for k in range(1, A))) | |
| cur = base["T"] | |
| for k in range(1, A): | |
| tidx[(k, "n")] = cur; cur += 1 | |
| tb.append(-k); tf.append(k); tp += [PR(i) + 1 for i in range(k)] | |
| for jj in range(k): | |
| tidx[(k, jj)] = cur; cur += 1 | |
| tb.append(-(1 + (k - 1 - jj))); tf.append(2 + (k - 1 - jj)) | |
| tp += [Y(jj) + 1, -(X(jj) + 1)] + [PR(i) + 1 for i in range(jj + 1, k)] | |
| emit(tb, tf, tp) | |
| # B14 | |
| block("C", A - 1) | |
| cp = [] | |
| for k in range(1, A): | |
| cp += [tidx[(k, "n")] + 1] + [tidx[(k, jj)] + 1 for jj in range(k)] | |
| emit(np.full(A - 1, -1), [k + 1 for k in range(1, A)], cp) | |
| C = lambda k: base["C"] + k - 1 | |
| # B15 | |
| ii = np.arange(A) | |
| block("AX", A) | |
| emit(np.full(A, -2), np.full(A, 2), np.stack([X(ii) + 1, Y(ii) + 1], -1)) | |
| block("NX", A) | |
| emit(np.zeros(A), np.full(A, 2), np.stack([-(X(ii) + 1), -(Y(ii) + 1)], -1)) | |
| block("XN", A) | |
| emit(np.full(A, -1), np.full(A, 2), | |
| np.stack([base["AX"] + ii + 1, base["NX"] + ii + 1], -1)) | |
| # B16 | |
| block("EQ", 1) | |
| emit([-A], [A], base["XN"] + ii + 1) | |
| # B17 | |
| block("R0", 3) | |
| emit([-1, 1, -2], [2, 2, 2], | |
| [Y(0) + 1, X(0) + 1, -(Y(0) + 1), -(X(0) + 1), | |
| base["R0"] + 1, base["R0"] + 2]) | |
| block("H", 3 * (A - 1)) | |
| hb, hp = [], [] | |
| for k in range(1, A): | |
| for b in (0, -1, -2): | |
| hb.append(b) | |
| hp += [Y(k) + 1, -(X(k) + 1), C(k) + 1] | |
| emit(hb, np.full(3 * (A - 1), 3), hp) | |
| H = lambda m, k: base["H"] + 3 * (k - 1) + (m - 1) | |
| block("R", A - 1) | |
| rp = [] | |
| for k in range(1, A): | |
| rp += [H(1, k) + 1, -(H(2, k) + 1), H(3, k) + 1] | |
| emit(np.full(A - 1, -1), np.full(A - 1, 3), rp) | |
| Rv = lambda k: base["R0"] + 2 if k == 0 else base["R"] + k - 1 | |
| block("LEQ", 1) | |
| top = A - 1 | |
| emit([-1], [4], [H(1, top) + 1, -(H(2, top) + 1), H(3, top) + 1, base["EQ"] + 1]) | |
| LEQ = base["LEQ"] | |
| # B18 | |
| ii = np.arange(A) | |
| block("TA", A) | |
| emit(np.full(A, -2), np.full(A, 3), | |
| np.stack([np.full(A, -(HALT + 1)), np.full(A, LEQ + 1), F(2, ii) + 1], -1)) | |
| block("TB", A) | |
| emit(np.full(A, -1), np.full(A, 3), | |
| np.stack([np.full(A, -(HALT + 1)), np.full(A, -(LEQ + 1)), base["Q"] + ii + 1], -1)) | |
| block("NP", A) | |
| emit(np.full(A, -1), np.full(A, 3), | |
| np.stack([base["HP"] + ii + 1, base["TA"] + ii + 1, base["TB"] + ii + 1], -1)) | |
| # B19 | |
| block("HA", 3) | |
| emit([-2, -1, -1], [3, 3, 3], | |
| [-(HALT + 1), LEQ + 1, base["AC"] + 1, | |
| -(HALT + 1), -(LEQ + 1), base["P"] + (S - 4) + 1, | |
| HALT + 1, base["HA"] + 1, base["HA"] + 2]) | |
| NH = base["HA"] + 2 | |
| # B20 | |
| j, t = np.meshgrid(J, np.arange(A), indexing="ij") | |
| rdata = np.array([Rv(A - 1 - tt) for tt in range(A)], dtype=np.int64) | |
| block("WR", S * A) | |
| emit(np.full(S * A, -2), np.full(S * A, 2), | |
| np.stack([rdata[t] + 1, base["W"] + j + 1], -1)) | |
| block("NM", S * A) | |
| emit(np.full(S * A, -1), np.full(S * A, 2), | |
| np.stack([base["KP"] + j * A + t + 1, base["WR"] + j * A + t + 1], -1)) | |
| nxt = np.concatenate([base["NP"] + np.arange(A), [NH], | |
| base["NM"] + np.arange(S * A)]) | |
| b = np.concatenate(bias); f = np.concatenate(fanin); p = np.concatenate(pred) | |
| assert len(b) == count[0] - n and int(f.sum()) == len(p) | |
| return encode(A, b, f, p, nxt, f"subleq{A}", layout(A), io_cells(A)) | |
| # constant signals, which Net folds into biases and which therefore appear only | |
| # as a next-state assignment | |
| CONST0, CONST1 = -1, -2 | |
| _DTORD = {"F64": 0, "F32": 1, "F16": 2, "I64": 3, "I32": 4, "I16": 5, "I8": 6, | |
| "U8": 7, "BOOL": 8} | |
| _TAG = {torch.float32: "F32", torch.float64: "F64", torch.int64: "I64", | |
| torch.int32: "I32", torch.int16: "I16", torch.int8: "I8"} | |
| def depths(net: Net) -> Dict[str, int]: | |
| """Depth of every unit: one more than the greatest depth of a predecessor.""" | |
| d: Dict[str, int] = {} | |
| order = topological(net) | |
| for g in order: | |
| ins, _ = net.gates[g] | |
| d[g] = 1 + max((d[s] for s, _ in ins if s in net.gates), default=0) | |
| return d | |
| def topological(net: Net) -> List[str]: | |
| """The units in some order consistent with the predecessor relation.""" | |
| gates = net.gates | |
| indeg = {g: 0 for g in gates} | |
| cons: Dict[str, List[str]] = {} | |
| for g, (ins, _) in gates.items(): | |
| for s, _w in ins: | |
| if s in gates: | |
| indeg[g] += 1 | |
| cons.setdefault(s, []).append(g) | |
| order = [g for g, k in indeg.items() if k == 0] | |
| i = 0 | |
| while i < len(order): | |
| for c in cons.get(order[i], []): | |
| if c in gates: | |
| indeg[c] -= 1 | |
| if indeg[c] == 0: | |
| order.append(c) | |
| i += 1 | |
| if len(order) != len(gates): | |
| raise ValueError("the predecessor relation has a cycle") | |
| return order | |
| def unit_order(net: Net, inputs: List[str]) -> Tuple[List[str], Dict[str, int]]: | |
| """The canonical order of the units, and the signal index of every signal.""" | |
| d = depths(net) | |
| by_level: Dict[int, List[str]] = {} | |
| for g in net.gates: | |
| by_level.setdefault(d[g], []).append(g) | |
| slot = {s: i for i, s in enumerate(inputs)} | |
| order: List[str] = [] | |
| for lv in sorted(by_level): | |
| keyed = [] | |
| for g in by_level[lv]: | |
| ins, b = net.gates[g] | |
| keyed.append(((b, tuple(sorted((slot[s], w) for s, w in ins))), g)) | |
| keyed.sort(key=lambda kv: kv[0]) | |
| seen: Dict[tuple, int] = {} | |
| for key, g in keyed: | |
| if key not in seen: | |
| seen[key] = len(inputs) + len(order) | |
| order.append(g) | |
| slot[g] = seen[key] | |
| return order, slot | |
| def tensors_of_net(net: Net, inputs: List[str], outputs: List[str]): | |
| """The five tensors of the serialization, in canonical unit order.""" | |
| order, slot = unit_order(net, inputs) | |
| bias, fanin, pred = [], [], [] | |
| for g in order: | |
| ins, b = net.gates[g] | |
| bias.append(b) | |
| fanin.append(len(ins)) | |
| for k, w in sorted((slot[s], w) for s, w in ins): | |
| assert w in (-1, 1), f"weight {w} is not an edge" | |
| pred.append(w * (k + 1)) | |
| nxt = [] | |
| for s in outputs: | |
| if s == "#0": | |
| nxt.append(CONST0) | |
| elif s == "#1": | |
| nxt.append(CONST1) | |
| else: | |
| nxt.append(slot[s]) | |
| i8, i16 = torch.int8, torch.int16 | |
| return { | |
| "net.bias": torch.tensor(bias, dtype=i8), | |
| "net.fanin": torch.tensor(fanin, dtype=i16), | |
| "net.next": torch.tensor(nxt, dtype=i16), | |
| "net.pred": torch.tensor(pred, dtype=i16), | |
| } | |
| def sigma_of_net(net: Net, inputs: List[str], outputs: List[str], | |
| machine: str, layout: Dict, io: Dict) -> bytes: | |
| """The serialization of an ordered clocked netlist.""" | |
| slot = {s: i for i, s in enumerate(inputs)} | |
| for g in net.gates: | |
| slot[g] = len(slot) | |
| bias, fanin, pred = [], [], [] | |
| for g, (ins, b) in net.gates.items(): | |
| bias.append(b) | |
| fanin.append(len(ins)) | |
| for s, w in ins: | |
| assert w in (-1, 1), f"weight {w} is not an edge" | |
| assert slot[s] < slot[g], "a predecessor is built after its unit" | |
| pred.append(w * (slot[s] + 1)) | |
| nxt = [CONST0 if s == "#0" else CONST1 if s == "#1" else slot[s] for s in outputs] | |
| return encode(layout["pc"][1], bias, fanin, pred, nxt, machine, layout, io) | |
| def parse(data: bytes) -> Tuple[Dict[str, torch.Tensor], Dict[str, str]]: | |
| """Header and tensors of a safetensors encoding, without a library call.""" | |
| n = struct.unpack("<Q", data[:8])[0] | |
| head = json.loads(data[8:8 + n]) | |
| meta = head.pop("__metadata__", {}) | |
| base = 8 + n | |
| out = {} | |
| dt = {"I8": (torch.int8, 1), "I16": (torch.int16, 2), | |
| "I32": (torch.int32, 4), "I64": (torch.int64, 8), | |
| "F32": (torch.float32, 4)} | |
| for name, spec in head.items(): | |
| ty, w = dt[spec["dtype"]] | |
| lo, hi = spec["data_offsets"] | |
| buf = bytearray(data[base + lo:base + hi]) | |
| t = torch.frombuffer(buf, dtype=ty).reshape(spec["shape"]) | |
| out[name] = t.clone() | |
| return out, meta | |
| def net_of_sigma(data: bytes) -> Tuple[Net, List[str], List[str], Dict[str, str]]: | |
| """The netlist a serialization denotes, with canonical signal names.""" | |
| t, meta = parse(data) | |
| n = int(t["manifest.state_bits"][0]) | |
| v = int(t["manifest.units"][0]) | |
| bias = t["net.bias"].tolist() | |
| fanin = t["net.fanin"].tolist() | |
| pred = t["net.pred"].tolist() | |
| nxt = t["net.next"].tolist() | |
| if len(bias) != v or len(fanin) != v: | |
| raise ValueError("unit count disagrees with the manifest") | |
| if len(pred) != sum(fanin): | |
| raise ValueError("predecessor count disagrees with the fan-in vector") | |
| if len(nxt) != n: | |
| raise ValueError("next-state vector disagrees with the manifest") | |
| inputs = [f"s{i}" for i in range(n)] | |
| name = lambda k: inputs[k] if k < n else f"u{k - n}" | |
| net = Net() | |
| p = 0 | |
| for j in range(v): | |
| ins = [] | |
| for _ in range(fanin[j]): | |
| e = pred[p] | |
| if e == 0: | |
| raise ValueError("predecessor entry carries no weight") | |
| k, w = abs(e) - 1, (1 if e > 0 else -1) | |
| if k >= n + v: | |
| raise ValueError("predecessor index out of range") | |
| if k >= n + j: | |
| raise ValueError("predecessor is not earlier in the order") | |
| ins.append((name(k), w)) | |
| p += 1 | |
| net.g(f"u{j}", ins, bias[j]) | |
| outputs = [] | |
| for k in nxt: | |
| if k == CONST0: | |
| outputs.append("#0") | |
| elif k == CONST1: | |
| outputs.append("#1") | |
| elif 0 <= k < n + v: | |
| outputs.append(name(k)) | |
| else: | |
| raise ValueError("next-state index out of range") | |
| return net, inputs, outputs, meta | |
| HALT_PC = 0xFF | |
| def subleq_state_names() -> List[str]: | |
| names = [f"pc{i}" for i in range(8)] + ["halt"] | |
| for j in range(256): | |
| names += [f"m{j}b{i}" for i in range(8)] | |
| return names | |
| def build_subleq_step_net() -> Tuple[Net, List[str], List[str]]: | |
| """The step netlist of the host: the member H_8 of the constant-depth family.""" | |
| return build(8) | |
| REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) | |
| HOST_PATH = os.path.join(REPO, "variants", "neural_subleq8io_netlist.safetensors") | |
| # device cells | |
| C_WR, C_EOT, C_RW, C_RD, R_IN, R_OUT = 0xF9, 0xFA, 0xFB, 0xFC, 0xFD, 0xFE | |
| # program variables | |
| Z, ONE, T1, T2, NEG1, EOK = 0xF0, 0xF1, 0xF2, 0xF3, 0xF4, 0xF5 | |
| # ============================================================================= | |
| # Recipe language (grammar and decoder), reused from constructor8 | |
| # ============================================================================= | |
| def describe(data: bytes) -> bytes: | |
| """Compile bytes into a recipe.""" | |
| tape = bytearray() | |
| i, n = 0, len(data) | |
| while i < n: | |
| j = i | |
| while j < n and data[j] == data[i] and j - i < 127: | |
| j += 1 | |
| if j - i >= 4: | |
| tape.append(256 - (j - i)) | |
| tape.append((256 - data[i]) % 256) | |
| i = j | |
| continue | |
| k = i | |
| while k < n and k - i < 127: | |
| m = k | |
| while m < n and data[m] == data[k] and m - k < 4: | |
| m += 1 | |
| if m - k >= 4: | |
| break | |
| k += 1 | |
| k = max(k, i + 1) | |
| tape.append(k - i) | |
| tape.extend((256 - x) % 256 for x in data[i:k]) | |
| i = k | |
| tape.append(0) | |
| return bytes(tape) | |
| def describe_literal(data: bytes) -> bytes: | |
| """The all-literal encoding used in the proof of the length bound.""" | |
| tape = bytearray() | |
| for i in range(0, len(data), 127): | |
| block = data[i:i + 127] | |
| tape.append(len(block)) | |
| tape.extend((256 - x) % 256 for x in block) | |
| tape.append(0) | |
| return bytes(tape) | |
| def decode(tape: bytes) -> bytes: | |
| """delta: the decoding map on well-formed recipes.""" | |
| out = bytearray() | |
| i = 0 | |
| while True: | |
| t = tape[i] | |
| i += 1 | |
| if t == 0: | |
| return bytes(out) | |
| if t <= 127: | |
| for _ in range(t): | |
| out.append((256 - tape[i]) % 256) | |
| i += 1 | |
| elif t == 128: | |
| raise ValueError("tag 128 is reserved") | |
| else: | |
| out.extend([(256 - tape[i]) % 256] * (256 - t)) | |
| i += 1 | |
| # ============================================================================= | |
| # Framing: ser and inst | |
| # ============================================================================= | |
| def lam(k: int) -> bytes: | |
| """Eight-byte little-endian length field.""" | |
| return struct.pack("<Q", k) | |
| def field(u: bytes) -> bytes: | |
| return lam(len(u)) + u | |
| def ser(sigma: bytes, m: bytes, tau: bytes) -> bytes: | |
| assert len(m) == 256 | |
| return field(sigma) + field(m) + tau | |
| def inst(s: bytes) -> Tuple[bytes, bytes, bytes]: | |
| """Partial inverse of ser: recover (sigma, m, tau).""" | |
| if len(s) < 8: | |
| raise ValueError("truncated") | |
| a = struct.unpack("<Q", s[:8])[0] | |
| if len(s) < 8 + a + 8: | |
| raise ValueError("truncated") | |
| sigma = s[8:8 + a] | |
| b = struct.unpack("<Q", s[8 + a:16 + a])[0] | |
| if b != 256: | |
| raise ValueError("memory image is not 256 bytes") | |
| m = s[16 + a:16 + a + b] | |
| if len(m) != b: # declared 256 bytes, fewer present | |
| raise ValueError("truncated memory image") | |
| tau = s[16 + a + b:] | |
| return sigma, m, tau | |
| # ============================================================================= | |
| # Programs | |
| # ============================================================================= | |
| def _emit(prog: List[Tuple[int, int, int]]) -> Dict[int, int]: | |
| mem = {} | |
| for idx, (a, b, c) in enumerate(prog): | |
| mem[idx * 3] = a | |
| mem[idx * 3 + 1] = b | |
| mem[idx * 3 + 2] = c | |
| return mem | |
| # The decoding loop. Addresses are 3k; every branch target below is written as | |
| # an instruction index and resolved to 3*index by _emit. | |
| # | |
| # k0 T2 <- 0 | |
| # k1 T1 <- 0 | |
| # k2 request read of the tag | |
| # k3 T1 <- -T | |
| # k4 T2 <- T | |
| # k5 branch to k7 when T = 0 or T >= 128 | |
| # k6 goto LITERAL | |
| # k7 branch to END when 256-T <= 0, that is T in {0,128} | |
| # k8 goto REPEAT with T1 = 256-T the run length | |
| # k9 LITERAL: request read of the next byte | |
| # k10 R_OUT <- b | |
| # k11 emit | |
| # k12 T2 <- T2-1; branch to k14 when the count is exhausted | |
| # k13 goto k9 | |
| # k14 goto k0 | |
| # k15 REPEAT: request read of the value byte | |
| # k16 R_OUT <- b | |
| # k17 emit | |
| # k18 T1 <- T1-1; branch to k14 when the count is exhausted | |
| # k19 goto k16 | |
| # k20 END | |
| DECODE_LOOP = [ | |
| (T2, T2, 3 * 1), # k0 | |
| (T1, T1, 3 * 2), # k1 | |
| (NEG1, C_RD, 3 * 3), # k2 | |
| (R_IN, T1, 3 * 4), # k3 | |
| (T1, T2, 3 * 5), # k4 | |
| (Z, T2, 3 * 7), # k5 | |
| (Z, Z, 3 * 9), # k6 | |
| (Z, T1, 3 * 20), # k7 | |
| (Z, Z, 3 * 15), # k8 | |
| (NEG1, C_RD, 3 * 10), # k9 | |
| (R_IN, R_OUT, 3 * 11), # k10 | |
| (NEG1, C_WR, 3 * 12), # k11 | |
| (ONE, T2, 3 * 14), # k12 | |
| (Z, Z, 3 * 9), # k13 | |
| (Z, Z, 3 * 0), # k14 | |
| (NEG1, C_RD, 3 * 16), # k15 | |
| (R_IN, R_OUT, 3 * 17), # k16 | |
| (NEG1, C_WR, 3 * 18), # k17 | |
| (ONE, T1, 3 * 14), # k18 | |
| (Z, Z, 3 * 16), # k19 | |
| ] | |
| # P: the constructor. The end token halts. | |
| P = DECODE_LOOP + [(Z, Z, HALT_PC)] | |
| # P*: the end token enters the rewind phase, then the copy phase. | |
| # k20 R: request rewind | |
| # k21 B0: request read | |
| # k22 B1: EOK <- 1-EOT; halt when the end of tape is reached | |
| # k23 B2: Z <- -b | |
| # k24 B3: R_OUT <- b | |
| # k25 B4: emit | |
| # k26 B5: Z <- 0 and go to B0 | |
| P_STAR = DECODE_LOOP + [ | |
| (NEG1, C_RW, 3 * 21), # k20 | |
| (NEG1, C_RD, 3 * 22), # k21 | |
| (C_EOT, EOK, HALT_PC), # k22 | |
| (R_IN, Z, 3 * 24), # k23 | |
| (Z, R_OUT, 3 * 25), # k24 | |
| (NEG1, C_WR, 3 * 26), # k25 | |
| (Z, Z, 3 * 21), # k26 | |
| ] | |
| # P_e: P with an end-of-tape guard after the tag read, so that the machine | |
| # halts on every tape, recipe or not. Instruction j3 assigns | |
| # M[EOK] <- M[EOK] - M[C_EOT]; a successful tag read has cleared C_EOT and | |
| # leaves EOK at 1, while a read at the end of the tape sets it and the result | |
| # 0 transfers control to the halt cell. | |
| P_TOTAL = [ | |
| (T2, T2, 3 * 1), # j0 | |
| (T1, T1, 3 * 2), # j1 | |
| (NEG1, C_RD, 3 * 3), # j2 read the tag | |
| (C_EOT, EOK, HALT_PC), # j3 halt if that read was past the end of the tape | |
| (R_IN, T1, 3 * 5), # j4 | |
| (T1, T2, 3 * 6), # j5 | |
| (Z, T2, 3 * 8), # j6 | |
| (Z, Z, 3 * 10), # j7 | |
| (Z, T1, 3 * 21), # j8 | |
| (Z, Z, 3 * 16), # j9 | |
| (NEG1, C_RD, 3 * 11), # j10 LITERAL | |
| (R_IN, R_OUT, 3 * 12), # j11 | |
| (NEG1, C_WR, 3 * 13), # j12 | |
| (ONE, T2, 3 * 15), # j13 | |
| (Z, Z, 3 * 10), # j14 | |
| (Z, Z, 3 * 0), # j15 | |
| (NEG1, C_RD, 3 * 17), # j16 REPEAT | |
| (R_IN, R_OUT, 3 * 18), # j17 | |
| (NEG1, C_WR, 3 * 19), # j18 | |
| (ONE, T1, 3 * 15), # j19 | |
| (Z, Z, 3 * 17), # j20 | |
| (Z, Z, HALT_PC), # j21 END | |
| ] | |
| def decode_any(tape: bytes, guard: bool) -> Tuple[bytes, bool]: | |
| """The output of the decoding loop on an arbitrary tape.""" | |
| out = bytearray() | |
| h, rin = 0, 0 | |
| while True: | |
| if h < len(tape): | |
| rin = tape[h] | |
| h += 1 | |
| eot = 0 | |
| else: | |
| eot = 1 | |
| if guard and eot: | |
| return bytes(out), True | |
| t = rin | |
| if not guard and eot: | |
| return bytes(out), t in (0, 128) | |
| if t == 0 or t == 128: | |
| return bytes(out), True | |
| reps = 1 if t > 128 else t | |
| count = (256 - t) if t > 128 else 1 | |
| for _ in range(reps): | |
| if h < len(tape): | |
| rin = tape[h] | |
| h += 1 | |
| out.extend([(256 - rin) % 256] * count) | |
| def memory_image(prog: List[Tuple[int, int, int]]) -> List[int]: | |
| """The 256-byte initial memory image holding a program and its constants.""" | |
| mem = [0] * 256 | |
| for addr, val in _emit(prog).items(): | |
| mem[addr] = val & 0xFF | |
| mem[Z] = 0 | |
| mem[ONE] = 1 | |
| mem[T1] = 0 | |
| mem[T2] = 0 | |
| mem[NEG1] = 0xFF | |
| mem[EOK] = 1 | |
| return mem | |
| M_P = memory_image(P) | |
| M_STAR = memory_image(P_STAR) | |
| def program_a(prog: List[Tuple[int, int, int]], A: int) -> List[Tuple[int, int, int]]: | |
| """A program moved to 2^A words: the cells from 0xF0 up name the top sixteen words.""" | |
| top = (1 << A) - 256 | |
| return [tuple(x + top if x >= 0xF0 else x for x in ins) for ins in prog] | |
| def memory_image_a(prog: List[Tuple[int, int, int]], A: int) -> List[int]: | |
| """The initial memory of 2^A words holding a program moved there and its constants.""" | |
| S = 1 << A | |
| top = S - 256 | |
| mem = [0] * S | |
| for addr, val in _emit(program_a(prog, A)).items(): | |
| mem[addr] = val | |
| for cell, val in ((Z, 0), (ONE, 1), (T1, 0), (T2, 0), (NEG1, S - 1), (EOK, 1)): | |
| mem[cell + top] = val | |
| return mem | |
| # ============================================================================= | |
| # Device | |
| # ============================================================================= | |
| class Tape: | |
| """Environment state (tau, h, omega) with the operations of the definition.""" | |
| def __init__(self, tau: bytes): | |
| self.tau = tau | |
| self.h = 0 | |
| self.out = bytearray() | |
| def apply(self, mem: List[int]) -> None: | |
| """One device application to the post-instruction memory image.""" | |
| if mem[C_WR] == 1: | |
| self.out.append(mem[R_OUT]) | |
| mem[R_OUT] = 0 | |
| if mem[C_RD] == 1: | |
| if self.h < len(self.tau): | |
| mem[R_IN] = self.tau[self.h] | |
| mem[C_EOT] = 0 | |
| self.h += 1 | |
| else: | |
| mem[C_EOT] = 1 | |
| if mem[C_RW] == 1: | |
| self.h = 0 | |
| mem[C_WR] = 0 | |
| mem[C_RD] = 0 | |
| mem[C_RW] = 0 | |
| # ============================================================================= | |
| # Evaluator 1: integer reference | |
| # ============================================================================= | |
| def run_reference(mem0: List[int], tau: bytes, max_steps: int = 1 << 34, | |
| expect: Optional[bytes] = None) -> Tuple[bytes, int]: | |
| mem = list(mem0) | |
| dev = Tape(tau) | |
| pc = 0 | |
| steps = 0 | |
| while pc != HALT_PC and steps < max_steps: | |
| A = mem[pc] | |
| B = mem[(pc + 1) & 0xFF] | |
| C = mem[(pc + 2) & 0xFF] | |
| r = (mem[B] - mem[A]) & 0xFF | |
| mem[B] = r | |
| pc = C if (r == 0 or r >= 0x80) else (pc + 3) & 0xFF | |
| n_before = len(dev.out) | |
| dev.apply(mem) | |
| if expect is not None and len(dev.out) > n_before: | |
| k = len(dev.out) - 1 | |
| if k >= len(expect) or dev.out[k] != expect[k]: | |
| raise AssertionError(f"stream diverged at byte {k}") | |
| steps += 1 | |
| return bytes(dev.out), steps | |
| def run_reference_a(mem0: List[int], tau: bytes, A: int, max_steps: int = 1 << 40) -> Tuple[bytes, int]: | |
| """SUBLEQ_A with the device on its top words, a read extending the byte by its sign.""" | |
| S, half = 1 << A, 1 << (A - 1) | |
| io = io_cells(A) | |
| wr, eot, rw, rd, rin, rout, halt = (io[k] for k in ('c_wr', 'c_eot', 'c_rw', 'c_rd', 'r_in', 'r_out', | |
| 'halt_pc')) | |
| mem, out = list(mem0), bytearray() | |
| pc = h = steps = 0 | |
| while pc != halt and steps < max_steps: | |
| a, b, c = mem[pc], mem[(pc + 1) % S], mem[(pc + 2) % S] | |
| r = (mem[b] - mem[a]) % S | |
| mem[b] = r | |
| pc = c if r == 0 or r >= half else (pc + 3) % S | |
| if mem[wr] == 1: | |
| out.append(mem[rout] & 0xFF) | |
| mem[rout] = 0 | |
| if mem[rd] == 1: | |
| if h < len(tau): | |
| v = tau[h] | |
| mem[rin] = v if v < 128 else v + S - 256 | |
| mem[eot] = 0 | |
| h += 1 | |
| else: | |
| mem[eot] = 1 | |
| if mem[rw] == 1: | |
| h = 0 | |
| mem[wr] = mem[rd] = mem[rw] = 0 | |
| steps += 1 | |
| return bytes(out), steps | |
| # ============================================================================= | |
| # The host netlist and its canonical serialization | |
| # ============================================================================= | |
| STATE_LAYOUT = {"pc": [0, 8], "halt": [8, 1], "mem": [9, 256, 8]} | |
| IO_CELLS = {"c_wr": C_WR, "c_eot": C_EOT, "c_rw": C_RW, "c_rd": C_RD, | |
| "r_in": R_IN, "r_out": R_OUT, "halt_pc": HALT_PC} | |
| STATE_BITS = 8 + 1 + 2048 | |
| def host_netlist(): | |
| """The clocked netlist of the host, assembled from its source description.""" | |
| return build_subleq_step_net() | |
| def sigma_host() -> bytes: | |
| """sigma(N_host): the canonical serialization of that netlist.""" | |
| net, inputs, outputs = host_netlist() | |
| return sigma_of_net(net, inputs, outputs, "subleq8", STATE_LAYOUT, | |
| IO_CELLS) | |
| def read_host() -> bytes: | |
| """The distributed serialization of the host.""" | |
| return open(HOST_PATH, "rb").read() | |
| def sha(b: bytes) -> str: | |
| return hashlib.sha256(b).hexdigest() | |
| def tau_star(sigma: bytes, m: List[int]) -> bytes: | |
| return describe(field(sigma) + field(bytes(m))) | |
| # ============================================================================= | |
| # Evaluators of the step map, each built from sigma alone | |
| # ============================================================================= | |
| class _Transducer: | |
| """State marshalling shared by the threshold evaluators.""" | |
| N = STATE_BITS | |
| _DEVCELLS = (C_WR, C_EOT, C_RW, C_RD, R_IN, R_OUT) | |
| def _vec(self, pc: int, mem: List[int]): | |
| torch = self.torch | |
| v = torch.zeros(self.N) | |
| for k in range(8): | |
| v[k] = (pc >> (7 - k)) & 1 | |
| for j in range(256): | |
| for k in range(8): | |
| v[9 + j * 8 + k] = (mem[j] >> (7 - k)) & 1 | |
| return v | |
| def _byte(vc, j: int) -> int: | |
| x = 0 | |
| for k in range(8): | |
| x = (x << 1) | int(vc[9 + j * 8 + k]) | |
| return x | |
| def _set_byte(self, v, j: int, val: int) -> None: | |
| for k in range(8): | |
| v[0, 9 + j * 8 + k] = float((val >> (7 - k)) & 1) | |
| def capture(self): | |
| """Capture one application of the map as a CUDA graph.""" | |
| torch = self.torch | |
| assert self.device.startswith("cuda") | |
| self.gin = torch.zeros(1, self.N, device=self.device) | |
| # every buffer the body writes must stay alive for the life of the | |
| # graph, or the allocator will hand its memory to something else and | |
| # the replay will overwrite that instead | |
| self.gbuf = self._buffers() | |
| for _ in range(5): | |
| self._body(self.gin) | |
| torch.cuda.synchronize() | |
| self.graph = torch.cuda.CUDAGraph() | |
| with torch.cuda.graph(self.graph): | |
| self.gout = self._body(self.gin) | |
| torch.cuda.synchronize() | |
| gen = torch.Generator(device=self.device).manual_seed(3) | |
| probe = (torch.rand(1, self.N, generator=gen, | |
| device=self.device) < 0.5).float() | |
| self.gin.copy_(probe) | |
| self.graph.replay() | |
| torch.cuda.synchronize() | |
| assert bool((self.gout == self._eager(probe)).all()), \ | |
| "the captured graph differs from the eager step" | |
| return self | |
| def _replay(self, v): | |
| # the state lives in gin, which is ordinary memory; gout belongs to the | |
| # graph's private pool and is only ever read as a whole | |
| if v.data_ptr() != self.gin.data_ptr(): | |
| self.gin.copy_(v) | |
| self.graph.replay() | |
| self.gin.copy_(self.gout) | |
| return self.gin | |
| def run(self, mem0: List[int], tau: bytes, max_steps: int, | |
| expect: Optional[bytes] = None, progress: int = 0, | |
| margin: bool = False) -> Tuple[bytes, int]: | |
| """Iterate the map with the device applied after each step.""" | |
| import time | |
| torch = self.torch | |
| v = self._vec(0, mem0).unsqueeze(0).to(self.device) | |
| dev = Tape(tau) | |
| cells = list(self._DEVCELLS) | |
| cell_t = torch.tensor([9 + j * 8 + k for j in cells for k in range(8)], | |
| device=self.device) | |
| pow2_t = torch.tensor([1 << (7 - k) for k in range(8)], | |
| device=self.device, dtype=torch.float32) | |
| shift_t = torch.tensor([7 - k for k in range(8)]) | |
| shadow = [0] * 256 | |
| self.min_margin = float("inf") | |
| n = 0 | |
| t0 = time.perf_counter() | |
| while n < max_steps: | |
| if margin: | |
| self._accumulate_margin(v) | |
| v = self.step(v) | |
| n += 1 | |
| vals = torch.cat([v[0, 8:9], | |
| (v[0, cell_t].reshape(len(cells), 8) * pow2_t) | |
| .sum(-1)]).to("cpu").to(torch.int64).tolist() | |
| for c, j in enumerate(cells): | |
| shadow[j] = vals[1 + c] | |
| before = len(dev.out) | |
| dev.apply(shadow) | |
| new = torch.tensor([shadow[j] for j in cells], dtype=torch.int64) | |
| v[0, cell_t] = (((new.unsqueeze(-1) >> shift_t) & 1) | |
| .reshape(-1).float().to(self.device)) | |
| if expect is not None and len(dev.out) > before: | |
| k = len(dev.out) - 1 | |
| if k >= len(expect) or dev.out[k] != expect[k]: | |
| raise AssertionError(f"stream diverged at byte {k}") | |
| if vals[0] >= 1: | |
| break | |
| if progress and n % progress == 0: | |
| rate = n / (time.perf_counter() - t0) | |
| print(f" {self.tag} {n:,} steps, {len(dev.out):,} bytes " | |
| f"({rate:,.0f} steps/s)", flush=True) | |
| self.seconds = time.perf_counter() - t0 | |
| return bytes(dev.out), n | |
| class NetEvaluator(_Transducer): | |
| """The netlist of sigma, evaluated unit by unit.""" | |
| tag = "net" | |
| def __init__(self, sigma: bytes, device: str = "cpu", graph: bool = False): | |
| import time | |
| import torch | |
| from reflect import Leveled | |
| t0 = time.perf_counter() | |
| self.torch = torch | |
| self.device = device | |
| net, inputs, outputs, meta = net_of_sigma(sigma) | |
| self.net, self.inputs, self.outputs, self.meta = net, inputs, outputs, meta | |
| assert len(inputs) == self.N, "state width disagrees with the layout" | |
| self.lev = Leveled(net, inputs, outputs, device=device) | |
| self.info = {"levels": len(self.lev.plan), "units": len(net.gates), | |
| "entries": sum(len(i) for i, _ in net.gates.values())} | |
| self.graph = None | |
| if graph: | |
| self.capture() | |
| self.build_seconds = time.perf_counter() - t0 | |
| def _buffers(self): | |
| return self.torch.zeros(self.lev.n_sig, 1, device=self.device) | |
| def _body(self, inp): | |
| V = self.gbuf | |
| lev = self.lev | |
| V.zero_() | |
| V[1] = 1.0 | |
| V[lev.in_slots] = inp.T | |
| for idx, w, b, out in lev.plan: | |
| g = V[idx] | |
| V[out] = ((g * w[:, :, None]).sum(1) + b[:, None] >= 0).float() | |
| return V[lev.out_slots].T | |
| def _eager(self, v): | |
| return self.lev.step(v) | |
| def step(self, v): | |
| if self.graph is not None: | |
| return self._replay(v) | |
| return self.lev.step(v) | |
| class LevEvaluator(_Transducer): | |
| """Lev(N) for the netlist N of sigma.""" | |
| tag = "lev" | |
| def __init__(self, sigma: bytes, device: str = "cpu", dense: bool = True, | |
| graph: bool = False): | |
| import time | |
| import torch | |
| from machines import compile_net | |
| t0 = time.perf_counter() | |
| self.torch = torch | |
| self.device = device | |
| self.dense = dense | |
| net, inputs, outputs, meta = net_of_sigma(sigma) | |
| self.net, self.inputs, self.outputs, self.meta = net, inputs, outputs, meta | |
| assert len(inputs) == self.N, "state width disagrees with the layout" | |
| layers, info = compile_net(net, inputs, outputs) | |
| for W, _ in layers: | |
| assert set(torch.unique(W).tolist()) <= {-1, 0, 1} | |
| self.info = dict(info) | |
| self.info["size"] = sum(int(W.shape[0]) for W, _ in layers) | |
| self.info["nonzero"] = sum(int((W != 0).sum()) for W, _ in layers) | |
| if dense: | |
| self.W = [W.to(device=device, dtype=torch.float32) | |
| for W, _ in layers] | |
| self.B = [b.to(device=device, dtype=torch.float32) | |
| for _, b in layers] | |
| else: | |
| pad = device.startswith("cuda") | |
| self.plan = [self._sparsify(W, b, device, pad) for W, b in layers] | |
| self.widths = [int(W.shape[0]) for W, _ in layers] | |
| self.graph = None | |
| if graph: | |
| self.capture() | |
| self.build_seconds = time.perf_counter() - t0 | |
| def _sparsify(W, b, device, pad): | |
| """One layer as the nonzero entries of its rows.""" | |
| import torch | |
| nz = (W != 0) | |
| counts = nz.sum(1) | |
| n = int(W.shape[0]) | |
| sizes = [int(counts.max())] if pad else sorted(set(counts.tolist())) | |
| groups = [] | |
| for k in sizes: | |
| rows = (torch.arange(n) if pad | |
| else torch.nonzero(counts == k, as_tuple=False).flatten()) | |
| idx = torch.zeros(len(rows), max(k, 1), dtype=torch.long) | |
| w = torch.zeros(len(rows), max(k, 1)) | |
| for r, row in enumerate(rows.tolist()): | |
| cols = torch.nonzero(nz[row], as_tuple=False).flatten() | |
| idx[r, :len(cols)] = cols | |
| w[r, :len(cols)] = W[row, cols] | |
| groups.append((rows.to(device), idx.to(device), | |
| w.to(device=device, dtype=torch.float32), | |
| b[rows].to(device=device, dtype=torch.float32))) | |
| return groups | |
| def _buffers(self): | |
| import torch | |
| return [torch.zeros(1, n, device=self.device) for n in self.widths] | |
| def _sparse_step(self, v, buf): | |
| x = v | |
| for groups, y in zip(self.plan, buf): | |
| for rows, idx, w, b in groups: | |
| y[:, rows] = ((x[:, idx] * w).sum(-1) + b >= 0).float() | |
| x = y | |
| return x | |
| def _body(self, inp): | |
| return self._sparse_step(inp, self.gbuf) | |
| def _eager(self, v): | |
| return self._sparse_step(v, [self.torch.zeros(v.shape[0], n, | |
| device=self.device) | |
| for n in self.widths]) | |
| def step(self, v): | |
| if self.dense: | |
| for W, b in zip(self.W, self.B): | |
| v = ((v @ W.T + b) >= 0).float() | |
| return v | |
| if self.graph is not None: | |
| return self._replay(v) | |
| return self._eager(v) | |
| def _accumulate_margin(self, v): | |
| y = v | |
| for W, b in zip(self.W, self.B): | |
| pre = y @ W.T + b | |
| m = float((pre + 0.5).abs().min()) | |
| if m < self.min_margin: | |
| self.min_margin = m | |
| y = (pre >= 0).float() | |
| def step_noisy(self, v, sigma: float, gen): | |
| """One step with additive Gaussian read noise per pre-activation and the comparator at -1/2.""" | |
| torch = self.torch | |
| if self.dense: | |
| for W, b in zip(self.W, self.B): | |
| pre = v @ W.T + b | |
| pre = pre + torch.randn(pre.shape, generator=gen, | |
| device=pre.device) * sigma | |
| v = (pre >= -0.5).float() | |
| return v | |
| x = v | |
| for groups, n in zip(self.plan, self.widths): | |
| y = torch.empty(x.shape[0], n, device=x.device) | |
| for rows, idx, w, b in groups: | |
| pre = (x[:, idx] * w).sum(-1) + b | |
| pre = pre + torch.randn(pre.shape, generator=gen, | |
| device=pre.device) * sigma | |
| y[:, rows] = (pre >= -0.5).float() | |
| x = y | |
| return x | |
| REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) # repo root; this module lives in src/ | |
| MODEL_PATH = os.path.join(REPO, "variants", "neural_subleq8io.safetensors") | |
| IN_DATA, IN_STROBE, OUT = 0xFD, 0xFC, 0xFE | |
| # ============================================================================= | |
| # Build the host file (family per-gate style, .inputs wiring, ternary strict) | |
| # ============================================================================= | |
| def build_host(save: bool = True) -> str: | |
| import build as B | |
| from build import quantize_tensors | |
| tensors: Dict[str, torch.Tensor] = {} | |
| B.add_decoder(tensors, 8, 256) | |
| B.add_memory_read_mux(tensors, 256) | |
| B.add_memory_write_cells(tensors, 256) | |
| B.add_subleq_core(tensors) | |
| tensors["manifest.data_bits"] = torch.tensor([8.0]) | |
| tensors["manifest.addr_bits"] = torch.tensor([8.0]) | |
| tensors["manifest.memory_bytes"] = torch.tensor([256.0]) | |
| tensors["manifest.pc_width"] = torch.tensor([8.0]) | |
| tensors["manifest.version"] = torch.tensor([4.0]) | |
| tensors, reg, stats = B.build_inputs(tensors) | |
| tensors, counts, _, _ = quantize_tensors(tensors, ternary=False) | |
| for k, t in tensors.items(): | |
| if k.endswith(".weight") and not k.startswith("manifest."): | |
| tf = t.float() if t.dtype.is_floating_point else t.to(torch.float32) | |
| assert (tf.abs() <= 1).all(), f"non-ternary weight {k}" | |
| meta = { | |
| "machine": "subleq8io", | |
| "weight_quantization": "ternary", | |
| "signal_registry": reg.to_metadata(), | |
| "io": json.dumps({"in_data": IN_DATA, "in_strobe": IN_STROBE, | |
| "out": OUT, "halt_pc": HALT_PC}), | |
| } | |
| if save: | |
| save_file(tensors, MODEL_PATH, metadata=meta) | |
| sz = os.path.getsize(MODEL_PATH) | |
| print(f" built {MODEL_PATH}: {stats['added']} wired gates, " | |
| f"{len(tensors)} tensors, {sz:,} bytes") | |
| return MODEL_PATH | |
| # ============================================================================= | |
| # Recipe codec (the construction-description language) | |
| # ============================================================================= | |
| def c8_describe(data: bytes) -> bytes: | |
| """Compile raw bytes into a construction recipe for the constructor.""" | |
| tape = bytearray() | |
| i, n = 0, len(data) | |
| while i < n: | |
| j = i | |
| while j < n and data[j] == data[i] and j - i < 127: | |
| j += 1 | |
| if j - i >= 4: | |
| tape.append(256 - (j - i)) # repeat tag 129..252 | |
| tape.append((256 - data[i]) % 256) # value, negated | |
| i = j | |
| continue | |
| k = i | |
| while k < n and k - i < 127: # literal until a run starts | |
| m = k | |
| while m < n and data[m] == data[k] and m - k < 4: | |
| m += 1 | |
| if m - k >= 4: | |
| break | |
| k += 1 | |
| k = max(k, i + 1) | |
| tape.append(k - i) # literal tag 1..127 | |
| tape.extend((256 - x) % 256 for x in data[i:k]) | |
| i = k | |
| tape.append(0) # END | |
| return bytes(tape) | |
| def c8_decode(tape: bytes) -> bytes: | |
| """Pure-Python reference semantics of the recipe language.""" | |
| out = bytearray() | |
| i = 0 | |
| while True: | |
| t = tape[i] | |
| i += 1 | |
| if t == 0: | |
| return bytes(out) | |
| if t <= 127: | |
| for _ in range(t): | |
| out.append((256 - tape[i]) % 256) | |
| i += 1 | |
| else: | |
| out.extend([(256 - tape[i]) % 256] * (256 - t)) | |
| i += 1 | |
| # ============================================================================= | |
| # The constructor program (21 SUBLEQ instructions + 4 variable cells) | |
| # ============================================================================= | |
| CONSTRUCTOR = [ | |
| # main loop: read tag T; T1 = -T (== run length for repeat tags), T2 = T | |
| (T2, T2, 0x03), # 00 T2 = 0 | |
| (T1, T1, 0x06), # 03 T1 = 0 | |
| (IN_DATA, T1, 0x09), # 06 T1 = -T | |
| (Z, IN_STROBE, 0x0C), # 09 consume tag byte | |
| (T1, T2, 0x0F), # 0C T2 = T | |
| (Z, T2, 0x15), # 0F if T == 0 or T >= 128 -> 15, else literal | |
| (Z, Z, 0x24), # 12 goto LITERAL | |
| (Z, T1, 0x1B), # 15 T1 = -T: leq only when T == 0 -> END | |
| (Z, Z, 0x30), # 18 goto REPEAT (T1 holds 256-T = run length) | |
| (Z, Z, HALT_PC), # 1B END: halt | |
| (Z, Z, 0x21), # 1E (pad) | |
| (Z, Z, 0x24), # 21 (pad) | |
| (IN_DATA, OUT, 0x27), # 24 LITERAL: emit 0 - tape_byte | |
| (Z, IN_STROBE, 0x2A), # 27 consume it | |
| (ONE, T2, 0x00), # 2A T2 -= 1; done -> main loop | |
| (Z, Z, 0x24), # 2D next literal byte | |
| (IN_DATA, OUT, 0x33), # 30 REPEAT: emit 0 - value (no strobe) | |
| (ONE, T1, 0x39), # 33 T1 -= 1; done -> 39 | |
| (Z, Z, 0x30), # 36 emit again | |
| (Z, IN_STROBE, 0x3C), # 39 consume the value byte | |
| (Z, Z, 0x00), # 3C goto main loop | |
| ] | |
| def constructor_image() -> List[int]: | |
| mem = [0] * 256 | |
| for idx, (a, b, c) in enumerate(CONSTRUCTOR): | |
| mem[idx * 3] = a | |
| mem[idx * 3 + 1] = b | |
| mem[idx * 3 + 2] = c | |
| mem[ONE] = 1 | |
| return mem | |
| # ============================================================================= | |
| # Backend 1: fast pure-integer reference with devices | |
| # ============================================================================= | |
| def ref_construct(tape: bytes, max_steps: int = 1 << 34, | |
| ) -> Tuple[bytes, int]: | |
| mem = constructor_image() | |
| mem[IN_DATA] = tape[0] if tape else 0 | |
| out = bytearray() | |
| head = 0 | |
| pc = 0 | |
| steps = 0 | |
| tn = len(tape) | |
| while pc != HALT_PC and steps < max_steps: | |
| A = mem[pc] | |
| Bc = mem[(pc + 1) & 0xFF] | |
| C = mem[(pc + 2) & 0xFF] | |
| r = (mem[Bc] - mem[A]) & 0xFF | |
| mem[Bc] = r | |
| pc = C if (r == 0 or r >= 0x80) else (pc + 3) & 0xFF | |
| if Bc == OUT: | |
| out.append(r) | |
| mem[OUT] = 0 | |
| elif Bc == IN_STROBE: | |
| head += 1 | |
| mem[IN_STROBE] = 0 | |
| mem[IN_DATA] = tape[head] if head < tn else 0 | |
| steps += 1 | |
| return bytes(out), steps | |
| # ============================================================================= | |
| # Backend 2: the gate-graph processor (shipped netlist) with devices | |
| # ============================================================================= | |
| class GateHost: | |
| def __init__(self, path: str = MODEL_PATH): | |
| from machines import SubleqThresholdCPU, load_tensors | |
| from eval import load_metadata | |
| T = load_tensors(path) | |
| reg = load_metadata(path)["signal_registry"] | |
| self.cpu = SubleqThresholdCPU(T, reg) | |
| def run(self, tape: bytes, max_steps: int) -> Tuple[bytes, int, dict]: | |
| s = {"pc": 0, "mem": constructor_image(), "halted": False} | |
| s["mem"][IN_DATA] = tape[0] if tape else 0 | |
| out = bytearray() | |
| head = 0 | |
| n = 0 | |
| while not s["halted"] and n < max_steps: | |
| Bc = s["mem"][(s["pc"] + 1) & 0xFF] | |
| s = self.cpu.step(s) | |
| if Bc == OUT: | |
| out.append(s["mem"][OUT]) | |
| s["mem"][OUT] = 0 | |
| elif Bc == IN_STROBE: | |
| head += 1 | |
| s["mem"][IN_STROBE] = 0 | |
| s["mem"][IN_DATA] = tape[head] if head < len(tape) else 0 | |
| n += 1 | |
| return bytes(out), n, s | |
| # ============================================================================= | |
| # Backend 3: the host compiled to recurrent ternary matrices (machine 1 tech) | |
| # ============================================================================= | |
| class MatrixHost: | |
| """The subleq8io processor as a recurrent ternary threshold network, the device applied between steps.""" | |
| N = 8 + 1 + 2048 | |
| def __init__(self, device: str = "cpu"): | |
| t0 = time.perf_counter() | |
| net, inputs, outputs = build_subleq_step_net() | |
| layers, info = compile_net(net, inputs, outputs) | |
| for W, Bt in layers: | |
| vals = set(torch.unique(W).tolist()) | |
| assert vals <= {-1, 0, 1} | |
| self.W = [W.to(device=device, dtype=torch.float32) for W, _ in layers] | |
| self.B = [Bt.to(device=device, dtype=torch.float32) for _, Bt in layers] | |
| self.device = device | |
| self.info = info | |
| print(f" host compiled to {info['layers']} ternary matrices " | |
| f"(max width {info['max_width']}, " | |
| f"{info['total_weights']:,} weights) in " | |
| f"{time.perf_counter() - t0:.1f}s on {device}") | |
| def _vec(self, pc: int, mem: List[int], halt: bool = False) -> torch.Tensor: | |
| v = torch.zeros(self.N) | |
| for k in range(8): | |
| v[k] = (pc >> (7 - k)) & 1 | |
| v[8] = 1.0 if halt else 0.0 | |
| for j in range(256): | |
| for k in range(8): | |
| v[9 + j * 8 + k] = (mem[j] >> (7 - k)) & 1 | |
| return v | |
| def step(self, v: torch.Tensor) -> torch.Tensor: | |
| for W, b in zip(self.W, self.B): | |
| v = ((v @ W.T + b) >= 0).float() | |
| return v | |
| def _byte(vcpu: torch.Tensor, j: int) -> int: | |
| x = 0 | |
| for k in range(8): | |
| x = (x << 1) | int(vcpu[9 + j * 8 + k]) | |
| return x | |
| def run(self, tape: bytes, max_steps: int, expect: Optional[bytes] = None, | |
| report_every: int = 100000) -> Tuple[bytes, int]: | |
| mem = constructor_image() | |
| mem[IN_DATA] = tape[0] if tape else 0 | |
| v = self._vec(0, mem).unsqueeze(0).to(self.device) | |
| out = bytearray() | |
| head = 0 | |
| n = 0 | |
| idx_pch = torch.arange(0, 9, device=self.device) | |
| t0 = time.perf_counter() | |
| while n < max_steps: | |
| front = v[0, idx_pch].tolist() | |
| if front[8] >= 0.5: | |
| break | |
| pcv = 0 | |
| for k in range(8): | |
| pcv = (pcv << 1) | int(front[k]) | |
| bidx = (pcv + 1) & 0xFF | |
| bbits = v[0, 9 + bidx * 8: 9 + bidx * 8 + 8].tolist() | |
| Bc = 0 | |
| for k in range(8): | |
| Bc = (Bc << 1) | int(bbits[k]) | |
| v = self.step(v) | |
| n += 1 | |
| if Bc == OUT: | |
| ob = v[0, 9 + OUT * 8: 9 + OUT * 8 + 8].tolist() | |
| val = 0 | |
| for k in range(8): | |
| val = (val << 1) | int(ob[k]) | |
| out.append(val) | |
| if expect is not None and out[-1] != expect[len(out) - 1]: | |
| raise AssertionError( | |
| f"stream diverged at byte {len(out) - 1}: " | |
| f"got {out[-1]:#04x} want {expect[len(out) - 1]:#04x}") | |
| v[0, 9 + OUT * 8: 9 + OUT * 8 + 8] = 0.0 | |
| elif Bc == IN_STROBE: | |
| head += 1 | |
| v[0, 9 + IN_STROBE * 8: 9 + IN_STROBE * 8 + 8] = 0.0 | |
| nxt = tape[head] if head < len(tape) else 0 | |
| for k in range(8): | |
| v[0, 9 + IN_DATA * 8 + k] = float((nxt >> (7 - k)) & 1) | |
| if report_every and n % report_every == 0: | |
| rate = n / (time.perf_counter() - t0) | |
| print(f" ... {n:,} recurrences, {len(out):,} bytes emitted " | |
| f"({rate:,.0f} instr/s)") | |
| return bytes(out), n | |
| def exhaustive_datapath(self) -> bool: | |
| """All 65,536 (x, y) operand pairs through the matrices in one batch.""" | |
| xs = torch.arange(65536) % 256 | |
| ys = torch.arange(65536) // 256 | |
| V = torch.zeros(65536, self.N) | |
| mem = [0] * 256 | |
| mem[0], mem[1], mem[2] = 0x90, 0x91, 0x30 | |
| v0 = self._vec(0, mem) | |
| V[:] = v0 | |
| for k in range(8): | |
| V[:, 9 + 0x90 * 8 + k] = ((xs >> (7 - k)) & 1).float() | |
| V[:, 9 + 0x91 * 8 + k] = ((ys >> (7 - k)) & 1).float() | |
| V = V.to(self.device) | |
| for W, b in zip(self.W, self.B): | |
| V = ((V @ W.T + b) >= 0).float() | |
| V = V.cpu() | |
| r = torch.zeros(65536, dtype=torch.long) | |
| for k in range(8): | |
| r = (r << 1) | V[:, 9 + 0x91 * 8 + k].long() | |
| pcv = torch.zeros(65536, dtype=torch.long) | |
| for k in range(8): | |
| pcv = (pcv << 1) | V[:, k].long() | |
| exp_r = (ys - xs) & 0xFF | |
| exp_leq = (exp_r == 0) | (exp_r >= 0x80) | |
| exp_pc = torch.where(exp_leq, torch.full_like(pcv, 0x30), | |
| torch.full_like(pcv, 3)) | |
| ok = bool((r == exp_r).all()) and bool((pcv == exp_pc).all()) | |
| print(f" {'OK ' if ok else 'FAIL'} matrix datapath exhaustive: " | |
| f"65,536/65,536 subtract results and branch decisions exact") | |
| return ok | |
| # ============================================================================= | |
| # Verification | |
| # ============================================================================= | |
| def c8_sha(b: bytes) -> str: | |
| return hashlib.sha256(b).hexdigest()[:16] | |
| def family_files() -> List[str]: | |
| files = [os.path.join(REPO, "neural_computer.safetensors")] | |
| vdir = os.path.join(REPO, "variants") | |
| files += sorted(os.path.join(vdir, f) for f in os.listdir(vdir) | |
| if f.endswith(".safetensors")) | |
| return files | |
| def verify(device: str) -> bool: | |
| ok = True | |
| print("\n[1/5] Host machine soundness (shipped netlist, exhaustive datapath)") | |
| from eval import NetlistEvaluator, load_metadata | |
| from machines import load_tensors | |
| T = load_tensors(MODEL_PATH) | |
| reg = load_metadata(MODEL_PATH)["signal_registry"] | |
| ne = NetlistEvaluator(T, reg, "subleq") | |
| a_v = torch.arange(65536) % 256 | |
| b_v = torch.arange(65536) // 256 | |
| ext = {} | |
| for k in range(8): | |
| ext[f"$a[{k}]"] = ((a_v >> k) & 1).float() | |
| ext[f"$b[{k}]"] = ((b_v >> k) & 1).float() | |
| ext[f"$pc[{k}]"] = torch.zeros(65536) | |
| ext[f"$c[{k}]"] = torch.zeros(65536) | |
| o = ne.run(ext) | |
| rr = sum(o[f"subleq.sub.fa{k}.ha2.sum.layer2"][:, 0].long() << k for k in range(8)) | |
| lq = o["subleq.leq"][:, 0].long() | |
| exp_r = (b_v - a_v) & 0xFF | |
| exp_l = ((exp_r == 0) | (exp_r >= 128)).long() | |
| good = bool((rr == exp_r).all()) and bool((lq == exp_l).all()) | |
| print(f" {'OK ' if good else 'FAIL'} gate netlist exhaustive: 65,536/65,536 exact") | |
| ok &= good | |
| print("\n[2/5] Recipe codec round-trip over the whole family") | |
| total = 0 | |
| for p in family_files(): | |
| data = open(p, "rb").read() | |
| tape = c8_describe(data) | |
| good = c8_decode(tape) == data | |
| ok &= good | |
| total += len(data) | |
| print(f" {'OK ' if good else 'FAIL'} {os.path.basename(p):<44} " | |
| f"{len(data):>10,} B -> tape {len(tape):>10,} B sha {c8_sha(data)}") | |
| print(f" ({total / 1e6:.0f} MB described and decoded back, byte-identical)") | |
| print("\n[3/5] Constructor program on the integer reference " | |
| "(output stream == target file, byte for byte)") | |
| targets = [MODEL_PATH, | |
| os.path.join(REPO, "variants", "neural_subleq8.safetensors"), | |
| os.path.join(REPO, "variants", "neural_matrix8.safetensors")] | |
| for p in targets: | |
| data = open(p, "rb").read() | |
| tape = c8_describe(data) | |
| t0 = time.perf_counter() | |
| out, steps = ref_construct(tape) | |
| dt = time.perf_counter() - t0 | |
| good = out == data | |
| ok &= good | |
| tag = "SELF-REPRODUCTION" if p == MODEL_PATH else "universal build" | |
| print(f" {'OK ' if good else 'FAIL'} {os.path.basename(p):<40} " | |
| f"{len(out):>10,} B in {steps:>12,} instructions ({dt:.1f}s) " | |
| f"[{tag}] sha {c8_sha(out)}") | |
| print("\n[4/5] Matrix backend (machine-1 compile of the host)") | |
| mh = MatrixHost(device=device) | |
| ok &= mh.exhaustive_datapath() | |
| print("\n[5/5] Backend lockstep on a shared prefix " | |
| "(gate graph vs matrices vs reference, first 600 cycles)") | |
| data = open(MODEL_PATH, "rb").read() | |
| tape = c8_describe(data) | |
| gh = GateHost() | |
| s = {"pc": 0, "mem": constructor_image(), "halted": False} | |
| s["mem"][IN_DATA] = tape[0] | |
| mem2 = constructor_image() | |
| mem2[IN_DATA] = tape[0] | |
| v = mh._vec(0, mem2).unsqueeze(0).to(device) | |
| headg = headm = 0 | |
| outg, outm = bytearray(), bytearray() | |
| good = True | |
| for n in range(600): | |
| Bg = s["mem"][(s["pc"] + 1) & 0xFF] | |
| s = gh.cpu.step(s) | |
| if Bg == OUT: | |
| outg.append(s["mem"][OUT]) | |
| s["mem"][OUT] = 0 | |
| elif Bg == IN_STROBE: | |
| headg += 1 | |
| s["mem"][IN_STROBE] = 0 | |
| s["mem"][IN_DATA] = tape[headg] if headg < len(tape) else 0 | |
| vc = v[0].cpu() | |
| pcm = 0 | |
| for k in range(8): | |
| pcm = (pcm << 1) | int(vc[k]) | |
| Bm = MatrixHost._byte(vc, (pcm + 1) & 0xFF) | |
| v = mh.step(v) | |
| if Bm == OUT: | |
| outm.append(MatrixHost._byte(v[0].cpu(), OUT)) | |
| v[0, 9 + OUT * 8: 9 + OUT * 8 + 8] = 0.0 | |
| elif Bm == IN_STROBE: | |
| headm += 1 | |
| v[0, 9 + IN_STROBE * 8: 9 + IN_STROBE * 8 + 8] = 0.0 | |
| nxt = tape[headm] if headm < len(tape) else 0 | |
| for k in range(8): | |
| v[0, 9 + IN_DATA * 8 + k] = float((nxt >> (7 - k)) & 1) | |
| vc = v[0].cpu() | |
| pcm = 0 | |
| for k in range(8): | |
| pcm = (pcm << 1) | int(vc[k]) | |
| memm = [MatrixHost._byte(vc, j) for j in range(256)] | |
| if pcm != s["pc"] or memm != s["mem"] or outg != outm: | |
| print(f" FAIL lockstep diverged at cycle {n}") | |
| good = False | |
| break | |
| print(f" {'OK ' if good else 'FAIL'} gate graph and matrix form agree " | |
| f"cycle-for-cycle for 600 cycles ({len(outg)} bytes emitted identically)") | |
| ok &= good | |
| print("\nCONSTRUCTOR VERIFICATION:", "PASS" if ok else "FAIL") | |
| return ok | |
| def self_reproduce(device: str, gen2: bool = True) -> bool: | |
| """Full self-reproduction at threshold fidelity on the matrix backend.""" | |
| data = open(MODEL_PATH, "rb").read() | |
| tape = c8_describe(data) | |
| print(f"\nSelf-reproduction on the threshold matrices: target " | |
| f"{len(data):,} bytes (sha {c8_sha(data)}), tape {len(tape):,} bytes") | |
| ref_out, ref_steps = ref_construct(tape) | |
| assert ref_out == data, "reference construction failed" | |
| print(f" reference: {ref_steps:,} instructions") | |
| mh = MatrixHost(device=device) | |
| t0 = time.perf_counter() | |
| out, n = mh.run(tape, max_steps=ref_steps + 64, expect=data) | |
| dt = time.perf_counter() - t0 | |
| ok = out == data and n == ref_steps | |
| print(f" GEN 1: emitted {len(out):,} bytes in {n:,} recurrences " | |
| f"({dt / 60:.1f} min, {n / dt:,.0f} instr/s) sha {c8_sha(out)}") | |
| print(f" {'OK ' if ok else 'FAIL'} output stream == host safetensors file, " | |
| f"byte for byte; instruction count matches reference exactly") | |
| if ok and gen2: | |
| off_path = os.path.join(REPO, "variants", "_offspring_subleq8io.safetensors") | |
| with open(off_path, "wb") as f: | |
| f.write(out) | |
| gh = GateHost(off_path) # boot the offspring's own gates | |
| probe_tape = c8_describe(out[:2048]) | |
| want = out[:2048] | |
| got, steps, _ = gh.run(probe_tape, max_steps=len(probe_tape) * 8 + 64) | |
| good2 = got == want | |
| print(f" GEN 2: offspring file booted as a machine (gate graph); " | |
| f"constructed the first {len(want):,} bytes of itself in " | |
| f"{steps:,} instructions: {'exact' if good2 else 'MISMATCH'}") | |
| os.remove(off_path) | |
| ok &= good2 | |
| print("SELF-REPRODUCTION:", "PASS" if ok else "FAIL") | |
| return ok | |
| def main_constructor8() -> int: | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("cmd", choices=["build", "verify", "self", "all"]) | |
| ap.add_argument("--device", default="cuda" if torch.cuda.is_available() else "cpu") | |
| args = ap.parse_args() | |
| rc = 0 | |
| if args.cmd in ("build", "all"): | |
| build_host(save=True) | |
| if args.cmd in ("verify", "all"): | |
| rc |= 0 if verify(args.device) else 1 | |
| if args.cmd in ("self", "all"): | |
| rc |= 0 if self_reproduce(args.device) else 1 | |
| return rc | |
| def cmd_family(): | |
| import hashlib | |
| import time | |
| for A in [int(a) for a in sys.argv[1:]] or [3, 4, 5, 8]: | |
| t0 = time.perf_counter() | |
| s = sigma(A) | |
| t1 = time.perf_counter() | |
| line = f"A={A}: sigma {len(s):,} bytes in {t1 - t0:.2f}s, sha256 {hashlib.sha256(s).hexdigest()[:16]}" | |
| if A <= 12: | |
| net, inp, out = build(A) | |
| same = family_sigma_of_net(net, inp, out, A) == s | |
| depth = {} | |
| for gname, (ins, _) in net.gates.items(): | |
| depth[gname] = 1 + max((depth.get(x, 0) for x, _ in ins), default=0) | |
| line += (f"; Net {len(net.gates):,} units, depth {max(depth.values())}, " | |
| f"edges {sum(len(i) for i, _ in net.gates.values()):,}; " | |
| f"equal to the Net's serialization: {same}") | |
| N2, i2, o2, _ = net_of_sigma(s) | |
| line += f"; parses to {len(N2.gates):,} units" | |
| print(line, flush=True) | |
| COMMANDS = {'family': cmd_family, 'constructor8': main_constructor8} | |
| 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]()) | |