Feat: N-15 min/max reductions; fork, signal, shutdown tests (#277)

docs/SPEC.md

@@ -892,9 +892,24 @@ All in `PM/commonTest`, so they run wherever the interpreter loads.
     `min(a, b)` compares with `<` and keeps the first element in iteration order on ties (including
     `+0.0` versus `-0.0`), NaN behaviour depends on position because comparisons with NaN are false,
     and chunked combination must reproduce that chain bit for bit; if an implementation cannot, they
-    are left to a later step. Step 2 implements `+` and `*` only: the compiled subset has no bitwise
+    are left to a later step. Step 2 implements `+` and `*` only (min/max follow in step 3 below): the compiled subset has no bitwise
     operator (the front end skips a function using `&`, `|` or `^` on ints, there is no IR operation),
     so `& | ^` are left until the subset has them.
+    **`min`/`max` (step 3).** The compiled subset has no `min(a, b)` / `max(a, b)` call (it stays a
+    generic object call, sequential). What it does compile is Python's own definition of
+    `min(acc, e)`, the conditional update `if e < acc: acc = e` (`e > acc` for max; `acc > e` and
+    `acc < e` are the same comparisons; no `else`, one statement, the assigned value is the compared
+    `e`). `<=` and `>=` are not min/max (the last of equal elements wins) and stay sequential. The
+    accumulator is a function-local i64 or f64 read nowhere else in the loop and assigned before it, `e` has
+    the accumulator's type and is built as for `+`/`*` (f64: constants, locals, `float(i)`, `b[i]`, `+ - *`,
+    negation). A loop may mix `+ *` and min/max statements. Each chunk folds from its own first element
+    with the same strict comparison; the partials combine in chunk order on the calling thread with the
+    earlier value on the left, so ties keep the first element in iteration order (including `+0.0` versus
+    `-0.0`). i64 is exact under any such grouping. **f64 is chunked only when no NaN is involved**: the
+    chain is not associative with a NaN (`[1.0] | [nan, 0.0]` folds to 1.0 chunked, 0.0 sequentially), so a
+    chunk that meets a NaN element (including one produced by `e`), or a NaN initial accumulator, discards
+    the region and the sequential loop runs, exactly as for an i64 overflow. Result and exception are
+    therefore Python's bit for bit for every NaN and signed-zero placement and thread count.
     A reduction loop is `acc = acc op e` statements only (no element-wise stores in the same loop, so
     a rerun needs no restore), each `acc` a function-local i64 read nowhere else in the loop and
     assigned before it, `e` built from i64 constants, locals, `b[i]` of a list or local int array
@@ -926,7 +941,7 @@ All in `PM/commonTest`, so they run wherever the interpreter loads.
     chunk, NaN/inf/-0.0), identical bytes for every thread count, failure at first/middle/last
     iteration, negative qualification cases, disabled-path test, reductions, race and GC tests,
     TSan/ASan, N-14, signals, fork and shutdown, and measurement tests for the threshold.
-  `Status: steps 1 and 2 implemented (#277): element-wise F64 stores a[i] = <F64 + - * and negation of constants, locals and b[i]> into list copy-in or local arrays (no failure path: eligible loops cannot fail or deopt); i64 reductions acc += e and acc *= e with the overflow rerun above (stores and reductions never in one loop); & | ^ and min/max reductions, float reductions, the poll inside a region and the rest planned (design)`.
+  `Status: steps 1, 2 and 3a implemented (#277): element-wise F64 stores a[i] = <F64 + - * and negation of constants, locals and b[i]> into list copy-in or local arrays (no failure path: eligible loops cannot fail or deopt); i64 reductions acc += e and acc *= e with the overflow rerun above (stores and reductions never in one loop); i64 and f64 min/max as `if e < acc: acc = e` (f64 with a NaN reruns sequentially); fork, signal and shutdown tests; & | ^ reductions, float + and * reductions, the poll inside a region and the rest planned (design)`.
 
 ---
 

python-multiplatform-ksp/src/main/python/typedpython/cgen.py

@@ -2151,7 +2151,19 @@ def _emit_region(self, s: ir.ForRange, ts: str, tn: str, mag: int) -> None:
         if info.reduce:
             if info.stored or len(info.reduce) != len(s.body):
                 raise CGenError(f"{f.name}: parallel reduction loop with stores or a missing reduction")
