diff --git a/Cargo.lock b/Cargo.lock index 59936218079..b2bdf0675f0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -22865,6 +22865,7 @@ dependencies = [ "encoding_rs", "fs", "futures 0.3.32", + "futures-lite 1.13.0", "fuzzy", "git", "gpui", diff --git a/crates/worktree/Cargo.toml b/crates/worktree/Cargo.toml index 5a2b6204e99..6ba5c1ebe0c 100644 --- a/crates/worktree/Cargo.toml +++ b/crates/worktree/Cargo.toml @@ -38,6 +38,7 @@ collections.workspace = true encoding_rs.workspace = true fs.workspace = true futures.workspace = true +futures-lite.workspace = true fuzzy.workspace = true git.workspace = true gpui.workspace = true diff --git a/crates/worktree/src/worktree.rs b/crates/worktree/src/worktree.rs index 02e3b2c38cb..16e5f480cea 100644 --- a/crates/worktree/src/worktree.rs +++ b/crates/worktree/src/worktree.rs @@ -19,6 +19,7 @@ use futures::{ select_biased, stream, task::Poll, }; +use futures_lite::future::yield_now; use fuzzy::CharBag; use git::{ BISECT_LOG, COMMIT_MESSAGE, DOT_GIT, FETCH_HEAD, FSMONITOR_DAEMON, GC_PID, GITIGNORE, @@ -7196,6 +7197,40 @@ fn read_file_header(file: &mut dyn Read, abs_path: &Path) -> Result<(Vec, bo Ok((header, reached_eof)) } +const STREAM_BLOCK_BYTES: usize = 1024 * 1024; + +async fn read_file_to_end( + file: &mut (dyn Read + Send), + content: &mut Vec, + abs_path: &Path, +) -> Result<()> { + let mut buf = vec![0u8; STREAM_BLOCK_BYTES]; + loop { + let mut block_len = 0; + while block_len < buf.len() { + let n = file + .read(&mut buf[block_len..]) + .with_context(|| format!("reading remaining bytes of the file {abs_path:?}"))?; + if n == 0 { + break; + } + block_len += n; + } + + if block_len == 0 { + break; + } + + content.extend_from_slice(&buf[..block_len]); + if block_len < buf.len() { + break; + } + + yield_now().await; + } + Ok(()) +} + pub async fn decode_file_text( fs: &dyn Fs, abs_path: &Path, @@ -7215,22 +7250,12 @@ pub async fn decode_file_text( // If the file is eligible for opening, read the rest of the file. let mut content = file_first_bytes; if !reached_eof { - let mut buf = [0u8; 8 * 1024]; - loop { - let n = file - .read(&mut buf) - .with_context(|| format!("reading remaining bytes of the file {abs_path:?}"))?; - if n == 0 { - break; - } - content.extend_from_slice(&buf[..n]); - } + read_file_to_end(&mut *file, &mut content, abs_path).await?; } let decoded = decode_text(content)?; Ok((decoded.text, decoded.encoding, decoded.has_bom)) } -const STREAM_BLOCK_BYTES: usize = 1024 * 1024; /// Reads and decodes a file straight into a [`Rope`]. /// The returned rope has already had its line endings normalized, the /// [`LineEnding`] detected before normalizing is returned alongside it. @@ -7255,7 +7280,7 @@ pub async fn decode_file_text_to_rope( if bom_encoding.is_none() && byte_content == ByteContent::Unknown && let Some((rope, line_ending)) = - stream_utf8_into_rope(&mut *file, prefix, reached_eof, abs_path)? + stream_utf8_into_rope(&mut *file, prefix, reached_eof, abs_path).await? { return Ok((rope, line_ending, encoding_rs::UTF_8, false)); } @@ -7273,8 +7298,8 @@ pub async fn decode_file_text_to_rope( /// Returns `None` if the file turns out not to be plain UTF-8, in which case the /// caller re-reads it and decodes it the slow way. `prefix` is the portion of /// the file already consumed from `file` for encoding detection. -fn stream_utf8_into_rope( - file: &mut dyn Read, +async fn stream_utf8_into_rope( + file: &mut (dyn Read + Send), prefix: Vec, reached_eof: bool, abs_path: &Path, @@ -7338,6 +7363,8 @@ fn stream_utf8_into_rope( if eof { break; } + + yield_now().await; } // At EOF everything should have been consumed. Anything left over is a @@ -7393,33 +7420,63 @@ mod tests { /// Streams `bytes` the way `decode_file_text_to_rope` would, returning the /// decoded text and detected line ending, or `None` if the fast path bailed. - fn stream(bytes: &[u8]) -> Option<(String, LineEnding)> { + async fn stream(bytes: &[u8]) -> Option<(String, LineEnding)> { let mut reader = std::io::Cursor::new(bytes.to_vec()); stream_utf8_into_rope(&mut reader, Vec::new(), false, Path::new("test")) + .await .unwrap() .map(|(rope, line_ending)| (rope.to_string(), line_ending)) } #[test] - fn test_stream_utf8_normalizes_line_endings() { + fn test_stream_utf8_yields_between_blocks() { + let mut reader = std::io::Cursor::new(vec![b'a'; STREAM_BLOCK_BYTES * 2]); + + assert!( + stream_utf8_into_rope(&mut reader, Vec::new(), false, Path::new("test")) + .now_or_never() + .is_none() + ); + assert_eq!(reader.position(), STREAM_BLOCK_BYTES as u64); + } + + #[test] + fn test_file_reading_yields_between_blocks() { + let mut reader = std::io::Cursor::new(vec![b'a'; STREAM_BLOCK_BYTES * 2]); + let mut content = Vec::new(); + + assert!( + read_file_to_end(&mut reader, &mut content, Path::new("test")) + .now_or_never() + .is_none() + ); + assert_eq!(reader.position(), STREAM_BLOCK_BYTES as u64); + assert_eq!(content.len(), STREAM_BLOCK_BYTES); + } + + #[gpui::test] + async fn test_stream_utf8_normalizes_line_endings() { let crlf = "one\r\ntwo\r\nthree\r\n".repeat(40); - let (text, line_ending) = stream(crlf.as_bytes()).unwrap(); + let (text, line_ending) = stream(crlf.as_bytes()).await.unwrap(); assert_eq!(text, crlf.replace("\r\n", "\n")); assert_eq!(line_ending, LineEnding::Windows); let cr = "one\rtwo\rthree\r".repeat(40); - assert_eq!(stream(cr.as_bytes()).unwrap().0, cr.replace('\r', "\n")); + assert_eq!( + stream(cr.as_bytes()).await.unwrap().0, + cr.replace('\r', "\n") + ); } - #[test] - fn test_stream_utf8_block_boundaries() { + #[gpui::test] + async fn test_stream_utf8_block_boundaries() { // A carriage return landing on the last byte of a block, with and // without its newline arriving in the next one. for (suffix, expected) in [("\r\ntail\n", "\ntail\n"), ("\rtail", "\ntail")] { let filler = "a".repeat(STREAM_BLOCK_BYTES - 1); let source = format!("{filler}{suffix}"); assert_eq!( - stream(source.as_bytes()).unwrap().0, + stream(source.as_bytes()).await.unwrap().0, format!("{filler}{expected}"), "suffix = {suffix:?}" ); @@ -7431,7 +7488,7 @@ mod tests { let filler = "a".repeat(STREAM_BLOCK_BYTES - split); let source = format!("{filler}{ch}tail"); assert_eq!( - stream(source.as_bytes()).unwrap().0, + stream(source.as_bytes()).await.unwrap().0, source, "ch = {ch:?}, split = {split}" ); @@ -7439,13 +7496,13 @@ mod tests { } } - #[test] - fn test_stream_utf8_falls_back_on_non_utf8() { + #[gpui::test] + async fn test_stream_utf8_falls_back_on_non_utf8() { // Each of these must bail so the caller re-reads and decodes the slow // way, rather than silently mangling the file. - assert_eq!(stream(b"hello \xff\xfeA"), None, "invalid utf-8"); - assert_eq!(stream(b"hello \xe2\x82"), None, "truncated at eof"); - assert_eq!(stream(b"plain \x1b$B text"), None, "iso-2022 escape"); + assert_eq!(stream(b"hello \xff\xfeA").await, None, "invalid utf-8"); + assert_eq!(stream(b"hello \xe2\x82").await, None, "truncated at eof"); + assert_eq!(stream(b"plain \x1b$B text").await, None, "iso-2022 escape"); } /// reproduction of issue #50785