Revert "write tests for algebraic UPat"

This reverts commit 6538935441.

.github/workflows/test.yml

@@ -452,8 +452,6 @@ jobs:
         run: CL=1 IGNORE_BEAM_CACHE=1 python3 -m pytest extra/optimization/test_beam_search.py
       - name: Test MLPerf stuff
         run: CL=1 python -m pytest -n=auto test/external/external_test_optim.py test/external/external_test_losses.py test/external/external_test_metrics.py test/external/external_test_datasets.py --durations=20
-      - name: NULL=1 beautiful_mnist_multigpu
-        run: MAX_BUFFER_SIZE=0 NULL=1 python examples/beautiful_mnist_multigpu.py
       - name: Test Bert training
         run: MAX_BUFFER_SIZE=0 NULL=1 DEFAULT_FLOAT=HALF BENCHMARK=10 BS=24 GPUS=4 BERT_LAYERS=2 MODEL=bert python3 examples/mlperf/model_train.py
       - name: Test llama 3 training

examples/mlperf/training_submission_v5.0/tinycorp/benchmarks/stable_diffusion/implementations/tinybox_8xMI300X/dev_run.sh

@@ -1,72 +0,0 @@
-#!/usr/bin/env bash
-
-DATETIME=${2:-$(date "+%m%d%H%M")}
-LOGFILE="${HOME}/logs/sd_mi300x_${DATETIME}.log"
-# UNET_CKPTDIR must be set: training saves checkpoints to this path, then a separate eval process scans this path to know which checkpoints to eval
-export UNET_CKPTDIR="${HOME}/stable_diffusion/training_checkpoints/${DATETIME}"
-mkdir -p "${HOME}/logs" "$UNET_CKPTDIR"
-
-# run this script in isolation when using the --bg flag
-if [[ "${1:-}" == "--bg" ]]; then
-  echo "logging output to $LOGFILE"
-  echo "saving UNet checkpoints to $UNET_CKPTDIR"
-  script_path="$(readlink -f "${BASH_SOURCE[0]}")"
-  nohup bash "$script_path" run "$DATETIME" >"$LOGFILE" 2>&1 & disown $!
-  exit 0
-fi
-
-# venv management
-if [[ -d .venv-sd-mlperf ]]; then
-  . .venv-sd-mlperf/bin/activate
-else
-  python3 -m venv .venv-sd-mlperf && . .venv-sd-mlperf/bin/activate
-  pip install --index-url https://download.pytorch.org/whl/cpu torch && pip install tqdm numpy ftfy regex pillow scipy wandb webdataset
-fi
-pip list
-apt list --installed | grep amdgpu
-rocm-smi --version
-modinfo amdgpu | grep version
-
-export BEAM=2 BEAM_UOPS_MAX=8000 BEAM_UPCAST_MAX=256 BEAM_LOCAL_MAX=1024 BEAM_MIN_PROGRESS=5 IGNORE_JIT_FIRST_BEAM=1 HCQDEV_WAIT_TIMEOUT_MS=300000
-export AMD_LLVM=0 # bf16 seems to require this
-export DATADIR="/raid/datasets/stable_diffusion"
-export CKPTDIR="/raid/weights/stable_diffusion"
-export EVAL_CKPT_DIR=$UNET_CKPTDIR
-export MODEL="stable_diffusion" PYTHONPATH="."
-export GPUS=8 BS=304
-export CONTEXT_BS=816 DENOISE_BS=600 DECODE_BS=384 INCEPTION_BS=560 CLIP_BS=240
-export WANDB=1
-export PARALLEL=4
-export PYTHONUNBUFFERED=1
-sudo rocm-smi -d 0 1 2 3 4 5 6 7 --setperfdeterminism 1500 || exit 1
-
-# Retry BEAM search if script fails before BEAM COMPLETE is printed, but don't retry after that
-run_retry(){ local try=0 max=5 code tmp py pgid kids
-  while :; do
-    tmp=$(mktemp)
-    setsid bash -c 'exec env "$@"' _ "$@" > >(tee -a "$LOGFILE" | tee "$tmp") 2>&1 &
-    py=$!; pgid=$(ps -o pgid= -p "$py" | tr -d ' ')
-    wait "$py"; code=$?
-    [[ -n "$pgid" ]] && { kill -TERM -"$pgid" 2>/dev/null; sleep 1; kill -KILL -"$pgid" 2>/dev/null; }
-    kids=$(pgrep -P "$py" || true)
-    while [[ -n "$kids" ]]; do
-      kill -TERM $kids 2>/dev/null; sleep 0.5
-      kids=$(for k in $kids; do pgrep -P "$k" || true; done)
-    done
-    grep -q 'BEAM COMPLETE' "$tmp" && { rm -f "$tmp"; return 1; }
-    rm -f "$tmp"
-    ((code==0)) && return 0
-    ((try>=max)) && return 2
-    ((try++)); sleep 90; echo "try = ${try}"
-  done
-}
-
-# Power limiting to 400W is only needed if GPUs fall out of sync (causing 2.2x increased train time) at higher power, which has been observed at 450W
-sudo rocm-smi -d 0 1 2 3 4 5 6 7 --setpoweroverdrive 750 && \
-run_retry TOTAL_CKPTS=7 python3 examples/mlperf/model_train.py; (( $? == 2 )) && { echo "training failed before BEAM completion"; exit 2; }
-sleep 90
-
-run_retry EVAL_SAMPLES=600 python3 examples/mlperf/model_eval.py; (( $? == 2 )) && { echo "eval failed before BEAM completion"; exit 2; }
-# Checkpoints will be evaluated in reverse chronological order, even if above training crashed early
-# STOP_IF_CONVERGED=1: Stop the eval after the first time convergence is detected; no more checkpoints will be evaluated after that.
-STOP_IF_CONVERGED=1 python3 examples/mlperf/model_eval.py

