minclose.py
Checks: LATENCY: Trees of units property; rule; observation.
Cited on: LATENCY.
To run it: save minclose.py, with chains.py, dags.py, graphs.py, manystream.py, poles.py, redundant.py, sublattice.py beside it; standard library only.
"""
minclose.py -- is a graph's feasible set a lattice in any signing?
QUESTION. sublattice.py's A arm (the arrivals of the units read twice
or more, F_T projected, every other unit existentially optimized), at
the shared classes of EXACTLY 6 units (dags.classes(6) filtered by
length), at (6, 2, 4), (35, 8, 27) and (4, 2, 2), each T from brute's
least to naive. Is A min-closed at every triple? To 5 units it is, on
177 testable triples; carried to 6, a TRANSPLANT.
KILL (frozen before code): "KILL A not min-closed at 6 units: N" over
the three cells and every kind; fires if N > 0, read only beside "A
testable at 6 units: M", UNREADABLE if M < 10.
CONTROLS. C1 the enumerator: alive_all = alive_product and least =
dags.brute over the first 12 6-unit classes at (6, 2, 4). C2 a real
failure caught: the F arm's (every unit's arrival) first non-min-closed
triple of sublattice's Y5, read through sublattice.read_pair, reads
not min-closed (the sweep's A-arm counter has no planted break). C3
sublattice's synthetic detector.
HAND STEP. The max witness h, c = (z, h), r = (h, c), (z, r) twice at
(6, 2, 4), T = 8: r's best leftover at arrival 7 behind h at 2 and 3,
and at arrival 6 behind h at 2 and 3.
FINDINGS (copied from the run below). C1 0 off, C2 (F closure
(False, True)), C3 green. HAND: r's best leftover at arrival 7 is
88577/455625 = 0.19441 behind h at 2 (delays 2, 2, 3) and 76133/455625
= 0.16710 behind h at 3 (3, 2, 2): the raise takes r's third delay;
a reader (z, r) at delay 1 is alive behind (2, 7), 13157/34171875, and
dead behind (3, 7). At arrival 6 the earlier head again leaves r more,
0.16645 against 0.00257.
X6, 995 classes at 3 cells: heads only 1,010 triples (A testable 92),
shared non-head 349 (36), above shared 2,936 (235): A testable at 6
units: 363. D not max-closed at 1,263 of 1,263 testable above-shared
triples, not min-closed at 98; A not max-closed at 48, not min-closed
at 5, every failure above shared.
KILL A not min-closed at 6 units: 5. FIRES (predicted 0). All 5 at
(6, 2, 4), and the product filter reads each set non-min-closed too.
The first, the MIN WITNESS: h0, h1, u2 = (z, h0), u3 = (h1, u2),
(z, u3), (h1, u3), T = 8, shared (h1, u3), arrivals (2, 7) and (3, 5)
alive, (2, 5) not. Every failure has 6 units, so the 5-unit sweeps
could not show one. So neither a greatest nor a least setting in
arrivals exists at (6, 2, 4).
SIGNED (frozen after the kill fired). Each shared unit's coordinate
times (-1)^depth, depth its longest chain of shared units below. S1
(all signs +, the sweep's failures again) is dropped: it compared one
set with itself. S2, frozen as "signed,
both witnesses closed", FAILED on its first run: the min witness
signed, (h1, -u3), is min-closed but not max-closed, (4, -5) and
(5, -6) alive, (5, -5) not: the precedence a_u3 >= a_h1 + 1, a
difference constraint unsigned, is broken by the flip. The frozen hand
step had read one pair as the set. S2 now checks that finding (the max
witness a lattice signed, the min witness not max-closed), and the
sweep runs behind it:
KILL signed A not a lattice: 2414 (testable 2650). FIRES, at every
depth-1 and depth-2 kind above shared (Y4 86 of 86, Y5 976 + 127, X6
845 + 380). Signed D not a lattice at 495: 0 at depth 1 to 5 units
(56 testable), 6 of 161 at X6's depth 1, every testable depth-2 triple.
Added after reading it, ANY SIGNING: per triple, does SOME sign vector
make the set a lattice? Testable with none: Y4 0 of 104 (D), 0 of 12
(A); Y5 126 of 1,200 (D), 0 of 165 (A); X6 342 of 1,391 (D), 8 of 363
(A). So no choice of signs, even per latency, makes F_T a lattice in
delays (from 5 units on) or arrivals (at 6 units), summed over the
cells each population runs. S3: the min witness's (a_h1, a_u3) at
T = 8 is {(2, 6), (2, 7), (3, 5), (3, 6), (3, 7), (4, 5), (4, 6),
(4, 7), (5, 6)}, neither min- nor max-closed, and in (a_h1, a_h1 -
a_u3) a lattice. Of its box [2, 5] x [5, 7] it misses (2, 5), (5, 5)
and (5, 7), read as "h1 late or u3 late", the precedence a_u3 >=
a_h1 + 1 and "h1 early or u3 early": within the box, each a clause in
two thresholds.
RUN RECORD. One run: 6/6 checks, 53.2 s, peak 10.9 MB; S1 dropped,
5/5 checks, 45.1 s.
"""
import itertools
import sys
import time
from collections import Counter
import dags
import graphs
import sublattice as sl
CHECKS = []
def check(name, ok, detail=""):
CHECKS.append(bool(ok))
tail = f" ({detail})" if detail else ""
print(f" {'PASS' if ok else 'FAIL'} {name}{tail}")
def cell_named(nm):
return next(c for c in graphs.cells() if graphs.name(c) == nm)
CELLS = ("(6, 2, 4)", "(35, 8, 27)", "(4, 2, 2)")
MAXW = ((None, None), (None, 0), (0, 1), (None, 2), (None, 2))
MINW = ((None, None), (None, None), (None, 0), (1, 2), (None, 3), (1, 3))
def six():
return [u for u in dags.classes(6) if len(u) == 6]
def controls():
print("C1..C3 the instrument")
cell = cell_named("(6, 2, 4)")
off = 0
t0 = time.time()
for units in six()[:12]:
d = dags.Dag(cell, units)
nv = d.naive()
a = sl.alive_all(d, nv)
b = sl.alive_product(d, nv)
off += sorted(a) != sorted(b)
off += min(r[0] for r in a) != dags.brute(d)[0]
check("C1 enumerator = product filter, least = brute, 12 six-unit "
"classes at (6, 2, 4)", off == 0,
f"{off} off, {time.time() - t0:.1f} s")
w = ((None, None), (None, None), (None, 0), (None, 0), (1, 2))
d, least, out = sl.read_pair(cell, w)
row = [o for o in out if o[0] == 6]
check("C2 the F arm's Y5 min failure at T = 6 read not min-closed",
len(row) == 1 and not row[0][4][0] and row[0][4][3],
f"F closure {row[0][4][:2] if row else None}")
syn = {"cross": {(1, 2), (2, 1)},
"square": {(1, 2), (2, 1), (2, 2), (1, 1)}}
got = {k: sl.closed(v)[:2] for k, v in syn.items()}
check("C3 detector: cross neither, square both",
got == {"cross": (False, False), "square": (True, True)},
f"{got}")
def hand():
print("\nHAND the max witness at (6, 2, 4), T = 8")
d = dags.Dag(cell_named("(6, 2, 4)"), MAXW)
best = {}
for Ls in itertools.product(*[range(0, x + 4) for x in d.Ls[:3]]):
left, arr = [None] * 3, [0] * 3
ok = True
for j in range(3):
arr[j] = d.arrival(j, Ls[j], arr)
r = d.leftover(j, Ls[j], left)
if r is None:
ok = False
break
left[j] = r
if not ok:
continue
key = (arr[0], arr[2])
if key not in best or left[2] > best[key][0]:
best[key] = (left[2], Ls)
for key in ((2, 6), (3, 6), (2, 7), (3, 7)):
v = best.get(key)
print(f" (a_h, a_r) = {key}: r's best leftover "
f"{None if v is None else (str(v[0]), float(v[0]), v[1])}")
for a_r in (6, 7):
L = 8 - a_r
for key in ((2, a_r), (3, a_r)):
v = best.get(key)
if v is None:
continue
left = [None] * 5
left[2] = v[0]
r = d.leftover(3, L, left)
print(f" reader (z, r) at delay {L} behind {key}: "
f"{None if r is None else (str(r), float(r))}")
def sweep(cells_):
t0 = time.time()
cls = six()
n = Counter()
first = {}
for nm in cells_:
cell = cell_named(nm)
for units in cls:
d, least, out = sl.read_pair(cell, units)
for T, k, cd, ca, cf in out:
n[(k, "triples")] += 1
for arm, c in (("D", cd), ("A", ca)):
n[(k, f"{arm} testable")] += c[3]
for i, s in ((0, "min"), (1, "max")):
if not c[i]:
n[(k, f"{arm} not {s}-closed")] += 1
key = (k, arm, s)
if key not in first:
first[key] = (nm, units, T,
sl.shared_units(d))
if (arm, s) == ("A", "min"):
n[("A min fails at", nm)] += 1
n[("A min re-read off", nm)] += reread(
d, T)
print(f"\nX6 {len(cls)} classes of 6 units at {len(cells_)} cells, "
f"{time.time() - t0:.1f} s")
for k in ("heads only", "shared non-head", "shared above shared"):
row = {s: n[(k, s)] for s in (
"triples", "D testable", "A testable", "D not max-closed",
"D not min-closed", "A not max-closed", "A not min-closed")}
print(f" {k:20s} {row}")
print(" A min failures per cell, and those the product filter "
"reads otherwise: " + str({x[1]: (n[x], n[("A min re-read off",
x[1])])
for x in n if x[0] == "A min fails at"}))
for key in sorted(first):
print(f" first {key}: {first[key]}")
return n
def reread(d, T):
"""1 if alive_product's shared-arrival set at T is min-closed
(disagreeing with alive_all's failure), else 0."""
S = sl.shared_units(d)
P = {tuple(r[2][j] for j in S) for r in sl.alive_product(d, T)}
return int(sl.closed(P)[0])
def depth(d, S):
"""d_j: the longest chain of shared units strictly below j."""
memo = {}
def f(j):
if j not in memo:
memo[j] = max([f(h) + 1 for h in S
if h != j and sl.above(d, j, h)], default=0)
return memo[j]
return {j: f(j) for j in S}
def projections(cell, units):
"""[(T, kind, max depth, parity signs, D set, A set)], the sets the
shared units' delays and arrivals at latency <= T."""
d = dags.Dag(cell, units)
nv = d.naive()
rows = sl.alive_all(d, nv)
least = min(r[0] for r in rows)
S = sl.shared_units(d)
dep = depth(d, S)
sg = tuple((-1) ** dep[j] for j in S)
k = sl.kind(d, S)
out = []
for T in range(least, nv + 1):
PD = {tuple(r[1][j] for j in S) for r in rows if r[0] <= T}
PA = {tuple(r[2][j] for j in S) for r in rows if r[0] <= T}
out.append((T, k, max(dep.values()), sg, PD, PA))
return out
def signed(P, sg):
return {tuple(g * x for g, x in zip(sg, p)) for p in P}
def lattice(P):
c = sl.closed(P)
return c[0] and c[1]
def signed_controls():
print("\nS2..S3 the signing")
cell = cell_named("(6, 2, 4)")
got = []
for w in (MAXW, MINW):
T, k, dp, sg, PD, PA = [o for o in projections(cell, w)
if o[0] == 8][0]
got += [sl.closed(PA)[:2], sl.closed(signed(PA, sg))[:2]]
check("S2 the two witnesses at T = 8, unsigned not closed; signed, "
"the max witness a lattice, the min witness not max-closed",
got[1] == (True, True) and got[3] == (True, False)
and got[0] != (True, True) and got[2] != (True, True), f"{got}")
T, k, dp, sg, PD, PA = [o for o in projections(cell, MINW)
if o[0] == 8][0]
print(f" the min witness's (a_h1, a_u3) at T = 8: {sorted(PA)}")
Q = {(a, a - u) for a, u in PA}
check("S3 the min witness at T = 8 a lattice in (a_h1, a_h1 - a_u3)",
lattice(Q) and not lattice(PA), f"{sorted(Q)}")
def signings(pops):
"""The parity signing's lattice counts per (kind, depth capped at 2),
testable read on the signed set; and, added after the parity kill
was read, per triple testable unsigned, whether SOME sign vector
(first coordinate +) makes the set a lattice."""
tot = Counter()
first = {}
for tag, cls, cells_ in pops:
t0 = time.time()
n = Counter()
for cell in cells_:
for units in cls:
for T, k, dp, sg, PD, PA in projections(cell, units):
key = (k, min(dp, 2))
n[key + ("triples",)] += 1
sgs = [(1,) + t for t in itertools.product(
(1, -1), repeat=len(sg) - 1)]
for arm, P in (("D", PD), ("A", PA)):
c = sl.closed(signed(P, sg))
n[key + (f"{arm} testable",)] += c[3]
bad = not (c[0] and c[1])
n[key + (f"{arm} not lattice",)] += bad
if bad and (arm, dp) not in first:
first[(arm, dp)] = (tag, graphs.name(cell),
units, T)
if not sl.closed(P)[3]:
continue
if not any(lattice(signed(P, s)) for s in sgs):
tot[(tag, arm, "no signing")] += 1
first.setdefault((tag, arm, "no signing"), (
graphs.name(cell), units, T))
tot[(tag, arm, "testable")] += 1
print(f"\n{tag} {len(cls)} classes at {len(cells_)} cells, "
f"{time.time() - t0:.1f} s (kind, depth >= 2 capped)")
for key in sorted({x[:2] for x in n}):
print(f" {str(key):28s} " + str(
{s_: n[key + (s_,)] for s_ in (
"triples", "D testable", "D not lattice",
"A testable", "A not lattice")}))
for arm in ("D", "A"):
tot[("signed", arm, "testable")] += sum(
v for x, v in n.items() if x[2] == f"{arm} testable")
tot[("signed", arm, "not lattice")] += sum(
v for x, v in n.items() if x[2] == f"{arm} not lattice")
for key in sorted(first, key=str):
print(f" first {key}: {first[key]}")
print(f"\nsigned D not a lattice: {tot[('signed', 'D', 'not lattice')]}"
f" (testable {tot[('signed', 'D', 'testable')]})")
print(f"KILL signed A not a lattice: "
f"{tot[('signed', 'A', 'not lattice')]} "
f"(testable {tot[('signed', 'A', 'testable')]})")
print("\nANY SIGNING testable triples no sign vector makes a lattice")
for tag, cls, cells_ in pops:
print(f" {tag}: " + ", ".join(
f"{arm} {tot[(tag, arm, 'no signing')]} of "
f"{tot[(tag, arm, 'testable')]}" for arm in ("D", "A")))
return tot
def main():
t0 = time.time()
controls()
if not all(CHECKS):
print("\nCONTROL FAILED: nothing below is read.")
return 1
hand()
n = sweep(CELLS)
kinds = ("heads only", "shared non-head", "shared above shared")
M = sum(n[(k, "A testable")] for k in kinds)
N = sum(n[(k, "A not min-closed")] for k in kinds)
print(f"\nA testable at 6 units: {M}")
print(f"KILL A not min-closed at 6 units: {N}"
f"{' (UNREADABLE: M < 10)' if M < 10 else ''}")
signed_controls()
if not all(CHECKS):
print("\nCONTROL FAILED: the signings are not read.")
return 1
cs = list(graphs.cells())
gain = lambda c, k: (lambda p: p[1] < p[0])(graphs.price(
c, graphs.SHAPES[k]))
sq = [c for c in cs if graphs.name(c) in ("(6, 2, 4)", "(35, 8, 27)")
or gain(c, "S") or gain(c, "Q")]
signings((("Y4", dags.classes(4), cs),
("Y5", [u for u in dags.classes(5) if len(u) == 5], sq),
("X6", six(), [cell_named(nm) for nm in CELLS])))
print(f"\n{sum(CHECKS)}/{len(CHECKS)} checks, {time.time() - t0:.1f} s")
return 0
if __name__ == "__main__":
sys.exit(main())