diff options
| -rw-r--r-- | core/module-batch.c | 77 | ||||
| -rw-r--r-- | tests/check.py | 70 |
2 files changed, 145 insertions, 2 deletions
diff --git a/core/module-batch.c b/core/module-batch.c index 0d02e45..ba19850 100644 --- a/core/module-batch.c +++ b/core/module-batch.c @@ -3143,6 +3143,77 @@ static int append_items(Batch *b, Frame *frame, unsigned n, uint64_t per) { static int batch_run(Frame *frame, Loop *loop, unsigned *done); +/* The sums into cells from round 0 at once (docs/design/025, stage 5): the + * round's sums in turn, a one- or four-byte cell's in integers while its e + * is a whole number below 2^32 and its sum stays in its bytes, an + * eight-byte cell's as a number while it is finite (the numbers the + * executor's adds would make, with no fault and no widening). The rows and + * the e's in four bytes first (vector code), up to the first round whose e + * is not one. Where one is not so, *round and *sum: the executor's way goes + * on from there. */ +static void sums_in_words(Batch *b, const double *const *rows, const double *const *es, const uint32_t *const *masks, + const PureCells *cells, unsigned *round, unsigned *sum) { + unsigned n = b->n, count = b->scatter_count, stop = n, spares = b->spare_count, buffers = b->buffers; + bool reverse = b->reverse, failed = b->failed; + uint32_t *row32[SUMS], *e32[SUMS]; + *round = 0, *sum = 0; + for (unsigned i = 0; i < count; ++i) { + uint8_t width = cells[i].width[b->scatters[i].field]; + if ((width != 8 && width != 4 && width != 1) || !(row32[i] = fresh(b)) || (width != 8 && !(e32[i] = fresh(b)))) { + b->spare_count = spares, b->buffers = buffers, b->failed = failed; /* (the rounds one by one) */ + return; + } + const double *at = rows[i], *e = es[i]; + for (unsigned l = 0; l < n; ++l) row32[i][l] = at[l] >= 0 && at[l] < 4294967296.0 ? (uint32_t)at[l] : 0; + if (width == 8) continue; + uint32_t bad = 0, *out = e32[i]; + for (unsigned l = 0; l < n; ++l) { + uint32_t u = e[l] >= 0 && e[l] < 4294967296.0 ? (uint32_t)e[l] : 0; + out[l] = u, bad |= (double)u != e[l]; + } + for (unsigned l = 0; bad && l < n; ++l) /* the first round (of those it runs in) whose e is not a whole number below 2^32 */ + if ((!masks[i] || masks[i][l]) && (double)out[l] != e[l]) { + unsigned r = reverse ? n - 1 - l : l; + stop = r < stop ? r : stop; + } + } + for (unsigned r = 0; r < stop; ++r) { + unsigned l = reverse ? n - 1 - r : r; + for (unsigned i = 0; i < count; ++i) { + if (masks[i] && !masks[i][l]) continue; + uint8_t width = cells[i].width[b->scatters[i].field]; + void *column = cells[i].columns[b->scatters[i].field]; + uint32_t row = row32[i][l]; + if (width == 8) { + double v; + memcpy(&v, (const double *)column + row, sizeof(v)); + v += es[i][l]; + if (!isfinite(v)) { + stop = r, *sum = i; + break; + } + memcpy((double *)column + row, &v, sizeof(v)); + } else if (width == 4) { + uint64_t v = (uint64_t)((uint32_t *)column)[row] + e32[i][l]; + if (v >> 32) { + stop = r, *sum = i; + break; + } + ((uint32_t *)column)[row] = (uint32_t)v; + } else { + uint32_t v = ((uint8_t *)column)[row] + e32[i][l]; + if (v > 255) { + stop = r, *sum = i; + break; + } + ((uint8_t *)column)[row] = (uint8_t)v; + } + } + } + *round = stop; + b->spare_count = spares, b->buffers = buffers; /* its four bytes let go */ +} + /* 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); @@ -3234,9 +3305,11 @@ static int batch_run(Frame *frame, Loop *loop, unsigned *done) { return STEP_NEXT; } } - for (unsigned r = 0; r < n; ++r) { + unsigned from_round, from_sum; + sums_in_words(b, rows, es, masks, cells, &from_round, &from_sum); + for (unsigned r = from_round; r < n; ++r) { /* the rest round by round, as the executor adds and sets */ unsigned l = round_of(b, r); - for (unsigned i = 0; i < b->scatter_count; ++i) { + for (unsigned i = r == from_round ? from_sum : 0; i < b->scatter_count; ++i) { const Scatter *sc = &b->scatters[i]; if (masks[i] && !masks[i][l]) continue; size_t row = (size_t)rows[i][l]; diff --git a/tests/check.py b/tests/check.py index 1d98f05..91ac114 100644 --- a/tests/check.py +++ b/tests/check.py @@ -1227,6 +1227,63 @@ on go { """ +# Sums into cells in whole numbers (SUMS_SOURCE, docs/design/025's stage 5): +# sixteen rows met again and again, a one-byte column widened to four in the +# middle of a batch, a four-byte one past 2^32 (to eight), and a masked +# fraction where the whole numbers stop. +SUMS_SOURCE = """record Total { + region = 0 + amount = 0 +} +state totals = 0 +state tinies = 0 +state bigs = 0 +state mixed = 0 +input go = message { + n: u32 + q: u32 +} +on go { + let t: list of Total by region + let u: list of Total by region + let w: list of Total by region + let m: list of Total by region + for r in 0 .. 16 { + t.add(Total { region: r }) + u.add(Total { region: r }) + w.add(Total { region: r }) + m.add(Total { region: r }) + } + for o in 0 .. go.n { + let region = o - floor(o / 16) * 16 + t[region].amount = t[region].amount + (o - floor(o / 1000) * 1000) + u[region].amount = u[region].amount + 1 + w[region].amount = w[region].amount + 1500000000 + if o - floor(o / go.q) * go.q == 0 { m[region].amount = m[region].amount + 0.5 } + } + let a = 0 + let b = 0 + let c = 0 + let d = 0 + for x in t { + a = a + x.amount * (x.region + 1) + } + for x in u { + b = b + x.amount * (x.region + 1) + } + for x in w { + c = c + x.amount + } + for x in m { + d = d + x.amount * (x.region + 1) + } + totals = a + tinies = b + bigs = c + mixed = d +} +""" + # Columns read in place (VIEW_SOURCE, docs/design/025's stage 4): a local # holding a row's field while the batch sets that field (its last value # written back after the sets), a set whose value is the same column three @@ -1758,6 +1815,19 @@ def module_assumptions(aup, pure, work): assert outputs[name] == outputs['batches'], ('reduce', name, next((a, b) for a, b in zip(outputs[name].splitlines(), outputs['batches'].splitlines()) if a != b)) faults = [json.loads(line)['fault'] for line in outputs['batches'].splitlines() if line.startswith('{"states"')] assert 'none' in faults and 'work' in faults, faults + source, program = work / 'sums.aup', work / 'sums.pmod' # sums into cells in whole numbers, against the rounds one by one + source.write_text(SUMS_SOURCE) + compile_module(aup, source, program) + given = ''.join(f'message go {struct.pack("<II", n, q).hex()}\nwork\n' for n, q in + ((3000, 7), (3000, 1), (5000, 70000), (100, 3), (40, 2), (2, 3))) + given += 'budget 30011\n' + given + 'budget 0\n' + outputs = {name: subprocess.run([str(pure / 'build/pure-run'), str(program)], input=given, capture_output=True, text=True, + timeout=120, env=dict(os.environ, **env)).stdout + for name, env in (('batches', {'PURE_JIT': '0'}), ('rounds', {'PURE_JIT': '0', 'PURE_BATCH': '0'}), ('all', {'PURE_JIT': 'all'}))} + for name in ('rounds', 'all'): + assert outputs[name] == outputs['batches'], ('sums', name, next((a, b) for a, b in zip(outputs[name].splitlines(), outputs['batches'].splitlines()) if a != b)) + faults = [json.loads(line)['fault'] for line in outputs['batches'].splitlines() if line.startswith('{"states"')] + assert 'none' in faults and 'work' in faults, faults source, program = work / 'view.aup', work / 'view.pmod' # columns read in place, against the rounds one by one source.write_text(VIEW_SOURCE) compile_module(aup, source, program) |