Batch executor, stage 4 of design 025: a column read at affine rows in memory's order in four or eight bytes is a view of the column itself (cells_at), no copy, read only while the pass runs; what outlives the pass (a local, a set's rows and value, an append's cells, a sum's rows and e, a filter's key, a read's rows) copies it (own); knapsack 4.0% fewer instructions, 6.6% fewer cycles; VIEW_SOURCE in check.py (a local holding a field the batch sets, a set from the same column in the order that overwrites what it reads, a set widening the column under a view), which differs and trips ASan's memcpy overlap when the copies are left out; seeds 1 2 3 5 23 (default and PURE_SELECT=all), the fixed sources (also sanitized) and the unit tests pass

2 files changed, 86 insertions, 4 deletions

diff --git a/core/module-batch.c b/core/module-batch.c
index 39a3672..8c342a4 100644
--- a/core/module-batch.c
+++ b/core/module-batch.c
@@ -36,6 +36,7 @@ typedef struct {
     uint8_t kind;
     bool u32;       /* lanes: whole numbers below 2^32 in four bytes each (else numbers) */
     bool temp;      /* lanes the stack alone holds: an instruction may leave its result in them */
+    bool view;      /* lanes a list's column itself (cells_at): read only, and only while the pass runs (own) */
     uint16_t owner; /* a position's list (a slot's bytes.owner) */
     double value, step;
     void *lane;
@@ -353,6 +354,18 @@ static const uint32_t *words(Batch *b, const Val *v) {
 }
 
 static Val lanes_val(void *lane, bool u32) { return (Val){.kind = LANES, .u32 = u32, .temp = true, .lane = lane}; }
+
+/* A view's lanes (a column read in place) as lanes of its own: what outlives
+ * the pass (a local, a set, an append, a sum into a cell, a filter's key, a
+ * read's rows) must not see its column set or widened at the batch's end. */
+static void own(Batch *b, Val *v) {
+    if (v->kind != LANES || !v->view) return;
+    void *out = fresh(b);
+    if (!out) return;
+    memcpy(out, v->lane, b->n * (v->u32 ? sizeof(uint32_t) : sizeof(double)));
+    uint16_t owner = v->owner;
+    *v = lanes_val(out, v->u32), v->owner = owner;
+}
 static Val gather(Batch *b, const Val *v);
 static Val expand(Batch *b, const Val *v);
 static Val expand_rows(Batch *b, const Val *v);
@@ -683,20 +696,23 @@ static Val cells_at(Batch *b, const PureCells *cells, unsigned field, const Val
         /* in range in every lane */
         ptrdiff_t first = (ptrdiff_t)rows->value, step = (ptrdiff_t)rows->step;
         b->backward += step < 0 ? 1 : -1;
+        if (step == 1 && (width == 8 || width == 4)) { /* the column itself, in memory's order: a view, no copy */
+            Val v = lanes_val((char *)column + first * width, width == 4);
+            v.temp = false, v.view = true;
+            return v;
+        }
         if (width == 8) {
             double *out = fresh(b);
             if (!out) return *rows;
             const double *from = (const double *)column + first;
-            if (step == 1) memcpy(out, from, n * sizeof(*out));
-            else for (unsigned l = 0; l < n; ++l) out[l] = from[(ptrdiff_t)l * step];
+            for (unsigned l = 0; l < n; ++l) out[l] = from[(ptrdiff_t)l * step];
             return lanes_val(out, false);
         }
         uint32_t *out = fresh(b);
         if (!out) return *rows;
         if (width == 4) {
             const uint32_t *from = (const uint32_t *)column + first;
-            if (step == 1) memcpy(out, from, n * sizeof(*out));
-            else if (step == -1) for (unsigned l = 0; l < n; ++l) out[l] = from[-(ptrdiff_t)l];
+            if (step == -1) for (unsigned l = 0; l < n; ++l) out[l] = from[-(ptrdiff_t)l];
             else for (unsigned l = 0; l < n; ++l) out[l] = from[(ptrdiff_t)l * step];
         } else {
             const uint8_t *from = (const uint8_t *)column + first;
@@ -742,6 +758,7 @@ static void read_of(Batch *b, uint32_t list, uint32_t field, bool known, const V
     }
     Read *r = &b->reads[b->read_count++];
     r->list = list, r->field = field, r->known = known, r->rows = b->sel ? expand_rows(b, rows) : *rows;
+    own(b, &r->rows);
     if (r->rows.kind == LANES) r->rows.temp = false;
 }
 
@@ -929,6 +946,7 @@ static void item_set(Batch *b, uint32_t list, uint32_t field) {
         return;
     }
     if (b->sel) rows = expand_rows(b, &rows), value = expand(b, &value); /* in the lanes around: the selection's mask tells which */
+    own(b, &rows), own(b, &value);
     if (value.kind == LANES) value.temp = false;
     if (rows.kind == LANES) rows.temp = false;
     b->stores[b->store_count++] = (Store){list, field, rows, value, b->sel ? b->selection.active : b->active};
@@ -952,6 +970,7 @@ static void local_store(Batch *b, uint32_t arg) {
         se->old[at] = b->local[at], se->old_state[at] = b->state[at], se->old_written[at] = b->written[at];
         b->state[at] = 0;
     }
+    own(b, &v);
     if (v.kind == LANES) v.temp = false;
     if (!b->active) {
         b->local[at] = v, b->state[at] = 1;
@@ -2005,6 +2024,7 @@ static bool open_filter(Batch *b, const PureOp *op, uint32_t pc) {
     }
     Filter *fl = &b->filters[f];
     Val *k = &b->stack[b->sp - 1];
+    own(b, k);
     k->temp = false; /* the filter keeps it (HAS gives a new value) */
     fl->key = *k, fl->mask = NULL, fl->sel = NULL, fl->none = false, fl->spread = b->parent != NULL, fl->first = b->append_count, fl->appends = 0;
     return true;
@@ -2108,6 +2128,7 @@ static bool pass(Batch *b, uint32_t start, uint32_t end) {
                 sc->list = list_of(set->arg), sc->field = item_field(set->arg), sc->add = sum->add, sc->set = sum->set;
                 sc->e = b->stack[--b->sp], sc->rows = b->stack[--b->sp], sc->mask = b->sel ? b->selection.active : b->active;
                 if (b->sel) sc->e = expand(b, &sc->e), sc->rows = expand_rows(b, &sc->rows);
+                own(b, &sc->e), own(b, &sc->rows);
                 sc->e.temp = sc->rows.temp = false;
             }
             continue;
@@ -2240,6 +2261,7 @@ static bool pass(Batch *b, uint32_t start, uint32_t end) {
             b->sp -= count;
             for (unsigned f = 0; f < count; ++f) {
                 a->cells[f] = b->stack[b->sp + f];
+                own(b, &a->cells[f]);
                 a->cells[f].temp = false;
             }
             break;
diff --git a/tests/check.py b/tests/check.py
index 2314c36..1d98f05 100644
--- a/tests/check.py
+++ b/tests/check.py
@@ -1227,6 +1227,53 @@ on go {
 """
 
 
+# Columns read in place (VIEW_SOURCE, docs/design/025's stage 4): a local
+# holding a row's field while the batch sets that field (its last value
+# written back after the sets), a set whose value is the same column three
+# rows over in the order that overwrites what it reads, and a set that widens
+# the column to eight bytes while a local holds a view of it.
+VIEW_SOURCE = """record Cell {
+  at = 0
+  v = 0
+}
+state last = 0
+state back = 0
+state wide = 0
+state sum = 0
+input go = message {
+  n: u32
+  q: u32
+}
+on go {
+  let t: list of Cell by at
+  for i in 0 .. go.n {
+    t.add(Cell { at: i, v: i * 3 })
+  }
+  let a = 0
+  for i in 0 .. go.n - 3 {
+    a = t[i].v
+    t[i].v = t[i + 3].v + 1
+  }
+  last = a
+  for k in 0 .. go.n - 3 {
+    let i = go.n - 1 - k
+    t[i].v = t[i - 3].v
+  }
+  back = t[go.n - 1].v + t[go.n - 2].v * 1000
+  let w = 0
+  for i in 0 .. go.n - 3 {
+    w = t[i].v
+    t[i].v = t[i + 3].v + go.q + 0.5
+  }
+  wide = w
+  let s = 0
+  for c in t {
+    s = s + c.v
+  }
+  sum = s
+}
+"""
+
 # Reductions that start with no lanes (REDUCE_SOURCE, docs/design/025): the
 # same e in every lane (a count, a uniform max and min, a fraction, a sum
 # past 2^53 one a lane), a count in a nested loop of the same rounds in
@@ -1711,6 +1758,19 @@ def module_assumptions(aup, pure, work):
         assert outputs[name] == outputs['batches'], ('reduce', name, next((a, b) for a, b in zip(outputs[name].splitlines(), outputs['batches'].splitlines()) if a != b))
     faults = [json.loads(line)['fault'] for line in outputs['batches'].splitlines() if line.startswith('{"states"')]
     assert 'none' in faults and 'work' in faults, faults
+    source, program = work / 'view.aup', work / 'view.pmod'  # columns read in place, against the rounds one by one
+    source.write_text(VIEW_SOURCE)
+    compile_module(aup, source, program)
+    given = ''.join(f'message go {struct.pack("<II", n, q).hex()}\nwork\n' for n, q in
+                    ((3000, 7), (3000, 70000), (2049, 1), (40, 3), (5, 2), (1500, 100000)))
+    given += 'budget 30011\n' + given + 'budget 0\n'
+    outputs = {name: subprocess.run([str(pure / 'build/pure-run'), str(program)], input=given, capture_output=True, text=True,
+                                    timeout=120, env=dict(os.environ, **env)).stdout
+               for name, env in (('batches', {'PURE_JIT': '0'}), ('rounds', {'PURE_JIT': '0', 'PURE_BATCH': '0'}), ('all', {'PURE_JIT': 'all'}))}
+    for name in ('rounds', 'all'):
+        assert outputs[name] == outputs['batches'], ('view', name, next((a, b) for a, b in zip(outputs[name].splitlines(), outputs['batches'].splitlines()) if a != b))
+    faults = [json.loads(line)['fault'] for line in outputs['batches'].splitlines() if line.startswith('{"states"')]
+    assert 'none' in faults and 'work' in faults, faults
     return len(runs)