-            for st, (acc, opn) in zip(sorted(s.body, key=lambda x: getattr(x, "target", "")), info.reduce):
+            def key(x):
+                return x.then[0].target if isinstance(x, ir.If) and len(x.then) == 1 else getattr(x, "target", "")
+            for st, (acc, opn) in zip(sorted(s.body, key=key), info.reduce):
+                if opn in ("MIN", "MAX"):
+                    c = st.cond if isinstance(st, ir.If) else None
+                    a = st.then[0] if isinstance(st, ir.If) and len(st.then) == 1 else None
+                    if (not isinstance(c, ir.Compare) or not isinstance(a, ir.Assign) or st.orelse
+                            or a.target != acc or acc == s.var or a.value.type not in (Type.I64, Type.F64)
+                            or self.types.get(acc) is not a.value.type):
+                        raise CGenError(f"{f.name}: parallel min/max statement is not `if e < acc: acc = e`")
+                    reds.append((acc, opn, a.value))
+                    self._par_collect(a.value, scalars, arrays)
+                    continue
                 v = st.value if isinstance(st, ir.Assign) else None
                 if (not isinstance(v, ir.BinOp) or st.target != acc or v.op.name != opn
                         or opn not in ("ADD", "MUL") or v.type is not Type.I64
@@ -2204,15 +2216,35 @@ def _emit_region(self, s: ir.ForRange, ts: str, tn: str, mag: int) -> None:
             decls.append(f"    {ct} {ident} = c->{ident};")
             sets.append(f"tp_pr.{ident} = {ident};")
         if reds:
-            fields += ["    unsigned char *tp_pst;"] + [f"    int64_t *tp_part{j};" for j in range(len(reds))]
+            fields += ["    unsigned char *tp_pst;"] + [f"    {_CT[self.types[a]]} *tp_part{j};"
+                                                       for j, (a, _, _) in enumerate(reds)]
         saved = (self.lines, self.temps, self.depth, self.brk)
         self.lines, self.temps, self.depth, self.brk = [], [], 2, []
         try:
-            self.stmts(s.body)
+            if reds:
+                # Assign statements go through the ordinary emitter; a MIN/MAX statement is emitted
+                # here: the chunk folds from its own first element with strict < (> for max), and a
+                # NaN element marks the chunk bad (the region is rerun sequentially, see tp_runtime.h).
+                by_acc = {acc: (opn, e) for acc, opn, e in reds}
+                for st in s.body:
+                    if isinstance(st, ir.If):
+                        acc = st.then[0].target
+                        opn, e = by_acc[acc]
+                        j = [r[0] for r in reds].index(acc)
+                        v = self.ev(e)
+                        ident, cmp = _ident("l", acc), "<" if opn == "MIN" else ">"
+                        if e.type is Type.F64:
+                            self.emit(f"if ({v} != {v}) {{ tp_rc = 1; goto tp_exit; }} /*tpmm*/")
+                        self.emit(f"if (!tp_has{j}) {{ {ident} = {v}; tp_has{j} = 1; }} /*tpmm*/")
+                        self.emit(f"else if ({v} {cmp} {ident}) {{ {ident} = {v}; }} /*tpmm*/")
+                    else:
+                        self.stmts((st,))
+            else:
+                self.stmts(s.body)
             body_lines, body_temps = self.lines, self.temps
         finally:
             self.lines, self.temps, self.depth, self.brk = saved
-        text = "\n".join(body_lines)
+        text = "\n".join(ln for ln in body_lines if "/*tpmm*/" not in ln)
         if reds:
             # the one failure path a reduction body has: the sequential overflow check, which here
             # marks the chunk bad (the worker's own tp_rc / tp_exit); nothing else may remain
@@ -2228,7 +2260,9 @@ def _emit_region(self, s: ir.ForRange, ts: str, tn: str, mag: int) -> None:
                "    uint64_t tp_k;", *decls]
         if reds:
             out += ["    int tp_rc = 0;", "    int tp_s = 0;"]
