pairwise.py
Checks: LATENCY: Trees of units property; rule; theorem; observation.
Cited on: LATENCY.
To run it: save pairwise.py, with chains.py, dags.py, drops.py, graphs.py, manystream.py, median.py, minclose.py, poles.py, raises.py, redundant.py, sublattice.py beside it; needs numpy.
"""
pairwise.py -- is a raise-free part its two-unit clauses beside the latency?
QUESTION. Below naive, RF is an FD set (median.sets) cut to p_u <= L*_u
at every unit (drops.cut). Against it: C2 = {p in the box
0 <= p_u <= L*_u : latency(p) <= T, every pair (p_u, p_v) in RF's
projection on (u, v)}, the latency d.arrival's (delay plus the latest
input's arrival) in numpy, read against it by K4; and CR, the
same with the pairs cut to (u, w), w reading u, every single unit's
range kept.
BY HAND (frozen before code). A median-closed set in a product of
chains is the set of points whose every two-unit projection lies in
its own (median is a majority operation; Baker and Pixley 1975, the
source not opened): x with every pair in R; induct on |I|, some r in
R agreeing with x on I; for |I| >= 3 take three indices of I and, by
induction, r_k in R agreeing with x on I less the k-th; at each index of
I two of the
three agree with x, so their median, in R, agrees on I. So C2 = RF on
every median-closed RF set and kill A can fire only where RF is not
closed (8 sets over X6, Y4, Y5, drops.py); X7, the seven-unit shared
classes, widens that arm at the four cells the 8 sit at. CR is not
settled by the theorem.
PREDICTIONS. Kill A misses (moderately); X7 adds not-closed sets
(moderately), M3 >= 10 (weakly). Kill B fires (weakly); if it does, B
on ancestor pairs misses (moderately). Aliveness cuts past single
units in many sets (strongly).
KILL A (frozen). "KILL raise-free parts not cut by two-unit clauses
beside the latency: N (sets M; not median-closed M3)" over X6 at the
441 cells, Y4, Y5 (median.populations) and X7 at the four cells;
fires if N > 0, UNREADABLE if M3 < 10.
KILL B (frozen). "KILL raise-free parts not cut by reader-pair clauses
beside the latency: N (sets M, cut past single units Mc)", same
populations; fires if N > 0, UNREADABLE if Mc < 10. Beside: the sets
with two or more drop sets; B on ancestor pairs; RF points the latency
or a pair excludes (an instrument bug, 0).
CONTROLS. K1 by theorem: C2 = RF on every closed RF set of X6 at the
four cells. K2 those cells rebuild drops.py's 8 not-closed sets, by
cell. K3 a plant: in closed testable RF sets there, the points on one
value triple whose deletion keeps every pair projection are deleted,
C2 read off the planted set, every one caught, >= 5. K4 the floors
point reads naive.
FINDINGS (copied from the runs below). K1 0 closed sets off (RF sets
1,742, 8 not closed), K2 8, K3 10 of 10, K4 0 off.
X6 at the 441 cells: RF sets 33,398 (two or more drop sets 5,492, not
closed 8, cut past single units 1,162), C2 off 0, reader pairs off 0.
Y4: 420 (42, 0, 8), off 0, 0. Y5: 2,100 (395, 0, 86), off 0, 0. X7 at
the four cells: 20,146 (5,041, 178, 1,276), C2 off 4 (closed 0),
reader pairs off 6, ancestor pairs off 4. Bugs 0.
KILL raise-free parts not cut by two-unit clauses beside the latency:
4 (sets 56,064; not median-closed 186). FIRES; the moderate miss
predicted wrong.
KILL raise-free parts not cut by reader-pair clauses beside the
latency: 6 (sets 56,064, cut past single units 2,532). FIRES, as
weakly predicted; the ancestor arm fires too (4), against the moderate
prediction. Cut past single units 2,532 of 56,064.
Added after the kills were read (a rebuild of the fires, printed by no function here): all
4 of kill A sit at (4, 2, 2), three graphs (three classes), each with
one extra point
whose only dead unit is the sink at delay 0 and which three drops
make: restoring any one of them to its floor gives a point of RF. So
the clause is three drop literals, [L_1 = 2] or [L_2 = 2] or
[L_6 = 1] on h, 1 = (z, h), 2 = (z, h), 3 = (1, 2), 4 = (1, 3),
5 = (1, 4), 6 = (2, 5) at T = 7, naive 8 (1 and 2 siblings, off one
path), and the same with 5 = (2, 4), 6 = (1, 5); on h, 1 = (z, h), 2 = 3
= (h, 1), 4 = (2, 3), 5 = (1, 4),
6 = (1, 5) at T = 6 and 7 it is on 1, 4, 6, one path. The other 2 of
kill B sit at (8, 2, 6): h, 1 = (z, h), 2 = (h, 1), 3 = (1, 2),
4 = (h, 3), 5 = (h, 4), 6 = (z, 5) at T = 11 and 12, naive 13, the
conflict not (4 and 6 dropped), 5 between them at its floor; at
T = 11 that RF set is median-closed, its pair off the reader edges.
RUN RECORD. Controls 4/4, 18.0 s, peak commit 21.6 MB. Rehearsed at 4
cells per population and X7 thinned, 19.8 s. Four shards, 223.6 to
234.5 s, peak commit 31.5 MB. The rebuild of the fires, 40.6 s at
(4, 2, 2), 143.6 s at the other three. Rerun from code/: controls
4/4, 18.2 s; four shards, 222.4 to 231.3 s, every count above
reproduced.
"""
import itertools
import json
import os
import sys
import time
from collections import Counter
os.environ.setdefault("OPENBLAS_NUM_THREADS", "1")
import numpy as np # noqa: E402
import dags # noqa: E402
import graphs # noqa: E402
import median # noqa: E402
import sublattice as sl # noqa: E402
from drops import cut # noqa: E402
from minclose import six # noqa: E402
FOUR = ("(6, 2, 4)", "(11, 3, 8)", "(8, 2, 6)", "(4, 2, 2)")
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 seven():
out = set()
for u in dags.shapes(7):
if dags.shared(u):
out.add(dags.canonical(u))
return sorted((c for c in out if len(c) == 7), key=lambda c: [tuple(
-1 if x is None else x for x in p) for p in c])
def named(names):
by = {graphs.name(c): c for c in graphs.cells()}
return [by[n] for n in names]
class Box:
"""The raise-free box of one graph, each point's latency, and the
unit pairs by relation."""
def __init__(self, d):
self.d = d
f = d.Ls
self.B = np.array(list(itertools.product(*[range(x + 1)
for x in f])),
dtype=np.int64)
arr = np.zeros_like(self.B)
for j, (a, b) in enumerate(d.units):
m = np.zeros(len(self.B), dtype=np.int64)
for x in (a, b):
if x is not None:
m = np.maximum(m, arr[:, x])
arr[:, j] = self.B[:, j] + m
self.lat = arr.max(1)
self.w = np.cumprod([1] + [x + 1 for x in f[::-1]])[::-1][1:]
below = [set() for _ in range(d.n)]
for j, u in enumerate(d.units):
for x in u:
if x is not None:
below[j] |= below[x] | {x}
self.below = below
self.allp = list(itertools.combinations(range(d.n), 2))
self.reader = [(u, w) for u, w in self.allp
if w in d.readers[u] or u in d.readers[w]]
self.anc = [(u, w) for u, w in self.allp
if u in below[w] or w in below[u]]
def idx(self, P):
X = np.array(sorted(P), dtype=np.int64)
return X @ self.w
def mask(self, P, T, pairs):
X = np.array(sorted(P), dtype=np.int64)
f = self.d.Ls
m = self.lat <= T
for u in range(self.d.n):
ok = np.zeros(f[u] + 1, dtype=bool)
ok[X[:, u]] = True
m &= ok[self.B[:, u]]
for u, v in pairs:
ok = np.zeros((f[u] + 1, f[v] + 1), dtype=bool)
ok[X[:, u], X[:, v]] = True
m &= ok[self.B[:, u], self.B[:, v]]
return m
def rel(self, u, w):
if w in self.d.readers[u] or u in self.d.readers[w]:
return "reader"
if u in self.below[w] or w in self.below[u]:
return "ancestor"
return "apart"
def missed(self, P, x):
"""Pairs whose RF projection the point x misses, by relation."""
out = []
for u, v in self.allp:
if not any(p[u] == x[u] and p[v] == x[v] for p in P):
out.append((u, v, self.rel(u, v)))
return out
def read_set(bx, RF, T, tag, out, first, nv):
d = bx.d
out[(tag, "M")] += 1
rf = np.zeros(len(bx.B), dtype=bool)
rf[bx.idx(RF)] = True
out[(tag, "bug")] += int((rf & (bx.lat > T)).sum() > 0)
drops = {frozenset(u for u in range(d.n) if p[u] < d.Ls[u]) for p in RF}
out[(tag, "D2")] += len(drops) >= 2
closed = median.fast(RF)[0]
out[(tag, "M3")] += not closed
arms = (("A", bx.allp), ("B", bx.reader), ("Banc", bx.anc),
("one", []))
for arm, pairs in arms:
m = bx.mask(RF, T, pairs)
out[(tag, "bug")] += int((rf & ~m).any())
extra = m & ~rf
if not extra.any():
continue
key = "Mc" if arm == "one" else "N" + arm
out[(tag, key)] += 1
if arm == "A":
out[(tag, "NA closed")] += closed
if arm in ("A", "B") and len(first.setdefault(arm + " " + tag,
[])) < 6:
x = tuple(int(t) for t in bx.B[np.flatnonzero(extra)[0]])
dx = sorted(u for u in range(d.n) if x[u] < d.Ls[u])
first[arm + " " + tag].append(
(graphs.name(d.cell), d.units, T, nv, d.Ls, x, dx,
int(extra.sum()), len(RF), bx.missed(RF, x)))
if arm == "B":
x = tuple(int(t) for t in bx.B[np.flatnonzero(extra)[0]])
for u, v, r in bx.missed(RF, x):
out[(tag, "B pair", r)] += 1
def sweep_one(tag, cell, units, out, first):
try:
d = dags.Dag(cell, units)
except AssertionError:
out[(tag, "skipped")] += 1
return
nv = d.naive()
rows = sl.alive_all(d, nv)
least = min(r[0] for r in rows)
bx = None
for T in range(least, nv):
RF, _, _ = cut(median.sets(d, nv, rows, T)["FD"], d.Ls)
if not RF:
continue
bx = bx or Box(d)
read_set(bx, RF, T, tag, out, first, nv)
def controls():
print("K1..K4 the instrument")
out, first = Counter(), {}
cells4 = named(FOUR)
for cell in cells4:
for units in six():
sweep_one("X6", cell, units, out, first)
check("K1 by theorem: C2 = RF on every closed RF set, X6 at the four "
"cells", out[("X6", "NA closed")] == 0 and out[("X6", "bug")] == 0,
f"{out[('X6', 'NA closed')]} closed off, {out[('X6', 'bug')]} "
f"bugs, sets {out[('X6', 'M')]}")
check("K2 the 8 not-closed RF sets rebuilt", out[("X6", "M3")] == 8,
f"{out[('X6', 'M3')]}")
planted = caught = 0
for cell in cells4:
for units in six():
d = dags.Dag(cell, units)
nv = d.naive()
rows = sl.alive_all(d, nv)
least = min(r[0] for r in rows)
for T in range(least, nv):
RF, _, _ = cut(median.sets(d, nv, rows, T)["FD"], d.Ls)
if len(RF) < 4 or not all(median.fast(RF)[:2]):
continue
bx = Box(d)
pr = lambda S: {(u, v, p[u], p[v]) for p in S
for u, v in bx.allp}
full = pr(RF)
hit = None
for U in itertools.combinations(range(d.n), 3):
for val in {tuple(p[u] for u in U) for p in RF}:
S = {p for p in RF
if tuple(p[u] for u in U) != val}
if S and pr(S) == full:
hit = S
break
if hit:
break
if hit is None:
continue
planted += 1
m = bx.mask(hit, T, bx.allp)
s = np.zeros(len(bx.B), dtype=bool)
s[bx.idx(hit)] = True
gone = bx.idx(RF - hit)
caught += bool(m[gone].all())
if planted >= 12:
break
if planted >= 12:
break
if planted >= 12:
break
check("K3 a deleted value triple keeping every pair, read back",
planted >= 5 and caught == planted, f"{caught} of {planted}")
off = 0
for cell in cells4:
for units in six()[:50]:
d = dags.Dag(cell, units)
bx = Box(d)
k = int(np.array(d.Ls, dtype=np.int64) @ bx.w)
off += int(bx.lat[k]) != d.naive()
check("K4 the floors point reads naive, 50 classes x four cells",
off == 0, f"{off} off")
def jobs(shard, ncells, X7):
"""(tag, cell, classes): X6 and Y4 at the 441 cells, Y5 at the 34,
X7 at the four cells, each population's cells dealt by index mod
4 (X7 dealt by class, the four cells each)."""
X6 = ("X6", six(), list(graphs.cells()))
pops = [X6] + [p for p in median.populations() if p[0] != "X6"]
for tag, cls, cells_ in pops:
if ncells:
step = max(1, len(cells_) // ncells)
cells_ = cells_[::step][:ncells]
for i, cell in enumerate(cells_):
if shard is None or i % 4 == shard:
yield tag, cell, cls
if X7 is not None:
cls = X7 if not ncells else X7[::max(1, len(X7) // (25 * ncells))]
part = [c for i, c in enumerate(cls)
if shard is None or i % 4 == shard]
for cell in named(FOUR):
yield "X7", cell, part
def run(shard, ncells):
out, first = Counter(), {}
t0 = time.time()
X7 = seven()
n = 0
for tag, cell, cls in jobs(shard, ncells, X7):
n += 1
for units in cls:
sweep_one(tag, cell, units, out, first)
return out, first, time.time() - t0, n
def dump(path, out, first, secs, n):
rec = {"out": [[list(k), v] for k, v in out.items()],
"first": first, "secs": secs, "cells": n}
with open(path, "w", encoding="utf-8", newline="\n") as fh:
json.dump(rec, fh, default=str)
def merge(paths):
out, first = Counter(), {}
for p in paths:
with open(p, encoding="utf-8") as fh:
rec = json.load(fh)
for k, v in rec["out"]:
out[tuple(k)] += v
for tag, xs in rec["first"].items():
first.setdefault(tag, []).extend(xs)
print(f" {os.path.basename(p)}: {rec['cells']} (pop, cell) jobs, "
f"{rec['secs']:.1f} s")
return out, first
def report(out, first):
tags = sorted({k[0] for k in out})
for tag in tags:
g = lambda *k: out[(tag,) + k]
print(f"\n {tag}: RF sets {g('M')} (two or more drop sets "
f"{g('D2')}, not median-closed {g('M3')}, cut past single "
f"units {g('Mc')}); C2 off {g('NA')} (of them closed "
f"{g('NA closed')}); reader pairs off {g('NB')}, ancestor "
f"pairs off {g('NBanc')}; bugs {g('bug')}; skipped "
f"{g('skipped')}")
bp = {k[2]: v for k, v in out.items() if k[:2] == (tag, "B pair")}
if bp:
print(f" pairs a B-extra point misses, by relation {bp}")
for arm in ("A", "B"):
for x in first.get(arm + " " + tag, [])[:3]:
print(f" first {arm} extra: {x}")
s = lambda k: sum(out[(t, k)] for t in tags)
print(f"\n RF points excluded by the latency or a pair (bugs): "
f"{s('bug')}")
print(f"KILL raise-free parts not cut by two-unit clauses beside the "
f"latency: {s('NA')} (sets {s('M')}; not median-closed "
f"{s('M3')}){' UNREADABLE' if s('M3') < 10 else ''}")
print(f"KILL raise-free parts not cut by reader-pair clauses beside "
f"the latency: {s('NB')} (sets {s('M')}, cut past single units "
f"{s('Mc')}){' UNREADABLE' if s('Mc') < 10 else ''}")
def main(argv):
t0 = time.time()
if argv and argv[0] == "--merge":
out, first = merge(argv[1:])
report(out, first)
return 0
if argv and argv[0] == "--shard":
shard, ncells, path = int(argv[1]), int(argv[2]), argv[3]
out, first, secs, n = run(None if shard < 0 else shard, ncells)
dump(path, out, first, secs, n)
print(f"shard {shard}: {secs:.1f} s")
return 0
controls()
if not all(CHECKS):
print("\nCONTROL FAILED: nothing below is read.")
return 1
print(f"\n{sum(CHECKS)}/{len(CHECKS)} checks, {time.time() - t0:.1f} s")
return 0
if __name__ == "__main__":
sys.exit(main(sys.argv[1:]))