test/test_assign.py

@@ -130,11 +130,6 @@ class TestAssign(unittest.TestCase):
   @unittest.expectedFailure
   def test_assign_changes_realized_alt(self): return self.test_assign_changes_alt(realize=True)
 
-  def test_assign_changes_buffer_alt(self):
-    a, b = [Tensor(Tensor(0).contiguous().realize().uop.as_buf()) for _ in range(2)]
-    Tensor.realize(a.contiguous().assign(1), b.contiguous().assign(2))
-    self.assertEqual((a + b).item(), 3)
-
   def test_assign_diamond_cycle(self):
     # NOTE: should *not* raise AssertionError from numpy
     with self.assertRaisesRegex(RuntimeError, "cycle"):

test/test_schedule.py

@@ -1925,15 +1925,6 @@ class TestSchedule(unittest.TestCase):
     run_schedule(check_schedule(loss, 4))
     np.testing.assert_allclose(loss.item(), 0.878309, atol=1e-5, rtol=1e-6)
 
-  def test_const_folding_alt(self):
-    t = Tensor.full((2,), 1.)
-    lt = (t < 0.)
-    a = Tensor.empty(2).assign(t*lt.where(-1., 0.))
-    b = Tensor.empty(2, dtype=dtypes.bool).assign(lt)
-    Tensor.realize(a, b)
-    self.assertEqual(a.tolist(), [0., 0.])
-    self.assertEqual(b.tolist(), [False, False])
-
   @unittest.skipIf(Device.DEFAULT == "WEBGPU", "Validation error on WebGPU")
   def test_mnist_val(self):
     from tinygrad.nn.datasets import mnist

test/test_tiny.py

@@ -16,8 +16,8 @@ class TestTiny(unittest.TestCase):
     self.assertListEqual(out.tolist(), [1.0, 2.0, 3.0])
 
   def test_elu(self):
-    out = Tensor([[1.,2],[3,4]]).sum(axis=1).elu()
-    self.assertListEqual(out.tolist(), [3.0, 7.0])
+    out = Tensor([1.,2,3]).sum().elu()
+    self.assertEqual(out.item(), 6.0)
 
   def test_plus(self):
     out = Tensor([1.,2,3]) + Tensor([4.,5,6])

tinygrad/runtime/ops_remote.py

@@ -176,7 +176,7 @@ class RemoteHandler:
     self.sessions: defaultdict[SessionKey, RemoteSession] = defaultdict(RemoteSession)
 
     try: self.ib_ctx: IBCtx|None = IBCtx(getenv("IB_DEV", 0))
-    except (RuntimeError, IndexError, AttributeError): self.ib_ctx = None
+    except (IndexError, AttributeError): self.ib_ctx = None
     self.ib_lock = asyncio.Lock()
     self.ib_conns: dict[str, IBConn|None] = {}
     self.iova_cache: dict[tuple[SessionKey, int], tuple[int, int, int]] = {}

tinygrad/runtime/support/ib.py

@@ -10,7 +10,7 @@ DEFAULT_PORT, DEFAULT_GID = getenv("DEFAULT_PORT", 1), getenv("DEFAULT_GID", 3)
 IOVA_ALIGN = resource.getpagesize()
 
 def checkz(x, ret=None):
