feat(compiler): a Job.parallel_for worker may read a state but not change one
Every pool thread runs the worker at once with the same states, so a mut state in a worker was a race nothing reported. check_worker_ref refuses a worker whose leading states include a mut one; threads.ludic's total moves into the words the worker is handed, under the mutex. The seeds are regenerated (ludic-dev reseed). ludic-dev test 305 passed, selfhost-test 33 passed. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
parent
7ceefa8c50
commit
7834b26ef8
6 changed files with 13354 additions and 13188 deletions
9
changes/worker-reads-states.md
Normal file
9
changes/worker-reads-states.md
Normal file
|
|
@ -0,0 +1,9 @@
|
||||||
|
bump: minor
|
||||||
|
type: feature
|
||||||
|
**A `Job.parallel_for` worker may read a state but not change one.** Every pool thread runs the
|
||||||
|
worker at once, and the runtime hands each call the same state, so a worker that took a state as
|
||||||
|
`mut` raced on it with nothing to say so. The compiler now refuses it (`fn bump: a worker runs on
|
||||||
|
every core at once, so it may read CountState but not change it`): a result goes into `ctx`, a
|
||||||
|
shared count through a `Sync` handle kept in the state. A function's signature already says what it
|
||||||
|
touches, so this is the whole check - the same one a parallel scheduler needs to run two systems at
|
||||||
|
once. `examples/library/threads.ludic` keeps its total in the words it hands the worker.
|
||||||
|
|
@ -19,7 +19,9 @@ started on first use) and the calling thread, so each index runs exactly once, i
|
||||||
on (`words`, `bytes` or a `pointer`). A worker computes on what it was handed and writes only its own
|
on (`words`, `bytes` or a `pointer`). A worker computes on what it was handed and writes only its own
|
||||||
index's results: it must not `spawn`, `despawn`, `push` onto a list another thread can see, or use
|
index's results: it must not `spawn`, `despawn`, `push` onto a list another thread can see, or use
|
||||||
Http or Audio. `spawn` and `despawn` on a worker stop the program with a located message. Shared
|
Http or Audio. `spawn` and `despawn` on a worker stop the program with a located message. Shared
|
||||||
counters go through `Sync.add`, shared totals behind a `Sync.mutex`.
|
counters go through `Sync.add`, shared totals behind a `Sync.mutex`. A worker may take states, but
|
||||||
|
only to read them: every thread runs it at once, so a worker that takes one as `mut` is refused at
|
||||||
|
compile time (a result goes into `ctx`, a count through a `Sync` handle kept in the state).
|
||||||
|
|
||||||
```ludic
|
```ludic
|
||||||
program Squares {
|
program Squares {
|
||||||
|
|
|
||||||
|
|
@ -9,17 +9,18 @@ program Threads {
|
||||||
state ThreadsState {
|
state ThreadsState {
|
||||||
calls: int = 0 # an atomic handle, made on the main thread before any work
|
calls: int = 0 # an atomic handle, made on the main thread before any work
|
||||||
lock: int = 0 # a mutex handle
|
lock: int = 0 # a mutex handle
|
||||||
total: long = 0 # guarded by `lock`
|
|
||||||
}
|
}
|
||||||
|
|
||||||
# out[i] = i * i: every index written exactly once, by whichever thread claimed it
|
# out[i] = i * i: every index written exactly once, by whichever thread claimed it
|
||||||
function square(i: int, out: words) -> void { out[i] = i * i }
|
function square(i: int, out: words) -> void { out[i] = i * i }
|
||||||
|
|
||||||
# count the call atomically, and add i to a shared total under the mutex
|
# count the call atomically, and add i to a shared total under the mutex. A worker may read a
|
||||||
function tally(threads_st: mut ThreadsState, i: int, ctx: pointer) -> void {
|
# state (the handles in it) but never change one - every thread runs it at once - so the total
|
||||||
|
# lives in what the worker is handed
|
||||||
|
function tally(threads_st: ThreadsState, i: int, total: words) -> void {
|
||||||
Sync.add(threads_st.calls, 1)
|
Sync.add(threads_st.calls, 1)
|
||||||
Sync.lock(threads_st.lock)
|
Sync.lock(threads_st.lock)
|
||||||
threads_st.total = threads_st.total + i
|
total[0] = total[0] + i
|
||||||
Sync.unlock(threads_st.lock)
|
Sync.unlock(threads_st.lock)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -43,10 +44,11 @@ program Threads {
|
||||||
|
|
||||||
threads_st.calls = Sync.atomic()
|
threads_st.calls = Sync.atomic()
|
||||||
threads_st.lock = Sync.mutex()
|
threads_st.lock = Sync.mutex()
|
||||||
Job.parallel_for(N, fn tally, null)
|
let total = words(1)
|
||||||
|
Job.parallel_for(N, fn tally, total)
|
||||||
if Sync.get(threads_st.calls) == N { print(4) }
|
if Sync.get(threads_st.calls) == N { print(4) }
|
||||||
let want: long = N * (N - 1) / 2
|
let want = N * (N - 1) / 2
|
||||||
if threads_st.total == want { print(5) }
|
if total[0] == want { print(5) }
|
||||||
|
|
||||||
let ran = words(N)
|
let ran = words(N)
|
||||||
Job.parallel_for(N, fn placed, ran)
|
Job.parallel_for(N, fn placed, ran)
|
||||||
|
|
|
||||||
|
|
@ -185,4 +185,11 @@ function check_worker_ref(e: Node) -> void {
|
||||||
var ok = len(d.kids) == s + 2 and llty(d.ty) == "void"
|
var ok = len(d.kids) == s + 2 and llty(d.ty) == "void"
|
||||||
if ok { ok = llty(d.kids[s].ty) == "i32" and llty(d.kids[s + 1].ty) == "ptr" }
|
if ok { ok = llty(d.kids[s].ty) == "i32" and llty(d.kids[s + 1].ty) == "ptr" }
|
||||||
if not ok { perr(`fn {e.s}: a worker function takes (i: int, ctx: pointer) and returns nothing`) }
|
if not ok { perr(`fn {e.s}: a worker function takes (i: int, ctx: pointer) and returns nothing`) }
|
||||||
|
# every thread runs it at once: a state it is handed is read by all of them, so none may change one
|
||||||
|
var k = 0
|
||||||
|
while k < s {
|
||||||
|
let p = d.kids[k]
|
||||||
|
if p.uns == 1 { perr(`fn {e.s}: a worker runs on every core at once, so it may read {p.ty} but not change it ({p.s}: mut {p.ty}); write results into ctx, shared counts through Sync.add`) }
|
||||||
|
k += 1
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
13253
selfhost/ludicc.seed.ll
13253
selfhost/ludicc.seed.ll
File diff suppressed because it is too large
Load diff
File diff suppressed because it is too large
Load diff
Loading…
Add table
Add a link
Reference in a new issue