Repository: vercel/next.js
Pull request: Batch ipc info messages to improve performance.

Batch the dependency messages sent from our nodejs worker process.

Some large projects using tailwind may report thousands of dependencies which can cause the node process to hang due to memory pressure.  We are addressing the memory pressure issue independently in #78462, but by reducing the number of small packets we can also improve performance.



Closes PACK-4447

Changed files:
filename: turbopack/crates/turbopack-node/js/src/ipc/index.ts
patch: @@ -95,6 +95,8 @@ function createIpc<TIncoming, TOutgoing>(
     process.exit(0);
   });
 
+  // TODO(lukesandberg): some of the messages being sent are very large and contain lots
+  //  of redundant information.  Consider adding gzip compression to our stream.
   function send(message: any): Promise<void> {
     // Reserve 4 bytes for our length prefix, we will over-write after encoding.
     const packet = Buffer.from("0000" + JSON.stringify(message), "utf8");

filename: turbopack/crates/turbopack-node/js/src/transforms/postcss.ts
patch: @@ -87,6 +87,10 @@ export default async function transform(
   });
 
   const assets = [];
+  const filePaths: string[] =[];
+  const buildFilePaths: string[] = [];
+  const directories: Array<[string, string]> = [];
+
   for (const msg of messages) {
     switch (msg.type) {
       case "asset":
@@ -104,36 +108,28 @@ export default async function transform(
         break;
       case "dependency":
       case "missing-dependency":
-        ipc.sendInfo({
-          type: "fileDependency",
-          path: toPath(msg.file),
-        });
+        filePaths.push(toPath(msg.file));
         break;
       case "build-dependency":
-        ipc.sendInfo({
-          type: "buildDependency",
-          path: toPath(msg.file),
-        });
+        buildFilePaths.push(toPath(msg.file));
         break;
       case "dir-dependency":
-        ipc.sendInfo({
-          type: "dirDependency",
-          path: toPath(msg.dir),
-          glob: msg.glob,
-        });
+        directories.push([toPath(msg.dir), msg.glob]);
         break;
       case "context-dependency":
-        ipc.sendInfo({
-          type: "dirDependency",
-          path: toPath(msg.file),
-          glob: "**",
-        });
+        directories.push([toPath(msg.dir), "**"]);
         break;
       default:
         // TODO: do we need to do anything here?
         break;
     }
   }
+  ipc.sendInfo({
+    type: "dependencies",
+    filePaths,
+    directories,
+    buildFilePaths,
+  });
   return {
     css,
     map: sourceMap ? JSON.stringify(map) : undefined,

filename: turbopack/crates/turbopack-node/js/src/transforms/webpack-loaders.ts
patch: @@ -19,27 +19,17 @@ import { type StructuredError } from "src/ipc";
 
 export type IpcInfoMessage =
   | {
-    type: "fileDependency";
-    path: string;
-  }
-  | {
-    type: "buildDependency";
-    path: string;
-  }
-  | {
-    type: "dirDependency";
-    path: string;
-    glob: string;
-  }
-  | {
-    type: "envDependency";
-    name: string;
-  }
+      type: 'dependencies'
+      envVariables?: string[]
+      directories?: Array<[string, string]>
+      filePaths?: string[],
+      buildFilePaths?: string[],
+    }
   | {
     type: "emittedError";
     severity: "warning" | "error";
     error: StructuredError;
-  }
+    }
   | {
     type: "log";
     time: number;
@@ -364,7 +354,8 @@ const transform = (
                   // TODO: do we need to handle this?
                   break
               }
-
+              // TODO(lukesandberg): should we batch these and flush lazily?
+              // turbopack just collects these and reports them when finishing the task.
               ipc.sendInfo({
                 type: "log",
                 time: Date.now(),
@@ -469,25 +460,15 @@ const transform = (
         },
       },
       (err, result) => {
-        for (const envVar of readEnvVars) {
-          ipc.sendInfo({
-            type: "envDependency",
-            name: envVar,
-          });
-        }
-        for (const dep of result.contextDependencies) {
-          ipc.sendInfo({
-            type: "dirDependency",
-            path: toPath(dep),
-            glob: "**",
-          });
-        }
-        for (const dep of result.fileDependencies) {
-          ipc.sendInfo({
-            type: "fileDependency",
-            path: toPath(dep),
-          });
-        }
+        ipc.sendInfo({
+          type: 'dependencies',
+          envVariables: Array.from(readEnvVars),
+          filePaths: result.fileDependencies.map(toPath),
+          directories: result.contextDependencies.map((dep) => [
+            toPath(dep),
+            '**',
+          ]),
+        });
         if (err) return reject(err);
         if (!result.result) return reject(new Error("No result from loaders"));
         const [source, map] = result.result;
@@ -518,17 +499,17 @@ function makeErrorEmitter(
       error:
         error instanceof Error
           ? {
-            name: error.name,
-            message: error.message,
-            stack: error.stack ? parseStackTrace(error.stack) : [],
-            cause: undefined,
-          }
+              name: error.name,
+              message: error.message,
+              stack: error.stack ? parseStackTrace(error.stack) : [],
+              cause: undefined,
+            }
           : {
             name: "Error",
-            message: error,
-            stack: [],
-            cause: undefined,
-          },
+              message: error,
+              stack: [],
+              cause: undefined,
+            },
     });
   };
 }