-  if x != 0: raise RuntimeError(f'{x} != 0 (errno {ctypes.get_errno()})')
+  assert x == 0, f'{x} != 0 (errno {ctypes.get_errno()})'
   return ret
 
 @dataclass(frozen=True)

tinygrad/schedule/rangeify.py

@@ -2,7 +2,7 @@ from typing import Any, cast, Iterator
 import functools, operator, itertools
 from dataclasses import dataclass, field
 from tinygrad.dtype import dtypes, PtrDType, ImageDType, AddrSpace
-from tinygrad.uop.ops import PatternMatcher, UPat, Ops, UOp, resolve, GroupOp, RewriteNotReady, _substitute, ssimplify, KernelInfo
+from tinygrad.uop.ops import PatternMatcher, UPat, Ops, UOp, resolve, GroupOp, ReprocessNode, _substitute, ssimplify, KernelInfo, BottomUpGate
 from tinygrad.uop.symbolic import sym, symbolic_simple
 from tinygrad.helpers import argsort, prod, all_same, pluralize, getenv, RANGEIFY, Context, flatten, dedup, unwrap, all_int, DEBUG, SPLIT_REDUCEOP
 from tinygrad.schedule.kernelize import Kernel
@@ -81,8 +81,8 @@ earliest_rewrites = PatternMatcher([
   # copy only to different device
   (UPat(Ops.COPY, src=(UPat.var("x"), UPat()), name="copy"), lambda x,copy: x.f(Ops.NOOP, tag=copy.tag) if x.device == copy.device else None),
 
-  # contiguous buffer is buffer, this is for *correctness* of assign, not just speed
-  (UPat(Ops.CONTIGUOUS, name="root", src=(UPat(Ops.BUFFER),)), lambda root: root.src[0].forced_reshape(root.shape).rtag(root.tag)),
+  # contiguous/buffer/copy/assign is already contiguous
+  #(UPat(Ops.CONTIGUOUS, name="root", src=(UPat((Ops.CONTIGUOUS, Ops.BUFFER, Ops.COPY, Ops.ASSIGN)),)), lambda root: root.src[0]),
 ])
 
 # *****************
@@ -151,6 +151,7 @@ class RangeifyContext:
   # block on parent until all children have been seen
   seen_children: dict[UOp, dict[int, UOp]] = field(default_factory=dict)
   seen_child: dict[UOp, Any] = field(default_factory=dict)
+  pending_children: dict[UOp, list[UOp]] = field(default_factory=dict)
   progress: int = 0
 
   # create ranges
@@ -270,13 +271,19 @@ def map_reduce(ctx:RangeifyContext, idx:UOp, red:UOp):
 
 def index_child(ctx:RangeifyContext, c:UOp, x:UOp, idx:UOp):
   if c not in ctx.seen_children: ctx.seen_children[c] = {}
-  ctx.seen_children[c][x.arg[0]] = idx
   # wait here until we have seen all the children
+  ctx.seen_children[c][x.arg[0]] = idx
+  print("see child", x.arg)
   if len(ctx.seen_children[c]) != x.arg[1]:
     ctx.progress += 1
     if ctx.progress > 10000: raise RuntimeError("children not making progress")
-    raise RewriteNotReady
+    # NOTE: we mark this here
+    print("BU GATE")
+    ctx.pending_children.setdefault(c, []).append(idx)
+    raise BottomUpGate
+    #raise RewriteNotReady
   ctx.progress = 0
+  print("CHILDREN", id(c))
 
   if c not in ctx.seen_child:
     all_rngs = list(zip(*[ch.src[1:] for ch in ctx.seen_children[c].values()]))
@@ -320,6 +327,11 @@ def index_child(ctx:RangeifyContext, c:UOp, x:UOp, idx:UOp):
 
 def children_gate(ctx:RangeifyContext, idx:UOp, c:UOp):
   if len(ctx.seen_children[c]) != c.arg: raise RuntimeError("all children should have been seen by now")
+  if len(pc:=ctx.pending_children[c]):
+    pcn = pc.pop()
+    print("reprocess", pcn.src[0].arg)
+    raise ReprocessNode(pcn)
+  print("COMPLETE", id(c))
   return idx.replace(src=(idx.src[0].src[0],)+idx.src[1:])
 
 def might_end_axis(idx:UOp):
