Brummel/Rust-Tests
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Spawn read_from_zip on blocking thread

Browse Source 
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<Arc<Vec<DataPoint<OhlcItem>>>>` to
`Vec<DataPoint<OhlcItem>>`.
The sender and receiver for data updates are also adjusted to handle
streaming
chunks.
This commit is contained in:
Michael Schimmel
2026-02-16 15:12:29 +01:00
parent e59fb661b9
commit 0e2bf7e2aa
1 changed files with 81 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main.rs
+81 -53
View File
@@ -124,19 +124,23 @@ impl DataServer {
Ok(shared_data) Ok(shared_data)
} }
async fn read_from_zip<R: bytemuck::Pod>(&self, path: &Path) -> anyhow::Result<Vec<R>> { async fn read_from_zip<R: bytemuck::Pod + Send + 'static>(&self, path: &Path) -> anyhow::Result<Vec<R>> {
let file = std::fs::File::open(path)?; let path = path.to_path_buf();
let mut archive = ZipArchive::new(file)?; tokio::task::spawn_blocking(move || {
for i in 0..archive.len() { let file = std::fs::File::open(&path)?;
let mut file = archive.by_index(i)?; let mut archive = ZipArchive::new(file)?;
if file.name().ends_with(".bin") { for i in 0..archive.len() {
let mut bin_content = vec![0u8; file.size() as usize]; let mut file = archive.by_index(i)?;
file.read_exact(&mut bin_content)?; if file.name().ends_with(".bin") {
let records: &[R] = cast_slice(&bin_content); let mut bin_content = vec![0u8; file.size() as usize];
return Ok(records.to_vec()); 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, rt: Runtime,
symbols: Vec<String>, symbols: Vec<String>,
selected_symbol: Option<String>, selected_symbol: Option<String>,
loaded_data: Option<Arc<Vec<DataPoint<OhlcItem>>>>, loaded_data: Vec<DataPoint<OhlcItem>>,
tx: std::sync::mpsc::Sender<Arc<Vec<DataPoint<OhlcItem>>>>, tx: std::sync::mpsc::Sender<Option<Arc<Vec<DataPoint<OhlcItem>>>>>,
rx: std::sync::mpsc::Receiver<Arc<Vec<DataPoint<OhlcItem>>>>, rx: std::sync::mpsc::Receiver<Option<Arc<Vec<DataPoint<OhlcItem>>>>>,
is_loading: bool, is_loading: bool,
status_msg: String, status_msg: String,
} }
@@ -173,7 +177,7 @@ impl DataViewerApp {
rt, rt,
symbols, symbols,
selected_symbol: None, selected_symbol: None,
loaded_data: None, loaded_data: Vec::new(),
tx, tx,
rx, rx,
is_loading: false, is_loading: false,
@@ -184,27 +188,44 @@ impl DataViewerApp {
fn load_symbol_async(&mut self, symbol: String) { fn load_symbol_async(&mut self, symbol: String) {
self.is_loading = true; self.is_loading = true;
self.status_msg = format!("Loading {}...", symbol); self.status_msg = format!("Loading {}...", symbol);
self.loaded_data = None; self.loaded_data.clear();
let server = Arc::clone(&self.server); let server = Arc::clone(&self.server);
let tx = self.tx.clone(); let tx = self.tx.clone();
let ctx = self.rt.handle().clone();
self.rt.spawn(async move { self.rt.spawn(async move {
let files = { let files: Vec<DataFile> = {
let symbols = server.symbols.read().unwrap(); let symbols = server.symbols.read().unwrap();
symbols.get(&symbol).cloned().unwrap_or_default() symbols.get(&symbol).cloned().unwrap_or_default()
}; };
let mut all_data = Vec::new(); let mut next_load: Option<tokio::task::JoinHandle<anyhow::Result<Arc<Vec<DataPoint<OhlcItem>>>>>> = None;
for file in files {
if file.symbol.ends_with(".m1") { for i in 0..files.len() {
if let Ok(data) = server.load_m1_data(&file).await { let file = &files[i];
all_data.extend_from_slice(&data); if !file.symbol.ends_with(".m1") {
} continue;
}
let data_res: anyhow::Result<Arc<Vec<DataPoint<OhlcItem>>>> = 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(); ctx.request_repaint();
} }
// Poll for finished loading tasks // Poll for finished loading tasks (Feed)
if let Ok(data) = self.rx.try_recv() { let mut received_any = false;
self.loaded_data = Some(data); while let Ok(msg) = self.rx.try_recv() {
self.is_loading = false; if let Some(chunk) = msg {
self.status_msg = format!( self.loaded_data.extend_from_slice(&chunk);
"Loaded {} records", received_any = true;
self.loaded_data.as_ref().unwrap().len() } else {
); self.is_loading = false;
}
}
if received_any {
self.status_msg = format!("Loaded {} records", self.loaded_data.len());
} }
let mut symbol_to_load = None; let mut symbol_to_load = None;
@@ -248,27 +274,29 @@ impl eframe::App for DataViewerApp {
if let Some(symbol) = &self.selected_symbol { if let Some(symbol) = &self.selected_symbol {
ui.heading(format!("Data: {}", symbol)); ui.heading(format!("Data: {}", symbol));
if self.is_loading { if self.is_loading {
ui.add(egui::Spinner::new()); ui.horizontal(|ui| {
ui.label("Streaming from network share..."); ui.add(egui::Spinner::new());
} else if let Some(data) = &self.loaded_data { ui.label("Streaming from network share...");
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.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 { } else {
ui.label("Bitte ein Symbol auswählen."); ui.label("Bitte ein Symbol auswählen.");
} }