A dev build's self-equality scan walks each container once, a free finds its block by the keyed probe, a registry listing waits out a growth, and a view's crossing and scalar read take fewer instructions.

runtime/flan_dev.c

@@ -1475,13 +1475,30 @@ static int flan_reg_snap(flan_reg_entry *e, flan_reg_entry *out) {
 /* The table-wide counter, read on the way into a scan and again on the way
  * out: a compaction between the two moved entries, so the scan saw some of
  * them twice and some not at all. */
+static uint64_t flan_reg_grows;   /* how many times the table has grown */
+
+static void flan_reg_wait(void);
 static int flan_reg_scan_open(uint64_t *at) {
   uint64_t g = __atomic_load_n(&flan_reg_epoch, __ATOMIC_ACQUIRE);
+  /* A growth copies the whole table and takes longer than a compaction, long
+     enough to use up a listing's attempts one wait at a time. So a listing
+     that meets one waits it out, up to a tenth of a second, rather than
+     counting each look as a lost attempt. */
+  for (int w = 0; (g & 1) && w < 400; w++) {
+    flan_reg_wait();
+    g = __atomic_load_n(&flan_reg_epoch, __ATOMIC_ACQUIRE);
+  }
   if (g & 1) return 0;
   *at = g;
   return 1;
 }
 
+/* Did the table grow since [grows0]? A walk that a growth overlapped is
+   retried without counting against its attempts: a growth ends. */
+static int flan_reg_grew(uint64_t grows0) {
+  return __atomic_load_n(&flan_reg_grows, __ATOMIC_ACQUIRE) != grows0;
+}
+
 static int flan_reg_scan_ok(uint64_t at) {
   __atomic_thread_fence(__ATOMIC_ACQUIRE);
   return __atomic_load_n(&flan_reg_epoch, __ATOMIC_ACQUIRE) == at;
@@ -1738,6 +1755,7 @@ static int flan_reg_grow(void) {
   }
   __atomic_store_n(&flan_reg, n, __ATOMIC_RELEASE);
   __atomic_store_n(&flan_reg_capv, ncap, __ATOMIC_RELEASE);
+  __atomic_store_n(&flan_reg_grows, flan_reg_grows + 1, __ATOMIC_RELEASE);
   __atomic_store_n(&flan_reg_epoch, (flan_reg_epoch | 1) + 1, __ATOMIC_RELEASE);
   return 1;
 }
@@ -1896,11 +1914,29 @@ int flan_dev_reg_overflowed(void) { return flan_reg_full; }
  * block handed out as a slice. */
 static flan_reg_entry *flan_reg_find(uintptr_t a);
 
+/* The entry for a block that starts at [base], live before dead, by the
+ * probe a free takes: (free s) always hands over a block's start, so the
+ * question is equality and a scan of the whole table — which grows — would
+ * make every free cost the table's size. NULL when no block starts there. */
+static flan_reg_entry *flan_reg_at_base(uintptr_t base) {
+  flan_reg_entry *dead = NULL;
+  int64_t cap = FLAN_REG_CAP, probe;
+  size_t s0 = flan_reg_slot(base);
+  for (probe = 0; probe < cap; probe++) {
+    size_t j = (s0 + (size_t)probe) & (size_t)(cap - 1);
+    if (flan_reg[j].base == 0) break;
+    if (flan_reg[j].base != base) continue;
+    if (flan_reg[j].died == 0) return &flan_reg[j];
+    if (dead == NULL) dead = &flan_reg[j];
+  }
+  return dead;
+}
+
 int32_t flan_dev_reg_owner_check(const void *p, const void *owner,
                                  const void **found) {
   flan_reg_entry *e;
   if (!flan_reg_on) return 0;
-  e = flan_reg_find((uintptr_t)p);
+  e = flan_reg_at_base((uintptr_t)p);
   if (e == NULL) return flan_reg_full ? 0 : 1;
   if (e->base != (uintptr_t)p) return 1;
   if (e->died != 0) return 3;
@@ -2153,8 +2189,9 @@ int32_t flan_dev_reg_at(const void *p, const char **type, int64_t *typelen,
      rearrangement they are all losing to — and leaving one bare retry in the
      file next to the note explaining why they are wrong is how the next
      person learns the rule has exceptions it does not have. */
+  int regrown = 0;
   for (attempt = 0; attempt < 8; attempt++) {
-    uint64_t at;
+    uint64_t at, grows0 = __atomic_load_n(&flan_reg_grows, __ATOMIC_ACQUIRE);
     int64_t i;
     have = 0;
     if (!flan_reg_scan_open(&at)) { flan_reg_wait(); continue; }
@@ -2168,6 +2205,7 @@ int32_t flan_dev_reg_at(const void *p, const char **type, int64_t *typelen,
     }
     if (flan_reg_scan_ok(at)) break;
     have = 0;
+    if (flan_reg_grew(grows0) && regrown++ < 64) attempt--;
     flan_reg_wait();
   }
   if (!have) return 0;
@@ -2238,8 +2276,9 @@ int64_t flan_dev_reg_by_type(int32_t live_only, int64_t *counts,
      the end of a string literal. The whole walk is retried when a compaction
      ran through the middle of it, since entries moved and the counts would
      hold some blocks twice and some not at all. */
+  int regrown = 0;
   for (attempt = 0; attempt < 8; attempt++) {
-    uint64_t at;
+    uint64_t at, grows0 = __atomic_load_n(&flan_reg_grows, __ATOMIC_ACQUIRE);
     n = 0;
     missed = 0;
     if (!flan_reg_scan_open(&at)) { flan_reg_wait(); continue; }
@@ -2270,7 +2309,11 @@ int64_t flan_dev_reg_by_type(int32_t live_only, int64_t *counts,
       }
       n++;
     }
-    if (!flan_reg_scan_ok(at)) { flan_reg_wait(); continue; }
+    if (!flan_reg_scan_ok(at)) {
+      if (flan_reg_grew(grows0) && regrown++ < 64) attempt--;
+      flan_reg_wait();
+      continue;
+    }
     /* A stable epoch and every slot copied: this is a table that existed. */
     if (missed == 0) return n;
     /* A stable epoch but slots that would not hold still. Worth another walk —

runtime/flan_dyn.c

@@ -2892,7 +2892,39 @@ flan_dyn flan_dyn_ge(flan_dyn a, flan_dyn b, const uint8_t *loc,
  * traps. So a dev build walks the one container, as deep as equality would,
  * and checks every view it holds — the same rule [eq_walk] applies at the
  * top. A release build keeps no guards, and takes the shortcut. */
-static void stale_scan(flan_dyn v, int depth) {
+/* The containers one scan has already walked. A container reached twice —
+ * shared, or holding itself — is walked once, so a scan is linear in what it
+ * can reach rather than exponential, and a cycle ends. Cleared at the start
+ * of each scan; open addressing over the object's address. */
+static flan_obj **scan_seen;
+static size_t     scan_cap, scan_n;
+
+static int scan_first_visit(flan_obj *o) {
+  size_t i, mask;
+  if (scan_n * 2 >= scan_cap) {
+    size_t ncap = scan_cap ? scan_cap * 2 : 64, j;
+    flan_obj **n = (flan_obj **)calloc(ncap, sizeof *n);
+    if (n == NULL) return 0;   /* no room to remember: stop descending */
+    for (j = 0; j < scan_cap; j++) {
+      size_t k;
+      if (scan_seen[j] == NULL) continue;
+      for (k = ((uintptr_t)scan_seen[j] >> 4) & (ncap - 1); n[k] != NULL;
+           k = (k + 1) & (ncap - 1)) {}
+      n[k] = scan_seen[j];
+    }
+    free(scan_seen);
+    scan_seen = n;
+    scan_cap = ncap;
+  }
+  mask = scan_cap - 1;
+  for (i = ((uintptr_t)o >> 4) & mask; scan_seen[i] != NULL; i = (i + 1) & mask)
+    if (scan_seen[i] == o) return 0;
+  scan_seen[i] = o;
+  scan_n++;
+  return 1;
+}
+
+static void stale_walk(flan_dyn v, int depth) {
   flan_obj *o;
   int64_t i, n;
   if (!dyn_boxed(v) || dyn_box(v) != BOX_OBJ || depth >= EQ_DEPTH) return;
@@ -2903,8 +2935,23 @@ static void stale_scan(flan_dyn v, int depth) {
     return;
   }
   if (o->kind != OBJ_VEC && o->kind != OBJ_MAP) return;
+  if (!scan_first_visit(o)) return;
   n = o->kind == OBJ_MAP ? o->len * 2 : o->len;
-  for (i = 0; i < n; i++) stale_scan(o->u.v.items[i], depth + 1);
+  for (i = 0; i < n; i++) stale_walk(o->u.v.items[i], depth + 1);
+}
+
+/* Only when some guarded view has ever been made: until then there is
+ * nothing a scan could find, and a program that never crosses a typed value
+ * into dyn pays nothing for it. */
+static int64_t views_guarded;
+
+static void stale_scan(flan_dyn v, int depth) {
+  if (views_guarded == 0) return;
+  if (scan_n > 0) {
+    memset(scan_seen, 0, scan_cap * sizeof *scan_seen);
+    scan_n = 0;
+  }
+  stale_walk(v, depth);
 }
 
 static int dyn_equal(flan_dyn a, flan_dyn b, int depth) {
@@ -3268,6 +3315,10 @@ extern struct flan_frame *flan_frame_head;   /* runtime/flan_dev.c */
  * would allocate and zero it for nothing, and a crossing is on the hot path.
  * VIEW_GUARDED in [gen] says the guard is there. */
 #define VIEW_GUARDED 0x100   /* bits 16-31 hold the element size */
+/* Bits 9-10: an element kind [flan_dyn_at] boxes inline. */
+#define VIEW_FAST_I64  1
+#define VIEW_FAST_F64  2
+#define VIEW_FAST_BOOL 3
 static inline view_guard *view_g(flan_obj *o) {
   return (o->gen & VIEW_GUARDED) ? (view_guard *)(o + 1) : NULL;
 }
@@ -3294,7 +3345,8 @@ static void guard_storage(view_guard *g, const void *p) {
 
 
 /* A new view record, its guard empty. */
-static flan_obj *view_new(void *base, int64_t len, const uint8_t *desc,
+static inline __attribute__((always_inline)) flan_obj *
+view_new(void *base, int64_t len, const uint8_t *desc,
                           int shape) {
   int dev = flan_dev_views_checked;
   flan_obj *o =
@@ -3308,9 +3360,16 @@ static flan_obj *view_new(void *base, int64_t len, const uint8_t *desc,
   if (shape != VIEW_STRUCT) {
     int64_t sz = desc_size(desc);
     if (sz > 0 && sz < 0x10000) o->gen |= (uint32_t)sz << 16;
+    /* The three element kinds [flan_dyn_at] reads without a call. */
+    o->gen |= (uint32_t)(*desc == 'l' ? VIEW_FAST_I64
+                         : *desc == 'd' ? VIEW_FAST_F64
+                         : *desc == '?' ? VIEW_FAST_BOOL : 0) << 9;
   }
   o->len = len;
-  if (dev) memset(view_g(o), 0, sizeof(view_guard));
+  if (dev) {
+    memset(view_g(o), 0, sizeof(view_guard));
+    views_guarded++;
+  }
   return o;
 }
 
@@ -3651,7 +3710,7 @@ static void view_write(const uint8_t *loc, int64_t loclen, const char *op,
 }
 
 /* Element [i] of a vec-shaped view. The caller has checked the bounds. */
-static uint8_t *view_elem_at(flan_obj *o, int64_t i) {
+static inline uint8_t *view_elem_at(flan_obj *o, int64_t i) {
   int64_t sz = (int64_t)(o->gen >> 16);
   if (sz == 0) sz = desc_size(o->u.view.desc);
   return (uint8_t *)view_base(o) + i * sz;
@@ -3704,7 +3763,8 @@ static uint8_t *view_field(const uint8_t *loc, int64_t loclen, const char *op,
  * for a struct view). [here] is the checker's word that the storage is the
  * calling function's own frame; otherwise a dev build finds the frame that
  * owns a stack address, or the registry block that holds a heap one. */
-static flan_dyn view_make(void *base, int64_t len, const uint8_t *desc,
+static inline __attribute__((always_inline)) flan_dyn
+view_make(void *base, int64_t len, const uint8_t *desc,
                           int shape, int32_t here) {
   flan_obj *o = view_new(base, len, desc, shape);
   view_guard *g = view_g(o);
@@ -3899,7 +3959,15 @@ flan_dyn flan_dyn_at(flan_dyn v, flan_dyn i, const uint8_t *loc,
   if (o->kind == OBJ_VIEW) {
     int64_t len = view_len(loc, loclen, "at", o);
     if (k < 0 || k >= len) trap_range(loc, loclen, "at", v, k, len);
-    return view_read(loc, loclen, "at", o, o->u.view.desc, view_elem_at(o, k));
+    {
+      uint8_t *p = view_elem_at(o, k);
+      switch ((o->gen >> 9) & 3) {
+      case VIEW_FAST_I64: { int64_t x; memcpy(&x, p, 8); return flan_dyn_from_i64(x); }
+      case VIEW_FAST_F64: { double x; memcpy(&x, p, 8); return flan_dyn_from_f64(x); }
+      case VIEW_FAST_BOOL: return flan_dyn_from_bool(*p ? 1 : 0);
+      default: return view_read(loc, loclen, "at", o, o->u.view.desc, p);
+      }
+    }
   }
   if (k < 0 || k >= o->len) trap_range(loc, loclen, "at", v, k, o->len);
   if (o->kind == OBJ_TEXT) return flan_dyn_from_i64(obj_text_bytes(o)[k]);

test/programs/dyn-view-any.flan

@@ -60,6 +60,11 @@
   (let [a [(i64 5) 6 7]]
     (via-slice (slice a 0 3))))
 
+(defn doubled [n i32] dyn
+  (let [v (the dyn [1 2])]
+    (dotimes [i n] (set v (the dyn [v v])))
+    v))
+
 (defn clobber [] i64
   (let [b [(i64 7) 8 9 10 11 12]]
     (+ (at b 0) (at b 5))))
@@ -310,4 +315,20 @@
       ;; has-key? on a value that is not a map, at its site.
       (= n 23)
       (do (println (has-key? (keep 5) :x)) 0)
+      ;; A container compared with itself, once a view exists: shared thirty
+      ;; levels deep, and holding itself. Each must answer at once.
+      (= n 24)
+      (let [a [(i64 1)]
+            k (keep a)
+            v (doubled 30)]
+        (println (= v v))
+        0)
+      (= n 25)
+      (let [a [(i64 1)]
+            k (keep a)
+            c (the dyn [1])]
+        (push c c)
+        (push c c)
+        (println (= c c))
+        0)
       :else (do (println "?") 1))))

test/test_acceptance.ml

@@ -5937,17 +5937,17 @@ level "1"
                a dyn view");
         ("9", "16777217 has no exact f32");
         ("10", "dyn put: a Point has no field :z. Its fields are :x :y");
-        ("14", "dyn-view-any.flan:272:39: dyn put: field :x of a Small is an \
+        ("14", "dyn-view-any.flan:277:39: dyn put: field :x of a Small is an \
                 i8, and 1.5 is a float — (put #Small{:x 1 :z true} :x 1.5)");
         ("15", "dyn put: field :z of a Small is a bool, and the value is nil \
                 — (put #Small{:x 1 :z true} :z nil)");
         ("16", "dyn put: field :x of a Small is an i8, which holds -128 to \
                 127, and 200 does not fit");
         ("19", "#Wide{:a 18000000000000000000 :b 1}\n");
-        ("19", "dyn-view-any.flan:287:18: dyn get: this u64 element is \
+        ("19", "dyn-view-any.flan:292:18: dyn get: this u64 element is \
                 18000000000000000000");
         ("20", "200 does not fit an i8 element, which holds -128 to 127");
-        ("23", "dyn-view-any.flan:312:20: dyn has-key?: int and keyword") ]
+        ("23", "dyn-view-any.flan:317:20: dyn has-key?: int and keyword") ]
     and any_stale =
       [ ("4", "this view points into a local of leak-local, and that call \
                has returned");
@@ -5958,9 +5958,9 @@ level "1"
                 call has returned");
         ("13", "this view points into a local of leak-slice-param, and that \
                 call has returned");
-        ("17", "dyn-view-any.flan:279:44: dyn print: this view points into a \
+        ("17", "dyn-view-any.flan:284:44: dyn print: this view points into a \
                 local of leak-local");
-        ("18", "dyn-view-any.flan:281:53: dyn length: this view points into \
+        ("18", "dyn-view-any.flan:286:53: dyn length: this view points into \
                 a local of leak-local");
         ("21", "5999\n");
         ("21", "this view's storage, a block of i64, has been released");
@@ -5989,6 +5989,21 @@ level "1"
                 saying %S\n" (name (", mode " ^ mode)) text code needle
            end)
         (any_traps @ if dev then any_stale else []);
+      (* A container compared with itself walks what it holds for stale
+         views in a dev build; shared and cyclic containers are walked once
+         each, so both answer in well under a second rather than in time
+         exponential in the sharing, or never. *)
+      List.iter
+        (fun mode ->
+           let t0 = Unix.gettimeofday () in
+           let code, text = run exe (Some mode) in
+           let dt = Unix.gettimeofday () -. t0 in
+           if code <> 0 || text <> "true\n" || dt > 1.0 then begin
+             incr failures;
+             Printf.printf "FAIL  %s\n        got: %S (exit %d) in %.2fs\n"
+               (name (", self-equality, mode " ^ mode)) text code dt
+           end)
+        [ "24"; "25" ];
       (* The collector takes back what it charged for a view: a leak here
          once doubled the heap's trigger forever. *)
       let code, text = run exe (Some "11") in