filename: turbopack/crates/turbopack-node/src/transforms/webpack.rs
patch: @@ -3,6 +3,7 @@ use std::mem::take;
 use anyhow::{bail, Context, Result};
 use async_trait::async_trait;
 use either::Either;
+use futures::try_join;
 use serde::{Deserialize, Serialize};
 use serde_json::{json, Value as JsonValue};
 use serde_with::serde_as;
@@ -348,18 +349,18 @@ pub struct LogInfo {
 #[derive(Deserialize, Debug)]
 #[serde(tag = "type", rename_all = "camelCase")]
 pub enum InfoMessage {
-    FileDependency {
-        path: RcStr,
-    },
-    BuildDependency {
-        path: RcStr,
-    },
-    DirDependency {
-        path: RcStr,
-        glob: RcStr,
-    },
-    EnvDependency {
-        name: RcStr,
+    // Sent to inform Turbopack about the dependencies of the task.
+    // All fields are `default` since it is ok for the client to
+    // simply omit instead of sending empty arrays.
+    Dependencies {
+        #[serde(default)]
+        env_variables: Vec<RcStr>,
+        #[serde(default)]
+        file_paths: Vec<RcStr>,
+        #[serde(default)]
+        directories: Vec<(RcStr, RcStr)>,
+        #[serde(default)]
+        build_file_paths: Vec<RcStr>,
     },
     EmittedError {
         severity: IssueSeverity,
@@ -472,29 +473,60 @@ impl EvaluateContext for WebpackLoaderContext {
         pool: &NodeJsPool,
     ) -> Result<()> {
         match data {
-            InfoMessage::FileDependency { path } => {
-                // TODO We might miss some changes that happened during execution
-                // Read dependencies to make them a dependencies of this task. This task will
-                // execute again when they change.
-                self.cwd.join(path).read().await?;
-            }
-            InfoMessage::BuildDependency { path } => {
-                // TODO We might miss some changes that happened during execution
-                BuildDependencyIssue {
-                    context_ident: self.context_ident_for_issue,
-                    path: self.cwd.join(path).to_resolved().await?,
+            InfoMessage::Dependencies {
+                env_variables,
+                file_paths,
+                directories,
+                build_file_paths,
+            } => {
+                // Track dependencies of the loader task
+                // TODO: Because these are reported _after_ the loader actually read the dependency
+                // there is a race condition where we may miss updates that race
+                // with the loader execution.
+
+                // Track all the subscriptions in parallel, since certain loaders like tailwind
+                // might add thousands of subscriptions.
+                let env_subscriptions = env_variables
+                    .iter()
+                    .map(|e| self.env.read(e.clone()))
+                    .try_join();
+                let file_subscriptions = file_paths
+                    .iter()
+                    .map(|p| self.cwd.join(p.clone()).read())
+                    .try_join();
+                let directory_subscriptions = directories
+                    .iter()
+                    .map(|(dir, glob)| {
+                        // TODO: there is some redundancy between `dir_dependency` and what
+                        // `read_glob` does, Introduce a new read_glob
+                        // option that will track all files the way
+                        // `dir_dependency` does but in a single traversal.
+                        dir_dependency(
+                            self.cwd
+                                .join(dir.clone())
+                                .read_glob(Glob::new(glob.clone()), false),
+                        )
+                    })
+                    .try_join();
+                let build_paths = build_file_paths
+                    .iter()
+                    .map(|path| self.cwd.join(path.clone()).to_resolved())
+                    .try_join();
+                let (resolved_build_paths, ..) = try_join!(
+                    build_paths,
+                    env_subscriptions,
+                    file_subscriptions,
+                    directory_subscriptions
+                )?;
+
+                for build_path in resolved_build_paths {
+                    BuildDependencyIssue {
+                        context_ident: self.context_ident_for_issue,
+                        path: build_path,
+                    }
+                    .resolved_cell()
+                    .emit();
                 }
-                .resolved_cell()
-                .emit();
-            }
-            InfoMessage::DirDependency { path, glob } => {
-                // TODO We might miss some changes that happened during execution
-                // Read dependencies to make them a dependencies of this task. This task will
-                // execute again when they change.
-                dir_dependency(self.cwd.join(path).read_glob(Glob::new(glob), false)).await?;
-            }
-            InfoMessage::EnvDependency { name } => {
-                self.env.read(name).await?;
             }
             InfoMessage::EmittedError { error, severity } => {
                 EvaluateEmittedErrorIssue {