diff options
| -rw-r--r-- | core/module-batch.c | 295 | ||||
| -rw-r--r-- | core/module-list.c | 20 | ||||
| -rw-r--r-- | core/module-list.h | 8 | ||||
| -rw-r--r-- | core/module-run.h | 6 | ||||
| -rw-r--r-- | core/module.c | 88 | ||||
| -rw-r--r-- | tests/check.py | 120 |
6 files changed, 517 insertions, 20 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]; diff --git a/core/module-list.c b/core/module-list.c index eb72c44..3df710a 100644 --- a/core/module-list.c +++ b/core/module-list.c @@ -1814,6 +1814,26 @@ bool pure_list_cells(PureList *list, PureCells *out) { bool pure_list_widen(PureList *list, unsigned field, uint8_t bytes) { return widen(list, field, bytes); } +bool pure_list_bits(const PureList *list, const uint64_t **bits, size_t *words) { + if (!list->members || !list->bitmap || list->layout.key < 0) return false; + *bits = list->members, *words = list->member_capacity; + return true; +} + +bool pure_list_dense(const PureList *list, PureCells *out) { + if (list->form != DENSE) return false; + const Layout *l = &list->layout; + out->count = list->room; + out->owned = false; + for (unsigned f = 0; f < l->fields; ++f) { + out->columns[f] = list->columns[f]; + out->width[f] = list->width[f] == 0 || ((l->kinds[f] == PURE_CELL_NUMBER || l->kinds[f] == PURE_CELL_BOOL) && + (list->width[f] == 1 || list->width[f] == 4 || list->width[f] == 8)) + ? list->width[f] : 255; + } + return true; +} + 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) { diff --git a/core/module-list.h b/core/module-list.h index c44366d..e2007a0 100644 --- a/core/module-list.h +++ b/core/module-list.h @@ -186,6 +186,14 @@ 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 *); +/* A list keyed by a number whose keys are a set of bits (every key a whole + * number below its words * 64: pure_list_contains's): the bits, for a + * batch's HAS (docs/design/024); false when its keys are kept otherwise. */ +bool pure_list_bits(const PureList *, const uint64_t **bits, size_t *words); +/* A DENSE list's cells at their keys (count: the keys its columns hold; + * width as pure_list_cells's; a key there when its bit is), for a batch's + * fields found by key; false when it is not DENSE. */ +bool pure_list_dense(const PureList *, PureCells *); /* 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 diff --git a/core/module-run.h b/core/module-run.h index 46494e4..67641ed 100644 --- a/core/module-run.h +++ b/core/module-run.h @@ -57,8 +57,10 @@ static inline bool number_op(uint8_t op, double a, double b, double *out) { /* An item's or a key's op->plan: its cell a text; a FIND joined to the ITEM_GET after it; a FIND whose row an ITEM_SET * takes; a LOOP whose block a batch may run (module-batch.c); a LOCAL_STORE ending x = e of a recurrence in such a block; * such a LOOP whose block begins if slot >= L.count { break } (L only grows in it: a batch runs to L's count as it opens); - * an EACH in a batch's block whose items' rounds may run apart (spread: no break, its locals its items' own or reductions). */ -enum { PLAN_TEXT = 1, PLAN_JOINED = 2, PLAN_ROW = 4, PLAN_BATCH = 8, PLAN_RECUR = 16, PLAN_GUARD = 32, PLAN_SPREAD = 64 }; + * an EACH in a batch's block whose items' rounds may run apart (spread: no break, its locals its items' own or reductions); + * a HAS in a batch's block beginning if !L.has(k) { ... } whose code only adds to L (a filter: L only grows, so a batch asks + * each round as it opens and again, in the rounds' order, at its end). */ +enum { PLAN_TEXT = 1, PLAN_JOINED = 2, PLAN_ROW = 4, PLAN_BATCH = 8, PLAN_RECUR = 16, PLAN_GUARD = 32, PLAN_SPREAD = 64, PLAN_FILTER = 128 }; /* An item instruction's argument: its list and its field. */ static inline uint32_t list_of(uint32_t arg) { return arg & 0xFFFF; } diff --git a/core/module.c b/core/module.c index 9403687..291a63b 100644 --- a/core/module.c +++ b/core/module.c @@ -686,6 +686,24 @@ PURE_COLD static bool block_has_lanes(const PureModule *m, PureBlock block, int return true; } +/* The end of the body of if !L.has(k) { ... } whose HAS is at h, when its + * code is a line of lane forms and appends (no jump, loop, break, local set, + * item set or HAS) within [h, end): a filter (module-batch.c); else 0. */ +PURE_COLD static uint32_t filter_body(const PureModule *m, uint32_t h, uint32_t end) { + const PureOp *op = &m->code[h]; + if (op->op != PURE_OP_LIST_HAS || h + 3 > end || op[1].op != PURE_OP_NOT || op[2].op != PURE_OP_JZ || op[2].arg > end || + op[2].arg <= h + 3 || m->height[h] != 1) + return 0; + for (uint32_t at = h + 3; at < op[2].arg; ++at) + switch (m->code[at].op) { + case PURE_OP_JZ: case PURE_OP_JMP: case PURE_OP_LOOP: case PURE_OP_EACH: case PURE_OP_BREAK: case PURE_OP_LOCAL_STORE: + case PURE_OP_ITEM_SET: case PURE_OP_LIST_HAS: + return 0; + default:; + } + return op[2].arg; +} + /* Whether the block [start, end) of a loop a batch runs (depth: the loops * around it the batch runs too) has batch forms throughout: lane forms, a * local set (a number, an index or a bool), read only after its round sets @@ -696,11 +714,13 @@ PURE_COLD static bool block_has_lanes(const PureModule *m, PureBlock block, int * in a nested loop). shape: 1 a nested loop, 2 a set. */ PURE_COLD static bool batch_block(const PureModule *m, uint32_t start, uint32_t end, int input, unsigned depth, uint64_t *sure, const uint64_t *set, const uint64_t *reduced, unsigned *shape) { - uint32_t arms = start; /* code before arms: in a branch's way */ + uint32_t arms = start, body = 0; /* code before arms: in a branch's way; before body: a filter's */ for (uint32_t at = start; at < end; ++at) { const PureOp *op = &m->code[at]; unsigned local = op->arg & 0xFF; uint64_t bit = UINT64_C(1) << (local % 64); + uint32_t filtered = op->op == PURE_OP_LIST_HAS && at >= arms ? filter_body(m, at, end) : 0; + if (filtered) body = filtered; switch (op->op) { case PURE_OP_LOOP: { uint32_t inner = at + 1, stop = inner + (op->arg >> 16); @@ -748,10 +768,10 @@ PURE_COLD static bool batch_block(const PureModule *m, uint32_t start, uint32_t 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; + if (depth && at >= body) return false; /* in a nested loop: a filter's alone */ 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; + *shape |= at < body ? 8 : 4; break; } default: @@ -771,6 +791,8 @@ PURE_COLD static int read_list(const PureOp *op) { } } +PURE_COLD static bool walk_spreads(const PureModule *m, uint32_t start, uint32_t end, uint32_t each, const uint64_t *reduced); + /* A count's loop or a walk whose block has batch forms throughout * (batch_block, after a count's guard: if slot >= L.count { break }, *guard) * runs a batch of its rounds at a time (module-batch.c, @@ -838,15 +860,55 @@ PURE_COLD static bool loop_batches(const PureModule *m, uint32_t pc, int input, for (unsigned w = 0; w < 4; ++w) exempt[w] = reduced[w] | recurs[w], reductions[w] = reduced[w]; unsigned shape = 0; bool any = recurs[0] | recurs[1] | recurs[2] | recurs[3]; - if (!batch_block(m, *guard ? start + 5 : start, end, input, 0, sure, set, exempt, &shape) || (shape & 1 && (shape & 6 || any))) + if (!batch_block(m, *guard ? start + 5 : start, end, input, 0, sure, set, exempt, &shape) || (shape & 1 && (shape & 6 || any)) || + (shape & 8 && shape & 6)) /* a filter's appends: no other append or set beside them */ 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), nor walked */ - 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; - if (walk && appended >> (loop->arg >> 8 & 0xFF) & 1) return false; + /* A list appended to is read nowhere else in the block (an earlier round's item not there yet), nor walked; one a + * filter's code alone adds to (grown) is read only as no earlier round's item would change: its count in the guard, + * its first, an item by its place when it has no key (added at its end), a field by key (found at once: not there, + * the batch gives its rounds back), and has only in its filter (asked again at the end). */ + uint32_t appended = 0, grown = 0, tested = 0, body = 0, each = 0, nest = 0; /* nest: a nested LOOP's end */ 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; + const PureOp *op = &m->code[at]; + uint32_t filtered = filter_body(m, at, end); + if (each && at >= each + 1 + (m->code[each].arg >> 16)) each = 0; + if (filtered) { /* in the block or in a walk its items' rounds spread (not a nested LOOP's), one a list */ + if (at < nest || (each && !walk_spreads(m, start, end, each, reduced)) || op->arg >= 32 || tested >> op->arg & 1) return false; + body = filtered, tested |= UINT32_C(1) << op->arg; + } + if (op->op == PURE_OP_EACH) each = at; + if (op->op == PURE_OP_LOOP) nest = at + 1 + (op->arg >> 16); + if (op->op == PURE_OP_LIST_APPEND) { + if (op->arg >= 32) return false; + if (at < body) grown |= UINT32_C(1) << op->arg; + else appended |= UINT32_C(1) << op->arg; + } + } + if (grown & appended || (walk && (appended | grown) >> (loop->arg >> 8 & 0xFF) & 1)) return false; + body = 0; + for (uint32_t at = start; at < end; ++at) { + const PureOp *op = &m->code[at]; + uint32_t filtered = filter_body(m, at, end); + int list = read_list(op); + if (filtered) body = filtered; + if (list < 0 || list >= 32 || !((appended | grown) >> list & 1)) continue; + if (appended >> list & 1) return false; + bool keyed = m->state_key[list] != 0; + switch (op->op) { + case PURE_OP_LIST_COUNT: if (!(*guard && at == start + 1)) return false; break; + case PURE_OP_INDEX_FIRST: break; + case PURE_OP_ITEM_GET: + if (keyed && !(at > start && m->code[at - 1].op == PURE_OP_LIST_FIND && m->code[at - 1].arg == (uint32_t)list)) return false; + break; + case PURE_OP_LIST_FIND: if (at + 1 >= end || m->code[at + 1].op != PURE_OP_ITEM_GET || list_of(m->code[at + 1].arg) != (uint32_t)list) return false; break; + case PURE_OP_LIST_HAS: if (!filtered || !(tested >> list & 1)) return false; break; + default: return false; + } + } + for (uint32_t at = start, b = 0, t = 0; at < end; ++at) { /* a tested list: added to in its filter's code alone */ + uint32_t filtered = filter_body(m, at, end); + if (filtered) b = filtered, t = m->code[at].arg; + if (m->code[at].op == PURE_OP_LIST_APPEND && tested >> (m->code[at].arg & 31) & 1 && !(at < b && m->code[at].arg == t)) return false; } return true; } @@ -894,9 +956,11 @@ PURE_COLD static void plan_folds(PureModule *m, PureBlock block, int input) { (op->plan = (op->plan & PLAN_SPREAD) | (loop_batches(m, pc, input, recurs, &guard, reduced) ? PLAN_BATCH | (guard ? PLAN_GUARD : 0) : 0)) & PLAN_BATCH) { m->batches = true; - for (uint32_t at = pc + 1; at < pc + 1 + (op->arg >> 16); ++at) /* its walks whose items may run apart */ + for (uint32_t at = pc + 1; at < pc + 1 + (op->arg >> 16); ++at) { /* its walks whose items may run apart; its filters */ if (m->code[at].op == PURE_OP_EACH && walk_spreads(m, pc + 1, pc + 1 + (op->arg >> 16), at, reduced)) m->code[at].plan |= PLAN_SPREAD; + if (filter_body(m, at, pc + 1 + (op->arg >> 16))) m->code[at].plan |= PLAN_FILTER; + } for (uint32_t at = pc + 1; at < pc + 1 + (op->arg >> 16); ++at) /* a recurrence's sets, in the loop's own block */ if (m->code[at].op == PURE_OP_LOOP || m->code[at].op == PURE_OP_EACH) at += m->code[at].arg >> 16; else if (m->code[at].op == PURE_OP_LOCAL_STORE && recurs[(m->code[at].arg & 0xFF) / 64] >> (m->code[at].arg & 63) & 1) @@ -1557,7 +1621,7 @@ PURE_COLD PureResult pure_module_load(const void *bytes, size_t size, PureModule PureOp *op = &out->code[pc]; if (op->op == PURE_OP_ITEM_GET || op->op == PURE_OP_ITEM_SET) op->plan = text_field(out, list_of(op->arg), item_field(op->arg)); else if (op->op == PURE_OP_LIST_FIND || op->op == PURE_OP_LIST_HAS) - op->plan = text_field(out, op->arg, out->state_key[op->arg] - 1u); + op->plan = (op->plan & PLAN_FILTER) | text_field(out, op->arg, out->state_key[op->arg] - 1u); } /* A FIND whose index the next instruction, an ITEM_GET of its list, takes * at once (no jump lands between them) reads the field by key, as one diff --git a/tests/check.py b/tests/check.py index 731c5f8..ddb4a7a 100644 --- a/tests/check.py +++ b/tests/check.py @@ -1390,6 +1390,96 @@ on go { } """ +# Filters in a batch's block (GRAPH_SOURCE): a graph built in the handler +# (an edge added if not there yet, keys from a recurrence), searched breadth +# first twice (a count's loop to the queue's count, each node's edges +# walked and spread, if !seen.has(e.to) { seen.add; queue.append }: asked +# as the batch opens and again in the rounds' order at its end), the depths +# summed by a walk; marks whose filter asks one key and adds another (one +# there already: EXISTS at its round); under work budgets that end within a +# batch and memory budgets that end in the filters' adds. +GRAPH_SOURCE = """record Edge { + link = 0 + to = 0 +} +record Seen { + node = 0 + depth = 0 +} +record Step { + node = 0 +} +state reached = 0 +state hops = -1 +state far = 0 +state marked = 0 +input go = message { + n: u32 + m: u32 + from: u32 + q: u32 +} +on go { + let edges: list of Edge by link + let seen: list of Seen by node + let queue: list of Step + let again: list of Seen by node + let next: list of Step + let marks: list of Seen by node + let order: list of Step + let x = 7 + for k in 0 .. go.m { + x = x * 48271 - floor(x * 48271 / 2147483647) * 2147483647 + let a = x - floor(x / go.n) * go.n + let b = floor(x / 1000) - floor(x / 1000 / go.n) * go.n + if !edges.has(a * 1048576 + b) { edges.add(Edge { link: a * 1048576 + b, to: b }) } + } + queue.append(Step { node: go.from }) + seen.add(Seen { node: go.from }) + for i in 0 .. go.n { + if i >= queue.count { break } + let q = queue.first + i + for e in edges[q.node * 1048576 .. q.node * 1048576 + 1048576] { + if !seen.has(e.to) { + seen.add(Seen { node: e.to, depth: seen[q.node].depth + 1 }) + queue.append(Step { node: e.to }) + } + } + } + next.append(Step { node: go.q }) + again.add(Seen { node: go.q, depth: 1 }) + for i in 0 .. go.n * 2 { + if i >= next.count { break } + let q = next.first + i + for e in edges[q.node * 1048576 + 1 .. q.node * 1048576 + 1048576] { + if !again.has(e.to) { + again.add(Seen { node: e.to, depth: again[q.node].depth * 2 + 0.5 }) + next.append(Step { node: e.to }) + } + } + } + let total = 0 + for s in seen { + total = total + s.depth * (s.node + 1) + } + for s in again { + total = total + s.depth + } + reached = seen.count * 1000 + again.count + hops = if seen.has(go.q) { seen[go.q].depth } else { -1 } + far = total + for v in 0 .. go.n { + order.append(Step { node: v * 7 - floor(v * 7 / go.n) * go.n }) + } + for i in 0 .. go.n { + if i >= order.count { break } + let p = order.first + i + if !marks.has(p.node) { marks.add(Seen { node: p.node * 3 - floor(p.node * 3 / (go.n + go.q)) * (go.n + go.q), depth: i }) } + } + marked = marks.count +} +""" + # 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 @@ -1513,6 +1603,20 @@ def module_assumptions(aup, pure, work): assert outputs[name] == outputs['batches'], ('front', 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 'work' in faults, faults + source, program = work / 'graph.aup', work / 'graph.pmod' # filters, against the rounds one by one + source.write_text(GRAPH_SOURCE) + compile_module(aup, source, program) + given = ''.join(f'message go {struct.pack("<IIII", n, m, a, q).hex()}\nwork\n' for n, m, a, q in + ((3000, 20000, 1, 7), (3000, 20000, 1, 0), (300, 1200, 0, 5), (2000, 3000, 7, 11), (40, 200, 3, 1))) + given += ''.join(f'budget {b}\n' + given for b in (200011, 900001)) + 'budget 0\n' + given += ''.join(f'memory {b}\n' + given for b in (150000, 600000)) + 'memory 0\n' + outputs = {name: subprocess.run([str(pure / 'build/pure-run'), str(program)], input=given, capture_output=True, text=True, + timeout=300, 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'], ('graph', 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 'work' 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) @@ -1654,6 +1758,9 @@ def tier_program(rng): lines.append(f'{pad} d.add(Cell {{ at: {rng.choice([name, f"{name} * 3", key(2, inner)])}, v: {number(1, inner)} }})') elif pick < 0.5: lines.append(f'{pad} t[{at()}].v = {number(1, inner)}') + elif pick < 0.55: # a filter at the batch's own level: d asked, then added to + k = rng.choice([name, f'{name} * 3', f'w{name}', key(2, inner)]) + lines.append(f'{pad} if !d.has({k}) {{ d.add(Cell {{ at: {k}, v: {number(1, inner)} }}) }}') elif pick < 0.6: # a sum into a cell (its rows read and set by the rounds in turn) cell = at() lines.append(f'{pad} t[{cell}].v = t[{cell}].v + {number(1, inner)}') @@ -1737,7 +1844,7 @@ def tier_program(rng): f'g[{name} .. {name}]', f'h[5000000 + {name} .. 5000100]', f'f[{name} .. 9]']) inner = loops + [name, w, 'c'] lines.append(f'{pad} for c in {walked} {{') - own = False + own = filtered = False for _ in range(rng.randint(1, 3)): pick = rng.random() if pick < 0.3: @@ -1748,8 +1855,17 @@ def tier_program(rng): own = True lines.append(f'{pad} let o{name} = {number(2, inner)}') lines.append(f'{pad} rc = rc + o{name}') - elif pick < 0.85: + elif pick < 0.75: lines.append(f'{pad} if {cond(1, inner)} {{ ra = ra + {rng.choice(["c.v", w])} }}') + elif pick < 0.92 and not filtered: # a filter: d asked, then added to (its key, or one that may be there) + filtered = True + k = rng.choice(['c.v', 'c.at', f'floor(c.v / 2)', f'c.v + {w}']) + added = k if rng.random() < 0.8 else f'{k} + 1' + lines.append(f'{pad} if !d.has({k}) {{') + lines.append(f'{pad} d.add(Cell {{ at: {added}, v: {rng.choice([w, "c.at", number(2, inner)])} }})') + if rng.random() < 0.5: + lines.append(f'{pad} e.append(Cell {{ at: {k}, v: d[{rng.choice(["c.v", "0", w])}].v }})') + lines.append(f'{pad} }}') else: lines.append(f'{pad} if {cond(1, inner)} {{ break }}') lines.append(f'{pad} }}') |