From 50a0a9e124a4faa55844cdfca687edbce4e14098 Mon Sep 17 00:00:00 2001 From: BREW <81044362+b-re-w@users.noreply.github.com> Date: Sun, 25 Oct 2026 16:25:00 +0900 Subject: [PATCH] Feat: N-15 min/max reductions with NaN rerun; fork, signal and shutdown tests (#277) --- docs/SPEC.md | 19 +- .../src/main/python/typedpython/cgen.py | 57 +- .../main/python/typedpython/runtime/API.md | 11 +- .../python/typedpython/runtime/tp_runtime.h | 47 ++ .../src/main/python/typedpython/verify.py | 61 +- .../src/test/python/test_parallel_loops.py | 636 +++++++++++++++++- 6 files changed, 809 insertions(+), 22 deletions(-) diff --git a/docs/SPEC.md b/docs/SPEC.md index cdbf9f60..bd0d3a7c 100644 --- a/docs/SPEC.md +++ b/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] = 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] = 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)`. --- diff --git a/python-multiplatform-ksp/src/main/python/typedpython/cgen.py b/python-multiplatform-ksp/src/main/python/typedpython/cgen.py index 9a5a5124..180f3b54 100644 --- a/python-multiplatform-ksp/src/main/python/typedpython/cgen.py +++ b/python-multiplatform-ksp/src/main/python/typedpython/cgen.py @@ -2150,7 +2150,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 @@ -2203,15 +2215,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 @@ -2227,7 +2259,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;") @@ -2238,6 +2272,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 += ["}", ""] @@ -2251,8 +2286,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) @@ -2266,8 +2301,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};") diff --git a/python-multiplatform-ksp/src/main/python/typedpython/runtime/API.md b/python-multiplatform-ksp/src/main/python/typedpython/runtime/API.md index a3de3edb..de10c720 100644 --- a/python-multiplatform-ksp/src/main/python/typedpython/runtime/API.md +++ b/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 diff --git a/python-multiplatform-ksp/src/main/python/typedpython/runtime/tp_runtime.h b/python-multiplatform-ksp/src/main/python/typedpython/runtime/tp_runtime.h index 0d11f897..1729d2a5 100644 --- a/python-multiplatform-ksp/src/main/python/typedpython/runtime/tp_runtime.h +++ b/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); diff --git a/python-multiplatform-ksp/src/main/python/typedpython/verify.py b/python-multiplatform-ksp/src/main/python/typedpython/verify.py index 15312762..07094fff 100644 --- a/python-multiplatform-ksp/src/main/python/typedpython/verify.py +++ b/python-multiplatform-ksp/src/main/python/typedpython/verify.py @@ -219,9 +219,12 @@ proven on an i64 non-buffer list or local array, and i64 + - *. Stores and reductions are never mixed in one loop. Those nodes can overflow, which is not a failure of the region: the runtime discards it and reruns the loop sequentially. (& | ^ are not in the compiled subset: the front - end lowers no bitwise operator, so there is no such reduction to mark; min/max are later.) + end lowers no bitwise operator, so there is no such reduction to mark.) + * Step 3 (min/max): a statement of the reduction body may be `if e < acc: acc = e` / `e > acc` (either + operand order, no else), `acc` an i64 or f64 local, `e` of its type (`_par_minmax`). That is Python's + `min(acc, e)`; the f64 NaN case reruns sequentially at run time (SPEC N-15). Anything else compiles sequentially, unchanged. This cannot find other eligible shapes (stencils, -calls, nested tiles, min/max); those are later steps. +calls, nested tiles, `min()` calls); those are later steps. Tuple. `Tuple(OBJ, elements)`: a tuple of OBJ expressions (a scalar needs Box), result OBJ, a reference creator in the ownership rule. Building one is not an effect. @@ -1694,8 +1697,8 @@ def ok(e: object) -> bool: return e.type is Type.F64 and e.op is UnaryOpKind.NEG and ok(e.operand) return False - if type(body[0]) is Assign: - return self._par_reduce(s, body, state, weight, array_ok, at_var) + if type(body[0]) in (Assign, If): + return self._par_reduce(s, body, state, weight, array_ok, at_var, ok) for st in body: if type(st) is not StoreIndex or not array_ok(st.array) or not at_var(st.index, st.proven): return None @@ -1706,26 +1709,41 @@ def ok(e: object) -> bool: return ParInfo(tuple(sorted(set(stored))), tuple(sorted(set(read))), weight[0]) def _par_reduce(self, s: ForRange, body: tuple, state: _State, weight: list, array_ok, - at_var) -> ParInfo | None: + at_var, ok_f64=None) -> ParInfo | None: """SPEC N-15 step 2: a body of only `acc = acc op e` (op ADD or MUL, either operand order), each `acc` a distinct i64 local of this function (not a loop variable, array or entry global) that is definitely assigned before the loop and appears nowhere else in the body, and `e` built from i64 Const, i64 Local (the loop variable, or any scalar the body does not assign), `b[var]` proven on an i64 non-buffer list or local array, and i64 + - * . Any of those nodes can overflow: the runtime then discards the region and reruns the loop sequentially. No store is allowed in - the same loop (a rerun then needs no restore).""" + the same loop (a rerun then needs no restore). + + Step 3 (min/max): a statement may instead be `if e < acc: acc = e` (MIN) or `if e > acc: acc = e` + (MAX), either operand order, no else, the assigned value equal to the compared `e`, `acc` an + I64 or F64 local like the others and `e` of the accumulator's type (F64: the step 1 float + expression set). That is Python's `min(acc, e)` / `max(acc, e)`: strict `<` keeps the first + element on ties. `<=` / `>=` are not it (the last of equal elements, which differs for -0.0).""" var = s.var if state is None: return None accs: dict[str, str] = {} + mm: dict[str, object] = {} # acc -> e, for MIN / MAX statements for st in body: + if type(st) is If: + hit = self._par_minmax(st) + if hit is None or hit[0] in accs: + return None + accs[hit[0]] = hit[1] + mm[hit[0]] = hit[2] + continue if (type(st) is not Assign or type(st.value) is not BinOp or st.value.type is not Type.I64 or st.value.op not in (BinOpKind.ADD, BinOpKind.MUL) or st.value.np or st.target in accs): return None accs[st.target] = st.value.op.name for acc in accs: - if (self.scalars.get(acc) is not Type.I64 or acc == var or acc in self.arrays + want = mm[acc].type if acc in mm else Type.I64 + if (self.scalars.get(acc) is not want or acc == var or acc in self.arrays or acc in self.larrays or acc in self.entry or acc not in state): return None read: list[str] = [] @@ -1750,8 +1768,18 @@ def ok(e: object) -> bool: and ok(e.left) and ok(e.right)) return False + def reads_acc(e: object) -> bool: + if type(e) is Local: + return e.name in accs + return any(reads_acc(x) for x in _exprs_of(e)) + for st in body: weight[0] += 2 + if type(st) is If: + e = mm[st.then[0].target] + if reads_acc(e) or not (ok(e) if e.type is Type.I64 else (ok_f64 is not None and ok_f64(e))): + return None + continue b = st.value if type(b.left) is Local and b.left.name == st.target and b.left.type is Type.I64: other = b.right @@ -1763,6 +1791,25 @@ def ok(e: object) -> bool: return None return ParInfo((), tuple(sorted(set(read))), weight[0], tuple(sorted(accs.items()))) + @staticmethod + def _par_minmax(st: If): + """(acc, "MIN" | "MAX", e) for `if e < acc: acc = e` and its mirrored spellings, else None.""" + c = st.cond + if (type(c) is not Compare or st.orelse or len(st.then) != 1 or type(st.then[0]) is not Assign + or c.op not in (CompareKind.LT, CompareKind.GT)): + return None + a = st.then[0] + if type(c.right) is Local and c.right.name == a.target: + e, kind = c.left, "MIN" if c.op is CompareKind.LT else "MAX" # e < acc, e > acc + elif type(c.left) is Local and c.left.name == a.target: + e, kind = c.right, "MIN" if c.op is CompareKind.GT else "MAX" # acc > e, acc < e + else: + return None + if e != a.value or e.type is not c.left.type or e.type is not c.right.type \ + or e.type not in (Type.I64, Type.F64): + return None + return a.target, kind, e + def s_Return(self, s: Return, state: _State): returns = self.fn.returns if s.value is None: diff --git a/python-multiplatform-ksp/src/test/python/test_parallel_loops.py b/python-multiplatform-ksp/src/test/python/test_parallel_loops.py index 3b4f037d..80fd6bf2 100644 --- a/python-multiplatform-ksp/src/test/python/test_parallel_loops.py +++ b/python-multiplatform-ksp/src/test/python/test_parallel_loops.py @@ -137,9 +137,11 @@ def walk(body): if isinstance(s, ir.ForRange): out.append(s) walk(s.body) - elif isinstance(s, (ir.If, ir.While)): + elif isinstance(s, ir.If): + walk(s.then) + walk(s.orelse) + elif isinstance(s, ir.While): walk(s.body) - walk(getattr(s, "orelse", ())) walk(f.body) return out @@ -838,3 +840,633 @@ def test_no_reduction_region_without_a_marker(tmp_path): src = RSOURCE.replace("# typedpython: parallel\n", "") _, c_text, _, _ = build(tmp_path, src, name="rnomark") assert "tp_par_run(" not in c_text and "__typedpython_parallel__" not in c_text + + +# ===================================================================================================== +# Step 3a (SPEC N-15, #277): min/max reductions, the conditional update `if e < acc: acc = e`. +# +# The compiled subset has no `min(a, b)` / `max(a, b)` call (it stays a generic object call, never +# parallel). The form it does compile is Python's own definition of `min(acc, e)`: `if e < acc: acc = e` +# (`e > acc` for max; `acc > e` and `acc < e` are the same comparison). Strict `<` keeps the first element +# on ties, which is Python's. Int: exact under chunked combination in chunk order. Float: the chunked +# combination equals the sequential chain only without NaN, so a NaN (in an element or in the initial +# accumulator) discards the region and the sequential loop runs; the counterexample is +# [1.0] | [nan, 0.0]: chunk fold gives 1.0, the sequential chain gives 0.0. +# +# Red record: written before the implementation; before it none of these loops was `par`. +# ===================================================================================================== + +MSOURCE = '''\ +# typedpython: compiled + +# typedpython: parallel +def mn_i(xs: list[int]) -> int: + m = xs[0] + for i in range(len(xs)): + if xs[i] < m: + m = xs[i] + return m + + +# typedpython: parallel +def mx_i(xs: list[int]) -> int: + m = xs[0] + for i in range(len(xs)): + if xs[i] > m: + m = xs[i] + return m + + +# typedpython: parallel +def mn_f(xs: list[float]) -> float: + m = xs[0] + for i in range(len(xs)): + if xs[i] < m: + m = xs[i] + return m + + +# typedpython: parallel +def mx_f(xs: list[float]) -> float: + m = xs[0] + for i in range(len(xs)): + if xs[i] > m: + m = xs[i] + return m + + +# typedpython: parallel +def mn_f_flipped(xs: list[float], init: float) -> float: + m = init + for i in range(len(xs)): + if m > xs[i]: + m = xs[i] + return m + + +# typedpython: parallel +def mx_f_flipped(xs: list[float], init: float) -> float: + m = init + for i in range(len(xs)): + if m < xs[i]: + m = xs[i] + return m + + +# typedpython: parallel +def mn_i_expr(xs: list[int], c: int) -> int: + m = c + for i in range(len(xs)): + if xs[i] * 3 - i < m: + m = xs[i] * 3 - i + return m + + +# typedpython: parallel +def mn_f_expr(xs: list[float], a: float) -> float: + m = a + for i in range(len(xs)): + if xs[i] * a - float(i) < m: + m = xs[i] * a - float(i) + return m + + +# typedpython: parallel +def mn_and_sum(xs: list[int]) -> int: + m = xs[0] + s = 0 + for i in range(len(xs)): + s += xs[i] + if xs[i] < m: + m = xs[i] + return m * 1000003 + s + + +# typedpython: parallel +def two_ext(xs: list[float]) -> float: + lo = xs[0] + hi = xs[0] + for i in range(len(xs)): + if xs[i] < lo: + lo = xs[i] + if xs[i] > hi: + hi = xs[i] + return hi - lo + + +# typedpython: parallel +def mn_i_init(xs: list[int], init: int) -> int: + m = init + for i in range(len(xs)): + if xs[i] < m: + m = xs[i] + return m + + +# typedpython: parallel +def mn_i_big_e(xs: list[int], c: int) -> int: + m = 0 + for i in range(len(xs)): + if xs[i] * c < m: + m = xs[i] * c + return m + + +def mn_unmarked(xs: list[int]) -> int: + m = xs[0] + for i in range(len(xs)): + if xs[i] < m: + m = xs[i] + return m + + +# typedpython: parallel +def n_le(xs: list[float]) -> float: + m = xs[0] + for i in range(len(xs)): + if xs[i] <= m: + m = xs[i] + return m + + +# typedpython: parallel +def n_other_value(xs: list[int]) -> int: + m = xs[0] + for i in range(len(xs)): + if xs[i] < m: + m = xs[i] + 1 + return m + + +# typedpython: parallel +def n_else(xs: list[int]) -> int: + m = xs[0] + s = 0 + for i in range(len(xs)): + if xs[i] < m: + m = xs[i] + else: + s += 1 + return m + s + + +# typedpython: parallel +def n_two_in_then(xs: list[int]) -> int: + m = xs[0] + s = 0 + for i in range(len(xs)): + if xs[i] < m: + m = xs[i] + s += 1 + return m + s + + +# typedpython: parallel +def n_acc_in_value(xs: list[int]) -> int: + m = xs[0] + for i in range(len(xs)): + if xs[i] < m: + m = xs[i] + m = m + 0 + return m + + +# typedpython: parallel +def n_cond_other(xs: list[int], t: int) -> int: + m = xs[0] + for i in range(len(xs)): + if xs[i] < t: + m = xs[i] + return m + + +# typedpython: parallel +def n_mixed_types(xs: list[int], init: float) -> float: + m = init + for i in range(len(xs)): + if xs[i] < m: + m = float(xs[i]) + return m + + +# typedpython: parallel +def n_acc_reads_acc(xs: list[int]) -> int: + m = xs[0] + for i in range(len(xs)): + if xs[i] < m: + m = xs[i] + elif xs[i] == m: + m = m + return m + + +# typedpython: parallel +def n_div(xs: list[float], d: float) -> float: + m = xs[0] + for i in range(len(xs)): + if xs[i] / d < m: + m = xs[i] / d + return m +''' + +MM_OK = {"mn_i": (("m", "MIN"),), "mx_i": (("m", "MAX"),), "mn_f": (("m", "MIN"),), + "mx_f": (("m", "MAX"),), "mn_f_flipped": (("m", "MIN"),), "mx_f_flipped": (("m", "MAX"),), + "mn_i_expr": (("m", "MIN"),), "mn_f_expr": (("m", "MIN"),), + "mn_and_sum": (("m", "MIN"), ("s", "ADD")), "two_ext": (("hi", "MAX"), ("lo", "MIN")), + "mn_i_init": (("m", "MIN"),), "mn_i_big_e": (("m", "MIN"),)} +MM_NO = ["n_le", "n_other_value", "n_else", "n_two_in_then", "n_acc_in_value", "n_cond_other", + "n_mixed_types", "n_acc_reads_acc", "n_div"] + + +@pytest.fixture(scope="module") +def mbuilt(tmp_path_factory): + return build(tmp_path_factory.mktemp("mmpar"), MSOURCE, name="mmparmod") + + +@pytest.fixture(scope="module") +def mmod(mbuilt): + return cbuild.load("mmparmod", mbuilt[2]) + + +@pytest.mark.parametrize("name", sorted(MM_OK)) +def test_minmax_loops_are_marked_par(mbuilt, name): + loops = [l for l in par_loops(fn(mbuilt, name)) if l.par is not None] + assert loops, name + assert loops[-1].par.reduce == MM_OK[name] and loops[-1].par.stored == () + + +@pytest.mark.parametrize("name", MM_NO) +def test_ineligible_minmax_shapes_stay_sequential(mbuilt, name): + f = fn(mbuilt, name) + assert f.parallel is True + loops = par_loops(f) + assert loops and all(l.par is None for l in loops), name + + +def test_unmarked_minmax_never_parallel(mbuilt): + assert all(l.par is None for l in par_loops(fn(mbuilt, "mn_unmarked"))) + + +def test_minmax_c_regions_only_for_eligible(mbuilt): + import re + owners = set(re.findall(r"static void tp_par_(\w+?)(?:_fast)?_\d+_body\(", mbuilt[1])) + assert owners == set(MM_OK), owners + + +def test_verifier_rejects_a_forged_minmax(mbuilt): + good = par_loops(fn(mbuilt, "mn_i"))[-1].par + proved, diags = _forge(mbuilt, "mn_i", dataclasses.replace(good, reduce=(("m", "MAX"),))) + assert not proved.functions and any(d.rule == PAR_RULE for d in diags) + for name in ("n_le", "n_other_value", "n_else", "n_cond_other"): + proved, diags = _forge(mbuilt, name, ir.ParInfo((), (), 3, (("m", "MIN"),))) + assert not proved.functions and any(d.rule == PAR_RULE for d in diags), name + proved, diags = _forge(mbuilt, "mn_i", good) + assert proved.functions and not diags + + +def test_naive_chunk_fold_differs_on_nan_so_float_nan_cannot_be_chunked(): + """The counterexample, in plain Python: why a NaN must send the float region to the sequential loop.""" + nan = float("nan") + + def fold(xs, acc): + for x in xs: + if x < acc: + acc = x + return acc + + seq = fold([1.0, nan, 0.0], 5.0) + chunk_first = fold([1.0], 5.0) + chunk_second = fold([nan, 0.0], 5.0) # starts from its own first element on a chunk + naive = chunk_second if chunk_second < chunk_first else chunk_first + assert seq == 0.0 + # the chunk starting at nan keeps nan, and `nan < 1.0` is false: the combine keeps 1.0 + p2 = nan + for x in [0.0]: + if x < p2: + p2 = x + assert p2 != p2 + combined = p2 if p2 < 1.0 else 1.0 + assert combined == 1.0 and combined != seq + assert naive in (0.0, 1.0) + + +MM_CHILD_BODY = """ + import math, random + I = m.__typedpython_interpreted__ + def eq(a, b): + if isinstance(a, float): + return isinstance(b, float) and struct.pack(" 1: + assert r["regions"] > 0 + assert r["reruns"] > 0 # NaN and overflow cases fell back + + +def test_float_minmax_without_nan_never_reruns(mbuilt): + r = run_child(mbuilt[2], """ + import random + rr = random.Random(9) + xs = [rr.uniform(-1e3, 1e3) for _ in range(400001)] + a = m.mn_f(xs); b = m.mx_f(xs); c = m.mn_f_flipped(xs, 2e3) + print(json.dumps({"ok": a == min(xs) and b == max(xs) and c == min(xs), + "regions": m.__typedpython_parallel__()[0], "reruns": m.__typedpython_parallel__()[3]})) + """, {"TYPEDPYTHON_THREADS": "4"}) + assert r["ok"] + if (os.cpu_count() or 1) > 1: + assert r["regions"] == 3 and r["reruns"] == 0 + + +def test_float_minmax_with_nan_reruns_once_per_call(mbuilt): + r = run_child(mbuilt[2], """ + xs = [1.0] * 400001 + xs[200000] = float("nan"); xs[200001] = 0.0 + a = m.mn_f(xs) + print(json.dumps({"a": a, "regions": m.__typedpython_parallel__()[0], "reruns": m.__typedpython_parallel__()[3]})) + """, {"TYPEDPYTHON_THREADS": "4"}) + assert r["a"] == 0.0 + if (os.cpu_count() or 1) > 1: + assert r["reruns"] == 1 + + +# ===================================================================================================== +# Step 3b (SPEC N-15, design §5.2 and §7.4): fork, signals and interpreter shutdown around a region. +# Every scenario runs in a CHILD interpreter with a timeout (a crash or a hang fails one test). +# Red record: these are new tests of behaviour steps 1 and 2 already implement (no polling inside a +# region, workers are not Python threads, per-call threads); none was red before the change. +# ===================================================================================================== + +BIG = 4_000_000 + + +def _threads_env(): + return {"TYPEDPYTHON_THREADS": "4"} + + +def test_fork_from_a_python_thread_while_another_thread_runs_a_region(built): + r = run_child(built[2], """ + import os, time, warnings, signal + warnings.simplefilter("ignore") + big = [float(k %% 977) for k in range(%d)] + running = threading.Event() + stop = threading.Event() + parent_bad = [] + def regionist(): + xs = list(big) + first = True + while not stop.is_set(): + ys = list(big) + if first: + running.set(); first = False + m.scale(ys, 2.0, 1.0) + if ys[5] != big[5] * 2.0 + 1.0 or ys[-1] != big[-1] * 2.0 + 1.0: + parent_bad.append(1) + t = threading.Thread(target=regionist) + t.start() + assert running.wait(60) + out = {} + def forker(): + pid = os.fork() + if pid == 0: + # the child has only this thread: a small call and a full region must both work + ok = True + try: + a = [1.5] * 10 + m.scale(a, 2.0, 1.0) + ok = ok and a == [4.0] * 10 + b = [float(k %% 977) for k in range(300001)] + m.scale(b, 2.0, 1.0) + ok = ok and b[7] == 15.0 + finally: + os._exit(0 if ok else 3) + out["pid"] = pid + f = threading.Thread(target=forker) + f.start(); f.join() + deadline = time.time() + 120 + status = None + while time.time() < deadline: + p, st = os.waitpid(out["pid"], os.WNOHANG) + if p: + status = st + break + time.sleep(0.01) + stop.set(); t.join() + print(json.dumps({"status": status, "exited": os.WIFEXITED(status) if status is not None else None, + "code": os.WEXITSTATUS(status) if status is not None and os.WIFEXITED(status) else None, + "parent_bad": parent_bad, "regions": m.__typedpython_parallel__()[0]})) + """ % BIG, _threads_env(), timeout=300) + assert r["exited"] is True and r["code"] == 0, r + assert r["parent_bad"] == [] + if (os.cpu_count() or 1) > 1: + assert r["regions"] >= 1 + + +SIGNAL_BODY = """ + import signal, time + main = threading.get_ident() + big = [float(k %% 977) for k in range(%d)] + seen = [] + def handler(signum, frame): + seen.append((threading.get_ident(), time.perf_counter())) + %s + signal.signal(signal.SIGALRM, handler) + %s +""" + + +def test_signal_handlers_run_on_the_main_thread_after_the_region(built): + r = run_child(built[2], SIGNAL_BODY % (BIG, "pass", "") + """ + marks = [] + for _ in range(3): + del seen[:] + xs = list(big) + signal.setitimer(signal.ITIMER_REAL, 0.002) + t0 = time.perf_counter() + m.scale(xs, 2.0, 1.0) + t1 = time.perf_counter() + time.sleep(0.05) # let a pending handler run + marks.append({"n": len(seen), "all_main": all(i == main for i, _ in seen), + "after_region": all(t >= t1 - 1e-4 for _, t in seen), "dur": t1 - t0, + "ok": xs[3] == big[3] * 2.0 + 1.0 and xs[-1] == big[-1] * 2.0 + 1.0}) + signal.setitimer(signal.ITIMER_REAL, 0) + print(json.dumps({"marks": marks, "regions": m.__typedpython_parallel__()[0]})) + """, _threads_env(), timeout=300) + for k in r["marks"]: + assert k["ok"] and k["all_main"], k + assert k["n"] == 1, k + if (os.cpu_count() or 1) > 1: + assert r["regions"] == 3 + assert all(k["after_region"] for k in r["marks"]), r["marks"] # no handler ran inside the region + + +def test_repeating_signals_never_run_a_handler_on_a_worker(built): + r = run_child(built[2], SIGNAL_BODY % (BIG, "pass", "") + """ + signal.setitimer(signal.ITIMER_REAL, 0.001, 0.001) + ok = True + for _ in range(6): + xs = list(big) + m.scale(xs, 2.0, 1.0) + ok = ok and xs[11] == big[11] * 2.0 + 1.0 + signal.setitimer(signal.ITIMER_REAL, 0) + print(json.dumps({"ok": ok, "n": len(seen), "all_main": all(i == main for i, _ in seen)})) + """, _threads_env(), timeout=300) + assert r["ok"] and r["all_main"] and r["n"] >= 1 + + +def test_keyboardinterrupt_during_a_region_matches_the_sequential_semantics(built): + """The interrupt is raised by the handler after the region ended: the list is either completely + scaled or untouched (the call raised before write-back), never partly written, and the exception + is the KeyboardInterrupt, as for the interpreter.""" + r = run_child(built[2], SIGNAL_BODY % (BIG, "raise KeyboardInterrupt", "") + """ + outcomes = [] + for _ in range(4): + xs = list(big) + raised = False + signal.setitimer(signal.ITIMER_REAL, 0.002) + try: + m.scale(xs, 2.0, 1.0) + time.sleep(0.2) + except KeyboardInterrupt: + raised = True + signal.setitimer(signal.ITIMER_REAL, 0) + full = all(xs[k] == big[k] * 2.0 + 1.0 for k in range(0, len(xs), 4093)) + untouched = xs == big + outcomes.append({"raised": raised, "full": full, "untouched": untouched, + "main": all(i == main for i, _ in seen)}) + print(json.dumps({"outcomes": outcomes})) + """, _threads_env(), timeout=300) + for o in r["outcomes"]: + assert o["raised"] and o["main"], o + assert o["full"] != o["untouched"], o # exactly one: all or nothing + + +def test_keyboardinterrupt_in_a_reduction_region(rbuilt): + r = run_child(rbuilt[2], SIGNAL_BODY % (BIG, "raise KeyboardInterrupt", "") + """ + ints = [k %% 1000 for k in range(%d)] + want = sum(ints) + res = [] + for _ in range(4): + got = None + raised = False + signal.setitimer(signal.ITIMER_REAL, 0.002) + try: + got = m.r_sum(ints) + time.sleep(0.2) + except KeyboardInterrupt: + raised = True + signal.setitimer(signal.ITIMER_REAL, 0) + res.append({"raised": raised, "got": got}) + plain = m.r_sum(ints) + print(json.dumps({"res": res, "plain": plain, "want": want})) + """ % BIG, _threads_env(), timeout=300) + assert r["plain"] == r["want"] + for o in r["res"]: + assert o["raised"] and o["got"] in (None, r["want"]), o # a result is the full sum or nothing + + +def test_interpreter_shutdown_while_a_daemon_thread_is_in_a_region(built): + for attempt in range(3): + r = run_child(built[2], """ + import time + big = [float(k %% 977) for k in range(%d)] + inside = threading.Event() + def spin(): + while True: + ys = list(big) + inside.set() + m.scale(ys, 2.0, 1.0) + t = threading.Thread(target=spin, daemon=True) + t.start() + assert inside.wait(60) + time.sleep(0.01 * attempt_delay) + print(json.dumps({"exiting": True})) + sys.stdout.flush() + """.replace("attempt_delay", str(attempt)) % BIG, _threads_env(), timeout=300) + assert r == {"exiting": True} + + +def test_interpreter_shutdown_with_a_daemon_reduction_region(rbuilt): + r = run_child(rbuilt[2], """ + import time + ints = list(range(%d)) + inside = threading.Event() + def spin(): + while True: + inside.set() + m.r_sum(ints) + t = threading.Thread(target=spin, daemon=True) + t.start() + assert inside.wait(60) + print(json.dumps({"exiting": True})) + sys.stdout.flush() + """ % BIG, _threads_env(), timeout=300) + assert r == {"exiting": True}