From 0e2bf7e2aac3d8e19842672265946f9d84cbdf22 Mon Sep 17 00:00:00 2001 From: Michael Schimmel Date: Mon, 16 Feb 2026 15:12:29 +0100 Subject: [PATCH] Spawn read_from_zip on blocking thread Adds `Send + 'static` bounds to `R` in `read_from_zip` and moves the file operations to a Tokio blocking task. This prevents blocking the async runtime when reading large zip archives. Also, updates the `load_symbol_async` function to handle data loading concurrently and stream chunks of data back to the UI. The `loaded_data` field is changed from `Option>>>` to `Vec>`. The sender and receiver for data updates are also adjusted to handle streaming chunks. --- src/main.rs | 134 +++++++++++++++++++++++++++++++--------------------- 1 file changed, 81 insertions(+), 53 deletions(-) diff --git a/src/main.rs b/src/main.rs index 534ab01..f0f62e3 100644 --- a/src/main.rs +++ b/src/main.rs @@ -124,19 +124,23 @@ impl DataServer { Ok(shared_data) } - async fn read_from_zip(&self, path: &Path) -> anyhow::Result> { - 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 records: &[R] = cast_slice(&bin_content); - return Ok(records.to_vec()); + async fn read_from_zip(&self, path: &Path) -> anyhow::Result> { + let path = path.to_path_buf(); + 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 records: &[R] = cast_slice(&bin_content); + return Ok(records.to_vec()); + } } - } - Err(anyhow::anyhow!("No .bin file found")) + Err(anyhow::anyhow!("No .bin file found")) + }) + .await? } } @@ -147,9 +151,9 @@ struct DataViewerApp { rt: Runtime, symbols: Vec, selected_symbol: Option, - loaded_data: Option>>>, - tx: std::sync::mpsc::Sender>>>, - rx: std::sync::mpsc::Receiver>>>, + loaded_data: Vec>, + tx: std::sync::mpsc::Sender>>>>, + rx: std::sync::mpsc::Receiver>>>>, is_loading: bool, status_msg: String, } @@ -173,7 +177,7 @@ impl DataViewerApp { rt, symbols, selected_symbol: None, - loaded_data: None, + loaded_data: Vec::new(), tx, rx, is_loading: false, @@ -184,27 +188,44 @@ impl DataViewerApp { fn load_symbol_async(&mut self, symbol: String) { self.is_loading = true; self.status_msg = format!("Loading {}...", symbol); - self.loaded_data = None; + self.loaded_data.clear(); let server = Arc::clone(&self.server); let tx = self.tx.clone(); - let ctx = self.rt.handle().clone(); self.rt.spawn(async move { - let files = { + let files: Vec = { let symbols = server.symbols.read().unwrap(); symbols.get(&symbol).cloned().unwrap_or_default() }; - let mut all_data = Vec::new(); - for file in files { - if file.symbol.ends_with(".m1") { - if let Ok(data) = server.load_m1_data(&file).await { - all_data.extend_from_slice(&data); - } + let mut next_load: Option>>>>> = None; + + for i in 0..files.len() { + let file = &files[i]; + if !file.symbol.ends_with(".m1") { + continue; + } + + let data_res: anyhow::Result>>> = if let Some(handle) = next_load.take() { + handle.await.map_err(|e| anyhow::anyhow!("Join error: {}", e)).and_then(|res| res) + } else { + server.load_m1_data(file).await + }; + + if i + 1 < files.len() { + let next_file = files[i + 1].clone(); + let server_clone = Arc::clone(&server); + next_load = Some(tokio::spawn(async move { + server_clone.load_m1_data(&next_file).await + })); + } + + if let Ok(data) = data_res { + let _ = tx.send(Some(data)); } } - let _ = tx.send(Arc::new(all_data)); + let _ = tx.send(None); // Finished }); } } @@ -215,14 +236,19 @@ impl eframe::App for DataViewerApp { ctx.request_repaint(); } - // Poll for finished loading tasks - if let Ok(data) = self.rx.try_recv() { - self.loaded_data = Some(data); - self.is_loading = false; - self.status_msg = format!( - "Loaded {} records", - self.loaded_data.as_ref().unwrap().len() - ); + // Poll for finished loading tasks (Feed) + let mut received_any = false; + while let Ok(msg) = self.rx.try_recv() { + if let Some(chunk) = msg { + self.loaded_data.extend_from_slice(&chunk); + received_any = true; + } else { + self.is_loading = false; + } + } + + if received_any { + self.status_msg = format!("Loaded {} records", self.loaded_data.len()); } let mut symbol_to_load = None; @@ -248,27 +274,29 @@ impl eframe::App for DataViewerApp { if let Some(symbol) = &self.selected_symbol { ui.heading(format!("Data: {}", symbol)); if self.is_loading { - ui.add(egui::Spinner::new()); - ui.label("Streaming from network share..."); - } else if let Some(data) = &self.loaded_data { - ui.label(format!("Successfully loaded {} records.", data.len())); - - // Show a small sample - egui::ScrollArea::vertical().show(ui, |ui| { - for i in 0..data.len().min(10000) { - let d = &data[i]; - // Copy fields to local variables to avoid unaligned references - 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 - )); - } + ui.horizontal(|ui| { + ui.add(egui::Spinner::new()); + ui.label("Streaming from network share..."); }); } + + ui.label(format!("Successfully loaded {} records.", self.loaded_data.len())); + + // Show a small sample + egui::ScrollArea::vertical().show(ui, |ui| { + let data = &self.loaded_data; + 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 + )); + } + }); } else { ui.label("Bitte ein Symbol auswählen."); }