Repository: vercel/next.js
Pull request: Turbopack: avoid blocking tasks for I/O work

### What?

avoid using blocking tasks for IO work as we can't control the priority of blocking tasks.

### Why?

We already control concurrency of fs access via a semaphore which helps to ensure we won't tie up all tokio threads and we are generally suspicious that moving this blocking work to blocking threads is worthwhile

Changed files:
filename: turbopack/crates/turbo-tasks-fs/src/lib.rs
patch: @@ -493,11 +493,11 @@ impl DiskFileSystemInner {
         if !already_created {
             let func = |p: &Path| std::fs::create_dir_all(p);
             retry_blocking(directory.to_path_buf(), func)
-                .concurrency_limited(&self.write_semaphore)
                 .instrument(tracing::info_span!(
                     "create directory",
                     name = display(directory.display())
                 ))
+                .concurrency_limited(&self.write_semaphore)
                 .await?;
             ApplyEffectsContext::with(|fs_context: &mut DiskFileSystemApplyContext| {
                 fs_context
@@ -708,11 +708,11 @@ impl FileSystem for DiskFileSystem {
 
         let _lock = self.inner.lock_path(&full_path).await;
         let content = match retry_blocking(full_path.clone(), |path: &Path| File::from_path(path))
-            .concurrency_limited(&self.inner.read_semaphore)
             .instrument(tracing::info_span!(
                 "read file",
                 name = display(full_path.display())
             ))
+            .concurrency_limited(&self.inner.read_semaphore)
             .await
         {
             Ok(file) => FileContent::new(file),
@@ -834,11 +834,11 @@ impl FileSystem for DiskFileSystem {
         let _lock = self.inner.lock_path(&full_path).await;
         let link_path =
             match retry_blocking(full_path.clone(), |path: &Path| std::fs::read_link(path))
-                .concurrency_limited(&self.inner.read_semaphore)
                 .instrument(tracing::info_span!(
                     "read symlink",
                     name = display(full_path.display())
                 ))
+                .concurrency_limited(&self.inner.read_semaphore)
                 .await
             {
                 Ok(res) => res,
@@ -956,11 +956,11 @@ impl FileSystem for DiskFileSystem {
             // not wasting cycles.
             let compare = content
                 .streaming_compare(&full_path)
-                .concurrency_limited(&inner.read_semaphore)
                 .instrument(tracing::info_span!(
                     "read file before write",
                     name = display(full_path.display())
                 ))
+                .concurrency_limited(&inner.read_semaphore)
                 .await?;
             if compare == FileComparison::Equal {
                 if !old_invalidators.is_empty() {
@@ -1024,23 +1024,23 @@ impl FileSystem for DiskFileSystem {
                         }
                         Ok::<(), io::Error>(())
                     })
-                    .concurrency_limited(&inner.write_semaphore)
                     .instrument(tracing::info_span!(
                         "write file",
                         name = display(full_path.display())
                     ))
+                    .concurrency_limited(&inner.write_semaphore)
                     .await
                     .with_context(|| format!("failed to write to {}", full_path.display()))?;
                 }
                 FileContent::NotFound => {
                     retry_blocking(full_path.clone().into_owned(), |path| {
                         std::fs::remove_file(path)
                     })
-                    .concurrency_limited(&inner.write_semaphore)
                     .instrument(tracing::info_span!(
                         "remove file",
                         name = display(full_path.display())
                     ))
+                    .concurrency_limited(&inner.write_semaphore)
                     .await
                     .or_else(|err| {
                         if err.kind() == ErrorKind::NotFound {
@@ -1139,11 +1139,11 @@ impl FileSystem for DiskFileSystem {
             let old_content = match retry_blocking(full_path.clone().into_owned(), |path| {
                 std::fs::read_link(path)
             })
-            .concurrency_limited(&inner.read_semaphore)
             .instrument(tracing::info_span!(
                 "read symlink before write",
                 name = display(full_path.display())
             ))
+            .concurrency_limited(&inner.read_semaphore)
             .await
             {
                 Ok(res) => Some((res.is_absolute(), res)),
@@ -1195,13 +1195,19 @@ impl FileSystem for DiskFileSystem {
                         // fails with EEXIST if the link already exists instead of overwriting it.
                         // Windows has similar behavior with junction points.
                         remove_symbolic_link_dir_helper(&full_path)
+                            .instrument(tracing::info_span!(
+                                "remove existing symlink before write",
+                                name = display(full_path.display())
+                            ))
                             .concurrency_limited(&inner.write_semaphore)
                             .await
                             .with_context(|| {
                                 anyhow!("removing existing symlink {} failed", full_path.display())
                             })?;
                     }
 
+                    let span =
+                        tracing::info_span!("create symlink", name = display(full_path.display()));
                     retry_blocking(target.clone(), move |target_path| {
                         let _span = tracing::info_span!(
                             "write symlink",
@@ -1221,6 +1227,8 @@ impl FileSystem for DiskFileSystem {
                             }
                         }
                     })
+                    .instrument(span)
+                    .concurrency_limited(&inner.write_semaphore)
                     .await
                     .with_context(|| {
                         #[cfg(not(windows))]
@@ -1243,6 +1251,10 @@ impl FileSystem for DiskFileSystem {
                 }
                 OsSpecificLinkContent::NotFound => {
                     remove_symbolic_link_dir_helper(&full_path)
+                        .instrument(tracing::info_span!(
+                            "remove symlink",
+                            name = display(full_path.display())
+                        ))
                         .concurrency_limited(&inner.write_semaphore)
                         .await
                         .with_context(|| anyhow!("removing {} failed", full_path.display()))?;
@@ -1271,11 +1283,11 @@ impl FileSystem for DiskFileSystem {
 
         let _lock = self.inner.lock_path(&full_path).await;
         let meta = retry_blocking(full_path.clone(), |path| std::fs::metadata(path))
-            .concurrency_limited(&self.inner.read_semaphore)
             .instrument(tracing::info_span!(
                 "read metadata",
                 name = display(full_path.display())
             ))
+            .concurrency_limited(&self.inner.read_semaphore)
             .await
             .with_context(|| format!("reading metadata for {}", full_path.display()))?;
 

filename: turbopack/crates/turbo-tasks-fs/src/retry.rs
patch: @@ -1,7 +1,6 @@
 use std::{
     io::{self, ErrorKind},
     path::{Path, PathBuf},
-    thread::sleep,
     time::Duration,
 };
 
@@ -12,25 +11,22 @@ where
     F: Fn(&Path) -> io::Result<R> + Send + 'static,
     R: Send + 'static,
 {
-    turbo_tasks::spawn_blocking(move || {
-        let mut attempt = 1;
+    let mut attempt = 1;
 
-        loop {
-            return match func(&path) {
-                Ok(r) => Ok(r),
-                Err(err) => {
-                    if attempt < MAX_RETRY_ATTEMPTS && can_retry(&err) {
-                        sleep(get_retry_wait_time(attempt));
-                        attempt += 1;
-                        continue;
-                    }
-
-                    Err(err)
+    loop {
+        return match func(&path) {
+            Ok(r) => Ok(r),
+            Err(err) => {
+                if attempt < MAX_RETRY_ATTEMPTS && can_retry(&err) {
+                    tokio::time::sleep(get_retry_wait_time(attempt)).await;
+                    attempt += 1;
+                    continue;
                 }
-            };
-        }
-    })
-    .await
+
+                Err(err)
+            }
+        };
+    }
 }
 
 fn can_retry(err: &io::Error) -> bool {