diff --git a/services/receiver/src/rd_poll.rs b/services/receiver/src/rd_poll.rs index 1466a42c..cf2fd535 100644 --- a/services/receiver/src/rd_poll.rs +++ b/services/receiver/src/rd_poll.rs @@ -48,19 +48,42 @@ fn signature(dir: &Path) -> Option { /// Copy each active RD file into a fresh temp directory and parse the copies. Copying /// first mitigates sharing violations / torn reads while RD writes the /// originals. Any copy or parse error is returned so the caller keeps last good. -fn snapshot_and_load(dir: &Path) -> Result { - let files = rd_dbf::resolve_rd_files(dir)?; - let tmp = tempfile::tempdir().map_err(|e| RdError::Io { - file: "".to_owned(), - source: e, - })?; - for file in files { - std::fs::copy(dir.join(file), tmp.path().join(file)).map_err(|e| RdError::Io { - file: file.to_owned(), +async fn snapshot_and_load(dir: &Path) -> Result { + let dir_owned = dir.to_path_buf(); + let (files, tmp) = tokio::task::spawn_blocking(move || { + let files = rd_dbf::resolve_rd_files(&dir_owned)?; + let tmp = tempfile::tempdir().map_err(|e| RdError::Io { + file: "".to_owned(), source: e, })?; + Ok::<_, RdError>((files, tmp)) + }) + .await + .map_err(|e| RdError::Io { + file: "".to_owned(), + source: std::io::Error::other(e), + })??; + + for file in files { + tokio::fs::copy(dir.join(file), tmp.path().join(file)) + .await + .map_err(|e| RdError::Io { + file: file.to_owned(), + source: e, + })?; } - rd_dbf::load_from_dir(tmp.path()) + + let tmp_path = tmp.path().to_path_buf(); + let import = tokio::task::spawn_blocking(move || rd_dbf::load_from_dir(&tmp_path)) + .await + .map_err(|e| RdError::Io { + file: "".to_owned(), + source: std::io::Error::other(e), + })??; + + tokio::task::spawn_blocking(move || drop(tmp)).await.ok(); + + Ok(import) } /// Run one poll pass against `dir`. @@ -81,18 +104,15 @@ pub async fn poll_once( return PollOutcome::Skipped; } - let dir_owned = dir.to_path_buf(); - let loaded = tokio::task::spawn_blocking(move || snapshot_and_load(&dir_owned)).await; - match loaded { - Ok(Ok(import)) => match crate::control_api::apply_rd_import(state, import).await { + match snapshot_and_load(dir).await { + Ok(import) => match crate::control_api::apply_rd_import(state, import).await { Ok(summary) => { *last = sig; PollOutcome::Imported(summary.resolvable_chips) } Err(e) => PollOutcome::Failed(e.to_string()), }, - Ok(Err(e)) => PollOutcome::Failed(e.to_string()), - Err(e) => PollOutcome::Failed(format!("import task failed: {e}")), + Err(e) => PollOutcome::Failed(e.to_string()), } } diff --git a/services/receiver/tests/bench_rd_poll.rs b/services/receiver/tests/bench_rd_poll.rs new file mode 100644 index 00000000..b596381b --- /dev/null +++ b/services/receiver/tests/bench_rd_poll.rs @@ -0,0 +1,46 @@ +use receiver::control_api::AppState; +use receiver::db::Db; +use receiver::rd_dbf::RD_FILES; +use receiver::rd_poll::poll_once; +use std::path::{Path, PathBuf}; +use std::time::Instant; + +fn fixtures_dir() -> PathBuf { + PathBuf::from(env!("CARGO_MANIFEST_DIR")) + .join("tests") + .join("fixtures") + .join("rd") +} + +fn copy_fixtures_to(dir: &Path) { + for file in RD_FILES { + std::fs::copy(fixtures_dir().join(file), dir.join(file)).unwrap(); + } +} + +#[tokio::test] +#[ignore] +async fn bench_rd_poll_once() { + let (state, _rx) = AppState::new(Db::open_in_memory().unwrap(), "recv".to_owned()); + let tmp = tempfile::tempdir().unwrap(); + copy_fixtures_to(tmp.path()); + + // Warmup + let mut last = None; + poll_once(&state, tmp.path(), &mut last).await; + + let iterations = 100; + let start = Instant::now(); + for _ in 0..iterations { + let mut last = None; // Force snapshot_and_load each iteration + poll_once(&state, tmp.path(), &mut last).await; + } + let elapsed = start.elapsed(); + let avg = elapsed / iterations; + + println!("\n=================================================="); + println!("BENCHMARK: poll_once snapshot_and_load ({iterations} iterations):"); + println!("Total time: {:?}", elapsed); + println!("Average time per poll: {:?}", avg); + println!("==================================================\n"); +}