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

Add Sync bound to DataServer Generic

Browse Source 
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.
This commit is contained in:
Michael Schimmel
2026-02-16 16:18:49 +01:00
parent 147b4b745c
commit b6a559b8bc
1 changed files with 79 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main.rs
+79 -42
View File
@@ -27,14 +27,14 @@ struct CacheEntry<T> {
last_accessed: Instant, last_accessed: Instant,
} }
pub struct DataServer<T: bytemuck::Pod + Send + 'static> { pub struct DataServer<T: bytemuck::Pod + Send + Sync + 'static> {
base_path: PathBuf, base_path: PathBuf,
symbols: RwLock<HashMap<String, Vec<DataFile>>>, symbols: RwLock<HashMap<String, Vec<DataFile>>>,
cache: RwLock<HashMap<PathBuf, CacheEntry<T>>>, cache: RwLock<HashMap<PathBuf, CacheEntry<T>>>,
sys: RwLock<System>, sys: RwLock<System>,
} }
impl<T: bytemuck::Pod + Send + 'static> DataServer<T> { impl<T: bytemuck::Pod + Send + Sync + 'static> DataServer<T> {
pub fn new(path: impl AsRef<Path>) -> Self { pub fn new(path: impl AsRef<Path>) -> Self {
let mut sys = System::new_all(); let mut sys = System::new_all();
sys.refresh_memory(); sys.refresh_memory();
@@ -110,13 +110,31 @@ impl<T: bytemuck::Pod + Send + 'static> DataServer<T> {
return Ok(Arc::clone(&entry.data)); return Ok(Arc::clone(&entry.data));
} }
} }
let raw_data = self.read_from_zip::<R>(&data_file.path).await?;
let mapped_data: Vec<T> = 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<T> = 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(); let mut cache = self.cache.write().unwrap();
cache.insert( cache.insert(
data_file.path.clone(), data_file.path.clone(),
@@ -154,8 +172,14 @@ impl<T: bytemuck::Pod + Send + 'static> DataServer<T> {
enum LoadedData { enum LoadedData {
None, None,
M1(Vec<DataPoint<OhlcItem>>), M1 {
Ticks(Vec<DataPoint<TickRecord>>), chunks: Vec<Arc<Vec<DataPoint<OhlcItem>>>>,
total_count: usize,
},
Ticks {
chunks: Vec<Arc<Vec<DataPoint<TickRecord>>>>,
total_count: usize,
},
} }
enum DataChunk { enum DataChunk {
@@ -214,9 +238,9 @@ impl DataViewerApp {
self.is_loading = true; self.is_loading = true;
self.status_msg = format!("Loading {}...", symbol); self.status_msg = format!("Loading {}...", symbol);
self.loaded_data = if symbol.ends_with(".m1") { self.loaded_data = if symbol.ends_with(".m1") {
LoadedData::M1(Vec::new()) LoadedData::M1 { chunks: Vec::new(), total_count: 0 }
} else { } else {
LoadedData::Ticks(Vec::new()) LoadedData::Ticks { chunks: Vec::new(), total_count: 0 }
}; };
let m1_server = Arc::clone(&self.m1_server); let m1_server = Arc::clone(&self.m1_server);
@@ -291,13 +315,15 @@ impl eframe::App for DataViewerApp {
if let Some(chunk) = msg { if let Some(chunk) = msg {
match chunk { match chunk {
DataChunk::M1(data) => { DataChunk::M1(data) => {
if let LoadedData::M1(vec) = &mut self.loaded_data { if let LoadedData::M1 { chunks, total_count } = &mut self.loaded_data {
vec.extend_from_slice(&data); *total_count += data.len();
chunks.push(data);
} }
} }
DataChunk::Ticks(data) => { DataChunk::Ticks(data) => {
if let LoadedData::Ticks(vec) = &mut self.loaded_data { if let LoadedData::Ticks { chunks, total_count } = &mut self.loaded_data {
vec.extend_from_slice(&data); *total_count += data.len();
chunks.push(data);
} }
} }
} }
@@ -309,8 +335,8 @@ impl eframe::App for DataViewerApp {
if received_any { if received_any {
let count = match &self.loaded_data { let count = match &self.loaded_data {
LoadedData::M1(v) => v.len(), LoadedData::M1 { total_count, .. } => *total_count,
LoadedData::Ticks(v) => v.len(), LoadedData::Ticks { total_count, .. } => *total_count,
_ => 0, _ => 0,
}; };
self.status_msg = format!("Loaded {} records", count); self.status_msg = format!("Loaded {} records", count);
@@ -346,33 +372,44 @@ impl eframe::App for DataViewerApp {
} }
match &self.loaded_data { match &self.loaded_data {
LoadedData::M1(data) => { LoadedData::M1 { chunks, total_count } => {
ui.label(format!("Successfully loaded {} OHLC records.", data.len())); ui.label(format!("Successfully loaded {} OHLC records.", total_count));
egui::ScrollArea::vertical().show(ui, |ui| { let row_height = ui.text_style_height(&egui::TextStyle::Body);
for i in 0..data.len().min(10000) { egui::ScrollArea::vertical().show_rows(ui, row_height, *total_count, |ui, row_range| {
let d = &data[i]; for i in row_range {
let open = d.data.open; // Find which chunk contains index i
let high = d.data.high; let mut current_idx = i;
let low = d.data.low; for chunk in chunks {
let close = d.data.close; if current_idx < chunk.len() {
ui.label(format!( let d = &chunk[current_idx];
"{}: O:{:.5} H:{:.5} L:{:.5} C:{:.5}", let open = d.data.open;
i, open, high, low, close 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) => { LoadedData::Ticks { chunks, total_count } => {
ui.label(format!("Successfully loaded {} Tick records.", data.len())); ui.label(format!("Successfully loaded {} Tick records.", total_count));
egui::ScrollArea::vertical().show(ui, |ui| { let row_height = ui.text_style_height(&egui::TextStyle::Body);
for i in 0..data.len().min(10000) { egui::ScrollArea::vertical().show_rows(ui, row_height, *total_count, |ui, row_range| {
let d = &data[i]; for i in row_range {
let ask = d.data.ask; let mut current_idx = i;
let bid = d.data.bid; for chunk in chunks {
ui.label(format!( if current_idx < chunk.len() {
"{}: A:{:.5} B:{:.5}", let d = &chunk[current_idx];
i, ask, bid 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();
}
} }
}); });
} }