dags.py
Checks: LATENCY: Trees of units property; rule; observation.
Cited on: LATENCY.
To run it: save dags.py, with chains.py, graphs.py, manystream.py, poles.py, redundant.py beside it; standard library only.
"""
dags.py -- a graph of on-line product units at one cell: least latency
by brute force over delay assignments, against the tree pass on its
unfolding and two polynomial passes.
QUESTION. graphs.py prices a TREE of product units by one pass of the
narrowed reader's programme. In a DAG a stream may be read by several
units; a unit takes one delay and hands one leftover, min(v(L), room),
to every reader (one inset serves every product reader). Where does
sharing cost, against the DAG's unfolding (each reader its own copy of
what it reads), and is the least latency found by a polynomial pass?
THE SETTING. graphs.py's: a unit reading U and V (fresh, largest
magnitude M^, inset 0; or produced, tau and the producer's leftover)
is alive at delay L where
v(L) = (w - 1) - (Lam w - tau_V eta_U - tau_U eta_V)/b^L >= 0,
Lam = tau_U + tau_V, each inset capped at w - 1. A DAG is a list of
units, unit j reading two distinct inputs, each None (a fresh stream of
its own) or an earlier unit's index. Arrival a_j = L_j + the later of
its produced inputs' arrivals (fresh: 0); LATENCY the largest arrival.
NAIVE: every unit at its floor L*, alive whatever its insets.
BRUTE (the instrument). Every assignment L_j in 0 .. L*_j + 3, depth-
first in index order, a dead unit pruned and any arrival at least the
best latency so far; seeded by naive + 1. Exact Fractions.
UNF(top). The largest over sinks of graphs.price on the sink's
unfolding, each unit at delays 0 .. L* + top. UNF(3) <= BRUTE (the
unfolding relaxes one delay per unit); sharing COSTS where BRUTE >
UNF(3). A DAG is RAISE-PRICED where UNF(0) > UNF(3): its unfolding
reads lower when a unit may sit above its floor.
PASS (the deadline pass). Target t; deadlines D_j first unbounded.
Round: bottom-up, B_j(s) for s = 0..t as graphs.py::step, a produced
input read at latency s - L valued at B_U(min(s - L, D_U)); top-down,
a unit nobody reads at s = t, any other at the least demand of its
readers capped by D_j, its delay the step's argmax there (least L among
ties), demanding s - L of each produced input; D_j = min(D_j, s_j).
Until no deadline moves; then that one assignment is evaluated exactly,
and t is feasible iff it is alive with latency <= t. PASS is the least
feasible t. Sound by construction, so PASS >= BRUTE; on a tree no
deadline moves past round one and PASS is the tree pass. PASS' is PASS
with a unit nobody reads at the least L alive at its s, not the argmax.
CLASSES. A unit list's class: isolated heads (a unit over two fresh
streams that nobody reads) dropped while another unit remains, then the
least list over topological relabellings, each pair sorted, fresh
first. Dropping one changes no latency (property): the remaining list's
unit 0 reads only fresh streams, so it is a head; a head reads fresh
inputs at inset 0, so its least arrival is its floor, the same at every head of a cell, and the latency is a max
over sinks. v is symmetric in (U, V). A list is SHARED where some unit
has two readers; the sweeps read each shared class once.
CONTROLS. K1 brute = graphs.py::price (naive and least) on C, P, S, Q
at all 441 cells. K2 brute on GE's tree (T(6), T(16)) at (8, 3, 5):
naive 7, least 5. K3 random trees of 2 to 8 units at both cells: brute
= graphs.py's pass. K4 (a print): over the family, the count where
brute exceeds UNF(3). K5 PASS >= BRUTE on every DAG of
the family. K6
every labelled shared list of 2..4 units at (6, 2, 4) and (8, 3, 5)
reads (naive, brute, UNF(3), UNF(0), PASS, PASS') as its class. Its
plants, each in a scratch copy run apart: dropping every sink, not only
isolated heads (as frozen), fails it at all 70 lists, but it collapses
lists toward one head, so it reads little; dropping one sink fails it
at 36 of the 70.
PREMISES (P) at (8, 3, 5), d = 1..16, exact, the deep regime's limits
(one input's leaf count to infinity) at finite leaf counts: G = (6, d)'s
floor and theta(d), the inset its deep input must hand for G at 0 (the
6-input's inset 0, and at the cap); a low G's leftover (insets at the
cap); G at its floor (zero insets); the spacer (1, d) at floor 1
(insets at cap) and at delay 2 (zero); the merge (d, d) at delay 0
(zero insets), its floor; rooms beside.
KILL (frozen before its code): "PASS > BRUTE" on one DAG of the
family, random DAGs of 3 to 8 units and the near-source corner (heads
x y shared by several (2, 2) units), at (8, 3, 5) and (8, 2, 6).
Predicted to fire, by a greedy argmax at a shared head near the
sources. Premise predicted: theta(16) within 0.01 of its limit, the
deep regime near 8 leaves.
FINDINGS (every number a print of the run below, the hand limits
marked).
Controls: K1 441 cells, 0 off; K2 (7, 5), 21 units; K3 300 random
trees at two cells, 0 off; K5 holds; K6 70 lists, 0 off.
Premises (P): at d = 16, theta 0.1067 (6-input inset 0) and 0.1018
(at cap) against the hand limit 0.0672 (not printed): the frozen
"within 0.01 at d = 16" MISSED, theta carrying a term w M^(d - 6)
(by hand).
theta < w - 1 only from d = 14 (at cap) or 15 (at 0), not near 8 as
predicted; a low G is alive at all only from d = 14 (+0.0010),
+0.0055 at 16 (hand limit 0.0100). G at its floor 0.1232, the
spacer 0.0530 and 0.1300. The merge's floor is 0 from d = 9 (0.0322; 0.0864 at 11).
So the deep regime needs deep inputs of 14 leaves or more, past every
class swept below: by hand, in 6 units an input unfolds to at most
13 leaves (2, 3, 5, 8, 13). The family's 8-unit DAGs can reach the
regime's leaf counts; whether any meets it there is unprinted.
The family (300 random and 300 corner per cell): sharing never costs
and PASS = BRUTE at all 1,200. The family's kill does not fire.
Classes: 17 shared classes to 4 units, 1,126 to 6.
X4, 4 units at the 441 cells: sharing costs at 8 (class, cell)
pairs, 6 at (6, 2, 4) and 2 at (35, 8, 27); PASS finds no t at 3.
The least, at (6, 2, 4): h = x y, c = (z, h),
r = (h, c), naive = brute 6, UNF(3) 5 (S's unfolding, its bare head
raised).
X6, 6 units at (8, 3, 5) and (8, 2, 6): no class costs; PASS = BRUTE
at all 2,252. Where sharing never costs, the tree pass run once per
unit (one B table per stream, polynomial) prices the DAG; outside
the family PASS finds no t at (6, 2, 4) (X4's third cost row).
RUN RECORD. main: 6/6 checks, 131.3 s (X4 67.5, X6 42.9), peak
11.4 MB, through the gate's guard.
"""
import random
import sys
import time
from collections import Counter
from fractions import Fraction as Fr
import chains
import graphs
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}")
class Dag:
"""Units in index order; each (i0, i1), an input None (fresh) or an
earlier index. Precomputes tau, range, room, floor per unit."""
def __init__(self, cell, units):
self.cell, self.units, self.n = cell, list(units), len(units)
self.tau, self.rng, self.room, self.Ls = [], [], [], []
self.readers = [[] for _ in units]
for j, (a, b) in enumerate(self.units):
assert a is None or b is None or a != b
ta, ra = self.inp(a)
tb, rb = self.inp(b)
rng = graphs.iprod(ra, rb)
assert cell.Mm + rng[0] >= 0 and cell.Mp - rng[1] >= 0
self.tau.append(ta * tb)
self.rng.append(rng)
self.room.append(chains.room(cell, rng))
self.Ls.append(graphs.floor(cell, ta, tb))
for x in (a, b):
if x is not None:
assert x < j
self.readers[x].append(j)
def inp(self, x):
c = self.cell
if x is None:
return c.Mh, (-c.Mm, c.Mp)
return self.tau[x], self.rng[x]
def taus(self, j):
a, b = self.units[j]
return self.inp(a)[0], self.inp(b)[0]
def leftover(self, j, L, left):
"""min(v, room) at delay L on the inputs' leftovers, None if
dead."""
c = self.cell
a, b = self.units[j]
tu, tv = self.taus(j)
eu = Fr(0) if a is None else min(left[a], c.w - 1)
ev = Fr(0) if b is None else min(left[b], c.w - 1)
r = graphs.v(c, tu + tv, L, tu, tv, eu, ev)
return min(r, self.room[j]) if r >= 0 else None
def arrival(self, j, L, arr):
a, b = self.units[j]
return L + max(0 if x is None else arr[x] for x in (a, b))
def evaluate(self, Ls):
"""(latency, alive) of one assignment."""
left, arr, ok = [None] * self.n, [0] * self.n, True
for j in range(self.n):
arr[j] = self.arrival(j, Ls[j], arr)
r = self.leftover(j, Ls[j], left)
if r is None:
ok, r = False, Fr(0)
left[j] = r
return max(arr), ok
def naive(self):
lat, ok = self.evaluate(self.Ls)
assert ok
return lat
def unfold(self, j):
"""The nested shape of the tree unit j unfolds to, each reader
its own copy of what it reads."""
a, b = self.units[j]
f = lambda x: graphs.FRESH if x is None else self.unfold(x)
return (f(a), f(b))
def sinks(self):
return [j for j in range(self.n) if not self.readers[j]]
def brute(d):
"""Least latency over L_j in 0..L*_j + 3, and one assignment."""
best = [d.naive() + 1, None]
left, arr, Ls = [None] * d.n, [0] * d.n, [0] * d.n
def go(j):
if j == d.n:
lat = max(arr)
if lat < best[0]:
best[0], best[1] = lat, list(Ls)
return
for L in range(0, d.Ls[j] + 4):
a = d.arrival(j, L, arr)
if a >= best[0]:
break
r = d.leftover(j, L, left)
if r is None:
continue
left[j], arr[j], Ls[j] = r, a, L
go(j + 1)
left[j], arr[j] = None, 0
go(0)
return best[0], best[1]
def feasible(d, t, least=False):
"""The deadline pass at target t: (ok, assignment). least: PASS',
a unit nobody reads at the least live L at its s, not the
argmax."""
c, n = d.cell, d.n
INF = t + 1
D = [INF] * n
for _ in range(n * (t + 2) + 2):
B, arg, low = [None] * n, [None] * n, [None] * n
for j in range(n):
a, b = d.units[j]
tu, tv = d.taus(j)
row, rarg, rlow = [], [], []
for s in range(t + 1):
best, first = (None, None), None
for L in range(0, d.Ls[j] + 4):
if s - L < 0:
break
es = []
for x in (a, b):
if x is None:
es.append(Fr(0))
continue
sx = min(s - L, D[x])
e = B[x][sx] if sx >= 0 else None
es.append(None if e is None else min(e, c.w - 1))
if None in es:
continue
r = graphs.v(c, tu + tv, L, tu, tv, es[0], es[1])
if r >= 0:
r = min(r, d.room[j])
first = L if first is None else first
if best[0] is None or r > best[0]:
best = (r, L)
row.append(best[0])
rarg.append(best[1])
rlow.append(first)
B[j], arg[j], low[j] = row, rarg, rlow
s_of = [None] * n
demand = [[] for _ in range(n)]
Ls = [None] * n
for j in reversed(range(n)):
s = min(demand[j]) if demand[j] else t
s = min(s, D[j])
if s < 0 or B[j][s] is None:
return False, None
pick = low if least and not d.readers[j] else arg
s_of[j], Ls[j] = s, pick[j][s]
for x in d.units[j]:
if x is not None:
demand[x].append(s - Ls[j])
newD = [min(D[j], s_of[j]) for j in range(n)]
if newD == D:
lat, ok = d.evaluate(Ls)
assert ok and lat <= t, "the pass's assignment"
return True, Ls
D = newD
raise AssertionError("deadlines did not settle")
def deadline_pass(d, least=False):
"""(least feasible t, assignment), or (None, None) where the pass
finds no t up to naive, the naive assignment among those missed."""
for t in range(0, d.naive() + 1):
ok, Ls = feasible(d, t, least)
if ok:
return t, Ls
return None, None
def tree_dag(cell, shape):
"""A nested shape as a Dag, every occurrence its own unit."""
units = []
def walk(node):
if node == graphs.FRESH:
return None
a, b = walk(node[0]), walk(node[1])
units.append((a, b))
return len(units) - 1
walk(shape)
return Dag(cell, units)
def random_dag(rng, n, p_fresh):
units = []
for j in range(n):
ins = []
while len(ins) < 2:
if j == 0 or rng.random() < p_fresh:
ins.append(None)
else:
x = rng.randrange(j)
if x not in ins:
ins.append(x)
units.append(tuple(ins))
return units
def random_tree(rng, n):
"""A random tree of n units: each new unit reads up to two roots."""
roots, units = [], []
for j in range(n):
ins = []
for _ in range(2):
if roots and rng.random() < 0.6:
ins.append(roots.pop(rng.randrange(len(roots))))
else:
ins.append(None)
units.append(tuple(ins))
roots.append(j)
return units
def corner_dag(rng, n):
"""Near-source corner: k heads x y, each read by several (2, 2)
units, those read by later units, n units in all."""
k = rng.randint(1, 3)
units = [(None, None)] * k
m = rng.randint(2, max(2, min(n - k, 4)))
for _ in range(m):
if k == 1:
units.append((0, None))
else:
a, b = rng.sample(range(k), 2)
units.append((a, b))
while len(units) < n:
j = len(units)
a, b = rng.sample(range(k, j), 2) if j - k >= 2 else (k, None)
if rng.random() < 0.3:
b = None
units.append((a, b))
return units
def unfold_least(d, top=3):
"""The tree pass on the DAG's unfolding, latency the largest over
the sinks (each reader its own copy of what it reads), each unit
at delays 0 .. L* + top."""
out = 0
for j in d.sinks():
_, least = graphs.price(d.cell, d.unfold(j), top=top)
out = max(out, least)
return out
def premises():
cell = chains.ms.Cell(8, 3, 5)
M, w = cell.Mh, cell.w
cap = w - 1
def rng_of(d):
r = (-cell.Mm, cell.Mp)
for _ in range(d - 1):
r = graphs.iprod(r, (-cell.Mm, cell.Mp))
return r
def unit(du, dv):
tu, tv = M ** du, M ** dv
room = chains.room(cell, graphs.iprod(rng_of(du), rng_of(dv)))
return tu, tv, graphs.floor(cell, tu, tv), room
f = lambda x: f"{float(x):+.4f}"
print("P premises at (8, 3, 5), exact; w - 1 = "
f"{float(cap):.4f}, 1/b = {1 / 8:.4f}")
print(" d | G=(6,d): L* theta(0) theta(cap) low@cap floor@0 room"
" | (1,d): L* @1cap @2zero | (d,d): L* @0zero room")
for d in range(1, 17):
tu, tv, Ls, room = unit(6, d)
lam = tu + tv
th0 = (lam * w - cap) / tu
thc = (lam * w - cap - tv * cap) / tu
low = graphs.v(cell, lam, 0, tu, tv, cap, cap)
flo = graphs.v(cell, lam, Ls, tu, tv, 0, 0)
su, sv, sL, _ = unit(1, d)
s1 = graphs.v(cell, su + sv, 1, su, sv, cap, cap)
s2 = graphs.v(cell, su + sv, 2, su, sv, 0, 0)
mu, mv, mL, mroom = unit(d, d)
m0 = graphs.v(cell, mu + mv, 0, mu, mv, 0, 0)
print(f" {d:3d} | {Ls:2d} {f(th0)} {f(thc)} {f(low)} {f(flo)} "
f"{f(room)} | {sL:2d} {f(s1)} {f(s2)} | {mL:2d} {f(m0)} "
f"{f(mroom)}")
def controls():
print("K1..K3 the instrument")
cs = list(graphs.cells())
off = 0
for cell in cs:
for key in ("C", "P", "S", "Q"):
d = tree_dag(cell, graphs.SHAPES[key])
got = (d.naive(), brute(d)[0])
off += got != graphs.price(cell, graphs.SHAPES[key])
check("K1 brute = graphs.price on C, P, S, Q", off == 0,
f"{len(cs)} cells, {off} off")
cell = chains.ms.Cell(8, 3, 5)
t0 = time.time()
d = tree_dag(cell, (graphs.tree(6), graphs.tree(16)))
got = (d.naive(), brute(d)[0])
check("K2 brute on (T(6), T(16)) at (8, 3, 5): naive 7, least 5",
got == (7, 5), f"{got}, {d.n} units, {time.time() - t0:.1f} s")
rng = random.Random(1818)
off, cnt = 0, 0
for cell in (chains.ms.Cell(8, 3, 5), chains.ms.Cell(8, 2, 6)):
for _ in range(150):
n = rng.randint(2, 8)
units = random_tree(rng, n)
d = Dag(cell, units)
lat = brute(d)[0]
want = unfold_least(d)
cnt += 1
off += lat != want
check("K3 brute = the tree pass on random trees of 2 to 8 units",
off == 0, f"{cnt} trees at two cells, {off} off")
off, cnt = 0, 0
for cell in (chains.ms.Cell(6, 2, 4), chains.ms.Cell(8, 3, 5)):
for n in range(2, 5):
for units in shapes(n):
if shared(units):
cnt += 1
off += (read_class(cell, units)
!= read_class(cell, canonical(units)))
check("K6 every labelled shared list of 2..4 units reads as its class",
off == 0 and cnt > 0, f"{cnt} lists at (6, 2, 4) and (8, 3, 5), "
f"{off} off")
def family(rng, cell, nr, nc):
for _ in range(nr):
yield "random", random_dag(rng, rng.randint(3, 8),
rng.choice((0.2, 0.35, 0.5)))
for _ in range(nc):
yield "corner", corner_dag(rng, rng.randint(4, 8))
def shapes(n):
"""Every DAG of n units, each unit an unordered pair of inputs from
fresh and the earlier units (fresh twice allowed: two streams)."""
def opts(j):
out = [(None, None)] + [(None, x) for x in range(j)]
out += [(x, y) for x in range(j) for y in range(x + 1, j)]
return out
def go(units):
if len(units) == n:
yield list(units)
return
for u in opts(len(units)):
units.append(u)
yield from go(units)
units.pop()
yield from go([])
def canonical(units):
"""A unit list's class: isolated heads dropped while another unit
remains, then the least relabelled list over topological orders,
each pair sorted, fresh first."""
units = [tuple(u) for u in units]
while len(units) > 1:
read = {x for u in units for x in u if x is not None}
iso = [j for j, u in enumerate(units)
if u == (None, None) and j not in read]
if not iso:
break
j = iso[-1]
f = lambda x: x if x is None or x < j else x - 1
units = [(f(a), f(b)) for a, b in units[:j] + units[j + 1:]]
n = len(units)
best = [None]
def go(new, out):
if len(out) == n:
if best[0] is None or out < best[0]:
best[0] = list(out)
return
cand = []
for j in range(n):
if j in new or any(x is not None and x not in new
for x in units[j]):
continue
p = tuple(sorted(-1 if x is None else new[x]
for x in units[j]))
cand.append((p, j))
m = min(c[0] for c in cand)
for k, j in cand:
if k != m:
continue
new[j] = len(out)
out.append(k)
go(new, out)
out.pop()
del new[j]
go({}, [])
return tuple(tuple(None if x < 0 else x for x in p) for p in best[0])
def shared(units):
"""Some unit read by two or more."""
c = Counter(x for u in units for x in u if x is not None)
return any(k > 1 for k in c.values())
def classes(nmax):
"""The shared classes of 2..nmax units, each once, least first."""
out = set()
for n in range(2, nmax + 1):
for units in shapes(n):
if shared(units):
out.add(canonical(units))
return sorted(out, key=lambda c: (len(c), [tuple(
-1 if x is None else x for x in p) for p in c]))
def read_class(cell, units):
"""(naive, brute, unf(3), unf(0), PASS, PASS') of one list."""
d = Dag(cell, units)
bl = brute(d)[0]
pl = deadline_pass(d)[0]
ql = deadline_pass(d, least=True)[0]
return (d.naive(), bl, unfold_least(d), unfold_least(d, 0), pl, ql)
def sweep(tag, nmax, cells_, extra=False):
"""Every shared class of 2..nmax units at each cell: brute against
the unfolded pass (cost) and PASS; extra: also the capped
unfolding (raise-priced) and PASS'."""
t0 = time.time()
cls = classes(nmax)
n = Counter()
where, wq, ex, ek = Counter(), Counter(), [], []
for cell in cells_:
for units in cls:
nv, bl, u3, u0, pl, ql = read_class(cell, units)
assert u3 <= bl and u3 <= u0
assert (pl is None or pl >= bl) and (ql is None or ql >= bl)
cost, raised = bl > u3, u0 > u3
n["pairs"] += 1
n["cost"] += cost
n["raise"] += raised
n["cost, not raise-priced"] += cost and not raised
n["PASS > BRUTE"] += pl is None or pl > bl
n["PASS no t"] += pl is None
n["PASS' > BRUTE"] += ql is None or ql > bl
n["PASS' no t"] += ql is None
if cost:
where[name(cell)] += 1
if ql is None or ql > bl:
wq[name(cell)] += 1
row = (name(cell), units, nv, bl, u3, u0, pl, ql)
if cost and len(ex) < 3:
ex.append(row)
if (cost and not raised or ql is None or ql > bl) \
and len(ek) < 6:
ek.append(row)
print(f"\n{tag} {len(cls)} shared classes of 2..{nmax} units at "
f"{len(cells_)} cells, {n['pairs']} (class, cell) pairs, "
f"{time.time() - t0:.1f} s")
keys = ["cost", "PASS > BRUTE", "PASS no t"]
if extra:
keys += ["raise", "cost, not raise-priced", "PASS' > BRUTE",
"PASS' no t"]
for k in keys:
print(f" {k:24s} {n[k]}")
print(f" cells where sharing costs: {dict(where)}")
if extra:
print(f" cells where PASS' > BRUTE: {dict(wq)}")
print(" [cell, class, naive, brute, unf(3)"
+ (", unf(0), PASS, PASS']" if extra else ", PASS]"))
cut = (lambda e: e) if extra else (lambda e: e[:5] + e[6:7])
for e in ex:
print(f" cost {cut(e)}")
for e in ek if extra else []:
print(f" KILL {e}")
return n
def name(cell):
return graphs.name(cell)
def main(nr=300, nc=300):
t0 = time.time()
premises()
controls()
if not all(CHECKS):
print("\nCONTROL FAILED: nothing below is read.")
return 1
rng = random.Random(18180)
for cell in (chains.ms.Cell(8, 3, 5), chains.ms.Cell(8, 2, 6)):
stats = {"random": [0, 0, 0, 0, 0], "corner": [0, 0, 0, 0, 0]}
kills, unsound = [], []
for kind, units in family(rng, cell, nr, nc):
d = Dag(cell, units)
nv = d.naive()
bl, bLs = brute(d)
pl, pLs = deadline_pass(d)
pl = d.naive() + 1 if pl is None else pl
ul = unfold_least(d)
s = stats[kind]
s[0] += 1
s[1] += bl < nv
s[2] += bl > ul
s[3] += pl > bl
s[4] += any(len(r) > 1 for r in d.readers)
if pl < bl:
unsound.append((units, pl, bl))
if pl > bl and len(kills) < 4:
kills.append((kind, units, nv, bl, bLs, pl, pLs))
print(f"\nF ({cell.b}, {cell.am}, {cell.ap}): family -> "
"[DAGs, brute < naive, brute > unfolded pass (K4), "
"PASS > BRUTE (the kill), with a shared unit]")
for kind, s in stats.items():
print(f" {kind:7s} {s}")
check(f"K5 PASS >= BRUTE at ({cell.b}, {cell.am}, {cell.ap})",
not unsound, f"{unsound[:2]}")
for kind, units, nv, bl, bLs, pl, pLs in kills:
print(f" KILL {kind}: {units} naive {nv}, brute {bl} "
f"{bLs}, pass {pl} {pLs}")
cs = list(graphs.cells())
two = [c for c in cs if name(c) in ("(8, 3, 5)", "(8, 2, 6)")]
sweep("X4", 4, cs)
sweep("X6", 6, two)
print(f"\n{sum(CHECKS)}/{len(CHECKS)} checks, {time.time() - t0:.1f} s")
return 0 if all(CHECKS) else 1
if __name__ == "__main__":
sys.exit(main(*(int(x) for x in sys.argv[1:])))