Batch executor stage 4, first part (024): an item of numbers and bools appended in a batch's block (to a list the block reads nowhere else, no index into it, no nested loop) is made at the batch's end round by round through the executor's own step: keys there already, the memory budget, its undo log as the executor's; a fault there is the executor's at that round (the event undone: its kind, its instruction and the work so far tell), pure_batch_run's third way out. module_tier's handlers append (e, d) in rounds apart and run under a memory budget too (seeds 1, 2, 3, 5, 23 alike); knapsack's table loop batched (13.8 to 13.7 ms: the appends cost as the executor's)

3 files changed, 89 insertions, 17 deletions

diff --git a/core/module-batch.c b/core/module-batch.c
index e83e3c1..a05577c 100644
--- a/core/module-batch.c
+++ b/core/module-batch.c
@@ -25,7 +25,7 @@
 #include "module-run.h"
 #include <stdlib.h>
 
-enum { BATCH = 1024, BUFFERS = 24, MASKS = 32, SPLITS = 8, READS = 16, STORES = 8, REDUCES = 8, LOCALS = 256, LEAST = 16, FAILURES = 3 };
+enum { BATCH = 1024, BUFFERS = 24, MASKS = 32, SPLITS = 8, READS = 16, STORES = 8, APPENDS = 8, REDUCES = 8, LOCALS = 256, LEAST = 16, FAILURES = 3 };
 
 enum { UNIFORM, AFFINE, LANES, REDUCE }; /* REDUCE: x in x = x op e (value: its reduction) */
 typedef struct {
@@ -44,6 +44,8 @@ typedef struct Mask { const uint32_t *lanes; const struct Mask *parent; struct M
 /* An item's field read at rows (known: else any row), and one set: what it sets and in which lanes (NULL: all). */
 typedef struct { uint32_t list, field; bool known; Val rows; } Read;
 typedef struct { uint32_t list, field; Val rows, value; Mask *mask; } Store;
+/* An item a LIST_APPEND at pc appends in the lanes of mask (NULL: all): its fields. */
+typedef struct { uint32_t pc; Mask *mask; unsigned count; Val cells[PURE_RECORD_FIELDS]; } Append;
 /* Lanes a branch sent ahead (NULL: every lane): they run again at pc. */
 typedef struct { uint32_t pc; Mask *mask; } Split;
 /* A reduction (x = x op e, x read no other way): each lane's op of its rounds'
@@ -89,6 +91,8 @@ typedef struct {
     void *spare[BUFFERS];       /* lanes no value holds any more */
     Read reads[READS];
     Store stores[STORES];
+    Append appends[APPENDS];
+    unsigned append_count;
 } Batch;
 
 static _Thread_local Batch batch;
@@ -929,6 +933,21 @@ static bool pass(Batch *b, uint32_t start, uint32_t end) {
         case PURE_OP_INDEX_MOVE: index_move(b); break;
         case PURE_OP_ITEM_GET: item_get(b, list_of(arg), item_field(arg)); break;
         case PURE_OP_ITEM_SET: item_set(b, list_of(arg), item_field(arg)); break;
+        case PURE_OP_LIST_APPEND: { /* kept for the end, made there round by round as the executor makes it */
+            unsigned count = m->record[m->state_ref[arg]].count;
+            if (b->append_count == APPENDS) {
+                b->failed = true;
+                break;
+            }
+            Append *a = &b->appends[b->append_count++];
+            a->pc = pc - 1, a->mask = b->active, a->count = count;
+            b->sp -= count;
+            for (unsigned f = 0; f < count; ++f) {
+                a->cells[f] = b->stack[b->sp + f];
+                a->cells[f].temp = false;
+            }
+            break;
+        }
         case PURE_OP_LIST_FIND: case PURE_OP_LIST_HAS:
             if (!find(b, op, pc - 1)) b->failed = true;
             else if (op->op == PURE_OP_LIST_FIND && op->plan & PLAN_JOINED) ++pc; /* its ITEM_GET with it */
@@ -996,6 +1015,10 @@ static void sweep(Batch *b) {
     for (unsigned s = 0; s < b->store_count; ++s)
         mark_val(kept, &b->stores[s].rows), mark_val(kept, &b->stores[s].value), mark_mask(b, kept, kept_mask, b->stores[s].mask);
     for (unsigned d = 0; d < b->sp; ++d) mark_val(kept, &b->stack[d]);
+    for (unsigned a = 0; a < b->append_count; ++a) {
+        mark_mask(b, kept, kept_mask, b->appends[a].mask);
+        for (unsigned f = 0; f < b->appends[a].count; ++f) mark_val(kept, &b->appends[a].cells[f]);
+    }
     b->spare_count = b->mask_free_count = 0;
     for (unsigned i = 0; i < b->buffers; ++i)
         if (!kept[i]) b->spare[b->spare_count++] = buffer[i];
@@ -1248,21 +1271,22 @@ static bool settle(Batch *b) {
     return true;
 }
 
-unsigned pure_batch_run(Frame *frame, Loop *loop) {
+int pure_batch_run(Frame *frame, Loop *loop, unsigned *done) {
     PureApp *app = frame->app;
     const PureModule *m = frame->m;
     uint64_t per = (uint64_t)(loop->end - loop->start) + 1, left = loop->count - loop->k;
     uint64_t budget = app->work_budget ? app->work_budget : PURE_WORK_DEFAULT, paid = 1 + (budget - app->work) / per;
+    *done = 0;
     if (left < LEAST || loop->integer) {
         loop->resume = UINT64_MAX;
-        return 0;
+        return STEP_NEXT;
     }
     unsigned n = (unsigned)(left < BATCH ? left : BATCH);
     if (paid < n) n = (unsigned)paid;
     double first = loop->slot->num;
     if (n < 2 || (first == 0 && signbit(first))) { /* the work left pays for one round, or the loop counts from -0: one round by the executor */
         loop->resume = loop->k + 1;
-        return 0;
+        return STEP_NEXT;
     }
     Batch *b = &batch;
     b->frame = frame, b->m = m, b->app = app, b->ev = frame->ev, b->n = n, b->first = first;
@@ -1271,7 +1295,7 @@ unsigned pure_batch_run(Frame *frame, Loop *loop) {
     b->slot = (uint32_t)(loop->slot - frame->ev->locals), b->end = loop->end;
     b->failed = b->none = b->items = false, b->active = NULL, b->nest = (Nest){0}, b->work = 0, b->var = NULL;
     b->sp = b->split_count = b->buffers = b->masks = b->read_count = b->store_count = b->spare_count = 0;
-    b->mask_free_count = b->reduce_count = 0;
+    b->mask_free_count = b->reduce_count = b->append_count = 0;
     memset(b->sets, 0, sizeof(b->sets));
     for (uint32_t pc = loop->start; pc < loop->end; ++pc) { /* the locals it sets (a nested loop's slot too), and whether it sets items */
         const PureOp *op = &m->code[pc];
@@ -1289,11 +1313,27 @@ unsigned pure_batch_run(Frame *frame, Loop *loop) {
                 unsigned at = w * 64 + (unsigned)__builtin_ctzll(bits);
                 if (b->state[at] == 2) mask_lanes(b, b->written[at]);
             }
+        for (unsigned a = 0; a < b->append_count; ++a) mask_lanes(b, b->appends[a].mask);
     }
     uint64_t work = (n - 1) * per + b->work; /* the rounds after the first, and every round of their nested loops */
     if (b->failed || work > budget - app->work || !settle(b)) {
         loop->resume = ++loop->failures >= FAILURES ? UINT64_MAX : loop->k + n; /* its rounds one by one, then batches again */
-        return 0;
+        return STEP_NEXT;
+    }
+    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;
+            }
+        }
     }
     app->work += work;
     for (unsigned r = 0; r < b->reduce_count; ++r) frame->ev->locals[b->reduce[r].local].num = b->reduce[r].result;
@@ -1307,5 +1347,6 @@ unsigned pure_batch_run(Frame *frame, Loop *loop) {
         }
     loop->slot->num = first + (double)(n - 1);
     if (b->backward > 0) loop->reverse = !loop->reverse; /* rows going down from lane to lane: the next batch's lanes the other way */
-    return n;
+    *done = n;
+    return STEP_NEXT;
 }
diff --git a/core/module-run.h b/core/module-run.h
index 3f8a9ca..0bc97eb 100644
--- a/core/module-run.h
+++ b/core/module-run.h
@@ -158,10 +158,12 @@ int pure_run_rest(Frame *, uint32_t pc);
 
 /* Loops a batch of rounds at a time (module-batch.c, docs/design/024): the
  * innermost loop's rounds from loop->k (set up and counted; its LOOP
- * planned PLAN_BATCH) run as one batch: how many (0: none ran, its block
- * runs round by round next, at least until round loop->resume), the frame
- * left as the last of them leaves it, their work counted. */
-unsigned pure_batch_run(Frame *, Loop *);
+ * planned PLAN_BATCH) run as one batch: *done how many (0: none ran, its
+ * block runs round by round next, at least until round loop->resume), the
+ * frame left as the last of them leaves it, their work counted. STEP_NEXT,
+ * or how an item it appends stopped at its round (as the executor's step:
+ * the event is undone, the work so far counted). */
+int pure_batch_run(Frame *, Loop *, unsigned *done);
 /* Whether the LOCAL_LOAD at pc, in a block ending at end, begins x = x op e
  * (x its local, op ADD, MIN or MAX, e computed above x with no branch or
  * loop): a reduction across a batch's rounds. *at: op's pc. */
diff --git a/core/module.c b/core/module.c
index 3bd9948..d6f202d 100644
--- a/core/module.c
+++ b/core/module.c
@@ -733,6 +733,15 @@ PURE_COLD static bool batch_block(const PureModule *m, uint32_t start, uint32_t
             *shape |= 2;
             break;
         }
+        case PURE_OP_LIST_APPEND: { /* an item of numbers and bools, its list read nowhere in the block (appended) */
+            for (uint32_t i = 0; i < m->states; ++i) /* an index into it: its moves (state_indexed is set after the plans) */
+                if (m->state_type[i] == PURE_STATE_INDEX && m->state_ref[i] == op->arg) return false;
+            if (depth) return false;
+            for (unsigned f = 0; f < m->record[m->state_ref[op->arg]].count; ++f)
+                if (field_type(m, op->arg, f) != T_NUM && field_type(m, op->arg, f) != T_BOOL) return false;
+            *shape |= 4;
+            break;
+        }
         default:
             if (!has_lanes(m, op, input)) return false;
         }
@@ -740,12 +749,22 @@ PURE_COLD static bool batch_block(const PureModule *m, uint32_t start, uint32_t
     return true;
 }
 
+/* The list an instruction of a batch's block reads, or -1. */
+PURE_COLD static int read_list(const PureOp *op) {
+    switch (op->op) {
+    case PURE_OP_ITEM_GET: case PURE_OP_ITEM_SET: return (int)list_of(op->arg);
+    case PURE_OP_LIST_FIND: case PURE_OP_LIST_HAS: case PURE_OP_LIST_COUNT: case PURE_OP_INDEX_FIRST: case PURE_OP_INDEX_LAST:
+    case PURE_OP_ROW: case PURE_OP_OUTER_ROW: return (int)(op->arg & 0xFFFF);
+    default: return -1;
+    }
+}
+
 /* A count's loop whose block has batch forms throughout (batch_block) runs a
  * batch of its rounds at a time (module-batch.c, docs/design/024), unless it
- * sets items and has a nested loop (not yet), or PURE_BATCH=0 in the
- * environment says none does (the executor alone, to measure it). A local is
- * a reduction when every read of it begins x = x op e and every set of it
- * ends one. */
+ * sets or appends items and has a nested loop (not yet), reads a list it
+ * appends to, or PURE_BATCH=0 in the environment says none does (the
+ * executor alone, to measure it). A local is a reduction when every read of
+ * it begins x = x op e and every set of it ends one. */
 PURE_COLD static bool loop_batches(const PureModule *m, uint32_t pc, int input) {
     static int off = -1;
     if (off < 0) {
@@ -774,7 +793,15 @@ PURE_COLD static bool loop_batches(const PureModule *m, uint32_t pc, int input)
     free(ends);
     for (unsigned w = 0; w < 4; ++w) reduced[w] &= set[w] & ~plain[w];
     unsigned shape = 0;
-    return batch_block(m, start, end, input, 0, sure, set, reduced, &shape) && shape != 3;
+    if (!batch_block(m, start, end, input, 0, sure, set, reduced, &shape) || (shape & 1 && shape & 6)) return false;
+    uint32_t appended = 0; /* a list appended to is read nowhere else in the block (an earlier round's item not there yet) */
+    for (uint32_t at = start; at < end; ++at)
+        if (m->code[at].op == PURE_OP_LIST_APPEND) appended |= UINT32_C(1) << m->code[at].arg;
+    for (uint32_t at = start; at < end; ++at) {
+        int list = read_list(&m->code[at]);
+        if (list >= 0 && list < 32 && appended >> list & 1) return false;
+    }
+    return true;
 }
 
 /* A fold in a verified block whose own block has lane forms throughout runs
@@ -2646,7 +2673,9 @@ static inline __attribute__((always_inline)) int iterate(Frame *frame, uint32_t
 static __attribute__((noinline, cold)) int batches(Frame *frame, Loop *loop, uint32_t *next) {
     unsigned depth = frame->depth;
     for (;;) {
-        unsigned done = pure_batch_run(frame, loop);
+        unsigned done;
+        int how = pure_batch_run(frame, loop, &done);
+        if (how != STEP_NEXT) return how;
         if (done) {
             if ((loop->k += done) >= loop->count) { /* it ended */
                 frame->arena->used = loop->mark;