diff --git a/crates/pstr-app/src/app.rs b/crates/pstr-app/src/app.rs index 926de74..81c6f8d 100644 --- a/crates/pstr-app/src/app.rs +++ b/crates/pstr-app/src/app.rs @@ -103,6 +103,8 @@ pub enum Action { RemoveDownload(DownloadKey, bool), /// Crawl one share, or every share. Crawl(Option), + /// Stop the crawls still listing. + StopCrawl, AddShare { name: String, url: String, @@ -422,7 +424,8 @@ pub struct App { pub connecting: bool, pub crawling: bool, /// Which share the crawl is on, and how far it has got. - crawl_progress: Option<(String, usize)>, + /// Files found so far, by share name, for each share still listing. + crawl_progress: std::collections::BTreeMap, pub downloads: Vec, pub offline_files: std::collections::HashSet, pub confirm_partial_delete: Option, @@ -486,7 +489,7 @@ impl App { opening: None, connecting: true, crawling: false, - crawl_progress: None, + crawl_progress: Default::default(), downloads: Vec::new(), offline_files: std::collections::HashSet::new(), confirm_partial_delete: None, @@ -551,6 +554,7 @@ impl App { seconds, } => { let name = self.share_name(&share_id); + self.crawl_progress.remove(&name); self.note( ctx, format!("{name}: {files} playable of {nodes} nodes in {seconds:.0}s"), @@ -559,11 +563,16 @@ impl App { } Event::CrawlFinished => { self.crawling = false; - self.crawl_progress = None; + self.crawl_progress.clear(); } Event::CrawlProgress { share, found } => { self.crawling = true; - self.crawl_progress = Some((share, found)); + self.crawl_progress.insert(share, found); + } + Event::CrawlStopped { share_id } => { + let name = self.share_name(&share_id); + self.crawl_progress.remove(&name); + self.note(ctx, format!("stopped crawling {name}"), false); } Event::Thumbnail { key, image } => self.thumbs.insert(key, image), Event::ThumbnailMissing { key } => self.thumbs.mark_missing(key), @@ -1058,6 +1067,7 @@ impl App { self.note(ctx, "crawling…", false); self.engine.crawl(share); } + Action::StopCrawl => self.engine.stop_crawl(), Action::AddShare { name, url, @@ -1266,12 +1276,15 @@ impl App { ui.with_layout(egui::Layout::right_to_left(egui::Align::Center), |ui| { ui.add_space(4.0); if self.crawling { + if ui + .small_button("Stop") + .on_hover_text("Stop listing; shares already listed are still stored") + .clicked() + { + actions.push(Action::StopCrawl); + } ui.add(egui::Spinner::new().size(16.0)); - ui.label(ui::muted(match &self.crawl_progress { - Some((share, 0)) => format!("crawling {share}"), - Some((share, found)) => format!("crawling {share} · {found} found"), - None => "crawling".to_owned(), - })); + ui.label(ui::muted(crawl_label(&self.crawl_progress))); } else if self.matching { ui.add(egui::Spinner::new().size(16.0)); ui.label(ui::muted("matching")); @@ -1621,6 +1634,22 @@ impl eframe::App for App { } } +/// What the top bar says while shares are being listed: one share by name, +/// several as a count, with the files found across all of them. +fn crawl_label(progress: &std::collections::BTreeMap) -> String { + let found: usize = progress.values().sum(); + let which = match progress.len() { + 0 => return "crawling".to_owned(), + 1 => progress.keys().next().cloned().unwrap_or_default(), + shares => format!("{shares} shares"), + }; + if found == 0 { + format!("crawling {which}") + } else { + format!("crawling {which} · {found} found") + } +} + /// Titles whose tile art is not what it was. /// /// Compared by URL: the same URL is the same picture, and anything else — @@ -1889,6 +1918,17 @@ mod tests { assert!(!Filter::Watching.admits(&film)); } + #[test] + fn the_crawl_label_names_one_share_and_counts_several() { + let mut progress = std::collections::BTreeMap::new(); + assert_eq!(crawl_label(&progress), "crawling"); + progress.insert("anime".to_owned(), 0); + assert_eq!(crawl_label(&progress), "crawling anime"); + progress.insert("films".to_owned(), 40); + progress.insert("anime".to_owned(), 2); + assert_eq!(crawl_label(&progress), "crawling 2 shares · 42 found"); + } + #[test] fn speed_steps_land_back_on_normal() { assert_eq!(step_speed(1.0, true), 1.25); diff --git a/crates/pstr-app/src/engine.rs b/crates/pstr-app/src/engine.rs index 9e4c123..5dec5ae 100644 --- a/crates/pstr-app/src/engine.rs +++ b/crates/pstr-app/src/engine.rs @@ -186,6 +186,8 @@ pub enum Event { CrawlProgress { share: String, found: usize }, /// Every requested crawl is done. CrawlFinished, + /// That share's crawl was stopped before it stored anything. + CrawlStopped { share_id: String }, /// A decoded Proton thumbnail, keyed as [`thumbnail_key`]. Thumbnail { key: String, @@ -314,6 +316,8 @@ pub struct Engine { download_permits: Arc, /// Whether the player page is on screen. See [`Engine::wake`]. picture_shown: Arc, + /// Bumped to stop the crawls in flight; each watches it from its start. + crawl_cancel: Arc>, events: Sender, ctx: egui::Context, } @@ -410,6 +414,7 @@ impl Engine { downloads: Arc::new(Mutex::new(Downloads::default())), download_permits: Arc::new(tokio::sync::Semaphore::new(DOWNLOAD_CONCURRENCY)), picture_shown: Arc::new(AtomicBool::new(false)), + crawl_cancel: Arc::new(tokio::sync::watch::channel(0).0), events, ctx, }; @@ -798,117 +803,148 @@ impl Engine { None => library.share_ids().map(str::to_string).collect(), }; - for share_id in targets { - let started = std::time::Instant::now(); - let name = self - .store - .list() - .ok() - .and_then(|shares| shares.into_iter().find(|share| share.id == share_id)) - .map_or_else(|| share_id.clone(), |share| share.name); - self.emit(Event::CrawlProgress { - share: name.clone(), - found: 0, - }); - // At most a few times a second: a wide level comes back as a burst - // of listings, and a frame per listing is a frame per folder. - let last = parking_lot::Mutex::new(std::time::Instant::now()); - let report = |found: usize| { - let mut last = last.lock(); - if last.elapsed() >= std::time::Duration::from_millis(250) { - *last = std::time::Instant::now(); - self.emit(Event::CrawlProgress { - share: name.clone(), - found, - }); - } - }; - let nodes = match library.crawl_reporting(&share_id, report).await { + // Every share at once: each is its own session against its own + // tree, and one large share no longer holds the rest back behind it. + let cancel = self.crawl_cancel.subscribe(); + futures::future::join_all( + targets + .into_iter() + .map(|share_id| self.crawl_share(&library, share_id, cancel.clone())), + ) + .await; + self.emit(Event::CrawlFinished); + } + + /// Stop every crawl still listing. Shares already past their listing + /// finish storing what they found. + pub fn stop_crawl(&self) { + self.crawl_cancel.send_modify(|generation| *generation += 1); + } + + /// Crawl one share and store what it holds. + async fn crawl_share( + &self, + library: &SharedLibrary, + share_id: String, + mut cancel: tokio::sync::watch::Receiver, + ) { + let started = std::time::Instant::now(); + let name = self + .store + .list() + .ok() + .and_then(|shares| shares.into_iter().find(|share| share.id == share_id)) + .map_or_else(|| share_id.clone(), |share| share.name); + self.emit(Event::CrawlProgress { + share: name.clone(), + found: 0, + }); + // At most a few times a second: a wide level comes back as a burst + // of listings, and a frame per listing is a frame per folder. + let last = parking_lot::Mutex::new(std::time::Instant::now()); + let report = |found: usize| { + let mut last = last.lock(); + if last.elapsed() >= std::time::Duration::from_millis(250) { + *last = std::time::Instant::now(); + self.emit(Event::CrawlProgress { + share: name.clone(), + found, + }); + } + }; + // Only the listing can be stopped. Past it the crawl is replacing the + // share's rows and its stale downloads, and stopping halfway through + // that would leave the two disagreeing. + let nodes = tokio::select! { + result = library.crawl_reporting(&share_id, report) => match result { Ok(nodes) => nodes, Err(error) => { self.fail(&format!("crawl {share_id}"), error); - continue; + return; } - }; - let rows = build_rows(&share_id, &nodes); - let files = rows.len(); - - // A writer opened before the crawl may be producing the revision - // this crawl is about to supersede. Quiesce all writers for the - // share before inspecting or deleting any revision paths. - let active_downloads: Vec<_> = self - .downloads - .lock() - .jobs - .keys() - .filter(|key| key.share_id == share_id) - .cloned() - .collect(); - for key in &active_downloads { - self.cancel_and_join(key).await; + }, + Ok(()) = cancel.changed() => { + self.emit(Event::CrawlStopped { share_id }); + return; } + }; + let rows = build_rows(&share_id, &nodes); + let files = rows.len(); - // Paths are derived from the old revision id, so inspect and - // remove stale bytes before changing or dropping their index rows. - let retained = match self.catalog.lock().offline_files_for_share(&share_id) { - Ok(files) => files, - Err(error) => { - self.fail(&format!("inspect downloads for {share_id}"), error); - continue; - } - }; - let revisions: HashMap<&str, Option<&str>> = rows - .iter() - .map(|row| (row.link_id.as_str(), row.active_revision_id.as_deref())) - .collect(); - let stale: Vec<_> = retained - .iter() - .filter(|(link_id, file)| { - revisions.get(link_id.as_str()).copied().flatten() - != Some(file.revision_id.as_str()) - }) - .map(|(link_id, file)| (link_id.clone(), file.clone())) - .collect(); - let mut cleanup_failed = false; - for (link_id, file) in &stale { - let path = self - .dirs - .offline_file(&share_id, link_id, &file.revision_id); - if let Err(error) = remove_if_present(&path).await { - self.fail(&format!("delete stale download for {share_id}"), error); - cleanup_failed = true; - break; - } - } - if cleanup_failed { - continue; - } - - let stored = { - let mut catalog = self.catalog.lock(); - catalog - .replace_share_retaining_offline(&share_id, &rows) - .and_then(|()| { - for (link_id, _) in &stale { - catalog.remove_offline_file(&share_id, link_id)?; - } - Ok(()) - }) - }; - if let Err(error) = stored { - self.fail(&format!("store {share_id}"), error); - continue; - } - - self.emit(Event::Crawled { - share_id, - nodes: nodes.len(), - files, - seconds: started.elapsed().as_secs_f64(), - }); - self.load_library(); + // A writer opened before the crawl may be producing the revision + // this crawl is about to supersede. Quiesce all writers for the + // share before inspecting or deleting any revision paths. + let active_downloads: Vec<_> = self + .downloads + .lock() + .jobs + .keys() + .filter(|key| key.share_id == share_id) + .cloned() + .collect(); + for key in &active_downloads { + self.cancel_and_join(key).await; } - self.emit(Event::CrawlFinished); + + // Paths are derived from the old revision id, so inspect and + // remove stale bytes before changing or dropping their index rows. + let retained = match self.catalog.lock().offline_files_for_share(&share_id) { + Ok(files) => files, + Err(error) => { + self.fail(&format!("inspect downloads for {share_id}"), error); + return; + } + }; + let revisions: HashMap<&str, Option<&str>> = rows + .iter() + .map(|row| (row.link_id.as_str(), row.active_revision_id.as_deref())) + .collect(); + let stale: Vec<_> = retained + .iter() + .filter(|(link_id, file)| { + revisions.get(link_id.as_str()).copied().flatten() + != Some(file.revision_id.as_str()) + }) + .map(|(link_id, file)| (link_id.clone(), file.clone())) + .collect(); + let mut cleanup_failed = false; + for (link_id, file) in &stale { + let path = self + .dirs + .offline_file(&share_id, link_id, &file.revision_id); + if let Err(error) = remove_if_present(&path).await { + self.fail(&format!("delete stale download for {share_id}"), error); + cleanup_failed = true; + break; + } + } + if cleanup_failed { + return; + } + + let stored = { + let mut catalog = self.catalog.lock(); + catalog + .replace_share_retaining_offline(&share_id, &rows) + .and_then(|()| { + for (link_id, _) in &stale { + catalog.remove_offline_file(&share_id, link_id)?; + } + Ok(()) + }) + }; + if let Err(error) = stored { + self.fail(&format!("store {share_id}"), error); + return; + } + + self.emit(Event::Crawled { + share_id, + nodes: nodes.len(), + files, + seconds: started.elapsed().as_secs_f64(), + }); + self.load_library(); } /// Record where playback got to.