1use std::{
2 cell::LazyCell,
3 collections::BTreeSet,
4 io::{BufRead, BufReader},
5 ops::Range,
6 path::{Path, PathBuf},
7 pin::pin,
8 sync::Arc,
9 time::Duration,
10};
11
12use anyhow::Context;
13use collections::HashSet;
14use fs::Fs;
15use futures::FutureExt as _;
16use futures::{SinkExt, StreamExt, select_biased, stream::FuturesOrdered};
17use gpui::{App, AppContext, AsyncApp, BackgroundExecutor, Entity, Priority, Task};
18use language::{Buffer, BufferSnapshot};
19use parking_lot::Mutex;
20use postage::oneshot;
21use rpc::{AnyProtoClient, proto};
22use smol::channel::{Receiver, Sender, bounded, unbounded};
23
24use util::{ResultExt, maybe, paths::compare_rel_paths, rel_path::RelPath};
25use worktree::{Entry, ProjectEntryId, Snapshot, Worktree, WorktreeSettings};
26
27use crate::{
28 Project, ProjectItem, ProjectPath, RemotelyCreatedModels,
29 buffer_store::BufferStore,
30 search::{SearchQuery, SearchResult},
31 worktree_store::WorktreeStore,
32};
33
34pub struct Search {
35 buffer_store: Entity<BufferStore>,
36 worktree_store: Entity<WorktreeStore>,
37 limit: usize,
38 kind: SearchKind,
39}
40
41/// Represents search setup, before it is actually kicked off with Search::into_results
42enum SearchKind {
43 /// Search for candidates by inspecting file contents on file system, avoiding loading the buffer unless we know that a given file contains a match.
44 Local {
45 fs: Arc<dyn Fs>,
46 worktrees: Vec<Entity<Worktree>>,
47 },
48 /// Query remote host for candidates. As of writing, the host runs a local search in "buffers with matches only" mode.
49 Remote {
50 client: AnyProtoClient,
51 remote_id: u64,
52 models: Arc<Mutex<RemotelyCreatedModels>>,
53 },
54 /// Run search against a known set of candidates. Even when working with a remote host, this won't round-trip to host.
55 OpenBuffersOnly,
56}
57
58/// Represents results of project search and allows one to either obtain match positions OR
59/// 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
60/// at most one match in each file.
61#[must_use]
62pub struct SearchResultsHandle {
63 results: Receiver<SearchResult>,
64 matching_buffers: Receiver<Entity<Buffer>>,
65 trigger_search: Box<dyn FnOnce(&mut App) -> Task<()> + Send + Sync>,
66}
67
68pub struct SearchResults<T> {
69 pub _task_handle: Task<()>,
70 pub rx: Receiver<T>,
71}
72impl SearchResultsHandle {
73 pub fn results(self, cx: &mut App) -> SearchResults<SearchResult> {
74 SearchResults {
75 _task_handle: (self.trigger_search)(cx),
76 rx: self.results,
77 }
78 }
79 pub fn matching_buffers(self, cx: &mut App) -> SearchResults<Entity<Buffer>> {
80 SearchResults {
81 _task_handle: (self.trigger_search)(cx),
82 rx: self.matching_buffers,
83 }
84 }
85}
86
87#[derive(Clone)]
88enum FindSearchCandidates {
89 Local {
90 fs: Arc<dyn Fs>,
91 /// Start off with all paths in project and filter them based on:
92 /// - Include filters
93 /// - Exclude filters
94 /// - Only open buffers
95 /// - Scan ignored files
96 /// Put another way: filter out files that can't match (without looking at file contents)
97 input_paths_rx: Receiver<InputPath>,
98 /// After that, if the buffer is not yet loaded, we'll figure out if it contains at least one match
99 /// based on disk contents of a buffer. This step is not performed for buffers we already have in memory.
100 confirm_contents_will_match_tx: Sender<MatchingEntry>,
101 confirm_contents_will_match_rx: Receiver<MatchingEntry>,
102 },
103 Remote,
104 OpenBuffersOnly,
105}
106
107impl Search {
108 pub fn local(
109 fs: Arc<dyn Fs>,
110 buffer_store: Entity<BufferStore>,
111 worktree_store: Entity<WorktreeStore>,
112 limit: usize,
113 cx: &mut App,
114 ) -> Self {
115 let worktrees = worktree_store.read(cx).visible_worktrees(cx).collect();
116 Self {
117 kind: SearchKind::Local { fs, worktrees },
118 buffer_store,
119 worktree_store,
120 limit,
121 }
122 }
123
124 pub(crate) fn remote(
125 buffer_store: Entity<BufferStore>,
126 worktree_store: Entity<WorktreeStore>,
127 limit: usize,
128 client_state: (AnyProtoClient, u64, Arc<Mutex<RemotelyCreatedModels>>),
129 ) -> Self {
130 Self {
131 kind: SearchKind::Remote {
132 client: client_state.0,
133 remote_id: client_state.1,
134 models: client_state.2,
135 },
136 buffer_store,
137 worktree_store,
138 limit,
139 }
140 }
141 pub(crate) fn open_buffers_only(
142 buffer_store: Entity<BufferStore>,
143 worktree_store: Entity<WorktreeStore>,
144 limit: usize,
145 ) -> Self {
146 Self {
147 kind: SearchKind::OpenBuffersOnly,
148 buffer_store,
149 worktree_store,
150 limit,
151 }
152 }
153
154 pub(crate) const MAX_SEARCH_RESULT_FILES: usize = 5_000;
155 pub(crate) const MAX_SEARCH_RESULT_RANGES: usize = 10_000;
156 /// Prepares a project search run. The resulting [`SearchResultsHandle`] has to be used to specify whether you're interested in matching buffers
157 /// or full search results.
158 pub fn into_handle(mut self, query: SearchQuery, cx: &mut App) -> SearchResultsHandle {
159 let mut open_buffers = HashSet::default();
160 let mut unnamed_buffers = Vec::new();
161 const MAX_CONCURRENT_BUFFER_OPENS: usize = 64;
162 let buffers = self.buffer_store.read(cx);
163 for handle in buffers.buffers() {
164 let buffer = handle.read(cx);
165 if !buffers.is_searchable(&buffer.remote_id()) {
166 continue;
167 } else if buffer
168 .file()
169 .is_some_and(|file| file.disk_state().is_deleted())
170 {
171 continue;
172 } else if let Some(entry_id) = buffer.entry_id(cx) {
173 open_buffers.insert(entry_id);
174 } else {
175 self.limit = self.limit.saturating_sub(1);
176 unnamed_buffers.push(handle)
177 };
178 }
179 let open_buffers = Arc::new(open_buffers);
180 let executor = cx.background_executor().clone();
181 let (tx, rx) = unbounded();
182 let (grab_buffer_snapshot_tx, grab_buffer_snapshot_rx) = unbounded();
183 let matching_buffers = grab_buffer_snapshot_rx.clone();
184 let trigger_search = Box::new(move |cx: &mut App| {
185 cx.spawn(async move |cx| {
186 for buffer in unnamed_buffers {
187 _ = grab_buffer_snapshot_tx.send(buffer).await;
188 }
189
190 let (find_all_matches_tx, find_all_matches_rx) =
191 bounded(MAX_CONCURRENT_BUFFER_OPENS);
192 let query = Arc::new(query);
193 let (candidate_searcher, tasks) = match self.kind {
194 SearchKind::OpenBuffersOnly => {
195 let open_buffers = cx.update(|cx| self.all_loaded_buffers(&query, cx));
196 let fill_requests = cx
197 .background_spawn(async move {
198 for buffer in open_buffers {
199 if let Err(_) = grab_buffer_snapshot_tx.send(buffer).await {
200 return;
201 }
202 }
203 })
204 .boxed_local();
205 (FindSearchCandidates::OpenBuffersOnly, vec![fill_requests])
206 }
207 SearchKind::Local {
208 fs,
209 ref mut worktrees,
210 } => {
211 let (get_buffer_for_full_scan_tx, get_buffer_for_full_scan_rx) =
212 unbounded();
213 let (confirm_contents_will_match_tx, confirm_contents_will_match_rx) =
214 bounded(64);
215 let (sorted_search_results_tx, sorted_search_results_rx) = unbounded();
216
217 let (input_paths_tx, input_paths_rx) = unbounded();
218 let tasks = vec![
219 cx.spawn(Self::provide_search_paths(
220 std::mem::take(worktrees),
221 query.clone(),
222 input_paths_tx,
223 sorted_search_results_tx,
224 ))
225 .boxed_local(),
226 Self::open_buffers(
227 self.buffer_store,
228 get_buffer_for_full_scan_rx,
229 grab_buffer_snapshot_tx,
230 cx.clone(),
231 )
232 .boxed_local(),
233 cx.background_spawn(Self::maintain_sorted_search_results(
234 sorted_search_results_rx,
235 get_buffer_for_full_scan_tx,
236 self.limit,
237 ))
238 .boxed_local(),
239 ];
240 (
241 FindSearchCandidates::Local {
242 fs,
243 confirm_contents_will_match_tx,
244 confirm_contents_will_match_rx,
245 input_paths_rx,
246 },
247 tasks,
248 )
249 }
250 SearchKind::Remote {
251 client,
252 remote_id,
253 models,
254 } => {
255 let (handle, rx) = self
256 .buffer_store
257 .update(cx, |this, _| this.register_project_search_result_handle());
258
259 let cancel_ongoing_search = util::defer({
260 let client = client.clone();
261 move || {
262 _ = client.send(proto::FindSearchCandidatesCancelled {
263 project_id: remote_id,
264 handle,
265 });
266 }
267 });
268 let request = client.request(proto::FindSearchCandidates {
269 project_id: remote_id,
270 query: Some(query.to_proto()),
271 limit: self.limit as _,
272 handle,
273 });
274
275 let buffer_store = self.buffer_store;
276 let guard = cx.update(|cx| {
277 Project::retain_remotely_created_models_impl(
278 &models,
279 &buffer_store,
280 &self.worktree_store,
281 cx,
282 )
283 });
284
285 let issue_remote_buffers_request = cx
286 .spawn(async move |cx| {
287 let _ = maybe!(async move {
288 request.await?;
289
290 let (buffer_tx, buffer_rx) = bounded(24);
291
292 let wait_for_remote_buffers = cx.spawn(async move |cx| {
293 while let Ok(buffer_id) = rx.recv().await {
294 let buffer =
295 buffer_store.update(cx, |buffer_store, cx| {
296 buffer_store
297 .wait_for_remote_buffer(buffer_id, cx)
298 });
299 buffer_tx.send(buffer).await?;
300 }
301 anyhow::Ok(())
302 });
303
304 let forward_buffers = cx.background_spawn(async move {
305 while let Ok(buffer) = buffer_rx.recv().await {
306 let _ =
307 grab_buffer_snapshot_tx.send(buffer.await?).await;
308 }
309 anyhow::Ok(())
310 });
311 let (left, right) = futures::future::join(
312 wait_for_remote_buffers,
313 forward_buffers,
314 )
315 .await;
316 left?;
317 right?;
318
319 drop(guard);
320 cancel_ongoing_search.abort();
321 anyhow::Ok(())
322 })
323 .await
324 .log_err();
325 })
326 .boxed_local();
327 (
328 FindSearchCandidates::Remote,
329 vec![issue_remote_buffers_request],
330 )
331 }
332 };
333
334 let should_find_all_matches = !tx.is_closed();
335
336 let _executor = executor.clone();
337 let worker_pool = executor.spawn(async move {
338 let num_cpus = _executor.num_cpus();
339
340 assert!(num_cpus > 0);
341 _executor
342 .scoped(|scope| {
343 let worker_count = (num_cpus - 1).max(1);
344 for _ in 0..worker_count {
345 let worker = Worker {
346 query: query.clone(),
347 open_buffers: open_buffers.clone(),
348 candidates: candidate_searcher.clone(),
349 find_all_matches_rx: find_all_matches_rx.clone(),
350 };
351 scope.spawn(worker.run());
352 }
353
354 drop(find_all_matches_rx);
355 drop(candidate_searcher);
356 })
357 .await;
358 });
359
360 let (sorted_matches_tx, sorted_matches_rx) = unbounded();
361 // The caller of `into_handle` decides whether they're interested in all matches (files that matched + all matching ranges) or
362 // just the files. *They are using the same stream as the guts of the project search do*.
363 // This means that we cannot grab values off of that stream unless it's strictly needed for making a progress in project search.
364 //
365 // Grabbing buffer snapshots is only necessary when we're looking for all matches. If the caller decided that they're not interested
366 // in all matches, running that task unconditionally would hinder caller's ability to observe all matching file paths.
367 let buffer_snapshots = if should_find_all_matches {
368 Some(
369 Self::grab_buffer_snapshots(
370 grab_buffer_snapshot_rx,
371 find_all_matches_tx,
372 sorted_matches_tx,
373 cx.clone(),
374 )
375 .boxed_local(),
376 )
377 } else {
378 drop(find_all_matches_tx);
379
380 None
381 };
382 let ensure_matches_are_reported_in_order = if should_find_all_matches {
383 Some(
384 Self::ensure_matched_ranges_are_reported_in_order(sorted_matches_rx, tx)
385 .boxed_local(),
386 )
387 } else {
388 drop(tx);
389 None
390 };
391
392 futures::future::join_all(
393 [worker_pool.boxed_local()]
394 .into_iter()
395 .chain(buffer_snapshots)
396 .chain(ensure_matches_are_reported_in_order)
397 .chain(tasks),
398 )
399 .await;
400 })
401 });
402
403 SearchResultsHandle {
404 results: rx,
405 matching_buffers,
406 trigger_search,
407 }
408 }
409
410 fn provide_search_paths(
411 worktrees: Vec<Entity<Worktree>>,
412 query: Arc<SearchQuery>,
413 tx: Sender<InputPath>,
414 results: Sender<oneshot::Receiver<ProjectPath>>,
415 ) -> impl AsyncFnOnce(&mut AsyncApp) {
416 async move |cx| {
417 _ = maybe!(async move {
418 let gitignored_tracker = PathInclusionMatcher::new(query.clone());
419 let include_ignored = query.include_ignored();
420 for worktree in worktrees {
421 let (mut snapshot, worktree_settings) = worktree
422 .read_with(cx, |this, _| {
423 Some((this.snapshot(), this.as_local()?.settings()))
424 })
425 .context("The worktree is not local")?;
426 if query.include_ignored() {
427 // Pre-fetch all of the ignored directories as they're going to be searched.
428 let mut entries_to_refresh = vec![];
429
430 for entry in snapshot.entries(query.include_ignored(), 0) {
431 if gitignored_tracker.should_scan_gitignored_dir(
432 entry,
433 &snapshot,
434 &worktree_settings,
435 ) {
436 entries_to_refresh.push(entry.path.clone());
437 }
438 }
439 let barrier = worktree.update(cx, |this, _| {
440 let local = this.as_local_mut()?;
441 let barrier = entries_to_refresh
442 .into_iter()
443 .map(|path| local.add_path_prefix_to_scan(path).into_future())
444 .collect::<Vec<_>>();
445 Some(barrier)
446 });
447 if let Some(barriers) = barrier {
448 futures::future::join_all(barriers).await;
449 }
450 snapshot = worktree.read_with(cx, |this, _| this.snapshot());
451 }
452 let tx = tx.clone();
453 let results = results.clone();
454
455 cx.background_executor()
456 .spawn(async move {
457 for entry in snapshot.files(include_ignored, 0) {
458 let (should_scan_tx, should_scan_rx) = oneshot::channel();
459
460 let Ok(_) = tx
461 .send(InputPath {
462 entry: entry.clone(),
463 snapshot: snapshot.clone(),
464 should_scan_tx,
465 })
466 .await
467 else {
468 return;
469 };
470 if results.send(should_scan_rx).await.is_err() {
471 return;
472 };
473 }
474 })
475 .await;
476 }
477 anyhow::Ok(())
478 })
479 .await;
480 }
481 }
482
483 async fn maintain_sorted_search_results(
484 rx: Receiver<oneshot::Receiver<ProjectPath>>,
485 paths_for_full_scan: Sender<ProjectPath>,
486 limit: usize,
487 ) {
488 let mut rx = pin!(rx);
489 let mut matched = 0;
490 while let Some(mut next_path_result) = rx.next().await {
491 let Some(successful_path) = next_path_result.next().await else {
492 // This file did not produce a match, hence skip it.
493 continue;
494 };
495 if paths_for_full_scan.send(successful_path).await.is_err() {
496 return;
497 };
498 matched += 1;
499 if matched >= limit {
500 break;
501 }
502 }
503 }
504
505 /// Background workers cannot open buffers by themselves, hence main thread will do it on their behalf.
506 async fn open_buffers(
507 buffer_store: Entity<BufferStore>,
508 rx: Receiver<ProjectPath>,
509 find_all_matches_tx: Sender<Entity<Buffer>>,
510 mut cx: AsyncApp,
511 ) {
512 let mut rx = pin!(rx.ready_chunks(64));
513 _ = maybe!(async move {
514 while let Some(requested_paths) = rx.next().await {
515 let mut buffers = buffer_store.update(&mut cx, |this, cx| {
516 requested_paths
517 .into_iter()
518 .map(|path| this.open_buffer(path, cx))
519 .collect::<FuturesOrdered<_>>()
520 });
521
522 while let Some(buffer) = buffers.next().await {
523 if let Some(buffer) = buffer.log_err() {
524 find_all_matches_tx.send(buffer).await?;
525 }
526 }
527 }
528 Result::<_, anyhow::Error>::Ok(())
529 })
530 .await;
531 }
532
533 async fn grab_buffer_snapshots(
534 rx: Receiver<Entity<Buffer>>,
535 find_all_matches_tx: Sender<(
536 Entity<Buffer>,
537 BufferSnapshot,
538 oneshot::Sender<(Entity<Buffer>, Vec<Range<language::Anchor>>)>,
539 )>,
540 results: Sender<oneshot::Receiver<(Entity<Buffer>, Vec<Range<language::Anchor>>)>>,
541 mut cx: AsyncApp,
542 ) {
543 _ = maybe!(async move {
544 while let Ok(buffer) = rx.recv().await {
545 let snapshot = buffer.read_with(&mut cx, |this, _| this.snapshot());
546 let (tx, rx) = oneshot::channel();
547 find_all_matches_tx.send((buffer, snapshot, tx)).await?;
548 results.send(rx).await?;
549 }
550 debug_assert!(rx.is_empty());
551 Result::<_, anyhow::Error>::Ok(())
552 })
553 .await;
554 }
555
556 async fn ensure_matched_ranges_are_reported_in_order(
557 rx: Receiver<oneshot::Receiver<(Entity<Buffer>, Vec<Range<language::Anchor>>)>>,
558 tx: Sender<SearchResult>,
559 ) {
560 use postage::stream::Stream;
561 _ = maybe!(async move {
562 let mut matched_buffers = 0;
563 let mut matches = 0;
564 while let Ok(mut next_buffer_matches) = rx.recv().await {
565 let Some((buffer, ranges)) = next_buffer_matches.recv().await else {
566 continue;
567 };
568
569 if matched_buffers > Search::MAX_SEARCH_RESULT_FILES
570 || matches > Search::MAX_SEARCH_RESULT_RANGES
571 {
572 _ = tx.send(SearchResult::LimitReached).await;
573 break;
574 }
575 matched_buffers += 1;
576 matches += ranges.len();
577
578 _ = tx.send(SearchResult::Buffer { buffer, ranges }).await?;
579 }
580 anyhow::Ok(())
581 })
582 .await;
583 }
584
585 fn all_loaded_buffers(&self, search_query: &SearchQuery, cx: &App) -> Vec<Entity<Buffer>> {
586 let worktree_store = self.worktree_store.read(cx);
587 let mut buffers = search_query
588 .buffers()
589 .into_iter()
590 .flatten()
591 .filter(|buffer| {
592 let b = buffer.read(cx);
593 if let Some(file) = b.file() {
594 if file.disk_state().is_deleted() {
595 return false;
596 }
597 if !search_query.match_path(file.path()) {
598 return false;
599 }
600 if !search_query.include_ignored()
601 && let Some(entry) = b
602 .entry_id(cx)
603 .and_then(|entry_id| worktree_store.entry_for_id(entry_id, cx))
604 && entry.is_ignored
605 {
606 return false;
607 }
608 }
609 true
610 })
611 .cloned()
612 .collect::<Vec<_>>();
613 buffers.sort_by(|a, b| {
614 let a = a.read(cx);
615 let b = b.read(cx);
616 match (a.file(), b.file()) {
617 (None, None) => a.remote_id().cmp(&b.remote_id()),
618 (None, Some(_)) => std::cmp::Ordering::Less,
619 (Some(_), None) => std::cmp::Ordering::Greater,
620 (Some(a), Some(b)) => compare_rel_paths((a.path(), true), (b.path(), true)),
621 }
622 });
623
624 buffers
625 }
626}
627
628struct Worker {
629 query: Arc<SearchQuery>,
630 open_buffers: Arc<HashSet<ProjectEntryId>>,
631 candidates: FindSearchCandidates,
632 /// Ok, we're back in background: run full scan & find all matches in a given buffer snapshot.
633 /// Then, when you're done, share them via the channel you were given.
634 find_all_matches_rx: Receiver<(
635 Entity<Buffer>,
636 BufferSnapshot,
637 oneshot::Sender<(Entity<Buffer>, Vec<Range<language::Anchor>>)>,
638 )>,
639}
640
641impl Worker {
642 async fn run(self) {
643 let (
644 input_paths_rx,
645 confirm_contents_will_match_rx,
646 mut confirm_contents_will_match_tx,
647 fs,
648 ) = match self.candidates {
649 FindSearchCandidates::Local {
650 fs,
651 input_paths_rx,
652 confirm_contents_will_match_rx,
653 confirm_contents_will_match_tx,
654 } => (
655 input_paths_rx,
656 confirm_contents_will_match_rx,
657 confirm_contents_will_match_tx,
658 Some(fs),
659 ),
660 FindSearchCandidates::Remote | FindSearchCandidates::OpenBuffersOnly => {
661 (unbounded().1, unbounded().1, unbounded().0, None)
662 }
663 };
664 // WorkerA: grabs a request for "find all matches in file/a" <- takes 5 minutes
665 // right after: WorkerB: grabs a request for "find all matches in file/b" <- takes 5 seconds
666 let mut find_all_matches = pin!(self.find_all_matches_rx.fuse());
667 let mut find_first_match = pin!(confirm_contents_will_match_rx.fuse());
668 let mut scan_path = pin!(input_paths_rx.fuse());
669
670 loop {
671 let handler = RequestHandler {
672 query: &self.query,
673 open_entries: &self.open_buffers,
674 fs: fs.as_deref(),
675 confirm_contents_will_match_tx: &confirm_contents_will_match_tx,
676 };
677 // Whenever we notice that some step of a pipeline is closed, we don't want to close subsequent
678 // steps straight away. Another worker might be about to produce a value that will
679 // be pushed there, thus we'll replace current worker's pipe with a dummy one.
680 // That way, we'll only ever close a next-stage channel when ALL workers do so.
681 select_biased! {
682 find_all_matches = find_all_matches.next() => {
683 let Some(matches) = find_all_matches else {
684 continue;
685 };
686 handler.handle_find_all_matches(matches).await;
687 },
688 find_first_match = find_first_match.next() => {
689 if let Some(buffer_with_at_least_one_match) = find_first_match {
690 handler.handle_find_first_match(buffer_with_at_least_one_match).await;
691 }
692 },
693 scan_path = scan_path.next() => {
694 if let Some(path_to_scan) = scan_path {
695 handler.handle_scan_path(path_to_scan).await;
696 } else {
697 // If we're the last worker to notice that this is not producing values, close the upstream.
698 confirm_contents_will_match_tx = bounded(1).0;
699 }
700
701 }
702 complete => {
703 break
704 },
705
706 }
707 }
708 }
709}
710
711struct RequestHandler<'worker> {
712 query: &'worker SearchQuery,
713 fs: Option<&'worker dyn Fs>,
714 open_entries: &'worker HashSet<ProjectEntryId>,
715 confirm_contents_will_match_tx: &'worker Sender<MatchingEntry>,
716}
717
718impl RequestHandler<'_> {
719 async fn handle_find_all_matches(
720 &self,
721 (buffer, snapshot, mut report_matches): (
722 Entity<Buffer>,
723 BufferSnapshot,
724 oneshot::Sender<(Entity<Buffer>, Vec<Range<language::Anchor>>)>,
725 ),
726 ) {
727 let ranges = self
728 .query
729 .search(&snapshot, None)
730 .await
731 .iter()
732 .map(|range| snapshot.anchor_before(range.start)..snapshot.anchor_after(range.end))
733 .collect::<Vec<_>>();
734
735 _ = report_matches.send((buffer, ranges)).await;
736 }
737
738 async fn handle_find_first_match(&self, mut entry: MatchingEntry) {
739 async move {
740 let abs_path = entry.worktree_root.join(entry.path.path.as_std_path());
741 let Some(file) = self
742 .fs
743 .context("Trying to query filesystem in remote project search")?
744 .open_sync(&abs_path)
745 .await
746 .log_err()
747 else {
748 return anyhow::Ok(());
749 };
750
751 let mut file = BufReader::new(file);
752 let file_start = file.fill_buf()?;
753
754 if let Err(Some(starting_position)) =
755 std::str::from_utf8(file_start).map_err(|e| e.error_len())
756 {
757 // Before attempting to match the file content, throw away files that have invalid UTF-8 sequences early on;
758 // That way we can still match files in a streaming fashion without having look at "obviously binary" files.
759 log::debug!(
760 "Invalid UTF-8 sequence in file {abs_path:?} \
761 at byte position {starting_position}"
762 );
763 return Ok(());
764 }
765
766 if self.query.detect(file).await.unwrap_or(false) {
767 // Yes, we should scan the whole file.
768 entry.should_scan_tx.send(entry.path).await?;
769 }
770 Ok(())
771 }
772 .await
773 .ok();
774 }
775
776 async fn handle_scan_path(&self, req: InputPath) {
777 _ = maybe!(async move {
778 let InputPath {
779 entry,
780 snapshot,
781 mut should_scan_tx,
782 } = req;
783
784 if entry.is_fifo || !entry.is_file() {
785 return Ok(());
786 }
787
788 if self.query.filters_path() {
789 let matched_path = if self.query.match_full_paths() {
790 let mut full_path = snapshot.root_name().to_owned();
791 full_path.push(&entry.path);
792 self.query.match_path(&full_path)
793 } else {
794 self.query.match_path(&entry.path)
795 };
796 if !matched_path {
797 return Ok(());
798 }
799 }
800
801 if self.open_entries.contains(&entry.id) {
802 // The buffer is already in memory and that's the version we want to scan;
803 // hence skip the dilly-dally and look for all matches straight away.
804 should_scan_tx
805 .send(ProjectPath {
806 worktree_id: snapshot.id(),
807 path: entry.path.clone(),
808 })
809 .await?;
810 } else {
811 self.confirm_contents_will_match_tx
812 .send(MatchingEntry {
813 should_scan_tx: should_scan_tx,
814 worktree_root: snapshot.abs_path().clone(),
815 path: ProjectPath {
816 worktree_id: snapshot.id(),
817 path: entry.path.clone(),
818 },
819 })
820 .await?;
821 }
822
823 anyhow::Ok(())
824 })
825 .await;
826 }
827}
828
829struct InputPath {
830 entry: Entry,
831 snapshot: Snapshot,
832 should_scan_tx: oneshot::Sender<ProjectPath>,
833}
834
835struct MatchingEntry {
836 worktree_root: Arc<Path>,
837 path: ProjectPath,
838 should_scan_tx: oneshot::Sender<ProjectPath>,
839}
840
841/// This struct encapsulates the logic to decide whether a given gitignored directory should be
842/// scanned based on include/exclude patterns of a search query (as include/exclude parameters may match paths inside it).
843/// It is kind-of doing an inverse of glob. Given a glob pattern like `src/**/` and a parent path like `src`, we need to decide whether the parent
844/// may contain glob hits.
845pub struct PathInclusionMatcher {
846 included: BTreeSet<PathBuf>,
847 query: Arc<SearchQuery>,
848}
849
850impl PathInclusionMatcher {
851 pub fn new(query: Arc<SearchQuery>) -> Self {
852 let mut included = BTreeSet::new();
853 // To do an inverse glob match, we split each glob into it's prefix and the glob part.
854 // For example, `src/**/*.rs` becomes `src/` and `**/*.rs`. The glob part gets dropped.
855 // Then, when checking whether a given directory should be scanned, we check whether it is a non-empty substring of any glob prefix.
856 if query.filters_path() {
857 included.extend(
858 query
859 .files_to_include()
860 .sources()
861 .flat_map(|glob| Some(wax::Glob::new(glob).ok()?.partition().0)),
862 );
863 }
864 Self { included, query }
865 }
866
867 pub fn should_scan_gitignored_dir(
868 &self,
869 entry: &Entry,
870 snapshot: &Snapshot,
871 worktree_settings: &WorktreeSettings,
872 ) -> bool {
873 if !entry.is_ignored || !entry.kind.is_unloaded() {
874 return false;
875 }
876 if !self.query.include_ignored() {
877 return false;
878 }
879 if worktree_settings.is_path_excluded(&entry.path) {
880 return false;
881 }
882 if !self.query.filters_path() {
883 return true;
884 }
885
886 let as_abs_path = LazyCell::new(move || snapshot.absolutize(&entry.path));
887 let entry_path = &entry.path;
888 // 3. Check Exclusions (Pruning)
889 // If the current path is a child of an excluded path, we stop.
890 let is_excluded = self.path_is_definitely_excluded(&entry_path, snapshot);
891
892 if is_excluded {
893 return false;
894 }
895
896 // 4. Check Inclusions (Traversal)
897 if self.included.is_empty() {
898 return true;
899 }
900
901 // We scan if the current path is a descendant of an include prefix
902 // OR if the current path is an ancestor of an include prefix (we need to go deeper to find it).
903 let is_included = self.included.iter().any(|prefix| {
904 let (prefix_matches_entry, entry_matches_prefix) = if prefix.is_absolute() {
905 (
906 prefix.starts_with(&**as_abs_path),
907 as_abs_path.starts_with(prefix),
908 )
909 } else {
910 RelPath::new(prefix, snapshot.path_style()).map_or((false, false), |prefix| {
911 (
912 prefix.starts_with(entry_path),
913 entry_path.starts_with(&prefix),
914 )
915 })
916 };
917
918 // Logic:
919 // 1. entry_matches_prefix: We are inside the target zone (e.g. glob: src/, current: src/lib/). Keep scanning.
920 // 2. prefix_matches_entry: We are above the target zone (e.g. glob: src/foo/, current: src/). Keep scanning to reach foo.
921 prefix_matches_entry || entry_matches_prefix
922 });
923
924 is_included
925 }
926 fn path_is_definitely_excluded(&self, path: &RelPath, snapshot: &Snapshot) -> bool {
927 if !self.query.files_to_exclude().sources().next().is_none() {
928 let mut path = if self.query.match_full_paths() {
929 let mut full_path = snapshot.root_name().to_owned();
930 full_path.push(path);
931 full_path
932 } else {
933 path.to_owned()
934 };
935 loop {
936 if self.query.files_to_exclude().is_match(&path) {
937 return true;
938 } else if !path.pop() {
939 return false;
940 }
941 }
942 } else {
943 false
944 }
945 }
946}
947
948type IsTerminating = bool;
949/// Adaptive batcher that starts eager (small batches) and grows batch size
950/// when items arrive quickly, reducing RPC overhead while preserving low latency
951/// for slow streams.
952pub struct AdaptiveBatcher<T> {
953 items: Sender<T>,
954 flush_batch: Sender<IsTerminating>,
955 _batch_task: Task<()>,
956}
957
958impl<T: 'static + Send> AdaptiveBatcher<T> {
959 pub fn new(cx: &BackgroundExecutor) -> (Self, Receiver<Vec<T>>) {
960 let (items, rx) = unbounded();
961 let (batch_tx, batch_rx) = unbounded();
962 let (flush_batch_tx, flush_batch_rx) = unbounded();
963 let flush_batch = flush_batch_tx.clone();
964 let executor = cx.clone();
965 let _batch_task = cx.spawn_with_priority(gpui::Priority::High, async move {
966 let mut current_batch = vec![];
967 let mut items_produced_so_far = 0_u64;
968
969 let mut _schedule_flush_after_delay: Option<Task<()>> = None;
970 let _time_elapsed_since_start_of_search = std::time::Instant::now();
971 let mut flush = pin!(flush_batch_rx);
972 let mut terminating = false;
973 loop {
974 select_biased! {
975 item = rx.recv().fuse() => {
976 match item {
977 Ok(new_item) => {
978 let is_fresh_batch = current_batch.is_empty();
979 items_produced_so_far += 1;
980 current_batch.push(new_item);
981 if is_fresh_batch {
982 // Chosen arbitrarily based on some experimentation with plots.
983 let desired_duration_ms = (20 * (items_produced_so_far + 2).ilog2() as u64).min(300);
984 let desired_duration = Duration::from_millis(desired_duration_ms);
985 let _executor = executor.clone();
986 let _flush = flush_batch_tx.clone();
987 let new_timer = executor.spawn_with_priority(Priority::High, async move {
988 _executor.timer(desired_duration).await;
989 _ = _flush.send(false).await;
990 });
991 _schedule_flush_after_delay = Some(new_timer);
992 }
993 }
994 Err(_) => {
995 // Items channel closed - send any remaining batch before exiting
996 if !current_batch.is_empty() {
997 _ = batch_tx.send(std::mem::take(&mut current_batch)).await;
998 }
999 break;
1000 }
1001 }
1002 }
1003 should_break_afterwards = flush.next() => {
1004 if !current_batch.is_empty() {
1005 _ = batch_tx.send(std::mem::take(&mut current_batch)).await;
1006 _schedule_flush_after_delay = None;
1007 }
1008 if should_break_afterwards.unwrap_or_default() {
1009 terminating = true;
1010 }
1011 }
1012 complete => {
1013 break;
1014 }
1015 }
1016 if terminating {
1017 // Drain any remaining items before exiting
1018 while let Ok(new_item) = rx.try_recv() {
1019 current_batch.push(new_item);
1020 }
1021 if !current_batch.is_empty() {
1022 _ = batch_tx.send(std::mem::take(&mut current_batch)).await;
1023 }
1024 break;
1025 }
1026 }
1027 });
1028 let this = Self {
1029 items,
1030 _batch_task,
1031 flush_batch,
1032 };
1033 (this, batch_rx)
1034 }
1035
1036 pub async fn push(&self, item: T) {
1037 _ = self.items.send(item).await;
1038 }
1039
1040 pub async fn flush(self) {
1041 _ = self.flush_batch.send(true).await;
1042 self._batch_task.await;
1043 }
1044}