From b6a559b8bc234171c1d3603eb0d826f8d205d100 Mon Sep 17 00:00:00 2001 From: Michael Schimmel Date: Mon, 16 Feb 2026 16:18:49 +0100 Subject: [PATCH] Add Sync bound to DataServer Generic The `DataServer` struct's generic type `T` now requires the `Sync` trait bound. This allows `DataServer` instances to be safely shared across multiple threads. Additionally, data loading from zip archives has been moved to a `spawn_blocking` task to avoid blocking the main Tokio runtime thread. This improves responsiveness when dealing with large data files. --- src/main.rs | 121 ++++++++++++++++++++++++++++++++++------------------ 1 file changed, 79 insertions(+), 42 deletions(-) diff --git a/src/main.rs b/src/main.rs index 00fb272..321278b 100644 --- a/src/main.rs +++ b/src/main.rs @@ -27,14 +27,14 @@ struct CacheEntry { last_accessed: Instant, } -pub struct DataServer { +pub struct DataServer { base_path: PathBuf, symbols: RwLock>>, cache: RwLock>>, sys: RwLock, } -impl DataServer { +impl DataServer { pub fn new(path: impl AsRef) -> Self { let mut sys = System::new_all(); sys.refresh_memory(); @@ -110,13 +110,31 @@ impl DataServer { return Ok(Arc::clone(&entry.data)); } } - let raw_data = self.read_from_zip::(&data_file.path).await?; - let mapped_data: Vec = raw_data - .into_iter() - .map(|r| r.to_data_point().into()) - .collect(); - let shared_data = Arc::new(mapped_data); + let path = data_file.path.clone(); + let shared_data = tokio::task::spawn_blocking(move || { + let file = std::fs::File::open(&path)?; + let mut archive = ZipArchive::new(file)?; + for i in 0..archive.len() { + let mut file = archive.by_index(i)?; + if file.name().ends_with(".bin") { + let mut bin_content = vec![0u8; file.size() as usize]; + file.read_exact(&mut bin_content)?; + let raw_data: &[R] = cast_slice(&bin_content); + + // Perform mapping inside blocking task + let mapped_data: Vec = raw_data + .iter() + .map(|&r| r.to_data_point().into()) + .collect(); + + return Ok(Arc::new(mapped_data)); + } + } + Err(anyhow::anyhow!("No .bin file found")) + }) + .await??; + let mut cache = self.cache.write().unwrap(); cache.insert( data_file.path.clone(), @@ -154,8 +172,14 @@ impl DataServer { enum LoadedData { None, - M1(Vec>), - Ticks(Vec>), + M1 { + chunks: Vec>>>, + total_count: usize, + }, + Ticks { + chunks: Vec>>>, + total_count: usize, + }, } enum DataChunk { @@ -214,9 +238,9 @@ impl DataViewerApp { self.is_loading = true; self.status_msg = format!("Loading {}...", symbol); self.loaded_data = if symbol.ends_with(".m1") { - LoadedData::M1(Vec::new()) + LoadedData::M1 { chunks: Vec::new(), total_count: 0 } } else { - LoadedData::Ticks(Vec::new()) + LoadedData::Ticks { chunks: Vec::new(), total_count: 0 } }; let m1_server = Arc::clone(&self.m1_server); @@ -291,13 +315,15 @@ impl eframe::App for DataViewerApp { if let Some(chunk) = msg { match chunk { DataChunk::M1(data) => { - if let LoadedData::M1(vec) = &mut self.loaded_data { - vec.extend_from_slice(&data); + if let LoadedData::M1 { chunks, total_count } = &mut self.loaded_data { + *total_count += data.len(); + chunks.push(data); } } DataChunk::Ticks(data) => { - if let LoadedData::Ticks(vec) = &mut self.loaded_data { - vec.extend_from_slice(&data); + if let LoadedData::Ticks { chunks, total_count } = &mut self.loaded_data { + *total_count += data.len(); + chunks.push(data); } } } @@ -309,8 +335,8 @@ impl eframe::App for DataViewerApp { if received_any { let count = match &self.loaded_data { - LoadedData::M1(v) => v.len(), - LoadedData::Ticks(v) => v.len(), + LoadedData::M1 { total_count, .. } => *total_count, + LoadedData::Ticks { total_count, .. } => *total_count, _ => 0, }; self.status_msg = format!("Loaded {} records", count); @@ -346,33 +372,44 @@ impl eframe::App for DataViewerApp { } match &self.loaded_data { - LoadedData::M1(data) => { - ui.label(format!("Successfully loaded {} OHLC records.", data.len())); - egui::ScrollArea::vertical().show(ui, |ui| { - for i in 0..data.len().min(10000) { - let d = &data[i]; - let open = d.data.open; - let high = d.data.high; - let low = d.data.low; - let close = d.data.close; - ui.label(format!( - "{}: O:{:.5} H:{:.5} L:{:.5} C:{:.5}", - i, open, high, low, close - )); + LoadedData::M1 { chunks, total_count } => { + ui.label(format!("Successfully loaded {} OHLC records.", total_count)); + let row_height = ui.text_style_height(&egui::TextStyle::Body); + egui::ScrollArea::vertical().show_rows(ui, row_height, *total_count, |ui, row_range| { + for i in row_range { + // Find which chunk contains index i + let mut current_idx = i; + for chunk in chunks { + if current_idx < chunk.len() { + let d = &chunk[current_idx]; + let open = d.data.open; + let high = d.data.high; + let low = d.data.low; + let close = d.data.close; + ui.label(format!("{}: O:{:.5} H:{:.5} L:{:.5} C:{:.5}", i, open, high, low, close)); + break; + } + current_idx -= chunk.len(); + } } }); } - LoadedData::Ticks(data) => { - ui.label(format!("Successfully loaded {} Tick records.", data.len())); - egui::ScrollArea::vertical().show(ui, |ui| { - for i in 0..data.len().min(10000) { - let d = &data[i]; - let ask = d.data.ask; - let bid = d.data.bid; - ui.label(format!( - "{}: A:{:.5} B:{:.5}", - i, ask, bid - )); + LoadedData::Ticks { chunks, total_count } => { + ui.label(format!("Successfully loaded {} Tick records.", total_count)); + let row_height = ui.text_style_height(&egui::TextStyle::Body); + egui::ScrollArea::vertical().show_rows(ui, row_height, *total_count, |ui, row_range| { + for i in row_range { + let mut current_idx = i; + for chunk in chunks { + if current_idx < chunk.len() { + let d = &chunk[current_idx]; + let ask = d.data.ask; + let bid = d.data.bid; + ui.label(format!("{}: A:{:.5} B:{:.5}", i, ask, bid)); + break; + } + current_idx -= chunk.len(); + } } }); }