make commit filtering an async job (#1842)

asyncgit/Cargo.toml

@@ -25,7 +25,7 @@ openssl-sys = { version = '0.9', features = ["vendored"], optional = true }
 rayon-core = "1.11"
 scopetime = { path = "../scopetime", version = "0.1" }
 serde = { version = "1.0", features = ["derive"] }
-shellexpand = "3.1" 
+shellexpand = "3.1"
 thiserror = "1.0"
 unicode-truncate = "0.2.0"
 url = "2.4"

asyncgit/src/filter_commits.rs

@@ -0,0 +1,149 @@
+use crate::{
+	asyncjob::{AsyncJob, RunParams},
+	error::Result,
+	sync::{self, CommitId, LogWalkerFilter, RepoPath},
+	AsyncGitNotification, ProgressPercent,
+};
+use std::{
+	sync::{Arc, Mutex},
+	time::{Duration, Instant},
+};
+
+///
+pub struct CommitFilterResult {
+	///
+	pub result: Vec<CommitId>,
+	///
+	pub duration: Duration,
+}
+
+enum JobState {
+	Request {
+		commits: Vec<CommitId>,
+		repo_path: RepoPath,
+	},
+	Response(Result<CommitFilterResult>),
+}
+
+///
+#[derive(Clone)]
+pub struct AsyncCommitFilterJob {
+	state: Arc<Mutex<Option<JobState>>>,
+	filter: LogWalkerFilter,
+}
+
+///
+impl AsyncCommitFilterJob {
+	///
+	pub fn new(
+		repo_path: RepoPath,
+		commits: Vec<CommitId>,
+		filter: LogWalkerFilter,
+	) -> Self {
+		Self {
+			state: Arc::new(Mutex::new(Some(JobState::Request {
+				repo_path,
+				commits,
+			}))),
+			filter,
+		}
+	}
+
+	///
+	pub fn result(&self) -> Option<Result<CommitFilterResult>> {
+		if let Ok(mut state) = self.state.lock() {
+			if let Some(state) = state.take() {
+				return match state {
+					JobState::Request { .. } => None,
+					JobState::Response(result) => Some(result),
+				};
+			}
+		}
+
+		None
+	}
+
+	fn run_request(
+		&self,
+		repo_path: &RepoPath,
+		commits: Vec<CommitId>,
+		params: &RunParams<AsyncGitNotification, ProgressPercent>,
+	) -> JobState {
+		let response = sync::repo(repo_path)
+			.map(|repo| self.filter_commits(&repo, commits, params))
+			.map(|(start, result)| CommitFilterResult {
+				result,
+				duration: start.elapsed(),
+			});
+
+		JobState::Response(response)
+	}
+
+	fn filter_commits(
+		&self,
+		repo: &git2::Repository,
+		commits: Vec<CommitId>,
+		params: &RunParams<AsyncGitNotification, ProgressPercent>,
+	) -> (Instant, Vec<CommitId>) {
+		let total_amount = commits.len();
+		let start = Instant::now();
+
+		let mut progress = ProgressPercent::new(0, total_amount);
+
+		let result = commits
+			.into_iter()
+			.enumerate()
+			.filter_map(|(idx, c)| {
+				let new_progress =
+					ProgressPercent::new(idx, total_amount);
+
+				if new_progress != progress {
+					Self::update_progress(params, new_progress);
+					progress = new_progress;
+				}
+
+				(*self.filter)(repo, &c)
+					.ok()
+					.and_then(|res| res.then_some(c))
+			})
+			.collect::<Vec<_>>();
+
+		(start, result)
+	}
+
+	fn update_progress(
+		params: &RunParams<AsyncGitNotification, ProgressPercent>,
+		new_progress: ProgressPercent,
+	) {
+		if let Err(e) = params.set_progress(new_progress) {
+			log::error!("progress error: {e}");
+		} else if let Err(e) =
+			params.send(AsyncGitNotification::CommitFilter)
+		{
+			log::error!("send error: {e}");
+		}
+	}
+}
+
+impl AsyncJob for AsyncCommitFilterJob {
+	type Notification = AsyncGitNotification;
+	type Progress = ProgressPercent;
+
+	fn run(
+		&mut self,
+		params: RunParams<Self::Notification, Self::Progress>,
+	) -> Result<Self::Notification> {
+		if let Ok(mut state) = self.state.lock() {
+			*state = state.take().map(|state| match state {
+				JobState::Request { commits, repo_path } => {
+					self.run_request(&repo_path, commits, &params)
+				}
+				JobState::Response(result) => {
+					JobState::Response(result)
+				}
+			});
+		}
+
+		Ok(AsyncGitNotification::CommitFilter)
+	}
+}

asyncgit/src/lib.rs

@@ -38,6 +38,7 @@ mod commit_files;
 mod diff;
 mod error;
 mod fetch_job;
+mod filter_commits;
 mod progress;
 mod pull;
 mod push;
@@ -57,6 +58,7 @@ pub use crate::{
 	diff::{AsyncDiff, DiffParams, DiffType},
 	error::{Error, Result},
 	fetch_job::AsyncFetchJob,
+	filter_commits::{AsyncCommitFilterJob, CommitFilterResult},
 	progress::ProgressPercent,
 	pull::{AsyncPull, FetchRequest},
 	push::{AsyncPush, PushRequest},
@@ -111,6 +113,8 @@ pub enum AsyncGitNotification {
 	Branches,
 	///
 	TreeFiles,
+	///
+	CommitFilter,
 }
 
 /// helper function to calculate the hash of an arbitrary type that implements the `Hash` trait

asyncgit/src/revlog.rs

@@ -81,6 +81,10 @@ impl AsyncLog {
 		start_index: usize,
 		amount: usize,
 	) -> Result<Vec<CommitId>> {
+		if self.partial_extract.load(Ordering::Relaxed) {
+			return Err(Error::Generic(String::from("Faulty usage of AsyncLog: Cannot partially extract items and rely on get_items slice to still work!")));
+		}
+
 		let list = &self.current.lock()?.commits;
 		let list_len = list.len();
 		let min = start_index.min(list_len);

asyncgit/src/sync/logwalker.rs

@@ -32,6 +32,7 @@ impl<'a> Ord for TimeOrderedCommit<'a> {
 	}
 }
 
+//TODO: since its used in more than just the log walker now, we should rename and put in its own file
 ///
 pub type LogWalkerFilter = Arc<
 	Box<dyn Fn(&Repository, &CommitId) -> Result<bool> + Send + Sync>,

src/app.rs

@@ -976,7 +976,7 @@ impl App {
 				self.reset_popup.open(id)?;
 			}
 			InternalEvent::CommitSearch(options) => {
-				self.revlog.search(options)?;
+				self.revlog.search(options);
 			}
 		};
 

src/components/commitlist.rs

@@ -97,6 +97,11 @@ impl CommitList {
 		self.items.clear();
 	}
 
+	///
+	pub fn copy_items(&self) -> Vec<CommitId> {
+		self.commits.clone()
+	}
+
 	///
 	pub fn set_tags(&mut self, tags: Tags) {
 		self.tags = Some(tags);

src/tabs/revlog.rs

@@ -18,8 +18,9 @@ use asyncgit::{
 		self, filter_commit_by_search, CommitId, LogFilterSearch,
 		LogFilterSearchOptions, RepoPathRef,
 	},
-	AsyncBranchesJob, AsyncGitNotification, AsyncLog, AsyncTags,
-	CommitFilesParams, FetchStatus,
+	AsyncBranchesJob, AsyncCommitFilterJob, AsyncGitNotification,
+	AsyncLog, AsyncTags, CommitFilesParams, FetchStatus,
+	ProgressPercent,
 };
 use crossbeam_channel::Sender;
 use crossterm::event::Event;
@@ -42,18 +43,14 @@ struct LogSearchResult {
 //TODO: deserves its own component
 enum LogSearch {
 	Off,
-	Searching(AsyncLog, LogFilterSearchOptions),
+	Searching(
+		AsyncSingleJob<AsyncCommitFilterJob>,
+		LogFilterSearchOptions,
+		Option<ProgressPercent>,
+	),
 	Results(LogSearchResult),
 }
 
-impl LogSearch {
-	fn set_background(&mut self) {
-		if let Self::Searching(log, _) = self {
-			log.set_background();
-		}
-	}
-}
-
 ///
 pub struct Revlog {
 	repo: RepoPathRef,
@@ -124,7 +121,7 @@ impl Revlog {
 	}
 
 	const fn is_search_pending(&self) -> bool {
-		matches!(self.search, LogSearch::Searching(_, _))
+		matches!(self.search, LogSearch::Searching(_, _, _))
 	}
 
 	///
@@ -134,8 +131,6 @@ impl Revlog {
 				self.list.clear();
 			}
 
-			self.update_search_state()?;
-
 			self.list
 				.refresh_extend_data(self.git_log.extract_items()?);
 
@@ -164,6 +159,9 @@ impl Revlog {
 			match ev {
 				AsyncGitNotification::CommitFiles
 				| AsyncGitNotification::Log => self.update()?,
+				AsyncGitNotification::CommitFilter => {
+					self.update_search_state();
+				}
 				AsyncGitNotification::Tags => {
 					if let Some(tags) = self.git_tags.last()? {
 						self.list.set_tags(tags);
@@ -242,10 +240,11 @@ impl Revlog {
 		}
 	}
 
-	pub fn search(
-		&mut self,
-		options: LogFilterSearchOptions,
-	) -> Result<()> {
+	pub fn search(&mut self, options: LogFilterSearchOptions) {
+		if !self.can_start_search() {
+			return;
+		}
+
 		if matches!(
 			self.search,
 			LogSearch::Off | LogSearch::Results(_)
@@ -256,47 +255,60 @@ impl Revlog {
 				LogFilterSearch::new(options.clone()),
 			);
 
-			let mut async_find = AsyncLog::new(
+			let mut job = AsyncSingleJob::new(self.sender.clone());
+			job.spawn(AsyncCommitFilterJob::new(
 				self.repo.borrow().clone(),
-				&self.sender,
-				Some(filter),
-			);
+				self.list.copy_items(),
+				filter,
+			));
 
-			assert_eq!(async_find.fetch()?, FetchStatus::Started);
-
-			self.search = LogSearch::Searching(async_find, options);
+			self.search = LogSearch::Searching(job, options, None);
 
 			self.list.set_highlighting(None);
 		}
-
-		Ok(())
 	}
 
-	fn update_search_state(&mut self) -> Result<bool> {
-		let changes = match &self.search {
-			LogSearch::Off | LogSearch::Results(_) => false,
-			LogSearch::Searching(search, options) => {
+	fn update_search_state(&mut self) {
+		match &mut self.search {
+			LogSearch::Off | LogSearch::Results(_) => (),
+			LogSearch::Searching(search, options, progress) => {
 				if search.is_pending() {
-					false
-				} else {
-					let results = search.extract_items()?;
-					let duration = search.get_last_duration()?;
+					//update progress
+					*progress = search.progress();
+				} else if let Some(search) = search
+					.take_last()
+					.and_then(|search| search.result())
+				{
+					match search {
+						Ok(search) => {
+							self.list.set_highlighting(Some(
+								Rc::new(
+									search
+										.result
+										.into_iter()
+										.collect::<IndexSet<_>>(),
+								),
+							));
 
-					self.list.set_highlighting(Some(Rc::new(
-						results.into_iter().collect::<IndexSet<_>>(),
-					)));
+							self.search =
+								LogSearch::Results(LogSearchResult {
+									options: options.clone(),
+									duration: search.duration,
+								});
+						}
+						Err(err) => {
+							self.queue.push(
+								InternalEvent::ShowErrorMsg(format!(
+									"search error: {err}",
+								)),
+							);
 
-					self.search =
-						LogSearch::Results(LogSearchResult {
-							options: options.clone(),
-							duration,
-						});
-					true
+							self.search = LogSearch::Off;
+						}
+					}
 				}
 			}
-		};
-
-		Ok(changes)
+		}
 	}
 
 	fn is_in_search_mode(&self) -> bool {
@@ -305,9 +317,14 @@ impl Revlog {
 
 	fn draw_search<B: Backend>(&self, f: &mut Frame<B>, area: Rect) {
 		let (text, title) = match &self.search {
-			LogSearch::Searching(_, options) => (
+			LogSearch::Searching(_, options, progress) => (
 				format!("'{}'", options.search_pattern.clone()),
-				String::from("(pending results...)"),
+				format!(
+					"({}%)",
+					progress
+						.map(|progress| progress.progress)
+						.unwrap_or_default()
+				),
 			),
 			LogSearch::Results(results) => {
 				let info = self.list.highlighted_selection_info();
@@ -351,6 +368,10 @@ impl Revlog {
 	fn can_leave_search(&self) -> bool {
 		self.is_in_search_mode() && !self.is_search_pending()
 	}
+
+	fn can_start_search(&self) -> bool {
+		!self.git_log.is_pending()
+	}
 }
 
 impl DrawableComponent for Revlog {
@@ -514,6 +535,7 @@ impl Component for Revlog {
 						},
 					);
 				} else if key_match(k, self.key_config.keys.log_find)
+					&& self.can_start_search()
 				{
 					self.queue
 						.push(InternalEvent::OpenLogSearchPopup);
@@ -662,7 +684,7 @@ impl Component for Revlog {
 		));
 		out.push(CommandInfo::new(
 			strings::commands::log_find_commit(&self.key_config),
-			true,
+			self.can_start_search(),
 			self.visible || force_all,
 		));
 
@@ -676,7 +698,6 @@ impl Component for Revlog {
 	fn hide(&mut self) {
 		self.visible = false;
 		self.git_log.set_background();
-		self.search.set_background();
 	}
 
 	fn show(&mut self) -> Result<()> {