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