diff --git a/docs/SPEC.md b/docs/SPEC.md index 85e0e471..c52eb7c3 100644 --- a/docs/SPEC.md +++ b/docs/SPEC.md @@ -869,7 +869,7 @@ All in `PM/commonTest`, so they run wherever the interpreter loads. - **N-15** TypedPython parallel execution of order-independent loops on free-threaded CPython (#151; `docs/design/typedpython-parallel-loops.md`). A compiled `for ... in range(...)` loop may run its iterations on several threads only if its result cannot depend on iteration order, in a function the - user marked (the spelling of the marking is settled at implementation), and then the + user marked with a `# typedpython: parallel` comment line directly above the `def` (above its decorators if any); module-level functions only, and then the result, the exception, the partial writes and the deopt behaviour are bit-for-bit those of the sequential loop (INTENT ยง1.4; N-12: every execution is indistinguishable from some valid CPython execution). @@ -895,20 +895,24 @@ All in `PM/commonTest`, so they run wherever the interpreter loads. (#153) awaits the owner's decision, and N-15 follows it. Partials combine in chunk order. - **Observability.** N-14 holds (a traced call runs the interpreted function, never a region); workers are not Python threads; the calling thread stays attached and polls the eval breaker - between micro-chunks when the function polls today; list writes become visible only at the - existing write-back. Outside the identity (like N-12's cyclic GC timing): the OS thread count, CPU + between micro-chunks when the function polls today (step 1 does not poll inside a region: a + signal waits until the region ends); the calling thread stays attached for the whole region, so a + stop-the-world waits for it too; list writes become visible only at the existing write-back. Outside the identity (like N-12's cyclic GC timing): the OS thread count, CPU time versus wall time, the multi-thread `os.fork()` warning while a region is active, and profilers seeing non-Python threads. - **Threads and cost.** v1 creates workers per region and joins them before the call returns (no pool, no thread outlives the call: nothing at fork or interpreter shutdown). Cap - `TYPEDPYTHON_THREADS` (1 disables; default 8 or the core count if lower). A region runs only above - a measured work threshold (`TP_PAR_MIN_WORK`), measured together with thread creation cost in a - quiet window and recorded on #151; below it the sequential code runs. Platforms without threads compile it out. + `TYPEDPYTHON_THREADS` (positive integer, 1 disables; default min(cores, 8); hard cap 64). A thread + that cannot be created is not an error: its chunk runs on the calling thread. Workers start with all + signals blocked. A region runs only above a measured work threshold (`TP_PAR_MIN_WORK`), which counts + weighted iterations (the trip count times the loop body's node count); a region also needs at least + 1024 iterations per thread. The threshold is measured together with thread creation cost in a quiet + window and recorded on #151; below it the sequential code runs. Platforms without threads compile it out. - **Tests** (written first): differential vs the interpreter (bit-exact, boundaries of trip count and 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: planned (design)`. + `Status: step 1 implemented (#277): element-wise F64 stores a[i] = into list copy-in or local arrays; no reductions, no failure path (eligible loops cannot fail or deopt); 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 cd9009f8..a817d0d3 100644 --- a/python-multiplatform-ksp/src/main/python/typedpython/cgen.py +++ b/python-multiplatform-ksp/src/main/python/typedpython/cgen.py @@ -512,6 +512,9 @@ def __init__(self, module: ir.Module): self.module = module self.display_path = "" self.functions = {f.name: f for f in module.functions} + # SPEC N-15: any ForRange.par in the module (the module then exports __typedpython_parallel__) + self.has_par = any(isinstance(st, ir.ForRange) and st.par is not None + for f in module.functions for st in _walk_stmts(f.body)) if len(self.functions) != len(module.functions): raise CGenError("duplicate compiled function names") self.names: list[str] = [] @@ -758,6 +761,8 @@ def generate(module: ir.Module, source_path: Path, display_path: str | None = No w("") if module.classes: w(_class_info_function(mg)) + if mg.has_par: + w(_par_stats_function()) w(_exec_function(mg)) w(_state_functions(mg)) w("static PyModuleDef_Slot tp_slots[] = {") @@ -929,6 +934,7 @@ def __init__(self, mg: _ModGen, f: ir.Function, fast: bool = False): self.depth = 1 self.temps: list[tuple[str, str]] = [] # (C type, name) self.arrays = {p.name: p for p in f.params if isinstance(p, ir.ArrayParam)} + self.par_defs: list[str] = [] # outlined region bodies (SPEC N-15), emitted before the impl self.local_arrays: dict[str, Type] = {} # name -> F64_ARRAY / I64_ARRAY self.brk: list = [] # per open loop: goto label for `break`, or None self.label_n = 0 @@ -2062,6 +2068,9 @@ def st_ForRange(self, s: ir.ForRange) -> None: self.emit(f" {tn} = ((uint64_t){ts} - (uint64_t){te} - 1u) / {mag} + 1u;") sign = "-" self.depth += 1 + par_open = s.par is not None + if par_open: + self._emit_region(s, ts, tn, abs(step)) vec = self._vec_plan(s) if step == 1 else None if self.snapshot: # no polling in a snapshot holder: a plain loop @@ -2095,9 +2104,111 @@ def st_ForRange(self, s: ir.ForRange) -> None: self.emit("}") if label in self.used_labels: self.emit(f"{label}: ;") + if par_open: + self.emit("}") # the sequential `else` of the parallel region self.depth -= 1 self.emit("}") + def _par_collect(self, e: ir.Expr, scalars: set, arrays: set) -> None: + if isinstance(e, ir.Local): + scalars.add(e.name) + elif isinstance(e, ir.Index): + arrays.add(e.array) + self._par_collect(e.index, scalars, arrays) + elif isinstance(e, ir.ToFloat): + self._par_collect(e.operand, scalars, arrays) + elif isinstance(e, ir.BinOp): + self._par_collect(e.left, scalars, arrays) + self._par_collect(e.right, scalars, arrays) + elif isinstance(e, ir.UnaryOp): + self._par_collect(e.operand, scalars, arrays) + elif not isinstance(e, ir.Const): + raise CGenError(f"{self.f.name}: {type(e).__name__} in a parallel loop body") + + def _emit_region(self, s: ir.ForRange, ts: str, tn: str, mag: int) -> None: + """SPEC N-15: opens `if ( ) { region } else {`; the caller closes the + else with the sequential loop it emits next. The body is the verified `par` loop body + (StoreIndex statements of F64 arithmetic), outlined into `tp_pbody_*`, which declares the + names it uses as locals copied from a context struct, so the existing statement and + expression emitters produce its code unchanged. Nothing in it can fail, deopt or build an + exception (the verifier's rule); cgen refuses text that tries.""" + f, info = self.f, s.par + if not f.parallel or not isinstance(info, ir.ParInfo) or s.step.value <= 0: + raise CGenError(f"{f.name}: ForRange.par on a loop that cannot be parallel (unmarked " + "function, bad step or no ParInfo)") + scalars: set[str] = set() + arrays: set[str] = set() + for st in s.body: + if not isinstance(st, ir.StoreIndex): + raise CGenError(f"{f.name}: parallel loop body has a {type(st).__name__}") + arrays.add(st.array) + self._par_collect(st.value, scalars, arrays) + scalars.discard(s.var) + for a in arrays: + if a not in self.arrays and a not in self.local_arrays: + raise CGenError(f"{f.name}: parallel loop array {a} is not an array") + if a in self.arrays and self.arrays[a].buffer: + raise CGenError(f"{f.name}: parallel loop over buffer array {a}") + for x in scalars: + if self.types.get(x) not in (Type.F64, Type.I64): + raise CGenError(f"{f.name}: parallel loop scalar {x} is not f64/i64") + k = len(self.par_defs) + tag = _ident("tp_par", f"{f.name}{'_fast' if self.fast else ''}_{k}") + ctx_t, body_f = f"{tag}_ctx", f"{tag}_body" + fields = [" int64_t tp_ts;"] + decls = [" int64_t " + _ident("l", s.var) + " = 0;"] + sets = [] + for a in sorted(arrays): + at = self.arrays[a].type if a in self.arrays else self.local_arrays[a] + ident = _ident("l", a) + fields.append(f" {_ARRAY_CT[at]} *{ident};") + if a in self.arrays: # a parameter is already a pointer + decls.append(f" {_ARRAY_CT[at]} *{ident} = c->{ident};") + sets.append(f"tp_pr.{ident} = {ident};") + else: # a local array is a struct: the worker's copy + decls.append(f" {_ARRAY_CT[at]} {ident} = *c->{ident};") + sets.append(f"tp_pr.{ident} = &{ident};") + for x in sorted(scalars): + ident, ct = _ident("l", x), _CT[self.types[x]] + fields.append(f" {ct} {ident};") + decls.append(f" {ct} {ident} = c->{ident};") + sets.append(f"tp_pr.{ident} = {ident};") + saved = (self.lines, self.temps, self.depth, self.brk) + self.lines, self.temps, self.depth, self.brk = [], [], 2, [] + try: + 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) + for banned in ("goto", "tp_rc", "tp_s ", "return", "tp_st", "tp_dict", "PyObject", "tp_release"): + if banned in text: + raise CGenError(f"{f.name}: parallel loop body emitted `{banned}`") + sign_var = _ident("l", s.var) + out = [f"typedef struct {{", *fields, f"}} {ctx_t};", "", + f"static void {body_f}(const void *tp_c, uint64_t tp_k0, uint64_t tp_k1)", "{", + f" const {ctx_t} *c = (const {ctx_t} *)tp_c;", + " uint64_t tp_k;", *decls] + for ct, n in body_temps: + out.append(f" {ct}{'' if ct.endswith('*') else ' '}{n} = 0;") + out.append(" (void)c;" + "".join(f" (void){_ident('l', x)};" for x in sorted(scalars))) + out.append(f" for (tp_k = tp_k0; tp_k < tp_k1; tp_k++) {{") + out.append(f" {sign_var} = (int64_t)((uint64_t)c->tp_ts + tp_k * UINT64_C({mag}));") + out += body_lines + out += [" }", "}", ""] + self.par_defs.append("\n".join(out)) + w = max(1, info.weight) + self.emit(f"if ({tn} >= (uint64_t)((TP_PAR_MIN_WORK + {w - 1}) / {w}) && tp_par_threads() > 1) {{") + self.depth += 1 + self.emit(f"{ctx_t} tp_pr;") + self.emit(f"tp_pr.tp_ts = {ts};") + for line in sets: + self.emit(line) + self.emit(f"(void)tp_par_run({body_f}, &tp_pr, {tn});") + self.emit(f"{_ident('l', s.var)} = (int64_t)((uint64_t){ts} + ({tn} - 1u) * UINT64_C({mag}));") + self.depth -= 1 + self.emit("} else {") + def st_Return(self, s: ir.Return) -> None: rt = self.f.returns if rt is Type.NONE: @@ -2217,8 +2328,13 @@ def generate(self) -> str: """The impl (preceded by its fast twin when bounded) and the wrapper.""" parts = [] if self.f.name in self.mg.bounded: - parts.append(_FnGen(self.mg, self.f, fast=True).impl()) - parts.append(self.impl()) + twin = _FnGen(self.mg, self.f, fast=True) + twin_text = twin.impl() + parts += twin.par_defs + parts.append(twin_text) + impl_text = self.impl() + parts += self.par_defs + parts.append(impl_text) parts.append(self.wrapper()) return "\n".join(parts) @@ -2631,6 +2747,23 @@ def _state_functions(mg: _ModGen) -> str: return "\n".join(L) +def _par_stats_function() -> str: + """`__typedpython_parallel__()`: (regions that used workers, workers started, failed thread + creations) since import, so a test can prove the parallel path ran (SPEC N-15).""" + return "\n".join([ + "static PyObject *tp_par_stats_fn(PyObject *tp_module, PyObject *tp_unused)", + "{", + " int64_t tp_v[3];", + " (void)tp_module; (void)tp_unused;", + " tp_par_stats(tp_v);", + " return Py_BuildValue(\"(LLL)\", (long long)tp_v[0], (long long)tp_v[1], (long long)tp_v[2]);", + "}", + "", + "static PyMethodDef tp_par_stats_def = {\"__typedpython_parallel__\", tp_par_stats_fn, METH_NOARGS, NULL};", + "", + ]) + + def _class_info_function(mg: _ModGen) -> str: """`__typedpython_class_info__()`: {class name: {compiled, captured_tag, tag, current, slow, refreshed, watched}}, introspection of the guards (module docstring).""" @@ -2792,6 +2925,12 @@ def _exec_function(mg: _ModGen) -> str: " if (tp_cf == NULL || PyDict_SetItemString(tp_dict, \"__typedpython_class_info__\", tp_cf) < 0) goto tp_done;", " tp_release(&tp_cf);", ] + if mg.has_par: + L += [ + " tp_cf = PyCFunction_NewEx(&tp_par_stats_def, tp_module, NULL);", + " if (tp_cf == NULL || PyDict_SetItemString(tp_dict, \"__typedpython_parallel__\", tp_cf) < 0) goto tp_done;", + " tp_release(&tp_cf);", + ] L += [ " tp_zero = PyLong_FromLong(0);", " if (tp_zero == NULL || PyDict_SetItemString(tp_dict, \"__typedpython_deopts__\", tp_zero) < 0) goto tp_done;", diff --git a/python-multiplatform-ksp/src/main/python/typedpython/frontend.py b/python-multiplatform-ksp/src/main/python/typedpython/frontend.py index a5314a03..2a1a9e58 100644 --- a/python-multiplatform-ksp/src/main/python/typedpython/frontend.py +++ b/python-multiplatform-ksp/src/main/python/typedpython/frontend.py @@ -237,9 +237,28 @@ def lower(path: Path) -> ir.Module: else: skipped[name] = result functions.sort(key=lambda f: f.source_line) + marked = _parallel_marked(tree, source) + functions = [dataclasses.replace(f, parallel=True) if f.name in marked else f for f in functions] return ir.Module(path.stem, tuple(functions), skipped, tuple(class_decls.values())) +PARALLEL_MARKER = "# typedpython: parallel" + + +def _parallel_marked(tree: ast.Module, source: str) -> set[str]: + """Module-level functions the user marked for SPEC N-15: the comment line `# typedpython: parallel` + directly above the `def` (above its decorators if it has any). A comment, like the module marker, + so the interpreted module is unchanged and needs no runtime name.""" + lines = source.splitlines() + out: set[str] = set() + for node in tree.body: + if isinstance(node, ast.FunctionDef): + first = min([node.lineno] + [d.lineno for d in node.decorator_list]) + if first >= 2 and lines[first - 2].strip() == PARALLEL_MARKER: + out.add(node.name) + return out + + def _module_info(tree: ast.Module) -> _ModuleInfo: bindings: dict[str, list[tuple[int, str]]] = {} diff --git a/python-multiplatform-ksp/src/main/python/typedpython/ir.py b/python-multiplatform-ksp/src/main/python/typedpython/ir.py index 45fc41bc..ba31d3c9 100644 --- a/python-multiplatform-ksp/src/main/python/typedpython/ir.py +++ b/python-multiplatform-ksp/src/main/python/typedpython/ir.py @@ -650,6 +650,22 @@ class While(Stmt): body: tuple[Stmt, ...] +@dataclass(frozen=True) +class ParInfo: + """Why a `ForRange` may run its iterations on several threads (SPEC N-15, step 1): the body is + only `a[var] = ` stores, `stored` are the arrays stored into (each at the loop + variable, proven in bounds, so distinct iterations write distinct elements) and `read` the arrays + read, all at the loop variable and proven in bounds. The expression nodes are Const, scalar + Local (not assigned in the body), ToFloat of an I64, F64 + - * and negation, so no node can fail, + deopt or touch a Python object, and `weight` is their count (the cost rule's per-iteration work). + Arrays are distinct objects (entry guard), so a store at `var` cannot alias a read of another + array.""" + + stored: tuple[str, ...] + read: tuple[str, ...] + weight: int = 1 + + @dataclass(frozen=True) class ForRange(Stmt): """`for var in range(start, stop, step)`: `step` is a non-zero I64 Const; `start`/`stop` are @@ -663,6 +679,10 @@ class ForRange(Stmt): stop: Expr step: Const body: tuple[Stmt, ...] + # SPEC N-15 (#151, #277): set only by `verify` (like `proven`), which recomputes it from the body + # and rejects a loop whose declared value differs (`verify/par-loop`). The front end leaves it + # None. Not None: the iterations may run on several C threads (see `ParInfo`). None: sequential. + par: ParInfo | None = None @dataclass(frozen=True) @@ -783,6 +803,9 @@ class Function: # rejects a function whose declared entries differ (`verify/virtual-escape`). The front end # leaves it empty. virtual: tuple[str, ...] = () + # SPEC N-15: the user marked this function with the comment line `# typedpython: parallel` + # directly above its `def`. Only a marked function can hold a `ForRange.par`. + parallel: bool = False @dataclass(frozen=True) 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 4cf0b7cf..52dc203e 100644 --- a/python-multiplatform-ksp/src/main/python/typedpython/runtime/API.md +++ b/python-multiplatform-ksp/src/main/python/typedpython/runtime/API.md @@ -472,6 +472,8 @@ static or module-state field must be added here): | module globals (class names, entry globals) | module dict | strong reads | `tp_global` and `tp_class_global_ok` use `PyDict_GetItemRef` | | a compiled class's instances | other threads' objects | CPython | `PyMember_GetOne` / `SetOne` are what the interpreter's member descriptors call (they take the object's lock on free-threaded builds); `tp_new_fixed` writes only its own fresh object | | `errno` (math helpers) | C library | thread-local | per thread by POSIX | +| `tp_par_cap_v`, `tp_par_regions_v`, `tp_par_started_v`, `tp_par_failed_v` | static in the module | atomic | relaxed loads and stores; the cap is computed identically by any thread that finds it unset and never changes after | +| a parallel region's arrays (`ForRange.par`) | caller's copied-in arrays | disjoint ranges | each worker writes `data[slot]` and `dirty[slot]` only for its own iterations (verifier: store index is the loop variable); the caller holds the list's critical section and joins every worker before the write-back | Borrowed references are never held across anything another thread can free. Replaced in this change: `tp_global`: `PyDict_GetItemWithError` (module globals, builtins) and `PyDict_GetItemString` @@ -488,6 +490,32 @@ Memory orders: relaxed loads and stores for the tags and the epoch (the synchron publishes a class or global change to another thread, a lock or an event, orders them), acquire-release for the epoch's increment. `tp_runtime.h` needs the GCC/Clang `__atomic` builtins. +## Parallel regions (SPEC N-15, issues #151, #277) +A `ForRange` the verifier marked `par` (`ir.ParInfo`; marked function, `a[i] = F64 expression` stores +only) 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 */ + #define TP_PAR_HARD_CAP 64 + #define TP_PAR_MIN_CHUNK 1024u /* fewest iterations worth a thread (placeholder) */ + #define TP_PAR_ALIGN 8u /* chunk starts are multiples of 8 iterations (one cache line of doubles) */ + #define TP_PAR_HAVE_THREADS /* defined with pthreads; -DTP_NO_THREADS compiles the threads out */ + #define TP_PAR_TEST_FAIL_CREATE /* test hook: every pthread_create reports failure */ + typedef void (*tp_par_fn)(const void *ctx, uint64_t lo, uint64_t hi); + int tp_par_threads(void); /* the cap, read once from TYPEDPYTHON_THREADS, else min(cores, 8) */ + int tp_par_run(tp_par_fn fn, const void *ctx, uint64_t n); /* runs [0, n) in contiguous chunks, joined before return; returns workers started */ + void tp_par_stats(int64_t out[3]); /* regions with workers, workers started, failed creations */ +`tp_par_run` always completes the whole range. The calling thread runs the first chunk; a chunk whose +`pthread_create` failed runs on the calling thread after it (fewer workers, down to none: the sequential +loop, no exception, not a semantic event). Workers start with all signals blocked, so Python signal +handlers run on Python threads only. The chunk function must be safe for any chunk on any thread; +generated code passes a context struct of raw array pointers and scalar copies. A module with a +`par` loop also exports `__typedpython_parallel__()` returning `(regions, workers_started, +failed_creations)`, so tests can prove the parallel path ran. Free-threaded state: the four counters and +the thread cap are atomics (relaxed); a region's chunk bounds, jobs and thread handles live on the +caller's stack. The caller does not poll the eval breaker inside a region (array functions do not poll +anyway, #152). + ## 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 typedpython.importgate.uninstall() 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 01c19e39..57e4334d 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 @@ -2915,4 +2915,167 @@ static inline int tp_traced(PyObject *module, PyThreadState *ts, PyCodeObject *c } } +/* ---- Parallel regions (SPEC N-15, #151, #277; runtime/API.md "Parallel regions") ------------------ + * A generated `ForRange` the verifier marked `par` hands its trip count to `tp_par_run`, which splits + * it into contiguous chunks and runs them on per-region C threads, joined before it returns. The + * chunk function touches no Python object and calls no Python API (the verifier proves the body is + * F64 arithmetic on raw arrays), so the workers need no thread state. The calling thread keeps its + * attached thread state, its critical section and its place in the call: it runs the first chunk + * itself. A thread that cannot be created is not an error: its chunk runs on the calling thread. + * Without pthreads (TP_NO_THREADS, wasm, Windows) tp_par_threads() is 1 and the region is dead code. */ +#ifndef TP_PAR_MIN_WORK +/* PLACEHOLDER, not a measurement: units are "weighted loop iterations" (trip count times the loop's + * node count `ParInfo.weight`). It is to be measured together with thread creation cost in a quiet + * window and recorded on #151 (design/typedpython-parallel-loops.md, decision 7). */ +#define TP_PAR_MIN_WORK 65536 +#endif +#ifndef TP_PAR_MAX_THREADS +#define TP_PAR_MAX_THREADS 8 /* default cap; TYPEDPYTHON_THREADS overrides, up to TP_PAR_HARD_CAP */ +#endif +#define TP_PAR_HARD_CAP 64 +#define TP_PAR_MIN_CHUNK 1024u /* fewest iterations worth a thread of their own (placeholder, #151) */ +#define TP_PAR_ALIGN 8u /* chunk starts are multiples of 8 iterations = one 64-byte line of doubles */ + +#if !defined(TP_NO_THREADS) && (defined(__unix__) || defined(__APPLE__)) && !defined(__EMSCRIPTEN__) +#define TP_PAR_HAVE_THREADS 1 +#include +#include +#include +#endif + +typedef void (*tp_par_fn)(const void *ctx, uint64_t lo, uint64_t hi); + +static int tp_par_cap_v = 0; /* 0: not read yet; then the thread cap (>= 1), set once */ +static int64_t tp_par_regions_v = 0; /* regions that ran with at least one worker */ +static int64_t tp_par_started_v = 0; /* worker threads started */ +static int64_t tp_par_failed_v = 0; /* pthread_create calls that failed (their chunk ran on the caller) */ + +/* The thread cap: TYPEDPYTHON_THREADS (a positive integer, 1 disables), else min(cores, 8); read once, + * immutable afterwards. Two threads racing on the first call compute the same value. */ +static inline int tp_par_threads(void) +{ + int c = __atomic_load_n(&tp_par_cap_v, __ATOMIC_RELAXED); + if (c != 0) { + return c; + } + c = 1; +#ifdef TP_PAR_HAVE_THREADS + { + const char *e = getenv("TYPEDPYTHON_THREADS"); + long v = 0; + if (e != NULL && e[0] != '\0') { + char *end = NULL; + v = strtol(e, &end, 10); + if (end == e || *end != '\0' || v < 1) { + v = 0; /* not a positive integer: the default */ + } + } + if (v == 0) { + long cores = sysconf(_SC_NPROCESSORS_ONLN); + v = cores < 1 ? 1 : cores; + if (v > TP_PAR_MAX_THREADS) { + v = TP_PAR_MAX_THREADS; + } + } + if (v > TP_PAR_HARD_CAP) { + v = TP_PAR_HARD_CAP; + } + c = (int)v; + } +#endif + __atomic_store_n(&tp_par_cap_v, c, __ATOMIC_RELAXED); + return c; +} + +/* out[0] regions that used workers, out[1] workers started, out[2] failed thread creations */ +static inline void tp_par_stats(int64_t out[3]) +{ + out[0] = __atomic_load_n(&tp_par_regions_v, __ATOMIC_RELAXED); + out[1] = __atomic_load_n(&tp_par_started_v, __ATOMIC_RELAXED); + out[2] = __atomic_load_n(&tp_par_failed_v, __ATOMIC_RELAXED); +} + +#ifdef TP_PAR_HAVE_THREADS +typedef struct { tp_par_fn fn; const void *ctx; uint64_t lo, hi; } tp_par_job; + +static void *tp_par_worker(void *p) +{ + const tp_par_job *j = (const tp_par_job *)p; + j->fn(j->ctx, j->lo, j->hi); + return NULL; +} +#endif + +/* Runs fn(ctx, lo, hi) over [0, n) in contiguous chunks and returns after every chunk is done (always + * the whole range, whatever the thread creation does). Returns the number of workers started (0: the + * range ran on the calling thread alone). The chunks are disjoint and `fn` must be safe to run on + * any thread for any chunk; results do not depend on the chunking. */ +static inline int tp_par_run(tp_par_fn fn, const void *ctx, uint64_t n) +{ +#ifdef TP_PAR_HAVE_THREADS + int t = tp_par_threads(); + uint64_t maxt = n / TP_PAR_MIN_CHUNK; + if ((uint64_t)t > maxt) { + t = (int)maxt; + } + if (t > 1) { + tp_par_job jobs[TP_PAR_HARD_CAP]; + pthread_t th[TP_PAR_HARD_CAP]; + unsigned char live[TP_PAR_HARD_CAP]; + uint64_t chunk = (n + (uint64_t)t - 1u) / (uint64_t)t; + int k, started = 0, jn = 0; + sigset_t all, old; + chunk = (chunk + (TP_PAR_ALIGN - 1u)) & ~(uint64_t)(TP_PAR_ALIGN - 1u); + for (k = 0; k < t; k++) { + uint64_t lo = (uint64_t)k * chunk; + if (lo >= n) { + break; + } + jobs[jn].fn = fn; + jobs[jn].ctx = ctx; + jobs[jn].lo = lo; + jobs[jn].hi = lo + chunk < n ? lo + chunk : n; + live[jn] = 0; + jn++; + } + /* workers start with every signal blocked, so a handler only ever runs on a Python thread */ + sigfillset(&all); + pthread_sigmask(SIG_SETMASK, &all, &old); + for (k = 1; k < jn; k++) { + int rc; +#ifdef TP_PAR_TEST_FAIL_CREATE + rc = 11; /* EAGAIN: tests make every creation fail */ +#else + rc = pthread_create(&th[k], NULL, tp_par_worker, &jobs[k]); +#endif + if (rc == 0) { + live[k] = 1; + started++; + } else { + __atomic_fetch_add(&tp_par_failed_v, 1, __ATOMIC_RELAXED); + } + } + pthread_sigmask(SIG_SETMASK, &old, NULL); + fn(ctx, jobs[0].lo, jobs[0].hi); + for (k = 1; k < jn; k++) { + if (!live[k]) { + fn(ctx, jobs[k].lo, jobs[k].hi); /* creation failed: this chunk runs here */ + } + } + for (k = 1; k < jn; k++) { + if (live[k]) { + pthread_join(th[k], NULL); + } + } + if (started > 0) { + __atomic_fetch_add(&tp_par_regions_v, 1, __ATOMIC_RELAXED); + __atomic_fetch_add(&tp_par_started_v, started, __ATOMIC_RELAXED); + } + return started; + } +#endif + fn(ctx, 0, n); + return 0; +} + #endif /* TP_RUNTIME_H */ diff --git a/python-multiplatform-ksp/src/main/python/typedpython/verify.py b/python-multiplatform-ksp/src/main/python/typedpython/verify.py index 4905ee17..1e34f73e 100644 --- a/python-multiplatform-ksp/src/main/python/typedpython/verify.py +++ b/python-multiplatform-ksp/src/main/python/typedpython/verify.py @@ -47,6 +47,9 @@ verify/array-open a function with an ArrayParam runs user code (#203) verify/buffer a malformed `ArrayParam.buffer` (only None and "array"), or a CopyArray of a buffer array (#149) + verify/par-loop a `ForRange.par` (SPEC N-15, #277) that the verifier does not recompute to + the same value: the function is not marked `parallel`, or the loop is + not eligible (see "Parallel loops" below) verify/class a class named by FieldGet/FieldSet/New/IsExact/CheckExact/Param.cls is not a (valid) ClassDecl of the module, a field is not one of its fields, a New names a class without trivial_init or has the wrong arity, or a @@ -192,6 +195,26 @@ proves constant indices and bounded index expressions; a negative index is never proven (the `proven` flag means 0 <= index < len). An array assigned twice or more has no known length. +Parallel loops (SPEC N-15, #277; step 1). `ForRange.par` is never trusted: `_par_info` recomputes it +for every loop; a declared value (the front end declares none) must equal it, else `verify/par-loop`. A loop +is eligible only if ALL of these hold: + * the function is marked (`Function.parallel`); + * the loop is `range(c >= 0, len(A), step > 0)` over an array `A` the body does not reassign (the + same condition that lets the verifier prove bounds, so every access below is `proven`); + * the body is a non-empty sequence of `StoreIndex` statements and nothing else (no Assign, no + nested loop, no If, no Break/Continue/Return, no call); + * every store is `a[var] = value` with the loop variable itself as index, `proven`, into an F64 + list or local array (not a buffer), so distinct iterations write distinct elements; + * `value` is built from F64 Const, an F64 scalar Local (nothing in the body assigns it), a ToFloat + of an I64 Const or Local, `Index(b, var)` proven on an F64 non-buffer array, F64 ADD/SUB/MUL + (not `np`), and F64 NEG. No division (ZeroDivisionError), no I64 arithmetic (overflow), no + comparison, no call, no object. So no node can fail, deopt or touch a Python object, and a read + of `b[var]` where `b` is another array cannot alias a written element (array parameters are + distinct lists by the entry guard, local arrays are distinct allocations), and a read of the + stored array itself is the same element the iteration writes. +Anything else compiles sequentially, unchanged. This cannot find other eligible shapes (reductions, +stencils, calls, nested tiles); 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. @@ -243,6 +266,7 @@ ) from .ir import CheckExact, CheckMethod, ClassDecl, FieldGet, FieldSet, Is, IsExact, New from .ir import BuildString, Contains, ForObj, Format, GetItem, NewDict, NewList, StoreItem +from .ir import ParInfo DA = "verify/definite-assignment" MISSING_RETURN = "verify/missing-return" @@ -267,6 +291,7 @@ BUFFER = "verify/buffer" CLASS = "verify/class" FIELD = "verify/field-unproved" +PAR_LOOP = "verify/par-loop" @dataclass(frozen=True) @@ -1609,7 +1634,65 @@ def s_ForRange(self, s: ForRange, state: _State): after: _State = state for br in loop.breaks: after = _meet(after, br) - return dataclasses.replace(s, start=start, stop=stop, body=body), after + par = self._par_info(s, step_ok, body, loop) + if s.par is not None and s.par != par: + self.error(PAR_LOOP, f"ForRange over '{var}' declares par={s.par!r}, but the verifier " + f"computes {par!r} (SPEC N-15)") + return dataclasses.replace(s, start=start, stop=stop, body=body, par=par), after + + def _par_info(self, s: ForRange, step_ok: bool, body: tuple, loop: _Loop) -> ParInfo | None: + """SPEC N-15 step 1: the ParInfo of this loop, or None (see "Parallel loops" above).""" + if self.fn.parallel is not True or not step_ok or loop.bounds_array is None: + return None + if self.diags or not isinstance(body, tuple) or not body: + return None + var = s.var + stored: list[str] = [] + read: list[str] = [] + weight = [0] + + def array_ok(name: str) -> bool: + t = self.arrays.get(name) or self.larrays.get(name) + return t is Type.F64_ARRAY and name not in self.buffers + + def at_var(ix: object, proven: object) -> bool: + return proven is True and type(ix) is Local and ix.name == var and ix.type is Type.I64 + + def ok(e: object) -> bool: + weight[0] += 1 + t = type(e) + if t is Const: + return e.type is Type.F64 and type(e.value) is float + if t is Local: + return (e.type is Type.F64 and e.name != var and e.name not in self.arrays + and e.name not in self.larrays and (e.name in self.scalars or e.name in self.entry)) + if t is ToFloat: + o = e.operand + return (e.type is Type.F64 and getattr(o, "type", None) is Type.I64 + and ((type(o) is Local and o.name not in self.arrays + and o.name not in self.larrays) + or (type(o) is Const and type(o.value) is int))) + if t is Index: + if e.type is Type.F64 and array_ok(e.array) and at_var(e.index, e.proven): + read.append(e.array) + return True + return False + if t is BinOp: + return (e.type is Type.F64 and e.left.type is Type.F64 and e.right.type is Type.F64 + and e.op in (BinOpKind.ADD, BinOpKind.SUB, BinOpKind.MUL) and not e.np + and e.inplace is False and ok(e.left) and ok(e.right)) + if t is UnaryOp: + return e.type is Type.F64 and e.op is UnaryOpKind.NEG and ok(e.operand) + return False + + 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 + weight[0] += 2 + if not ok(st.value) or st.value.type is not Type.F64: + return None + stored.append(st.array) + return ParInfo(tuple(sorted(set(stored))), tuple(sorted(set(read))), weight[0]) def s_Return(self, s: Return, state: _State): returns = self.fn.returns diff --git a/python-multiplatform-ksp/src/test/python/test_parallel_loops.py b/python-multiplatform-ksp/src/test/python/test_parallel_loops.py new file mode 100644 index 00000000..1aa28652 --- /dev/null +++ b/python-multiplatform-ksp/src/test/python/test_parallel_loops.py @@ -0,0 +1,419 @@ +"""TypedPython parallel order-independent loops, step 1 (SPEC N-15, #151, #277). + +Step 1 scope: in a function marked by the comment line `# typedpython: parallel` directly above its +`def`, a `for i in range(len(a))` loop whose body is only stores `a[i] = ` (the same +index i, each proven in bounds), over list copy-in or local arrays, with values built from constants, +loop-invariant scalars, `float(i)`, reads `b[i]` at the same i (proven) and F64 + - * and negation, +runs on several C threads above TP_PAR_MIN_WORK. The verifier recomputes the set (`ForRange.par`) and +rejects a declared one that differs (`verify/par-loop`). Everything else compiles as before. + +Red record: written before the implementation. Before it, `ir.ForRange` had no `par`, `ir.Function` +had no `parallel`, the marker comment was ignored and the generated module had no +`__typedpython_parallel__`; every test below that touches those failed (see the report in #277). +""" +from __future__ import annotations + +import dataclasses +import json +import os +import struct +import subprocess +import sys +import sysconfig +import textwrap +from pathlib import Path + +import pytest + +from typedpython import cbuild, cgen, frontend, ir, verify + +FREE_THREADED = bool(sysconfig.get_config_var("Py_GIL_DISABLED")) +PAR_RULE = "verify/par-loop" + +SOURCE = '''\ +# typedpython: compiled + +# typedpython: parallel +def scale(xs: list[float], a: float, b: float) -> None: + for i in range(len(xs)): + xs[i] = xs[i] * a + b + + +def scale_unmarked(xs: list[float], a: float, b: float) -> None: + for i in range(len(xs)): + xs[i] = xs[i] * a + b + + +# typedpython: parallel +def ramp(xs: list[float], a: float) -> None: + for i in range(len(xs)): + xs[i] = -(float(i) * a) + xs[i] + + +# typedpython: parallel +def two_stores(xs: list[float], ys: list[float], a: float) -> None: + for i in range(len(xs)): + xs[i] = xs[i] + a + xs[i] = xs[i] * xs[i] + + +# typedpython: parallel +def neighbour(xs: list[float]) -> None: + for i in range(len(xs) - 1): + xs[i] = xs[i + 1] + + +# typedpython: parallel +def total(xs: list[float]) -> float: + s: float = 0.0 + for i in range(len(xs)): + s += xs[i] + return s + + +def helper(x: float) -> float: + return x * 2.0 + + +# typedpython: parallel +def with_call(xs: list[float]) -> None: + for i in range(len(xs)): + xs[i] = helper(xs[i]) + + +# typedpython: parallel +def divides(xs: list[float], d: float) -> None: + for i in range(len(xs)): + xs[i] = xs[i] / d + + +# typedpython: parallel +def local_fill(a: float) -> float: + t = [0.0] * 300000 + for i in range(len(t)): + t[i] = float(i) * a + return t[7] + t[299999] + + +# typedpython: parallel +def local_dyn(n: int, a: float) -> float: + t = [1.0] * n + for i in range(len(t)): + t[i] = t[i] + float(i) * a + return t[n - 1] +''' + + +def build(tmp: Path, source: str = SOURCE, flags: tuple[str, ...] = cbuild.DEFAULT_FLAGS, name: str = "parmod"): + path = tmp / f"{name}.py" + path.write_text(source) + lowered = frontend.lower(path) + proved, diags = verify.verify(lowered) + c_source = cgen.generate(proved, path) + so = cbuild.build(c_source, name, tmp / "out", flags=flags) + return proved, c_source, so, lowered + + +@pytest.fixture(scope="module") +def built(tmp_path_factory): + d = tmp_path_factory.mktemp("par") + return build(d) + + +@pytest.fixture(scope="module") +def mod(built): + return cbuild.load("parmod", built[2]) + + +def fn(built, name): + return next(f for f in built[0].functions if f.name == name) + + +def par_loops(f): + out = [] + + def walk(body): + for s in body: + if isinstance(s, ir.ForRange): + out.append(s) + walk(s.body) + elif isinstance(s, (ir.If, ir.While)): + walk(s.body) + walk(getattr(s, "orelse", ())) + walk(f.body) + return out + + +def bits(xs): + return struct.pack(f"<{len(xs)}d", *xs) + + +def data(n, seed=1): + import random + r = random.Random(seed) + out = [r.uniform(-1e3, 1e3) for _ in range(n)] + for k, v in enumerate((float("nan"), float("inf"), -0.0, 5e-324, float("-inf"))): + if k < n: + out[(k * 7919) % n] = v + return out + + +def regions(m): + return m.__typedpython_parallel__()[0] + + +SIZES = [0, 1, 2, 7, 1000, 200003, 1048576] + + +# --- marking and eligibility (IR / verifier markers, not timing) ------------------------------------- + +def test_marker_comment_sets_function_parallel(built): + assert fn(built, "scale").parallel is True + assert fn(built, "scale_unmarked").parallel is False + + +@pytest.mark.parametrize("name", ["scale", "ramp", "two_stores", "local_fill", "local_dyn"]) +def test_eligible_loops_are_marked_par(built, name): + loops = par_loops(fn(built, name)) + assert loops and all(l.par is not None for l in loops), name + + +@pytest.mark.parametrize("name", ["neighbour", "total", "with_call", "divides"]) +def test_ineligible_loops_stay_sequential(built, name): + f = fn(built, 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_function_never_parallel(built): + assert all(l.par is None for l in par_loops(fn(built, "scale_unmarked"))) + + +def test_front_end_never_sets_par(built): + lowered = built[3] + for f in lowered.functions: + assert all(l.par is None for l in par_loops(f)), f.name + + +def test_verifier_rejects_a_declared_par_on_an_ineligible_loop(built): + f = fn(built, "neighbour") + loop = par_loops(f)[0] + fake = dataclasses.replace(loop, par=ir.ParInfo(("xs",), ())) + bad = dataclasses.replace(f, body=tuple(fake if s is loop else s for s in f.body)) + mod = dataclasses.replace(built[0], functions=(bad,), skipped={}) + proved, diags = verify.verify(mod) + assert not proved.functions + assert any(d.rule == PAR_RULE for d in diags), diags + + +def test_verifier_rejects_a_declared_par_in_an_unmarked_function(built): + f = fn(built, "scale") + loop = par_loops(f)[0] + fake = dataclasses.replace(loop, par=loop.par or ir.ParInfo(("xs",), ("xs",))) + bad = dataclasses.replace(f, parallel=False, body=tuple(fake if s is loop else s for s in f.body)) + mod = dataclasses.replace(built[0], functions=(bad,), skipped={}) + proved, diags = verify.verify(mod) + assert not proved.functions + assert any(d.rule == PAR_RULE for d in diags), diags + + +def test_verifier_recomputes_par_when_it_is_missing(built): + """The front end leaves `par` empty; the proved module has it, set by the verifier alone.""" + assert par_loops(fn(built, "scale"))[0].par is not None + + +# --- generated C ------------------------------------------------------------------------------------- + +def test_c_has_a_region_only_for_eligible_loops(built): + c_text = built[1] + assert "tp_par_run" in c_text + import re + owners = set(re.findall(r"static void tp_par_(\w+?)(?:_fast)?_\d+_body\(", c_text)) + assert owners == {"scale", "ramp", "two_stores", "local_fill", "local_dyn"}, owners + assert "TP_PAR_MIN_WORK" in c_text + + +def test_no_par_code_without_a_marker(tmp_path): + src = SOURCE.replace("# typedpython: parallel\n", "") + _, c_text, _, _ = build(tmp_path, src, name="nomark") + assert "tp_par_run(" not in c_text and "__typedpython_parallel__" not in c_text + + +# --- differential results vs the interpreter --------------------------------------------------------- + +@pytest.mark.parametrize("n", SIZES) +def test_scale_matches_interpreter_bit_for_bit(mod, n): + xs = data(n) + want, got = list(xs), list(xs) + mod.__typedpython_interpreted__["scale"](want, 1.0000001, -3.25) + mod.scale(got, 1.0000001, -3.25) + assert bits(got) == bits(want) + + +@pytest.mark.parametrize("n", [0, 5, 200003]) +def test_ramp_matches(mod, n): + xs = data(n, 3) + want, got = list(xs), list(xs) + mod.__typedpython_interpreted__["ramp"](want, 0.1) + mod.ramp(got, 0.1) + assert bits(got) == bits(want) + + +@pytest.mark.parametrize("n", [0, 3, 200003]) +def test_two_stores_match(mod, n): + xs = data(n, 4) + ys = [0.0] * 4 + want, got = list(xs), list(xs) + mod.__typedpython_interpreted__["two_stores"](want, list(ys), 0.3) + mod.two_stores(got, list(ys), 0.3) + assert bits(got) == bits(want) + + +def test_local_array_loops_match(mod): + for a in (0.5, -2.75, float("inf"), float("nan")): + w = mod.__typedpython_interpreted__["local_fill"](a) + g = mod.local_fill(a) + assert struct.pack(" 1: + assert regions(mod) == r0 + 1 # above: one region + r1 = regions(mod) + mod.scale_unmarked([1.0] * 1048576, 2.0, 1.0) + mod.neighbour([1.0] * 1048576) + mod.total([1.0] * 1048576) + mod.with_call([1.0] * 1048576) + assert regions(mod) == r1 # unmarked / ineligible: never + + +def test_list_untouched_when_the_call_raises_before_the_loop(mod): + with pytest.raises(TypeError): + mod.scale([1.0] * 100000, 2.0) # wrong arity deopts to the interpreter's TypeError + + +# --- child processes: thread counts, creation failure, concurrency ----------------------------------- + +CHILD = """\ +import importlib.machinery, importlib.util, json, struct, sys, threading +path = sys.argv[1] +name = path.rsplit("/", 1)[-1].split(".")[0] +loader = importlib.machinery.ExtensionFileLoader(name, path) +spec = importlib.util.spec_from_file_location(name, path, loader=loader) +m = importlib.util.module_from_spec(spec) +loader.exec_module(m) +""" + + +def run_child(so: Path, body: str, env: dict | None = None, timeout: int = 300) -> dict: + e = dict(os.environ) + e.pop("TYPEDPYTHON_THREADS", None) + e.update(env or {}) + proc = subprocess.run([sys.executable, "-c", CHILD + textwrap.dedent(body), str(so)], + capture_output=True, text=True, timeout=timeout, env=e) + assert proc.returncode == 0, f"child failed ({proc.returncode}):\n{proc.stdout}\n{proc.stderr}" + return json.loads(proc.stdout.strip().splitlines()[-1]) + + +THREAD_BODY = """ + xs = [((k * 2654435761) % 1000003) / 977.0 - 500.0 for k in range(400001)] + m.scale(xs, 1.0000003, 0.125) + print(json.dumps({"hex": struct.pack("<%dd" % len(xs), *xs).hex(), "regions": m.__typedpython_parallel__()[0]})) +""" + + +def test_identical_bytes_for_every_thread_count(built): + so = built[2] + results = {t: run_child(so, THREAD_BODY, {"TYPEDPYTHON_THREADS": str(t)}) for t in (1, 2, 3, 7, 64)} + first = results[1] + assert first["regions"] == 0 # 1 disables the feature + for t, r in results.items(): + assert r["hex"] == first["hex"], t + if (os.cpu_count() or 1) > 1: + assert results[2]["regions"] == 1 and results[7]["regions"] == 1 + + +def test_thread_creation_failure_falls_back_to_sequential(tmp_path): + """Every pthread_create fails (test hook): the call still completes, correctly, on the caller.""" + _, _, so, _ = build(tmp_path, flags=cbuild.DEFAULT_FLAGS + ("-DTP_PAR_TEST_FAIL_CREATE=1",), name="parfail") + r = run_child(so, THREAD_BODY, {"TYPEDPYTHON_THREADS": "4"}) + ok = run_child(built_so_for_compare(tmp_path), THREAD_BODY, {"TYPEDPYTHON_THREADS": "1"}) + assert r["hex"] == ok["hex"] + info = run_child(so, """ + m.scale([1.0] * 400001, 2.0, 1.0) + print(json.dumps(list(m.__typedpython_parallel__()))) + """, {"TYPEDPYTHON_THREADS": "4"}) + assert info[2] > 0 # creation failures were counted, none raised + + +def built_so_for_compare(tmp_path): + return build(tmp_path, name="parcmp")[2] + + +@pytest.mark.skipif(not FREE_THREADED, reason="needs a free-threaded CPython") +def test_concurrent_calls_from_several_python_threads(built): + r = run_child(built[2], """ + N = 6 + base = [((k * 40503) % 65521) / 13.0 for k in range(262144)] + want = list(base) + m.__typedpython_interpreted__["scale"](want, 1.5, -0.5) + bad = [] + errors = [] + def work(k): + def go(): + try: + for _ in range(6): + xs = list(base) + m.scale(xs, 1.5, -0.5) + if xs != want: + bad.append(k) + except BaseException as e: + errors.append(repr(e)) + return go + ts = [threading.Thread(target=work(k)) for k in range(N)] + for t in ts: t.start() + for t in ts: t.join() + shared = [1.0] * 262144 + shared_bad = [] + def sh(): + for _ in range(4): + m.scale(shared, 1.0, 0.0) # identity: whole-call atomic under the list's section + if shared != [1.0] * 262144: + shared_bad.append(1) + ts = [threading.Thread(target=sh) for _ in range(4)] + for t in ts: t.start() + for t in ts: t.join() + print(json.dumps({"bad": bad, "errors": errors, "shared_bad": shared_bad, + "gil": sys._is_gil_enabled(), "regions": m.__typedpython_parallel__()[0]})) + """, {"TYPEDPYTHON_THREADS": "4"}) + assert r["bad"] == [] and r["errors"] == [] and r["shared_bad"] == [] + assert r["gil"] is False + if (os.cpu_count() or 1) > 1: + assert r["regions"] >= 6 * 6