diff --git a/crates/console/prometeu-drivers/src/asset.rs b/crates/console/prometeu-drivers/src/asset.rs index 5adb1f12..b80ddff2 100644 --- a/crates/console/prometeu-drivers/src/asset.rs +++ b/crates/console/prometeu-drivers/src/asset.rs @@ -2,11 +2,11 @@ use crate::memory_banks::{GlyphBankPoolInstaller, SceneBankPoolInstaller, SoundBankPoolInstaller}; use prometeu_hal::AssetBridge; use prometeu_hal::asset::{ - AssetCodec, AssetEntry, AssetId, AssetLoadError, AssetOpStatus, BankTelemetry, BankType, - HandleId, LoadStatus, PreloadEntry, SCENE_DECODED_LAYER_OVERHEAD_BYTES_V1, - SCENE_HEADER_BYTES_V1, SCENE_LAYER_COUNT_V1, SCENE_LAYER_HEADER_BYTES_V1, - SCENE_PAYLOAD_MAGIC_V1, SCENE_PAYLOAD_VERSION_V1, SCENE_TILE_RECORD_BYTES_V1, SlotRef, - SlotStats, + AssetBacklogInfo, AssetBacklogPosition, AssetCodec, AssetEntry, AssetId, AssetLoadError, + AssetOpStatus, AssetTargetStatus, BankTelemetry, BankType, HandleId, LoadStatus, PreloadEntry, + SCENE_DECODED_LAYER_OVERHEAD_BYTES_V1, SCENE_HEADER_BYTES_V1, SCENE_LAYER_COUNT_V1, + SCENE_LAYER_HEADER_BYTES_V1, SCENE_PAYLOAD_MAGIC_V1, SCENE_PAYLOAD_VERSION_V1, + SCENE_TILE_RECORD_BYTES_V1, SlotRef, SlotStats, }; use prometeu_hal::cartridge::AssetsPayloadSource; use prometeu_hal::color::Color; @@ -18,10 +18,10 @@ use prometeu_hal::scene_layer::{ParallaxFactor, SceneLayer}; use prometeu_hal::sound_bank::SoundBank; use prometeu_hal::tile::Tile; use prometeu_hal::tilemap::TileMap; -use std::collections::HashMap; +use std::collections::{HashMap, VecDeque}; use std::io::Read; use std::sync::atomic::{AtomicUsize, Ordering}; -use std::sync::{Arc, Mutex, RwLock}; +use std::sync::{Arc, Condvar, Mutex, RwLock}; use std::thread; use std::time::Instant; @@ -31,6 +31,136 @@ type StagingMap = HashMap>; type AssetTable = HashMap; type HandleTable = HashMap; +type TargetHandleTable = HashMap; +type TargetGenerationTable = HashMap; + +const ASSET_PROGRESS_IDLE: u16 = 0; +const ASSET_PROGRESS_LOADING: u16 = 1_000; +const ASSET_PROGRESS_DONE: u16 = 10_000; +const ASSET_TELEMETRY_SAMPLE_WINDOW: usize = 32; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub struct AssetLatencyPercentiles { + pub p50_micros: u64, + pub p95_micros: u64, +} + +#[derive(Debug, Clone, PartialEq, Eq, Default)] +pub struct AssetPipelineTelemetry { + pub submitted_requests: u64, + pub started_jobs: u64, + pub completed_jobs: u64, + pub failed_jobs: u64, + pub canceled_requests: u64, + pub superseded_requests: u64, + pub stale_results_discarded: u64, + pub current_backlog_depth: usize, + pub max_backlog_depth: usize, + pub active_progress: u16, + pub glyph_latency: AssetLatencyPercentiles, + pub sound_latency: AssetLatencyPercentiles, + pub scene_latency: AssetLatencyPercentiles, + pub last_asset_id: Option, + pub last_asset_latency: AssetLatencyPercentiles, +} + +#[derive(Default)] +struct AssetPipelineTelemetryState { + snapshot: AssetPipelineTelemetry, + bank_samples: HashMap>, + asset_samples: HashMap>, +} + +impl AssetPipelineTelemetryState { + fn record_submitted(&mut self, backlog_depth: usize) { + self.snapshot.submitted_requests += 1; + self.snapshot.current_backlog_depth = backlog_depth; + self.snapshot.max_backlog_depth = self.snapshot.max_backlog_depth.max(backlog_depth); + } + + fn record_started(&mut self, backlog_depth: usize) { + self.snapshot.started_jobs += 1; + self.snapshot.current_backlog_depth = backlog_depth; + self.snapshot.active_progress = ASSET_PROGRESS_LOADING; + } + + fn record_completed(&mut self, asset_id: AssetId, bank_type: BankType, duration_micros: u64) { + self.snapshot.completed_jobs += 1; + self.snapshot.active_progress = ASSET_PROGRESS_IDLE; + self.record_latency_sample(asset_id, bank_type, duration_micros); + } + + fn record_failed(&mut self, asset_id: AssetId, bank_type: BankType, duration_micros: u64) { + self.snapshot.failed_jobs += 1; + self.snapshot.active_progress = ASSET_PROGRESS_IDLE; + self.record_latency_sample(asset_id, bank_type, duration_micros); + } + + fn record_canceled(&mut self) { + self.snapshot.canceled_requests += 1; + } + + fn record_superseded(&mut self, count: usize) { + self.snapshot.superseded_requests += count as u64; + } + + fn record_stale_discard(&mut self) { + self.snapshot.stale_results_discarded += 1; + self.snapshot.superseded_requests += 1; + self.snapshot.active_progress = ASSET_PROGRESS_IDLE; + } + + fn record_latency_sample( + &mut self, + asset_id: AssetId, + bank_type: BankType, + duration_micros: u64, + ) { + push_sample( + self.bank_samples.entry(bank_type).or_default(), + duration_micros, + ASSET_TELEMETRY_SAMPLE_WINDOW, + ); + push_sample( + self.asset_samples.entry(asset_id).or_default(), + duration_micros, + ASSET_TELEMETRY_SAMPLE_WINDOW, + ); + self.snapshot.glyph_latency = percentile_snapshot(self.bank_samples.get(&BankType::GLYPH)); + self.snapshot.sound_latency = percentile_snapshot(self.bank_samples.get(&BankType::SOUNDS)); + self.snapshot.scene_latency = percentile_snapshot(self.bank_samples.get(&BankType::SCENE)); + self.snapshot.last_asset_id = Some(asset_id); + self.snapshot.last_asset_latency = percentile_snapshot(self.asset_samples.get(&asset_id)); + } +} + +fn push_sample(samples: &mut VecDeque, value: u64, window: usize) { + samples.push_back(value); + while samples.len() > window { + samples.pop_front(); + } +} + +fn percentile_snapshot(samples: Option<&VecDeque>) -> AssetLatencyPercentiles { + let Some(samples) = samples else { + return AssetLatencyPercentiles::default(); + }; + if samples.is_empty() { + return AssetLatencyPercentiles::default(); + } + + let mut sorted = samples.iter().copied().collect::>(); + sorted.sort_unstable(); + AssetLatencyPercentiles { + p50_micros: percentile_value(&sorted, 50), + p95_micros: percentile_value(&sorted, 95), + } +} + +fn percentile_value(sorted: &[u64], percentile: usize) -> u64 { + let index = ((sorted.len() - 1) * percentile).div_ceil(100); + sorted[index] +} #[derive(Clone, Default)] pub struct GlyphAssetSlotIndex { @@ -182,8 +312,11 @@ impl BankPolicy { pub struct AssetManager { assets: Arc>, handles: Arc>, + target_handles: Arc>, + target_generations: Arc>, next_handle_id: Mutex, assets_data: Arc>, + pipeline_telemetry: Arc>, /// Narrow hardware interfaces gfx_installer: Arc, @@ -203,14 +336,205 @@ pub struct AssetManager { /// Residency policy for scene banks. scene_policy: BankPolicy, + load_worker: AssetLoadWorker, + // Commits that are ready to be applied at the next frame boundary. - pending_commits: Mutex>, + pending_commits: Mutex>, } struct LoadHandleInfo { _asset_id: AssetId, slot: SlotRef, status: LoadStatus, + request_generation: u64, + progress: u16, +} + +#[derive(Clone)] +struct AssetLoadJob { + handle_id: HandleId, + asset_id: AssetId, + slot: SlotRef, + request_generation: u64, + entry: AssetEntry, +} + +struct AssetLoadJobResources<'a> { + handles: &'a Arc>, + target_generations: &'a Arc>, + assets_data: &'a Arc>, + pipeline_telemetry: &'a Arc>, + gfx_policy: &'a BankPolicy, + sound_policy: &'a BankPolicy, + scene_policy: &'a BankPolicy, +} + +#[derive(Default)] +struct AssetLoadQueueState { + pending: VecDeque, + active: Option, + shutdown: bool, +} + +#[derive(Default)] +struct AssetLoadQueue { + state: Mutex, + ready: Condvar, +} + +impl AssetLoadQueue { + fn submit(&self, job: AssetLoadJob) -> usize { + let mut state = self.state.lock().unwrap(); + if state.shutdown { + return 0; + } + let before_len = state.pending.len(); + state.pending.retain(|pending| pending.slot != job.slot); + let removed = before_len - state.pending.len(); + state.pending.push_back(job); + self.ready.notify_one(); + removed + } + + fn next_job(&self) -> Option { + let mut state = self.state.lock().unwrap(); + loop { + if let Some(job) = state.pending.pop_front() { + state.active = Some(job.clone()); + return Some(job); + } + if state.shutdown { + return None; + } + state = self.ready.wait(state).unwrap(); + } + } + + fn clear_pending(&self) { + self.state.lock().unwrap().pending.clear(); + } + + fn complete_active(&self, handle_id: HandleId, request_generation: u64) { + let mut state = self.state.lock().unwrap(); + if state.active.as_ref().is_some_and(|job| { + job.handle_id == handle_id && job.request_generation == request_generation + }) { + state.active = None; + } + } + + fn pending_count(&self) -> usize { + self.state.lock().unwrap().pending.len() + } + + fn active_job(&self) -> Option { + self.state.lock().unwrap().active.clone() + } + + fn pending_position(&self, handle_id: HandleId) -> Option { + self.state + .lock() + .unwrap() + .pending + .iter() + .position(|job| job.handle_id == handle_id) + .map(|index| index + 1) + } + + fn move_pending(&self, handle_id: HandleId, new_position: usize) -> bool { + if new_position == 0 { + return false; + } + + let mut state = self.state.lock().unwrap(); + let Some(current_index) = state.pending.iter().position(|job| job.handle_id == handle_id) + else { + return false; + }; + let Some(job) = state.pending.remove(current_index) else { + return false; + }; + let insert_index = new_position.saturating_sub(1).min(state.pending.len()); + state.pending.insert(insert_index, job); + true + } + + fn shutdown(&self) { + let mut state = self.state.lock().unwrap(); + state.pending.clear(); + state.shutdown = true; + self.ready.notify_all(); + } +} + +struct AssetLoadWorker { + queue: Arc, + join_handle: Option>, +} + +impl AssetLoadWorker { + fn new( + handles: Arc>, + target_generations: Arc>, + assets_data: Arc>, + pipeline_telemetry: Arc>, + gfx_policy: BankPolicy, + sound_policy: BankPolicy, + scene_policy: BankPolicy, + ) -> Self { + let queue = Arc::new(AssetLoadQueue::default()); + let worker_queue = Arc::clone(&queue); + let join_handle = thread::spawn(move || { + let resources = AssetLoadJobResources { + handles: &handles, + target_generations: &target_generations, + assets_data: &assets_data, + pipeline_telemetry: &pipeline_telemetry, + gfx_policy: &gfx_policy, + sound_policy: &sound_policy, + scene_policy: &scene_policy, + }; + while let Some(job) = worker_queue.next_job() { + AssetManager::process_load_job(job.clone(), &resources); + worker_queue.complete_active(job.handle_id, job.request_generation); + } + }); + + Self { queue, join_handle: Some(join_handle) } + } + + fn submit_counting_superseded(&self, job: AssetLoadJob) -> usize { + self.queue.submit(job) + } + + fn clear_pending(&self) { + self.queue.clear_pending(); + } + + fn pending_count(&self) -> usize { + self.queue.pending_count() + } + + fn active_job(&self) -> Option { + self.queue.active_job() + } + + fn pending_position(&self, handle_id: HandleId) -> Option { + self.queue.pending_position(handle_id) + } + + fn move_pending(&self, handle_id: HandleId, new_position: usize) -> bool { + self.queue.move_pending(handle_id, new_position) + } +} + +impl Drop for AssetLoadWorker { + fn drop(&mut self) { + self.queue.shutdown(); + if let Some(join_handle) = self.join_handle.take() { + let _ = join_handle.join(); + } + } } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -240,6 +564,24 @@ impl AssetBridge for AssetManager { fn cancel(&self, handle: HandleId) -> AssetOpStatus { self.cancel(handle) } + fn backlog_info(&self) -> AssetBacklogInfo { + self.backlog_info() + } + fn backlog_position(&self, handle: HandleId) -> AssetBacklogPosition { + self.backlog_position(handle) + } + fn backlog_move(&self, handle: HandleId, new_position: usize) -> AssetOpStatus { + self.backlog_move(handle, new_position) + } + fn backlog_promote(&self, handle: HandleId) -> AssetOpStatus { + self.backlog_promote(handle) + } + fn backlog_demote(&self, handle: HandleId) -> AssetOpStatus { + self.backlog_demote(handle) + } + fn target_status(&self, bank_type: BankType, slot: usize) -> AssetTargetStatus { + self.target_status(bank_type, slot) + } fn apply_commits(&self) { self.apply_commits() } @@ -333,6 +675,23 @@ impl AssetManager { asset_map.insert(entry.asset_id, entry); } + let handles = Arc::new(RwLock::new(HashMap::new())); + let target_generations = Arc::new(RwLock::new(HashMap::new())); + let assets_data = Arc::new(RwLock::new(assets_data)); + let pipeline_telemetry = Arc::new(Mutex::new(AssetPipelineTelemetryState::default())); + let gfx_policy = BankPolicy::new(); + let sound_policy = BankPolicy::new(); + let scene_policy = BankPolicy::new(); + let load_worker = AssetLoadWorker::new( + Arc::clone(&handles), + Arc::clone(&target_generations), + Arc::clone(&assets_data), + Arc::clone(&pipeline_telemetry), + gfx_policy.clone(), + sound_policy.clone(), + scene_policy.clone(), + ); + Self { assets: Arc::new(RwLock::new(asset_map)), gfx_installer, @@ -342,12 +701,16 @@ impl AssetManager { glyph_slot_index: GlyphAssetSlotIndex::new(), sound_slots: Arc::new(RwLock::new(std::array::from_fn(|_| None))), scene_slots: Arc::new(RwLock::new(std::array::from_fn(|_| None))), - gfx_policy: BankPolicy::new(), - sound_policy: BankPolicy::new(), - scene_policy: BankPolicy::new(), - handles: Arc::new(RwLock::new(HashMap::new())), + gfx_policy, + sound_policy, + scene_policy, + handles, + target_handles: Arc::new(RwLock::new(HashMap::new())), + target_generations, next_handle_id: Mutex::new(1), - assets_data: Arc::new(RwLock::new(assets_data)), + assets_data, + pipeline_telemetry, + load_worker, pending_commits: Mutex::new(Vec::new()), } } @@ -435,6 +798,71 @@ impl AssetManager { } } + fn handle_for_slot(&self, slot: SlotRef) -> HandleId { + if let Some(handle_id) = self.target_handles.read().unwrap().get(&slot).copied() { + return handle_id; + } + + let mut target_handles = self.target_handles.write().unwrap(); + if let Some(handle_id) = target_handles.get(&slot).copied() { + return handle_id; + } + + let mut next_id = self.next_handle_id.lock().unwrap(); + let handle_id = *next_id; + *next_id += 1; + target_handles.insert(slot, handle_id); + handle_id + } + + fn bump_target_generation(&self, slot: SlotRef) -> u64 { + let mut generations = self.target_generations.write().unwrap(); + let generation = generations.entry(slot).or_insert(0); + *generation += 1; + *generation + } + + fn current_target_generation( + target_generations: &Arc>, + slot: SlotRef, + ) -> u64 { + target_generations.read().unwrap().get(&slot).copied().unwrap_or(0) + } + + fn is_current_target_generation( + target_generations: &Arc>, + slot: SlotRef, + request_generation: u64, + ) -> bool { + Self::current_target_generation(target_generations, slot) == request_generation + } + + fn clear_staging_for_handle(&self, handle_id: HandleId) { + self.gfx_policy.take_staging(handle_id); + self.sound_policy.take_staging(handle_id); + self.scene_policy.take_staging(handle_id); + } + + fn target_for_handle(&self, handle_id: HandleId) -> Option { + self.target_handles + .read() + .unwrap() + .iter() + .find_map(|(slot, target_handle)| (*target_handle == handle_id).then_some(*slot)) + } + + fn slot_asset_id(&self, slot: SlotRef) -> Option { + match slot.asset_type { + BankType::GLYPH => self.gfx_slots.read().unwrap().get(slot.index).and_then(|s| *s), + BankType::SOUNDS => self.sound_slots.read().unwrap().get(slot.index).and_then(|s| *s), + BankType::SCENE => self.scene_slots.read().unwrap().get(slot.index).and_then(|s| *s), + } + } + + fn idle_target_state(&self, slot: SlotRef) -> LoadStatus { + if self.slot_asset_id(slot).is_some() { LoadStatus::COMMITTED } else { LoadStatus::EMPTY } + } + pub fn load(&self, asset_id: AssetId, slot_index: usize) -> Result { if slot_index >= 16 { return Err(AssetLoadError::SlotIndexInvalid); @@ -449,9 +877,9 @@ impl AssetManager { BankType::SCENE => SlotRef::scene(slot_index), }; - let mut next_id = self.next_handle_id.lock().unwrap(); - let handle_id = *next_id; - *next_id += 1; + let handle_id = self.handle_for_slot(slot); + let request_generation = self.bump_target_generation(slot); + self.clear_staging_for_handle(handle_id); // Check if already resident (Dedup) let already_resident = match entry.bank_type { @@ -484,132 +912,264 @@ impl AssetManager { if already_resident { self.handles.write().unwrap().insert( handle_id, - LoadHandleInfo { _asset_id: asset_id, slot, status: LoadStatus::READY }, + LoadHandleInfo { + _asset_id: asset_id, + slot, + status: LoadStatus::READY, + request_generation, + progress: ASSET_PROGRESS_DONE, + }, ); + self.pipeline_telemetry.lock().unwrap().record_submitted(0); + self.pipeline_telemetry.lock().unwrap().record_completed(asset_id, entry.bank_type, 0); return Ok(handle_id); } // Not resident, start loading self.handles.write().unwrap().insert( handle_id, - LoadHandleInfo { _asset_id: asset_id, slot, status: LoadStatus::PENDING }, + LoadHandleInfo { + _asset_id: asset_id, + slot, + status: LoadStatus::PENDING, + request_generation, + progress: ASSET_PROGRESS_IDLE, + }, ); - let handles = self.handles.clone(); - let assets_data = self.assets_data.clone(); - let entry_clone = entry.clone(); - - // Capture policies for the worker thread - let gfx_policy = self.gfx_policy.clone(); - let sound_policy = self.sound_policy.clone(); - let scene_policy = self.scene_policy.clone(); - - thread::spawn(move || { - // Update status to LOADING - { - let mut handles_map = handles.write().unwrap(); - if let Some(h) = handles_map.get_mut(&handle_id) { - if h.status == LoadStatus::PENDING { - h.status = LoadStatus::LOADING; - } else { - return; - } - } else { - return; - } - } - - match entry_clone.bank_type { - BankType::GLYPH => { - let result = Self::perform_load_glyph_bank(&entry_clone, assets_data); - if let Ok(tilebank) = result { - let bank_arc = Arc::new(tilebank); - let resident_arc = gfx_policy.put_resident( - asset_id, - bank_arc, - entry_clone.decoded_size as usize, - ); - gfx_policy.stage( - handle_id, - resident_arc, - entry_clone.decoded_size as usize, - ); - - let mut handles_map = handles.write().unwrap(); - if let Some(h) = handles_map.get_mut(&handle_id) { - if h.status == LoadStatus::LOADING { - h.status = LoadStatus::READY; - } - } - } else { - let mut handles_map = handles.write().unwrap(); - if let Some(h) = handles_map.get_mut(&handle_id) { - h.status = LoadStatus::ERROR; - } - } - } - BankType::SOUNDS => { - let result = Self::perform_load_sound_bank(&entry_clone, assets_data); - if let Ok(soundbank) = result { - let bank_arc = Arc::new(soundbank); - let resident_arc = sound_policy.put_resident( - asset_id, - bank_arc, - entry_clone.decoded_size as usize, - ); - sound_policy.stage( - handle_id, - resident_arc, - entry_clone.decoded_size as usize, - ); - - let mut handles_map = handles.write().unwrap(); - if let Some(h) = handles_map.get_mut(&handle_id) { - if h.status == LoadStatus::LOADING { - h.status = LoadStatus::READY; - } - } - } else { - let mut handles_map = handles.write().unwrap(); - if let Some(h) = handles_map.get_mut(&handle_id) { - h.status = LoadStatus::ERROR; - } - } - } - BankType::SCENE => { - let result = Self::perform_load_scene_bank(&entry_clone, assets_data); - if let Ok(scenebank) = result { - let bank_arc = Arc::new(scenebank); - let resident_arc = scene_policy.put_resident( - asset_id, - bank_arc, - entry_clone.decoded_size as usize, - ); - scene_policy.stage( - handle_id, - resident_arc, - entry_clone.decoded_size as usize, - ); - - let mut handles_map = handles.write().unwrap(); - if let Some(h) = handles_map.get_mut(&handle_id) { - if h.status == LoadStatus::LOADING { - h.status = LoadStatus::READY; - } - } - } else { - let mut handles_map = handles.write().unwrap(); - if let Some(h) = handles_map.get_mut(&handle_id) { - h.status = LoadStatus::ERROR; - } - } - } - } + let superseded = self.load_worker.submit_counting_superseded(AssetLoadJob { + handle_id, + asset_id, + slot, + request_generation, + entry, }); + let backlog_depth = self.load_worker.pending_count(); + let mut telemetry = self.pipeline_telemetry.lock().unwrap(); + telemetry.record_submitted(backlog_depth); + telemetry.record_superseded(superseded); Ok(handle_id) } + fn process_load_job(job: AssetLoadJob, resources: &AssetLoadJobResources<'_>) { + if !Self::is_current_target_generation( + resources.target_generations, + job.slot, + job.request_generation, + ) { + resources.pipeline_telemetry.lock().unwrap().record_stale_discard(); + return; + } + + { + let mut handles_map = resources.handles.write().unwrap(); + let Some(handle) = handles_map.get_mut(&job.handle_id) else { + return; + }; + if handle.request_generation != job.request_generation + || handle.status != LoadStatus::PENDING + { + return; + } + handle.status = LoadStatus::LOADING; + handle.progress = ASSET_PROGRESS_LOADING; + } + let started_at = Instant::now(); + resources.pipeline_telemetry.lock().unwrap().record_started(0); + + match job.entry.bank_type { + BankType::GLYPH => { + let result = + Self::perform_load_glyph_bank(&job.entry, Arc::clone(resources.assets_data)); + match result { + Ok(tilebank) => { + if !Self::is_current_target_generation( + resources.target_generations, + job.slot, + job.request_generation, + ) { + resources.pipeline_telemetry.lock().unwrap().record_stale_discard(); + return; + } + let bank_arc = Arc::new(tilebank); + let resident_arc = resources.gfx_policy.put_resident( + job.asset_id, + bank_arc, + job.entry.decoded_size as usize, + ); + if !Self::is_current_target_generation( + resources.target_generations, + job.slot, + job.request_generation, + ) { + resources.pipeline_telemetry.lock().unwrap().record_stale_discard(); + return; + } + resources.gfx_policy.stage( + job.handle_id, + resident_arc, + job.entry.decoded_size as usize, + ); + Self::complete_load_job(resources.handles, &job, LoadStatus::READY); + Self::record_load_job_finished( + resources.pipeline_telemetry, + &job, + started_at, + LoadStatus::READY, + ); + } + Err(_) => { + Self::complete_load_job(resources.handles, &job, LoadStatus::ERROR); + Self::record_load_job_finished( + resources.pipeline_telemetry, + &job, + started_at, + LoadStatus::ERROR, + ); + } + } + } + BankType::SOUNDS => { + let result = + Self::perform_load_sound_bank(&job.entry, Arc::clone(resources.assets_data)); + match result { + Ok(soundbank) => { + if !Self::is_current_target_generation( + resources.target_generations, + job.slot, + job.request_generation, + ) { + resources.pipeline_telemetry.lock().unwrap().record_stale_discard(); + return; + } + let bank_arc = Arc::new(soundbank); + let resident_arc = resources.sound_policy.put_resident( + job.asset_id, + bank_arc, + job.entry.decoded_size as usize, + ); + if !Self::is_current_target_generation( + resources.target_generations, + job.slot, + job.request_generation, + ) { + resources.pipeline_telemetry.lock().unwrap().record_stale_discard(); + return; + } + resources.sound_policy.stage( + job.handle_id, + resident_arc, + job.entry.decoded_size as usize, + ); + Self::complete_load_job(resources.handles, &job, LoadStatus::READY); + Self::record_load_job_finished( + resources.pipeline_telemetry, + &job, + started_at, + LoadStatus::READY, + ); + } + Err(_) => { + Self::complete_load_job(resources.handles, &job, LoadStatus::ERROR); + Self::record_load_job_finished( + resources.pipeline_telemetry, + &job, + started_at, + LoadStatus::ERROR, + ); + } + } + } + BankType::SCENE => { + let result = + Self::perform_load_scene_bank(&job.entry, Arc::clone(resources.assets_data)); + match result { + Ok(scenebank) => { + if !Self::is_current_target_generation( + resources.target_generations, + job.slot, + job.request_generation, + ) { + resources.pipeline_telemetry.lock().unwrap().record_stale_discard(); + return; + } + let bank_arc = Arc::new(scenebank); + let resident_arc = resources.scene_policy.put_resident( + job.asset_id, + bank_arc, + job.entry.decoded_size as usize, + ); + if !Self::is_current_target_generation( + resources.target_generations, + job.slot, + job.request_generation, + ) { + resources.pipeline_telemetry.lock().unwrap().record_stale_discard(); + return; + } + resources.scene_policy.stage( + job.handle_id, + resident_arc, + job.entry.decoded_size as usize, + ); + Self::complete_load_job(resources.handles, &job, LoadStatus::READY); + Self::record_load_job_finished( + resources.pipeline_telemetry, + &job, + started_at, + LoadStatus::READY, + ); + } + Err(_) => { + Self::complete_load_job(resources.handles, &job, LoadStatus::ERROR); + Self::record_load_job_finished( + resources.pipeline_telemetry, + &job, + started_at, + LoadStatus::ERROR, + ); + } + } + } + } + } + + fn record_load_job_finished( + pipeline_telemetry: &Arc>, + job: &AssetLoadJob, + started_at: Instant, + status: LoadStatus, + ) { + let duration_micros = started_at.elapsed().as_micros() as u64; + let mut telemetry = pipeline_telemetry.lock().unwrap(); + match status { + LoadStatus::READY => { + telemetry.record_completed(job.asset_id, job.entry.bank_type, duration_micros) + } + LoadStatus::ERROR => { + telemetry.record_failed(job.asset_id, job.entry.bank_type, duration_micros) + } + _ => {} + } + } + + fn complete_load_job( + handles: &Arc>, + job: &AssetLoadJob, + status: LoadStatus, + ) { + let mut handles_map = handles.write().unwrap(); + if let Some(handle) = handles_map.get_mut(&job.handle_id) { + if handle.request_generation == job.request_generation + && handle.status == LoadStatus::LOADING + { + handle.status = status; + handle.progress = ASSET_PROGRESS_DONE; + } + } + } + fn perform_load_glyph_bank( entry: &AssetEntry, assets_data: Arc>, @@ -946,12 +1506,160 @@ impl AssetManager { } pub fn status(&self, handle: HandleId) -> LoadStatus { - self.handles + if let Some(status) = self.handles.read().unwrap().get(&handle).map(|h| h.status) { + return status; + } + + self.target_for_handle(handle) + .map(|slot| self.idle_target_state(slot)) + .unwrap_or(LoadStatus::UnknownHandle) + } + + pub fn backlog_info(&self) -> AssetBacklogInfo { + let active = self.load_worker.active_job(); + let active_progress = active + .as_ref() + .and_then(|job| self.handles.read().unwrap().get(&job.handle_id).map(|h| h.progress)) + .unwrap_or(ASSET_PROGRESS_IDLE); + AssetBacklogInfo { + status: AssetOpStatus::Ok, + pending_count: self.load_worker.pending_count() as u32, + active_handle: active.as_ref().map(|job| job.handle_id).unwrap_or(0), + active_asset_id: active.as_ref().map(|job| job.asset_id), + active_bank_type: active.as_ref().map(|job| job.slot.asset_type), + active_slot: active.as_ref().map(|job| job.slot.index), + active_progress, + } + } + + pub fn backlog_position(&self, handle: HandleId) -> AssetBacklogPosition { + let Some(slot) = self.target_for_handle(handle) else { + return AssetBacklogPosition { + status: AssetOpStatus::UnknownHandle, + state: LoadStatus::UnknownHandle, + position: 0, + progress: 0, + }; + }; + + if let Some(position) = self.load_worker.pending_position(handle) { + return AssetBacklogPosition { + status: AssetOpStatus::Ok, + state: LoadStatus::QUEUED, + position: position as u32, + progress: self + .handles + .read() + .unwrap() + .get(&handle) + .map(|h| h.progress) + .unwrap_or(ASSET_PROGRESS_IDLE), + }; + } + + if self.load_worker.active_job().as_ref().is_some_and(|job| job.handle_id == handle) { + return AssetBacklogPosition { + status: AssetOpStatus::Ok, + state: LoadStatus::ACTIVE, + position: 0, + progress: self + .handles + .read() + .unwrap() + .get(&handle) + .map(|h| h.progress) + .unwrap_or(ASSET_PROGRESS_LOADING), + }; + } + + AssetBacklogPosition { + status: AssetOpStatus::Ok, + state: self + .handles + .read() + .unwrap() + .get(&handle) + .map(|h| h.status) + .unwrap_or_else(|| self.idle_target_state(slot)), + position: 0, + progress: self + .handles + .read() + .unwrap() + .get(&handle) + .map(|h| h.progress) + .unwrap_or(ASSET_PROGRESS_IDLE), + } + } + + pub fn backlog_move(&self, handle: HandleId, new_position: usize) -> AssetOpStatus { + if self.target_for_handle(handle).is_none() { + return AssetOpStatus::UnknownHandle; + } + if self.load_worker.move_pending(handle, new_position) { + AssetOpStatus::Ok + } else { + AssetOpStatus::InvalidState + } + } + + pub fn backlog_promote(&self, handle: HandleId) -> AssetOpStatus { + self.backlog_move(handle, 1) + } + + pub fn backlog_demote(&self, handle: HandleId) -> AssetOpStatus { + if self.target_for_handle(handle).is_none() { + return AssetOpStatus::UnknownHandle; + } + let new_position = self.load_worker.pending_count(); + if new_position == 0 { + return AssetOpStatus::InvalidState; + } + self.backlog_move(handle, new_position) + } + + pub fn target_status(&self, bank_type: BankType, slot: usize) -> AssetTargetStatus { + if slot >= 16 { + return AssetTargetStatus { + status: AssetOpStatus::InvalidState, + asset_id: None, + handle: 0, + state: LoadStatus::INVALID, + position: 0, + progress: 0, + }; + } + + let slot_ref = match bank_type { + BankType::GLYPH => SlotRef::gfx(slot), + BankType::SOUNDS => SlotRef::audio(slot), + BankType::SCENE => SlotRef::scene(slot), + }; + let handle = self.handle_for_slot(slot_ref); + let position = self.backlog_position(handle); + let asset_id = self + .handles .read() .unwrap() .get(&handle) - .map(|h| h.status) - .unwrap_or(LoadStatus::UnknownHandle) + .map(|h| h._asset_id) + .or_else(|| self.slot_asset_id(slot_ref)); + + AssetTargetStatus { + status: AssetOpStatus::Ok, + asset_id, + handle, + state: position.state, + position: position.position, + progress: position.progress, + } + } + + pub fn pipeline_telemetry(&self) -> AssetPipelineTelemetry { + let mut snapshot = self.pipeline_telemetry.lock().unwrap().snapshot.clone(); + snapshot.current_backlog_depth = self.load_worker.pending_count(); + snapshot.active_progress = self.backlog_info().active_progress; + snapshot } pub fn commit(&self, handle: HandleId) -> AssetOpStatus { @@ -960,7 +1668,7 @@ impl AssetManager { return AssetOpStatus::UnknownHandle; }; if h.status == LoadStatus::READY { - self.pending_commits.lock().unwrap().push(handle); + self.pending_commits.lock().unwrap().push((handle, h.request_generation)); AssetOpStatus::Ok } else { AssetOpStatus::InvalidState @@ -969,23 +1677,32 @@ impl AssetManager { pub fn cancel(&self, handle: HandleId) -> AssetOpStatus { let mut final_status = AssetOpStatus::UnknownHandle; - let mut handles_map = self.handles.write().unwrap(); - if let Some(h) = handles_map.get_mut(&handle) { - final_status = match h.status { - LoadStatus::PENDING | LoadStatus::LOADING | LoadStatus::READY => AssetOpStatus::Ok, - LoadStatus::CANCELED => AssetOpStatus::Ok, - _ => AssetOpStatus::InvalidState, - }; - match h.status { - LoadStatus::PENDING | LoadStatus::LOADING | LoadStatus::READY => { - h.status = LoadStatus::CANCELED; + let mut canceled_slot = None; + { + let mut handles_map = self.handles.write().unwrap(); + if let Some(h) = handles_map.get_mut(&handle) { + final_status = match h.status { + LoadStatus::PENDING | LoadStatus::LOADING | LoadStatus::READY => { + AssetOpStatus::Ok + } + LoadStatus::CANCELED => AssetOpStatus::Ok, + _ => AssetOpStatus::InvalidState, + }; + match h.status { + LoadStatus::PENDING | LoadStatus::LOADING | LoadStatus::READY => { + h.status = LoadStatus::CANCELED; + h.progress = ASSET_PROGRESS_DONE; + canceled_slot = Some(h.slot); + } + _ => {} } - _ => {} } } - self.gfx_policy.take_staging(handle); - self.sound_policy.take_staging(handle); - self.scene_policy.take_staging(handle); + if let Some(slot) = canceled_slot { + self.bump_target_generation(slot); + self.pipeline_telemetry.lock().unwrap().record_canceled(); + } + self.clear_staging_for_handle(handle); final_status } @@ -993,9 +1710,9 @@ impl AssetManager { let mut pending = self.pending_commits.lock().unwrap(); let mut handles = self.handles.write().unwrap(); - for handle_id in pending.drain(..) { + for (handle_id, request_generation) in pending.drain(..) { if let Some(h) = handles.get_mut(&handle_id) { - if h.status == LoadStatus::READY { + if h.status == LoadStatus::READY && h.request_generation == request_generation { match h.slot.asset_type { BankType::GLYPH => { if let Some((bank, _)) = self.gfx_policy.take_staging(handle_id) { @@ -1131,6 +1848,11 @@ impl AssetManager { self.sound_policy.clear(); self.scene_policy.clear(); self.handles.write().unwrap().clear(); + self.target_handles.write().unwrap().clear(); + for generation in self.target_generations.write().unwrap().values_mut() { + *generation += 1; + } + self.load_worker.clear_pending(); self.pending_commits.lock().unwrap().clear(); self.gfx_slots.write().unwrap().fill(None); self.glyph_slot_index.clear(); @@ -1526,6 +2248,268 @@ mod tests { assert!(Arc::ptr_eq(bank1, bank2)); } + #[test] + fn test_asset_load_uses_stable_handle_for_target_slot() { + let banks = Arc::new(MemoryBanks::new()); + let gfx_installer = Arc::clone(&banks) as Arc; + let sound_installer = Arc::clone(&banks) as Arc; + let scene_installer = Arc::clone(&banks) as Arc; + + let data = test_glyph_asset_data(); + let asset_entry = test_glyph_asset_entry("test_glyphs", 16, 16); + + let am = AssetManager::new( + vec![asset_entry], + AssetsPayloadSource::from_bytes(data), + gfx_installer, + sound_installer, + scene_installer, + ); + + let first_handle = am.load(0, 0).unwrap(); + let second_handle = am.load(0, 0).unwrap(); + + assert_eq!(first_handle, second_handle); + + let start = Instant::now(); + while am.status(second_handle) != LoadStatus::READY && start.elapsed().as_secs() < 5 { + thread::sleep(std::time::Duration::from_millis(10)); + } + assert_eq!(am.status(second_handle), LoadStatus::READY); + } + + #[test] + fn test_asset_load_supersedes_previous_request_for_target_slot() { + let banks = Arc::new(MemoryBanks::new()); + let gfx_installer = Arc::clone(&banks) as Arc; + let sound_installer = Arc::clone(&banks) as Arc; + let scene_installer = Arc::clone(&banks) as Arc; + + let first_payload = test_glyph_asset_data(); + let second_payload = test_glyph_asset_data(); + let second_offset = first_payload.len(); + let mut data = first_payload; + data.extend_from_slice(&second_payload); + + let first_asset = test_glyph_asset_entry("first_glyphs", 16, 16); + let mut second_asset = test_glyph_asset_entry("second_glyphs", 16, 16); + second_asset.asset_id = 1; + second_asset.offset = second_offset as u64; + + let am = AssetManager::new( + vec![first_asset, second_asset], + AssetsPayloadSource::from_bytes(data), + gfx_installer, + sound_installer, + scene_installer, + ); + + let first_handle = am.load(0, 0).unwrap(); + let second_handle = am.load(1, 0).unwrap(); + + assert_eq!(first_handle, second_handle); + + let start = Instant::now(); + while am.status(second_handle) != LoadStatus::READY && start.elapsed().as_secs() < 5 { + thread::sleep(std::time::Duration::from_millis(10)); + } + assert_eq!(am.status(second_handle), LoadStatus::READY); + + assert_eq!(am.commit(second_handle), AssetOpStatus::Ok); + am.apply_commits(); + + let slot = am.slot_info(SlotRef::gfx(0)); + assert_eq!(slot.asset_id, Some(1)); + } + + #[test] + fn test_asset_backlog_queue_reorders_pending_requests() { + let queue = AssetLoadQueue::default(); + let entry = test_glyph_asset_entry("queued_glyphs", 16, 16); + let job = |handle_id: HandleId, slot: usize| AssetLoadJob { + handle_id, + asset_id: handle_id as AssetId, + slot: SlotRef::gfx(slot), + request_generation: handle_id as u64, + entry: entry.clone(), + }; + + queue.submit(job(1, 0)); + queue.submit(job(2, 1)); + queue.submit(job(3, 2)); + + assert_eq!(queue.pending_position(3), Some(3)); + assert!(queue.move_pending(3, 1)); + assert_eq!(queue.pending_position(3), Some(1)); + assert_eq!(queue.pending_position(1), Some(2)); + assert!(queue.move_pending(3, queue.pending_count())); + assert_eq!(queue.pending_position(3), Some(3)); + } + + #[test] + fn test_asset_backlog_queue_keeps_only_latest_request_per_target() { + let queue = AssetLoadQueue::default(); + let entry = test_glyph_asset_entry("queued_glyphs", 16, 16); + let job = |handle_id: HandleId, request_generation: u64| AssetLoadJob { + handle_id, + asset_id: handle_id as AssetId, + slot: SlotRef::gfx(0), + request_generation, + entry: entry.clone(), + }; + + assert_eq!(queue.submit(job(1, 1)), 0); + assert_eq!(queue.submit(job(2, 2)), 1); + + assert_eq!(queue.pending_count(), 1); + assert_eq!(queue.pending_position(1), None); + assert_eq!(queue.pending_position(2), Some(1)); + } + + #[test] + fn test_asset_target_status_exposes_empty_and_ready_slot_handle() { + let banks = Arc::new(MemoryBanks::new()); + let gfx_installer = Arc::clone(&banks) as Arc; + let sound_installer = Arc::clone(&banks) as Arc; + let scene_installer = Arc::clone(&banks) as Arc; + + let data = test_glyph_asset_data(); + let asset_entry = test_glyph_asset_entry("test_glyphs", 16, 16); + + let am = AssetManager::new( + vec![asset_entry], + AssetsPayloadSource::from_bytes(data), + gfx_installer, + sound_installer, + scene_installer, + ); + + let empty = am.target_status(BankType::GLYPH, 0); + assert_eq!(empty.status, AssetOpStatus::Ok); + assert_eq!(empty.asset_id, None); + assert_eq!(empty.state, LoadStatus::EMPTY); + assert_ne!(empty.handle, 0); + assert_eq!(am.status(empty.handle), LoadStatus::EMPTY); + + let handle = am.load(0, 0).unwrap(); + assert_eq!(handle, empty.handle); + + let start = Instant::now(); + while am.status(handle) != LoadStatus::READY && start.elapsed().as_secs() < 5 { + thread::sleep(std::time::Duration::from_millis(10)); + } + + let ready = am.target_status(BankType::GLYPH, 0); + assert_eq!(ready.status, AssetOpStatus::Ok); + assert_eq!(ready.asset_id, Some(0)); + assert_eq!(ready.handle, handle); + assert_eq!(ready.state, LoadStatus::READY); + } + + #[test] + fn test_asset_pipeline_progress_and_telemetry_snapshot_for_success() { + let banks = Arc::new(MemoryBanks::new()); + let gfx_installer = Arc::clone(&banks) as Arc; + let sound_installer = Arc::clone(&banks) as Arc; + let scene_installer = Arc::clone(&banks) as Arc; + + let data = test_glyph_asset_data(); + let asset_entry = test_glyph_asset_entry("test_glyphs", 16, 16); + + let am = AssetManager::new( + vec![asset_entry], + AssetsPayloadSource::from_bytes(data), + gfx_installer, + sound_installer, + scene_installer, + ); + + let empty = am.target_status(BankType::GLYPH, 0); + assert_eq!(empty.progress, ASSET_PROGRESS_IDLE); + + let handle = am.load(0, 0).unwrap(); + let start = Instant::now(); + while am.status(handle) != LoadStatus::READY && start.elapsed().as_secs() < 5 { + thread::sleep(std::time::Duration::from_millis(10)); + } + + let position = am.backlog_position(handle); + assert_eq!(position.state, LoadStatus::READY); + assert_eq!(position.progress, ASSET_PROGRESS_DONE); + + let telemetry = am.pipeline_telemetry(); + assert_eq!(telemetry.submitted_requests, 1); + assert_eq!(telemetry.started_jobs, 1); + assert_eq!(telemetry.completed_jobs, 1); + assert_eq!(telemetry.failed_jobs, 0); + assert_eq!(telemetry.active_progress, ASSET_PROGRESS_IDLE); + assert_eq!(telemetry.last_asset_id, Some(0)); + } + + #[test] + fn test_asset_pipeline_progress_and_telemetry_snapshot_for_failure() { + let banks = Arc::new(MemoryBanks::new()); + let gfx_installer = Arc::clone(&banks) as Arc; + let sound_installer = Arc::clone(&banks) as Arc; + let scene_installer = Arc::clone(&banks) as Arc; + + let asset_entry = test_glyph_asset_entry("broken_glyphs", 16, 16); + + let am = AssetManager::new( + vec![asset_entry], + AssetsPayloadSource::from_bytes(Vec::new()), + gfx_installer, + sound_installer, + scene_installer, + ); + + let handle = am.load(0, 0).unwrap(); + let start = Instant::now(); + while am.status(handle) != LoadStatus::ERROR && start.elapsed().as_secs() < 5 { + thread::sleep(std::time::Duration::from_millis(10)); + } + + let position = am.backlog_position(handle); + assert_eq!(position.state, LoadStatus::ERROR); + assert_eq!(position.progress, ASSET_PROGRESS_DONE); + + let telemetry = am.pipeline_telemetry(); + assert_eq!(telemetry.submitted_requests, 1); + assert_eq!(telemetry.started_jobs, 1); + assert_eq!(telemetry.completed_jobs, 0); + assert_eq!(telemetry.failed_jobs, 1); + assert_eq!(telemetry.last_asset_id, Some(0)); + } + + #[test] + fn test_asset_backlog_reorder_rejects_non_queued_request() { + let banks = Arc::new(MemoryBanks::new()); + let gfx_installer = Arc::clone(&banks) as Arc; + let sound_installer = Arc::clone(&banks) as Arc; + let scene_installer = Arc::clone(&banks) as Arc; + + let data = test_glyph_asset_data(); + let asset_entry = test_glyph_asset_entry("test_glyphs", 16, 16); + + let am = AssetManager::new( + vec![asset_entry], + AssetsPayloadSource::from_bytes(data), + gfx_installer, + sound_installer, + scene_installer, + ); + + let handle = am.load(0, 0).unwrap(); + let start = Instant::now(); + while am.status(handle) != LoadStatus::READY && start.elapsed().as_secs() < 5 { + thread::sleep(std::time::Duration::from_millis(10)); + } + + assert_eq!(am.backlog_promote(handle), AssetOpStatus::InvalidState); + assert_eq!(am.backlog_demote(handle), AssetOpStatus::InvalidState); + assert_eq!(am.backlog_move(999, 1), AssetOpStatus::UnknownHandle); + } + #[test] fn test_sound_asset_loading() { let banks = Arc::new(MemoryBanks::new()); diff --git a/crates/console/prometeu-hal/src/asset.rs b/crates/console/prometeu-hal/src/asset.rs index aa451b0f..173cfbe6 100644 --- a/crates/console/prometeu-hal/src/asset.rs +++ b/crates/console/prometeu-hal/src/asset.rs @@ -112,6 +112,12 @@ pub enum LoadStatus { CANCELED = 4, ERROR = 5, UnknownHandle = 6, + QUEUED = 7, + ACTIVE = 8, + SUPERSEDED = 9, + EMPTY = 10, + INVALID = 11, + BackendUnavailable = 12, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -129,6 +135,36 @@ pub enum AssetOpStatus { Ok = 0, UnknownHandle = 1, InvalidState = 2, + Superseded = 3, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct AssetBacklogInfo { + pub status: AssetOpStatus, + pub pending_count: u32, + pub active_handle: HandleId, + pub active_asset_id: Option, + pub active_bank_type: Option, + pub active_slot: Option, + pub active_progress: u16, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct AssetBacklogPosition { + pub status: AssetOpStatus, + pub state: LoadStatus, + pub position: u32, + pub progress: u16, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct AssetTargetStatus { + pub status: AssetOpStatus, + pub asset_id: Option, + pub handle: HandleId, + pub state: LoadStatus, + pub position: u32, + pub progress: u16, } #[derive(Debug, Clone, Serialize, Deserialize)] diff --git a/crates/console/prometeu-hal/src/asset_bridge.rs b/crates/console/prometeu-hal/src/asset_bridge.rs index 5c80e3fe..ad4f51f3 100644 --- a/crates/console/prometeu-hal/src/asset_bridge.rs +++ b/crates/console/prometeu-hal/src/asset_bridge.rs @@ -1,6 +1,7 @@ use crate::asset::{ - AssetEntry, AssetId, AssetLoadError, AssetOpStatus, BankTelemetry, HandleId, LoadStatus, - PreloadEntry, SlotRef, SlotStats, + AssetBacklogInfo, AssetBacklogPosition, AssetEntry, AssetId, AssetLoadError, AssetOpStatus, + AssetTargetStatus, BankTelemetry, BankType, HandleId, LoadStatus, PreloadEntry, SlotRef, + SlotStats, }; use crate::cartridge::AssetsPayloadSource; @@ -15,6 +16,12 @@ pub trait AssetBridge { fn status(&self, handle: HandleId) -> LoadStatus; fn commit(&self, handle: HandleId) -> AssetOpStatus; fn cancel(&self, handle: HandleId) -> AssetOpStatus; + fn backlog_info(&self) -> AssetBacklogInfo; + fn backlog_position(&self, handle: HandleId) -> AssetBacklogPosition; + fn backlog_move(&self, handle: HandleId, new_position: usize) -> AssetOpStatus; + fn backlog_promote(&self, handle: HandleId) -> AssetOpStatus; + fn backlog_demote(&self, handle: HandleId) -> AssetOpStatus; + fn target_status(&self, bank_type: BankType, slot: usize) -> AssetTargetStatus; fn apply_commits(&self); fn bank_telemetry(&self) -> Vec; fn slot_info(&self, slot: SlotRef) -> SlotStats; diff --git a/crates/console/prometeu-hal/src/syscalls.rs b/crates/console/prometeu-hal/src/syscalls.rs index 73ae59cf..618e20d4 100644 --- a/crates/console/prometeu-hal/src/syscalls.rs +++ b/crates/console/prometeu-hal/src/syscalls.rs @@ -69,6 +69,12 @@ pub enum Syscall { AssetStatus = 0x6002, AssetCommit = 0x6003, AssetCancel = 0x6004, + AssetBacklogInfo = 0x6005, + AssetBacklogPosition = 0x6006, + AssetBacklogMove = 0x6007, + AssetBacklogPromote = 0x6008, + AssetBacklogDemote = 0x6009, + AssetTargetStatus = 0x600A, BankInfo = 0x6101, } diff --git a/crates/console/prometeu-hal/src/syscalls/domains/asset.rs b/crates/console/prometeu-hal/src/syscalls/domains/asset.rs index 26c7ee30..849c7632 100644 --- a/crates/console/prometeu-hal/src/syscalls/domains/asset.rs +++ b/crates/console/prometeu-hal/src/syscalls/domains/asset.rs @@ -24,4 +24,40 @@ pub(crate) const ENTRIES: &[SyscallRegistryEntry] = &[ .caps(caps::ASSET) .non_deterministic() .cost(20), + SyscallRegistryEntry::builder(Syscall::AssetBacklogInfo, "asset", "backlog_info") + .args(0) + .rets(7) + .caps(caps::ASSET) + .non_deterministic() + .cost(5), + SyscallRegistryEntry::builder(Syscall::AssetBacklogPosition, "asset", "backlog_position") + .args(1) + .rets(4) + .caps(caps::ASSET) + .non_deterministic() + .cost(5), + SyscallRegistryEntry::builder(Syscall::AssetBacklogMove, "asset", "backlog_move") + .args(2) + .rets(1) + .caps(caps::ASSET) + .non_deterministic() + .cost(10), + SyscallRegistryEntry::builder(Syscall::AssetBacklogPromote, "asset", "backlog_promote") + .args(1) + .rets(1) + .caps(caps::ASSET) + .non_deterministic() + .cost(10), + SyscallRegistryEntry::builder(Syscall::AssetBacklogDemote, "asset", "backlog_demote") + .args(1) + .rets(1) + .caps(caps::ASSET) + .non_deterministic() + .cost(10), + SyscallRegistryEntry::builder(Syscall::AssetTargetStatus, "asset", "target_status") + .args(2) + .rets(6) + .caps(caps::ASSET) + .non_deterministic() + .cost(5), ]; diff --git a/crates/console/prometeu-hal/src/syscalls/registry.rs b/crates/console/prometeu-hal/src/syscalls/registry.rs index 65e4d3ad..3d0a08a6 100644 --- a/crates/console/prometeu-hal/src/syscalls/registry.rs +++ b/crates/console/prometeu-hal/src/syscalls/registry.rs @@ -53,6 +53,12 @@ impl Syscall { 0x6002 => Some(Self::AssetStatus), 0x6003 => Some(Self::AssetCommit), 0x6004 => Some(Self::AssetCancel), + 0x6005 => Some(Self::AssetBacklogInfo), + 0x6006 => Some(Self::AssetBacklogPosition), + 0x6007 => Some(Self::AssetBacklogMove), + 0x6008 => Some(Self::AssetBacklogPromote), + 0x6009 => Some(Self::AssetBacklogDemote), + 0x600A => Some(Self::AssetTargetStatus), 0x6101 => Some(Self::BankInfo), _ => None, } @@ -109,6 +115,12 @@ impl Syscall { Self::AssetStatus => "AssetStatus", Self::AssetCommit => "AssetCommit", Self::AssetCancel => "AssetCancel", + Self::AssetBacklogInfo => "AssetBacklogInfo", + Self::AssetBacklogPosition => "AssetBacklogPosition", + Self::AssetBacklogMove => "AssetBacklogMove", + Self::AssetBacklogPromote => "AssetBacklogPromote", + Self::AssetBacklogDemote => "AssetBacklogDemote", + Self::AssetTargetStatus => "AssetTargetStatus", Self::BankInfo => "BankInfo", } } diff --git a/crates/console/prometeu-hal/src/syscalls/tests.rs b/crates/console/prometeu-hal/src/syscalls/tests.rs index 3625b5ad..e4e22814 100644 --- a/crates/console/prometeu-hal/src/syscalls/tests.rs +++ b/crates/console/prometeu-hal/src/syscalls/tests.rs @@ -258,6 +258,30 @@ fn status_first_syscall_signatures_are_pinned() { assert_eq!(asset_cancel.arg_slots, 1); assert_eq!(asset_cancel.ret_slots, 1); + let asset_backlog_info = meta_for(Syscall::AssetBacklogInfo); + assert_eq!(asset_backlog_info.arg_slots, 0); + assert_eq!(asset_backlog_info.ret_slots, 7); + + let asset_backlog_position = meta_for(Syscall::AssetBacklogPosition); + assert_eq!(asset_backlog_position.arg_slots, 1); + assert_eq!(asset_backlog_position.ret_slots, 4); + + let asset_backlog_move = meta_for(Syscall::AssetBacklogMove); + assert_eq!(asset_backlog_move.arg_slots, 2); + assert_eq!(asset_backlog_move.ret_slots, 1); + + let asset_backlog_promote = meta_for(Syscall::AssetBacklogPromote); + assert_eq!(asset_backlog_promote.arg_slots, 1); + assert_eq!(asset_backlog_promote.ret_slots, 1); + + let asset_backlog_demote = meta_for(Syscall::AssetBacklogDemote); + assert_eq!(asset_backlog_demote.arg_slots, 1); + assert_eq!(asset_backlog_demote.ret_slots, 1); + + let asset_target_status = meta_for(Syscall::AssetTargetStatus); + assert_eq!(asset_target_status.arg_slots, 2); + assert_eq!(asset_target_status.ret_slots, 6); + let bank_info = meta_for(Syscall::BankInfo); assert_eq!(bank_info.arg_slots, 1); assert_eq!(bank_info.ret_slots, 2); diff --git a/crates/console/prometeu-system/src/lib.rs b/crates/console/prometeu-system/src/lib.rs index 98c39b85..c0c052f0 100644 --- a/crates/console/prometeu-system/src/lib.rs +++ b/crates/console/prometeu-system/src/lib.rs @@ -7,7 +7,13 @@ pub use crash_report::CrashReport; pub use os::{LifecycleError, LifecycleOperation, SystemOS}; pub use programs::{NativeShellApp, PrometeuHub, SystemProfileAction, SystemProfileUpdate}; pub use prometeu_hal::{RenderWorkerBackend, RenderWorkerFrameSink}; +pub use services::async_work::{ + AsyncWorkActiveJob, AsyncWorkCancelToken, AsyncWorkJobContext, AsyncWorkJobId, + AsyncWorkJobKind, AsyncWorkJobOutcome, AsyncWorkLane, AsyncWorkLaneConfig, + AsyncWorkLaneController, AsyncWorkLaneError, AsyncWorkLaneTelemetry, AsyncWorkPriority, +}; pub use services::fs; +pub use services::memcard::MemcardAsyncLaneOperation; pub use services::process; pub use services::task; pub use services::vm_runtime::{ diff --git a/crates/console/prometeu-system/src/services/async_work.rs b/crates/console/prometeu-system/src/services/async_work.rs new file mode 100644 index 00000000..55c61cd9 --- /dev/null +++ b/crates/console/prometeu-system/src/services/async_work.rs @@ -0,0 +1,601 @@ +use std::collections::VecDeque; +use std::panic::{AssertUnwindSafe, catch_unwind}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::mpsc::{self, Receiver, RecvTimeoutError}; +use std::sync::{Arc, Condvar, Mutex}; +use std::thread::{self, JoinHandle}; +use std::time::Duration; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub struct AsyncWorkJobId(u64); + +impl AsyncWorkJobId { + pub const ZERO: Self = Self(0); + + pub fn new(value: u64) -> Self { + Self(value) + } + + pub fn get(self) -> u64 { + self.0 + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum AsyncWorkJobKind { + Asset, + Memcard, + Fs, + Test, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum AsyncWorkPriority { + MemcardCommitWrite, + FsWriteConfig, + AssetLoad, + NonCriticalReadList, +} + +impl AsyncWorkPriority { + const fn rank(self) -> u8 { + match self { + Self::MemcardCommitWrite => 0, + Self::FsWriteConfig => 1, + Self::AssetLoad => 2, + Self::NonCriticalReadList => 3, + } + } +} + +impl AsyncWorkJobKind { + pub const fn default_priority(self) -> AsyncWorkPriority { + match self { + Self::Memcard => AsyncWorkPriority::MemcardCommitWrite, + Self::Fs => AsyncWorkPriority::FsWriteConfig, + Self::Asset => AsyncWorkPriority::AssetLoad, + Self::Test => AsyncWorkPriority::NonCriticalReadList, + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum AsyncWorkJobOutcome { + Completed, + Canceled, + Failed, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum AsyncWorkLaneError { + Shutdown, + WorkerPanic, + ShutdownTimeout, + InternalFailure, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct AsyncWorkLaneConfig { + pub shutdown_timeout: Duration, +} + +impl Default for AsyncWorkLaneConfig { + fn default() -> Self { + Self { shutdown_timeout: Duration::from_millis(250) } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct AsyncWorkActiveJob { + pub id: AsyncWorkJobId, + pub kind: AsyncWorkJobKind, + pub priority: AsyncWorkPriority, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub struct AsyncWorkLaneTelemetry { + pub submitted_jobs: u64, + pub started_jobs: u64, + pub completed_jobs: u64, + pub canceled_jobs: u64, + pub failed_jobs: u64, + pub shutdown_discards: u64, + pub pending_depth: usize, + pub max_pending_depth: usize, + pub active_job_id: Option, + pub last_submitted_job_id: Option, + pub last_started_job_id: Option, + pub last_closed_job_id: Option, +} + +impl AsyncWorkLaneTelemetry { + fn record_submitted(&mut self, job_id: AsyncWorkJobId, pending_depth: usize) { + self.submitted_jobs += 1; + self.pending_depth = pending_depth; + self.max_pending_depth = self.max_pending_depth.max(pending_depth); + self.last_submitted_job_id = Some(job_id); + } + + fn record_started(&mut self, job_id: AsyncWorkJobId, pending_depth: usize) { + self.started_jobs += 1; + self.pending_depth = pending_depth; + self.active_job_id = Some(job_id); + self.last_started_job_id = Some(job_id); + } + + fn record_closed(&mut self, job_id: AsyncWorkJobId, outcome: AsyncWorkJobOutcome) { + match outcome { + AsyncWorkJobOutcome::Completed => self.completed_jobs += 1, + AsyncWorkJobOutcome::Canceled => self.canceled_jobs += 1, + AsyncWorkJobOutcome::Failed => self.failed_jobs += 1, + } + self.active_job_id = None; + self.last_closed_job_id = Some(job_id); + } + + fn record_shutdown_discards(&mut self, count: usize) { + self.shutdown_discards += count as u64; + self.pending_depth = 0; + } +} + +#[derive(Clone, Debug)] +pub struct AsyncWorkCancelToken { + canceled: Arc, +} + +impl AsyncWorkCancelToken { + fn new() -> Self { + Self { canceled: Arc::new(AtomicBool::new(false)) } + } + + pub fn cancel(&self) { + self.canceled.store(true, Ordering::Relaxed); + } + + pub fn is_canceled(&self) -> bool { + self.canceled.load(Ordering::Relaxed) + } +} + +#[derive(Clone, Debug)] +pub struct AsyncWorkJobContext { + pub job_id: AsyncWorkJobId, + pub kind: AsyncWorkJobKind, + pub priority: AsyncWorkPriority, + cancel_token: AsyncWorkCancelToken, +} + +impl AsyncWorkJobContext { + pub fn is_canceled(&self) -> bool { + self.cancel_token.is_canceled() + } +} + +type AsyncWorkFn = Box AsyncWorkJobOutcome + Send + 'static>; + +struct PendingAsyncWorkJob { + id: AsyncWorkJobId, + kind: AsyncWorkJobKind, + priority: AsyncWorkPriority, + cancel_token: AsyncWorkCancelToken, + work: AsyncWorkFn, +} + +struct RunningAsyncWorkJob { + id: AsyncWorkJobId, + kind: AsyncWorkJobKind, + priority: AsyncWorkPriority, + cancel_token: AsyncWorkCancelToken, + work: AsyncWorkFn, +} + +#[derive(Default)] +struct AsyncWorkLaneState { + next_job_id: u64, + pending: VecDeque, + active: Option, + active_cancel_token: Option, + shutdown_requested: bool, + telemetry: AsyncWorkLaneTelemetry, +} + +#[derive(Default)] +pub struct AsyncWorkLane { + state: Mutex, + ready: Condvar, +} + +impl AsyncWorkLane { + pub fn submit( + &self, + kind: AsyncWorkJobKind, + work: impl FnOnce(AsyncWorkJobContext) -> AsyncWorkJobOutcome + Send + 'static, + ) -> Result { + self.submit_with_priority(kind, kind.default_priority(), work) + } + + pub fn submit_with_priority( + &self, + kind: AsyncWorkJobKind, + priority: AsyncWorkPriority, + work: impl FnOnce(AsyncWorkJobContext) -> AsyncWorkJobOutcome + Send + 'static, + ) -> Result { + let mut state = self.state.lock().map_err(|_| AsyncWorkLaneError::InternalFailure)?; + if state.shutdown_requested { + return Err(AsyncWorkLaneError::Shutdown); + } + + state.next_job_id += 1; + let id = AsyncWorkJobId::new(state.next_job_id); + let cancel_token = AsyncWorkCancelToken::new(); + let job = PendingAsyncWorkJob { id, kind, priority, cancel_token, work: Box::new(work) }; + let insert_at = state + .pending + .iter() + .position(|pending| priority.rank() < pending.priority.rank()) + .unwrap_or(state.pending.len()); + state.pending.insert(insert_at, job); + let pending_depth = state.pending.len(); + state.telemetry.record_submitted(id, pending_depth); + self.ready.notify_one(); + Ok(id) + } + + pub fn submit_memcard_commit_write( + &self, + work: impl FnOnce(AsyncWorkJobContext) -> AsyncWorkJobOutcome + Send + 'static, + ) -> Result { + self.submit_with_priority( + AsyncWorkJobKind::Memcard, + AsyncWorkPriority::MemcardCommitWrite, + work, + ) + } + + pub fn submit_fs_write_config( + &self, + work: impl FnOnce(AsyncWorkJobContext) -> AsyncWorkJobOutcome + Send + 'static, + ) -> Result { + self.submit_with_priority(AsyncWorkJobKind::Fs, AsyncWorkPriority::FsWriteConfig, work) + } + + pub fn submit_asset_load( + &self, + work: impl FnOnce(AsyncWorkJobContext) -> AsyncWorkJobOutcome + Send + 'static, + ) -> Result { + self.submit_with_priority(AsyncWorkJobKind::Asset, AsyncWorkPriority::AssetLoad, work) + } + + pub fn submit_non_critical_read_list( + &self, + kind: AsyncWorkJobKind, + work: impl FnOnce(AsyncWorkJobContext) -> AsyncWorkJobOutcome + Send + 'static, + ) -> Result { + self.submit_with_priority(kind, AsyncWorkPriority::NonCriticalReadList, work) + } + + pub fn request_cancel(&self, job_id: AsyncWorkJobId) -> bool { + let state = self.state.lock().unwrap(); + if let Some(token) = state + .active + .filter(|active| active.id == job_id) + .and(state.active_cancel_token.as_ref()) + { + token.cancel(); + return true; + } + + for pending in &state.pending { + if pending.id == job_id { + pending.cancel_token.cancel(); + return true; + } + } + + false + } + + pub fn active_job(&self) -> Option { + self.state.lock().unwrap().active + } + + pub fn pending_depth(&self) -> usize { + self.state.lock().unwrap().pending.len() + } + + pub fn telemetry(&self) -> AsyncWorkLaneTelemetry { + self.state.lock().unwrap().telemetry + } + + pub fn request_shutdown(&self) { + let mut state = self.state.lock().unwrap(); + state.shutdown_requested = true; + if let Some(token) = &state.active_cancel_token { + token.cancel(); + } + self.ready.notify_all(); + } + + fn wait_take(&self) -> Option { + self.wait_take_with_hook(|| {}) + } + + fn wait_take_with_hook(&self, mut before_wait: impl FnMut()) -> Option { + let mut state = self.state.lock().unwrap(); + loop { + if state.shutdown_requested { + let discarded = state.pending.len(); + state.pending.clear(); + state.telemetry.record_shutdown_discards(discarded); + return None; + } + + if let Some(pending) = state.pending.pop_front() { + let active = AsyncWorkActiveJob { + id: pending.id, + kind: pending.kind, + priority: pending.priority, + }; + state.active = Some(active); + state.active_cancel_token = Some(pending.cancel_token.clone()); + let pending_depth = state.pending.len(); + state.telemetry.record_started(pending.id, pending_depth); + return Some(RunningAsyncWorkJob { + id: pending.id, + kind: pending.kind, + priority: pending.priority, + cancel_token: pending.cancel_token, + work: pending.work, + }); + } + + before_wait(); + state = self.ready.wait(state).unwrap(); + } + } + + fn record_finished(&self, job_id: AsyncWorkJobId, outcome: AsyncWorkJobOutcome) { + let mut state = self.state.lock().unwrap(); + if state.active.is_some_and(|active| active.id == job_id) { + state.active = None; + state.active_cancel_token = None; + } + state.telemetry.record_closed(job_id, outcome); + } +} + +pub struct AsyncWorkLaneController { + config: AsyncWorkLaneConfig, + lane: Arc, + handle: Option>, + done_rx: Receiver>, +} + +impl AsyncWorkLaneController { + pub fn start(config: AsyncWorkLaneConfig, lane: Arc) -> Self { + let worker_lane = Arc::clone(&lane); + let (done_tx, done_rx) = mpsc::channel(); + let handle = thread::spawn(move || { + let result = catch_unwind(AssertUnwindSafe(|| run_async_work_loop(worker_lane))) + .map_err(|_| AsyncWorkLaneError::WorkerPanic); + let _ = done_tx.send(result); + }); + + Self { config, lane, handle: Some(handle), done_rx } + } + + pub fn lane(&self) -> Arc { + Arc::clone(&self.lane) + } + + pub fn stop(&mut self) -> Result<(), AsyncWorkLaneError> { + self.lane.request_shutdown(); + match self.done_rx.recv_timeout(self.config.shutdown_timeout) { + Ok(result) => { + if let Some(handle) = self.handle.take() + && handle.join().is_err() + { + return Err(AsyncWorkLaneError::WorkerPanic); + } + result + } + Err(RecvTimeoutError::Timeout) => Err(AsyncWorkLaneError::ShutdownTimeout), + Err(RecvTimeoutError::Disconnected) => Err(AsyncWorkLaneError::InternalFailure), + } + } +} + +impl Drop for AsyncWorkLaneController { + fn drop(&mut self) { + if self.handle.is_some() { + let _ = self.stop(); + } + } +} + +fn run_async_work_loop(lane: Arc) { + while let Some(job) = lane.wait_take() { + let context = AsyncWorkJobContext { + job_id: job.id, + kind: job.kind, + priority: job.priority, + cancel_token: job.cancel_token, + }; + let result = catch_unwind(AssertUnwindSafe(|| (job.work)(context))) + .unwrap_or(AsyncWorkJobOutcome::Failed); + lane.record_finished(job.id, result); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::mpsc; + + #[test] + fn async_work_lane_runs_jobs_in_submission_order() { + let lane = Arc::new(AsyncWorkLane::default()); + let mut controller = + AsyncWorkLaneController::start(AsyncWorkLaneConfig::default(), Arc::clone(&lane)); + let (tx, rx) = mpsc::channel(); + + lane.submit(AsyncWorkJobKind::Test, { + let tx = tx.clone(); + move |_| { + tx.send(1).unwrap(); + AsyncWorkJobOutcome::Completed + } + }) + .unwrap(); + lane.submit(AsyncWorkJobKind::Test, { + let tx = tx.clone(); + move |_| { + tx.send(2).unwrap(); + AsyncWorkJobOutcome::Completed + } + }) + .unwrap(); + + assert_eq!(rx.recv().unwrap(), 1); + assert_eq!(rx.recv().unwrap(), 2); + controller.stop().unwrap(); + + let telemetry = lane.telemetry(); + assert_eq!(telemetry.submitted_jobs, 2); + assert_eq!(telemetry.started_jobs, 2); + assert_eq!(telemetry.completed_jobs, 2); + } + + #[test] + fn async_work_lane_runs_higher_priority_io_before_assets() { + let lane = Arc::new(AsyncWorkLane::default()); + let (tx, rx) = mpsc::channel(); + + lane.submit_asset_load({ + let tx = tx.clone(); + move |ctx| { + tx.send((ctx.kind, ctx.priority)).unwrap(); + AsyncWorkJobOutcome::Completed + } + }) + .unwrap(); + lane.submit_non_critical_read_list(AsyncWorkJobKind::Fs, { + let tx = tx.clone(); + move |ctx| { + tx.send((ctx.kind, ctx.priority)).unwrap(); + AsyncWorkJobOutcome::Completed + } + }) + .unwrap(); + lane.submit_memcard_commit_write({ + let tx = tx.clone(); + move |ctx| { + tx.send((ctx.kind, ctx.priority)).unwrap(); + AsyncWorkJobOutcome::Completed + } + }) + .unwrap(); + lane.submit_fs_write_config({ + let tx = tx.clone(); + move |ctx| { + tx.send((ctx.kind, ctx.priority)).unwrap(); + AsyncWorkJobOutcome::Completed + } + }) + .unwrap(); + + let mut controller = + AsyncWorkLaneController::start(AsyncWorkLaneConfig::default(), Arc::clone(&lane)); + + assert_eq!( + rx.recv().unwrap(), + (AsyncWorkJobKind::Memcard, AsyncWorkPriority::MemcardCommitWrite) + ); + assert_eq!(rx.recv().unwrap(), (AsyncWorkJobKind::Fs, AsyncWorkPriority::FsWriteConfig)); + assert_eq!(rx.recv().unwrap(), (AsyncWorkJobKind::Asset, AsyncWorkPriority::AssetLoad)); + assert_eq!( + rx.recv().unwrap(), + (AsyncWorkJobKind::Fs, AsyncWorkPriority::NonCriticalReadList) + ); + controller.stop().unwrap(); + } + + #[test] + fn async_work_lane_has_one_active_job() { + let lane = Arc::new(AsyncWorkLane::default()); + let mut controller = + AsyncWorkLaneController::start(AsyncWorkLaneConfig::default(), Arc::clone(&lane)); + let (first_started_tx, first_started_rx) = mpsc::channel(); + let (release_first_tx, release_first_rx) = mpsc::channel(); + let (second_started_tx, second_started_rx) = mpsc::channel(); + + let first = lane + .submit(AsyncWorkJobKind::Test, move |_| { + first_started_tx.send(()).unwrap(); + release_first_rx.recv().unwrap(); + AsyncWorkJobOutcome::Completed + }) + .unwrap(); + let second = lane + .submit(AsyncWorkJobKind::Test, move |_| { + second_started_tx.send(()).unwrap(); + AsyncWorkJobOutcome::Completed + }) + .unwrap(); + + first_started_rx.recv().unwrap(); + assert_eq!(lane.active_job().unwrap().id, first); + assert_eq!(lane.pending_depth(), 1); + assert!(second_started_rx.try_recv().is_err()); + + release_first_tx.send(()).unwrap(); + second_started_rx.recv().unwrap(); + assert_eq!(lane.telemetry().last_started_job_id, Some(second)); + controller.stop().unwrap(); + } + + #[test] + fn async_work_lane_supports_cooperative_cancellation() { + let lane = Arc::new(AsyncWorkLane::default()); + let mut controller = + AsyncWorkLaneController::start(AsyncWorkLaneConfig::default(), Arc::clone(&lane)); + let (started_tx, started_rx) = mpsc::channel(); + + let job_id = lane + .submit(AsyncWorkJobKind::Test, move |ctx| { + started_tx.send(()).unwrap(); + while !ctx.is_canceled() { + std::thread::yield_now(); + } + AsyncWorkJobOutcome::Canceled + }) + .unwrap(); + + started_rx.recv().unwrap(); + assert!(lane.request_cancel(job_id)); + controller.stop().unwrap(); + assert_eq!(lane.telemetry().canceled_jobs, 1); + } + + #[test] + fn async_work_lane_rejects_submit_after_shutdown() { + let lane = Arc::new(AsyncWorkLane::default()); + lane.request_shutdown(); + + let result = lane.submit(AsyncWorkJobKind::Test, |_| AsyncWorkJobOutcome::Completed); + assert_eq!(result, Err(AsyncWorkLaneError::Shutdown)); + } + + #[test] + fn async_work_lane_shutdown_discards_pending_jobs() { + let lane = Arc::new(AsyncWorkLane::default()); + lane.submit(AsyncWorkJobKind::Test, |_| AsyncWorkJobOutcome::Completed).unwrap(); + assert_eq!(lane.pending_depth(), 1); + + lane.request_shutdown(); + assert!(lane.wait_take().is_none()); + assert_eq!(lane.pending_depth(), 0); + assert_eq!(lane.telemetry().shutdown_discards, 1); + } +} diff --git a/crates/console/prometeu-system/src/services/fs/mod.rs b/crates/console/prometeu-system/src/services/fs/mod.rs index dcb16257..bed0b115 100644 --- a/crates/console/prometeu-system/src/services/fs/mod.rs +++ b/crates/console/prometeu-system/src/services/fs/mod.rs @@ -8,4 +8,4 @@ pub use fs_backend::FsBackend; pub use fs_entry::FsEntry; pub use fs_error::FsError; pub use fs_state::FsState; -pub use virtual_fs::VirtualFS; +pub use virtual_fs::{FsAsyncLaneOperation, VirtualFS}; diff --git a/crates/console/prometeu-system/src/services/fs/virtual_fs.rs b/crates/console/prometeu-system/src/services/fs/virtual_fs.rs index dc09d3f8..fdc6b6f6 100644 --- a/crates/console/prometeu-system/src/services/fs/virtual_fs.rs +++ b/crates/console/prometeu-system/src/services/fs/virtual_fs.rs @@ -1,4 +1,5 @@ use crate::fs::{FsBackend, FsEntry, FsError}; +use crate::services::async_work::{AsyncWorkJobKind, AsyncWorkPriority}; /// Virtual Filesystem (VFS) interface for Prometeu. /// @@ -13,6 +14,29 @@ pub struct VirtualFS { backend: Option>, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum FsAsyncLaneOperation { + Write, + Delete, + Config, + Read, + ListDir, + Exists, +} + +impl FsAsyncLaneOperation { + pub const fn job_kind(self) -> AsyncWorkJobKind { + AsyncWorkJobKind::Fs + } + + pub const fn priority(self) -> AsyncWorkPriority { + match self { + Self::Write | Self::Delete | Self::Config => AsyncWorkPriority::FsWriteConfig, + Self::Read | Self::ListDir | Self::Exists => AsyncWorkPriority::NonCriticalReadList, + } + } +} + impl Default for VirtualFS { fn default() -> Self { Self::new() @@ -242,4 +266,18 @@ mod tests { assert!(matches!(vfs.delete("/"), Err(FsError::PermissionDenied))); assert_eq!(calls.delete.load(Ordering::Relaxed), 0); } + + #[test] + fn fs_async_lane_operations_classify_priority_without_public_api_changes() { + assert_eq!(FsAsyncLaneOperation::Write.job_kind(), AsyncWorkJobKind::Fs); + assert_eq!(FsAsyncLaneOperation::Write.priority(), AsyncWorkPriority::FsWriteConfig); + assert_eq!(FsAsyncLaneOperation::Delete.priority(), AsyncWorkPriority::FsWriteConfig); + assert_eq!(FsAsyncLaneOperation::Config.priority(), AsyncWorkPriority::FsWriteConfig); + assert_eq!(FsAsyncLaneOperation::Read.priority(), AsyncWorkPriority::NonCriticalReadList); + assert_eq!( + FsAsyncLaneOperation::ListDir.priority(), + AsyncWorkPriority::NonCriticalReadList + ); + assert_eq!(FsAsyncLaneOperation::Exists.priority(), AsyncWorkPriority::NonCriticalReadList); + } } diff --git a/crates/console/prometeu-system/src/services/memcard.rs b/crates/console/prometeu-system/src/services/memcard.rs index a53b5436..5ade70eb 100644 --- a/crates/console/prometeu-system/src/services/memcard.rs +++ b/crates/console/prometeu-system/src/services/memcard.rs @@ -1,4 +1,5 @@ use crate::fs::{FsError, VirtualFS}; +use crate::services::async_work::{AsyncWorkJobKind, AsyncWorkPriority}; use std::collections::HashMap; pub const MEMCARD_SLOT_COUNT: usize = 32; @@ -50,6 +51,30 @@ pub struct MemcardWriteResult { pub bytes_written: u32, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum MemcardAsyncLaneOperation { + SlotWrite, + SlotCommit, + SlotClear, + SlotRead, + SlotStat, +} + +impl MemcardAsyncLaneOperation { + pub const fn job_kind(self) -> AsyncWorkJobKind { + AsyncWorkJobKind::Memcard + } + + pub const fn priority(self) -> AsyncWorkPriority { + match self { + Self::SlotWrite | Self::SlotCommit | Self::SlotClear => { + AsyncWorkPriority::MemcardCommitWrite + } + Self::SlotRead | Self::SlotStat => AsyncWorkPriority::NonCriticalReadList, + } + } +} + #[derive(Debug, Clone)] struct SlotImage { payload: Vec, @@ -428,4 +453,29 @@ mod tests { let stat = mem.slot_stat(&fs, 7, 2); assert_eq!(stat.state, MemcardSlotState::Corrupt); } + + #[test] + fn memcard_async_lane_operations_classify_priority_without_changing_public_api() { + assert_eq!(MemcardAsyncLaneOperation::SlotWrite.job_kind(), AsyncWorkJobKind::Memcard); + assert_eq!( + MemcardAsyncLaneOperation::SlotWrite.priority(), + AsyncWorkPriority::MemcardCommitWrite + ); + assert_eq!( + MemcardAsyncLaneOperation::SlotCommit.priority(), + AsyncWorkPriority::MemcardCommitWrite + ); + assert_eq!( + MemcardAsyncLaneOperation::SlotClear.priority(), + AsyncWorkPriority::MemcardCommitWrite + ); + assert_eq!( + MemcardAsyncLaneOperation::SlotRead.priority(), + AsyncWorkPriority::NonCriticalReadList + ); + assert_eq!( + MemcardAsyncLaneOperation::SlotStat.priority(), + AsyncWorkPriority::NonCriticalReadList + ); + } } diff --git a/crates/console/prometeu-system/src/services/mod.rs b/crates/console/prometeu-system/src/services/mod.rs index 6dbdfab5..efe827c6 100644 --- a/crates/console/prometeu-system/src/services/mod.rs +++ b/crates/console/prometeu-system/src/services/mod.rs @@ -1,3 +1,4 @@ +pub mod async_work; pub mod fs; pub mod memcard; pub mod process; diff --git a/crates/console/prometeu-system/src/services/vm_runtime/dispatch.rs b/crates/console/prometeu-system/src/services/vm_runtime/dispatch.rs index 932b8589..b650f02f 100644 --- a/crates/console/prometeu-system/src/services/vm_runtime/dispatch.rs +++ b/crates/console/prometeu-system/src/services/vm_runtime/dispatch.rs @@ -757,6 +757,64 @@ impl NativeInterface for VmRuntimeHost<'_> { ret.push_int(status as i64); Ok(()) } + Syscall::AssetBacklogInfo => { + let info = platform.assets().backlog_info(); + ret.push_int(info.status as i64); + ret.push_int(info.pending_count as i64); + ret.push_int(info.active_handle as i64); + ret.push_int(info.active_asset_id.unwrap_or(0) as i64); + ret.push_int(info.active_bank_type.map(|bank| bank as i64).unwrap_or(0)); + ret.push_int(info.active_slot.map(|slot| slot as i64).unwrap_or(0)); + ret.push_int(info.active_progress as i64); + Ok(()) + } + Syscall::AssetBacklogPosition => { + let position = platform.assets().backlog_position(expect_int(args, 0)? as u32); + ret.push_int(position.status as i64); + ret.push_int(position.state as i64); + ret.push_int(position.position as i64); + ret.push_int(position.progress as i64); + Ok(()) + } + Syscall::AssetBacklogMove => { + let new_position = expect_non_negative_usize(args, 1, "new_position")?; + let status = + platform.assets().backlog_move(expect_int(args, 0)? as u32, new_position); + ret.push_int(status as i64); + Ok(()) + } + Syscall::AssetBacklogPromote => { + let status = platform.assets().backlog_promote(expect_int(args, 0)? as u32); + ret.push_int(status as i64); + Ok(()) + } + Syscall::AssetBacklogDemote => { + let status = platform.assets().backlog_demote(expect_int(args, 0)? as u32); + ret.push_int(status as i64); + Ok(()) + } + Syscall::AssetTargetStatus => { + let bank_type = match expect_int(args, 0)? { + 0 => BankType::GLYPH, + 1 => BankType::SOUNDS, + 2 => BankType::SCENE, + other => { + return Err(VmFault::Trap( + TRAP_TYPE, + format!("Invalid asset bank type: {}", other), + )); + } + }; + let slot = expect_non_negative_usize(args, 1, "slot")?; + let target = platform.assets().target_status(bank_type, slot); + ret.push_int(target.status as i64); + ret.push_int(target.asset_id.unwrap_or(0) as i64); + ret.push_int(target.handle as i64); + ret.push_int(target.state as i64); + ret.push_int(target.position as i64); + ret.push_int(target.progress as i64); + Ok(()) + } Syscall::BankInfo => { let asset_type = match expect_int(args, 0)? as u32 { 0 => BankType::GLYPH, diff --git a/crates/console/prometeu-system/src/services/vm_runtime/tests_asset_bank.rs b/crates/console/prometeu-system/src/services/vm_runtime/tests_asset_bank.rs index 45ae5a2e..6fd57e59 100644 --- a/crates/console/prometeu-system/src/services/vm_runtime/tests_asset_bank.rs +++ b/crates/console/prometeu-system/src/services/vm_runtime/tests_asset_bank.rs @@ -180,6 +180,102 @@ fn tick_asset_status_unknown_handle_returns_status_not_crash() { assert_eq!(vm.operand_stack_top(1), vec![Value::Int64(LoadStatus::UnknownHandle as i64)]); } +#[test] +fn tick_asset_target_status_empty_slot_returns_status_first_payload() { + let mut runtime = VirtualMachineRuntime::new(None); + let mut log_service = LogService::new(4096); + let mut fs = VirtualFS::new(); + let mut fs_state = FsState::Unmounted; + let mut memcard = MemcardService::new(); + let mut open_files: HashMap = HashMap::new(); + let mut next_handle = 1; + let mut vm = VirtualMachine::default(); + let mut platform = TestPlatform::new(); + let signals = InputSignals::default(); + let code = assemble("PUSH_I32 0\nPUSH_I32 0\nHOSTCALL 0\nHALT").expect("assemble"); + let program = serialized_single_function_module( + code, + vec![SyscallDecl { + module: "asset".into(), + name: "target_status".into(), + version: 1, + arg_slots: 2, + ret_slots: 6, + }], + ); + let cartridge = cartridge_with_program(program, caps::ASSET); + + runtime.initialize_vm(&mut log_service, &mut vm, &cartridge).expect("runtime must initialize"); + let report = runtime.tick( + &mut log_service, + &mut fs, + &mut fs_state, + &mut memcard, + &mut open_files, + &mut next_handle, + &mut vm, + &signals, + &mut platform, + ); + + assert!(report.is_none(), "target_status must not crash for empty slots"); + assert!(vm.is_halted()); + assert_eq!( + vm.operand_stack_top(6), + vec![ + Value::Int64(0), + Value::Int64(0), + Value::Int64(LoadStatus::EMPTY as i64), + Value::Int64(1), + Value::Int64(0), + Value::Int64(AssetOpStatus::Ok as i64), + ] + ); +} + +#[test] +fn tick_asset_backlog_promote_unknown_handle_returns_status_not_crash() { + let mut runtime = VirtualMachineRuntime::new(None); + let mut log_service = LogService::new(4096); + let mut fs = VirtualFS::new(); + let mut fs_state = FsState::Unmounted; + let mut memcard = MemcardService::new(); + let mut open_files: HashMap = HashMap::new(); + let mut next_handle = 1; + let mut vm = VirtualMachine::default(); + let mut platform = TestPlatform::new(); + let signals = InputSignals::default(); + let code = assemble("PUSH_I32 999\nHOSTCALL 0\nHALT").expect("assemble"); + let program = serialized_single_function_module( + code, + vec![SyscallDecl { + module: "asset".into(), + name: "backlog_promote".into(), + version: 1, + arg_slots: 1, + ret_slots: 1, + }], + ); + let cartridge = cartridge_with_program(program, caps::ASSET); + + runtime.initialize_vm(&mut log_service, &mut vm, &cartridge).expect("runtime must initialize"); + let report = runtime.tick( + &mut log_service, + &mut fs, + &mut fs_state, + &mut memcard, + &mut open_files, + &mut next_handle, + &mut vm, + &signals, + &mut platform, + ); + + assert!(report.is_none(), "unknown backlog handle must not crash"); + assert!(vm.is_halted()); + assert_eq!(vm.operand_stack_top(1), vec![Value::Int64(AssetOpStatus::UnknownHandle as i64)]); +} + #[test] fn tick_bank_info_returns_slot_summary_not_json() { let mut runtime = VirtualMachineRuntime::new(None); diff --git a/discussion/index.ndjson b/discussion/index.ndjson index f2a5c205..d40b78be 100644 --- a/discussion/index.ndjson +++ b/discussion/index.ndjson @@ -1,4 +1,4 @@ -{"type":"meta","next_id":{"DSC":43,"AGD":44,"DEC":34,"PLN":123,"LSN":50,"CLSN":1}} +{"type":"meta","next_id":{"DSC":43,"AGD":44,"DEC":35,"PLN":129,"LSN":51,"CLSN":1}} {"type":"discussion","id":"DSC-0039","status":"abandoned","ticket":"render-pipeline-family-and-future-3d","title":"Render Pipeline Family and Future 3D","created_at":"2026-06-04","updated_at":"2026-06-04","tags":["gfx","renderer","runtime","architecture","pipeline"],"agendas":[{"id":"AGD-0039","file":"AGD-0039-render-pipeline-family-and-future-3d.md","status":"abandoned","created_at":"2026-06-04","updated_at":"2026-06-04","_override_reason":"User explicitly chose to close this agenda without a new decision because DSC-0038 already established enough architecture for future extension, and 3D is intentionally deferred."}],"decisions":[],"plans":[],"lessons":[],"_override_reason":"User explicitly chose to close this agenda without a new decision because DSC-0038 already established enough architecture for future extension, and 3D is intentionally deferred."} {"type":"discussion","id":"DSC-0040","status":"done","ticket":"vm-render-parallel-execution-boundary","title":"VM and Render Parallel Execution Boundary","created_at":"2026-06-04","updated_at":"2026-06-06","tags":["runtime","renderer","vm","concurrency","architecture","perf"],"agendas":[],"decisions":[],"plans":[],"lessons":[{"id":"LSN-0048","file":"discussion/lessons/DSC-0040-vm-render-parallel-execution-boundary/LSN-0048-render-workers-need-a-closed-packet-contract.md","status":"done","created_at":"2026-06-06","updated_at":"2026-06-06"}]} {"type":"discussion","id":"DSC-0041","status":"open","ticket":"foreground-stack-game-pause-shell-vm-backed","title":"Foreground Stack, Game Pause, and VM-Backed Shell Coexistence","created_at":"2026-06-05","updated_at":"2026-06-05","tags":["runtime","os","lifecycle","shell","game","vm","foreground","architecture"],"agendas":[{"id":"AGD-0041","file":"AGD-0041-foreground-stack-game-pause-shell-vm-backed.md","status":"open","created_at":"2026-06-05","updated_at":"2026-06-05"}],"decisions":[],"plans":[],"lessons":[]} @@ -18,9 +18,9 @@ {"type":"discussion","id":"DSC-0006","status":"open","ticket":"vm-owned-random-service","title":"Agenda - VM-Owned Random Service","created_at":"2026-03-27","updated_at":"2026-03-27","tags":[],"agendas":[{"id":"AGD-0005","file":"AGD-0005-vm-owned-random-service.md","status":"open","created_at":"2026-03-27","updated_at":"2026-03-27"}],"decisions":[],"plans":[],"lessons":[]} {"type":"discussion","id":"DSC-0007","status":"open","ticket":"app-home-filesystem-surface-and-semantics","title":"Agenda - App Home Filesystem Surface and Semantics","created_at":"2026-03-27","updated_at":"2026-03-27","tags":[],"agendas":[{"id":"AGD-0006","file":"AGD-0006-app-home-filesystem-surface-and-semantics.md","status":"open","created_at":"2026-03-27","updated_at":"2026-03-27"}],"decisions":[],"plans":[],"lessons":[]} {"type":"discussion","id":"DSC-0008","status":"done","ticket":"perf-runtime-telemetry-hot-path","title":"Agenda - [PERF] Runtime Telemetry Hot Path","created_at":"2026-03-27","updated_at":"2026-06-04","tags":[],"agendas":[],"decisions":[],"plans":[],"lessons":[{"id":"LSN-0026","file":"discussion/lessons/DSC-0008-perf-runtime-telemetry-hot-path/LSN-0026-push-based-telemetry-model.md","status":"done","created_at":"2026-04-10","updated_at":"2026-04-10"}]} -{"type":"discussion","id":"DSC-0009","status":"open","ticket":"perf-async-background-work-lanes-for-assets-and-fs","title":"Agenda - [PERF] Async Background Work Lanes for Assets and FS","created_at":"2026-03-27","updated_at":"2026-03-27","tags":[],"agendas":[{"id":"AGD-0008","file":"AGD-0008-perf-async-background-work-lanes-for-assets-and-fs.md","status":"open","created_at":"2026-03-27","updated_at":"2026-03-27"}],"decisions":[],"plans":[],"lessons":[]} +{"type":"discussion","id":"DSC-0009","status":"done","ticket":"perf-async-background-work-lanes-for-assets-and-fs","title":"Agenda - [PERF] Async Background Work Lanes for Assets and FS","created_at":"2026-03-27","updated_at":"2026-07-01","tags":["perf","asset","fs","async","scheduler","runtime"],"agendas":[],"decisions":[],"plans":[],"lessons":[{"id":"LSN-0050","file":"discussion/lessons/DSC-0009-perf-async-background-work-lanes-for-assets-and-fs/LSN-0050-serial-async-lanes-bound-backlog-complexity.md","status":"done","created_at":"2026-07-01","updated_at":"2026-07-01"}]} {"type":"discussion","id":"DSC-0010","status":"done","ticket":"perf-host-desktop-frame-pacing-and-presentation","title":"Agenda - [PERF] Host Desktop Frame Pacing and Presentation","created_at":"2026-03-27","updated_at":"2026-04-20","tags":[],"agendas":[],"decisions":[],"plans":[],"lessons":[{"id":"LSN-0036","file":"discussion/lessons/DSC-0010-perf-host-desktop-frame-pacing-and-presentation/LSN-0036-frame-publication-and-host-invalidation-must-be-separate.md","status":"done","created_at":"2026-04-20","updated_at":"2026-04-20"}]} -{"type":"discussion","id":"DSC-0011","status":"open","ticket":"perf-gfx-render-pipeline-and-dirty-regions","title":"Agenda - [PERF] GFX Render Pipeline and Dirty Regions","created_at":"2026-03-27","updated_at":"2026-03-27","tags":[],"agendas":[{"id":"AGD-0010","file":"AGD-0010-perf-gfx-render-pipeline-and-dirty-regions.md","status":"open","created_at":"2026-03-27","updated_at":"2026-03-27"}],"decisions":[],"plans":[],"lessons":[]} +{"type":"discussion","id":"DSC-0011","status":"abandoned","ticket":"perf-gfx-render-pipeline-and-dirty-regions","title":"Agenda - [PERF] GFX Render Pipeline and Dirty Regions","created_at":"2026-03-27","updated_at":"2026-06-28","tags":[],"agendas":[{"id":"AGD-0010","file":"AGD-0010-perf-gfx-render-pipeline-and-dirty-regions.md","status":"abandoned","created_at":"2026-03-27","updated_at":"2026-06-28","_override_reason":"User explicitly chose to remove this agenda because it no longer applies to the current render scenario; future optimizations can be discussed when needed."}],"decisions":[],"plans":[],"lessons":[],"_override_reason":"User explicitly chose to abandon this discussion because it no longer applies to the current render scenario; future optimizations can be discussed when needed."} {"type":"discussion","id":"DSC-0012","status":"done","ticket":"perf-runtime-introspection-syscalls","title":"Agenda - [PERF] Runtime Introspection Syscalls","created_at":"2026-03-27","updated_at":"2026-04-19","tags":["perf","runtime","syscall","telemetry","debug","asset"],"agendas":[],"decisions":[],"plans":[],"lessons":[{"id":"LSN-0034","file":"discussion/lessons/DSC-0012-perf-runtime-introspection-syscalls/LSN-0034-host-owned-debug-boundaries.md","status":"done","created_at":"2026-04-19","updated_at":"2026-04-19"}]} {"type":"discussion","id":"DSC-0013","status":"done","ticket":"perf-host-debug-overlay-isolation","title":"Agenda - [PERF] Host Debug Overlay Isolation","created_at":"2026-03-27","updated_at":"2026-04-10","tags":[],"agendas":[],"decisions":[],"plans":[],"lessons":[{"id":"LSN-0027","file":"discussion/lessons/DSC-0013-perf-host-debug-overlay-isolation/LSN-0027-host-debug-overlay-isolation.md","status":"done","created_at":"2026-04-10","updated_at":"2026-04-10"}]} {"type":"discussion","id":"DSC-0024","status":"done","ticket":"generic-memory-bank-slot-contract","title":"Agenda - Generic Memory Bank Slot Contract","created_at":"2026-04-10","updated_at":"2026-04-10","tags":["runtime","asset","memory-bank","slots","host"],"agendas":[],"decisions":[],"plans":[],"lessons":[{"id":"LSN-0029","file":"discussion/lessons/DSC-0024-generic-memory-bank-slot-contract/LSN-0029-slot-first-bank-telemetry-belongs-in-asset-manager.md","status":"done","created_at":"2026-04-10","updated_at":"2026-04-10"}]} diff --git a/discussion/lessons/DSC-0009-perf-async-background-work-lanes-for-assets-and-fs/LSN-0050-serial-async-lanes-bound-backlog-complexity.md b/discussion/lessons/DSC-0009-perf-async-background-work-lanes-for-assets-and-fs/LSN-0050-serial-async-lanes-bound-backlog-complexity.md new file mode 100644 index 00000000..71df825b --- /dev/null +++ b/discussion/lessons/DSC-0009-perf-async-background-work-lanes-for-assets-and-fs/LSN-0050-serial-async-lanes-bound-backlog-complexity.md @@ -0,0 +1,100 @@ +--- +id: LSN-0050 +ticket: perf-async-background-work-lanes-for-assets-and-fs +title: Serial Async Lanes Bound Backlog Complexity +created: 2026-07-01 +tags: [runtime, asset, async, scheduler, telemetry] +--- + +## Context + +The async asset work moved from request-owned worker creation toward a +runtime-owned serial lane. The important shift is not just "use a worker +thread"; it is the introduction of a third logical execution lane with explicit +ownership, priority, cancellation, progress, and telemetry. + +This work followed `DEC-0034` and plans `PLN-0123` through `PLN-0128`. The +published model separates: + +- main runtime execution, where guest-visible state is committed; +- render worker execution, where closed render packets are consumed; +- async IO/decode/persistence work, where one background job is active at a + time. + +## Key Decisions + +### Async Work Lane and Asset Backlog Contract + +**What:** Asset IO/decode and compatible persistence work use a runtime-owned +serial async lane. Asset requests are queued by target bank slot, not by +unbounded transient request identity. + +**Why:** A serial lane keeps the runtime observable and portable. It preserves +the intended hardware mental model while avoiding a desktop-biased thread pool +or one OS thread per asset request. + +**Trade-offs:** The lane intentionally sacrifices parallel decode throughput in +exchange for bounded state, deterministic ownership, simpler telemetry, and +clear priority rules. If future implementations add more physical parallelism, +they still need to preserve the logical serial contract where the spec requires +it. + +## Patterns and Algorithms + +The asset backlog is bounded by target identity. A request targets a concrete +bank type and slot. A newer request for the same target supersedes the older +pending request, and an older active result is discarded when its generation is +no longer current. + +Stable handles observe slots. A handle should not be treated as a worker job or +thread token. The durable identity is the bank slot plus request generation: +slot state says what is resident; request state says what is queued, active, +ready, canceled, superseded, or failed. + +Commit remains a main-lane operation. The background lane may read and decode, +but publication into resident runtime state happens at predictable ownership +points. This keeps VM execution, render handoff, and frame observation from +racing with background mutation. + +Progress and telemetry are closure-oriented. Use integer progress and update +expensive aggregate telemetry when jobs close, not inside decode loops. Tests +should synchronize on explicit state transitions rather than sleeps. + +Priority is part of the lane contract. Memcard commit/write work can outrank +ordinary asset loads, FS write/config work can be represented internally, and +non-critical read/list work belongs below asset loads. Public FS semantics +remain owned by the filesystem discussion; sharing the lane is not permission +to decide the FS API here. + +## Pitfalls + +Do not hide concurrency behind per-request `thread::spawn`. That appears simple +locally but loses boundedness, priority, cancellation, and telemetry. + +Do not let a handle mean "the current background job." Handles must survive +empty slots, completed work, cancellation, superseding, errors, and already +resident fast paths. + +Do not install background results directly from the worker. Decode completion is +not the same as publication. Use request generation checks before committing a +result. + +Do not turn operational states into traps. Queued, active, canceled, +superseded, error, and backend-unavailable outcomes belong in status-first +surfaces unless the caller violated the structural ABI. + +Do not reopen FS public API scope while wiring internal async-lane consumers. +The lane can support FS-style work without deciding request/poll semantics. + +## Takeaways + +- A runtime async lane is an ownership boundary, not just an implementation + thread. +- Backlog keying by target slot bounds complexity better than queue length + limits exposed to the guest. +- Stable slot handles plus request generations prevent stale async results from + mutating newer state. +- Main-lane commit keeps background IO/decode compatible with deterministic + runtime publication. +- Telemetry belongs at state transitions and job closure, not inside hot decode + loops. diff --git a/discussion/workflow/agendas/AGD-0006-app-home-filesystem-surface-and-semantics.md b/discussion/workflow/agendas/AGD-0006-app-home-filesystem-surface-and-semantics.md index 7ff84a23..588a197c 100644 --- a/discussion/workflow/agendas/AGD-0006-app-home-filesystem-surface-and-semantics.md +++ b/discussion/workflow/agendas/AGD-0006-app-home-filesystem-surface-and-semantics.md @@ -22,6 +22,19 @@ Sem um contrato claro de `home` por app, a API tende a crescer com semantica inc 1. Todo `app` acessa somente sua `home` logica. 2. Nunca ha acesso direto ao filesystem global do host pela userland. 3. O runtime `fs` interno continua cobrindo tanto `game` quanto `app`. +4. Existe uma async IO lane compartilhavel por assets, memcard e FS. Esta + agenda deve decidir a API publica de FS considerando essa lane, mas a + existencia da lane ja esta fechada por `DEC-0034`. + +## Fronteira com a Async IO Lane + +`DEC-0034` fecha a existencia de uma lane serial para trabalho async de IO. FS +pode consumir essa lane internamente para escrita, configuracao, leitura ou +listagem, conforme prioridade operacional definida pelo runtime. + +Esta agenda continua dona da decisao sobre API publica de FS para `app home`. +Portanto, a existencia da lane nao implica criar agora request handles, polling +publico ou novas syscalls de FS. Esses shapes permanecem em aberto aqui. ## Alvo da Discussao @@ -116,6 +129,8 @@ No perfil `app` (`home` sandbox), esta agenda passa a ser a fonte normativa para 2. `rename` entra no MVP ou pode ficar para fase seguinte? 3. Qual conjunto minimo de metadados garante portabilidade real entre hosts? 4. Qual grau de atomicidade e obrigatorio para escrita de arquivo no v1? +5. Quais operacoes de FS devem consumir a async IO lane e quais permanecem + sincrono-aparentes para a userland? ## Dependencias diff --git a/discussion/workflow/agendas/AGD-0008-perf-async-background-work-lanes-for-assets-and-fs.md b/discussion/workflow/agendas/AGD-0008-perf-async-background-work-lanes-for-assets-and-fs.md deleted file mode 100644 index 28eee281..00000000 --- a/discussion/workflow/agendas/AGD-0008-perf-async-background-work-lanes-for-assets-and-fs.md +++ /dev/null @@ -1,85 +0,0 @@ ---- -id: AGD-0008 -ticket: perf-async-background-work-lanes-for-assets-and-fs -title: Agenda - [PERF] Async Background Work Lanes for Assets and FS -status: open -created: 2026-03-27 -resolved: -decision: -tags: [] ---- - -# Agenda - [PERF] Async Background Work Lanes for Assets and FS - -## Problema - -`asset.load()` hoje cria uma thread do SO por requisicao. Ao mesmo tempo, `fs` ainda nao tem uma politica clara para sync assincrono barato em hardware simples. - -O projeto precisa de paralelismo para IO/decode/sync, mas o target inclui handheld DIY e hardware barato, onde um pool grande ou explosao de threads pode piorar a latencia em vez de melhorar. - -## Dor - -- `thread::spawn` por request escala mal e cria jitter. -- assets e `fs` competem por IO sem uma politica unica de fila/prioridade. -- sem lane dedicada, operacoes de background tendem a vazar custo para o main loop. -- sem disciplina, o host desktop vira referencia errada para hardware fraco. - -## Hotspots Atuais - -- [asset.rs](/Users/niltonconstantino/personal/workspace.personal/intrepid/prometeu/runtime/crates/console/prometeu-drivers/src/asset.rs#L353) -- [tick.rs](/Users/niltonconstantino/personal/workspace.personal/intrepid/prometeu/runtime/crates/console/prometeu-system/src/virtual_machine_runtime/tick.rs#L53) -- [runner.rs](/Users/niltonconstantino/personal/workspace.personal/intrepid/prometeu/runtime/crates/host/prometeu-host-desktop-winit/src/runner.rs#L315) - -## Alvo da Discussao - -Fechar um modelo de execucao assincrona para `asset` e `fs` que seja previsivel em hardware simples. - -## O Que Precisa Ser Definido - -1. Topologia de workers. - Escolher entre: - - uma thread dedicada para `asset` e outra para `fs`; - - um worker unico multiplexando filas; - - um pool minimo e fixo; - - proibicao explicita de `spawn` por request. - -2. Separacao por dominio. - Decidir se `asset` e `fs` compartilham scheduler/fila ou se cada dominio tem lane propria. - -3. Politica de prioridade. - Definir: - - prioridade de loads visuais vs audio; - - prioridade de sync de save/config; - - limite de trabalho por frame para install/commit. - -4. Modelo de retorno. - Fechar como o guest observa backlog, cancelamento e saturacao: - - status imediato de fila cheia; - - status de pending; - - metrica de backlog; - - politica de cancelamento. - -5. Orçamento para hardware barato. - Definir quantas threads o runtime pode assumir como baseline. - -## Open Questions de Arquitetura - -1. O v1 precisa de lanes separadas para `asset` e `fs` ou basta uma fila central com classes de prioridade? -2. Decode de asset fica no mesmo worker do IO ou em fase distinta? -3. Commit/install no device continua no main thread por determinismo? - -## Dependencias - -- `../specs/09-events-and-concurrency.md` -- `../specs/15-asset-management.md` -- `../specs/16a-syscall-policies.md` -- `014-app-home-filesystem-surface-and-semantics.md` - -## Criterio de Saida Desta Agenda - -Pode virar PR quando houver decisao escrita sobre: - -- numero e tipo de workers aceitos no baseline; -- fila/prioridade de `asset` e `fs`; -- proibicao ou aceitacao limitada de `thread::spawn` por request; -- modelo de status/telemetria para backlog e saturacao. diff --git a/discussion/workflow/agendas/AGD-0010-perf-gfx-render-pipeline-and-dirty-regions.md b/discussion/workflow/agendas/AGD-0010-perf-gfx-render-pipeline-and-dirty-regions.md deleted file mode 100644 index 42307ee1..00000000 --- a/discussion/workflow/agendas/AGD-0010-perf-gfx-render-pipeline-and-dirty-regions.md +++ /dev/null @@ -1,409 +0,0 @@ ---- -id: AGD-0010 -ticket: perf-gfx-render-pipeline-and-dirty-regions -title: Agenda - [PERF] GFX Render Pipeline and Dirty Regions -status: open -created: 2026-03-27 -resolved: -decision: -tags: [] ---- - -# Agenda - [PERF] GFX Render Pipeline and Dirty Regions - -## Problema - -O renderer `gfx` recompõe a cena inteira a cada frame logico, mesmo quando a mudanca visual e pequena. - -Hoje `render_all()` reconstrói buckets de sprites, escaneia os 512 sprites, redesenha layers e aplica dois passes fullscreen de fade sem politica de invalidacao. - -Ao mesmo tempo, a arquitetura atual ja opera como um framebuffer destrutivo em memoria: draws escrevem diretamente no buffer de trabalho e operacoes posteriores sobrescrevem o que estiver por baixo. A discussao nao e migrar para um modelo tipo GPU, scene graph ou retained rendering. - -Existe ainda um driver de produto importante: o objetivo continua sendo viabilizar hardware handheld proprio com limitacoes reais de orcamento e com economia de memoria o mais agressiva possivel. Isso significa que ganhos de CPU nao podem ser avaliados isoladamente; custo de RAM adicional, buffers extras e estruturas auxiliares entram diretamente no criterio de aceitacao. - -## Dor - -- custo visual basico cresce demais para hardware simples. -- pequenos updates pagam preco de full redraw. -- fade, HUD e world composition ficam sempre no caminho critico. - -## Hotspots Atuais - -- [gfx.rs](/Users/niltonconstantino/personal/workspace.personal/intrepid/prometeu/runtime/crates/console/prometeu-drivers/src/gfx.rs#L563) -- [gfx.rs](/Users/niltonconstantino/personal/workspace.personal/intrepid/prometeu/runtime/crates/console/prometeu-drivers/src/gfx.rs#L671) - -## Alvo da Discussao - -Definir ate onde o v1 pode sofisticar o modelo de desenho destrutivo em memoria para reduzir custo e manter previsibilidade, sem migrar para um renderer de estilo GPU. - -## O Que Precisa Ser Definido - -1. Perfil real de custo. - Medir separadamente: - - rebuild de buckets; - - rasterizacao de layers; - - rasterizacao de sprites; - - fades fullscreen; - - conversao/copia no host; - - apresentacao no host. - -2. Granularidade minima de invalidacao compativel com framebuffer destrutivo. - Escolher entre: - - dirty flag por frame inteiro; - - dirty flags por subsistema (`world`, `hud`, `fade`, `sprites`); - - dirty regions restritas e derivadas de primitivas/sprites; - - combinacao minima viavel. - -3. Buckets e traversal de sprites. - Decidir se os buckets continuam sendo reconstruidos por frame ou se viram estrutura incremental. - -4. Fades. - Definir se fade neutro precisa bypass total e se fade parcial pode operar por regiao/camada. - -5. Cache de composicao. - Delimitar se layers estaticas/HUD podem manter cache intermediario. - -6. Primitivas e contrato operacional. - Definir quais draws devem continuar sendo vistos como escrita destrutiva direta no framebuffer e onde valem aceleracoes internas sem mudar a semantica observavel. - -7. Meta de custo. - Fechar qual teto de pixels/trabalho por frame e aceitavel no baseline. - -8. Orcamento de memoria. - Definir quais otimizacoes aceitam memoria adicional e quais precisam caber no modelo mais economico possivel para handheld barato. - -## Open Questions de Arquitetura - -1. Quais partes do custo atual estao no `render_all()` e quais partes estao na conversao/apresentacao do host? -2. Qual o maior ganho de curto prazo dentro do modelo atual: buckets, fades, HUD, primitivas ou host copy/convert? -3. Dirty regions restritas quebram alguma expectativa de determinismo visivel ou sao apenas um detalhe interno de implementacao? -4. O HUD deve continuar no mesmo pipeline da world scene? -5. Quais primitivas merecem tratamento especial para reduzir overdraw sem mudar a semantica de sobrescrita? -6. Qual o teto aceitavel de memoria extra para caches, buffers separados ou estruturas auxiliares no handheld alvo? - -## Problemas de Medio e Grande Porte Identificados - -### 1. Composicao completa do framebuffer a cada frame logico - -`render_all()` recompõe o `back` inteiro por software, mesmo quando a mudanca visual e pequena. - -Impacto: - -- custo cresce com a quantidade de pixels tocados, nao apenas com a quantidade de estado alterado; -- layers, sprites e fades disputam o mesmo budget de CPU; -- o modelo atual favorece previsibilidade sem ainda ter atalhos internos suficientes. - -### 2. Rasterizacao pixel a pixel de tilemaps - -Cada layer visivel percorre tiles e, para cada tile nao vazio, resolve indices e escreve pixel a pixel no framebuffer. - -Impacto: - -- o custo real esta mais na escrita de pixels do que na manutencao do `TileMap`; -- scroll e barato como estado, mas caro na hora de compor a imagem se tudo for repintado; -- layers estaticas continuam pagando custo de rasterizacao em toda recomposicao. - -### 3. Rasterizacao pixel a pixel de sprites e escolha por bucket full scan - -Os 512 sprites sao varridos para reconstruir buckets e os sprites ativos sao desenhados pixel a pixel. - -Impacto: - -- quando poucos sprites mudam, ainda existe custo fixo de rebuild dos buckets; -- overdraw entre sprites e layers pode explodir o numero de writes no `back`; -- a escolha de sprites ativos/prioridade ainda e simples demais para cenarios com alta ocupacao. - -### 4. Fades fullscreen com custo proporcional ao frame inteiro - -`scene fade` e `hud fade` percorrem o framebuffer inteiro quando ativos. - -Impacto: - -- custo grande e previsivel, mas potencialmente desproporcional; -- fade neutro ja pode ser bypass, mas fade parcial ainda custa uma passada completa; -- dois passes fullscreen no mesmo frame podem virar gargalo antes de layers ou sprites. - -### 5. HUD no mesmo caminho critico da composicao principal - -O HUD e redesenhado como tilemap depois da world scene e antes do `hud fade`. - -Impacto: - -- HUD estatico continua gerando trabalho por frame; -- mistura responsabilidades de world/UI no mesmo budget de composicao; -- dificulta isolar ganho de cache ou dirtying so para interface. - -### 6. Conversao obrigatoria de RGB565 para RGBA8 no host - -No `RedrawRequested`, o host percorre todo o `front_buffer` e converte para o frame do `pixels`. - -Impacto: - -- ha custo de leitura do framebuffer inteiro e escrita de um segundo buffer inteiro; -- esse custo existe mesmo quando o host esta apenas apresentando o mesmo quadro logico; -- pode competir com o custo do renderer sem estar visivel na discussao do `gfx`. - -### 7. Pipeline serializado no host desktop - -`tick`, `render_all()`, conversao de formato e apresentacao ocorrem no mesmo fluxo operacional. - -Impacto: - -- o frame time final acumula custo de runtime, composicao, copy/convert e present; -- falta desacoplamento para esconder latencia entre producao e apresentacao; -- dificulta atribuir gargalo sem instrumentacao por etapa. - -## Estudo Inicial de Possiveis Optimizacoes - -### A. Dirty flags por subsistema em vez de dirty rect generico - -Ideia: - -- flags independentes para `world`, `sprites`, `hud`, `scene_fade`, `hud_fade`, `host_present`. - -Vantagens: - -- preserva a semantica de framebuffer destrutivo; -- reduz recomposicao desnecessaria quando so um subsistema mudou; -- e muito mais simples de validar do que dirty regions arbitrarias. - -Riscos: - -- exige definir dependencias claras entre world, sprites, HUD e fades; -- pode induzir falsos positivos conservadores, o que e aceitavel no v1. - -### B. Rebuild incremental ou condicional dos buckets de sprite - -Ideia: - -- so reconstruir buckets quando algum `GfxSetSprite` mudar atividade, prioridade, banco ou tile; -- opcionalmente manter contadores/sinais de mutacao da OAM. - -Vantagens: - -- elimina custo fixo por frame quando OAM permanece estavel; -- combina com a ideia de sprites como estado, nao como draw list efemera. - -Riscos: - -- precisa invalidacao correta para evitar bucket desatualizado; -- ganho pode ser pequeno se o custo dominante estiver no draw dos sprites, nao no rebuild. - -### C. Separar world e HUD em buffers logicos distintos - -Ideia: - -- manter um buffer da world scene e um buffer do HUD, compondo no final apenas quando necessario. - -Vantagens: - -- HUD estatico deixa de participar da recomposicao da world; -- permite cache e fade especificos para cada dominio; -- se alinha bem com a separacao conceitual entre cena e interface. - -Riscos: - -- aumenta uso de memoria e pontos de sincronizacao; -- precisa manter semantica clara de sobrescrita entre mundo e HUD. -- pode ser inviavel no perfil de handheld barato se exigir buffers adicionais permanentes. - -### D. Cache de layer estatica ou composicao parcial da world - -Ideia: - -- layers que nao mudam podem manter imagem intermediaria pronta; -- scroll, tile updates ou troca de bank invalidam o cache correspondente. - -Vantagens: - -- reduz rasterizacao repetida de cenario estavel; -- aproxima o ganho de hardware tile-based sem abandonar software raster. - -Riscos: - -- cache amplo demais vira complexidade estrutural; -- se scroll muda constantemente, o ganho cai bastante. -- pode consumir memoria demais para um hardware com orcamento agressivo. - -### E. Otimizacao de fades - -Ideia: - -- bypass total para fade neutro; -- opcionalmente aplicar fade apenas sobre buffers relevantes; -- estudar LUTs ou blend mais barato por pixel. - -Vantagens: - -- ataca um custo fullscreen claramente delimitado; -- baixo risco conceitual. - -Riscos: - -- fade parcial por regiao pode complicar demais o contrato; -- LUT ajuda aritmetica, mas nao elimina custo de varrer memoria. - -### F. Melhorias em primitivas e spans de memoria - -Ideia: - -- acelerar `fill_rect`, linhas horizontais/verticais, clears e possiveis blits contiguos; -- explorar caminhos de row spans contiguos em vez de loops mais genericos. - -Vantagens: - -- melhora direta do modelo de desenho destrutivo; -- baixo risco arquitetural; -- cria fundacao para outros atalhos internos. - -Riscos: - -- ganho localizado se o workload dominante vier de tile/sprite compositing; -- precisa medir por primitive class. - -### G. Reduzir custo de copy/convert no host - -Ideia: - -- medir separadamente `draw_rgb565_to_rgba8` e `pixels.render()`; -- estudar formato mais proximo do host ou conversao mais barata; -- evitar redraw host quando nao houver novo front relevante. - -Vantagens: - -- ataca custo fora do `gfx` que ainda entra no frame time total; -- pode render ganho imediato no desktop host. - -Riscos: - -- parte do custo depende da stack `pixels/wgpu`, nao apenas do runtime; -- alguns ganhos podem ser especificos do host desktop e nao do contrato do console. - -## Restricao de Plataforma - -Qualquer recomendacao desta agenda deve ser filtrada por um criterio adicional: - -- priorizar solucoes que melhorem custo sem multiplicar buffers; -- tratar memoria extra como recurso escasso de primeira classe; -- preferir flags, metadados pequenos e estruturas incrementais a caches grandes; -- evitar solucoes cuja performance dependa de assumir um host desktop mais forte do que o handheld alvo. - -Direcao atual da discussao: - -- nesta etapa, buffers extras devem ficar fora; -- o foco deve ser otimizar ao maximo o pipeline existente; -- cache intermediario ou novos buffers so entram em estudo depois que as otimizacoes de baixo custo de memoria forem medidas e esgotadas. - -## Sugestao / Recomendacao Atual - -Priorizar o estudo em camadas, nesta ordem: - -1. instrumentar o pipeline por etapa; -2. validar dirty flags por subsistema como mecanismo minimo de invalidacao; -3. testar rebuild condicional de buckets e bypass/isolamento de fades; -4. otimizar primitivas e caminhos contiguos de escrita no framebuffer atual; -5. estudar separacao world/HUD e cache de layer apenas se os dados justificarem e o teto de memoria permitir; -5. tratar copy/convert do host como frente paralela de otimizacao, nao como substituto da analise do `gfx`. - -## Achados Consolidados Ate Aqui - -1. O contrato base continua sendo framebuffer destrutivo em memoria. - Draws escrevem diretamente no buffer de trabalho e operacoes posteriores sobrescrevem o que estiver por baixo. A agenda nao aponta para migracao a um renderer tipo GPU, retained mode ou compositor sofisticado. - -2. O `present()` atual nao parece ser o problema principal. - No `gfx`, `present()` faz swap de buffers, nao copia completa de pixels. - -3. O custo suspeito esta na recomposicao e na apresentacao, nao na manutencao do estado logico. - Escritas de `TileMap`, scroll e `GfxSetSprite` sao pequenas e pontuais; o custo potencialmente dominante esta em rasterizar pixels, aplicar fades fullscreen e converter o framebuffer para o host. - -4. Os maiores suspeitos atuais de custo sao: - - rasterizacao de layers; - - rasterizacao de sprites; - - fades fullscreen; - - conversao `RGB565 -> RGBA8` no host; - - `pixels.render()` / apresentacao no host. - -5. O pipeline atual deve ser otimizado antes de considerar buffers extras. - A direcao aceita da discussao e esgotar primeiro flags pequenas, bypasses, melhorias de primitive paths e ajustes incrementais no pipeline existente. - -6. Restricao de plataforma pesa tanto quanto CPU. - Como o alvo e um handheld proprio com limitacoes de orcamento e memoria, solucoes que consumam RAM adicional significativa devem ficar fora desta fase. - -7. A instrumentacao minima desejada para retomar o estudo ficou definida. - Quando houver workload representativo, queremos medir ao menos: - - `render_all_total_us` - - `bucket_rebuild_us` - - `layer_raster_us` - - `sprite_raster_us` - - `scene_fade_us` - - `hud_raster_us` - - `hud_fade_us` - - `host_convert_us` - - `host_present_us` - -8. A retencao de metricas deve ser barata e orientada a analise posterior. - A preferencia atual e por acumuladores e ring buffer pequeno de snapshots, sem logging pesado por frame no hot path. - -9. Ring buffer no `TileMap` nao e prioridade nesta fase. - O custo suspeito nao esta na manutencao da grade logica do mapa, e sim na composicao da janela visivel no framebuffer. Mudar a estrutura do mapa so faria sentido se o gargalo principal estivesse em streaming/atualizacao estrutural do cenario, o que nao e a hipotese atual. - -10. A direcao de otimizacao mais promissora e padronizar copias massivas para o framebuffer atual. - Em vez de atacar apenas tilemaps com loops pixel a pixel, a discussao passa a favorecer um padrao geral de blit/copia massiva com: - - calculo previo de offsets e spans; - - fast paths para casos contiguos; - - trabalho por linha/chunk em vez de trabalho atomizado por pixel quando possivel; - - reaproveitamento dessa infraestrutura para tilemaps, sprites e outras operacoes de desenho destrutivo. - -11. Migrar o `back` para `RGBA8888` nao e direcao aceita neste momento. - Isso aumentaria custo de memoria e largura de banda no alvo handheld, alem de deslocar a otimizacao para o host desktop. O contrato interno continua preferencialmente em `RGB565`, mesmo considerando fades e uma futura pipeline de lights. - -## Nova Direcao Tecnica Em Estudo - -Quando esta agenda for reaberta, a linha principal de investigacao deve considerar: - -- manter o `TileMap` simples como estado logico; -- assumir que a janela visivel provavelmente continuara sendo recomposta na maior parte dos frames; -- concentrar a otimizacao em como essa recomposicao escreve no `back`; -- desenhar uma infraestrutura compartilhada de copias massivas/blits para o framebuffer atual; -- usar essa infraestrutura nao apenas para tilemaps, mas como base comum para raster de sprites e outros caminhos de desenho. - -Pergunta orientadora da reabertura: - -- como transformar o renderer de um pipeline de writes atomizados por pixel em um pipeline com fast paths de spans/chunks, sem perder a semantica de framebuffer destrutivo e sem pagar memoria extra relevante? - -## Status de Standby - -Esta agenda deve ficar em espera ate existir um game ou workload real que exercite de forma representativa: - -- tilemaps com scroll e composicao de world; -- sprites em quantidade suficiente para testar buckets e overdraw; -- HUD ativo; -- fades; -- apresentacao completa no host. - -Antes disso, qualquer conclusao sobre gargalo ou prioridade de otimizacao tende a ser prematura. - -## Proximo Gatilho Para Reabrir - -Reabrir esta agenda quando houver: - -- um game jogavel ou cena de teste que percorra o pipeline completo; -- dados de workload mais proximos do uso real; -- necessidade concreta de justificar uma rodada de instrumentacao e profiling. - -## Dependencias - -- `../specs/04-gfx-peripheral.md` -- `../specs/11-portability-and-cross-platform-execution.md` - -## Criterio de Saida Desta Agenda - -Pode virar PR quando houver decisao escrita sobre: - -- filosofia explicita de framebuffer destrutivo como contrato base; -- plano de instrumentacao para localizar o custo dominante do pipeline; -- nivel minimo de invalidacao no v1; -- politica de rebuild de buckets de sprites; -- bypass/cache de fade e HUD; -- politica para otimizar primitivas sem mudar a semantica observavel; -- meta de custo para o render pipeline. diff --git a/docs/specs/runtime/09-events-and-concurrency.md b/docs/specs/runtime/09-events-and-concurrency.md index 9337ad90..497c25c0 100644 --- a/docs/specs/runtime/09-events-and-concurrency.md +++ b/docs/specs/runtime/09-events-and-concurrency.md @@ -127,7 +127,31 @@ Shutdown is explicit and bounded. A shutdown request wakes a waiting worker, causes pending work that will not be consumed to be discarded, and reports a typed failure if the worker cannot join within the configured timeout. -## 8 Determinism and Best Practices +## 8 Async Asset and IO Work Lane + +The asset/IO async work lane is not a machine-visible event source and does not +introduce guest callbacks. It is an implementation-side lane for asset +IO/decode/materialization and compatible persistence IO work. + +The lane is serial: + +- it has at most one active job; +- it keeps an ordered backlog; +- it is separate from the render worker; +- it must not create one OS thread per guest-visible asset request. + +Asset jobs are keyed by target `bank_type/slot`. A newer request for the same +target supersedes the older request. Superseding is an operational status and +does not execute guest code. + +Asset install/commit into resident banks happens on the main runtime lane at +predictable ownership points. The async lane prepares materialized results; it +does not publish resident graphics/audio/scene state directly. + +FS and game persistence services may consume this lane for IO-style work, but +public FS API shape is defined by the FS/app-home contract, not by this chapter. + +## 9 Determinism and Best Practices PROMETEU encourages: @@ -142,7 +166,7 @@ PROMETEU discourages: - hidden timing channels; - ambiguous out-of-band execution. -## 9 Relationship to Other Specs +## 10 Relationship to Other Specs - [`09a-coroutines-and-cooperative-scheduling.md`](09a-coroutines-and-cooperative-scheduling.md) defines coroutine lifecycle and scheduling behavior. - [`10-debug-inspection-and-profiling.md`](10-debug-inspection-and-profiling.md) defines observability and diagnostics surfaces. diff --git a/docs/specs/runtime/15-asset-management.md b/docs/specs/runtime/15-asset-management.md index d2e52306..699c499d 100644 --- a/docs/specs/runtime/15-asset-management.md +++ b/docs/specs/runtime/15-asset-management.md @@ -393,14 +393,129 @@ Fault boundary: Rules: -- `handle` is valid only when `load` status is `OK`; +- `handle` is returned when `load` status is `OK`; +- `handle` represents a stable bank slot target, not a worker thread; +- a known handle remains queryable even when the slot is empty, has no active + request, or has a superseded request; - failed `load` returns `handle = 0`; - `commit` and `cancel` must not be silent no-op for unknown/invalid handle state. - `asset.load` resolves the target bank type from `asset_table` using `asset_id`; - public callers must not supply `asset_name` or `bank_type` to `asset.load`; - slot validation and residency/lifecycle rejection remain in `asset` status space and are not delegated to `bank`. -### 11.2 Minimum status tables +### 11.2 Async work lane and backlog + +Asset loading uses the runtime async work lane. This lane is separate from the +VM/main lane and separate from the render worker lane. + +The asset async lane is serial: + +- it executes at most one active asset job at a time; +- it keeps an ordered backlog of pending requests; +- it must not create one OS thread per `asset.load` request. + +Asset requests are keyed by the target `bank_type/slot`. + +Rules: + +- each `bank_type/slot` has at most one current request; +- a newer request for the same `bank_type/slot` supersedes the previous request; +- if the previous request is queued, it is removed from the backlog; +- if the previous request is active, the lane should cancel cooperatively when + the current phase supports cheap cancellation; +- if active work cannot stop cheaply, it may finish, but the result must be + discarded when its generation no longer matches the target request generation; +- if the target already contains the requested `asset_id` as a valid resident + asset, `asset.load` returns a ready handle without adding a backlog entry. + +The effective backlog size is bounded by the sum of targetable bank slots, +because only one current request can exist per `bank_type/slot`. The runtime +does not expose a guest-visible `queue_full` status for the normal asset +backlog path. + +### 11.3 Handle state + +An asset handle observes one stable bank slot target. Its observable state has +two parts: + +```text +handle: + bank_type + slot + +slot_state: + loaded_asset_id + resident_state + slot_generation + +request_state: + requested_asset_id + request_generation + state + backlog_position + progress +``` + +`slot_state` describes what is currently resident in the target slot. +`request_state` describes the current or most recent request for that target. + +Mutating operations such as commit, cancel, promote, demote, and move must act +on the current request generation. They must not accidentally mutate a newer +request through an older handle view. + +### 11.4 Backlog inspection and ordering + +The asset backlog surface may expose these status-first operations: + +- `asset.backlog_info() -> (status, pending_count, active_handle, active_asset_id, active_bank_type, active_slot, active_progress)` +- `asset.backlog_position(handle) -> (status, state, position, progress)` +- `asset.backlog_move(handle, new_position) -> status` +- `asset.backlog_promote(handle) -> status` +- `asset.backlog_demote(handle) -> status` +- `asset.target_status(bank_type, slot) -> (status, asset_id, handle, state, position, progress)` + +`asset.backlog_promote(handle)` is a shortcut for moving a queued request to +position `1`, the first pending position after the active job. + +`asset.backlog_demote(handle)` is a shortcut for moving a queued request to the +end of the pending backlog. + +### 11.5 Progress and telemetry + +Asset progress uses integer progress, not floating point. The preferred scale is +`0..10000`. + +The initial phase model is: + +```text +queued -> 0 +read -> 0..4000 +decode -> 4000..9000 +stage -> 9000..10000 +ready -> 10000 +``` + +If a phase cannot report internal progress, it keeps the previous progress mark +and advances at phase completion. The runtime must not invent false precision +for non-linear decode phases. + +Minimum telemetry: + +- current backlog depth; +- target/request position; +- active job progress; +- jobs submitted; +- jobs completed; +- jobs failed; +- jobs canceled; +- jobs superseded; +- job duration; +- percentiles by `bank_type`; +- lightweight percentiles or small-window samples by `asset_id`. + +Percentiles are updated when a job closes, not inside the inner decode loop. + +### 11.6 Minimum status tables `asset.load` request statuses: @@ -418,9 +533,16 @@ Rules: - `4` = `CANCELED` - `5` = `ERROR` - `6` = `UNKNOWN_HANDLE` +- `7` = `QUEUED` +- `8` = `ACTIVE` +- `9` = `SUPERSEDED` +- `10` = `EMPTY` +- `11` = `INVALID` +- `12` = `BACKEND_UNAVAILABLE` `asset.commit` and `asset.cancel` operation statuses: - `0` = `OK` - `1` = `UNKNOWN_HANDLE` - `2` = `INVALID_STATE` +- `3` = `SUPERSEDED` diff --git a/docs/specs/runtime/16-host-abi-and-syscalls.md b/docs/specs/runtime/16-host-abi-and-syscalls.md index 34ac736f..5944b4c4 100644 --- a/docs/specs/runtime/16-host-abi-and-syscalls.md +++ b/docs/specs/runtime/16-host-abi-and-syscalls.md @@ -203,6 +203,12 @@ Canonical operations in v1 are: - `asset.status(handle) -> status` - `asset.commit(handle) -> status` - `asset.cancel(handle) -> status` +- `asset.backlog_info() -> (status, pending_count, active_handle, active_asset_id, active_bank_type, active_slot, active_progress)` +- `asset.backlog_position(handle) -> (status, state, position, progress)` +- `asset.backlog_move(handle, new_position) -> status` +- `asset.backlog_promote(handle) -> status` +- `asset.backlog_demote(handle) -> status` +- `asset.target_status(bank_type, slot) -> (status, asset_id, handle, state, position, progress)` For `asset.load`: @@ -210,6 +216,18 @@ For `asset.load`: - `slot` is the target slot index; - bank kind is resolved from `asset_table` by `asset_id`, not supplied by the caller. +Asset handles represent stable bank slot targets. A handle can be queried even +when its slot has no resident asset or active request. Internally the handle +state separates resident `slot_state` from current `request_state`. + +The asset backlog is keyed by `bank_type/slot`. New requests for the same target +supersede older requests for that target. `superseded` is an operational status, +not a structural trap. + +`asset.backlog_promote(handle)` and `asset.backlog_demote(handle)` are +convenience operations over backlog movement. They do not introduce a second +ordering model. + ### Bank diagnostics surface (`bank`, v1) `DEC-0009` narrows the public bank contract: diff --git a/docs/specs/runtime/16a-syscall-policies.md b/docs/specs/runtime/16a-syscall-policies.md index 1efb10ef..71e58c45 100644 --- a/docs/specs/runtime/16a-syscall-policies.md +++ b/docs/specs/runtime/16a-syscall-policies.md @@ -44,6 +44,9 @@ Normal operational success and operational failure conditions should be represen Examples: - asset not yet loaded; +- asset request queued or active; +- asset request superseded by a newer request for the same bank slot; +- asset backend unavailable; - audio voice unavailable; - persistent storage full. @@ -107,6 +110,11 @@ Game memcard operations (`mem.*`) are status-first and use `fs` capability in v1 `mem` remains layered on runtime `fs`; no parallel persistence channel is introduced. Domain surface, status catalog and slot semantics are defined in [`08-save-memory-and-memcard.md`](08-save-memory-and-memcard.md). +Asset backlog operations are status-first. `queued`, `active`, `ready`, +`canceled`, `superseded`, `empty`, `invalid`, decode failure, and backend +unavailability are operational results. They must not be reclassified as `Trap` +unless the caller violates the structural ABI contract. + ## 3 Interaction with the Garbage Collector The VM heap and host-managed memory are separate.