class_name AdtWaterLoadPipelineState extends RefCounted ## Owns ADT water request, active-task and worker-result bookkeeping. The caller ## owns WorkerThreadPool operations, ADT parsing, permits and scene finalization. var _pending_requests: Array[Dictionary] = [] var _queued_tile_keys: Dictionary = {} var _active_task_ids_by_tile_key: Dictionary = {} var _result_mutex := Mutex.new() var _results: Array[Dictionary] = [] ## Appends one tile/path request in FIFO order. Invalid, queued or active tile ## keys are rejected without mutation. func enqueue_request(tile_key: String, adt_path: String) -> bool: if tile_key.is_empty() or adt_path.is_empty(): return false if _queued_tile_keys.has(tile_key) or _active_task_ids_by_tile_key.has(tile_key): return false _pending_requests.append({ "key": tile_key, "path": adt_path, }) _queued_tile_keys[tile_key] = true return true ## Removes and returns the oldest request, also releasing its queued-key marker. ## An empty Dictionary means no pending request exists. func take_next_request() -> Dictionary: if _pending_requests.is_empty(): return {} var request: Dictionary = _pending_requests.pop_front() _queued_tile_keys.erase(String(request.get("key", ""))) return request ## Removes all pending requests for one tile without affecting active work. ## Returns the number of removed FIFO entries. func cancel_pending_requests(tile_key: String) -> int: if tile_key.is_empty(): return 0 _queued_tile_keys.erase(tile_key) var removed_count := 0 for request_index in range(_pending_requests.size() - 1, -1, -1): var request: Dictionary = _pending_requests[request_index] if String(request.get("key", "")) != tile_key: continue _pending_requests.remove_at(request_index) removed_count += 1 return removed_count ## Records one caller-started worker task. The opaque task ID is retained until ## the caller waits for and completes it. func remember_active_task(tile_key: String, task_id: int) -> bool: if tile_key.is_empty() or _active_task_ids_by_tile_key.has(tile_key): return false _active_task_ids_by_tile_key[tile_key] = task_id return true ## Returns whether a worker task is active for the tile. func has_active_task(tile_key: String) -> bool: return not tile_key.is_empty() and _active_task_ids_by_tile_key.has(tile_key) ## Returns the opaque active task ID, or -1 for an unknown tile. func active_task_id_for(tile_key: String) -> int: return int(_active_task_ids_by_tile_key.get(tile_key, -1)) ## Removes and returns one completed task ID, or -1 for an unknown tile. func complete_active_task(tile_key: String) -> int: if tile_key.is_empty() or not _active_task_ids_by_tile_key.has(tile_key): return -1 var task_id := int(_active_task_ids_by_tile_key[tile_key]) _active_task_ids_by_tile_key.erase(tile_key) return task_id ## Returns a detached snapshot of opaque task IDs for orderly shutdown waiting. func active_task_ids_snapshot() -> Array: return _active_task_ids_by_tile_key.values().duplicate() ## Publishes one parsed ADT result from a worker thread. Payload identity is ## retained; the caller must not mutate it after publication. func publish_result(tile_key: String, adt_path: String, water_data: Dictionary) -> bool: if tile_key.is_empty() or adt_path.is_empty(): return false _result_mutex.lock() _results.append({ "key": tile_key, "path": adt_path, "data": water_data, }) _result_mutex.unlock() return true ## Returns whether a worker result is currently ready. func has_result() -> bool: _result_mutex.lock() var result_available := not _results.is_empty() _result_mutex.unlock() return result_available ## Removes and returns the oldest worker result. An empty Dictionary means no ## result was ready. Intended for the serialized renderer main thread. func pop_result() -> Dictionary: _result_mutex.lock() var result: Dictionary = {} if not _results.is_empty(): result = _results.pop_front() _result_mutex.unlock() return result ## Returns pending plus active load count for existing renderer metrics. func total_load_count() -> int: return _pending_requests.size() + _active_task_ids_by_tile_key.size() ## Returns the pending request count. func pending_request_count() -> int: return _pending_requests.size() ## Returns the active worker-task count. func active_task_count() -> int: return _active_task_ids_by_tile_key.size() ## Clears bookkeeping and queued results without interrupting worker tasks. ## Orderly shutdown must wait the task-ID snapshot before calling this method. func clear() -> void: _pending_requests.clear() _queued_tile_keys.clear() _active_task_ids_by_tile_key.clear() _result_mutex.lock() _results.clear() _result_mutex.unlock() ## Returns detached FIFO/task/result scalar diagnostics without water payloads. func diagnostic_snapshot() -> Dictionary: var pending_requests: Array[Dictionary] = [] for request in _pending_requests: pending_requests.append(request.duplicate()) var active_tile_keys: Array = _active_task_ids_by_tile_key.keys() active_tile_keys.sort() _result_mutex.lock() var result_records: Array[Dictionary] = [] for result in _results: result_records.append({ "key": String(result.get("key", "")), "path": String(result.get("path", "")), }) _result_mutex.unlock() return { "pending_requests": pending_requests, "active_tile_keys": active_tile_keys.duplicate(), "result_records": result_records, }