Batch: filters, if !L.has(k) { a line appending } in a batch's block or in a walk's spread items (PLAN_FILTER; L only grown there, one filter a list, none in a nested LOOP): each round asked as the batch opens, its code run in the rounds it said no to, those rounds kept (k and the appends' cells) and asked again at the end in the rounds' order (L as it is and what the kept rounds before add), the rest added at once (pure_list_add_many) once the lists take them and the memory budget all of them (else, or a key added there already, the batch gives its rounds back); a list such a filter grows read in the block only as no earlier round's item would change (its count in the guard, its first, a place when it has no key, a field by key). Keys kept as bits (pure_list_bits) asked by a loop of lanes, fields of a DENSE list found at their keys (pure_list_dense); an outer lane's affine value in four bytes. The suite's BFS batched: PURE_JIT=0 93 to 39.6 ms (the compiled code 7.7, C 10). module_tier's walks filter d (seeds 1, 2, 3, 5, 23 alike: 5,806 filters decided in seed 23's); GRAPH_SOURCE: a graph built by a filter, searched twice, marks whose filter adds another key (EXISTS), under work and memory budgets

1 files changed, 291 insertions, 4 deletions

diff --git a/core/module-batch.c b/core/module-batch.c
index d151642..56e8291 100644
--- a/core/module-batch.c
+++ b/core/module-batch.c
@@ -63,6 +63,23 @@ typedef struct { uint32_t again, resume, add, set; } Sum;
 typedef struct { uint32_t list, field, add, set; Val rows, e; Mask *mask; } Scatter;
 /* 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;
+/* A filter, if !L.has(k) { its code } (PLAN_FILTER: L only grows in the
+ * batch's block; its code a line, appending): its HAS, its code's end, L; k
+ * and the lanes its code runs in this pass; its code's appends (their
+ * instructions, lists and cells' counts). The rounds its code ran in are
+ * kept (filtered: a round's number, k, each append's cells), asked again in
+ * the rounds' order at the end, their appends made for those still not there. */
+enum { FILTERS = 2, FILTER_APPENDS = 4, FILTERED = 1 << 16 };
+typedef struct {
+    uint32_t has, end, list;
+    Val key;
+    Mask *mask;
+    bool none, spread;
+    unsigned first, appends, made, stride; /* made: its appends, as its kept rounds hold their cells */
+    uint32_t pc[FILTER_APPENDS], to[FILTER_APPENDS];
+    unsigned cells[FILTER_APPENDS];
+    size_t kept;
+} Filter;
 /* 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'
@@ -128,9 +145,13 @@ typedef struct {
     unsigned sum_count, scatter_count;
     Sum sums[SUMS];
     Scatter scatters[SUMS];
+    Filter filters[FILTERS];
+    unsigned filter_count;
 } Batch;
 
 static _Thread_local Batch batch;
+static _Thread_local double *filtered[FILTERS]; /* each filter's kept rounds (Filter), grown as needed */
+static _Thread_local size_t filtered_room[FILTERS];
 
 /* A lane's round in the batch (and a round's lane). */
 static inline unsigned round_of(const Batch *b, unsigned l) { return b->reverse ? b->n - 1 - l : l; }
@@ -704,6 +725,49 @@ static bool find(Batch *b, const PureOp *op, uint32_t pc) {
         }
         return true;
     }
+    const uint64_t *bits;
+    size_t words;
+    PureCells at_keys;
+    if ((has || (joined && pure_list_dense(keyed, &at_keys) && at_keys.width[field] != 255)) && pure_list_bits(keyed, &bits, &words)) {
+        /* keys as bits: each lane's asked at once (a key whole, not below 0, below words * 64), a field read at its key */
+        const uint32_t *run = running(b);
+        void *out = fresh(b);
+        if (!out) return true;
+        uint8_t bad[BATCH];
+        size_t top = words * 64;
+        if (has && x->kind == LANES && x->u32) { /* keys in four bytes: whole, not below 0 */
+            const uint32_t *keys = x->lane;
+            for (unsigned l = 0, n = b->n; l < n; ++l) {
+                size_t at = keys[l] < top ? keys[l] : 0;
+                ((uint32_t *)out)[l] = keys[l] < top && bits[at >> 6] >> (at & 63) & 1;
+            }
+            *x = lanes_val(out, true);
+            return true;
+        }
+        for (unsigned l = 0; l < b->n; ++l) {
+            double k = lane_of(x, l);
+            size_t at = k >= 0 && k < (double)top && k == floor(k) ? (size_t)k : top;
+            bool there = at < top && bits[at >> 6] >> (at & 63) & 1;
+            if (has) ((uint32_t *)out)[l] = there;
+            else {
+                bad[l] = !there;
+                at = there ? at : 0;
+                uint8_t width = at_keys.width[field];
+                const void *column = at_keys.columns[field];
+                double v = width == 8 ? ((const double *)column)[at] : width == 4 ? ((const uint32_t *)column)[at]
+                         : width == 1 ? ((const uint8_t *)column)[at] : (double)at; /* (width 0: the key itself) */
+                ((double *)out)[l] = v;
+            }
+        }
+        (void)run;
+        if (!has && any_running(b, bad)) return false;
+        if (joined) {
+            Val rows = {.kind = UNIFORM};
+            read_of(b, list, field, false, &rows);
+        }
+        *x = lanes_val(out, has);
+        return true;
+    }
     double *out = fresh(b); /* any other: each lane's key found as the executor finds it */
     const uint32_t *run = running(b);
     if (!out) return true;
@@ -781,7 +845,11 @@ static Val outer_lanes(Batch *b, const Val *v) {
     void *out = fresh(b);
     if (!out) return *v;
     Val r;
-    if (v->kind == AFFINE) {
+    double last = v->value + v->step * (b->outer_n - 1);
+    if (v->kind == AFFINE && whole32(v->value) && whole32(last)) { /* whole below 2^32 in every outer lane: four bytes */
+        for (unsigned j = 0; j < b->n; ++j) ((uint32_t *)out)[j] = (uint32_t)(v->value + v->step * parent[j]);
+        r = lanes_val(out, true);
+    } else if (v->kind == AFFINE) {
         for (unsigned j = 0; j < b->n; ++j) ((double *)out)[j] = v->value + v->step * parent[j];
         r = lanes_val(out, false);
     } else if (v->u32) {
@@ -1256,6 +1324,62 @@ static bool sums(Batch *b, uint32_t start, uint32_t end) {
 static void nested(Batch *b, const PureOp *op, uint32_t pc);
 static void walk_nested(Batch *b, const PureOp *op, uint32_t pc);
 
+/* The filter whose HAS is at pc, as this pass reaches it: k (the top) kept, its appends from here. false: more than it holds. */
+static bool open_filter(Batch *b, const PureOp *op, uint32_t pc) {
+    unsigned f = 0;
+    while (f < b->filter_count && b->filters[f].has != pc) ++f;
+    if (f == b->filter_count) {
+        if (f == FILTERS) return false;
+        b->filters[b->filter_count++] = (Filter){.has = pc, .end = b->m->code[pc + 2].arg, .list = op->arg, .key = uniform(0, 0)};
+    }
+    Filter *fl = &b->filters[f];
+    Val *k = &b->stack[b->sp - 1];
+    k->temp = false; /* the filter keeps it (HAS gives a new value) */
+    fl->key = *k, fl->mask = NULL, fl->none = false, fl->spread = b->parent != NULL, fl->first = b->append_count, fl->appends = 0;
+    return true;
+}
+
+/* Each filter's rounds whose code ran this pass (spread: a spread walk's
+ * filters', this chunk's; else the batch's own), in the rounds' order: its
+ * round, k and its appends' cells kept; its appends then let go. false: more
+ * rounds than it keeps, or no memory. */
+static bool keep_filtered(Batch *b) {
+    unsigned least = b->append_count;
+    for (unsigned f = 0; f < b->filter_count; ++f) {
+        Filter *fl = &b->filters[f];
+        if (!fl->appends || fl->spread != (b->parent != NULL)) continue;
+        least = fl->first < least ? fl->first : least;
+        if (fl->none) continue;
+        unsigned stride = 2;
+        for (unsigned a = 0; a < fl->appends; ++a) stride += fl->cells[a];
+        fl->stride = stride, fl->made = fl->appends;
+        const uint32_t *mask = fl->mask ? mask_lanes(b, fl->mask) : NULL;
+        for (unsigned k = 0; k < b->n; ++k) {
+            unsigned l = b->parent ? k : round_of(b, k);
+            if (mask && !mask[l]) continue;
+            if (fl->kept == FILTERED) return false;
+            if ((fl->kept + 1) * stride > filtered_room[f]) {
+                size_t room = filtered_room[f] ? 2 * filtered_room[f] : 4096;
+                double *grown = realloc(filtered[f], room * sizeof(double));
+                if (!grown) return false;
+                filtered[f] = grown, filtered_room[f] = room;
+            }
+            double *r = filtered[f] + fl->kept++ * stride;
+            unsigned o = b->parent ? b->parent[l] : 0;
+            r[0] = b->parent ? (double)(b->outer_reverse ? b->outer_n - 1 - o : o) : (double)k;
+            r[1] = lane_of(&fl->key, l);
+            unsigned at = 2;
+            for (unsigned a = 0; a < fl->appends; ++a) {
+                const Append *ap = &b->appends[fl->first + a];
+                for (unsigned c = 0; c < ap->count; ++c) r[at++] = lane_of(&ap->cells[c], l);
+            }
+        }
+        fl->appends = 0, fl->key = uniform(0, 0), fl->mask = NULL;
+    }
+    b->append_count = least;
+    return true;
+}
+
 /* The block from start to end once for the batch's lanes: false where it failed. */
 static bool pass(Batch *b, uint32_t start, uint32_t end) {
     const PureModule *m = b->m;
@@ -1374,6 +1498,12 @@ static bool pass(Batch *b, uint32_t start, uint32_t end) {
             }
             Append *a = &b->appends[b->append_count++];
             a->pc = pc - 1, a->mask = b->active, a->count = count;
+            for (unsigned f = 0; f < b->filter_count; ++f) /* a filter's code's */
+                if (pc - 1 > b->filters[f].has && pc - 1 < b->filters[f].end) {
+                    Filter *fl = &b->filters[f];
+                    if (fl->appends == FILTER_APPENDS) b->failed = true;
+                    else fl->pc[fl->appends] = pc - 1, fl->to[fl->appends] = arg, fl->cells[fl->appends] = count, ++fl->appends;
+                }
             b->sp -= count;
             for (unsigned f = 0; f < count; ++f) {
                 a->cells[f] = b->stack[b->sp + f];
@@ -1382,10 +1512,19 @@ static bool pass(Batch *b, uint32_t start, uint32_t end) {
             break;
         }
         case PURE_OP_LIST_FIND: case PURE_OP_LIST_HAS:
+            if (op->plan & PLAN_FILTER && !open_filter(b, op, pc - 1)) {
+                b->failed = true;
+                break;
+            }
             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 */
             break;
-        case PURE_OP_JZ: branch(b, arg); break;
+        case PURE_OP_JZ:
+            branch(b, arg);
+            if (pc >= 3 && m->code[pc - 3].op == PURE_OP_LIST_HAS && m->code[pc - 3].plan & PLAN_FILTER) /* a filter's: its lanes */
+                for (unsigned f = 0; f < b->filter_count; ++f)
+                    if (b->filters[f].has == pc - 3) b->filters[f].mask = b->active, b->filters[f].none = b->none;
+            break;
         case PURE_OP_LOOP: nested(b, op, pc - 1), pc += arg >> 16; break;
         case PURE_OP_EACH: walk_nested(b, op, pc - 1), pc += arg >> 16; break;
         case PURE_OP_BREAK: { /* the running lanes leave the nested loop */
@@ -1445,6 +1584,7 @@ static void sweep(Batch *b) {
     mark_lanes(kept, b->var);
     mark_lanes(kept, b->nest.count), mark_lanes(kept, b->nest.broken), mark_val(kept, &b->nest.low), mark_lanes(kept, b->nest.rows);
     mark_lanes(kept, b->parent), mark_lanes(kept, b->item_rows);
+    for (unsigned f = 0; f < b->filter_count; ++f) mark_val(kept, &b->filters[f].key), mark_mask(b, kept, kept_mask, b->filters[f].mask);
     for (unsigned r = 0; r < b->reduce_count; ++r) mark_lanes(kept, b->reduce[r].partial), mark_lanes(kept, b->reduce[r].count);
     for (unsigned r = 0; r < b->read_count; ++r) mark_val(kept, &b->reads[r].rows);
     for (unsigned s = 0; s < b->store_count; ++s)
@@ -1576,6 +1716,10 @@ static void spread(Batch *b, uint32_t start, uint32_t end, const uint32_t *count
         memset(b->own, 0, sizeof(b->own));
         if (!pass(b, start, end)) break;
         arrive(b, end);
+        if (b->filter_count && !keep_filtered(b)) {
+            b->failed = true;
+            break;
+        }
         for (unsigned w = 0; w < LOCALS / 64; ++w) /* each own local: the last item's that set it */
             for (uint64_t bits = b->own[w]; bits; bits &= bits - 1) {
                 unsigned at = w * 64 + (unsigned)__builtin_ctzll(bits), j = m;
@@ -1867,6 +2011,133 @@ static bool settle(Batch *b) {
 
 enum { CHUNK = 128 }; /* items a call of pure_list_add_many takes */
 
+static _Thread_local uint64_t *added_keys;  /* decide's keys added to a filter's list by its kept rounds */
+static _Thread_local double *scratch;       /* and the keys its rounds add, for pure_list_takes */
+static _Thread_local size_t added_room, scratch_room;
+static const uint64_t NO_KEY = UINT64_C(0x7FF8000000000001); /* a NaN: never a key */
+
+static uint64_t key_of(double k) {
+    if (k == 0) k = 0; /* -0 and +0 are one key */
+    uint64_t bits;
+    memcpy(&bits, &k, sizeof(bits));
+    return bits;
+}
+
+/* Whether a list keyed by a number has key k now, as HAS asks it. */
+static bool list_has(const PureList *list, unsigned key, double k) {
+    const uint64_t *bits;
+    size_t words;
+    if (pure_list_bits(list, &bits, &words)) {
+        size_t top = words * 64, at = k >= 0 && k < (double)top && k == floor(k) ? (size_t)k : top;
+        return at < top && bits[at >> 6] >> (at & 63) & 1;
+    }
+    bool found;
+    if (!pure_list_contains(list, k, &found)) pure_list_seek(list, key, (PureCell){.num = k}, &found);
+    return found;
+}
+
+/* Whether k is in (add: put in) decide's keys, mask + 1 slots. */
+static bool added(uint64_t bits, size_t mask, bool add) {
+    size_t at = (size_t)((bits ^ bits >> 33) * UINT64_C(0xff51afd7ed558ccd) >> 20) & mask;
+    for (; added_keys[at] != NO_KEY; at = (at + 1) & mask)
+        if (added_keys[at] == bits) return true;
+    if (add) added_keys[at] = bits;
+    return false;
+}
+
+/* Each filter's kept rounds asked again in the rounds' order, its list as it
+ * is and what the kept rounds before add to it: a round whose k is there by
+ * now is left out (its number -1, as the executor's code skips it then);
+ * then whether each list takes the rest at once (pure_list_takes) and the
+ * memory budget all of them. false: the batch gives its rounds back (a key
+ * an append adds there already, a list that does not take them so, the
+ * memory: the executor's rounds tell where). */
+static bool decide(Batch *b) {
+    const PureModule *m = b->m;
+    PureApp *app = b->app;
+    size_t stored = 0, extra = 0;
+    for (unsigned f = 0; f < b->filter_count; ++f) {
+        Filter *fl = &b->filters[f];
+        if (!fl->kept) continue;
+        const PureList *list = app->lists[fl->list];
+        unsigned key = m->state_key[fl->list] - 1u;
+        size_t slots = 64;
+        while (slots < 2 * fl->kept * (fl->made + 1)) slots *= 2;
+        if (slots > added_room) {
+            uint64_t *grown = realloc(added_keys, slots * sizeof(*grown));
+            if (!grown) return false;
+            added_keys = grown, added_room = slots;
+        }
+        if (fl->kept > scratch_room) {
+            double *grown = realloc(scratch, fl->kept * sizeof(*grown));
+            if (!grown) return false;
+            scratch = grown, scratch_room = fl->kept;
+        }
+        for (size_t i = 0; i < slots; ++i) added_keys[i] = NO_KEY;
+        for (size_t i = 0; i < fl->kept; ++i) {
+            double *r = filtered[f] + i * fl->stride;
+            if (list_has(list, key, r[1]) || added(key_of(r[1]), slots - 1, false)) {
+                r[0] = -1;
+                continue;
+            }
+            for (unsigned a = 0, at = 2; a < fl->made; at += fl->cells[a++]) /* its adds to the list: keys not there yet */
+                if (fl->to[a] == fl->list) {
+                    double k = r[at + key];
+                    if (list_has(list, key, k) || added(key_of(k), slots - 1, true)) return false;
+                }
+        }
+        for (unsigned a = 0, at = 2; a < fl->made; at += fl->cells[a++]) {
+            uint32_t to = fl->to[a];
+            int keyed = m->state_key[to] ? (int)m->state_key[to] - 1 : -1;
+            size_t count = 0, s, e;
+            for (size_t i = 0; i < fl->kept; ++i) {
+                const double *r = filtered[f] + i * fl->stride;
+                if (r[0] < 0) continue;
+                if (keyed >= 0) scratch[count] = r[at + (unsigned)keyed];
+                ++count;
+            }
+            if (!(m->state_flags[to] & PURE_LIST_ROWS >> 8) || m->state_indexed[to] ||
+                !pure_list_takes(app->lists[to], keyed, count, scratch, &s, &e))
+                return false;
+            stored += s, extra += e;
+        }
+    }
+    return !((app->memory && app->grown + (int64_t)stored > (int64_t)app->memory) || app->spent + (int64_t)(stored + extra) > (int64_t)app->room);
+}
+
+/* The filters' appends their kept rounds still make (decide), each list's
+ * at once in the rounds' order. STEP_NEXT, or STEP_REFUSED without memory. */
+static int apply_filters(Batch *b, Frame *frame, uint64_t per) {
+    PureApp *app = b->app;
+    for (unsigned f = 0; f < b->filter_count; ++f) {
+        Filter *fl = &b->filters[f];
+        for (unsigned a = 0, at = 2; fl->kept && a < fl->made; at += fl->cells[a++]) {
+            PureList *list = app->lists[fl->to[a]];
+            size_t was = pure_list_size(list), had = pure_list_extra(list);
+            double cells[PURE_RECORD_FIELDS][CHUNK];
+            const double *fields[PURE_RECORD_FIELDS];
+            for (unsigned c = 0; c < fl->cells[a]; ++c) fields[c] = cells[c];
+            for (size_t i = 0; i < fl->kept;) {
+                size_t count = 0, first = i;
+                for (; i < fl->kept && count < CHUNK; ++i) {
+                    const double *r = filtered[f] + i * fl->stride;
+                    if (r[0] < 0) continue;
+                    for (unsigned c = 0; c < fl->cells[a]; ++c) cells[c][count] = r[at + c];
+                    ++count;
+                }
+                if (count && pure_list_add_many(list, count, fields) != count) { /* no memory (decide left no key there) */
+                    app->work += (uint64_t)filtered[f][first * fl->stride] * per;
+                    frame->status = PURE_ALLOCATION, frame->fault_pc = fl->pc[a], frame->sp = 0;
+                    return STEP_REFUSED;
+                }
+            }
+            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);
+        }
+    }
+    return STEP_NEXT;
+}
+
 /* 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)
@@ -2018,7 +2289,18 @@ static int append_items(Batch *b, Frame *frame, unsigned n, uint64_t per) {
     return STEP_NEXT;
 }
 
+static int batch_run(Frame *frame, Loop *loop, unsigned *done);
+
+/* A batch (batch_run), the filters' rounds and keys let go after it (held only while one runs: a thread keeps none). */
 int pure_batch_run(Frame *frame, Loop *loop, unsigned *done) {
+    int how = batch_run(frame, loop, done);
+    if (added_room || scratch_room) free(added_keys), free(scratch), added_keys = NULL, scratch = NULL, added_room = scratch_room = 0;
+    for (unsigned f = 0; f < FILTERS; ++f)
+        if (filtered_room[f]) free(filtered[f]), filtered[f] = NULL, filtered_room[f] = 0;
+    return how;
+}
+
+static int batch_run(Frame *frame, Loop *loop, unsigned *done) {
     PureApp *app = frame->app;
     const PureModule *m = frame->m;
     bool walking = loop->walking; /* a walk: its rounds its places, their work counted as it opened */
@@ -2055,7 +2337,7 @@ int pure_batch_run(Frame *frame, Loop *loop, unsigned *done) {
     b->walk = walking ? (int)loop->walk.list : -1;
     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 = b->append_count = 0;
+    b->mask_free_count = b->reduce_count = b->append_count = b->filter_count = 0;
     b->recur_count = b->update_count = b->record_count = b->step_count = b->sum_count = b->scatter_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 */
@@ -2071,6 +2353,7 @@ int pure_batch_run(Frame *frame, Loop *loop, unsigned *done) {
         b->failed = true; /* the recurrences ahead, round by round; the sums into cells */
     if (!b->failed && pass(b, guard ? loop->start + 5 : loop->start, loop->end)) {
         arrive(b, loop->end);
+        if (b->filter_count && !keep_filtered(b)) b->failed = true; /* the filters' rounds in the batch's own lanes */
         for (unsigned w = 0; w < LOCALS / 64; ++w) /* the lanes each local was set in, made before anything changes */
             for (uint64_t bits = b->sets[w]; bits; bits &= bits - 1) {
                 unsigned at = w * 64 + (unsigned)__builtin_ctzll(bits);
@@ -2080,7 +2363,7 @@ int pure_batch_run(Frame *frame, Loop *loop, unsigned *done) {
         for (unsigned i = 0; i < b->scatter_count; ++i) mask_lanes(b, b->scatters[i].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)) {
+    if (b->failed || work > budget - app->work || (b->filter_count && !decide(b)) || !settle(b)) {
         loop->resume = ++loop->failures >= FAILURES ? UINT64_MAX : round + n; /* its rounds one by one, then batches again */
         return STEP_NEXT;
     }
@@ -2138,6 +2421,10 @@ int pure_batch_run(Frame *frame, Loop *loop, unsigned *done) {
         int how = append_items(b, frame, n, per);
         if (how != STEP_NEXT) return how;
     }
+    if (b->filter_count) {
+        int how = apply_filters(b, frame, 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;
     for (unsigned x = 0; x < b->recur_count; ++x) frame->ev->locals[b->recur_local[x]].num = b->recur_value[x];