Linked repository source
fixtures/native_components/subsystem/reload_worker.odin
This read-only view comes from the file in the Gododin repository. It retains its design, example or evidence context.
Download the original file · Return to the guide
// The reload fixture's pure background worker (Phase 15 baseline consumers).
// It is a deliberately compact copy of the Cavern Lab worker's proven shape:
// one seeded computation at a time on a replaceable pending slot, a bounded
// one-record handoff, cooperative cancellation at loop checkpoints, a plain
// pthread or Win32 thread created directly with default attributes (the
// engine process rejects core:thread's scheduler-attribute setup), and a
// destroy that cancels, joins, and drains before the worker's memory is
// released. The computation is allocation-free and deterministic in (seed,
// iterations, version); the compiled-in version marker is what lets the
// harness prove the reloaded image's worker code — not a stale entry —
// answered.
//
// This package imports neither Godot, Gododin, nor Raylib.
package reload_worker
import "core:mem"
import "core:sync"
import "core:time"
// The harness rewrites this revision marker on every accepted save, so a
// result can only carry the version of the image that computed it.
WORKER_VERSION :: 0
Status :: enum i32 {
Idle,
Running,
Completed,
Cancelled,
Failed,
}
Error :: enum i32 {
None,
Rejected,
Unavailable,
}
// Vec3 states the locked float_32 Vector3 layout as plain Odin data; typed
// vector inputs borrow the caller's engine storage through this shape.
Vec3 :: struct #align(4) {
x, y, z: f32,
}
// Result is an owned, immutable record transferred exactly once by take.
Result :: struct {
request_id: u64,
hash: u64,
version: i64,
compute_usec: i64,
cancelled: bool,
}
Worker :: struct {
live: bool,
owner_allocator: mem.Allocator,
mutex: sync.Mutex,
wake: sync.Cond,
thread_started: bool,
thread_handle: Worker_Thread_Handle,
// --- guarded by mutex ---
shutdown: bool,
worker_exited: bool,
next_request_id: u64,
active: bool,
active_id: u64,
active_seed: i64,
active_iters: i64,
pending: bool,
pending_id: u64,
pending_seed: i64,
pending_iters: i64,
ready: ^Result,
status: Status,
error: Error,
active_cancelled: bool,
dropped_pending: u64,
cancel_token: bool,
// --- written by the worker, read after join or under mutex ---
compute_usec: i64,
}
worker_create :: proc(allocator: mem.Allocator) -> (^Worker, bool) {
storage, allocation_error := mem.alloc(size_of(Worker), align_of(Worker), allocator)
if allocation_error != nil || storage == nil do return nil, false
worker := cast(^Worker)storage
worker.live = true
worker.owner_allocator = allocator
return worker, true
}
// worker_destroy cancels, joins, and drains before the memory is released.
// The bool is the drain receipt: true only when the handoff was empty at the
// end and every free succeeded.
worker_destroy :: proc(worker: ^^Worker) -> bool {
if worker == nil || worker^ == nil do return false
w := worker^
if !w.live do return false
w.live = false
sync.mutex_lock(&w.mutex)
w.shutdown = true
w.active_cancelled = true
sync.atomic_store(&w.cancel_token, true)
sync.cond_broadcast(&w.wake)
sync.mutex_unlock(&w.mutex)
if w.thread_started {
worker_thread_join(w.thread_handle)
w.thread_started = false
}
drain_ok := true
if w.ready != nil {
record := w.ready
w.ready = nil
if mem.free(rawptr(record), w.owner_allocator) != nil do drain_ok = false
}
allocator := w.owner_allocator
if mem.free(rawptr(w), allocator) != nil do return false
worker^ = nil
return drain_ok
}
// worker_submit validates synchronously, coalesces onto the replaceable
// pending slot while a computation is active, and starts the worker thread on
// first use.
worker_submit :: proc(worker: ^Worker, seed, iterations: i64) -> (i64, Error) {
if worker == nil || !worker.live do return -1, .Unavailable
if seed < 0 || iterations < 1_000 || iterations > 1_000_000_000 do return -1, .Rejected
sync.mutex_lock(&worker.mutex)
defer sync.mutex_unlock(&worker.mutex)
if worker.worker_exited do return -1, .Unavailable
id := worker.next_request_id + 1
if !worker.thread_started {
if !worker_start_thread(worker) {
return -1, .Unavailable
}
}
if worker.active {
if worker.pending do worker.dropped_pending += 1
worker.pending = true
worker.pending_id = id
worker.pending_seed = seed
worker.pending_iters = iterations
} else {
worker.active = true
worker.active_id = id
worker.active_seed = seed
worker.active_iters = iterations
worker.status = .Running
sync.cond_signal(&worker.wake)
}
worker.next_request_id = id
return i64(id), .None
}
// worker_cancel drops the pending request outright and asks the active
// computation to stop at its next loop checkpoint. It reports whether anything
// was cancellable.
worker_cancel :: proc(worker: ^Worker) -> i64 {
if worker == nil || !worker.live do return 0
sync.mutex_lock(&worker.mutex)
defer sync.mutex_unlock(&worker.mutex)
requested := false
if worker.pending {
worker.pending = false
worker.dropped_pending += 1
requested = true
}
if worker.active {
worker.active_cancelled = true
sync.atomic_store(&worker.cancel_token, true)
requested = true
}
if requested do return 1
return 0
}
// worker_status reports the worker-level status without consuming anything.
worker_status :: proc(worker: ^Worker) -> Status {
if worker == nil || !worker.live do return .Failed
sync.mutex_lock(&worker.mutex)
defer sync.mutex_unlock(&worker.mutex)
if worker.worker_exited do return .Failed
return worker.status
}
// worker_take exchanges the bounded handoff: it returns the one retained
// record and transfers its ownership to the caller. It never blocks.
worker_take :: proc(worker: ^Worker) -> (result: Result, ok: bool) {
if worker == nil || !worker.live do return result, false
sync.mutex_lock(&worker.mutex)
record := worker.ready
if record == nil {
sync.mutex_unlock(&worker.mutex)
return result, false
}
worker.ready = nil
if !worker.active && !worker.pending {
worker.status = .Idle
}
sync.mutex_unlock(&worker.mutex)
result = record^
if mem.free(rawptr(record), worker.owner_allocator) != nil do return result, false
return result, true
}
// worker_start_thread creates the worker thread with default attributes
// (joinable, inherited scheduling, default stack) and starts it immediately;
// the platform sibling owns the plain thread handle the pure owner joins.
worker_start_thread :: proc(worker: ^Worker) -> bool {
if !worker_thread_spawn(worker) do return false
worker.thread_started = true
return true
}
worker_main :: proc(raw: rawptr) {
w := cast(^Worker)raw
for {
sync.mutex_lock(&w.mutex)
for !w.active && !w.shutdown {
sync.cond_wait(&w.wake, &w.mutex)
}
if w.shutdown && !w.active {
sync.mutex_unlock(&w.mutex)
break
}
id := w.active_id
seed := w.active_seed
iters := w.active_iters
// Carry this request's cancellation intent into the token: a cancel()
// that raced the pickup keeps its effect, and the intent stays set
// until completion so the early exit is reported as a cancellation
// rather than published as a result.
sync.atomic_store(&w.cancel_token, w.active_cancelled)
sync.mutex_unlock(&w.mutex)
started := time.tick_now()
hash := worker_compute(w, seed, iters)
compute_usec := i64(time.duration_microseconds(time.tick_since(started)))
sync.mutex_lock(&w.mutex)
cancelled := w.active_cancelled
w.active = false
w.active_id = 0
if cancelled {
w.status = .Cancelled
} else {
record, ok := worker_new_record(w)
if !ok {
w.status = .Failed
w.error = .Unavailable
} else {
record.request_id = id
record.hash = hash
record.version = WORKER_VERSION
record.compute_usec = compute_usec
record.cancelled = false
// The handoff is bounded at one record: a newer result displaces
// an unconsumed older one, which is released here, not leaked.
if w.ready != nil {
_ = mem.free(rawptr(w.ready), w.owner_allocator)
}
w.ready = record
w.status = .Completed
w.error = .None
}
}
// The finished request can no longer observe a cancellation; a late
// cancel() must not leak into the next request.
w.active_cancelled = false
// A request submitted while this one ran is promoted now instead of
// being silently dropped.
if w.pending && !w.shutdown {
w.active = true
w.active_id = w.pending_id
w.active_seed = w.pending_seed
w.active_iters = w.pending_iters
w.pending = false
w.status = .Running
}
sync.mutex_unlock(&w.mutex)
}
sync.mutex_lock(&w.mutex)
w.worker_exited = true
sync.mutex_unlock(&w.mutex)
}
// worker_compute folds a splitmix64 stream over the iteration range with a
// cancel checkpoint every chunk; the version marker enters the final mix so a
// result identifies the image that produced it.
worker_compute :: proc(w: ^Worker, seed, iterations: i64) -> u64 {
chunk :: 16_384
state := u64(seed) ~ 0x9e3779b97f4a7c15
accumulator := u64(WORKER_VERSION)
index := 0
for index < int(iterations) {
bound := min(int(iterations), index + chunk)
for index < bound {
state += 0x9e3779b97f4a7c15
z := state
z = (z ~ (z >> 30)) * 0xbf58476d1ce4e5b9
z = (z ~ (z >> 27)) * 0x94d049bb133111eb
z = z ~ (z >> 31)
accumulator = rotl64(accumulator ~ z, 23) + 0x165667b19e3779f9
index += 1
}
if sync.atomic_load(&w.cancel_token) {
return 0
}
}
return accumulator ~ rotl64(state, 7)
}
worker_new_record :: proc(w: ^Worker) -> (^Result, bool) {
storage, allocation_error := mem.alloc(size_of(Result), align_of(Result), w.owner_allocator)
if allocation_error != nil || storage == nil do return nil, false
return cast(^Result)storage, true
}
rotl64 :: proc(x: u64, n: int) -> u64 {
return (x << uint(n)) | (x >> (64 - uint(n)))
}