""" 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:]))