-            out += [f"    int64_t {_ident('l', acc)} = {'1' if opn == 'MUL' else '0'};" for acc, opn, _ in reds]
+            out += [f"    {_CT[self.types[acc]]} {_ident('l', acc)} = {'1' if opn == 'MUL' else '0'};"
+                    for acc, opn, _ in reds]
+            out += [f"    int tp_has{j} = 0;" for j, (_, opn, _) in enumerate(reds) if opn in ("MIN", "MAX")]
             out.append("    (void)tp_rc; (void)tp_s;")
         for ct, n in body_temps:
             out.append(f"    {ct}{'' if ct.endswith('*') else ' '}{n} = 0;")
@@ -2239,6 +2273,7 @@ def _emit_region(self, s: ir.ForRange, ts: str, tn: str, mag: int) -> None:
         out.append("    }")
         if reds:
             out += [f"    c->tp_part{j}[tp_ci] = {_ident('l', acc)};" for j, (acc, _, _) in enumerate(reds)]
+            out += [f"    if (!tp_has{j}) goto tp_exit;" for j, (_, opn, _) in enumerate(reds) if opn in ("MIN", "MAX")]
             out += ["    c->tp_pst[tp_ci] = TP_PAR_OK;", "    return;", "tp_exit:",
                     "    c->tp_pst[tp_ci] = TP_PAR_BAD;"]
         out += ["}", ""]
@@ -2252,8 +2287,8 @@ def _emit_region(self, s: ir.ForRange, ts: str, tn: str, mag: int) -> None:
         self.emit(f"{ctx_t} tp_pr;")
         if reds:
             self.emit("unsigned char tp_pst[TP_PAR_HARD_CAP] = {0};")
-            for j in range(len(reds)):
-                self.emit(f"int64_t tp_pp{j}[TP_PAR_HARD_CAP];")
+            for j, (acc, _, _) in enumerate(reds):
+                self.emit(f"{_CT[self.types[acc]]} tp_pp{j}[TP_PAR_HARD_CAP];")
         self.emit(f"tp_pr.tp_ts = {ts};")
         for line in sets:
             self.emit(line)
@@ -2267,8 +2302,12 @@ def _emit_region(self, s: ir.ForRange, ts: str, tn: str, mag: int) -> None:
             # (a chunk, or here) leaves every accumulator untouched and the sequential loop runs.
             self.emit(f"{pok} = 1;")
             for j, (acc, opn, _) in enumerate(reds):
-                self.emit(f"int64_t tp_rv{j} = {_ident('l', acc)};")
-                self.emit(f"{pok} = {pok} && tp_par_combine(TP_PAR_OP_{opn}, &tp_rv{j}, tp_pp{j}, tp_pst) == 0;")
+                self.emit(f"{_CT[self.types[acc]]} tp_rv{j} = {_ident('l', acc)};")
+                if opn in ("MIN", "MAX"):
+                    fn = "tp_par_combine_ext_f64" if self.types[acc] is Type.F64 else "tp_par_combine_ext_i64"
+                    self.emit(f"{pok} = {pok} && {fn}({1 if opn == 'MAX' else 0}, &tp_rv{j}, tp_pp{j}, tp_pst) == 0;")
+                else:
+                    self.emit(f"{pok} = {pok} && tp_par_combine(TP_PAR_OP_{opn}, &tp_rv{j}, tp_pp{j}, tp_pst) == 0;")
             self.emit(f"if ({pok}) {{")
             for j, (acc, _, _) in enumerate(reds):
                 self.emit(f"    {_ident('l', acc)} = tp_rv{j};")

python-multiplatform-ksp/src/main/python/typedpython/runtime/API.md

@@ -492,7 +492,7 @@ for the epoch's increment. `tp_runtime.h` needs the GCC/Clang `__atomic` builtin
 
 ## Parallel regions (SPEC N-15, issues #151, #277)
 A `ForRange` the verifier marked `par` (`ir.ParInfo`; marked function, `a[i] = F64 expression` stores
-or i64 `+ *` reductions) runs its iterations on per-region C threads above a work threshold. Workers touch no Python
+or i64 `+ *` reductions, or min/max `if e < acc: acc = e` on an i64 or f64) runs its iterations on per-region C threads above a work threshold. Workers touch no Python
 object and no Python API, are not Python threads and never outlive the call.
     #define TP_PAR_MIN_WORK 65536          /* PLACEHOLDER, to be measured on #151: weighted iterations (trip count * ParInfo.weight) */
     #define TP_PAR_MAX_THREADS 8           /* default cap; TYPEDPYTHON_THREADS overrides (1 disables), hard cap 64 */
