Batch executor stage 2 (024): a loop in the batch's block runs its rounds for every lane together (round t for the lanes with more than t rounds that did not break out; BREAK leaves it lane by lane; its slot as each round sets it), and x = x op e (op +, min, max; x read no other way) is a reduction: each lane's own from op's identity (-0 a sum's), folded in the rounds' order at the batch's end (a sum's lanes of several whole numbers only, below 2^53); nested rounds' work counted for each lane and checked before anything changes; lanes and masks no holder keeps freed between nested rounds. Planned at load: a local read before set in its loop's scope, or not a reduction's, keeps the loop round by round. PURE_JIT=0: roots 130.5 to 16.8 ms, matmul 914.7 to 68.2. module_tier makes nested loops with breaks and reductions (seeds 1, 2, 3, 23: 549 programs alike); NESTED_SOURCE

4 files changed, 450 insertions, 47 deletions

diff --git a/core/module-batch.c b/core/module-batch.c
index dd9df1a..c4cd15f 100644
--- a/core/module-batch.c
+++ b/core/module-batch.c
@@ -25,9 +25,9 @@
 #include "module-run.h"
 #include <stdlib.h>
 
-enum { BATCH = 1024, BUFFERS = 16, MASKS = 12, SPLITS = 8, READS = 16, STORES = 8, LOCALS = 256, LEAST = 16, FAILURES = 3 };
+enum { BATCH = 1024, BUFFERS = 24, MASKS = 32, SPLITS = 8, READS = 16, STORES = 8, REDUCES = 8, LOCALS = 256, LEAST = 16, FAILURES = 3 };
 
-enum { UNIFORM, AFFINE, LANES };
+enum { UNIFORM, AFFINE, LANES, REDUCE }; /* REDUCE: x in x = x op e (value: its reduction) */
 typedef struct {
     uint8_t kind;
     bool u32;       /* lanes: whole numbers below 2^32 in four bytes each (else numbers) */
@@ -46,13 +46,20 @@ typedef struct { uint32_t list, field; bool known; Val rows; } Read;
 typedef struct { uint32_t list, field; Val rows, value; Mask *mask; } Store;
 /* 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'
+ * e (from op's identity), how many it had, whether one was not whole or a
+ * lane's past 2^53 (then a lane's two may not sum as the rounds would). */
+typedef struct { unsigned local; uint8_t op; double *partial; uint32_t *count; bool loose; double result; } Reduce;
+/* A loop the batch's block runs, its rounds together: the lanes that came
+ * to it, how many rounds each has, from what, and those that broke out. */
+typedef struct { Mask *entry; uint32_t *count, *broken; Val low; } Nest;
 
 typedef struct {
     Frame *frame;
     const PureModule *m;
     PureApp *app;
     Event *ev;
-    uint32_t slot;
+    uint32_t slot, end;         /* the loop's slot, and where its block ends */
     unsigned n;
     double first;               /* the loop's value in the batch's first round */
     bool reverse;               /* lane l runs round n - 1 - l (else round l) */
@@ -66,6 +73,13 @@ typedef struct {
     Val local[LOCALS];
     Mask *active;               /* the lanes running here (NULL: every lane) */
     Mask mask[MASKS];
+    Mask *mask_free[MASKS];     /* masks no lane set holds any more */
+    unsigned mask_free_count;
+    Reduce reduce[REDUCES];
+    unsigned reduce_count;
+    Nest nest;                  /* the nested loop running (nest.count NULL: none) */
+    uint64_t work;              /* the nested loops' rounds' work */
+    bool items;                 /* the block sets items: its reads are kept to check against them */
     bool none;                  /* no lane runs here */
     Split splits[SPLITS];
     unsigned split_count, buffers, masks, read_count, store_count, spare_count;
@@ -107,11 +121,11 @@ static void release(Batch *b, const Val *v) {
 }
 
 static Mask *new_mask(Batch *b, const uint32_t *lanes, const Mask *parent, unsigned count) {
-    if (b->masks == MASKS) {
+    if (!b->mask_free_count && b->masks == MASKS) {
         b->failed = true;
         return NULL;
     }
-    Mask *m = &b->mask[b->masks++];
+    Mask *m = b->mask_free_count ? b->mask_free[--b->mask_free_count] : &b->mask[b->masks++];
     *m = (Mask){lanes, parent, NULL, count};
     return m;
 }
@@ -318,8 +332,15 @@ static void number_lanes(Batch *b, uint8_t op, Val *x, const Val *y) {
     *x = lanes_val(out, false);
 }
 
+static void reduce_step(Batch *b, Reduce *r, const Val *e);
+
 static void binary(Batch *b, uint8_t op) {
     Val y = b->stack[--b->sp], *x = &b->stack[b->sp - 1];
+    if (x->kind == REDUCE) { /* x = x op e: e into each running lane's */
+        reduce_step(b, &b->reduce[(unsigned)x->value], &y);
+        release(b, &y);
+        return;
+    }
     if (x->kind == UNIFORM && y.kind == UNIFORM) {
         double value = 0;
         if (!number_op(op, x->value, y.value, &value)) b->failed = true; /* every running lane faults */
@@ -468,6 +489,7 @@ static bool stored(const Batch *b, uint32_t list, uint32_t field) {
 }
 
 static void read_of(Batch *b, uint32_t list, uint32_t field, bool known, const Val *rows) {
+    if (!b->items) return; /* nothing set: no read to check */
     if (b->read_count == READS) {
         b->failed = true;
         return;
@@ -615,6 +637,7 @@ static void item_set(Batch *b, uint32_t list, uint32_t field) {
 static void local_store(Batch *b, uint32_t arg) {
     unsigned at = arg & 0xFF;
     Val v = b->stack[--b->sp];
+    if (v.kind == REDUCE) return; /* x = x op e: in its lanes already */
     if (at == b->slot) {
         b->failed = true;
         return;
@@ -637,9 +660,17 @@ static void local_store(Batch *b, uint32_t arg) {
     if (b->state[at] == 2 && !(b->written[at] = union_of(b, b->written[at], b->active))) b->state[at] = 1;
 }
 
-static void local_load(Batch *b, uint32_t arg) {
+static Reduce *reduction(Batch *b, unsigned local, uint8_t op);
+
+static void local_load(Batch *b, uint32_t arg, uint32_t pc) {
     unsigned at = arg & 0xFF;
     Val v;
+    uint32_t by;
+    if (b->sets[at / 64] >> (at % 64) & 1 && !b->state[at] && at != b->slot && pure_batch_reduces(b->m, pc, b->end, &by)) {
+        Reduce *r = reduction(b, at, b->m->code[by].op); /* not set yet: an earlier round's, as x = x op e reads it */
+        if (r) b->stack[b->sp++] = (Val){.kind = REDUCE, .value = (double)(r - b->reduce)};
+        return;
+    }
     if (at == b->slot) {
         double start = b->reverse ? b->first + (b->n - 1) : b->first, step = b->reverse ? -1 : 1;
         if (!affine_of(b, start, step, &v)) {
@@ -714,8 +745,79 @@ static void branch(Batch *b, uint32_t pc) {
     else b->none = true;
 }
 
+/* ---- Reductions */
+
+bool pure_batch_reduces(const PureModule *m, uint32_t pc, uint32_t end, uint32_t *at) {
+    unsigned h = m->height[pc], x = m->code[pc].arg & 0xFF;
+    for (uint32_t i = pc + 1; i + 1 < end; ++i) {
+        const PureOp *op = &m->code[i];
+        if (op->op == PURE_OP_JZ || op->op == PURE_OP_JMP || op->op == PURE_OP_LOOP || op->op == PURE_OP_BREAK || m->height[i] < h + 1)
+            return false;
+        if (m->height[i + 1] == h + 1) { /* x taken: by op, with e */
+            *at = i;
+            return m->height[i] == h + 2 && (op->op == PURE_OP_ADD || op->op == PURE_OP_MIN || op->op == PURE_OP_MAX) &&
+                   m->code[i + 1].op == PURE_OP_LOCAL_STORE && (m->code[i + 1].arg & 0xFF) == x;
+        }
+    }
+    return false;
+}
+
+/* Local's reduction by op, begun from op's identity in every lane (-0 a sum's: -0 + e is e). */
+static Reduce *reduction(Batch *b, unsigned local, uint8_t op) {
+    for (unsigned r = 0; r < b->reduce_count; ++r)
+        if (b->reduce[r].local == local) {
+            if (b->reduce[r].op == op) return &b->reduce[r];
+            b->failed = true; /* x = x + e here, x = max(x, e) there: not one reduction */
+            return NULL;
+        }
+    double *partial = b->reduce_count < REDUCES ? fresh(b) : NULL;
+    uint32_t *count = partial ? fresh(b) : NULL;
+    if (!count) {
+        b->failed = true;
+        return NULL;
+    }
+    double identity = op == PURE_OP_ADD ? -0.0 : op == PURE_OP_MIN ? INFINITY : -INFINITY;
+    for (unsigned l = 0; l < b->n; ++l) partial[l] = identity, count[l] = 0;
+    Reduce *r = &b->reduce[b->reduce_count++];
+    *r = (Reduce){local, op, partial, count, false, 0};
+    return r;
+}
+
+/* x = x op e in each running lane: e into its partial. */
+static void reduce_step(Batch *b, Reduce *r, const Val *e) {
+    const uint32_t *run = running(b);
+    double *p = r->partial;
+    uint32_t *c = r->count;
+    unsigned n = b->n;
+    if (e->kind == UNIFORM && r->op == PURE_OP_ADD) { /* a count's +1, say */
+        double v = e->value;
+        if (!run) for (unsigned l = 0; l < n; ++l) p[l] += v, c[l] += 1;
+        else for (unsigned l = 0; l < n; ++l) p[l] = run[l] ? p[l] + v : p[l], c[l] += run[l];
+        r->loose |= v != floor(v);
+    } else {
+        const double *es = numbers(b, e);
+        if (!es) return;
+        uint8_t fraction = 0;
+        for (unsigned l = 0; l < n; ++l) {
+            if (run && !run[l]) continue;
+            double v = es[l];
+            p[l] = r->op == PURE_OP_ADD ? p[l] + v : r->op == PURE_OP_MIN ? minimum(p[l], v) : maximum(p[l], v);
+            ++c[l], fraction |= v != floor(v);
+        }
+        if (es != e->lane) b->spare[b->spare_count++] = (void *)es;
+        r->loose |= r->op == PURE_OP_ADD && fraction;
+    }
+    if (r->op == PURE_OP_ADD) { /* past 2^53, a lane's sum of whole numbers may round */
+        uint8_t far = 0;
+        for (unsigned l = 0; l < n; ++l) far |= !(fabs(p[l]) <= 0x1p53);
+        r->loose |= far;
+    }
+}
+
 /* ---- The pass */
 
+static void nested(Batch *b, const PureOp *op, uint32_t pc);
+
 /* 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;
@@ -763,7 +865,7 @@ static bool pass(Batch *b, uint32_t start, uint32_t end) {
             for (unsigned d = b->sp - 1 - arg; d + 1 < b->sp; ++d) release(b, &b->stack[d]);
             b->stack[b->sp - 1 - arg] = b->stack[b->sp - 1], b->sp -= arg;
             break;
-        case PURE_OP_LOCAL_LOAD: local_load(b, arg); break;
+        case PURE_OP_LOCAL_LOAD: local_load(b, arg, pc - 1); break;
         case PURE_OP_LOCAL_STORE: local_store(b, arg); break;
         case PURE_OP_NEG: case PURE_OP_NOT: case PURE_OP_FLOOR: unary(b, op->op); break;
         case PURE_OP_CLAMP: clamp(b); break;
@@ -775,6 +877,19 @@ static bool pass(Batch *b, uint32_t start, uint32_t end) {
             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_LOOP: nested(b, op, pc - 1), pc += arg >> 16; break;
+        case PURE_OP_BREAK: { /* the running lanes leave the nested loop */
+            const uint32_t *run = running(b);
+            uint32_t *broken = b->nest.broken;
+            if (!broken) {
+                b->failed = true; /* none: no batch form (the load plans none such) */
+                break;
+            }
+            if (!run) for (unsigned l = 0; l < b->n; ++l) broken[l] = 1;
+            else for (unsigned l = 0; l < b->n; ++l) broken[l] |= run[l];
+            b->none = true;
+            break;
+        }
         case PURE_OP_JMP: send(b, arg, b->active), b->none = true; break;
         case PURE_OP_ADD: case PURE_OP_SUB: case PURE_OP_MUL: case PURE_OP_DIV: case PURE_OP_MOD: case PURE_OP_MIN:
         case PURE_OP_MAX: case PURE_OP_LT: case PURE_OP_LE: case PURE_OP_GT: case PURE_OP_GE: case PURE_OP_EQ:
@@ -785,6 +900,112 @@ static bool pass(Batch *b, uint32_t start, uint32_t end) {
     return !b->failed;
 }
 
+
+/* ---- Nested loops: their rounds together */
+
+static void mark_lanes(bool *kept, const void *lanes) {
+    uintptr_t at = (uintptr_t)lanes, first = (uintptr_t)&buffer[0][0], past = (uintptr_t)&buffer[BUFFERS - 1][BATCH + 24];
+    if (at >= first && at < past) kept[(at - first) / sizeof(buffer[0])] = true;
+}
+
+static void mark_mask(Batch *b, bool *kept, bool *kept_mask, const Mask *m) {
+    for (; m && !kept_mask[m - b->mask]; m = m->parent) {
+        kept_mask[m - b->mask] = true;
+        mark_lanes(kept, m->lanes);
+        if (m->sibling) mark_mask(b, kept, kept_mask, m->sibling);
+    }
+}
+
+static void mark_val(bool *kept, const Val *v) {
+    if (v->kind == LANES) mark_lanes(kept, v->lane);
+}
+
+/* Between a nested loop's rounds (the stack empty): the lanes and masks
+ * nothing holds (no local, reduction, nest, read, set or running lanes) free again. */
+static void sweep(Batch *b) {
+    bool kept[BUFFERS] = {false}, kept_mask[MASKS] = {false};
+    for (unsigned w = 0; w < LOCALS / 64; ++w)
+        for (uint64_t bits = b->sets[w]; bits; bits &= bits - 1) {
+            unsigned at = w * 64 + (unsigned)__builtin_ctzll(bits);
+            if (b->state[at]) mark_val(kept, &b->local[at]);
+            if (b->state[at] == 2) mark_mask(b, kept, kept_mask, b->written[at]);
+        }
+    mark_mask(b, kept, kept_mask, b->active);
+    mark_mask(b, kept, kept_mask, b->nest.entry);
+    mark_lanes(kept, b->nest.count), mark_lanes(kept, b->nest.broken), mark_val(kept, &b->nest.low);
+    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)
+        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]);
+    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];
+    for (unsigned i = 0; i < b->masks; ++i)
+        if (!kept_mask[i]) b->mask_free[b->mask_free_count++] = &b->mask[i];
+}
+
+/* A nested loop's slot in round t: low + t in the lanes running it (low
+ * itself in round 0, as the LOOP sets it), the others' as it was. */
+static void set_slot(Batch *b, unsigned slot, const Val *low, uint32_t t, const uint32_t *live, unsigned alive) {
+    Val v;
+    if (alive == b->n && low->kind == UNIFORM) v = uniform(t ? low->value + t : low->value, 0);
+    else if (!(alive == b->n && low->kind == AFFINE && affine_of(b, low->value + t, low->step, &v))) {
+        double *out = fresh(b);
+        if (!out) return;
+        for (unsigned l = 0; l < b->n; ++l)
+            out[l] = live[l] ? (t ? lane_of(low, l) + t : lane_of(low, l))
+                   : b->state[slot] ? lane_of(&b->local[slot], l) : b->ev->locals[slot].num;
+        v = lanes_val(out, false);
+        v.temp = false;
+    }
+    b->local[slot] = v, b->state[slot] = 1;
+}
+
+/* The LOOP at pc in the batch's block: its rounds for every lane running
+ * here together, round t for the lanes with more than t rounds that did not
+ * break out, its work counted for each (the batch's end checks it). */
+static void nested(Batch *b, const PureOp *op, uint32_t pc) {
+    Val high = b->stack[--b->sp], low = b->stack[--b->sp];
+    uint32_t start = pc + 1, end = start + (op->arg >> 16);
+    unsigned slot = op->arg & 0xFF, n = b->n;
+    uint32_t *count = fresh(b), *broken = fresh(b);
+    if (!count || !broken) return;
+    const uint32_t *entry = running(b);
+    uint8_t bad[BATCH];
+    for (unsigned l = 0; l < n; ++l) { /* as the LOOP opens: whole ends within 2^53 */
+        double lo = lane_of(&low, l), hi = lane_of(&high, l);
+        bad[l] = !(fabs(lo) <= 0x1p53 && fabs(hi) <= 0x1p53 && lo == floor(lo) && hi == floor(hi)) || hi - lo > 0x1p31;
+        count[l] = bad[l] || (entry && !entry[l]) || hi <= lo ? 0 : (uint32_t)(hi - lo), broken[l] = 0;
+    }
+    release(b, &high);
+    if (any_running(b, bad)) {
+        b->failed = true;
+        return;
+    }
+    if (low.kind == LANES) low.temp = false; /* the nest keeps it */
+    Mask *outer = b->active;
+    Nest saved = b->nest;
+    b->nest = (Nest){outer, count, broken, low};
+    uint64_t per = (uint64_t)(end - start) + 1;
+    for (uint32_t t = 0; !b->failed; ++t) {
+        uint32_t *live = fresh(b);
+        if (!live) break;
+        unsigned alive = 0;
+        for (unsigned l = 0; l < n; ++l) live[l] = (t < count[l]) & (broken[l] ^ 1), alive += live[l];
+        if (!alive) break;
+        b->work += alive * per; /* each lane's round: its block and its LOOP, as written */
+        set_slot(b, slot, &low, t, live, alive);
+        b->active = alive == n ? NULL : new_mask(b, live, outer, alive);
+        b->none = false;
+        if (!pass(b, start, end)) break;
+        arrive(b, end);
+        sweep(b);
+    }
+    b->nest = saved;
+    b->active = outer, b->none = false;
+}
+
 /* ---- The end: the sets, if no read of the batch would have seen one */
 
 /* A lane's round in the batch. */
@@ -869,9 +1090,30 @@ static uint8_t width_needed(double v) {
     return v < 256 ? 1 : v < 4294967296.0 ? 4 : 8;
 }
 
+/* Each reduction's value after the batch's rounds, as they would leave it
+ * (false: they would fault, or a lane's sum may not round as theirs). */
+static bool fold(Batch *b) {
+    for (unsigned i = 0; i < b->reduce_count; ++i) {
+        Reduce *r = &b->reduce[i];
+        double c = b->ev->locals[r->local].num; /* as the batch began */
+        bool several = false;
+        for (unsigned l = 0; l < b->n; ++l) several |= r->count[l] > 1;
+        if (r->op == PURE_OP_ADD && several && (r->loose || !(c == floor(c) && fabs(c) <= 0x1p53))) return false;
+        for (unsigned k = 0; k < b->n; ++k) { /* in the rounds' order */
+            unsigned l = round_of(b, k);
+            if (!r->count[l]) continue;
+            c = r->op == PURE_OP_ADD ? c + r->partial[l] : r->op == PURE_OP_MIN ? minimum(c, r->partial[l]) : maximum(c, r->partial[l]);
+            if (!isfinite(c) || (several && fabs(c) > 0x1p53)) return false;
+        }
+        r->result = c;
+    }
+    return true;
+}
+
 static bool settle(Batch *b) {
     PureCells cells[STORES];
     const uint32_t *mask[STORES];
+    if (!fold(b)) return false;
     for (unsigned i = 0; i < b->store_count; ++i) {
         Store *s = &b->stores[i];
         PureList *list = b->app->lists[s->list];
@@ -948,16 +1190,20 @@ unsigned pure_batch_run(Frame *frame, Loop *loop) {
     Batch *b = &batch;
     b->frame = frame, b->m = m, b->app = app, b->ev = frame->ev, b->n = n, b->first = first;
     b->reverse = loop->reverse, b->backward = 0;
-    b->slot = (uint32_t)(loop->slot - frame->ev->locals);
-    b->failed = b->none = false, b->active = NULL;
+    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->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;
     memset(b->sets, 0, sizeof(b->sets));
-    for (uint32_t pc = loop->start; pc < loop->end; ++pc)
-        if (m->code[pc].op == PURE_OP_LOCAL_STORE) {
-            unsigned at = m->code[pc].arg & 0xFF;
+    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];
+        b->items |= op->op == PURE_OP_ITEM_SET;
+        if (op->op == PURE_OP_LOCAL_STORE || op->op == PURE_OP_LOOP) {
+            unsigned at = op->arg & 0xFF;
             b->sets[at / 64] |= UINT64_C(1) << (at % 64);
             b->state[at] = 0;
         }
+    }
     if (pass(b, loop->start, loop->end)) {
         arrive(b, loop->end);
         for (unsigned w = 0; w < LOCALS / 64; ++w) /* the lanes each local was set in, made before anything changes */
@@ -966,11 +1212,13 @@ unsigned pure_batch_run(Frame *frame, Loop *loop) {
                 if (b->state[at] == 2) mask_lanes(b, b->written[at]);
             }
     }
-    if (b->failed || !settle(b)) {
+    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;
     }
-    app->work += (n - 1) * per;
+    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 w = 0; w < LOCALS / 64; ++w) /* each local the block set: as the last round to set it left it */
         for (uint64_t bits = b->sets[w]; bits; bits &= bits - 1) {
             unsigned at = w * 64 + (unsigned)__builtin_ctzll(bits), last = n - 1;
diff --git a/core/module-run.h b/core/module-run.h
index 1e48072..3f8a9ca 100644
--- a/core/module-run.h
+++ b/core/module-run.h
@@ -162,6 +162,10 @@ int pure_run_rest(Frame *, uint32_t pc);
  * 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 *);
+/* 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. */
+bool pure_batch_reduces(const PureModule *, uint32_t pc, uint32_t end, uint32_t *at);
 
 /* Where a list keeps what compiled code reads of it (module-list.c owns the
  * layout and says where): its item count, its key set (the words, how many,
diff --git a/core/module.c b/core/module.c
index c6eb9be..3bd9948 100644
--- a/core/module.c
+++ b/core/module.c
@@ -686,43 +686,51 @@ PURE_COLD static bool block_has_lanes(const PureModule *m, PureBlock block, int
     return true;
 }
 
-/* A count's loop whose block has batch forms throughout runs a batch of its
- * rounds at a time (module-batch.c, docs/design/024): lane forms, a local
- * set (a number, an index or a bool) and read only after every round sets
- * it, a branch with nothing else on the stack to a place in the block, and
- * a number or bool set in an item (not its key). PURE_BATCH=0 in the
- * environment: none does (the executor alone, to measure it). */
-PURE_COLD static bool loop_batches(const PureModule *m, uint32_t pc, int input) {
-    static int off = -1;
-    if (off < 0) {
-        const char *batches = getenv("PURE_BATCH");
-        off = batches && !strcmp(batches, "0");
-    }
-    const PureOp *loop = &m->code[pc];
-    uint32_t start = pc + 1, end = start + (loop->arg >> 16), arms = start;
-    if (off || loop->arg & PURE_LOOP_INTEGER || start >= end || m->height[start]) return false;
-    uint64_t set[4] = {0}, sure[4] = {0}; /* the locals the block sets; set before here in every round */
-    for (uint32_t at = start; at < end; ++at)
-        if (m->code[at].op == PURE_OP_LOCAL_STORE) set[(m->code[at].arg & 0xFF) / 64] |= UINT64_C(1) << (m->code[at].arg & 63);
-    for (uint32_t at = start; at < end; ++at) { /* a local read before a round sets it: an earlier round's (no batch has it) */
-        const PureOp *op = &m->code[at];
-        unsigned local = op->arg & 0xFF;
-        if ((op->op == PURE_OP_JZ || op->op == PURE_OP_JMP) && op->arg > arms) arms = op->arg; /* code before arms: in a branch's way */
-        if (op->op == PURE_OP_LOCAL_STORE && at >= arms) sure[local / 64] |= UINT64_C(1) << (local % 64);
-        if (op->op == PURE_OP_LOCAL_LOAD && (set[local / 64] & ~sure[local / 64]) >> (local % 64) & 1) return false;
-    }
+/* 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
+ * it (sure: set for certain since the round began) or a reduction's
+ * (reduced: x = x op e alone), a branch with nothing else on the stack to a
+ * place in the block, a loop in it (one deep: its rounds run together, a
+ * BREAK leaving it), and a number or bool set in an item (not its key; not
+ * 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 */
     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);
         switch (op->op) {
-        case PURE_OP_LOCAL_STORE:
-            if (local_type(op->arg) == T_I64) return false;
+        case PURE_OP_LOOP: {
+            uint32_t inner = at + 1, stop = inner + (op->arg >> 16);
+            if (depth || op->arg & PURE_LOOP_INTEGER || stop > end || m->height[inner]) return false;
+            uint64_t scope[4];
+            memcpy(scope, sure, sizeof(scope));
+            scope[local / 64] |= bit; /* its slot, set as each round begins */
+            *shape |= 1;
+            if (!batch_block(m, inner, stop, input, depth + 1, scope, set, reduced, shape)) return false;
+            at = stop - 1;
+            break;
+        }
+        case PURE_OP_BREAK:
+            if (!depth) return false; /* a nested loop's alone */
             break;
         case PURE_OP_JZ: case PURE_OP_JMP:
             if (op->arg > end || m->height[at] != (op->op == PURE_OP_JZ)) return false;
+            if (op->arg > arms) arms = op->arg;
+            break;
+        case PURE_OP_LOCAL_STORE:
+            if (local_type(op->arg) == T_I64) return false;
+            if (at >= arms) sure[local / 64] |= bit;
+            break;
+        case PURE_OP_LOCAL_LOAD:
+            if (!has_lanes(m, op, input) || set[local / 64] & ~sure[local / 64] & ~reduced[local / 64] & bit) return false;
             break;
         case PURE_OP_ITEM_SET: {
             unsigned char type = field_type(m, list_of(op->arg), item_field(op->arg));
-            if ((type != T_NUM && type != T_BOOL) || item_field(op->arg) == m->state_key[list_of(op->arg)] - 1u) return false;
+            if (depth || (type != T_NUM && type != T_BOOL) || item_field(op->arg) == m->state_key[list_of(op->arg)] - 1u) return false;
+            *shape |= 2;
             break;
         }
         default:
@@ -732,6 +740,43 @@ PURE_COLD static bool loop_batches(const PureModule *m, uint32_t pc, int input)
     return true;
 }
 
+/* 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. */
+PURE_COLD static bool loop_batches(const PureModule *m, uint32_t pc, int input) {
+    static int off = -1;
+    if (off < 0) {
+        const char *batches = getenv("PURE_BATCH");
+        off = batches && !strcmp(batches, "0");
+    }
+    const PureOp *loop = &m->code[pc];
+    uint32_t start = pc + 1, end = start + (loop->arg >> 16);
+    if (off || loop->arg & PURE_LOOP_INTEGER || start >= end || m->height[start]) return false;
+    uint8_t *ends = calloc(end - start, 1); /* the sets that end x = x op e */
+    if (!ends) return false;
+    uint64_t set[4] = {0}, sure[4] = {0}, reduced[4] = {0}, plain[4] = {0}; /* plain: read or set otherwise */
+    for (uint32_t at = start; at < end; ++at) {
+        const PureOp *op = &m->code[at];
+        uint64_t bit = UINT64_C(1) << (op->arg & 63);
+        unsigned word = (op->arg & 0xFF) / 64;
+        uint32_t by;
+        if (op->op == PURE_OP_LOCAL_STORE || op->op == PURE_OP_LOOP) set[word] |= bit;
+        if (op->op == PURE_OP_LOCAL_LOAD && local_type(op->arg) == T_NUM && pure_batch_reduces(m, at, end, &by))
+            reduced[word] |= bit, ends[by + 1 - start] = 1;
+        else if (op->op == PURE_OP_LOCAL_LOAD)
+            plain[word] |= bit;
+    }
+    for (uint32_t at = start; at < end; ++at)
+        if (m->code[at].op == PURE_OP_LOCAL_STORE && !ends[at - start]) plain[(m->code[at].arg & 0xFF) / 64] |= UINT64_C(1) << (m->code[at].arg & 63);
+    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;
+}
+
 /* A fold in a verified block whose own block has lane forms throughout runs
  * a leaf of its list at a time; a count's loop so planned, a batch of its
  * rounds. */
diff --git a/tests/check.py b/tests/check.py
index f330bae..ff04e09 100644
--- a/tests/check.py
+++ b/tests/check.py
@@ -1219,6 +1219,74 @@ on go {
 """
 
 
+# Nested loops a batch runs together, and reductions (NESTED_SOURCE): Newton
+# steps breaking out lane by lane, a sum of fractions one a round, a nested
+# loop of 0 to 6 rounds a lane whose division by 0 is a fault in some
+# lane's round (q < 4), max and min, a sum of fractions several a lane (the
+# batch gives its rounds back), -0 summed, a grid's sums.
+NESTED_SOURCE = """state total = 0
+state steps = 0
+state most = 0
+state least = 0
+state frac = 0
+state zero = 0
+state grid = 0
+input go = message {
+  n: u32
+  q: u32
+}
+on go {
+  let sum = 0
+  let count = 0
+  for i in 1 .. go.n + 1 {
+    let r = i
+    for s in 0 .. 64 {
+      let next = (r + i / r) / 2
+      if next >= r { break }
+      r = next
+      count = count + 1
+    }
+    sum = sum + r
+  }
+  let high = 0
+  let low = 0
+  for i in 0 .. go.n {
+    let w = 0
+    for k in 0 .. i - floor(i / 7) * 7 {
+      w = w + 1000 / (k + go.q - 3)
+    }
+    high = max(high, w)
+    low = min(low, 0 - w)
+  }
+  let f = 0
+  for i in 0 .. go.n {
+    for k in 0 .. 3 {
+      f = f + 0.1
+    }
+  }
+  let z = 0 * (0 - 1)
+  for i in 0 .. go.n {
+    z = z + 0 * (0 - 1)
+  }
+  let g = 0
+  for j in 0 .. go.n {
+    let c = 0
+    for k in 0 .. 20 {
+      c = c + (j * 7 + k) * (k + 1)
+    }
+    g = g + c
+  }
+  total = sum
+  steps = count
+  most = high
+  least = low
+  frac = f
+  zero = z
+  grid = g
+}
+"""
+
+
 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,
@@ -1270,6 +1338,19 @@ def module_assumptions(aup, pure, work):
         assert outputs[name] == outputs['batches'], ('batch', 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 / '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)
+    given = ''.join(f'message go {struct.pack("<II", n, q).hex()}\nwork\n' for n, q in
+                    ((2000, 9), (2000, 2), (300, 100), (17, 9), (5, 9), (1500, 3)))
+    given += 'budget 30011\n' + given + 'budget 400000\n' + given
+    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'], ('nested', 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
     return len(runs)
 
 
@@ -1278,7 +1359,8 @@ def tier_program(rng):
     numbers in locals and states, a keyed list read and set by key, one
     whose keys came out of order read by key alone (HASHED), one of small
     keys read by key alone (full, or not where keys are left out), ifs and
-    breaks, loops whose rounds are apart (a batch's, 024), with what fails
+    breaks, loops whose rounds are apart (a batch's, 024), some with a loop
+    run in each and reductions (ra, rb, rc), with what fails
     (a key not there, not whole, 0 divided by, a cell past its column's
     width) as likely as not."""
     names = iter('jklpqrxyz')
@@ -1327,8 +1409,8 @@ def tier_program(rng):
     def statements(depth, loops, indent, inside):
         lines = []
         for _ in range(rng.randint(1, 3)):
-            kind = rng.choice(['s', 's', 'u', 'acc', 'if', 'loop', 'walk', 'set', 'set', 'break', 'sweep', 'lanes', 'lanes'] if depth < 3
-                              else ['s', 'u', 'set', 'break'])
+            kind = rng.choice(['s', 's', 'u', 'acc', 'if', 'loop', 'walk', 'set', 'set', 'break', 'sweep', 'lanes', 'lanes', 'nest', 'nest']
+                              if depth < 3 else ['s', 'u', 'set', 'break'])
             pad = '  ' * indent
             if kind == 's':
                 lines.append(f'{pad}s = s + {number(1, loops)}')
@@ -1382,6 +1464,30 @@ def tier_program(rng):
                         lines.append(f'{pad}  if {cond(1, inner)} {{ t[{at()}].v = {number(1, inner)} }}' +
                                      (f' else {{ t[{at()}].v = {number(1, inner)} }}' if rng.random() < 0.4 else ''))
                 lines.append(f'{pad}}}')
+            elif kind == 'nest':  # rounds apart, a loop in each run together, reductions (024's stage 2)
+                outer, inner = next(names, None), next(names, None)
+                if not inner:
+                    continue
+                w = f'w{outer}'
+                lines.append(f'{pad}for {outer} in {rng.choice(["0", "1", "go.q"])} .. {rng.choice(["go.n", "go.m", "go.n + 3"])} {{')
+                lines.append(f'{pad}  let {w} = {number(2, loops + [outer])}')
+                high = rng.choice([outer, f'{outer} + 2', 'go.q', '3', f'floor({outer} / 2)', '64'])
+                lines.append(f'{pad}  for {inner} in {rng.choice(["0", "1", outer])} .. {high} {{')
+                both = loops + [outer, inner, w]
+                for _ in range(rng.randint(1, 3)):
+                    pick = rng.random()
+                    if pick < 0.25:
+                        lines.append(f'{pad}    if {cond(1, both)} {{ break }}')
+                    elif pick < 0.55:
+                        lines.append(f'{pad}    {w} = {w} + {number(2, both)}')
+                    elif pick < 0.8:
+                        lines.append(f'{pad}    rc = rc + {number(2, both)}')
+                    else:
+                        lines.append(f'{pad}    if {cond(1, both)} {{ {w} = {number(2, both)} }}')
+                lines.append(f'{pad}  }}')
+                lines.append(f'{pad}  ra = ra + {w}')
+                lines.append(f'{pad}  rb = {rng.choice(["max", "min"])}(rb, {number(2, loops + [outer, w])})')
+                lines.append(f'{pad}}}')
             elif kind == 'walk' and 'c' not in loops:
                 lines.append(f'{pad}for c in f {{')
                 lines += statements(depth + 1, loops + ['c'], indent + 1, True)
@@ -1395,9 +1501,9 @@ def tier_program(rng):
         '  for i in 0 .. go.n {', f'    t.add(Cell {{ at: i, v: {rng.choice(["i", "0", "(i * 3)", "1"])} }})',
         '    f.append(Cell { at: i * 2, v: i })', '    h.add(Cell { at: i * 7919 - floor(i * 7919 / 1009) * 1009 + 5000000, v: i * 5 })',
         f'    g.add(Cell {{ at: {rng.choice(["i", "i", "i + floor(i / 40)"])}, v: i * 3 }})', '  }',
-        '  let s = 0', '  let u = 1',
+        '  let s = 0', '  let u = 1', '  let ra = 0', '  let rb = 0', '  let rc = 0',
         '  for i in 0 .. go.m {', *body, '  }',
-        '  out = s', '  acc = acc + u', '}', ''])
+        '  out = s + ra + rb + rc', '  acc = acc + u', '}', ''])
 
 
 def module_tier(aup, pure, work, programs=150):