@@ -344,7 +356,7 @@ pm_rangeify = pm_mops+PatternMatcher([
   (UPat(Ops.INDEX, src=(UPat(Ops.REALIZE, src=(UPat(),), name="x"),), allow_any_len=True, name="idx"), map_partial_realize),
 
   # if there are new ended children, tag the SINK
-  (UPat(Ops.INDEX, src=(UPat(Ops.CHILD, src=(UPat(name="c"), ), name="x"),), allow_any_len=True, name="idx"), index_child),
+  (UPat(Ops.INDEX, src=(UPat(Ops.CHILD, src=(UPat(Ops.CHILDREN, name="c"), ), name="x"),), allow_any_len=True, name="idx"), index_child),
   (UPat(Ops.INDEX, src=(UPat(Ops.CHILDREN, name="c"),), allow_any_len=True, name="idx"), children_gate),
 
   # if we come across this, remove it. it was a CHILD unused in an INDEX
@@ -749,9 +761,7 @@ def get_rangeify_map(sink:UOp) -> dict[UOp, UOp]:
   tsink = graph_rewrite(tsink, pm_rangeify, ctx=(rangeify_ctx:=RangeifyContext()), bottom_up=True, name="rangeify")
   # NOTE: sym (vs symbolic_simple) breaks things here because ranges with len 1 aren't handled right
   tsink = graph_rewrite(tsink, symbolic_simple+pm_reduce_unparented, name="symbolic")  # this supports const folding
-  tsink = graph_rewrite(tsink, pm_cleanups, bottom_up=True, name="remove costly buffers")
-  # TODO: can you substitute and remove costly buffers at the same time?
-  tsink = graph_rewrite(tsink, pm_substitute_recurse, bottom_up=True, name="run substitutes")
+  tsink = graph_rewrite(tsink, pm_cleanups+pm_substitute_recurse, bottom_up=True, name="remove costly buffers")
   tsink = graph_rewrite(tsink, pm_limit_bufs, ctx=rangeify_ctx, name="limit buffers")
 
   # rebuild the sink with all the BUFFERIZEs with tags, this is what's ending up in the tensor graph

tinygrad/uop/__init__.py

@@ -10,7 +10,6 @@ class FastEnum(IntEnum):
 class Ops(FastEnum):
   # uops that aren't rendered
   NOOP = auto(); SINK = auto(); UNIQUE = auto(); DEVICE = auto(); KERNEL = auto(); PRECAST = auto(); REWRITE_ERROR = auto()  # noqa: E702
-  SENTINEL = auto()
 
   # track children
   CHILD = auto(); CHILDREN = auto() # noqa: E702

tinygrad/uop/ops.py

@@ -1023,9 +1023,13 @@ if TRACK_MATCH_STATS or PROFILE:
 
 # *** simple graph rewrite engine ***
 
-SENTINEL = UOp(Ops.SENTINEL)
 class RewriteNotReady(Exception): pass
 class BottomUpGate(Exception): pass
+class ReprocessNode(Exception):
+  def __init__(self, node):
+    self.node = node
+    super().__init__(self, "reprocess node")
+
 class RewriteContext:
   def __init__(self, pm, bpm, ctx=None):
     self.pm: PatternMatcher|None = pm
@@ -1036,58 +1040,55 @@ class RewriteContext:
     self.replace: dict[UOp, UOp] = {}
 
   def cached_pm_rewrite(self, x:UOp):
-    if (ret:=self.pm_cache.get(x,SENTINEL)) is not SENTINEL: return ret
+    if (ret:=self.pm_cache.get(x,False)) is not False: return ret
     ret = self.pm_cache[x] = cast(PatternMatcher, self.pm).rewrite(x, self.ctx)
     return ret
 
   def cached_bpm_rewrite(self, x:UOp):
-    if (ret:=self.bpm_cache.get(x,SENTINEL)) is not SENTINEL: return ret
+    if (ret:=self.bpm_cache.get(x,False)) is not False: return ret
     ret = self.bpm_cache[x] = cast(PatternMatcher, self.bpm).rewrite(x, self.ctx)
     return ret
 
+  def canon(self, u: UOp) -> UOp:
+    # chase replace chains with path compression
+    path = []
+    while True:
+      v = self.replace.get(u)
+      if v is None or v is u:  # no redirect or self
+        rep = u
+        break
+      path.append(u)
+      u = v
+    for x in path: self.replace[x] = rep
+    return rep
+
   def unified_rewrite(self, root:UOp) -> UOp:
     stack: collections.deque[tuple[UOp, int, UOp]] = collections.deque([(root, 0, root)])
-    on_stack = {root}  # all UOps either on the stack or in self.replace, i.e. dont have to be placed again
-    REWRITE_STACK_LIMIT = getenv("REWRITE_STACK_LIMIT", 250000)
     while stack:
-      if len(stack) > REWRITE_STACK_LIMIT: raise RuntimeError("infinite loop in graph_rewrite (stack too big)")
+      if len(stack) > getenv("REWRITE_STACK_LIMIT", 250000): raise RuntimeError("infinite loop in graph_rewrite (stack too big)")
       n, stage, new_n = stack.pop()
+      #n, new_n = self.canon(n), self.canon(new_n)
+      #print(len(stack), stage)
       if n in self.replace: continue  # skip any nodes we have seen
-      if stage == 0:
-        # if bottom up, we rewrite this node early. in both cases, we add its parents to the stack
-        if self.bpm is not None:
-          # apply rewrite rules until a fixed point is reached. may return `uop` itself if PatternMatcher doesn't match
-          test_n: UOp|None = n
-          seen = set()
+      try:
+        if stage == 0:
           try:
-            while test_n is not None:
-              if test_n in seen: raise RuntimeError("infinite loop in fixed_point_rewrite")
-              seen.add(test_n)
-              new_n, test_n = test_n, self.cached_bpm_rewrite(test_n)
-          except RewriteNotReady:
-            # try the full thing again later
-            stack.appendleft((n, 0, n))
-            continue
-          except BottomUpGate:
-            # if the bpm matching raised a gate, we are done with this node and dont continue down the srcs
-            self.replace[n] = new_n
-            continue
-        stack.append((n, 1, new_n))
-        for x in reversed(new_n.src):
-          if x in on_stack: continue
-          stack.append((x, 0, x))
-          on_stack.add(x)
-      elif stage == 1:
-        tmp = []
-        for x in new_n.src:
-          if (rx:=self.replace.get(x, SENTINEL)) is SENTINEL:
-            # if some new sources aren't ready, we try this again later
-            stack.appendleft((n, 1, new_n))
-            break
-          tmp.append(rx)
-        else:
-          # in stage 1, once all srcs are rewritten, rebuild (if changed) or run top-down rewrite
-          if (new_src:=tuple(tmp)) == new_n.src:
+            # if bottom up, we rewrite this node early. in both cases, we add its parents to the stack
+            if self.bpm is not None:
+              # apply rewrite rules until a fixed point is reached. may return `uop` itself if PatternMatcher doesn't match
+              test_n: UOp|None = n
+              seen = set()
+              while test_n is not None:
+                if test_n in seen: raise RuntimeError("infinite loop in fixed_point_rewrite")
+                seen.add(test_n)
+                new_n, test_n = test_n, self.cached_bpm_rewrite(test_n)
+            stack.append((n, 1, new_n))
+            for x in reversed(new_n.src): stack.append((x, 0, x))
+          # if the bpm matching raised a gate, we are done with this node and dont continue down the srcs
+          except BottomUpGate: self.replace[n] = new_n
+        elif stage == 1:
+          new_src = tuple([self.replace[x] for x in new_n.src])
+          if new_src == new_n.src:
             # if top down, do the rewrite. if no rewrite or bottom up, we are done rewriting this node so we add it to the dict
             if self.pm is None or (new_src_n:=self.cached_pm_rewrite(new_n)) is None:
               self.replace[n] = new_n
@@ -1098,14 +1099,35 @@ class RewriteContext:
           # trigger a rewrite of new_src_n, then after that rewrite is done, link it back to n
           stack.append((n, 2, new_src_n))
           stack.append((new_src_n, 0, new_src_n))
-      else:
-        # in stage 2, we link the result of new_n to the result of n
-        if (replaced_new_n:=self.replace.get(new_n, SENTINEL)) is SENTINEL:
-          # not ready, try the link later
-          stack.appendleft((n, 2, new_n))
         else:
-          # otherwise we are done
-          self.replace[n] = replaced_new_n
+          # in stage 2, we link the result of new_n to the result of n
+          self.replace[n] = self.replace[new_n]
+      except ReprocessNode as e:
+        assert e.node is self.replace[e.node]
+
+        # invalidate node and all children
+        invalid = [e.node]
+        tset = [e.node]
+        while len(tset):
+          u: UOp = tset.pop()
+          for c in u.children:
+            if (pc:=c()) is not None:
+              tset.append(pc)
+              invalid.append(pc)
+        print(len(invalid))
+        #for s in list(stack):
+        #  if s[0] in invalid or s[2] in invalid:
+        #    stack.remove(s)
+        #    print("ISSUE")
+        for u in invalid:
+          if u in self.replace:
+            print("del")
+            del self.replace[u]
+            #stack.append((u, 0, u))
+        #stack.append((e.node, 0, e.node))
+        #del self.replace[e.node]
+        stack.clear()
+        stack.append((root, 0, root))
     return self.replace[root]
 
 @track_matches