Linked repository source
examples/cavern_lab/subsystem/background.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
// Background generation owner for the pure cavern kernel. It implements the
// frozen Phase 13 budgets for one generator instance:
//
// storage — one worker workspace (18 MiB class) plus at most one owned
// result copy (width*height bytes) and one unconsumed handoff
// record; the peak is tracked by the worker's own tracking
// allocator and observable through worker_peak_memory.
// requests — at most one active computation and one replaceable pending
// request per worker; the bounded handoff retains at most one
// completed record, and a newly completed record replaces an
// unconsumed older one (counted as superseded).
// cancellation — the worker observes the cooperative token at the kernel's
// loop boundaries, so acknowledgement is bounded by one
// smoothing pass; the observed acknowledgement latency is
// recorded exactly once per cancelled computation.
// publication — the main thread only ever takes a finished record with
// worker_take, a non-blocking exchange, so the publishing
// budget is the engine-side copy performed by the caller.
// shutdown — worker_destroy sets shutdown, cancels the active request,
// joins the thread, drains the handoff and workspace, and
// refuses to report success while the tracking allocator still
// holds bytes.
//
// The worker computes on pure Odin data with its own allocator context. It
// never touches Godot storage, callback capabilities, or borrowed spans.
package cavern
import "core:mem"
import "core:sync"
import "core:time"
Background_Status :: enum i32 {
Idle,
Running,
Completed,
Cancelled,
Failed,
}
// Background_Result is an owned, immutable record. request_id pairs it with
// the submit that produced it so the consumer can discard obsolete output.
// cells is owned by the receiver and must be released exactly once through
// worker_release_result with the owning worker.
Background_Result :: struct {
request_id: u64,
error: Error,
diagnostics: Diagnostics,
compute_usec: i64,
cancellation_ack_usec: i64,
cells: []u8,
}
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_settings: Settings,
pending: bool,
pending_id: u64,
pending_settings: Settings,
ready: ^Background_Result,
superseded_results: u64,
dropped_pending: u64,
status: Background_Status,
error: Error,
active_cancelled: bool,
cancel_requested: bool,
cancel_stamp: time.Tick,
// Controlled fixture-only injection counters, armed through
// worker_debug_inject and consumed at pickup. Guarded by the mutex; no
// product path arms them.
debug_generation_failures: i32,
debug_workspace_failures: i32,
// --- written by the worker, read after join or under mutex ---
cancel_token: bool,
compute_usec: i64,
cancellation_ack_usec: i64,
peak_memory: i64,
tracking: mem.Tracking_Allocator,
worker_workspace: ^Workspace,
}
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
mem.tracking_allocator_init(&worker.tracking, allocator, allocator)
return worker, true
}
// worker_destroy closes, cancels, and drains before the worker's memory is
// released. The returned bool is the drain receipt: true only when the tracking
// allocator holds zero bytes after the handoff record and workspace are freed.
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
// The intent flag is what a pickup consults; setting the raw token alone
// would be wiped when the worker picked up an already-submitted request.
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 record.cells != nil {
delete(record.cells, mem.tracking_allocator(&w.tracking))
}
if mem.free(rawptr(record), mem.tracking_allocator(&w.tracking)) != nil do drain_ok = false
}
if w.worker_workspace != nil {
if !workspace_destroy(&w.worker_workspace) do drain_ok = false
}
if w.tracking.current_memory_allocated != 0 do drain_ok = false
mem.tracking_allocator_destroy(&w.tracking)
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. It returns the new request id, or a negative id with the named
// subsystem error.
worker_submit :: proc(worker: ^Worker, settings: Settings) -> (i64, Error) {
if worker == nil || !worker.live do return -1, .Workspace_Unavailable
if validation := settings_valid(settings); validation != .None do return -1, validation
sync.mutex_lock(&worker.mutex)
defer sync.mutex_unlock(&worker.mutex)
if worker.worker_exited do return -1, .Workspace_Unavailable
id := worker.next_request_id + 1
if !worker.thread_started {
if !worker_start_thread(worker) {
return -1, .Workspace_Unavailable
}
}
if worker.active {
if worker.pending do worker.dropped_pending += 1
worker.pending = true
worker.pending_id = id
worker.pending_settings = settings
} else {
worker.active = true
worker.active_id = id
worker.active_settings = settings
worker.status = .Running
sync.cond_signal(&worker.wake)
}
worker.next_request_id = id
return i64(id), .None
}
// worker_cancel cancels the pending request outright and asks the active
// computation to stop at its next loop boundary. 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 {
// The intent flag survives a race with the worker's token reset at
// pickup: a cancellation requested before the pickup still cancels the
// request it was aimed at.
worker.active_cancelled = true
worker.cancel_requested = true
worker.cancel_stamp = time.tick_now()
sync.atomic_store(&worker.cancel_token, true)
requested = true
}
if requested do return 1
return 0
}
// worker_take exchanges the bounded handoff: it returns the one retained
// record and transfers its ownership (including cells) to the caller. It never
// blocks and never waits.
worker_take :: proc(worker: ^Worker) -> (result: Background_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
sync.mutex_unlock(&worker.mutex)
result = record^
if mem.free(rawptr(record), mem.tracking_allocator(&worker.tracking)) != nil do return result, false
return result, true
}
// worker_release_result frees the owned cells of a taken record exactly once.
// Obsolete results are released through this path without ever being published
// to Godot.
worker_release_result :: proc(worker: ^Worker, cells: []u8) {
if worker == nil do return
if cells != nil {
delete(cells, mem.tracking_allocator(&worker.tracking))
}
}
worker_status :: proc(worker: ^Worker) -> Background_Status {
if worker == nil do return .Idle
sync.mutex_lock(&worker.mutex)
defer sync.mutex_unlock(&worker.mutex)
return worker.status
}
worker_error :: proc(worker: ^Worker) -> Error {
if worker == nil do return .None
sync.mutex_lock(&worker.mutex)
defer sync.mutex_unlock(&worker.mutex)
return worker.error
}
worker_latest_request_id :: proc(worker: ^Worker) -> u64 {
if worker == nil do return 0
sync.mutex_lock(&worker.mutex)
defer sync.mutex_unlock(&worker.mutex)
return worker.next_request_id
}
worker_superseded_results :: proc(worker: ^Worker) -> u64 {
if worker == nil do return 0
sync.mutex_lock(&worker.mutex)
defer sync.mutex_unlock(&worker.mutex)
return worker.superseded_results
}
worker_dropped_pending :: proc(worker: ^Worker) -> u64 {
if worker == nil do return 0
sync.mutex_lock(&worker.mutex)
defer sync.mutex_unlock(&worker.mutex)
return worker.dropped_pending
}
worker_compute_usec :: proc(worker: ^Worker) -> i64 {
if worker == nil do return 0
sync.mutex_lock(&worker.mutex)
defer sync.mutex_unlock(&worker.mutex)
return worker.compute_usec
}
worker_cancellation_ack_usec :: proc(worker: ^Worker) -> i64 {
if worker == nil do return 0
sync.mutex_lock(&worker.mutex)
defer sync.mutex_unlock(&worker.mutex)
return worker.cancellation_ack_usec
}
// worker_debug_inject arms the controlled fixture-only failure injection on
// this worker: the next generation_failures requests finish with the kernel's
// Validation_Failed result — a completed computation whose cells are dropped
// exactly as the kernel's own internal validation failure drops them — and
// the next workspace_failures worker workspace creations fail, leaving that
// worker permanently unusable. Counters are consumed at pickup under the
// worker mutex, so arming from any thread is exact. No product path arms them.
worker_debug_inject :: proc(worker: ^Worker, generation_failures, workspace_failures: i32) {
if worker == nil || !worker.live do return
sync.mutex_lock(&worker.mutex)
worker.debug_generation_failures += generation_failures
worker.debug_workspace_failures += workspace_failures
sync.mutex_unlock(&worker.mutex)
}
worker_peak_memory :: proc(worker: ^Worker) -> i64 {
if worker == nil do return 0
sync.mutex_lock(&worker.mutex)
defer sync.mutex_unlock(&worker.mutex)
return worker.peak_memory
}
// 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(data: rawptr) {
worker := cast(^Worker)data
context.allocator = mem.tracking_allocator(&worker.tracking)
workspace, created := workspace_create(context.allocator)
// Controlled fixture injection, consumed under the mutex: a failed
// workspace creation leaves this worker permanently unusable, which the
// lifecycle campaign and the pure tests exercise.
fail_workspace := false
sync.mutex_lock(&worker.mutex)
if worker.debug_workspace_failures > 0 {
worker.debug_workspace_failures -= 1
fail_workspace = true
}
sync.mutex_unlock(&worker.mutex)
if created && fail_workspace {
_ = workspace_destroy(&workspace)
created = false
}
sync.mutex_lock(&worker.mutex)
worker.worker_workspace = workspace
if !created {
worker.error = .Workspace_Unavailable
worker.worker_exited = true
sync.mutex_unlock(&worker.mutex)
return
}
sync.mutex_unlock(&worker.mutex)
for {
sync.mutex_lock(&worker.mutex)
for !worker.active && !worker.shutdown {
sync.cond_wait(&worker.wake, &worker.mutex)
}
if worker.shutdown && !worker.active {
worker.worker_exited = true
sync.mutex_unlock(&worker.mutex)
return
}
request := worker.active_settings
request_id := worker.active_id
// Consume this request's cancellation intent: a cancel() that raced
// the pickup keeps its effect through the intent flag rather than the
// raw token.
sync.atomic_store(&worker.cancel_token, worker.active_cancelled)
worker.active_cancelled = false
// Controlled fixture injection, decided under the mutex at pickup.
fail_request := false
if worker.debug_generation_failures > 0 {
worker.debug_generation_failures -= 1
fail_request = true
}
sync.mutex_unlock(&worker.mutex)
started := time.tick_now()
generated := generate_controlled(workspace, request, &worker.cancel_token)
if fail_request && generated.error == .None {
// Drop the finished cells exactly as the kernel's own internal
// validation failure does, and surface its validation error.
generated = {error=.Validation_Failed}
}
compute := i64(time.duration_microseconds(time.tick_since(started)))
if generated.error == .None {
cells := make([]u8, len(generated.cells))
copy(cells, generated.cells)
generated.cells = cells
} else {
generated.cells = nil
}
if generated.error == .Cancelled {
sync.mutex_lock(&worker.mutex)
if worker.cancel_requested {
ack := i64(time.duration_microseconds(time.tick_since(worker.cancel_stamp)))
// The measurement's resolution is one microsecond; a faster
// acknowledgement still counts as one.
if ack < 1 do ack = 1
worker.cancellation_ack_usec = ack
worker.cancel_requested = false
}
sync.mutex_unlock(&worker.mutex)
}
worker_publish(worker, request_id, generated, compute)
}
}
worker_publish :: proc(worker: ^Worker, request_id: u64, generated: Result, compute_usec: i64) {
// The worker owns this call: it runs only on the worker thread after the
// computation returned, and the mutex only guards the handoff fields.
sync.mutex_lock(&worker.mutex)
defer sync.mutex_unlock(&worker.mutex)
if worker.ready != nil {
// The handoff is bounded at one record. Replace the unconsumed older
// result so the newest work is always the one available, and count the
// displaced record as superseded.
older := worker.ready
worker.ready = nil
if older.cells != nil {
delete(older.cells, mem.tracking_allocator(&worker.tracking))
}
_ = mem.free(rawptr(older), mem.tracking_allocator(&worker.tracking))
worker.superseded_results += 1
}
if !worker.shutdown {
record, record_error := mem.alloc(size_of(Background_Result), align_of(Background_Result), mem.tracking_allocator(&worker.tracking))
if record_error == nil && record != nil {
stored := cast(^Background_Result)record
stored^ = Background_Result{
request_id = request_id,
error = generated.error,
diagnostics = generated.diagnostics,
compute_usec = compute_usec,
cancellation_ack_usec = worker.cancellation_ack_usec,
cells = generated.cells,
}
worker.ready = stored
} else if generated.cells != nil {
delete(generated.cells, mem.tracking_allocator(&worker.tracking))
}
} else if generated.cells != nil {
delete(generated.cells, mem.tracking_allocator(&worker.tracking))
}
#partial switch generated.error {
case .None:
worker.status = .Completed
case .Cancelled:
worker.status = .Cancelled
case:
worker.status = .Failed
}
worker.error = generated.error
worker.compute_usec = compute_usec
worker.peak_memory = worker.tracking.peak_memory_allocated
worker.active = false
// The finished request can no longer observe a cancellation; a late
// cancel() must not leak into the next request.
worker.active_cancelled = false
if worker.pending && !worker.shutdown {
worker.active = true
worker.active_id = worker.pending_id
worker.active_settings = worker.pending_settings
worker.pending = false
worker.status = .Running
}
}