Batch: a batch's appends made at once (pure_list_add_many, a call a chunk of 128) when every list takes its items so (an event's list, no index into it, one append to it: FLAT not keyed at its end, DENSE by keys it holds at its count, HASHED by number; never past its arrays' limit) and the memory budget takes them all (each item only grows what it counts): columns grown and widened once, the first key there already the executor's EXISTS at its round; else round by round as before, tried at once again after a round that changed a list's form. 50,000 DENSE adds 5.18 to 0.65 ms (the compiled code 0.28). APPEND_SOURCE: keys there already at rounds q and p, keys past a DENSE list's from round p, two appends to one list, widened columns, under memory budgets (seeds 1, 2, 3, 5, 23 alike: 31,402 batches at once in seed 23's, 292 by the budget round by round)

4 files changed, 351 insertions, 14 deletions

diff --git a/core/module-batch.c b/core/module-batch.c
index 6fe4a19..9905d43 100644
--- a/core/module-batch.c
+++ b/core/module-batch.c
@@ -1536,6 +1536,159 @@ static bool settle(Batch *b) {
     return true;
 }
 
+enum { CHUNK = 128 }; /* items a call of pure_list_add_many takes */
+
+/* v in the k rounds from r on, in their order (each round's lane's). */
+static void rounds_of(const Batch *b, const Val *v, unsigned r, unsigned k, double *out) {
+    if (v->kind == UNIFORM)
+        for (unsigned i = 0; i < k; ++i) out[i] = v->value;
+    else if (!b->reverse) {
+        if (v->kind == AFFINE)
+            for (unsigned i = 0; i < k; ++i) out[i] = v->value + v->step * (r + i);
+        else if (v->u32)
+            for (unsigned i = 0; i < k; ++i) out[i] = ((const uint32_t *)v->lane)[r + i];
+        else
+            memcpy(out, (const double *)v->lane + r, k * sizeof(double));
+    } else {
+        unsigned top = b->n - 1 - r; /* round r's lane, the lanes going down */
+        if (v->kind == AFFINE)
+            for (unsigned i = 0; i < k; ++i) out[i] = v->value + v->step * (top - i);
+        else if (v->u32)
+            for (unsigned i = 0; i < k; ++i) out[i] = ((const uint32_t *)v->lane)[top - i];
+        else
+            for (unsigned i = 0; i < k; ++i) out[i] = ((const double *)v->lane)[top - i];
+    }
+}
+
+/* The rounds from *r on that mask runs (NULL: every one), at most CHUNK of
+ * them, v's in each into out (rounds: which, when mask is not NULL); *r
+ * past the last. How many. */
+static unsigned chunk_of(const Batch *b, const Val *v, const uint32_t *mask, unsigned *r, unsigned n, double *out, unsigned *rounds) {
+    if (!mask) {
+        unsigned k = n - *r < CHUNK ? n - *r : CHUNK;
+        rounds_of(b, v, *r, k, out);
+        *r += k;
+        return k;
+    }
+    unsigned k = 0;
+    for (; *r < n && k < CHUNK; ++*r) {
+        unsigned l = round_of(b, *r);
+        if (mask[l]) rounds[k] = *r, out[k++] = lane_of(v, l);
+    }
+    return k;
+}
+
+/* The batch's appends from round from on at once, each list's items by
+ * one call a chunk (pure_list_add_many), when every list takes its items
+ * so and the memory budget takes them all (each item only grows what the
+ * budget counts, so no round in between runs out): as the executor's steps
+ * make them round by round, the first key there already (an append's at
+ * its round, the earliest) its fault. false, nothing changed: make them
+ * round by round. */
+static bool at_once(Batch *b, Frame *frame, unsigned from, unsigned n, uint64_t per, int *how) {
+    PureApp *app = b->app;
+    const PureModule *m = b->m;
+    double keys[BATCH];
+    size_t stored = 0, extra = 0;
+    for (unsigned i = 0; i < b->append_count; ++i) {
+        const Append *a = &b->appends[i];
+        uint32_t arg = m->code[a->pc].arg;
+        int key = m->state_key[arg] ? (int)m->state_key[arg] - 1 : -1;
+        if (!(m->state_flags[arg] & PURE_LIST_ROWS >> 8) || m->state_indexed[arg]) return false; /* an event's list, no index into it */
+        for (unsigned j = 0; j < i; ++j)
+            if (m->code[b->appends[j].pc].arg == arg) return false; /* two appends to one list: their items in turn */
+        const uint32_t *mask = mask_lanes(b, a->mask);
+        size_t count = 0, s, e;
+        if (!mask) {
+            count = n - from;
+            if (key >= 0) rounds_of(b, &a->cells[key], from, n - from, keys);
+        } else
+            for (unsigned r = from; r < n; ++r) {
+                unsigned l = round_of(b, r);
+                if (!mask[l]) continue;
+                if (key >= 0) keys[count] = lane_of(&a->cells[key], l);
+                ++count;
+            }
+        if (!pure_list_takes(app->lists[arg], key, count, keys, &s, &e)) return false;
+        stored += s, extra += e;
+    }
+    if ((app->memory && app->grown + (int64_t)stored > (int64_t)app->memory) || app->spent + (int64_t)(stored + extra) > (int64_t)app->room)
+        return false; /* it runs out at some round: there, round by round */
+    unsigned fault = n, faulting = 0;
+    for (unsigned i = 0; i < b->append_count; ++i) {
+        const Append *a = &b->appends[i];
+        PureList *list = app->lists[m->code[a->pc].arg];
+        const uint32_t *mask = mask_lanes(b, a->mask);
+        size_t was = pure_list_size(list), had = pure_list_extra(list);
+        double cells[PURE_RECORD_FIELDS][CHUNK];
+        const double *fields[PURE_RECORD_FIELDS];
+        unsigned rounds[CHUNK];
+        for (unsigned f = 0; f < a->count; ++f) fields[f] = cells[f];
+        for (unsigned r = from; r < n && r < fault;) {
+            unsigned at = r, count = 0;
+            for (unsigned f = 0; f < a->count; ++f) {
+                at = r;
+                count = chunk_of(b, &a->cells[f], mask, &at, n, cells[f], rounds);
+            }
+            if (!mask)
+                for (unsigned k = 0; k < count; ++k) rounds[k] = r + k;
+            r = at;
+            size_t added = count ? pure_list_add_many(list, count, fields) : 0;
+            if (added == SIZE_MAX) { /* no memory: the event stops there, as the executor's step would */
+                app->work += (uint64_t)rounds[0] * per;
+                frame->status = PURE_ALLOCATION, frame->fault_pc = a->pc, frame->sp = 0;
+                *how = STEP_REFUSED;
+                return true;
+            }
+            if (added < count) { /* its key there already: the executor's fault at that round, unless one comes before it */
+                if (rounds[added] < fault) fault = rounds[added], faulting = i;
+                break;
+            }
+        }
+        app->grown += (int64_t)pure_list_size(list) - (int64_t)was; /* as grows counts them, item by item */
+        app->spent += (int64_t)(pure_list_size(list) + pure_list_extra(list)) - (int64_t)(was + had);
+    }
+    *how = STEP_NEXT;
+    if (fault < n) { /* the event's undone: only the fault and the work so far tell */
+        app->work += (uint64_t)fault * per;
+        frame->kind = PURE_FAULT_EXISTS, frame->fault_pc = b->appends[faulting].pc, frame->sp = 0;
+        *how = STEP_FAULT;
+    }
+    return true;
+}
+
+/* The batch's appends round by round, as the executor's steps make them:
+ * at once where they may (at_once), tried again after a round that changed
+ * a list's form (a keyed list's first items make it DENSE or HASHED). */
+static int append_items(Batch *b, Frame *frame, unsigned n, uint64_t per) {
+    PureApp *app = b->app;
+    unsigned forms[APPENDS];
+    bool again = true;
+    for (unsigned r = 0; r < n; ++r) {
+        if (again) {
+            int how;
+            if (at_once(b, frame, r, n, per, &how)) return how;
+            for (unsigned i = 0; i < b->append_count; ++i) forms[i] = pure_list_form(app->lists[b->m->code[b->appends[i].pc].arg]);
+            again = false;
+        }
+        unsigned l = round_of(b, r);
+        for (unsigned i = 0; i < b->append_count; ++i) {
+            const Append *a = &b->appends[i];
+            const uint32_t *mask = mask_lanes(b, a->mask);
+            if (mask && !mask[l]) continue;
+            for (unsigned f = 0; f < a->count; ++f) frame->stack[f].num = lane_of(&a->cells[f], l);
+            frame->sp = a->count;
+            int how = pure_run_step(frame, a->pc);
+            if (how != STEP_NEXT) { /* its fault at its round: the event's undone, so only how and the work so far tell */
+                app->work += (uint64_t)r * per;
+                return how;
+            }
+            again |= pure_list_form(app->lists[b->m->code[a->pc].arg]) != forms[i];
+        }
+    }
+    return STEP_NEXT;
+}
+
 int pure_batch_run(Frame *frame, Loop *loop, unsigned *done) {
     PureApp *app = frame->app;
     const PureModule *m = frame->m;
@@ -1643,20 +1796,9 @@ int pure_batch_run(Frame *frame, Loop *loop, unsigned *done) {
         }
         for (unsigned i = 0; i < b->scatter_count; ++i) pure_list_touched(b->app->lists[b->scatters[i].list]);
     }
-    for (unsigned r = 0; r < n && b->append_count; ++r) { /* the appends round by round, each as the executor's step makes it */
-        unsigned l = round_of(b, r);
-        for (unsigned i = 0; i < b->append_count; ++i) {
-            const Append *a = &b->appends[i];
-            const uint32_t *mask = mask_lanes(b, a->mask);
-            if (mask && !mask[l]) continue;
-            for (unsigned f = 0; f < a->count; ++f) frame->stack[f].num = lane_of(&a->cells[f], l);
-            frame->sp = a->count;
-            int how = pure_run_step(frame, a->pc);
-            if (how != STEP_NEXT) { /* its fault at its round: the event's undone, so only how and the work so far tell */
-                app->work += (uint64_t)r * per;
-                return how;
-            }
-        }
+    if (b->append_count) {
+        int how = append_items(b, frame, n, per);
+        if (how != STEP_NEXT) return how;
     }
     app->work += work;
     for (unsigned r = 0; r < b->reduce_count; ++r) frame->ev->locals[b->reduce[r].local].num = b->reduce[r].result;
diff --git a/core/module-list.c b/core/module-list.c
index 6da4a43..eb72c44 100644
--- a/core/module-list.c
+++ b/core/module-list.c
@@ -1816,6 +1816,129 @@ bool pure_list_widen(PureList *list, unsigned field, uint8_t bytes) { return wid
 
 void pure_list_touched(PureList *list) { ++list->version; }
 
+bool pure_list_takes(const PureList *list, int key, size_t n, const double *keys, size_t *stored, size_t *extra) {
+    const Layout *l = &list->layout;
+    if (!list->arrays || list->pack || list->copy || list->journaled || l->key != key || n > PURE_LIST_ITEMS - list->count) return false;
+    for (unsigned f = 0; f < l->fields; ++f)
+        if (l->kinds[f] != PURE_CELL_NUMBER && l->kinds[f] != PURE_CELL_BOOL) return false;
+    size_t row = row_width(l), capacity = list->member_capacity;
+    if (key < 0) { /* at its end, FLAT: its arrays grown as flat_insert grows them, never past ARRAYS_LARGE (a tree then) */
+        if (list->form != FLAT) return false;
+        for (size_t c = list->count, room = list->room; c < list->count + n; ++c)
+            if (c == room) {
+                if (2 * room * row > ARRAYS_LARGE) return false;
+                room = room ? 2 * room : 16;
+            }
+    } else if (list->form == DENSE && list->bitmap) { /* every key one it holds at its count now (it holds more as it grows) */
+        size_t limit = dense_limit(list), top = 0;
+        for (size_t i = 0; i < n; ++i) {
+            if (!small_key(keys[i]) || (keys[i] == 0 && signbit(keys[i])) || (size_t)keys[i] >= limit) return false;
+            if ((size_t)keys[i] > top) top = (size_t)keys[i];
+        }
+        if (n && top / 64 >= capacity) /* its bits grown as dense_add grows them */
+            for (capacity = capacity ? capacity : 1; capacity <= top / 64;) capacity *= 2;
+    } else if (list->form == HASHED && list->text_key < 0) { /* its table doubled as hashed_add doubles it, never past ARRAYS_LARGE (made ordered then) */
+        for (size_t c = list->count; c < list->count + n; ++c)
+            if (hashed_full(c + 1, capacity)) {
+                if (2 * capacity * row > ARRAYS_LARGE) return false;
+                capacity *= 2;
+            }
+    } else
+        return false;
+    *stored = n * l->number_bytes;
+    *extra = (capacity - list->member_capacity) * sizeof(*list->members);
+    return true;
+}
+
+/* n numbers into a column of cells bytes wide from row at on. */
+static void column_out(void *column, uint8_t bytes, size_t at, const double *values, size_t n) {
+    switch (bytes) {
+    case 1: for (size_t i = 0; i < n; ++i) ((uint8_t *)column)[at + i] = (uint8_t)values[i]; break;
+    case 4: for (size_t i = 0; i < n; ++i) ((uint32_t *)column)[at + i] = (uint32_t)values[i]; break;
+    case 8: memcpy((double *)column + at, values, n * sizeof(double)); break;
+    }
+}
+
+size_t pure_list_add_many(PureList *list, size_t n, const double *const *fields) {
+    const Layout *l = &list->layout;
+    for (unsigned f = 0; f < l->fields; ++f) { /* every column wide enough for them all first: widths only grow, as arrays_fit grows them */
+        uint8_t now = list->width[f], need = now;
+        if ((now != 1 && now != 4) || !narrows(l->kinds[f])) continue;
+        for (size_t i = 0; i < n && need < 8; ++i) {
+            uint8_t w = width_of(fields[f][i]);
+            if (w > need) need = w;
+        }
+        if (need > now && !widen(list, f, need)) return SIZE_MAX;
+    }
+    size_t done = 0;
+    if (l->key < 0) { /* at its end */
+        size_t room = list->room;
+        while (room < list->count + n) room = room ? 2 * room : 16;
+        if (room != list->room && !arrays_room(list, room)) return SIZE_MAX;
+        for (unsigned f = 0; f < l->fields; ++f) column_out(list->columns[f], list->width[f], list->count, fields[f], n);
+        done = n;
+        list->count += n;
+    } else if (list->form == DENSE) { /* at their keys, each set in its bits in turn: one there already ends them */
+        const double *keys = fields[l->key];
+        size_t room = list->room, top = 0, row = row_width(l);
+        for (size_t i = 0; i < n; ++i) { /* the room dense_add would leave, key after key */
+            size_t key = (size_t)keys[i], limit = 4 * (list->count + i + 1) + DENSE_SLACK / row;
+            if (limit > ARRAYS_LARGE / row) limit = ARRAYS_LARGE / row;
+            if (key >= room) {
+                size_t grown = room + room / 2;
+                if (grown <= key) grown = key + 1 + (key + 1) / 2;
+                room = grown > limit ? limit : grown;
+            }
+            if (key > top) top = key;
+        }
+        if (room != list->room && !arrays_room(list, room)) return SIZE_MAX;
+        if (n && top / 64 >= list->member_capacity) {
+            size_t capacity = list->member_capacity ? list->member_capacity : 1;
+            while (capacity <= top / 64) capacity *= 2;
+            uint64_t *grown = realloc(list->members, capacity * sizeof(*grown));
+            if (!grown) return SIZE_MAX;
+            memset(grown + list->member_capacity, 0, (capacity - list->member_capacity) * sizeof(*grown));
+            list->members = grown, list->member_capacity = capacity;
+        }
+        for (; done < n; ++done) {
+            size_t key = (size_t)keys[done];
+            uint64_t bit = UINT64_C(1) << (key % 64);
+            if (list->members[key / 64] & bit) break;
+            list->members[key / 64] |= bit;
+        }
+        for (unsigned f = 0; f < l->fields; ++f) {
+            void *column = list->columns[f];
+            const double *values = fields[f];
+            switch (list->width[f]) {
+            case 1: for (size_t i = 0; i < done; ++i) ((uint8_t *)column)[(size_t)keys[i]] = (uint8_t)values[i]; break;
+            case 4: for (size_t i = 0; i < done; ++i) ((uint32_t *)column)[(size_t)keys[i]] = (uint32_t)values[i]; break;
+            case 8: for (size_t i = 0; i < done; ++i) ((double *)column)[(size_t)keys[i]] = values[i]; break;
+            }
+        }
+        list->member_count += done;
+        list->count += done;
+    } else { /* HASHED, at their keys' slots: one there already ends them */
+        for (; done < n; ++done) {
+            if (hashed_full(list->count + 1, list->member_capacity) && !hashed_table(list, 2 * list->member_capacity)) {
+                list->version += done;
+                list->bytes += done * l->number_bytes;
+                return SIZE_MAX;
+            }
+            uint64_t bits = key_bits(fields[l->key][done]);
+            size_t mask = list->member_capacity - 1, at = member_slot(bits, mask);
+            while (list->members[at] != VACANT && list->members[at] != bits) at = (at + 1) & mask;
+            if (list->members[at] == bits) break;
+            list->members[at] = bits;
+            for (unsigned f = 0; f < l->fields; ++f) cell_out(list->columns[f], list->width[f], at, (PureCell){.num = fields[f][done]});
+            ++list->member_count;
+            ++list->count;
+        }
+    }
+    list->version += done;
+    list->bytes += done * l->number_bytes;
+    return done;
+}
+
 bool pure_list_settle(PureList *list) {
     if (list->pack || list->copy || list->arrays || list->form != TREE || list->layout.key < 0 || list->count < PACK_ITEMS ||
         list->searched_version != list->version || list->searched < list->count / PACK_SEARCHES)
diff --git a/core/module-list.h b/core/module-list.h
index 28c1936..c44366d 100644
--- a/core/module-list.h
+++ b/core/module-list.h
@@ -186,6 +186,18 @@ bool pure_list_cells(PureList *, PureCells *);
 bool pure_list_widen(PureList *, unsigned field, uint8_t bytes);
 /* A batch set cells of it in place: it changed (its version). */
 void pure_list_touched(PureList *);
+/* Whether n items of numbers and bools go into an event's list at once
+ * (pure_list_add_many) as one after another they would go, the list
+ * staying in its form: a FLAT list not keyed (key -1), at its end; a DENSE
+ * one by keys it holds at its count now; a HASHED one by number (keys: the
+ * items' key field's cells, in order). *stored and *extra: what
+ * pure_list_size and pure_list_extra grow by once they are in. */
+bool pure_list_takes(const PureList *, int key, size_t n, const double *keys, size_t *stored, size_t *extra);
+/* The n items (fields[f][i]: item i's field f) into a list that takes
+ * them, as pure_list_insert at its end or pure_list_add puts each in turn
+ * (its columns grown and widened once): how many went in before one whose
+ * key it had already (n: all of them); SIZE_MAX when memory ran out. */
+size_t pure_list_add_many(PureList *, size_t n, const double *const *fields);
 /* What it keeps its items in now: 0 a tree, 1 FLAT arrays, 2 DENSE (for tests and counts). */
 unsigned pure_list_form(const PureList *);
 /* Field of the item whose key_field is key, in a list ordered by key_field
diff --git a/tests/check.py b/tests/check.py
index 744b505..1fe7079 100644
--- a/tests/check.py
+++ b/tests/check.py
@@ -1338,6 +1338,53 @@ on go {
 """
 
 
+# Items a batch appends at once (APPEND_SOURCE): keys there already, met
+# at round q of a DENSE list and at round p of a HASHED one (the
+# executor's fault at that round); keys past what a DENSE list holds from
+# round p on (it changes its form there, round by round); a list appended
+# to twice a round (round by round); columns widened by the items; all
+# under memory budgets that run out in the middle of a batch.
+APPEND_SOURCE = """record Cell {
+  at = 0
+  v = 0
+}
+state items = 0
+state last = 0
+input go = message {
+  n: u32
+  q: u32
+  p: u32
+}
+on go {
+  let d: list of Cell by at
+  let e: list of Cell by at
+  let h: list of Cell by at
+  let u: list of Cell
+  let w: list of Cell
+  for i in 0 .. go.n {
+    w.append(Cell { at: i, v: i * 300 })
+    d.add(Cell { at: i - floor(i / go.q) * go.q, v: i * 0.5 })
+  }
+  for i in 0 .. go.n {
+    e.add(Cell { at: i + floor(i / go.p) * 1000000, v: i })
+    h.add(Cell { at: i * 7919 - floor(i * 7919 / go.p) * go.p + 5000000, v: 0 - i })
+  }
+  for i in 0 .. go.n {
+    u.append(Cell { at: i, v: i })
+    u.append(Cell { at: 0 - i, v: i * 70000 })
+  }
+  items = d.count + e.count * 10000 + h.count * 100000000 + u.count + w.count
+  let s = 0
+  for c in u {
+    s = s + c.v
+  }
+  for c in w {
+    s = s + c.v * 2
+  }
+  last = s
+}
+"""
+
 def module_assumptions(aup, pure, work):
     """The compiled code's integers (Pure docs/design/022) where they stop
     holding: a loop whose columns hold small whole numbers until, halfway,
@@ -1401,6 +1448,19 @@ def module_assumptions(aup, pure, work):
         assert outputs[name] == outputs['batches'], ('recur', 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 'divide' in faults and 'none' in faults and 'non_finite' in faults and 'memory' in faults, faults
+    source, program = work / 'append.aup', work / 'append.pmod'  # items appended at once, against the rounds one by one
+    source.write_text(APPEND_SOURCE)
+    compile_module(aup, source, program)
+    given = ''.join(f'message go {struct.pack("<III", n, q, p).hex()}\nwork\n' for n, q, p in
+                    ((3000, 70000, 70000), (3000, 1500, 70000), (3000, 70000, 2000), (3000, 70000, 40), (90, 70000, 70000), (20, 3, 5)))
+    given += ''.join(f'memory {b}\n' + given for b in (30000, 90000, 200000)) + 'memory 0\nbudget 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'], ('append', 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 'exists' in faults and 'none' in faults and 'memory' in faults and 'work' in faults, faults
     source, program = work / 'nested.aup', work / 'nested.pmod'  # nested loops together and reductions, against the rounds one by one
     source.write_text(NESTED_SOURCE)
     compile_module(aup, source, program)