use std::{ cell::LazyCell, collections::BTreeSet, io::{BufRead, BufReader}, ops::Range, path::{Path, PathBuf}, pin::pin, sync::Arc, time::Duration,}; use anyhow::Context;use async_channel::{Receiver, Sender, bounded, unbounded};use collections::HashSet;use fs::Fs;use futures::FutureExt as _;use futures::{SinkExt, StreamExt, select_biased, stream::FuturesOrdered};use gpui::{App, AppContext, AsyncApp, BackgroundExecutor, Entity, Priority, Task};use language::{Buffer, BufferSnapshot, Point};use parking_lot::Mutex;use postage::oneshot;use rpc::{AnyProtoClient, proto}; use util::{ResultExt, maybe, paths::compare_rel_paths, rel_path::RelPath};use worktree::{Entry, ProjectEntryId, Snapshot, Worktree, WorktreeSettings}; use crate::{ Project, ProjectItem, ProjectPath, RemotelyCreatedModels, buffer_store::BufferStore, search::{LineHint, SearchQuery, SearchResult}, worktree_store::WorktreeStore,}; pub struct Search { buffer_store: Entity<BufferStore>, worktree_store: Entity<WorktreeStore>, limit: usize, kind: SearchKind,} /// Represents search setup, before it is actually kicked off with Search::into_resultsenum SearchKind { /// Search for candidates by inspecting file contents on file system, avoiding loading the buffer unless we know that a given file contains a match. Local { fs: Arc<dyn Fs>, worktrees: Vec<Entity<Worktree>>, }, /// Query remote host for candidates. As of writing, the host runs a local search in "buffers with matches only" mode. Remote { client: AnyProtoClient, remote_id: u64, models: Arc<Mutex<RemotelyCreatedModels>>, }, /// Run search against a known set of candidates. Even when working with a remote host, this won't round-trip to host. OpenBuffersOnly,} /// Represents results of project search and allows one to either obtain match positions OR/// just the handles to buffers that may match the search. Grabbing the handles is cheaper than obtaining full match positions, because in that case we'll look for/// at most one match in each file.#[must_use]pub struct SearchResultsHandle { results: Receiver<SearchResult>, matching_buffers: Receiver<(Entity<Buffer>, LineHint)>, trigger_search: Box<dyn FnOnce(&mut App) -> Task<()> + Send + Sync>,} pub struct SearchResults<T> { pub task_handle: Task<()>, pub rx: Receiver<T>,}impl SearchResultsHandle { pub fn results(self, cx: &mut App) -> SearchResults<SearchResult> { SearchResults { task_handle: (self.trigger_search)(cx), rx: self.results, } } pub fn matching_buffers(self, cx: &mut App) -> SearchResults<(Entity<Buffer>, LineHint)> { SearchResults { task_handle: (self.trigger_search)(cx), rx: self.matching_buffers, } }} #[derive(Clone)]enum FindSearchCandidates { Local { fs: Arc<dyn Fs>, /// Start off with all paths in project and filter them based on: /// - Include filters /// - Exclude filters /// - Only open buffers /// - Scan ignored files /// Put another way: filter out files that can't match (without looking at file contents) input_paths_rx: Receiver<InputPath>, /// After that, if the buffer is not yet loaded, we'll figure out if it contains at least one match /// based on disk contents of a buffer. This step is not performed for buffers we already have in memory. confirm_contents_will_match_tx: Sender<MatchingEntry>, confirm_contents_will_match_rx: Receiver<MatchingEntry>, }, Remote, OpenBuffersOnly,} impl Search { pub fn local( fs: Arc<dyn Fs>, buffer_store: Entity<BufferStore>, worktree_store: Entity<WorktreeStore>, limit: usize, cx: &mut App, ) -> Self { let worktrees = worktree_store.read(cx).visible_worktrees(cx).collect(); Self { kind: SearchKind::Local { fs, worktrees }, buffer_store, worktree_store, limit, } } pub(crate) fn remote( buffer_store: Entity<BufferStore>, worktree_store: Entity<WorktreeStore>, limit: usize, client_state: (AnyProtoClient, u64, Arc<Mutex<RemotelyCreatedModels>>), ) -> Self { Self { kind: SearchKind::Remote { client: client_state.0, remote_id: client_state.1, models: client_state.2, }, buffer_store, worktree_store, limit, } } pub(crate) fn open_buffers_only( buffer_store: Entity<BufferStore>, worktree_store: Entity<WorktreeStore>, limit: usize, ) -> Self { Self { kind: SearchKind::OpenBuffersOnly, buffer_store, worktree_store, limit, } } pub(crate) const MAX_SEARCH_RESULT_FILES: usize = 5_000; pub const MAX_SEARCH_RESULT_RANGES: usize = 10_000; /// Prepares a project search run. The resulting [`SearchResultsHandle`] has to be used to specify whether you're interested in matching buffers /// or full search results. pub fn into_handle(mut self, query: SearchQuery, cx: &mut App) -> SearchResultsHandle { let mut open_buffers = HashSet::default(); let mut unnamed_buffers = Vec::new(); const MAX_CONCURRENT_BUFFER_OPENS: usize = 64; let buffers = self.buffer_store.read(cx); for handle in buffers.buffers() { let buffer = handle.read(cx); if !buffers.is_searchable(&buffer.remote_id()) { continue; } else if buffer .file() .is_some_and(|file| file.disk_state().is_deleted()) { continue; } else if let Some(entry_id) = buffer.entry_id(cx) { open_buffers.insert(entry_id); } else { self.limit = self.limit.saturating_sub(1); unnamed_buffers.push(handle) }; } let open_buffers = Arc::new(open_buffers); let executor = cx.background_executor().clone(); let (tx, rx) = unbounded(); let (grab_buffer_snapshot_tx, grab_buffer_snapshot_rx) = unbounded::<(Entity<Buffer>, LineHint)>(); let matching_buffers = grab_buffer_snapshot_rx.clone(); let trigger_search = Box::new(move |cx: &mut App| { cx.spawn(async move |cx| { for buffer in unnamed_buffers { _ = grab_buffer_snapshot_tx .send((buffer, LineHint::default())) .await; } let (find_all_matches_tx, find_all_matches_rx) = bounded(MAX_CONCURRENT_BUFFER_OPENS); let query = Arc::new(query); let (candidate_searcher, tasks) = match self.kind { SearchKind::OpenBuffersOnly => { let open_buffers = cx.update(|cx| self.all_loaded_buffers(&query, cx)); let fill_requests = cx .background_spawn(async move { for buffer in open_buffers { if let Err(_) = grab_buffer_snapshot_tx .send((buffer, LineHint::default())) .await { return; } } }) .boxed_local(); (FindSearchCandidates::OpenBuffersOnly, vec![fill_requests]) } SearchKind::Local { fs, ref mut worktrees, } => { let (get_buffer_for_full_scan_tx, get_buffer_for_full_scan_rx) = unbounded(); let (confirm_contents_will_match_tx, confirm_contents_will_match_rx) = bounded(64); let (sorted_search_results_tx, sorted_search_results_rx) = unbounded(); let (input_paths_tx, input_paths_rx) = unbounded(); let tasks = vec![ cx.spawn(Self::provide_search_paths( std::mem::take(worktrees), query.clone(), input_paths_tx, sorted_search_results_tx, tx.clone(), )) .boxed_local(), Self::open_buffers( self.buffer_store, get_buffer_for_full_scan_rx, grab_buffer_snapshot_tx, cx.clone(), ) .boxed_local(), cx.background_spawn(Self::maintain_sorted_search_results( sorted_search_results_rx, get_buffer_for_full_scan_tx, self.limit, )) .boxed_local(), ]; ( FindSearchCandidates::Local { fs, confirm_contents_will_match_tx, confirm_contents_will_match_rx, input_paths_rx, }, tasks, ) } SearchKind::Remote { client, remote_id, models, } => { let (handle, rx) = self .buffer_store .update(cx, |this, _| this.register_project_search_result_handle()); let cancel_ongoing_search = util::defer({ let client = client.clone(); move || { _ = client.send(proto::FindSearchCandidatesCancelled { project_id: remote_id, handle, }); } }); let request = client.request(proto::FindSearchCandidates { project_id: remote_id, query: Some(query.to_proto()), limit: self.limit as _, handle, }); let buffer_store = self.buffer_store; let guard = cx.update(|cx| { Project::retain_remotely_created_models_impl( &models, &buffer_store, &self.worktree_store, cx, ) }); let issue_remote_buffers_request = cx .spawn(async move |cx| { let _ = maybe!(async move { request.await?; let (buffer_tx, buffer_rx) = bounded(24); let wait_for_remote_buffers = cx.spawn(async move |cx| { while let Ok(buffer_id) = rx.recv().await { let buffer = buffer_store.update(cx, |buffer_store, cx| { buffer_store .wait_for_remote_buffer(buffer_id, cx) }); buffer_tx.send(buffer).await?; } anyhow::Ok(()) }); let forward_buffers = cx.background_spawn(async move { while let Ok(buffer) = buffer_rx.recv().await { let _ = grab_buffer_snapshot_tx .send((buffer.await?, LineHint::default())) .await; } anyhow::Ok(()) }); let (left, right) = futures::future::join( wait_for_remote_buffers, forward_buffers, ) .await; left?; right?; drop(guard); cancel_ongoing_search.abort(); anyhow::Ok(()) }) .await .log_err(); }) .boxed_local(); ( FindSearchCandidates::Remote, vec![issue_remote_buffers_request], ) } }; let should_find_all_matches = !tx.is_closed(); let _executor = executor.clone(); let worker_pool = executor.spawn(async move { let num_cpus = _executor.num_cpus(); assert!(num_cpus > 0); _executor .scoped(|scope| { let worker_count = (num_cpus - 1).max(1); for _ in 0..worker_count { let worker = Worker { query: query.clone(), open_buffers: open_buffers.clone(), candidates: candidate_searcher.clone(), find_all_matches_rx: find_all_matches_rx.clone(), }; scope.spawn(worker.run()); } drop(find_all_matches_rx); drop(candidate_searcher); }) .await; }); let (sorted_matches_tx, sorted_matches_rx) = unbounded(); // The caller of `into_handle` decides whether they're interested in all matches (files that matched + all matching ranges) or // just the files. *They are using the same stream as the guts of the project search do*. // This means that we cannot grab values off of that stream unless it's strictly needed for making a progress in project search. // // Grabbing buffer snapshots is only necessary when we're looking for all matches. If the caller decided that they're not interested // in all matches, running that task unconditionally would hinder caller's ability to observe all matching file paths. let buffer_snapshots = if should_find_all_matches { Some( Self::grab_buffer_snapshots( grab_buffer_snapshot_rx, find_all_matches_tx, sorted_matches_tx, cx.clone(), ) .boxed_local(), ) } else { drop(find_all_matches_tx); None }; let ensure_matches_are_reported_in_order = if should_find_all_matches { Some( Self::ensure_matched_ranges_are_reported_in_order(sorted_matches_rx, tx) .boxed_local(), ) } else { drop(tx); None }; futures::future::join_all( [worker_pool.boxed_local()] .into_iter() .chain(buffer_snapshots) .chain(ensure_matches_are_reported_in_order) .chain(tasks), ) .await; }) }); SearchResultsHandle { results: rx, matching_buffers, trigger_search, } } fn provide_search_paths( worktrees: Vec<Entity<Worktree>>, query: Arc<SearchQuery>, tx: Sender<InputPath>, results: Sender<oneshot::Receiver<(ProjectPath, LineHint)>>, results_tx: Sender<SearchResult>, ) -> impl AsyncFnOnce(&mut AsyncApp) { async move |cx| { _ = maybe!(async move { let gitignored_tracker = PathInclusionMatcher::new(query.clone()); let include_ignored = query.include_ignored(); for worktree in worktrees { let scan_complete = worktree.read_with(cx, |worktree, _| { worktree.as_local().map(|local| local.scan_complete()) }); if let Some(scan_complete) = scan_complete { let mut scan_complete = pin!(scan_complete); if scan_complete.as_mut().now_or_never().is_none() { _ = results_tx.send(SearchResult::WaitingForScan).await; scan_complete.await; _ = results_tx.send(SearchResult::Searching).await; } } let (mut snapshot, worktree_settings) = worktree .read_with(cx, |this, _| { Some((this.snapshot(), this.as_local()?.settings())) }) .context("The worktree is not local")?; if query.include_ignored() { // Pre-fetch all of the ignored directories as they're going to be searched. let mut entries_to_refresh = vec![]; for entry in snapshot.entries(query.include_ignored(), 0) { if gitignored_tracker.should_scan_gitignored_dir( entry, &snapshot, &worktree_settings, ) { entries_to_refresh.push(entry.path.clone()); } } let barrier = worktree.update(cx, |this, _| { let local = this.as_local_mut()?; let barrier = entries_to_refresh .into_iter() .map(|path| local.add_path_prefix_to_scan(path).into_future()) .collect::<Vec<_>>(); Some(barrier) }); if let Some(barriers) = barrier { futures::future::join_all(barriers).await; } snapshot = worktree.read_with(cx, |this, _| this.snapshot()); } let tx = tx.clone(); let results = results.clone(); cx.background_executor() .spawn(async move { for entry in snapshot.files(include_ignored, 0) { let (should_scan_tx, should_scan_rx) = oneshot::channel(); let Ok(_) = tx .send(InputPath { entry: entry.clone(), snapshot: snapshot.clone(), should_scan_tx, }) .await else { return; }; if results.send(should_scan_rx).await.is_err() { return; }; } }) .await; } anyhow::Ok(()) }) .await;Lines 1–500 of 1089. Load more
use std::{ cell::LazyCell, collections::BTreeSet, io::{BufRead, BufReader}, ops::Range, path::{Path, PathBuf}, pin::pin, sync::Arc, time::Duration, }; use anyhow::Context; use async_channel::{Receiver, Sender, bounded, unbounded}; use collections::HashSet; use fs::Fs; use futures::FutureExt as _; use futures::{SinkExt, StreamExt, select_biased, stream::FuturesOrdered}; use gpui::{App, AppContext, AsyncApp, BackgroundExecutor, Entity, Priority, Task}; use language::{Buffer, BufferSnapshot, Point}; use parking_lot::Mutex; use postage::oneshot; use rpc::{AnyProtoClient, proto}; use util::{ResultExt, maybe, paths::compare_rel_paths, rel_path::RelPath}; use worktree::{Entry, ProjectEntryId, Snapshot, Worktree, WorktreeSettings}; use crate::{ Project, ProjectItem, ProjectPath, RemotelyCreatedModels, buffer_store::BufferStore, search::{LineHint, SearchQuery, SearchResult}, worktree_store::WorktreeStore, }; pub struct Search { buffer_store: Entity<BufferStore>, worktree_store: Entity<WorktreeStore>, limit: usize, kind: SearchKind, } /// Represents search setup, before it is actually kicked off with Search::into_results enum SearchKind { /// Search for candidates by inspecting file contents on file system, avoiding loading the buffer unless we know that a given file contains a match. Local { fs: Arc<dyn Fs>, worktrees: Vec<Entity<Worktree>>, }, /// Query remote host for candidates. As of writing, the host runs a local search in "buffers with matches only" mode. Remote { client: AnyProtoClient, remote_id: u64, models: Arc<Mutex<RemotelyCreatedModels>>, }, /// Run search against a known set of candidates. Even when working with a remote host, this won't round-trip to host. OpenBuffersOnly, } /// Represents results of project search and allows one to either obtain match positions OR /// just the handles to buffers that may match the search. Grabbing the handles is cheaper than obtaining full match positions, because in that case we'll look for /// at most one match in each file. #[must_use] pub struct SearchResultsHandle { results: Receiver<SearchResult>, matching_buffers: Receiver<(Entity<Buffer>, LineHint)>, trigger_search: Box<dyn FnOnce(&mut App) -> Task<()> + Send + Sync>, } pub struct SearchResults<T> { pub task_handle: Task<()>, pub rx: Receiver<T>, } impl SearchResultsHandle { pub fn results(self, cx: &mut App) -> SearchResults<SearchResult> { SearchResults { task_handle: (self.trigger_search)(cx), rx: self.results, } } pub fn matching_buffers(self, cx: &mut App) -> SearchResults<(Entity<Buffer>, LineHint)> { SearchResults { task_handle: (self.trigger_search)(cx), rx: self.matching_buffers, } } } #[derive(Clone)] enum FindSearchCandidates { Local { fs: Arc<dyn Fs>, /// Start off with all paths in project and filter them based on: /// - Include filters /// - Exclude filters /// - Only open buffers /// - Scan ignored files /// Put another way: filter out files that can't match (without looking at file contents) input_paths_rx: Receiver<InputPath>, /// After that, if the buffer is not yet loaded, we'll figure out if it contains at least one match /// based on disk contents of a buffer. This step is not performed for buffers we already have in memory. confirm_contents_will_match_tx: Sender<MatchingEntry>, confirm_contents_will_match_rx: Receiver<MatchingEntry>, }, Remote, OpenBuffersOnly, } impl Search { pub fn local( fs: Arc<dyn Fs>, buffer_store: Entity<BufferStore>, worktree_store: Entity<WorktreeStore>, limit: usize, cx: &mut App, ) -> Self { let worktrees = worktree_store.read(cx).visible_worktrees(cx).collect(); Self { kind: SearchKind::Local { fs, worktrees }, buffer_store, worktree_store, limit, } } pub(crate) fn remote( buffer_store: Entity<BufferStore>, worktree_store: Entity<WorktreeStore>, limit: usize, client_state: (AnyProtoClient, u64, Arc<Mutex<RemotelyCreatedModels>>), ) -> Self { Self { kind: SearchKind::Remote { client: client_state.0, remote_id: client_state.1, models: client_state.2, }, buffer_store, worktree_store, limit, } } pub(crate) fn open_buffers_only( buffer_store: Entity<BufferStore>, worktree_store: Entity<WorktreeStore>, limit: usize, ) -> Self { Self { kind: SearchKind::OpenBuffersOnly, buffer_store, worktree_store, limit, } } pub(crate) const MAX_SEARCH_RESULT_FILES: usize = 5_000; pub const MAX_SEARCH_RESULT_RANGES: usize = 10_000; /// Prepares a project search run. The resulting [`SearchResultsHandle`] has to be used to specify whether you're interested in matching buffers /// or full search results. pub fn into_handle(mut self, query: SearchQuery, cx: &mut App) -> SearchResultsHandle { let mut open_buffers = HashSet::default(); let mut unnamed_buffers = Vec::new(); const MAX_CONCURRENT_BUFFER_OPENS: usize = 64; let buffers = self.buffer_store.read(cx); for handle in buffers.buffers() { let buffer = handle.read(cx); if !buffers.is_searchable(&buffer.remote_id()) { continue; } else if buffer .file() .is_some_and(|file| file.disk_state().is_deleted()) { continue; } else if let Some(entry_id) = buffer.entry_id(cx) { open_buffers.insert(entry_id); } else { self.limit = self.limit.saturating_sub(1); unnamed_buffers.push(handle) }; } let open_buffers = Arc::new(open_buffers); let executor = cx.background_executor().clone(); let (tx, rx) = unbounded(); let (grab_buffer_snapshot_tx, grab_buffer_snapshot_rx) = unbounded::<(Entity<Buffer>, LineHint)>(); let matching_buffers = grab_buffer_snapshot_rx.clone(); let trigger_search = Box::new(move |cx: &mut App| { cx.spawn(async move |cx| { for buffer in unnamed_buffers { _ = grab_buffer_snapshot_tx .send((buffer, LineHint::default())) .await; } let (find_all_matches_tx, find_all_matches_rx) = bounded(MAX_CONCURRENT_BUFFER_OPENS); let query = Arc::new(query); let (candidate_searcher, tasks) = match self.kind { SearchKind::OpenBuffersOnly => { let open_buffers = cx.update(|cx| self.all_loaded_buffers(&query, cx)); let fill_requests = cx .background_spawn(async move { for buffer in open_buffers { if let Err(_) = grab_buffer_snapshot_tx .send((buffer, LineHint::default())) .await { return; } } }) .boxed_local(); (FindSearchCandidates::OpenBuffersOnly, vec![fill_requests]) } SearchKind::Local { fs, ref mut worktrees, } => { let (get_buffer_for_full_scan_tx, get_buffer_for_full_scan_rx) = unbounded(); let (confirm_contents_will_match_tx, confirm_contents_will_match_rx) = bounded(64); let (sorted_search_results_tx, sorted_search_results_rx) = unbounded(); let (input_paths_tx, input_paths_rx) = unbounded(); let tasks = vec![ cx.spawn(Self::provide_search_paths( std::mem::take(worktrees), query.clone(), input_paths_tx, sorted_search_results_tx, tx.clone(), )) .boxed_local(), Self::open_buffers( self.buffer_store, get_buffer_for_full_scan_rx, grab_buffer_snapshot_tx, cx.clone(), ) .boxed_local(), cx.background_spawn(Self::maintain_sorted_search_results( sorted_search_results_rx, get_buffer_for_full_scan_tx, self.limit, )) .boxed_local(), ]; ( FindSearchCandidates::Local { fs, confirm_contents_will_match_tx, confirm_contents_will_match_rx, input_paths_rx, }, tasks, ) } SearchKind::Remote { client, remote_id, models, } => { let (handle, rx) = self .buffer_store .update(cx, |this, _| this.register_project_search_result_handle()); let cancel_ongoing_search = util::defer({ let client = client.clone(); move || { _ = client.send(proto::FindSearchCandidatesCancelled { project_id: remote_id, handle, }); } }); let request = client.request(proto::FindSearchCandidates { project_id: remote_id, query: Some(query.to_proto()), limit: self.limit as _, handle, }); let buffer_store = self.buffer_store; let guard = cx.update(|cx| { Project::retain_remotely_created_models_impl( &models, &buffer_store, &self.worktree_store, cx, ) }); let issue_remote_buffers_request = cx .spawn(async move |cx| { let _ = maybe!(async move { request.await?; let (buffer_tx, buffer_rx) = bounded(24); let wait_for_remote_buffers = cx.spawn(async move |cx| { while let Ok(buffer_id) = rx.recv().await { let buffer = buffer_store.update(cx, |buffer_store, cx| { buffer_store .wait_for_remote_buffer(buffer_id, cx) }); buffer_tx.send(buffer).await?; } anyhow::Ok(()) }); let forward_buffers = cx.background_spawn(async move { while let Ok(buffer) = buffer_rx.recv().await { let _ = grab_buffer_snapshot_tx .send((buffer.await?, LineHint::default())) .await; } anyhow::Ok(()) }); let (left, right) = futures::future::join( wait_for_remote_buffers, forward_buffers, ) .await; left?; right?; drop(guard); cancel_ongoing_search.abort(); anyhow::Ok(()) }) .await .log_err(); }) .boxed_local(); ( FindSearchCandidates::Remote, vec![issue_remote_buffers_request], ) } }; let should_find_all_matches = !tx.is_closed(); let _executor = executor.clone(); let worker_pool = executor.spawn(async move { let num_cpus = _executor.num_cpus(); assert!(num_cpus > 0); _executor .scoped(|scope| { let worker_count = (num_cpus - 1).max(1); for _ in 0..worker_count { let worker = Worker { query: query.clone(), open_buffers: open_buffers.clone(), candidates: candidate_searcher.clone(), find_all_matches_rx: find_all_matches_rx.clone(), }; scope.spawn(worker.run()); } drop(find_all_matches_rx); drop(candidate_searcher); }) .await; }); let (sorted_matches_tx, sorted_matches_rx) = unbounded(); // The caller of `into_handle` decides whether they're interested in all matches (files that matched + all matching ranges) or // just the files. *They are using the same stream as the guts of the project search do*. // This means that we cannot grab values off of that stream unless it's strictly needed for making a progress in project search. // // Grabbing buffer snapshots is only necessary when we're looking for all matches. If the caller decided that they're not interested // in all matches, running that task unconditionally would hinder caller's ability to observe all matching file paths. let buffer_snapshots = if should_find_all_matches { Some( Self::grab_buffer_snapshots( grab_buffer_snapshot_rx, find_all_matches_tx, sorted_matches_tx, cx.clone(), ) .boxed_local(), ) } else { drop(find_all_matches_tx); None }; let ensure_matches_are_reported_in_order = if should_find_all_matches { Some( Self::ensure_matched_ranges_are_reported_in_order(sorted_matches_rx, tx) .boxed_local(), ) } else { drop(tx); None }; futures::future::join_all( [worker_pool.boxed_local()] .into_iter() .chain(buffer_snapshots) .chain(ensure_matches_are_reported_in_order) .chain(tasks), ) .await; }) }); SearchResultsHandle { results: rx, matching_buffers, trigger_search, } } fn provide_search_paths( worktrees: Vec<Entity<Worktree>>, query: Arc<SearchQuery>, tx: Sender<InputPath>, results: Sender<oneshot::Receiver<(ProjectPath, LineHint)>>, results_tx: Sender<SearchResult>, ) -> impl AsyncFnOnce(&mut AsyncApp) { async move |cx| { _ = maybe!(async move { let gitignored_tracker = PathInclusionMatcher::new(query.clone()); let include_ignored = query.include_ignored(); for worktree in worktrees { let scan_complete = worktree.read_with(cx, |worktree, _| { worktree.as_local().map(|local| local.scan_complete()) }); if let Some(scan_complete) = scan_complete { let mut scan_complete = pin!(scan_complete); if scan_complete.as_mut().now_or_never().is_none() { _ = results_tx.send(SearchResult::WaitingForScan).await; scan_complete.await; _ = results_tx.send(SearchResult::Searching).await; } } let (mut snapshot, worktree_settings) = worktree .read_with(cx, |this, _| { Some((this.snapshot(), this.as_local()?.settings())) }) .context("The worktree is not local")?; if query.include_ignored() { // Pre-fetch all of the ignored directories as they're going to be searched. let mut entries_to_refresh = vec![]; for entry in snapshot.entries(query.include_ignored(), 0) { if gitignored_tracker.should_scan_gitignored_dir( entry, &snapshot, &worktree_settings, ) { entries_to_refresh.push(entry.path.clone()); } } let barrier = worktree.update(cx, |this, _| { let local = this.as_local_mut()?; let barrier = entries_to_refresh .into_iter() .map(|path| local.add_path_prefix_to_scan(path).into_future()) .collect::<Vec<_>>(); Some(barrier) }); if let Some(barriers) = barrier { futures::future::join_all(barriers).await; } snapshot = worktree.read_with(cx, |this, _| this.snapshot()); } let tx = tx.clone(); let results = results.clone(); cx.background_executor() .spawn(async move { for entry in snapshot.files(include_ignored, 0) { let (should_scan_tx, should_scan_rx) = oneshot::channel(); let Ok(_) = tx .send(InputPath { entry: entry.clone(), snapshot: snapshot.clone(), should_scan_tx, }) .await else { return; }; if results.send(should_scan_rx).await.is_err() { return; }; } }) .await; } anyhow::Ok(()) }) .await;