Linked repository source
examples/voxel_stream/subsystem/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 pure voxel worker owns bounded queue snapshots and result storage.
// Remesh submissions copy into one of WORKER_QUEUE preallocated payload slots.
// Each of WORKER_MAILBOX result slots owns a chunk and a Mesh_Workspace. The
// producer reserves a free result slot before starting a job; a slow consumer
// applies backpressure instead of silently losing another chunk's completion.
// worker_take leases immutable slices until worker_release_result. Cancelling
// queued, active or published work only discards explicitly obsolete requests;
// taken leases stay valid. Cancellation and shutdown wake the producer.
// No job, result publication, take or release allocates after worker_create.
// Every worker callback installs its own Odin context and has no Godot access.
package voxel
import "core:mem"
import "core:thread"
import "base:runtime"
import "core:sync"
import "core:time"
WORKER_QUEUE :: 128
WORKER_MAILBOX :: 4
Job_Kind :: enum u8 {
Generate,
Remesh,
}
// Remesh inputs are copied before submit returns. Generate inputs are nil.
Job_Spec :: struct {
kind: Job_Kind,
coords: [3]i32,
params: Params,
lod: u8,
voxels: ^Chunk,
aprons: ^Aprons,
}
// A taken result leases worker-owned storage until worker_release_result.
// Generate exposes its chunk so the caller can copy it into its own store;
// Remesh exposes only the mesh. Neither payload may be freed independently.
// request_id identifies both the request and the current use of a result slot.
Worker_Result :: struct {
request_id: u64,
kind: Job_Kind,
coords: [3]i32,
error: Error,
compute_usec: i64,
lod: u8,
skipped: bool,
voxels: ^Chunk,
mesh: Mesh_Data,
voxel_digest: u64,
mesh_digest: u64,
lease_owner: ^Worker,
lease_slot: i32,
}
Job :: struct {
request_id: u64,
kind: Job_Kind,
coords: [3]i32,
params: Params,
lod: u8,
payload_slot: i32,
}
Worker_Job_Payload :: struct {
used: bool,
voxels: Chunk,
aprons: Aprons,
}
Worker_Result_State :: enum u8 {
Free,
Computing,
Published,
Leased,
}
Worker_Result_Slot :: struct {
state: Worker_Result_State,
record: Worker_Result,
voxels: Chunk,
workspace: Mesh_Workspace,
}
Worker_Status :: enum i32 {
Idle,
Running,
Completed,
Failed,
}
Worker :: struct {
live: bool,
owner_allocator: mem.Allocator,
mutex: sync.Mutex,
wake: sync.Cond,
thread_started: bool,
thread_handle: ^thread.Thread,
// Queue, result states and lease transitions are mutex-guarded. Payloads
// are immutable while queued; a computing slot belongs to the producer.
shutdown: bool,
worker_exited: bool,
// Set only at create and read by the producer thread; digests are the
// verification oracles result consumers may compare against.
digests: bool,
// Request IDs start here. Pools give members disjoint bases so pooled
// result IDs stay globally unique for exact-request matching.
id_base: u64,
next_request_id: u64,
head, tail, count: i32,
ring: [WORKER_QUEUE]Job,
payloads: [WORKER_QUEUE]Worker_Job_Payload,
results: [WORKER_MAILBOX]Worker_Result_Slot,
mailbox: [WORKER_MAILBOX]i32,
mailbox_count: i32,
active_slot: i32,
active_coords: [3]i32,
active_cancelled: bool,
dropped_jobs: u64,
status: Worker_Status,
error: Error,
compute_usec: i64,
tracking: mem.Tracking_Allocator,
created_allocations: i64,
// Only the producer accesses the apron workspace.
worker_aprons: Aprons,
}
// All bounded payloads and mesh buffers exist before the first submit. The
// producer thread starts lazily; a failed start cannot strand a materialized
// payload.
// digests_enabled chooses whether results carry chunk/mesh digest oracles:
// consumers that verify published bytes themselves can skip that worker-side
// fold. Verification-minded consumers pass true. id_base shifts this worker's
// request-ID space; standalone workers pass 0.
worker_create :: proc(allocator: mem.Allocator, digests_enabled: bool, id_base: u64) -> (^Worker, bool) {
storage, allocation_error := mem.alloc(size_of(Worker), align_of(Worker), allocator)
if allocation_error != nil || storage == nil {
return nil, false
}
worker := cast(^Worker)storage
worker.live = true
worker.active_slot = -1
worker.digests = digests_enabled
worker.id_base = id_base
worker.next_request_id = id_base
worker.owner_allocator = allocator
// The producer's thread record is created here, suspended: the worker
// contract refuses any allocation after create, so the record joins the
// create-time allocation budget, where a failure strands no payload
// because none exists yet. core:thread allocates the record from
// context.allocator, which an engine callback does not establish, so the
// create allocator is pinned explicitly.
outer := context.allocator
context.allocator = allocator
worker.thread_handle = thread.create(worker_thread_entry)
context.allocator = outer
if worker.thread_handle == nil {
_ = worker_destroy(&worker)
return nil, false
}
worker.thread_handle.data = worker
mem.tracking_allocator_init(&worker.tracking, allocator, allocator)
for &slot in worker.results {
if !mesh_workspace_create(mem.tracking_allocator(&worker.tracking), &slot.workspace) {
_ = worker_destroy(&worker)
return nil, false
}
}
worker.created_allocations = worker.tracking.total_allocation_count
return worker, true
}
// Stop and join before releasing any storage. An outstanding taken lease makes
// destroy return false and retain the closed owner: release that lease, then
// retry destroy. No result borrow is invalidated behind the consumer's back.
worker_destroy :: proc(worker: ^^Worker) -> bool {
if worker == nil || worker^ == nil {
return false
}
w := worker^
sync.mutex_lock(&w.mutex)
w.live = false
w.shutdown = true
w.active_cancelled = true
drain_ring_locked(w)
sync.cond_broadcast(&w.wake)
sync.mutex_unlock(&w.mutex)
if w.thread_handle != nil {
// destroy joins and frees the thread record. A never-started producer
// must be resumed here first: joining starts it itself while holding
// the thread record's non-recursive mutex, which self-deadlocks.
if !w.thread_started do thread.start(w.thread_handle)
thread.destroy(w.thread_handle)
w.thread_handle = nil
w.thread_started = false
}
for &slot in w.results {
if slot.state == .Leased {
return false
}
}
for &slot in w.results {
mesh_workspace_destroy(&slot.workspace, mem.tracking_allocator(&w.tracking))
}
drain_ok := w.tracking.current_memory_allocated == 0
mem.tracking_allocator_destroy(&w.tracking)
allocator := w.owner_allocator
if mem.free(rawptr(w), allocator) != nil {
return false
}
worker^ = nil
return drain_ok
}
// Non-waiting admission: a full queue refuses the request; the caller retains
// its state and may retry. Remesh priority is selected by worker_submit_front.
worker_submit :: proc(worker: ^Worker, spec: Job_Spec) -> (i64, Error) {
return worker_submit_at(worker, spec, false)
}
worker_submit_front :: proc(worker: ^Worker, spec: Job_Spec) -> (i64, Error) {
return worker_submit_at(worker, spec, true)
}
worker_submit_at :: proc(worker: ^Worker, spec: Job_Spec, front: bool) -> (i64, Error) {
if worker == nil || !worker.live {
return -1, .Workspace_Unavailable
}
if spec.kind != .Generate && spec.kind != .Remesh {
return -1, .Validation_Failed
}
if spec.kind == .Remesh && (spec.voxels == nil || spec.aprons == nil) {
return -1, .Validation_Failed
}
if spec.coords[1] != 0 {
return -1, .Coords_Out_Of_Range
}
if spec.lod > 2 {
return -1, .Validation_Failed
}
if params_valid(spec.params) != .None {
return -1, .Params_Out_Of_Range
}
sync.mutex_lock(&worker.mutex)
defer sync.mutex_unlock(&worker.mutex)
if worker.shutdown || worker.worker_exited {
return -1, .Workspace_Unavailable
}
if worker.count >= WORKER_QUEUE {
worker.dropped_jobs += 1
return -1, .Queue_Full
}
if !worker.thread_started && !worker_start_thread(worker) {
return -1, .Workspace_Unavailable
}
id := worker.next_request_id + 1
job := Job{
request_id = id, kind = spec.kind, coords = spec.coords,
params = spec.params, lod = spec.lod, payload_slot = -1,
}
if spec.kind == .Remesh {
for &payload, i in worker.payloads {
if payload.used do continue
payload.used = true
copy(payload.voxels[:], spec.voxels[:])
copy(payload.aprons[:], spec.aprons[:])
job.payload_slot = i32(i)
break
}
// An active job has already returned its payload slot; at most count
// slots are occupied, so admission above guarantees this free slot.
assert(job.payload_slot >= 0)
}
if front {
worker.head = (worker.head - 1 + WORKER_QUEUE) % WORKER_QUEUE
worker.ring[worker.head] = job
} else {
worker.ring[worker.tail] = job
worker.tail = (worker.tail + 1) % WORKER_QUEUE
}
worker.count += 1
worker.next_request_id = id
worker.status = .Running
sync.cond_signal(&worker.wake)
return i64(id), .None
}
worker_free_job_locked :: proc(worker: ^Worker, job: ^Job) {
if job.payload_slot >= 0 {
worker.payloads[job.payload_slot].used = false
}
job^ = Job{payload_slot = -1}
}
worker_coords_outside :: proc(coords, min, max: [3]i32) -> bool {
return coords[0] < min[0] || coords[0] > max[0] ||
coords[2] < min[2] || coords[2] > max[2]
}
// Cancel queued, active and published requests outside the inclusive x/z
// range. The active computation finishes but cannot publish. Taken leases
// remain immutable until released. Returns the number newly cancelled.
worker_cancel_outside :: proc(worker: ^Worker, min, max: [3]i32) -> i64 {
return worker_cancel_selected(worker, min, max, false)
}
worker_cancel_all :: proc(worker: ^Worker) -> i64 {
return worker_cancel_selected(worker, {}, {}, true)
}
worker_cancel_selected :: proc(worker: ^Worker, min, max: [3]i32, all: bool) -> i64 {
if worker == nil || !worker.live {
return 0
}
sync.mutex_lock(&worker.mutex)
defer sync.mutex_unlock(&worker.mutex)
removed := i64(0)
kept := i32(0)
for i in 0..<worker.count {
index := (worker.head + i) % WORKER_QUEUE
job := worker.ring[index]
if all || worker_coords_outside(job.coords, min, max) {
removed += 1
worker_free_job_locked(worker, &worker.ring[index])
} else {
worker.ring[(worker.head + kept) % WORKER_QUEUE] = job
kept += 1
}
}
worker.count = kept
worker.tail = (worker.head + kept) % WORKER_QUEUE
kept = 0
for i in 0..<worker.mailbox_count {
slot_index := worker.mailbox[i]
slot := &worker.results[slot_index]
if all || worker_coords_outside(slot.record.coords, min, max) {
removed += 1
slot.state = .Free
slot.record = {}
} else {
worker.mailbox[kept] = slot_index
kept += 1
}
}
worker.mailbox_count = kept
if worker.active_slot >= 0 && !worker.active_cancelled &&
(all || worker_coords_outside(worker.active_coords, min, max)) {
worker.active_cancelled = true
removed += 1
}
if worker.count == 0 && worker.active_slot < 0 && worker.mailbox_count == 0 {
worker.status = .Idle
}
sync.cond_broadcast(&worker.wake)
return removed
}
// Pop the oldest result without waiting for computation. The returned slices
// stay valid across further submits/takes, until this specific lease is released.
worker_take :: proc(worker: ^Worker) -> (result: Worker_Result, ok: bool) {
if worker == nil || !worker.live {
return result, false
}
sync.mutex_lock(&worker.mutex)
defer sync.mutex_unlock(&worker.mutex)
if worker.mailbox_count == 0 {
return result, false
}
slot_index := worker.mailbox[0]
for i in 1..<worker.mailbox_count {
worker.mailbox[i - 1] = worker.mailbox[i]
}
worker.mailbox_count -= 1
slot := &worker.results[slot_index]
assert(slot.state == .Published)
slot.state = .Leased
return slot.record, true
}
// Release only the named live lease; repeated/copied releases cannot release a
// reused slot. A closed worker still accepts lease returns before final destroy.
worker_release_result :: proc(worker: ^Worker, result: ^Worker_Result) {
if worker == nil || result == nil {
return
}
sync.mutex_lock(&worker.mutex)
defer sync.mutex_unlock(&worker.mutex)
if result.lease_owner == worker && result.lease_slot >= 0 && result.lease_slot < WORKER_MAILBOX {
slot := &worker.results[result.lease_slot]
if slot.state == .Leased && slot.record.request_id == result.request_id {
slot.state = .Free
slot.record = {}
sync.cond_signal(&worker.wake)
}
}
result^ = {}
}
worker_status :: proc(worker: ^Worker) -> Worker_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
}
// Backpressure never supersedes another request. Retained for existing stats.
worker_superseded_results :: proc(worker: ^Worker) -> u64 {
return 0
}
worker_dropped_jobs :: proc(worker: ^Worker) -> u64 {
if worker == nil do return 0
sync.mutex_lock(&worker.mutex)
defer sync.mutex_unlock(&worker.mutex)
return worker.dropped_jobs
}
// Count pending, computing, published and leased requests exactly once.
worker_work_count :: proc(worker: ^Worker) -> i64 {
if worker == nil do return 0
sync.mutex_lock(&worker.mutex)
defer sync.mutex_unlock(&worker.mutex)
count := i64(worker.count)
for &slot in worker.results {
if slot.state != .Free do count += 1
}
return count
}
// Includes the embedded queue/chunk storage as well as tracked mesh buffers.
worker_peak_memory :: proc(worker: ^Worker) -> i64 {
if worker == nil do return 0
return i64(size_of(Worker)) + worker.tracking.peak_memory_allocated
}
worker_warm_allocations :: proc(worker: ^Worker) -> i64 {
if worker == nil do return 0
return worker.tracking.total_allocation_count - worker.created_allocations
}
worker_start_thread :: proc(worker: ^Worker) -> bool {
// The record was created suspended at worker_create; first submit only
// resumes it, so start cannot fail or allocate here.
if worker.thread_handle == nil || worker.thread_started {
return worker.thread_started
}
thread.start(worker.thread_handle)
worker.thread_started = true
return true
}
worker_thread_entry :: proc(t: ^thread.Thread) {
context = runtime.default_context()
worker_main(cast(^Worker)t.data)
}
worker_free_result_slot_locked :: proc(worker: ^Worker) -> i32 {
for &slot, i in worker.results {
if slot.state == .Free do return i32(i)
}
return -1
}
worker_main :: proc(data: rawptr) {
worker := cast(^Worker)data
context.allocator = mem.tracking_allocator(&worker.tracking)
for {
sync.mutex_lock(&worker.mutex)
result_slot := worker_free_result_slot_locked(worker)
for !worker.shutdown && (worker.count == 0 || result_slot < 0) {
sync.cond_wait(&worker.wake, &worker.mutex)
result_slot = worker_free_result_slot_locked(worker)
}
if worker.shutdown {
worker.worker_exited = true
sync.mutex_unlock(&worker.mutex)
return
}
job := worker.ring[worker.head]
worker.head = (worker.head + 1) % WORKER_QUEUE
worker.count -= 1
slot := &worker.results[result_slot]
slot.state = .Computing
worker.active_slot = result_slot
worker.active_coords = job.coords
worker.active_cancelled = false
if job.kind == .Remesh {
payload := &worker.payloads[job.payload_slot]
copy(slot.voxels[:], payload.voxels[:])
copy(worker.worker_aprons[:], payload.aprons[:])
payload.used = false
}
sync.mutex_unlock(&worker.mutex)
started := time.tick_now()
result := worker_run_job(worker, job, result_slot)
result.compute_usec = i64(time.duration_microseconds(time.tick_since(started)))
worker_publish(worker, result_slot, result)
}
}
worker_run_job :: proc(worker: ^Worker, job: Job, slot_index: i32) -> Worker_Result {
result := Worker_Result{
request_id = job.request_id, kind = job.kind, coords = job.coords,
lod = job.lod, lease_owner = worker, lease_slot = slot_index,
}
slot := &worker.results[slot_index]
center := &slot.voxels
aprons := &worker.worker_aprons
if job.kind == .Generate {
// Apron generation uses the slot chunk as scratch; generate its center
// last, directly in the eventual result storage.
if !generate_aprons(aprons, job.coords, job.params) ||
!generate_chunk(center, job.coords, job.params) {
result.error = .Validation_Failed
return result
}
result.voxels = center
}
if worker.digests {
result.voxel_digest = chunk_digest(center)
}
view := Mesher_View{chunk = center, aprons = aprons}
if !mesh_chunk_into(view, job.coords, job.params, job_lod_step(job), &slot.workspace, &result.mesh) {
result.error = .Validation_Failed
return result
}
// Use exactly the direct mesher's empty geometry and AABB. This metadata
// counts empty chunks; it never substitutes a second meshing algorithm.
result.skipped = chunk_is_all_air(center)
if worker.digests {
result.mesh_digest = mesh_digest(&result.mesh)
}
return result
}
job_lod_step :: proc(job: Job) -> i32 {
return lod_step(i32(job.lod))
}
// Publication needs no allocation and cannot overflow: this job reserved its
// slot before running. The only discarded result was explicitly cancelled.
worker_publish :: proc(worker: ^Worker, slot_index: i32, result: Worker_Result) {
sync.mutex_lock(&worker.mutex)
defer sync.mutex_unlock(&worker.mutex)
slot := &worker.results[slot_index]
if worker.shutdown || worker.active_cancelled {
slot.record = {}
slot.state = .Free
} else {
assert(worker.mailbox_count < WORKER_MAILBOX)
slot.record = result
slot.state = .Published
worker.mailbox[worker.mailbox_count] = slot_index
worker.mailbox_count += 1
worker.error = result.error
worker.status = .Completed if result.error == .None else .Failed
}
worker.active_slot = -1
worker.active_cancelled = false
worker.compute_usec = result.compute_usec
if worker.count == 0 && worker.mailbox_count == 0 {
worker.status = .Idle
}
}
drain_ring_locked :: proc(w: ^Worker) {
for w.count > 0 {
worker_free_job_locked(w, &w.ring[w.head])
w.head = (w.head + 1) % WORKER_QUEUE
w.count -= 1
}
w.tail = w.head
}