diff --git a/Cargo.lock b/Cargo.lock index 6878297d87..2e9cba4902 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -14080,6 +14080,7 @@ dependencies = [ "anyhow", "askpass", "async-channel 2.5.0", + "async-lock", "async-trait", "base64 0.22.1", "buffer_diff", diff --git a/crates/collab/tests/integration/git_tests.rs b/crates/collab/tests/integration/git_tests.rs index c0f7932b66..1443566112 100644 --- a/crates/collab/tests/integration/git_tests.rs +++ b/crates/collab/tests/integration/git_tests.rs @@ -21,7 +21,7 @@ use gpui::{ }; use project::{ ProjectPath, - git_store::{CommitDataState, Repository}, + git_store::{CommitDataState, MAX_CONCURRENT_BLOB_READS, Repository}, }; use rand::{SeedableRng, rngs::StdRng}; use serde_json::json; @@ -449,6 +449,87 @@ async fn test_project_diff(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) }); } +#[gpui::test] +async fn test_blob_read_rpcs_are_bounded_on_host( + executor: BackgroundExecutor, + cx_a: &mut TestAppContext, + cx_b: &mut TestAppContext, + cx_c: &mut TestAppContext, +) { + let mut server = TestServer::start(executor.clone()).await; + let client_a = server.create_client(cx_a, "user_a").await; + let client_b = server.create_client(cx_b, "user_b").await; + let client_c = server.create_client(cx_c, "user_c").await; + server + .create_room(&mut [(&client_a, cx_a), (&client_b, cx_b), (&client_c, cx_c)]) + .await; + let active_call_a = cx_a.read(ActiveCall::global); + + const FILE_COUNT: usize = MAX_CONCURRENT_BLOB_READS + 4; + let names = (0..FILE_COUNT) + .map(|index| format!("f{index:02}.txt")) + .collect::>(); + + let mut tree = serde_json::Map::new(); + tree.insert(".git".to_owned(), json!({})); + for (index, name) in names.iter().enumerate() { + tree.insert(name.clone(), json!(format!("new-{index:02}\n"))); + } + client_a + .fs() + .insert_tree(path!("/dir"), serde_json::Value::Object(tree)) + .await; + + let merge_base = names + .iter() + .enumerate() + .map(|(index, name)| (name.as_str(), format!("old-{index:02}\n"))) + .collect::>(); + let dot_git = Path::new(path!("/dir/.git")); + let oids = client_a + .fs() + .set_merge_base_content_for_repo(dot_git, &merge_base); + let gate = client_a.fs().install_blob_read_gate_for_repo(dot_git); + + let (project_local, worktree_id) = client_a.build_local_project(path!("/dir"), cx_a).await; + let project_id = active_call_a + .update(cx_a, |call, cx| { + call.share_project(project_local.clone(), cx) + }) + .await + .unwrap(); + let project_b = client_b.join_remote_project(project_id, cx_b).await; + let project_c = client_c.join_remote_project(project_id, cx_c).await; + executor.run_until_parked(); + + // Two guests exceed the cap combined; only the host's own limiter holds them to it. + let mut diffs = Vec::with_capacity(names.len() * 2); + for (project, cx) in [(&project_b, &mut *cx_b), (&project_c, &mut *cx_c)] { + let git_store = project.read_with(cx, |project, _| project.git_store().clone()); + let repo = project.read_with(cx, |project, cx| project.active_repository(cx).unwrap()); + for (index, name) in names.iter().enumerate() { + let buffer = project + .update(cx, |project, cx| { + project.open_buffer((worktree_id, rel_path(name)), cx) + }) + .await + .unwrap(); + diffs.push(git_store.update(cx, |git_store, cx| { + git_store.open_diff_since(Some(oids[index]), buffer, repo.clone(), cx) + })); + } + } + executor.run_until_parked(); + + assert_eq!(gate.peak_concurrent(), MAX_CONCURRENT_BLOB_READS); + + gate.open(); + executor.run_until_parked(); + for diff in diffs { + diff.await.unwrap(); + } +} + #[gpui::test] async fn test_remote_git_worktrees( executor: BackgroundExecutor, diff --git a/crates/fs/src/fake_git_repo.rs b/crates/fs/src/fake_git_repo.rs index 6bdffbed71..fae26d0a28 100644 --- a/crates/fs/src/fake_git_repo.rs +++ b/crates/fs/src/fake_git_repo.rs @@ -5,6 +5,7 @@ use anyhow::{Context as _, Result, bail}; use async_channel::Sender; use collections::{HashMap, HashSet}; use futures::FutureExt as _; +use futures::channel::oneshot; use futures::future::{self, BoxFuture, join_all}; use git::repository::GitCommitTemplate; use git::{ @@ -81,12 +82,14 @@ pub struct FakeGitRepositoryState { pub commit_data: HashMap, pub stash_entries: GitStash, pub commit_template: Option, + pub blob_read_gate: Option, } impl FakeGitRepositoryState { pub fn new(event_emitter: async_channel::Sender) -> Self { FakeGitRepositoryState { event_emitter, + blob_read_gate: None, head_contents: Default::default(), index_contents: Default::default(), unmerged_paths: Default::default(), @@ -112,6 +115,96 @@ impl FakeGitRepositoryState { } } +#[derive(Clone, Default, Debug)] +pub struct FakeBlobReadGate(Arc>); + +#[derive(Default, Debug)] +struct BlobReadGateState { + open: bool, + peak: usize, + next_id: u64, + waiters: Vec, +} + +#[derive(Debug)] +struct Waiter { + id: u64, + oid: Oid, + sender: oneshot::Sender<()>, +} + +impl FakeBlobReadGate { + async fn wait(&self, oid: Oid) { + let (_guard, receiver) = { + let mut inner = self.0.lock(); + if inner.open { + return; + } + let id = inner.next_id; + inner.next_id += 1; + let (sender, receiver) = oneshot::channel(); + inner.waiters.push(Waiter { id, oid, sender }); + inner.peak = inner.peak.max(inner.waiters.len()); + ( + WaiterGuard { + state: self.0.clone(), + id, + }, + receiver, + ) + }; + receiver.await.ok(); + } + + pub fn peak_concurrent(&self) -> usize { + self.0.lock().peak + } + + pub fn waiting(&self) -> usize { + self.0.lock().waiters.len() + } + + pub fn is_waiting(&self, oid: Oid) -> bool { + self.0.lock().waiters.iter().any(|waiter| waiter.oid == oid) + } + + pub fn release(&self, oid: Oid) -> bool { + let mut inner = self.0.lock(); + let Some(position) = inner.waiters.iter().position(|waiter| waiter.oid == oid) else { + return false; + }; + let waiter = inner.waiters.remove(position); + drop(inner); + waiter.sender.send(()).ok(); + true + } + + pub fn open(&self) { + let waiters = { + let mut inner = self.0.lock(); + inner.open = true; + std::mem::take(&mut inner.waiters) + }; + for waiter in waiters { + waiter.sender.send(()).ok(); + } + } +} + +struct WaiterGuard { + state: Arc>, + id: u64, +} + +impl Drop for WaiterGuard { + fn drop(&mut self) { + self.state + .lock() + .waiters + .retain(|waiter| waiter.id != self.id); + } +} + impl FakeGitRepository { fn with_state_async(&self, write: bool, f: F) -> BoxFuture<'static, Result> where @@ -171,9 +264,16 @@ impl GitRepository for FakeGitRepository { } fn load_blob_content(&self, oid: git::Oid) -> BoxFuture<'_, Result>> { - self.with_state_async(false, move |state| { - state.oids.get(&oid).cloned().context("oid does not exist") - }) + let content_and_gate = self.with_state_async(false, move |state| { + Ok((state.oids.get(&oid).cloned(), state.blob_read_gate.clone())) + }); + async move { + let (content, gate) = content_and_gate.await?; + if let Some(gate) = gate { + gate.wait(oid).await; + } + content.context("oid does not exist") + } .boxed() } diff --git a/crates/fs/src/fs.rs b/crates/fs/src/fs.rs index 120d6e26a3..167e63c98e 100644 --- a/crates/fs/src/fs.rs +++ b/crates/fs/src/fs.rs @@ -55,6 +55,8 @@ mod fake_git_repo; #[cfg(feature = "test-support")] use collections::{BTreeMap, btree_map}; #[cfg(feature = "test-support")] +pub use fake_git_repo::FakeBlobReadGate; +#[cfg(feature = "test-support")] use fake_git_repo::{FakeCommitDataEntry, FakeGitRepositoryState}; #[cfg(feature = "test-support")] use git::{ @@ -2444,20 +2446,32 @@ impl FakeFs { &self, dot_git: &Path, contents_by_path: &[(&str, String)], - ) { + ) -> Vec { self.with_git_state(dot_git, true, |state| { use git::Oid; state.merge_base_contents.clear(); - let oids = (1..) - .map(|n| n.to_string()) - .map(|n| Oid::from_bytes(n.repeat(20).as_bytes()).unwrap()); - for ((path, content), oid) in contents_by_path.iter().zip(oids) { + let mut assigned = Vec::with_capacity(contents_by_path.len()); + for (index, (path, content)) in contents_by_path.iter().enumerate() { + let mut bytes = [0u8; 20]; + bytes[..4].copy_from_slice(&((index as u32) + 1).to_be_bytes()); + let oid = Oid::from_bytes(&bytes).unwrap(); state.merge_base_contents.insert(repo_path(path), oid); state.oids.insert(oid, content.as_bytes().to_vec()); + assigned.push(oid); } + assigned + }) + .unwrap() + } + + pub fn install_blob_read_gate_for_repo(&self, dot_git: &Path) -> FakeBlobReadGate { + let gate = FakeBlobReadGate::default(); + self.with_git_state(dot_git, false, |state| { + state.blob_read_gate = Some(gate.clone()); }) .unwrap(); + gate } pub fn set_blame_for_repo(&self, dot_git: &Path, blames: Vec<(RepoPath, git::blame::Blame)>) { diff --git a/crates/git_ui/src/diff_multibuffer.rs b/crates/git_ui/src/diff_multibuffer.rs index dc51973759..6c9aeb35ca 100644 --- a/crates/git_ui/src/diff_multibuffer.rs +++ b/crates/git_ui/src/diff_multibuffer.rs @@ -10,6 +10,7 @@ use editor::{ EditorEvent, EditorSettings, SelectionEffects, SplittableEditor, actions::GoToHunk, multibuffer_context_lines, scroll::Autoscroll, }; +use futures::{FutureExt as _, StreamExt as _, stream}; use futures_lite::future::yield_now; use git::{repository::RepoPath, status::FileStatus}; use gpui::{ @@ -36,6 +37,10 @@ use workspace::{ }; use ztracing::instrument; +/// Loading every changed file at once makes the first one land no sooner than the +/// last, leaving a large diff on a spinner. Throughput flattens past this point. +const MAX_CONCURRENT_BUFFER_LOADS: usize = 16; + struct BufferSubscriptions { _diff: Entity, display_buffer: Entity, @@ -679,17 +684,28 @@ impl DiffMultibuffer { let mut buffers_to_fold = Vec::new(); - for (path_key, entry) in entries { - if let Some(loaded_buffer) = entry.load.await.log_err() { - // We might be lagging behind enough that all future entry.load futures are no longer pending. - // If that is the case, this task will never yield, starving the foreground thread of execution time. - yield_now().await; + // Ordered, so excerpts don't shift under the reader as later files arrive. + let mut loads = stream::iter(entries.into_iter().map(|(path_key, entry)| { + let diff_buffer_list::DiffBuffer { + repo_path, + file_status, + load, + } = entry; + load.map(move |loaded| (path_key, repo_path, file_status, loaded)) + })) + .buffered(MAX_CONCURRENT_BUFFER_LOADS); + + while let Some((path_key, repo_path, file_status, loaded)) = loads.next().await { + // Buffered loads can already be complete when we poll them, in which + // case this loop never awaits and starves the foreground thread. + yield_now().await; + if let Some(loaded_buffer) = loaded.log_err() { cx.update(|window, cx| { this.update(cx, |this, cx| { if let Some(buffer_id) = this.register_buffer( - entry.repo_path, + repo_path, path_key, - entry.file_status, + file_status, loaded_buffer.display_buffer, loaded_buffer.main_buffer, loaded_buffer.diff, diff --git a/crates/git_ui/src/project_diff.rs b/crates/git_ui/src/project_diff.rs index 5ef3c1a808..38d08718c8 100644 --- a/crates/git_ui/src/project_diff.rs +++ b/crates/git_ui/src/project_diff.rs @@ -996,9 +996,9 @@ mod tests { use buffer_diff::DiffHunkSecondaryStatus; use db::indoc; use editor::test::editor_test_context::{EditorTestContext, assert_state_with_diff}; - use gpui::TestAppContext; + use gpui::{TestAppContext, VisualTestContext}; use multi_buffer::PathKey; - use project::FakeFs; + use project::{FakeFs, git_store::MAX_CONCURRENT_BLOB_READS}; use serde_json::json; use settings::{DiffViewStyle, GitPanelGroupBy, GitPanelSortBy, SettingsStore}; use std::path::Path; @@ -1031,6 +1031,7 @@ mod tests { use zed_actions::git as git_actions; + use crate::branch_diff::BranchDiff; use crate::project_diff::{self, ProjectDiff}; #[test] @@ -1541,6 +1542,138 @@ mod tests { ); } + #[gpui::test(iterations = 10)] + async fn test_excerpts_are_ordered_by_path(cx: &mut TestAppContext) { + init_test(cx); + + const FILE_COUNT: usize = 20; + + let names = (0..FILE_COUNT) + .map(|index| format!("f{index:02}.txt")) + .collect::>(); + + let fs = FakeFs::new(cx.executor()); + let mut tree = serde_json::Map::new(); + tree.insert(".git".to_owned(), json!({})); + for (index, name) in names.iter().enumerate() { + tree.insert(name.clone(), json!(format!("new-{index:02}\n"))); + } + fs.insert_tree(path!("/project"), serde_json::Value::Object(tree)) + .await; + + let head = names + .iter() + .enumerate() + .map(|(index, name)| (name.as_str(), format!("old-{index:02}\n"))) + .collect::>(); + fs.set_head_and_index_for_repo(Path::new(path!("/project/.git")), &head); + + let project = Project::test(fs, [Path::new(path!("/project"))], cx).await; + let (multi_workspace, cx) = + cx.add_window_view(|window, cx| MultiWorkspace::test_new(project, window, cx)); + let workspace = multi_workspace.read_with(cx, |mw, _| mw.workspace().clone()); + cx.run_until_parked(); + + cx.focus(&workspace); + cx.update(|window, cx| { + window.dispatch_action(project_diff::Diff.boxed_clone(), cx); + }); + cx.run_until_parked(); + + let item = workspace.update(cx, |workspace, cx| { + workspace.active_item_as::(cx).unwrap() + }); + let editor = item.read_with(cx, |item, cx| item.editor(cx).read(cx).rhs_editor().clone()); + let text = editor.update(cx, |editor, cx| { + editor.buffer().read(cx).snapshot(cx).text() + }); + + // Unique per-file content, so appearance order is excerpt order. + let actual = text + .lines() + .filter(|line| line.starts_with("new-")) + .collect::>(); + let expected = (0..FILE_COUNT) + .map(|index| format!("new-{index:02}")) + .collect::>(); + assert_eq!(actual, expected); + } + + #[gpui::test] + async fn test_merge_base_loading_is_incremental(cx: &mut TestAppContext) { + init_test(cx); + + const FILE_COUNT: usize = MAX_CONCURRENT_BLOB_READS + 4; + + let names = (0..FILE_COUNT) + .map(|index| format!("f{index:02}.txt")) + .collect::>(); + + let project_root = Path::new(path!("/project")); + let dot_git = project_root.join(".git"); + + let fs = FakeFs::new(cx.executor()); + let mut tree = serde_json::Map::new(); + tree.insert(".git".to_owned(), json!({})); + for (index, name) in names.iter().enumerate() { + tree.insert(name.clone(), json!(format!("new-{index:02}\n"))); + } + fs.insert_tree(project_root, serde_json::Value::Object(tree)) + .await; + + let merge_base = names + .iter() + .enumerate() + .map(|(index, name)| (name.as_str(), format!("old-{index:02}\n"))) + .collect::>(); + let oids = fs.set_merge_base_content_for_repo(&dot_git, &merge_base); + let gate = fs.install_blob_read_gate_for_repo(&dot_git); + + let project = Project::test(fs, [project_root], cx).await; + project + .update(cx, |project, cx| project.git_scans_complete(cx)) + .await; + let (multi_workspace, cx) = + cx.add_window_view(|window, cx| MultiWorkspace::test_new(project.clone(), window, cx)); + let workspace = multi_workspace.read_with(cx, |mw, _| mw.workspace().clone()); + + let branch_diff = cx + .update(|window, cx| { + BranchDiff::new_with_default_branch(project.clone(), workspace, window, cx) + }) + .await + .unwrap(); + cx.run_until_parked(); + + assert!(gate.waiting() > 0 && gate.waiting() < FILE_COUNT); + + let editor = + branch_diff.read_with(cx, |diff, cx| diff.editor(cx).read(cx).rhs_editor().clone()); + let shown = |cx: &mut VisualTestContext| { + editor + .update(cx, |editor, cx| { + editor.buffer().read(cx).snapshot(cx).text() + }) + .lines() + .filter(|line| line.starts_with("new-")) + .map(str::to_owned) + .collect::>() + }; + + assert!(shown(cx).is_empty()); + + assert!(gate.release(oids[0])); + cx.run_until_parked(); + assert_eq!(shown(cx), vec!["new-00".to_owned()]); + + gate.open(); + cx.run_until_parked(); + let expected = (0..FILE_COUNT) + .map(|index| format!("new-{index:02}")) + .collect::>(); + assert_eq!(shown(cx), expected); + } + #[gpui::test] async fn test_go_to_prev_hunk_multibuffer(cx: &mut TestAppContext) { init_test(cx); diff --git a/crates/project/Cargo.toml b/crates/project/Cargo.toml index a0e6d1e418..20f6cd6abc 100644 --- a/crates/project/Cargo.toml +++ b/crates/project/Cargo.toml @@ -37,6 +37,7 @@ aho-corasick.workspace = true anyhow.workspace = true askpass.workspace = true async-channel.workspace = true +async-lock.workspace = true async-trait.workspace = true base64.workspace = true buffer_diff.workspace = true diff --git a/crates/project/src/git_store.rs b/crates/project/src/git_store.rs index aa539e1840..afcb419d1e 100644 --- a/crates/project/src/git_store.rs +++ b/crates/project/src/git_store.rs @@ -15,6 +15,7 @@ use crate::{ }; use anyhow::{Context as _, Result, anyhow, bail}; use askpass::{AskPassDelegate, EncryptedPassword, IKnowWhatIAmDoingAndIHaveReadTheDocs}; +use async_lock::Semaphore; use buffer_diff::{ BufferDiff, DiffHunk, DiffHunkSecondaryStatus, DiffOperations, PendingHunk, PendingSense, }; @@ -115,11 +116,14 @@ pub struct GitStore { diffs: HashMap>, buffer_ids_by_index_text_buffer_id: HashMap, shared_diffs: HashMap>, + blob_read_limiter: Arc, _subscriptions: Vec, } const MIN_PARKED_REPOSITORY_DEPTH: usize = 2; +pub const MAX_CONCURRENT_BLOB_READS: usize = 16; + #[derive(Clone, Debug)] pub struct ParkedRepository { worktree_id: WorktreeId, @@ -683,6 +687,7 @@ pub struct Repository { unshallow_state: UnshallowState, commit_message_buffer: Option>, git_store: WeakEntity, + blob_read_limiter: Arc, // For a local repository, holds paths that have had worktree events since the last status scan completed, // and that should be examined during the next status scan. paths_needing_status_update: Vec>, @@ -1019,6 +1024,7 @@ impl GitStore { _subscriptions, loading_diffs: HashMap::default(), shared_diffs: HashMap::default(), + blob_read_limiter: Arc::new(Semaphore::new(MAX_CONCURRENT_BLOB_READS)), diffs: HashMap::default(), buffer_ids_by_index_text_buffer_id: HashMap::default(), } @@ -2912,6 +2918,7 @@ impl GitStore { let id = RepositoryId(next_repository_id.fetch_add(1, atomic::Ordering::Release)); let git_store = cx.weak_entity(); + let blob_read_limiter = self.blob_read_limiter.clone(); let repo = cx.new(|cx| { let mut repo = Repository::local( id, @@ -2923,6 +2930,7 @@ impl GitStore { fs, is_trusted, git_store, + blob_read_limiter, cx, ); if let Some(updates_tx) = updates_tx.as_ref() { @@ -3391,6 +3399,7 @@ impl GitStore { .map(|p| Path::new(p).into()); let mut repo_subscription = None; + let blob_read_limiter = this.blob_read_limiter.clone(); let repo = this.repositories.entry(id).or_insert_with(|| { let git_store = cx.weak_entity(); let repo = cx.new(|cx| { @@ -3403,6 +3412,7 @@ impl GitStore { ProjectId(update.project_id), client, git_store, + blob_read_limiter, cx, ) }); @@ -6530,6 +6540,7 @@ impl Repository { fs: Arc, is_trusted: bool, git_store: WeakEntity, + blob_read_limiter: Arc, cx: &mut Context, ) -> Self { let snapshot = RepositorySnapshot::empty( @@ -6544,6 +6555,7 @@ impl Repository { let mut repo = Repository { this: cx.weak_entity(), git_store, + blob_read_limiter, snapshot, unshallow_state: UnshallowState::default(), pending_ops: Default::default(), @@ -6575,6 +6587,7 @@ impl Repository { project_id: ProjectId, client: AnyProtoClient, git_store: WeakEntity, + blob_read_limiter: Arc, cx: &mut Context, ) -> Self { let snapshot = RepositorySnapshot::empty( @@ -6597,6 +6610,7 @@ impl Repository { unshallow_state: UnshallowState::default(), commit_message_buffer: None, git_store, + blob_read_limiter, pending_ops: Default::default(), paths_needing_status_update: Default::default(), job_sender, @@ -10381,10 +10395,15 @@ impl Repository { }) } - fn load_blob_content(&mut self, oid: Oid, cx: &App) -> Task> { + /// Bypasses the serial git job queue: blobs are content-addressed, so this read + /// doesn't depend on the status snapshot, and a diff issues one per changed file. + fn load_blob_content(&self, oid: Oid, cx: &App) -> Task> { let repository_id = self.snapshot.id; - let rx = self.send_job("load_blob_content", None, move |state, _| async move { - match state { + let repository_state = self.repository_state.clone(); + let blob_read_limiter = self.blob_read_limiter.clone(); + cx.background_spawn(async move { + let _permit = blob_read_limiter.acquire_arc().await; + match repository_state.await.map_err(|err| anyhow::anyhow!(err))? { RepositoryState::Local(LocalRepositoryState { backend, .. }) => { decode_git_text(backend.load_blob_content(oid).await?) } @@ -10399,8 +10418,7 @@ impl Repository { Ok(response.content) } } - }); - cx.spawn(|_: &mut AsyncApp| async move { rx.await? }) + }) } fn paths_changed( @@ -11361,7 +11379,7 @@ impl Repository { mod tests { use super::*; use crate::Project; - use fs::{FakeFs, Fs}; + use fs::{FakeBlobReadGate, FakeFs, Fs}; use git::repository::{RepoPath, repo_path}; use gpui::proptest::prelude::*; use gpui::{TestAppContext, UpdateGlobal}; @@ -11876,6 +11894,95 @@ mod tests { assert!(weak_display_diff_list.upgrade().is_none()); } + async fn setup_gated_blob_reads( + cx: &mut TestAppContext, + count: usize, + ) -> (FakeBlobReadGate, Entity, Vec) { + use util::path; + + let project_root = Path::new(path!("/project")); + let dot_git = project_root.join(".git"); + let fs = FakeFs::new(cx.executor()); + fs.insert_tree(project_root, json!({ ".git": {} })).await; + + let entries = (0..count) + .map(|index| (format!("f{index:02}.txt"), format!("blob-{index:02}\n"))) + .collect::>(); + let entry_refs = entries + .iter() + .map(|(name, content)| (name.as_str(), content.clone())) + .collect::>(); + let oids = fs.set_merge_base_content_for_repo(&dot_git, &entry_refs); + let gate = fs.install_blob_read_gate_for_repo(&dot_git); + + let project = Project::test(fs, [project_root], cx).await; + project + .update(cx, |project, cx| project.git_scans_complete(cx)) + .await; + let repository = + project.read_with(cx, |project, cx| project.active_repository(cx).unwrap()); + (gate, repository, oids) + } + + #[gpui::test] + async fn test_blob_reads_are_bounded(cx: &mut TestAppContext) { + init_test(cx); + let (gate, repository, oids) = + setup_gated_blob_reads(cx, MAX_CONCURRENT_BLOB_READS + 4).await; + + let reads = oids + .iter() + .map(|oid| { + repository.update(cx, |repository, cx| repository.load_blob_content(*oid, cx)) + }) + .collect::>(); + cx.run_until_parked(); + + assert_eq!(gate.peak_concurrent(), MAX_CONCURRENT_BLOB_READS); + assert_eq!(gate.waiting(), MAX_CONCURRENT_BLOB_READS); + + gate.open(); + cx.run_until_parked(); + for read in reads { + read.await.unwrap(); + } + } + + #[gpui::test] + async fn test_cancelled_blob_read_releases_permit(cx: &mut TestAppContext) { + init_test(cx); + let (gate, repository, oids) = + setup_gated_blob_reads(cx, MAX_CONCURRENT_BLOB_READS + 1).await; + + let mut holding = oids[..MAX_CONCURRENT_BLOB_READS] + .iter() + .map(|oid| { + repository.update(cx, |repository, cx| repository.load_blob_content(*oid, cx)) + }) + .collect::>(); + cx.run_until_parked(); + assert_eq!(gate.waiting(), MAX_CONCURRENT_BLOB_READS); + + // One more read can't get a permit, so it never reaches the backend. + let blocked_oid = oids[MAX_CONCURRENT_BLOB_READS]; + let _blocked = repository.update(cx, |repository, cx| { + repository.load_blob_content(blocked_oid, cx) + }); + cx.run_until_parked(); + assert!(!gate.is_waiting(blocked_oid)); + + let cancelled_oid = oids[0]; + drop(holding.remove(0)); + cx.run_until_parked(); + assert!(gate.is_waiting(blocked_oid)); + assert!(!gate.is_waiting(cancelled_oid)); + assert_eq!(gate.waiting(), MAX_CONCURRENT_BLOB_READS); + assert_eq!(gate.peak_concurrent(), MAX_CONCURRENT_BLOB_READS); + + gate.open(); + cx.run_until_parked(); + } + #[gpui::test] async fn test_append_pattern_to_ignore_file_creates_and_deduplicates(cx: &mut TestAppContext) { let fs: Arc = FakeFs::new(cx.executor()); diff --git a/crates/project/src/git_store/diff_buffer_list.rs b/crates/project/src/git_store/diff_buffer_list.rs index ee0421b587..b338f69a4c 100644 --- a/crates/project/src/git_store/diff_buffer_list.rs +++ b/crates/project/src/git_store/diff_buffer_list.rs @@ -1,7 +1,7 @@ use anyhow::Result; use buffer_diff::BufferDiff; use collections::{HashMap, HashSet}; -use futures::StreamExt; +use futures::{FutureExt, StreamExt, future::LocalBoxFuture}; use git::{ repository::RepoPath, status::{DiffTreeType, FileStatus, StatusCode, TrackedStatus, TreeDiff, TreeDiffStatus}, @@ -390,8 +390,10 @@ impl DiffBufferList { repo: Entity, git_store: Entity, cx: &Context, - ) -> Task> { - cx.spawn(async move |_, cx| { + ) -> LocalBoxFuture<'static, Result> { + let mut cx = cx.to_async(); + async move { + let cx = &mut cx; let buffer = git_store .update(cx, |git_store, cx| { git_store.buffer_store.update(cx, |buffer_store, cx| { @@ -466,7 +468,8 @@ impl DiffBufferList { diff: changes, conflict_set, }) - }) + } + .boxed_local() } } @@ -536,9 +539,9 @@ pub struct LoadedDiffBuffer { pub conflict_set: Option>, } -#[derive(Debug)] pub struct DiffBuffer { pub repo_path: RepoPath, pub file_status: FileStatus, - pub load: Task>, + /// Not started until polled, so the consumer controls load concurrency. + pub load: LocalBoxFuture<'static, Result>, }