@@ -526,7 +526,14 @@ writes `part[ci]` and `st[ci] = TP_PAR_OK`; a failed check sets `st[ci] = TP_PAR
 After `tp_par_run` the caller combines every accumulator with `tp_par_combine`; if any accumulator fails
 (a BAD chunk or an overflow in the combination) no accumulator and no loop variable is written, the rerun
 counter is incremented and the sequential loop runs from the loop's start state, which deopts or not
-exactly as today. (`& | ^` and min/max are not reductions yet: the compiled subset has no bitwise operator.)
+exactly as today. (`& | ^` are not reductions yet: the compiled subset has no bitwise operator.)
+    int  tp_par_combine_ext_i64(int is_max, int64_t *acc, const int64_t *part, const unsigned char *st);
+    int  tp_par_combine_ext_f64(int is_max, double *acc, const double *part, const unsigned char *st);
+A min/max accumulator (`if e < acc: acc = e`, `>` for max) is folded per chunk from the chunk's own first element with the
+same strict comparison and combined in chunk order with the earlier value on the left, which equals the sequential chain
+for i64 and for f64 without NaN (ties and signed zeros keep the first). An f64 chunk that meets a NaN element sets
+`TP_PAR_BAD`, and `tp_par_combine_ext_f64` returns 1 for a NaN initial accumulator: the loop then runs sequentially
+(`[1.0] | [nan, 0.0]` is the case a chunked fold gets wrong).
 
 ## Development-mode import gate (Python API, not C; SPEC N-7b)
     typedpython.importgate.install(source_roots)   # meta path finder; checks project modules before they run

python-multiplatform-ksp/src/main/python/typedpython/runtime/tp_runtime.h

@@ -3116,6 +3116,53 @@ static inline int tp_par_combine(int op, int64_t *acc, const int64_t *part, cons
     return 0;
 }
 
+/* min / max (the conditional update `if e < acc: acc = e`, `>` for max). Each part[k] is its chunk's
+ * extreme taken from the chunk's own first element with the same strict comparison. Combining in chunk
+ * order with the earlier value on the left gives the sequential chain's result exactly for an i64 (a
+ * total order: the first of equal elements wins in either grouping) and for an f64 WITHOUT NaN (ties,
+ * including +0.0 / -0.0, keep the earlier chunk's element because the comparison is strict). With a NaN
+ * the chain is not associative ([1.0] | [nan, 0.0]: chunks give 1.0, the sequential loop gives 0.0), so
+ * a chunk that saw a NaN element is marked bad by the worker, and a NaN initial accumulator is refused
+ * here: both return 1 and the caller reruns the loop sequentially. */
+static inline int tp_par_combine_ext_i64(int is_max, int64_t *acc, const int64_t *part, const unsigned char *st)
+{
+    int64_t v = *acc;
+    int k;
+    for (k = 0; k < TP_PAR_HARD_CAP; k++) {
+        if (st[k] == TP_PAR_BAD) {
+            return 1;
+        }
+    }
+    for (k = 0; k < TP_PAR_HARD_CAP && st[k] == TP_PAR_OK; k++) {
+        if (is_max ? part[k] > v : part[k] < v) {
+            v = part[k];
+        }
+    }
+    *acc = v;
+    return 0;
+}
+
+static inline int tp_par_combine_ext_f64(int is_max, double *acc, const double *part, const unsigned char *st)
+{
+    double v = *acc;
+    int k;
+    if (v != v) {
+        return 1;
+    }
+    for (k = 0; k < TP_PAR_HARD_CAP; k++) {
+        if (st[k] == TP_PAR_BAD) {
+            return 1;
+        }
+    }
+    for (k = 0; k < TP_PAR_HARD_CAP && st[k] == TP_PAR_OK; k++) {
+        if (is_max ? part[k] > v : part[k] < v) {
+            v = part[k];
+        }
+    }
+    *acc = v;
+    return 0;
+}
+
 static inline void tp_par_rerun_note(void)
 {
     __atomic_fetch_add(&tp_par_reruns_v, 1, __ATOMIC_RELAXED);